LangChain 是当前最流行的 LLM 应用开发框架之一,为开发者提供了构建 AI 应用所需的核心抽象层——从简单的 LLM 调用、Prompt 管理,到复杂的 Agent 编排、多步推理与外部工具集成。自 2022 年底开源以来,LangChain 已迭代至 1.0 稳定版本,生态扩展出 LangGraph、LangSmith、LangMem 等子项目,形成了一个完整的 LLM 应用开发生态。
LangChain 围绕几个核心抽象构建,这些抽象构成了所有高级功能的基础:
LangChain 提供统一的模型接口,封装了不同 LLM 提供商的 API 差异:
统一接口意味着切换模型提供商只需更改模型名称和 API Key,无需重写业务逻辑。
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
# 统一接口,不同提供商
gpt4 = ChatOpenAI(model="gpt-4o", temperature=0)
claude = ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0)
Prompt 管理是 LLM 应用开发的基础。LangChain 提供:
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个{role}专家,请用{language}回答。"),
("human", "{question}"),
])
# 使用
chain = prompt | model
result = chain.invoke({
"role": "金融分析",
"language": "中文",
"question": "解释杜邦分析法"
})
LLM 输出格式不稳定的痛点通过解析器解决:
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
class AnalysisResult(BaseModel):
summary: str = Field(description="总结")
risk_level: str = Field(description="风险等级: low/medium/high")
suggestions: list[str] = Field(description="建议列表")
parser = PydanticOutputParser(pydantic_object=AnalysisResult)
LCEL 是 LangChain 的核心设计创新,它通过管道操作符 | 将组件连接成声明式执行链。每个 LCEL 组件都实现 Runnable 接口,提供标准化的 invoke、batch、stream 和 astream 方法。
所有 LCEL 组件都继承自 Runnable,这意味着:
from langchain_core.runnables import RunnablePassthrough
# 声明式管道
chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
)
# 统一调用接口
result = chain.invoke("什么是机器学习?")
for chunk in chain.stream("解释强化学习"):
print(chunk, end="")
| 组合器 | 类型 | 作用 | 用例 |
|---|---|---|---|
RunnablePassthrough() |
透传 | 原样传递输入 | 提取输入字段 |
RunnableParallel |
并行 | 多个 Runnable 并行执行 | 同时检索多源 |
RunnableBranch |
条件分支 | 根据条件选择执行路径 | 路由不同任务 |
RunnableLambda |
自定义函数 | 将任意函数包装为 Runnable | 自定义逻辑 |
itemgetter |
字段提取 | 从字典中提取指定字段 | 提取上下文 |
RunnableMap |
映射 | 对输入应用函数映射 | 数据转换 |
from langchain_core.runnables import RunnableParallel
# 并行执行示例:多角度分析
chain = RunnableParallel(
summary=summary_prompt | model,
sentiment=sentiment_prompt | model,
keywords=keywords_prompt | model_cheap,
)
LCEL 天然支持流式输出,这在交互式应用中至关重要:
async for chunk in chain.astream({"query": "写一首关于程序员的诗"}):
print(chunk, end="", flush=True)
通过 abatch、ainvoke、astream 统一支持异步,适合 Web 服务等 I/O 密集型场景。
Chain 是 LangChain 最早的核心概念——将多个 LLM 调用或工具操作编排为可复用的处理流程。
| Chain | 用途 | 适用场景 |
|---|---|---|
LLMChain |
单步 Prompt+LLM 调用 | 基础问答 |
SimpleSequentialChain |
顺序执行,前一步输出为后一步输入 | 流水线处理 |
SequentialChain |
顺序执行,支持多输入多输出 | 复杂多步任务 |
RouterChain |
根据输入路由到不同子链 | 多意图分类 |
TransformChain |
自定义转换逻辑 | 数据预处理 |
from langchain.chains import SequentialChain, TransformChain
# 步骤1:文本清洗
def clean_text(inputs):
text = inputs["raw_text"]
cleaned = text.strip().replace("\r\n", "\n")
return {"cleaned_text": cleaned}
clean_chain = TransformChain(
input_variables=["raw_text"],
output_variables=["cleaned_text"],
transform=clean_text
)
# 步骤2:总结
summary_chain = (
ChatPromptTemplate.from_template("总结以下内容:\n{cleaned_text}")
| model
| StrOutputParser()
)
# 步骤3:提取关键点
keypoint_chain = (
ChatPromptTemplate.from_template("提取关键论点:\n{cleaned_text}")
| model
| StrOutputParser()
)
# 组合为流水线
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
pipeline = (
RunnablePassthrough()
| clean_chain
| RunnableParallel(
summary=summary_chain,
keypoints=keypoint_chain
)
)
复杂链通常需要状态传递。LangChain 通过 RunnableParallel 和 RunnablePassthrough.assign 实现链式状态管理:
from langchain_core.runnables import RunnablePassthrough
chain = (
RunnablePassthrough.assign(
cleaned=lambda x: clean_text(x),
timestamp=lambda _: datetime.now().isoformat()
)
| RunnableParallel(
summary=summary_chain,
analysis=analysis_chain
)
)
Tools 是 LangChain 赋予 LLM 与外部世界交互能力的核心机制。每个 Tool 封装了一个函数或 API,并附带清晰的描述供 LLM 理解何时调用。
from langchain_core.tools import tool
# 方式1:装饰器(推荐)
@tool
def search_knowledge_base(query: str) -> str:
"""搜索知识库获取相关信息。输入应为具体的搜索关键词。"""
# 实际调用搜索 API
return f"关于 '{query}' 的结果..."
# 方式2:Tool 类
from langchain_core.tools import Tool
calculator = Tool(
name="calculator",
func=eval,
description="执行数学运算,输入应为 Python 表达式"
)
每个 Tool 定义包含:
from pydantic import BaseModel, Field
class EmailInput(BaseModel):
to: str = Field(description="收件人邮箱")
subject: str = Field(description="邮件主题")
body: str = Field(description="邮件正文")
@tool(args_schema=EmailInput)
def send_email(to: str, subject: str, body: str) -> str:
"""发送电子邮件。"""
# smtp_client.send(to, subject, body)
return f"邮件已发送至 {to}"
LangChain 提供 700+ 集成工具,涵盖:
| 工具类型 | 示例 |
|---|---|
| 搜索引擎 | Tavily Search、Google Search、Bing Search |
| 代码解释器 | Python REPL、JS REPL |
| 数据处理 | CSV Agent、Pandas Agent、SQL Agent |
| 文件操作 | 文件读写、PDF 解析 |
| API 调用 | Arxiv、Wikipedia、Wolfram Alpha |
| 数据库 | SQL、向量数据库、图数据库 |
记忆机制使 LLM 应用具备跨轮对话的上下文保持能力。LangChain 提供多种 Memory 实现,适用于不同场景。
| Memory 类型 | 存储内容 | 适用场景 |
|---|---|---|
ConversationBufferMemory |
原始完整对话 | 简单对话 |
ConversationBufferWindowMemory |
最近 k 轮对话 | 控制 Token 消耗 |
ConversationSummaryMemory |
压缩摘要 | 长对话 |
ConversationSummaryBufferMemory |
摘要+缓冲 | 平衡细节与 Token |
VectorStoreRetrieverMemory |
语义检索 | 超长上下文/个性化 |
ConversationTokenBufferMemory |
基于 Token 数截断 | 精确控制预算 |
PostgresChatMessageHistory |
持久化存储 | 生产环境 |
对于需要长期记忆的场景,使用 VectorStoreRetrieverMemory:
from langchain.memory import VectorStoreRetrieverMemory
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
vectorstore = Chroma(
embedding_function=OpenAIEmbeddings(),
collection_name="user_memories"
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
memory = VectorStoreRetrieverMemory(
retriever=retriever,
memory_key="history",
input_key="input"
)
# 存储记忆
memory.save_context({"input": "我喜欢科幻小说"}, {"output": "已记录"})
# 检索记忆
relevant = memory.load_memory_variables({"input": "推荐一本书"})
LangMem 是 LangChain 生态中专门用于 Agent 长期记忆的子项目(2025 年推出),支持:
from langmem import create_memory_manager
manager = create_memory_manager(
model="gpt-4o",
max_tokens=2000
)
# 存储跨会话记忆
manager.add_memory(
user_id="hugo",
content="用户是一名软件架构师,关注 AI/LLM 技术"
)
Agent 是 LangChain 的核心价值所在——让 LLM 自主决定调用哪些工具、以什么顺序调用、以及如何整合结果。
Agent 遵循"思考-行动-观察"循环:
用户输入 → LLM 思考(Reason)→ 选择行动(Act)→ 调用工具 → 观察结果(Observe)
↓
LLM 再次思考 → 是否还需要行动?
↓
最终输出
| 版本 | Agent 类型 | 核心机制 | 状态 |
|---|---|---|---|
| 早期 | zero-shot-react-description |
ReAct 范式 | 经典 |
| 0.1 | openai-functions |
Function Calling | 经典 |
| 0.2 | openai-tools |
Tool Calling | 推荐 |
| 0.3+ | LangGraph Agent | 图编排 | 生产推荐 |
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_community.tools import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
# 1. 定义工具
tools = [
TavilySearchResults(max_results=3),
search_knowledge_base,
calculator
]
# 2. 定义 Prompt
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个智能助手,可以使用以下工具回答问题:{tools}"),
("placeholder", "{chat_history}"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
# 3. 创建 Agent
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 4. 执行
result = agent_executor.invoke({
"input": "搜索2025年AI框架的最新趋势,然后总结"
})
| 参数 | 说明 | 默认值 | 建议 |
|---|---|---|---|
max_iterations |
最大循环次数 | 15 | 5-10 |
max_execution_time |
超时(秒) | 无 | 60-300 |
early_stopping_method |
停止策略 | force |
generate 更稳定 |
handle_parsing_errors |
解析错误处理 | False | True |
return_intermediate_steps |
返回中间步骤 | False | 调试时开启 |
import requests
from langchain_core.tools import tool
@tool
def query_internal_api(query: str) -> str:
"""查询内部订单系统。查询格式:order_id 或 customer_email"""
api_key = os.environ.get("INTERNAL_API_KEY")
resp = requests.post(
"https://api.internal.com/query",
json={"query": query},
headers={"Authorization": f"Bearer {api_key}"},
timeout=10
)
return resp.json()["result"]
LangGraph 是 LangChain 生态中 Agent 编排的核心组件,支持用有向图构建复杂、可定制的 Agent 工作流。可以将其理解为"Agent 的状态机"。
传统 LangChain Agent 的局限:
LangGraph 通过图结构解决以上所有问题。
from langgraph.graph import StateGraph, State
from typing import TypedDict, Annotated, Sequence
import operator
class AgentState(TypedDict):
messages: Annotated[Sequence, operator.add]
next_step: str
# 定义节点
def call_model(state: AgentState):
response = model.invoke(state["messages"])
return {"messages": [response]}
def call_tool(state: AgentState):
tool_call = state["messages"][-1].tool_calls[0]
result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
return {"messages": [ToolMessage(result, tool_call_id=tool_call["id"])]}
# 条件边:决定下一步
def should_continue(state: AgentState) -> str:
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return "end"
# 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", call_tool)
graph.set_entry_point("agent")
graph.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "end": END}
)
graph.add_edge("tools", "agent")
# 编译
app = graph.compile()
LangGraph 的一大特色是内置中断机制,可在任意节点暂停等待人类审批:
from langgraph.checkpoint import MemorySaver
# 启用持久化和中断
graph_with_interrupts = graph.compile(
checkpointer=MemorySaver(),
interrupt_before=["tools"] # 在调用工具前暂停
)
# 执行(会暂停在调用工具前)
for event in graph_with_interrupts.stream(
{"messages": [HumanMessage("帮我转账1000元给张三")]},
config
):
# 人工审批后继续...
pass
LangGraph 0.3(2025 年发布)引入了预构建 Agent 体系,将常见模式封装为即插即用组件:
| 预构建 Agent | 用途 |
|---|---|
create_react_agent |
标准 ReAct 模式 |
LangGraph Supervisor |
多 Agent 监督架构 |
LangGraph Swarm |
蜂群式多 Agent 协作 |
Trustcall |
可靠的结构化信息抽取 |
create_tool_agent |
通用工具调用 Agent |
from langgraph.prebuilt import create_react_agent
# 一行创建标准 Agent
agent = create_react_agent(
model=model,
tools=tools,
prompt="你是一个数据分析助手"
)
LangChain 提供完整的 RAG 流水线抽象,从文档加载到最终生成全覆盖。
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
# 1. 加载文档
loader = TextLoader("documents/tech_manual.txt")
docs = loader.load()
# 2. 文档切分
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
chunks = splitter.split_documents(docs)
# 3. 存储向量
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=OpenAIEmbeddings()
)
# 4. 检索
retriever = vectorstore.as_retriever(
search_type="mmr", # 最大边际相关性
search_kwargs={"k": 4, "fetch_k": 20}
)
# 5. 生成(LCEL 链)
from langchain_core.runnables import RunnablePassthrough
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| ChatPromptTemplate.from_template(
"基于以下上下文回答问题:\n{context}\n\n问题:{question}"
)
| model
| StrOutputParser()
)
| 策略 | 描述 | LangChain 实现 |
|---|---|---|
| 查询重写 | 优化用户查询后再检索 | RunnableBranch + 重写 Prompt |
| 多向量检索 | 摘要检索 + 全文返回 | ParentDocumentRetriever |
| 融合检索 | 关键词 + 语义混合 | EnsembleRetriever |
| 自我反思 | 检索后自查是否需要补充 | SelfQueryRetriever |
| Agent RAG | 用 Agent 动态决定检索策略 | Agent + Retriever Tool |
LangChain 已从单一框架演变为完整的 LLM 应用平台:
LangChain 生态
├── LangChain(核心框架)
│ ├── Models(模型抽象)
│ ├── Prompts(提示管理)
│ ├── Chains(流水线)
│ ├── Tools(工具集成)
│ ├── Memory(记忆系统)
│ └── Agents(Agent 运行时)
├── LangGraph(Agent 编排)
│ ├── 有向图执行引擎
│ ├── 持久化与中断
│ ├── 预构建 Agent
│ └── 多 Agent 架构
├── LangSmith(可观测性平台)
│ ├── Tracing(追踪)
│ ├── Evaluation(评估)
│ └── Monitoring(监控)
├── LangMem(长期记忆)
├── Deep Agents(深度 Agent SDK)
└── 700+ 集成(提供商、向量库、工具)
LangSmith 是 LangChain 的运维平台,提供:
# 一行开启追踪
from langsmith import trace
with trace("my-agent-run") as rt:
result = agent_executor.invoke({"input": "分析这份财务报表"})
from langchain_core.runnables import RunnableConfig
from langchain_core.exceptions import OutputParserException
def safe_invoke(chain, input_data, max_retries=2):
for attempt in range(max_retries):
try:
return chain.invoke(input_data)
except OutputParserException as e:
if attempt == max_retries - 1:
raise
continue
except Exception as e:
print(f"Attempt {attempt + 1} failed: {e}")
raise
ConversationTokenBufferMemory 精确控制上下文窗口max_iterations 避免 Agent 陷入无限循环RunnableBranch 先走轻量模型过滤return_intermediate_steps=False 减少传输数据langchain.globals.set_llm_cache)batch 合并stream 而非 invokeainvoke / astream| 特性 | LangChain | LlamaIndex | CrewAI | AutoGen |
|---|---|---|---|---|
| RAG 能力 | 强(全流水线) | 最强(数据为中心) | 弱 | 中 |
| Agent 编排 | LangGraph(强) | 中 | 专精 | 专精 |
| 工具生态 | 700+ | 300+ | 有限 | 有限 |
| 学习曲线 | 中等 | 中等 | 低 | 高 |
| 生产就绪度 | 高 | 高 | 中 | 中 |
| 多 Agent | LangGraph Supervisor | 有限 | 专精 | 专精 |
| 可观测性 | LangSmith | 基础 | 有限 | 基础 |
选择建议:
| 版本 | 时间 | 重要变更 |
|---|---|---|
| 0.0.x | 2023 | 初始版本,Chains 为核心 |
| 0.1.x | 2024 Q1 | LCEL 引入,Runnable 接口 |
| 0.2.x | 2024 Q3 | 模块化拆分(langchain-community/openai/anthropic) |
| 0.3.x | 2024 Q4 | LangGraph 0.3 预构建 Agent |
| 1.0 | 2025 Q1 | 稳定版本,核心 API 冻结 |
| 2025+ | 2025-2026 | LangMem、Deep Agents、多模态支持 |
LangChain 从最初的 Chain 编排框架演进为完整的 LLM 应用平台,其核心价值在于:
对于构建复杂 LLM 应用的开发者,LangChain 生态是目前最成熟的选择。建议从 LCEL 链入手,逐步过渡到 LangGraph Agent,最后通过 LangSmith 保障生产质量。
此页面为 AI 知识体系 的一部分,内容持续更新中。
参考资源: