写在前面
2024年以来,LangGraph 作为 LangChain 生态的重要扩展,逐渐成为构建复杂 Agent 工作流的首选工具。然而,社区中存在一个普遍的误解:把 LangGraph 视为 LangChain 的"升级版"或"替代者"。这种认知是错误的——两者本质上是不同层次的设计:LangChain 是库(Library),LangGraph 是框架(Framework)。本文将深入剖析两者的定位差异、适用场景,并提供完整的代码对比,帮助你在实际项目中做出正确的技术选型。
一、问题背景:为什么需要澄清这个误解?
1.1 社区中的常见困惑
当你准备构建一个 LLM 应用时,你可能会遇到这样的困惑:
我在用 LangChain,但听说 LangGraph 更强大,应该切换吗?
LangChain 和 LangGraph 哪个更好?
它们有什么区别?我该用哪个?这些问题反映了一个深层次的认知混乱。让我用一个类比来解释:
LangChain 就像是一套完整的 工具箱,里面有锤子、螺丝刀、扳手、电钻……每种工具都有明确的用途。
LangGraph 就像是一套 工作台系统,它提供了工作台、夹具、导轨,让你把工具组合起来完成复杂的制作任务。
你不会问"我应该用锤子还是工作台?"——它们解决的是不同层次的问题。LangChain vs LangGraph 也是如此。
1.2 技术演进的背景
LangChain 在 2023 年初发布,最初的定位是简化 LLM 应用的开发——通过 Chain、Prompt Template、Memory 等抽象,让开发者快速搭建基于 LLM 的应用。
随着应用场景的复杂度提升,开发者开始遇到以下问题:
多轮对话状态管理:如何在多个 LLM 调用之间保持上下文?
循环与分支逻辑:如何实现"如果...那么..."的条件判断?
Agent 的可控性:如何让 Agent 的行为更加可预测、可调试?
这些问题促使了 LangGraph 的诞生——它专注于解决有状态(Stateful)工作流的问题,与 LangChain 形成互补。
二、LangChain 定位分析:LLM 应用的工具箱
2.1 LangChain 是什么?
LangChain 是一个用于构建 LLM 应用的模块化库。它的核心设计理念是:将 LLM 应用的各种能力分解为独立的、可组合的组件。
2.2 核心组件一览
┌─────────────────────────────────────────────────────────────┐
│ LangChain 生态 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Prompts │ │ Models │ │ Output │ │
│ │ 提示词管理 │ │ 模型接口 │ │ 解析器 │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Chains │ │ Tools │ │ Memory │ │
│ │ 链式调用 │ │ 工具 │ │ 记忆 │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Indexes │ │ Agents │ │ Callbacks │ │
│ │ 索引 │ │ 代理 │ │ 回调系统 │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘2.3 LangChain 的设计哲学
LangChain 的设计哲学可以概括为三个关键词:组合、抽象、简洁。
组合(Composition):每个组件都是独立的,可以自由组合
抽象(Abstraction):通过统一的接口隐藏底层复杂度
简洁(Simplicity):用最少的代码完成最多的工作
2.4 典型使用场景
LangChain 非常适合以下场景:
2.5 代码示例:使用 LangChain 构建 RAG 应用
下面是一个典型的 LangChain RAG 应用示例:
"""
LangChain RAG 应用示例
使用 LangChain 实现一个简单的文档问答系统
"""
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.text_splitter import CharacterTextSplitter
# === 配置 ===
# 注意:在生产环境中,请使用环境变量管理 API Key
OPENAI_API_KEY = "your-api-key" # 替换为你的 API Key
DOCUMENT_PATH = "documents/manual.txt" # 替换为你的文档路径
# === 1. 加载文档 ===
def load_documents(path: str) -> list:
"""
加载文本文档并分割成块
Args:
path: 文档路径
Returns:
分割后的文档块列表
"""
loader = TextLoader(path, encoding="utf-8")
documents = loader.load()
# 使用字符分割器将文档分割成小块
text_splitter = CharacterTextSplitter(
chunk_size=1000, # 每块的最大字符数
chunk_overlap=200, # 块之间的重叠字符数
separator="\n\n" # 分割符
)
texts = text_splitter.split_documents(documents)
print(f"✅ 已加载并分割 {len(texts)} 个文档块")
return texts
# === 2. 创建向量存储 ===
def create_vectorstore(documents: list):
"""
使用 OpenAI Embeddings 创建向量存储
Args:
documents: 文档块列表
Returns:
向量存储对象
"""
embeddings = OpenAIEmbeddings(
openai_api_key=OPENAI_API_KEY
)
# 创建 FAISS 向量数据库
vectorstore = FAISS.from_documents(
documents=documents,
embedding=embeddings
)
print("✅ 向量存储创建成功")
return vectorstore
# === 3. 构建问答链 ===
def build_qa_chain(vectorstore):
"""
构建检索增强问答链
Args:
vectorstore: 向量存储对象
Returns:
问答链对象
"""
# 初始化 LLM
llm = ChatOpenAI(
model="gpt-4",
openai_api_key=OPENAI_API_KEY,
temperature=0.7 # 控制回答的随机性
)
# 创建检索器
retriever = vectorstore.as_retriever(
search_kwargs={"k": 3} # 返回最相关的 3 个文档块
)
# 构建 RetrievalQA 链
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff", # 将所有文档块拼接到一起
retriever=retriever,
return_source_documents=True # 返回源文档用于验证
)
print("✅ 问答链构建成功")
return qa_chain
# === 4. 主函数 ===
def main():
"""
主函数:加载文档、构建链、运行问答
"""
# 步骤1:加载文档
documents = load_documents(DOCUMENT_PATH)
# 步骤2:创建向量存储
vectorstore = create_vectorstore(documents)
# 步骤3:构建问答链
qa_chain = build_qa_chain(vectorstore)
# 步骤4:运行问答
print("\n" + "=" * 50)
print("💬 文档问答系统已准备就绪")
print("=" * 50)
while True:
query = input("\n请输入问题(输入 'exit' 退出):")
if query.lower() in ["exit", "quit", "q"]:
print("👋 再见!")
break
if not query.strip():
print("⚠️ 请输入有效的问题")
continue
# 执行问答
result = qa_chain({"query": query})
print("\n📝 回答:")
print(result["result"])
# 显示参考来源
print("\n📚 参考来源:")
for i, doc in enumerate(result["source_documents"], 1):
print(f" [{i}] {doc.page_content[:200]}...")
if __name__ == "__main__":
main()2.6 LangChain 的局限性
尽管 LangChain 功能强大,但它在以下场景中存在局限性:
复杂的循环逻辑:Chain 本质上是线性的,难以实现"如果失败则重试"的逻辑
细粒度状态管理:多个组件之间的状态传递不够透明
工作流可视化:难以直观地展示复杂的工作流程
这些问题正是 LangGraph 试图解决的。
三、LangGraph 定位分析:有状态 Agent 的编排框架
3.1 LangGraph 是什么?
LangGraph 是一个用于构建有状态、多步骤 Agent 工作流的框架。它的核心设计理念是:将 LLM 应用建模为一张图(Graph),图中的节点是操作,边是状态流转。
3.2 核心概念
┌─────────────────────────────────────────────────────────────┐
│ LangGraph 核心概念 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ │ State │ ◀── 整个应用的"全局上下文",存储所有数据 │
│ └────┬────┘ │
│ │ │
│ ▼ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Node │ ──▶ │ Node │ ──▶ │ Node │ │
│ │ 节点A │ │ 节点B │ │ 节点C │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │ │ │ │
│ └──────────────┴──────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────┐ │
│ │ Edge │ ◀── 定义节点间的流转规则 │
│ │ 边 │ │
│ └─────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘3.3 LangGraph vs LangChain 对比
3.4 代码示例:使用 LangGraph 构建研究 Agent
下面是一个使用 LangGraph 构建的多步骤研究 Agent:
"""
LangGraph 研究 Agent 示例
实现一个"搜索 → 分析 → 总结 → 审核"的工作流
"""
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
import os
# === 配置 ===
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your-api-key")
# === 1. 定义状态结构 ===
class ResearchState(TypedDict):
"""
研究 Agent 的状态结构
使用 Annotated 定义 messages 字段,add_messages 负责
自动合并新旧消息,实现消息历史的管理
"""
messages: Annotated[list, add_messages] # 对话历史
query: str # 研究主题
search_results: list # 搜索结果
analysis: str # 分析结果
report: str # 最终报告
quality_score: float # 质量评分
revision_count: int # 修改次数
# === 2. 初始化 LLM ===
llm = ChatOpenAI(
model="gpt-4",
openai_api_key=OPENAI_API_KEY,
temperature=0.7
)
# === 3. 定义节点函数 ===
def search_node(state: ResearchState) -> dict:
"""
搜索节点:执行信息检索
Args:
state: 当前状态
Returns:
状态更新字典
"""
query = state["query"]
prompt = f"""
请为以下研究主题搜索相关信息,列出 5 个关键点:
主题:{query}
请用结构化的方式输出搜索结果。
"""
response = llm.invoke(prompt)
return {
"search_results": [response.content],
"messages": [response]
}
def analyze_node(state: ResearchState) -> dict:
"""
分析节点:对搜索结果进行深度分析
Args:
state: 当前状态
Returns:
状态更新字典
"""
query = state["query"]
search_results = state.get("search_results", [""])[-1]
prompt = f"""
基于以下搜索结果,对研究主题进行深度分析:
主题:{query}
搜索结果:{search_results}
请分析:
1. 主要观点和发现
2. 关键趋势和模式
3. 潜在的争议点或不确定性
"""
response = llm.invoke(prompt)
return {
"analysis": response.content,
"messages": [response]
}
def write_report_node(state: ResearchState) -> dict:
"""
报告节点:撰写最终研究报告
Args:
state: 当前状态
Returns:
状态更新字典
"""
query = state["query"]
analysis = state.get("analysis", "")
prompt = f"""
基于以下分析,撰写一份完整的研究报告:
主题:{query}
分析结果:{analysis}
报告要求:
- 结构清晰,分为引言、主体、结论
- 包含具体的数据和例子
- 字数控制在 500-800 字
"""
response = llm.invoke(prompt)
return {
"report": response.content,
"messages": [response]
}
def review_node(state: ResearchState) -> dict:
"""
审核节点:评估报告质量,决定是否需要修改
Args:
state: 当前状态
Returns:
状态更新字典
"""
report = state.get("report", "")
revision_count = state.get("revision_count", 0)
prompt = f"""
请评估以下研究报告的质量:
报告内容:
{report}
评估标准:
1. 内容完整性(0-10分)
2. 逻辑清晰度(0-10分)
3. 数据支撑(0-10分)
4. 语言表达(0-10分)
最终评分(平均分):
是否需要修改?(评分低于7分需要修改)
"""
response = llm.invoke(prompt)
# 简单解析评分(实际项目中应该使用更严格的解析逻辑)
try:
# 尝试从响应中提取数字评分
lines = response.content.split('\n')
for line in lines:
if '最终评分' in line or '平均分' in line:
# 提取第一个数字
for char in line:
if char.isdigit():
score = float(char)
break
else:
score = 7.0
break
else:
score = 7.0
except:
score = 7.0
return {
"quality_score": score,
"messages": [response]
}
def route_after_review(state: ResearchState) -> str:
"""
路由函数:根据审核结果决定下一步
Args:
state: 当前状态
Returns:
下一个节点的名称
"""
score = state.get("quality_score", 0)
revision_count = state.get("revision_count", 0)
# 如果评分低于 7 分且修改次数未超过 3 次,则重新生成
if score < 7.0 and revision_count < 3:
return "revise_report"
else:
return END
def revise_report_node(state: ResearchState) -> dict:
"""
修改节点:根据反馈修改报告
Args:
state: 当前状态
Returns:
状态更新字典
"""
current_report = state.get("report", "")
messages = state["messages"]
# 获取之前的审核反馈
feedback = ""
for msg in reversed(messages):
if hasattr(msg, 'content') and '评估' in msg.content:
feedback = msg.content
break
prompt = f"""
请根据以下反馈修改研究报告:
当前报告:
{current_report}
审核反馈:
{feedback}
请在保持报告整体结构的同时,改进被指出不足的部分。
"""
response = llm.invoke(prompt)
new_revision_count = revision_count + 1
return {
"report": response.content,
"revision_count": new_revision_count,
"messages": [response]
}
# === 4. 构建工作流图 ===
def build_research_graph():
"""
构建研究 Agent 工作流图
工作流设计:
START -> search -> analyze -> write_report -> review -> [END|revise_report]
↑ │
└──────────────┘
"""
# 创建状态图
workflow = StateGraph(ResearchState)
# 注册所有节点
workflow.add_node("search", search_node)
workflow.add_node("analyze", analyze_node)
workflow.add_node("write_report", write_report_node)
workflow.add_node("review", review_node)
workflow.add_node("revise_report", revise_report_node)
# 定义静态边(固定流程)
workflow.add_edge(START, "search")
workflow.add_edge("search", "analyze")
workflow.add_edge("analyze", "write_report")
workflow.add_edge("write_report", "review")
# 定义条件边(动态决策)
workflow.add_conditional_edges(
source="review",
path=route_after_review,
path_map={
"revise_report": "revise_report",
END: END
}
)
# 修改后回到审核节点
workflow.add_edge("revise_report", "review")
# 编译图
return workflow.compile()
# === 5. 主函数 ===
def main():
"""
主函数:运行研究 Agent
"""
print("🔬 研究 Agent 已启动")
print("=" * 50)
# 构建工作流
app = build_research_graph()
# 定义初始状态
initial_state = {
"messages": [],
"query": "人工智能对就业市场的影响",
"search_results": [],
"analysis": "",
"report": "",
"quality_score": 0.0,
"revision_count": 0
}
# 执行工作流
print("\n📊 开始研究...\n")
result = app.invoke(initial_state)
# 输出结果
print("\n" + "=" * 50)
print("📄 最终研究报告")
print("=" * 50)
print(result["report"])
print(f"\n📈 质量评分: {result['quality_score']}/10")
print(f"🔄 修改次数: {result['revision_count']}")
if __name__ == "__main__":
main()3.5 LangGraph 的适用场景
LangGraph 特别适合以下场景:
四、架构对比:同一个需求的不同实现
4.1 需求描述
需求:构建一个"先搜索相关信息,再生成回答"的简单问答系统。
4.2 使用 LangChain 实现
"""
LangChain 实现:搜索增强问答
"""
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate
# === 配置 ===
OPENAI_API_KEY = "your-api-key"
# === 初始化组件 ===
llm = ChatOpenAI(model="gpt-4", openai_api_key=OPENAI_API_KEY)
search_tool = DuckDuckGoSearchRun()
# === 定义提示词模板 ===
search_prompt = PromptTemplate(
input_variables=["question"],
template="请为以下问题搜索相关信息,输出3个关键点:\n问题:{question}"
)
answer_prompt = PromptTemplate(
input_variables=["search_result", "question"],
template="""基于以下搜索结果回答问题:
搜索结果:
{search_result}
问题:{question}
请给出清晰、准确的回答。"""
)
# === 构建链 ===
search_chain = LLMChain(llm=llm, prompt=search_prompt, output_key="search_result")
answer_chain = LLMChain(llm=llm, prompt=answer_prompt, output_key="answer")
# === 组合链 ===
overall_chain = SequentialChain(
chains=[search_chain, answer_chain],
input_variables=["question"],
output_variables=["search_result", "answer"],
verbose=True
)
# === 执行 ===
result = overall_chain({"question": "什么是大语言模型?"})
print(result["answer"])4.3 使用 LangGraph 实现
"""
LangGraph 实现:搜索增强问答
"""
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
# === 配置 ===
OPENAI_API_KEY = "your-api-key"
# === 定义状态 ===
class QaState(TypedDict):
messages: Annotated[list, add_messages]
question: str
search_result: str
answer: str
# === 初始化组件 ===
llm = ChatOpenAI(model="gpt-4", openai_api_key=OPENAI_API_KEY)
search_tool = DuckDuckGoSearchRun()
# === 定义节点 ===
def search_node(state: QaState) -> dict:
"""搜索节点"""
question = state["question"]
# 使用搜索工具
result = search_tool.run(f"搜索:{question}")
return {"search_result": result}
def answer_node(state: QaState) -> dict:
"""回答节点"""
question = state["question"]
search_result = state["search_result"]
prompt = f"""基于以下搜索结果回答问题:
搜索结果:
{search_result}
问题:{question}
请给出清晰、准确的回答。"""
response = llm.invoke(prompt)
return {"answer": response.content, "messages": [response]}
# === 构建图 ===
workflow = StateGraph(QaState)
workflow.add_node("search", search_node)
workflow.add_node("answer", answer_node)
workflow.add_edge(START, "search")
workflow.add_edge("search", "answer")
workflow.add_edge("answer", END)
app = workflow.compile()
# === 执行 ===
result = app.invoke({
"messages": [],
"question": "什么是大语言模型?",
"search_result": "",
"answer": ""
})
print(result["answer"])4.4 对比分析
五、从 LangChain 迁移到 LangGraph 的典型路径
5.1 迁移决策树
┌─────────────────────┐
│ 当前使用 LangChain │
└──────────┬──────────┘
│
▼
┌─────────────────────┐
│ 需求复杂度如何? │
└──────────┬──────────┘
│
┌──────────────────┼──────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 简单任务 │ │ 中等复杂度 │ │ 高度复杂 │
│ (RAG、问答) │ │ (多步骤流程) │ │ (循环、协作) │
└─────────────┘ └─────────────┘ └─────────────┘
│ │ │
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 继续使用 │ │ 考虑迁移 │ │ 必须迁移 │
│ LangChain │ │ LangGraph │ │ LangGraph │
└─────────────┘ └─────────────┘ └─────────────┘5.2 迁移策略
策略一:渐进式迁移
不要一次性重写整个系统,而是逐步将关键流程迁移到 LangGraph。
"""
渐进式迁移示例:在 LangChain 基础上引入 LangGraph
"""
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
# === 保留现有的 LangChain 组件 ===
# 假设我们已经有了一个 RAG 链
llm = ChatOpenAI(model="gpt-4", api_key="your-key")
rag_chain = RetrievalQA.from_chain_type(llm=llm, retriever=your_retriever)
# === 定义 LangGraph 状态 ===
class EnhancedState(TypedDict):
query: str
rag_result: str
final_answer: str
# === 将 LangChain 链封装为 LangGraph 节点 ===
def rag_node(state: EnhancedState) -> dict:
"""使用现有的 LangChain RAG 链"""
result = rag_chain.invoke({"query": state["query"]})
return {"rag_result": result["result"]}
def enhance_node(state: EnhancedState) -> dict:
"""增强回答节点"""
prompt = f"增强以下回答,添加更多细节:\n{state['rag_result']}"
enhanced = llm.invoke(prompt)
return {"final_answer": enhanced.content}
# === 构建 LangGraph 工作流 ===
workflow = StateGraph(EnhancedState)
workflow.add_node("rag", rag_node)
workflow.add_node("enhance", enhance_node)
workflow.add_edge(START, "rag")
workflow.add_edge("rag", "enhance")
workflow.add_edge("enhance", END)
app = workflow.compile()
# 现在你可以同时使用原有的 rag_chain 和新的 app策略二:混合使用
LangChain 和 LangGraph 可以共存,发挥各自的优势。
"""
混合使用示例:LangChain 提供工具能力,LangGraph 提供编排能力
"""
from langgraph.graph import StateGraph, START
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
# === LangChain 提供工具 ===
search_tool = DuckDuckGoSearchRun()
wiki_tool = WikipediaQueryRun()
# === LangGraph 编排工作流 ===
tool_node = ToolNode([search_tool, wiki_tool])
workflow = StateGraph(...)
# ... 省略定义
# 完美结合:LangChain 的工具 + LangGraph 的编排5.3 迁移检查清单
在迁移过程中,确保完成以下检查:
状态设计:明确定义 State 结构,列出所有需要的字段
节点拆分:将现有逻辑拆分为独立的节点函数
边定义:明确节点之间的流转关系(静态/条件)
异常处理:定义失败场景的处理逻辑
测试覆盖:为每个节点编写单元测试
性能验证:对比迁移前后的响应时间和资源消耗
六、总结与建议
6.1 核心结论
经过以上分析,我们可以得出以下结论:
LangChain 和 LangGraph 不是替代关系,而是互补关系
LangChain 专注于组件能力(工具、链、内存)
LangGraph 专注于流程编排(状态、节点、边)
技术选型的关键在于问题复杂度
简单任务:优先使用 LangChain,开发效率更高
复杂工作流:使用 LangGraph,掌控力更强
最佳实践是混合使用
在 LangGraph 工作流中调用 LangChain 组件
让每个工具各司其职
6.2 技术选型建议
┌─────────────────────────────────────────────────────────────┐
│ 技术选型决策指南 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 你的需求是... 推荐方案 │
│ ───────────────────────────────────────────────────────── │
│ │
│ 简单的 RAG 问答 LangChain │
│ 带记忆的对话 LangChain + Memory │
│ 多步骤数据分析 LangGraph │
│ 需要循环/重试的工作流 LangGraph │
│ 人机协作流程 LangGraph │
│ 复杂的多 Agent 协作 LangGraph │
│ 需要精细控制的状态管理 LangGraph │
│ │
│ 不确定?先从 LangChain 开始,根据需要迁移 │
│ │
└─────────────────────────────────────────────────────────────┘6.3 写在最后
技术选型没有绝对的对错,只有适合与不适合。LangChain 和 LangGraph 都是优秀的工具,关键在于理解它们的设计哲学和使用场景。
我的建议是:先掌握 LangChain 的基本用法,理解 LLM 应用的各种组件能力;再学习 LangGraph 的核心概念,学会用"图"的思维来设计工作流。当你能够自如地在两者之间切换时,你就已经成为了真正的 LLM 应用架构师。
评论区