在 AI 应用开发中，高效整合大模型、向量数据库、工具调用与会话记忆是核心需求。LangChain 作为开源的大模型应用框架，提供了一站式解决方案，能快速串联各类组件，降低开发门槛。本文基于实战经验，从基础组件入手，逐步深入到 RAG 系统、多 Agent 开发与会话记忆优化，附完整可运行代码，帮助开发者快速上手 LangChain 全链路开发。

一、LangChain 核心组件快速上手

LangChain 的核心优势在于组件化设计，常用核心组件包括提示词模板、输出解析器、链式调用等，是构建复杂应用的基础。

1. 提示词模板（PromptTemplate）

提示词模板用于标准化大模型输入，支持动态变量填充与默认值设置，避免重复编写冗余提示词。

from langchain_core.prompts import PromptTemplate

# 基础模板：固定格式+动态变量
template = """你是一位专业的{domain}顾问，请用{language}回答：
问题：{question}
回答:"""

# 实例化模板并填充变量
prompt = PromptTemplate(
    template=template,
    input_variables=["domain", "language", "question"]
)

# 生成完整提示词
print(prompt.format(domain="机器学习", language="中文", question="如何使用LangChain构建RAG？"))

# 带默认值的模板
template2 = """分析用户情绪（默认分析类型：{analysis_type}）
用户输入:{user_input}
分析结果"""

prompt2 = PromptTemplate(
    template=template2,
    input_variables=["user_input"],
    partial_variables={"analysis_type": "sentiment"}  # 预设默认值
)
print(prompt2.format(user_input="今天天气真不错！"))

2. 输出解析器（OutputParser）

大模型输出默认是字符串，解析器可将其转换为列表、JSON 等结构化格式，方便后续处理。

from langchain_core.output_parsers import (
    StrOutputParser,
    CommaSeparatedListOutputParser,
    JsonOutputParser
)
from langchain_openai import ChatOpenAI
from pydantic import SecretStr

# 初始化大模型（对接通义千问）
model = ChatOpenAI(
    model="qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=SecretStr("你的API密钥"),
    temperature=0.7
)

# 1. 字符串解析器（基础款）
str_parser = StrOutputParser()
chain = prompt | model | str_parser
print(chain.invoke({"domain": "Python", "language": "中文", "question": "列表推导式语法"}))

# 2. 逗号分隔列表解析器
list_parser = CommaSeparatedListOutputParser()
list_prompt = PromptTemplate(
    template="列举5个{topic}相关技术，{format_instructions}",
    input_variables=["topic"],
    partial_variables={"format_instructions": list_parser.get_format_instructions()}
)
list_chain = list_prompt | model | list_parser
print(list_chain.invoke({"topic": "大模型应用"}))

# 3. JSON解析器（结构化输出）
json_parser = JsonOutputParser()
json_prompt = PromptTemplate(
    template="回答问题并返回JSON格式：{{'answer':'答案','confidence':置信度}}，问题：{question}",
    input_variables=["question"]
)
json_chain = json_prompt | model | json_parser
result = json_chain.invoke({"question": "地球半径是多少？"})
print(f"答案：{result['answer']}，置信度：{result['confidence']}")

3. 链式调用（LCEL）

LangChain Expression Language（LCEL）通过|符号串联组件，实现 "提示词→模型→解析器" 的一站式流程，代码简洁且易维护。

# 流式输出链（实时返回结果）
from langchain_core.prompts import ChatPromptTemplate

stream_prompt = ChatPromptTemplate.from_template("用100字解释{concept}")
stream_chain = stream_prompt | model | str_parser
# 流式迭代输出
for chunk in stream_chain.stream({"concept": "向量数据库"}):
    print(chunk, end="", flush=True)

二、RAG 系统构建：文档加载→向量存储→检索增强

检索增强生成（RAG）是 LangChain 的核心应用场景，能让大模型结合外部文档回答问题，解决知识时效性与准确性问题。

1. 文档加载与预处理

支持加载文本、PDF、Word、网页等多种格式文档，通过文本分割器拆分长文档，提升检索精度。

from langchain_community.document_loaders import (
    TextLoader, PyPDFLoader, Docx2txtLoader, WebBaseLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1. 加载网页文档（以Milvus官方文档为例）
loader = WebBaseLoader(
    ["https://milvus.io/docs/zh/overview.md"],
    requests_kwargs={"headers": {"Accept-Language": "zh-CN"}}
)
docs = loader.load()

# 2. 文本分割（递归分割，保留上下文）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1024,  # 每个片段最大字符数
    chunk_overlap=20   # 片段重叠字符数
)
split_docs = text_splitter.split_documents(docs)

2. 向量存储与检索（Milvus 实战）

向量数据库是 RAG 的核心，本文以 Milvus 为例，实现文档向量入库与相似性检索。

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_milvus import Milvus

# 初始化嵌入模型（通义千问文本嵌入）
embeddings = DashScopeEmbeddings(
    model="text-embedding-v2",
    dashscope_api_key="你的API密钥"
)

# 向量入库（Milvus）
vector_store = Milvus.from_documents(
    documents=split_docs,
    embedding=embeddings,
    collection_name="doc_qa_db",
    connection_args={"uri": "http://192.168.64.137:19530"},
    drop_old=True  # 覆盖旧集合
)

# 构建检索器
retriever = vector_store.as_retriever(search_kwargs={"k": 3})  # 返回Top3相似文档

# 基础相似性检索
query = "Milvus如何创建Collection？"
docs = retriever.invoke(query)
for doc in docs:
    print(f"相关文档：{doc.page_content[:100]}")

3. 高级检索：MMR 与多查询优化

通过 MMR（最大边际相关性）提升结果多样性，多查询检索器扩大检索范围，提升召回率。

# MMR检索（平衡相似度与多样性）
mmr_results = vector_store.max_marginal_relevance_search(
    query=query,
    k=3,
    fetch_k=10,  # 候选集大小
    lambda_mult=0.4  # 0→侧重多样性，1→侧重相似度
)

# 多查询检索器（LLM生成多个查询角度）
from langchain.retrievers import MultiQueryRetriever

multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=retriever,
    llm=model
)
multi_results = multi_query_retriever.invoke(query)
print(f"多查询检索到的文档数：{len(multi_results)}")

4. 完整 RAG 链构建

串联检索器、提示词模板与大模型，实现 "检索→生成" 全流程。

from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough

# 定义RAG提示词模板
rag_prompt = PromptTemplate.from_template("""
基于以下上下文回答问题，最多10句话，简洁准确：
{context}
问题：{question}
""")

# 构建RAG链
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | rag_prompt
    | model
    | str_parser
)

# 执行问答
result = rag_chain.invoke("Milvus的Docker安装命令是什么？")
print(result)

三、工具调用与多 Agent 开发

LangChain 支持大模型自动调用外部工具（如计算器、联网搜索、数据库查询），结合多 Agent 架构实现复杂任务协作。

1. 自定义工具开发

通过@tool装饰器快速定义工具，支持参数校验与异常处理。

from langchain_core.tools import tool
from pydantic import Field, BaseModel

# 工具参数校验模型
class CalculatorInput(BaseModel):
    a: int = Field(..., description="第一个整数")
    b: int = Field(..., description="第二个整数")

# 自定义乘法工具
@tool("multiply-tool", args_schema=CalculatorInput, description="计算两个整数的乘法")
def multiply(a: int, b: int) -> int:
    return a * b

# 联网搜索工具（对接SearchApi）
import os
from langchain_community.utilities import SearchApiAPIWrapper

os.environ["SEARCHAPI_API_KEY"] = "你的API密钥"
search = SearchApiAPIWrapper()

@tool("web-search", return_direct=True)
def web_search(query: str) -> str:
    """获取实时信息（如股价、新闻、最新版本）"""
    result = search.results(query)
    return "\n\n".join([f"来源：{res['title']}\n内容：{res['snippet']}" for res in result['organic_results']])

2. 单 Agent 工具调用

让大模型根据问题自动选择工具，实现 "思考→行动→观察" 的闭环。

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# 工具列表
tools = [multiply, web_search]

# Agent提示词模板
agent_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是智能助手，必要时调用工具解决问题"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 创建Agent
agent = create_tool_calling_agent(llm=model, tools=tools, prompt=agent_prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 测试Agent（实时股价查询+计算）
agent_executor.invoke({
    "input": "比亚迪今天股价是多少？再计算123*456"
})

3. 多 Agent 协作（RunnableBranch 路由）

通过路由机制将不同类型问题分配给对应专业 Agent，提升回答精准度。

from langchain_core.runnables import RunnableBranch, RunnableLambda

# 定义专业子链
# 技术问题子链
tech_prompt = ChatPromptTemplate.from_template("你是技术专家，回答：{input}")
tech_chain = tech_prompt | model | str_parser

# 财务问题子链
billing_prompt = ChatPromptTemplate.from_template("你是财务专员，回答：{input}")
billing_chain = billing_prompt | model | str_parser

# 路由判断函数
def is_tech_question(input: dict) -> bool:
    tech_keywords = ["技术", "安装", "错误", "代码"]
    return any(keyword in input["input"] for keyword in tech_keywords)

def is_billing_question(input: dict) -> bool:
    billing_keywords = ["账单", "支付", "发票"]
    return any(keyword in input["input"] for keyword in billing_keywords)

# 构建路由链
branch_chain = RunnableBranch(
    (is_tech_question, tech_chain),
    (is_billing_question, billing_chain),
    lambda x: "请输入技术或财务相关问题"  # 默认回复
)

# 测试路由
print(branch_chain.invoke({"input": "如何解决LangChain工具调用错误？"}))
print(branch_chain.invoke({"input": "如何查询本月发票？"}))

四、会话记忆优化：从短期记忆到长期持久化

会话记忆让 Agent 能记住上下文，支持多轮对话，本文涵盖内存记忆、Redis 持久化与多租户隔离。

1. 基础会话记忆（ConversationSummaryMemory）

自动总结对话历史，避免记忆过长导致的性能问题。

from langchain.memory import ConversationSummaryMemory

# 初始化记忆
memory = ConversationSummaryMemory(
    llm=model,
    return_messages=True,
    memory_key="chat_history"
)

# 构建带记忆的对话链
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是AI助手，基于历史对话回答：{chat_history}"),
    ("human", "{input}")
])

chat_chain = (
    RunnablePassthrough.assign(
        chat_history=lambda _: memory.load_memory_variables({})["chat_history"]
    )
    | chat_prompt
    | model
    | str_parser
)

# 多轮对话
queries = [
    "我叫小明，是程序员",
    "什么是LangChain？",
    "我是谁？从事什么工作？"  # 测试记忆能力
]

for query in queries:
    result = chat_chain.invoke({"input": query})
    print(f"用户：{query}")
    print(f"AI：{result}\n")
    memory.save_context({"input": query}, {"output": result})

2. Redis 持久化记忆（多用户隔离）

通过 Redis 存储对话历史，支持多用户、多会话隔离，重启程序不丢失记忆。

from langchain_redis import RedisChatMessageHistory
from langchain_core.runnables import RunnableWithMessageHistory

# Redis配置
REDIS_URL = "redis://127.0.0.1:6379"

# 获取会话历史（多用户+多会话隔离）
def get_session_history(user_id: str, session_id: str):
    return RedisChatMessageHistory(
        f"{user_id}_{session_id}",
        redis_url=REDIS_URL
    )

# 构建带持久化记忆的链
with_memory_chain = RunnableWithMessageHistory(
    chat_chain,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    history_factory_config=[
        {"id": "user_id", "annotation": str, "name": "用户ID"},
        {"id": "session_id", "annotation": str, "name": "会话ID"}
    ]
)

# 不同用户会话测试
user1_resp1 = with_memory_chain.invoke(
    {"input": "我叫小红"},
    config={"configurable": {"user_id": "1", "session_id": "1"}}
)

user2_resp1 = with_memory_chain.invoke(
    {"input": "我叫小李"},
    config={"configurable": {"user_id": "2", "session_id": "1"}}
)

# 验证隔离性
user1_resp2 = with_memory_chain.invoke(
    {"input": "我是谁？"},
    config={"configurable": {"user_id": "1", "session_id": "1"}}
)
print(user1_resp2)  # 输出：你叫小红

五、调试与监控：LangSmith 实战

LangSmith 是 LangChain 官方调试工具，支持链路追踪、性能监控与异常告警。

import os

# 配置LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "你的LangSmith密钥"
os.environ["LANGCHAIN_PROJECT"] = "langchain-demo"

# 执行链（自动追踪）
result = rag_chain.invoke("LangSmith如何查看调用链路？")
print(result)

配置完成后，访问LangSmith 官网即可查看：

• 调用链路：输入输出、耗时、Token 用量
• 监控数据：成功率、响应时间分布
• 异常告警：超时、失败率阈值提醒

六、实战总结与进阶方向

本文覆盖 LangChain 核心组件、RAG 系统、工具调用、多 Agent 与会话记忆，通过实战案例展示了从基础到进阶的全链路开发。进阶方向推荐：

1. 性能优化：向量缓存（CacheBackedEmbeddings）、索引优化（Milvus HNSW 索引）
2. 复杂 Agent：LangGraph 状态管理、多 Agent 协作（如分工执行复杂任务）
3. 多模态 RAG：整合图片、音频等多模态文档，扩展应用场景
4. 部署落地：FastAPI 封装接口、Docker 容器化部署、K8s 集群扩展