LangChain与LangGraph:从链式编排到图式工作流的技术演进
最近在准备大模型应用开发面试的朋友,可能都遇到过这样一个问题:“LangChain 和 LangGraph 有什么区别?” 这个问题看似基础,却直接考验你对当前大模型应用开发技术栈的理解深度。很多人会回答“LangGraph 是 LangChain 用来构建 Agent 的库”,这个答案没错,但太浅了,面试官听完只会觉得你停留在“会用”的层面,没理解背后的设计哲学和工程取舍。
真正拉开差距的回答,需要讲清楚: LangChain 解决的是“如何连接”的问题,而 LangGraph 解决的是“如何编排”的问题。 前者是构建应用的“积木”和“胶水”,后者是为这些积木设计“工作流程图”和“状态机”。理解这个区别,你才能知道在什么场景下该选谁,以及如何设计出更健壮、更可控的 AI 应用。
本文将带你深入 LangChain 和 LangGraph 的核心,不仅回答面试题,更帮你建立起一套从单体链式调用到复杂多智能体协作的系统性认知。我们会从概念、架构、代码示例到实战场景,完整拆解两者的区别与联系,让你在面试和实际项目中都能游刃有余。
1. 核心区别:从“链式思维”到“图式思维”
要理解两者的区别,首先要跳出“LangGraph 是 LangChain 的一部分”这个固有认知。虽然它们同出一源,但解决的问题域和抽象层级截然不同。
LangChain 的本质是“组件化”与“标准化连接”。 想象你要组装一台电脑,LangChain 就是为你提供了 CPU(LLM)、内存(Memory)、硬盘(Vector Store)、各种外设(Tools)以及标准化的接口(如 LCEL)。它的核心价值在于:
- 标准化接口 :无论底层用的是 OpenAI 还是 Anthropic 的模型,通过 LangChain 的
BaseChatModel接口,你的调用代码几乎不变。 - 丰富组件 :提供了大量开箱即用的模块,如文本分割器、向量化器、各种检索器、以及海量的第三方工具集成。
- 链式组合 :通过 LCEL(LangChain Expression Language)将多个组件像管道一样连接起来,形成可预测的线性或简单分支流程,例如
检索 -> 加工 -> 生成。
它的思维模式是 “链” (Chain) ,流程通常是顺序或带有简单条件分支的。
LangGraph 的本质是“有状态的工作流编排”。 继续用电脑的比喻,如果你的任务只是开机、打开浏览器、搜索一个词,那么链式思维足够了。但如果你要运行一个复杂的游戏,里面涉及角色状态更新、事件触发、多线程物理计算、渲染循环,这就需要更精细的调度。LangGraph 就是为这种复杂、有状态、可能循环或并行执行的 AI 智能体工作流设计的。 它的核心抽象是 “图” (Graph) 和 “状态” (State) :
- 节点 (Nodes) :代表一个执行单元,可以是一个工具调用、一次 LLM 生成,或任何函数。
- 边 (Edges) :决定工作流的走向。可以是条件边(根据上一步结果决定下一步),也可以是固定边。
- 状态 (State) :一个贯穿整个工作流的共享上下文对象。每个节点读取并更新状态的一部分,这使得智能体有了“记忆”和“上下文”。
- 循环 (Cycles) :图支持循环,这是实现“思考-行动-观察”这类 ReAct 模式 Agent 的关键。
它的思维模式是 “状态机” (State Machine) ,流程可以是循环、并行、带有复杂条件判断的。
一个简单的类比:
- LangChain 像 乐高积木和说明书 。说明书(Chain)告诉你按什么顺序拼接积木(Components)来搭出一个房子(简单应用)。
- LangGraph 像 一个自动化工厂的流程图 。它定义了原材料(输入)如何经过不同的加工站(节点),每个站如何根据半成品的状态(State)决定加工方式,并且产品可能需要送回某个站进行二次加工(循环),最终产出成品(复杂应用)。
理解了这层本质区别,我们再来深入看它们各自的核心架构。
2. LangChain 深度解析:组件化与链式编排
2.1 核心架构与概念
LangChain 的架构围绕几个核心概念构建,理解它们是使用它的基础:
- Schema :数据结构的基石。主要包括
Document(用于存储文本片段及其元数据)、Message(HumanMessage,AIMessage,SystemMessage等,用于与LLM对话)和Example(用于few-shot学习)。 - Models :大模型抽象层。将不同供应商(OpenAI, Anthropic, Cohere等)的模型统一封装为
LLM(补全模型)和ChatModel(聊天模型)接口。 - Prompts :提示词管理。提供
PromptTemplate、ChatPromptTemplate等,支持变量注入和少量示例管理,实现提示词与代码逻辑分离。 - Indexes :数据索引与检索。核心是
Vectorstore(向量数据库)和Retriever(检索器),用于构建外部知识库,实现 RAG(检索增强生成)。 - Memory :记忆管理。用于在多次交互中维护上下文,如
ConversationBufferMemory、ConversationSummaryMemory。 - Chains :链式组合。这是 LangChain 早期的核心编排方式,将上述组件按顺序组合,如
LLMChain、SequentialChain。 - Agents :智能代理。通过 LLM 决定调用哪些工具(Tools),并按照“思考-行动-观察”的循环执行。这是 LangChain 向复杂工作流迈出的第一步。
- LCEL (LangChain Expression Language) :新一代的、声明式的链构建方式。它用
|操作符连接组件,更 Pythonic,支持流式输出、异步、批量处理等高级特性。
2.2 典型使用模式与代码示例
让我们通过一个经典的 RAG 应用示例,来看 LangChain 如何工作。
场景 :基于本地文档的问答系统。
# 文件:basic_rag.py
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
# 1. 加载与处理文档 (Indexes)
loader = TextLoader("./state_of_the_union.txt")
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = text_splitter.split_documents(documents)
# 2. 创建向量存储 (Indexes)
vectorstore = Chroma.from_documents(
documents=splits,
embedding=OpenAIEmbeddings(model="text-embedding-3-small")
)
retriever = vectorstore.as_retriever()
# 3. 定义提示词 (Prompts)
template = """你是一个专业的问答助手。请根据以下上下文回答问题。
如果你不知道答案,就诚实地回答不知道,不要编造信息。
上下文:
{context}
问题:{question}
请用中文给出有帮助的答案:"""
prompt = ChatPromptTemplate.from_template(template)
# 4. 选择模型 (Models)
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 5. 使用 LCEL 构建链 (Chains)
rag_chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
)
# 6. 运行链
question = "总统在国情咨文中提到了哪些关于人工智能的政策?"
answer = rag_chain.invoke(question)
print(f"问题:{question}")
print(f"答案:{answer}")
代码解读 :
- 组件化 :每一步都使用了 LangChain 提供的标准组件(Loader, Splitter, Vectorstore, Model, Prompt)。
- LCEL 链 :使用
|操作符将检索器、提示词模板、模型和输出解析器连接成一个可执行的管道rag_chain。RunnablePassthrough()用于传递用户的问题。 - 线性流程 :这是一个典型的线性流程:输入问题 -> 检索上下文 -> 组装提示词 -> 调用模型 -> 解析输出。
2.3 LangChain 的优势与局限
优势 :
- 生态丰富 :拥有最大最全的集成库,几乎任何你能想到的工具或数据库都有对应的 Connector。
- 入门友好 :高级 API 封装良好,几行代码就能实现 RAG 或简单 Agent。
- 标准化 :统一的接口让切换底层模型或向量数据库成本极低。
局限 :
- 复杂流程编排能力弱 :对于需要循环、复杂条件分支、并行执行或精细状态管理的多步骤工作流,原生的
Chain或简单Agent会变得非常笨拙,代码可读性和可维护性下降。 - 状态管理隐式 :在链中传递状态通常依赖于将上一个节点的输出作为下一个节点的输入,对于需要长期维护和跨节点共享的复杂状态(如对话历史、中间决策结果)管理不便。
- 调试和可视化困难 :当链变得复杂时,理解和调试数据流经的路径比较困难。
这正是 LangGraph 要解决的问题。
3. LangGraph 深度解析:基于图的有状态工作流
3.1 核心概念:State, Node, Edge
LangGraph 引入了几个关键概念,将工作流建模为一个有向图:
- State (状态) :一个定义好的 Pydantic 模型或 TypedDict,描述了在整个工作流中共享和更新的所有数据。这是 LangGraph 的灵魂,它使得工作流有了“记忆”。
- Node (节点) :一个函数,它接收当前的
State,执行一些操作(如调用 LLM、运行工具),并返回一个包含对State更新内容的字典。 - Edge (边) :决定在某个节点执行完毕后,下一个应该执行哪个节点。分为:
- 条件边 (Conditional Edge) :根据
State中的某个值来决定流向。 - 固定边 (Fixed Edge) :始终流向指定的下一个节点。
- 条件边 (Conditional Edge) :根据
3.2 构建一个简单的 LangGraph 工作流
让我们构建一个比简单链更智能的“研究助手” Agent。它需要先决定是否需要联网搜索,如果需要,则搜索并总结;如果不需要,则直接回答。
# 文件:research_agent_graph.py
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import HumanMessage, SystemMessage
import operator
# 1. 定义状态结构
class AgentState(TypedDict):
question: str # 用户问题
needs_search: Annotated[str, operator.add] = "unknown" # 是否需要搜索
search_results: Annotated[str, operator.add] = "" # 搜索结果
final_answer: Annotated[str, operator.add] = "" # 最终答案
# Annotated 和 operator.add 用于指定如何更新该字段(这里是字符串拼接)
# 2. 初始化工具和模型
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
search_tool = TavilySearchResults(max_results=3)
# 3. 定义节点函数
def decide_search_node(state: AgentState) -> dict:
"""决策节点:判断是否需要联网搜索"""
system_msg = SystemMessage(content="你是一个研究助手。请判断以下问题是否需要通过联网搜索最新信息来回答。只需回答 'yes' 或 'no'。")
human_msg = HumanMessage(content=state["question"])
response = llm.invoke([system_msg, human_msg])
decision = response.content.strip().lower()
return {"needs_search": decision}
def search_node(state: AgentState) -> dict:
"""搜索节点:执行搜索并获取结果"""
if state["needs_search"] != "yes":
return {"search_results": "未执行搜索。"}
results = search_tool.invoke(state["question"])
# 简化处理,将结果拼接成字符串
result_str = "\n\n".join([f"来源:{r['title']}\n内容:{r['content']}" for r in results])
return {"search_results": f"搜索到的信息:\n{result_str}"}
def answer_node(state: AgentState) -> dict:
"""回答节点:综合所有信息生成最终答案"""
messages = [
SystemMessage(content="你是一个专业的研究助手。请根据用户的问题和已有的信息(如果有的话)给出准确、全面的回答。"),
HumanMessage(content=f"问题:{state['question']}\n\n相关信息:{state['search_results']}")
]
response = llm.invoke(messages)
return {"final_answer": response.content}
# 4. 构建图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("decide_search", decide_search_node)
workflow.add_node("search", search_node)
workflow.add_node("answer", answer_node)
# 设置入口点
workflow.set_entry_point("decide_search")
# 添加边(包括条件边)
workflow.add_conditional_edges(
"decide_search",
# 这是一个路由函数,根据状态决定下一个节点
lambda state: "search" if state["needs_search"] == "yes" else "answer",
{
"search": "search", # 如果返回 "search",则跳转到 search 节点
"answer": "answer" # 如果返回 "answer",则直接跳转到 answer 节点
}
)
workflow.add_edge("search", "answer") # 搜索完成后,固定流向回答节点
workflow.add_edge("answer", END) # 回答完成后,结束工作流
# 编译图
app = workflow.compile()
# 5. 运行图
initial_state = AgentState(question="2024年巴黎奥运会中国代表团获得了多少枚金牌?")
result = app.invoke(initial_state)
print(f"最终答案:\n{result['final_answer']}")
print("\n--- 工作流状态追踪 ---")
print(f"是否需要搜索:{result['needs_search']}")
代码解读 :
- 定义状态 :
AgentState明确了工作流中需要流转的所有数据。 - 节点即函数 :每个节点都是一个纯函数,接收状态,返回对状态的更新。逻辑清晰,易于单元测试。
- 条件路由 :
add_conditional_edges是关键,它让decide_search_node的输出决定了工作流是走搜索 -> 回答路径,还是直接跳转到回答路径。 - 可视化潜力 :由于工作流被明确定义为图,它可以被轻松可视化,便于理解和调试。
3.3 实现循环与持久化:ReAct Agent 示例
LangGraph 最强大的能力之一是轻松实现循环,比如经典的 ReAct(Reasoning + Acting)Agent。
# 文件:react_agent_graph.py
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
import operator
# 定义状态,主要维护消息历史
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add] # 消息列表
# 初始化
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools([TavilySearchResults(max_results=2)])
tool_map = {"tavily_search_results_json": TavilySearchResults(max_results=2)}
# 定义两个核心节点
def call_model(state: AgentState):
"""调用模型节点:让LLM根据历史决定是回复还是调用工具"""
response = llm.invoke(state["messages"])
return {"messages": [response]} # 将模型响应添加到历史
def call_tool(state: AgentState):
"""调用工具节点:执行模型选择的工具"""
last_message = state["messages"][-1]
tool_calls = last_message.tool_calls
tool_messages = []
for tool_call in tool_calls:
tool_name = tool_call["name"]
if tool_name not in tool_map:
raise ValueError(f"未知工具:{tool_name}")
tool = tool_map[tool_name]
try:
output = tool.invoke(tool_call["args"])
except Exception as e:
output = f"工具调用出错:{e}"
tool_messages.append(ToolMessage(content=str(output), tool_call_id=tool_call["id"]))
return {"messages": tool_messages}
# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)
workflow.set_entry_point("agent")
# 关键:条件边实现循环
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
"""根据最后一条消息判断下一步:调用工具还是结束"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools" # 有工具调用,去执行工具
return "__end__" # 没有工具调用,结束
workflow.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "__end__": END}
)
workflow.add_edge("tools", "agent") # 工具执行完后,回到agent节点继续思考
app = workflow.compile()
# 运行
initial_state = AgentState(messages=[HumanMessage(content="用中文告诉我,OpenAI 最新的多模态模型叫什么?有什么特点?")])
for event in app.stream(initial_state, stream_mode="values"):
event["messages"][-1].pretty_print()
这个 Agent 会持续在“思考(调用模型)”和“行动(调用工具)”之间循环,直到模型认为不需要再调用工具,给出最终答案。这种循环逻辑在纯 LangChain 中实现起来要复杂得多。
4. 对比表格:LangChain vs LangGraph 全景图
| 特性维度 | LangChain | LangGraph |
|---|---|---|
| 核心抽象 | 链 (Chain)、组件 (Component) | 图 (Graph)、节点 (Node)、状态 (State) |
| 编排范式 | 声明式管道(LCEL)、顺序/简单分支 | 有向图、支持循环、条件分支、并行 |
| 状态管理 | 隐式,通过输入/输出在链间传递 | 显式 ,通过强类型的 State 对象全局管理 |
| 适用场景 | 线性任务、RAG、简单对话、工具调用 | 复杂、多步骤工作流 、ReAct Agent、多智能体协作、需持久化状态的任务 |
| 代码复杂度 | 简单场景下极简,复杂流程下陡增 | 初期学习曲线稍高,但复杂流程下更清晰、易维护 |
| 可调试性 | 一般,需依赖日志 | 优秀 ,图结构易于可视化,状态变化可追踪 |
| 与 LangChain 关系 | 基础生态与组件库 | 基于 LangChain 组件,提供高级编排框架 |
| 心智模型 | “组装流水线” | “设计状态机” |
5. 如何选择?实战场景指南
面试时被问到区别,最终要落到“怎么用”上。你可以这样回答:
“在项目中,我的选择标准基于工作流的复杂性:”
-
选择 LangChain 当:
- 任务简单线性 :例如,标准的 RAG 流程(加载-分割-嵌入-检索-生成)。
- 快速原型验证 :需要利用其丰富的集成快速搭出一个可用的 Demo。
- 主要使用其生态组件 :比如只需要它的
TextSplitter、Vectorstore或某个特定的工具集成,编排用更轻量的方式(如直接函数调用)。
-
选择 LangGraph 当:
- 工作流包含循环或复杂条件逻辑 :例如,一个需要反复尝试、自我修正的代码生成 Agent,或者一个根据用户反馈调整策略的对话系统。
- 需要精细管理长期状态 :例如,一个游戏 NPC Agent,需要记住玩家的偏好、之前的对话和任务完成状态。
- 构建多智能体系统 :多个 Agent 需要协作、竞争或顺序执行,它们之间的交互和状态传递用图来建模非常直观。
- 追求生产环境的健壮性和可维护性 :图的结构化定义使得代码更模块化,状态流更清晰,便于团队协作和后期扩展。
一个更具体的例子:客服工单处理系统
- LangChain 部分 :负责单个技能。例如,用
LLMChain实现“分类工单”,用RetrievalQA实现“从知识库查找解决方案”。 - LangGraph 部分 :负责整体流程编排。定义状态(工单ID、用户问题、当前处理阶段、已尝试方案、客服备注等),设计节点(分类节点、检索节点、转人工判断节点、回复生成节点、归档节点),并通过边来控制流程(如果知识库解决则归档,否则判断是否转人工)。
6. 常见问题与排查思路
在实际使用中,你会遇到一些典型问题。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
LangChain: LCEL 链调用报错 KeyError |
链中 RunnablePassthrough 或字典输入输出键名不匹配。 |
检查链的输入字典结构,确保每个节点所需的键都能在上游节点的输出中找到。 | 使用 RunnableParallel 显式组合多个输入源,或在节点函数中做好键名映射。 |
| LangChain: 检索结果不相关 | 文本分割块大小不合适,或嵌入模型不匹配,或检索器配置问题。 | 1. 检查分割后的文本块是否完整。 2. 确认索引和查询时使用的嵌入模型是否一致。 3. 调整 retriever 的 search_kwargs (如 k 值)。 |
优化分割策略(按语义或章节),确保嵌入模型一致,尝试混合检索(MMR)以提高多样性。 |
| LangGraph: 状态更新不符合预期 | State 中字段的更新操作符(如 operator.add )使用错误,或节点返回值格式不对。 |
打印每个节点执行前后的状态,检查 Annotated 注解是否正确。 |
确保节点函数返回的字典键与 State 字段名对应,理解 add (拼接)、 replace (替换)等操作符的语义。 |
| LangGraph: 图陷入无限循环 | 条件边的逻辑有误,导致无法满足结束条件。 | 在 should_continue 或条件边函数中添加详细的日志,检查状态变量的变化。 |
设置最大循环次数( interrupt_before / after ),或在状态中增加 steps 计数器并在条件中判断。 |
| 通用: 工具调用失败 | API 密钥未设置、网络问题、工具参数格式错误。 | 首先在 LangChain/LangGraph 环境外单独测试工具调用。 | 检查环境变量,验证工具初始化参数,查看工具的官方文档和错误信息。 |
| 通用: LLM 响应慢或超时 | 模型本身延迟高、提示词过于复杂、网络延迟。 | 使用流式输出先获取部分结果,或在本地测试小提示词。 | 优化提示词,考虑使用更快的模型(如 gpt-4o-mini ),设置合理的超时参数,启用缓存。 |
7. 最佳实践与工程化建议
无论是使用 LangChain 还是 LangGraph,遵循一些最佳实践能让你的项目更稳健。
-
环境与依赖管理
- 使用虚拟环境(
venv,conda)。 - 使用
requirements.txt或pyproject.toml严格锁定核心包版本,避免因 LangChain 生态快速迭代导致的意外破坏。 - 将 API 密钥等敏感信息存储在环境变量或安全的配置管理中,切勿硬编码。
- 使用虚拟环境(
-
提示词工程
- 与代码分离 :将提示词模板放在单独的配置文件(如 YAML、JSON)或数据库中,便于管理和 A/B 测试。
- 结构化输出 :优先要求 LLM 返回 JSON 等结构化数据,并使用
PydanticOutputParser或JsonOutputParser进行解析和验证,提高下游代码的可靠性。 - 少样本示例 :在提示词中包含高质量的示例,能显著提升复杂任务的性能。
-
应用架构
- 分层设计 :将数据层(向量库)、业务逻辑层(链/图编排)、表示层(API/UI)分离。
- 为 LangGraph 设计清晰的状态 :花时间精心设计
State结构,它是工作流的“单一数据源”。避免在节点间通过全局变量等隐式方式传递信息。 - 节点职责单一 :每个节点函数应只做一件事,并做好错误处理。这有利于测试和复用。
-
可观测性与监控
- 启用日志 :利用 LangChain/LangGraph 的回调(
callbacks)系统,记录关键步骤、耗时和 Token 使用情况。 - 追踪与可视化 :LangSmith 是官方提供的绝佳平台,可以可视化链和图的执行过程,调试问题事半功倍。对于生产系统,需将关键指标(延迟、费用、错误率)接入现有监控体系。
- 为 LangGraph 添加检查点 :利用
checkpointer实现工作流的持久化,允许中断后从中间状态恢复,这对长时运行的任务至关重要。
- 启用日志 :利用 LangChain/LangGraph 的回调(
-
性能与成本
- 缓存 :对 Embedding 模型、LLM 的相同请求使用缓存(如
InMemoryCache,SQLiteCache),节省成本和时间。 - 异步优化 :对于 I/O 密集型操作(如批量调用 LLM、并发检索),使用异步接口(
ainvoke,abatch)提升吞吐。 - 精简上下文 :在 RAG 中,精心设计检索策略,只返回最相关的片段,避免因上下文过长导致的高成本和低性能。
- 缓存 :对 Embedding 模型、LLM 的相同请求使用缓存(如
回到最初的面试问题,“LangChain 和 LangGraph 有什么区别?” 你现在可以给出一个层次丰富的答案: LangChain 是构建大模型应用的“标准零件库”和“装配线”,它降低了连接各种组件的门槛;而 LangGraph 是设计复杂智能体工作流的“流程图”和“状态机”,它解决了超越线性流程的、需要持久化状态和复杂逻辑编排的挑战。 在技术选型上,它们不是二选一,而是相辅相成。LangChain 的丰富生态是地基,而当你需要建造多层、多房间、带有自动循环系统(复杂工作流)的建筑时,LangGraph 提供了更强大的蓝图和施工框架。
对于学习者,建议从 LangChain 入手,掌握 LCEL 和核心组件的使用,建立起对大模型应用开发的基本感觉。当你开始设计需要多次工具调用、循环推理或多角色协作的项目时,就是深入 LangGraph 的最佳时机。理解两者的差异和结合点,你就能在快速演进的大模型应用开发领域中,更好地架构和实现你的想法。
更多推荐




所有评论(0)