LangChain vs LangGraph:完整对比指南
两个流行的 LLM 应用框架的全面对比,帮助开发者为项目做出明智的选择。
概述
什么是 LangChain?
LangChain 是一个开源框架,旨在简化由大语言模型(LLM)驱动的应用程序开发。它为链提供了标准化接口, 与各种工具集成,并为常见用例提供端到端的链。
什么是 LangGraph?
LangGraph 是构建在 LangChain 之上的库,支持创建有状态的、多角色的 LLM 应用程序。 它通过基于图的工作流扩展了 LangChain,使构建复杂的、循环的应用程序变得更加容易。
为什么要对比它们?
虽然两个框架都是为 LLM 应用开发设计的,但它们服务于不同的需求,在不同的场景中表现出色。 了解它们的差异有助于您为特定用例选择正确的工具。
核心概念与设计哲学
LangChain:链式思维
LangChain 遵循线性的、基于链的方法,操作按顺序依次执行。它围绕"链"的概念构建—— 即可以链接在一起创建复杂工作流的操作序列。
核心概念:
- Chains(链):链接在一起的顺序操作
- Prompts(提示):基于模板的输入格式化
- LLMs:语言模型集成
- Memory(记忆):跨调用的状态持久化
- Agents(代理):动态决策组件
LangGraph:基于图的状态思维
LangGraph 引入了基于图的方法,节点表示操作,边定义它们之间的流程。 这使得更复杂的、循环的工作流成为可能,并内置状态管理。
核心概念:
- Nodes(节点):单个操作或函数
- Edges(边):节点之间的连接
- State(状态):图中共享的数据
- Conditional Edges(条件边):基于条件的动态路由
- Persistence(持久化):内置状态检查点
对比表格
| 特性 | LangChain | LangGraph |
|---|---|---|
| 学习曲线 | 对初学者更友好 | 较陡峭但更一致 |
| 状态管理 | 记忆类 | 内置状态模式 |
| 工作流复杂度 | 适合线性链 | 擅长复杂图结构 |
| 循环与迭代 | 有限支持 | 完全支持 |
| 集成能力 | 100+ 集成 | LangChain 集成 |
| 调试能力 | 回调 + LangSmith | 状态快照 + LangSmith |
| 生产部署 | LangServe | 兼容 LangServe |
| 多智能体系统 | 可实现但复杂 | 原生支持 |
| 文档完善度 | 丰富 | 持续增长中 |
学习曲线
LangChain 学习路径
LangChain 对初学者来说学习曲线相对平缓。基于链的概念很直观,文档提供了许多示例。 然而,随着应用程序变得越来越复杂,管理状态和流程可能变得具有挑战性。
初学者:2-3 天 - 基本链和提示
中级:1-2 周 - 代理、工具和记忆
高级:2-4 周 - 自定义链和复杂集成
LangGraph 学习路径
LangGraph 由于其基于图的范式,初始学习曲线较陡。然而,一旦理解,它为复杂应用程序提供了更一致的心智模型。
初学者:3-5 天 - 基本图和状态
中级:1-2 周 - 条件路由和持久化
高级:2-3 周 - 复杂的多代理系统
代码对比
Level 1:简单问答链
让我们从最简单的用例开始:基本的问答链。
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
# 定义提示模板
prompt = ChatPromptTemplate.from_template(
"回答以下问题:{question}"
)
# 创建模型
model = ChatOpenAI(model="gpt-4")
# 构建链
chain = prompt | model | StrOutputParser()
# 调用链
result = chain.invoke({"question": "什么是 LangChain?"})
print(result) from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict
# 定义状态
class State(TypedDict):
question: str
answer: str
# 定义节点
def generate_answer(state: State) -> State:
model = ChatOpenAI(model="gpt-4")
response = model.invoke(state["question"])
return {"answer": response.content}
# 构建图
graph = StateGraph(State)
graph.add_node("generate", generate_answer)
graph.set_entry_point("generate")
graph.add_edge("generate", END)
# 编译并运行
app = graph.compile()
result = app.invoke({"question": "什么是 LangGraph?"})
print(result["answer"]) Level 2:多轮对话
添加对话历史以在多轮交互中保持上下文。
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, AIMessage
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
# 创建模型和记忆
model = ChatOpenAI(model="gpt-4")
memory = ConversationBufferMemory(return_messages=True)
# 创建对话链
chain = ConversationChain(llm=model, memory=memory)
# 多轮对话
response1 = chain.predict(input="你好,我是小明")
response2 = chain.predict(input="我叫什么名字?")
print(response2) # 会记住"小明" from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, List
class State(TypedDict):
messages: List
def chat(state: State) -> State:
model = ChatOpenAI(model="gpt-4")
response = model.invoke(state["messages"])
return {"messages": state["messages"] + [response]}
# 构建图
graph = StateGraph(State)
graph.add_node("chat", chat)
graph.set_entry_point("chat")
graph.add_edge("chat", END)
app = graph.compile()
# 带状态的多轮对话
result1 = app.invoke({
"messages": [HumanMessage("你好,我是小明")]
})
result2 = app.invoke({
"messages": result1["messages"] + [HumanMessage("我叫什么名字?")]
}) Level 3:RAG(检索增强生成)
实现带有文档检索和生成的 RAG 管道。
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain_community.document_loaders import TextLoader
# 加载并分割文档
loader = TextLoader("docs.txt")
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
texts = text_splitter.split_documents(documents)
# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(texts, embeddings)
# 创建 RAG 链
llm = ChatOpenAI(model="gpt-4")
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
retriever=vectorstore.as_retriever()
)
# 查询
result = qa_chain.invoke({"query": "文档是关于什么的?"}) from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from typing import TypedDict, List
class RAGState(TypedDict):
query: str
documents: List[str]
answer: str
def retrieve(state: RAGState) -> RAGState:
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(persist_directory="./chroma", embedding_function=embeddings)
docs = vectorstore.similarity_search(state["query"], k=3)
return {"documents": [d.page_content for d in docs]}
def generate(state: RAGState) -> RAGState:
llm = ChatOpenAI(model="gpt-4")
context = "\n\n".join(state["documents"])
prompt = f"上下文:{context}\n\n问题:{state['query']}"
response = llm.invoke(prompt)
return {"answer": response.content}
# 构建图
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)
app = graph.compile()
result = app.invoke({"query": "文档是关于什么的?"}) Level 4:工具调用与函数绑定
让 LLM 能够使用外部工具和函数。
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
# 定义自定义工具
@tool
def get_weather(city: str) -> str:
"""获取城市天气"""
return f"{city}天气:晴朗,22°C"
@tool
def calculate(expression: str) -> str:
"""计算数学表达式"""
return str(eval(expression))
# 创建工具列表
tools = [get_weather, calculate]
# 创建代理
llm = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有用的助手。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)
# 执行
result = agent_executor.invoke({"input": "北京天气怎么样?"}) from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from typing import TypedDict, List, Any
# 定义工具
@tool
def get_weather(city: str) -> str:
"""获取城市天气"""
return f"{city}天气:晴朗,22°C"
@tool
def calculate(expression: str) -> str:
"""计算数学表达式"""
return str(eval(expression))
tools = [get_weather, calculate]
# 定义状态
class AgentState(TypedDict):
messages: List[Any]
# 创建绑定工具的模型
llm = ChatOpenAI(model="gpt-4")
llm_with_tools = llm.bind_tools(tools)
# 定义代理节点
def agent(state: AgentState) -> AgentState:
response = llm_with_tools.invoke(state["messages"])
return {"messages": state["messages"] + [response]}
# 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))
graph.set_entry_point("agent")
# 条件路由
def should_continue(state):
if state["messages"][-1].tool_calls:
return "tools"
return END
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")
app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "北京天气怎么样?"}]}) Level 5:多代理协作
构建多个代理协作解决复杂问题的系统。
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import Tool
from langchain_core.prompts import ChatPromptTemplate
# 创建专业代理
def create_agent(role: str, expertise: str):
llm = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
("system", f"你是{role}。{expertise}"),
("human", "{input}"),
])
return prompt | llm
# 创建代理
researcher = create_agent("研究员", "你负责收集信息。")
analyst = create_agent("分析师", "你负责分析数据。")
writer = create_agent("撰写者", "你负责撰写报告。")
# 顺序执行(简化版)
topic = "2024年AI趋势"
research = researcher.invoke({"input": f"研究{topic}"})
analysis = analyst.invoke({"input": f"分析:{research.content}"})
report = writer.invoke({"input": f"基于分析撰写报告:{analysis.content}"}) from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, List, Optional
class MultiAgentState(TypedDict):
topic: str
research: Optional[str]
analysis: Optional[str]
report: Optional[str]
feedback: Optional[str]
llm = ChatOpenAI(model="gpt-4")
def researcher_node(state: MultiAgentState) -> MultiAgentState:
prompt = f"深入研究这个主题:{state['topic']}"
response = llm.invoke(prompt)
return {"research": response.content}
def analyst_node(state: MultiAgentState) -> MultiAgentState:
prompt = f"分析这份研究:{state['research']}"
response = llm.invoke(prompt)
return {"analysis": response.content}
def writer_node(state: MultiAgentState) -> MultiAgentState:
prompt = f"基于分析撰写报告:{state['analysis']}"
response = llm.invoke(prompt)
return {"report": response.content}
def reviewer_node(state: MultiAgentState) -> MultiAgentState:
prompt = f"审核这份报告并提供反馈:{state['report']}"
response = llm.invoke(prompt)
return {"feedback": response.content}
# 条件:是否需要修订
def should_revise(state):
if "修订" in state["feedback"].lower():
return "writer"
return END
# 构建图
graph = StateGraph(MultiAgentState)
graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node)
graph.add_node("writer", writer_node)
graph.add_node("reviewer", reviewer_node)
graph.set_entry_point("researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", "reviewer")
graph.add_conditional_edges("reviewer", should_revise)
app = graph.compile()
result = app.invoke({"topic": "2024年AI趋势"}) Level 6:流式输出
实现实时流式输出以提供更好的用户体验。
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# 创建支持流式的模型
model = ChatOpenAI(model="gpt-4", streaming=True)
prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的故事")
chain = prompt | model
# 流式输出
for chunk in chain.stream({"topic": "勇敢的骑士"}):
print(chunk.content, end="", flush=True) from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict
class StreamState(TypedDict):
topic: str
story: str
async def generate_story(state: StreamState) -> StreamState:
model = ChatOpenAI(model="gpt-4")
prompt = f"讲一个关于{state['topic']}的故事"
story_parts = []
async for chunk in model.astream(prompt):
story_parts.append(chunk.content)
print(chunk.content, end="", flush=True)
return {"story": "".join(story_parts)}
# 构建图
graph = StateGraph(StreamState)
graph.add_node("generate", generate_story)
graph.set_entry_point("generate")
graph.add_edge("generate", END)
app = graph.compile()
# 流式执行
import asyncio
asyncio.run(app.ainvoke({"topic": "勇敢的骑士"})) Level 7:错误处理与重试逻辑
实现带有自动重试的健壮错误处理。
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from tenacity import retry, stop_after_attempt, wait_exponential
# 使用重试装饰器包装
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_llm_with_retry(prompt: str) -> str:
model = ChatOpenAI(model="gpt-4")
try:
response = model.invoke(prompt)
return response.content
except Exception as e:
print(f"错误:{e},正在重试...")
raise
# 使用
try:
result = call_llm_with_retry("什么是 LangChain?")
except Exception as e:
print(f"重试后失败:{e}") from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Optional
class RetryState(TypedDict):
prompt: str
response: Optional[str]
error: Optional[str]
attempts: int
MAX_RETRIES = 3
def call_llm(state: RetryState) -> RetryState:
model = ChatOpenAI(model="gpt-4")
try:
response = model.invoke(state["prompt"])
return {"response": response.content, "error": None}
except Exception as e:
return {"error": str(e), "attempts": state.get("attempts", 0) + 1}
def handle_error(state: RetryState) -> RetryState:
print(f"第 {state['attempts']} 次尝试失败:{state['error']}")
return state
def should_retry(state: RetryState):
if state.get("response"):
return END
if state.get("attempts", 0) >= MAX_RETRIES:
return END
return "call_llm"
# 构建图
graph = StateGraph(RetryState)
graph.add_node("call_llm", call_llm)
graph.add_node("handle_error", handle_error)
graph.set_entry_point("call_llm")
graph.add_conditional_edges("call_llm", should_retry)
graph.add_edge("handle_error", "call_llm")
app = graph.compile()
result = app.invoke({"prompt": "什么是 LangGraph?", "attempts": 0}) 架构深入分析
LangChain 架构
LangChain 遵循分层架构,包含三个主要层次:
- 集成层:与外部服务的连接(LLM、向量存储、工具)
- 核心层:基础抽象(提示、链、记忆)
- 应用层:常见用例的预构建解决方案
基于链的设计促进组合——较小的链可以组合成较大的链。 然而,这可能导致深层嵌套的结构,难以调试和维护。
LangGraph 架构
LangGraph 引入了基于图的架构,包含以下组件:
- 状态模式:定义共享状态结构的 TypedDict
- 节点:接收和更新状态的函数
- 边:定义节点之间的流程(静态或条件)
- 检查点:用于状态快照的持久化层
图结构对流程控制更加显式,使复杂应用程序更容易理解和调试。 状态管理内置在框架中,而不是作为附加功能。
性能基准测试
基于社区基准测试和内部测试,以下是典型的性能特征:
| 指标 | LangChain | LangGraph |
|---|---|---|
| 简单链开销 | ~5ms | ~15ms |
| 复杂工作流(5+ 步骤) | ~50ms 开销 | ~30ms 开销 |
| 状态序列化 | 手动 | 自动(~10ms) |
| 内存使用 | 简单场景更低 | 基线略高 |
| 流式延迟 | ~50ms 首个 token | ~55ms 首个 token |
注意:这些是近似值,会根据具体用例、LLM 提供商和基础设施而有所不同。
用例分析
LangChain 最适用场景
1. 简单聊天机器人
如果你正在构建具有基本对话历史的简单聊天机器人, LangChain 的对话链实现快速且满足大多数需求。
2. 文档问答系统
对于文档加载一次并重复查询的 RAG 应用, LangChain 的检索链效果良好且易于设置。
3. 原型开发与实验
当你需要快速测试想法或构建 MVP 时,LangChain 丰富的 模板和示例支持快速开发。
LangGraph 最适用场景
1. 多代理系统
当多个代理需要协作、传递信息或相互迭代工作时, LangGraph 的图结构提供了必要的控制流程。
2. 带分支的复杂工作流
需要根据中间结果进行动态路由的应用程序, 从 LangGraph 的条件边和显式状态管理中受益。
3. 长时间运行的流程
可能跨越数小时或数天的工作流需要持久化和恢复能力, LangGraph 通过其检查点系统提供这些功能。
4. 人机协作系统
当在特定点需要人工审批或干预时,LangGraph 的 中断和恢复功能使实现变得简单。
迁移指南
从 LangChain 迁移到 LangGraph
如果您正在考虑从 LangChain 迁移到 LangGraph,以下是一般方法:
- 识别状态:定义需要在步骤之间持久化的数据
- 将链映射到节点:每个链成为图中的一个节点
- 定义边:根据工作流逻辑连接节点
- 处理记忆:用状态模式替换记忆类
- 增量测试:一次迁移一个组件
同时使用两者
请记住,LangGraph 构建在 LangChain 之上,因此您可以继续使用 LangChain 组件(LLM、工具、检索器)在 LangGraph 节点中。这允许渐进式 迁移和两全其美的解决方案。
灵活性与可扩展性
LangChain 灵活性
- 丰富的集成生态系统(100+ 集成)
- 易于交换组件(LLM、向量存储等)
- 链组合允许构建复杂工作流
- 可以通过扩展基类创建自定义链
LangGraph 灵活性
- 图结构允许任何工作流拓扑
- 条件边支持动态路由
- 内置持久化和状态管理
- 更好地支持循环和迭代
状态管理与记忆
LangChain 状态管理
LangChain 使用记忆类在调用之间持久化状态。有各种记忆类型可用(ConversationBufferMemory、 ConversationSummaryMemory 等),但它们可能难以配置,并且可能无法很好地处理所有场景。
LangGraph 状态管理
LangGraph 通过其基于 TypedDict 的状态模式内置状态管理。状态在节点之间传递, 可以在每个步骤修改。检查点系统允许工作流的持久化和恢复。
错误处理与调试
LangChain 错误处理
- 链调用周围的 try-catch 块
- 用于日志和监控的回调
- LangSmith 集成用于追踪
LangGraph 错误处理
- 用于处理失败的错误节点
- 内置图执行的重试逻辑
- 更好的执行流程可见性
- 用于调试的状态快照
性能与效率
两个框架在基本操作上有相似的性能特征。主要差异: p>
- LangChain:简单链开销较低;状态管理可能变得复杂
- LangGraph:初始开销稍高;复杂、有状态工作流性能更好
社区与文档
LangChain 生态系统
- 更大的社区和更多资源
- 丰富的文档和示例
- 更多第三方集成
- LangSmith 用于可观测性
LangGraph 生态系统
- 增长的社区
- 核心概念文档完善
- 与 LangChain 生态系统紧密集成
- LangSmith 集成用于追踪
生产环境适用性
LangChain 在生产中
- 被许多生产应用使用
- LangServe 用于部署
- 适合简单的线性工作流
- 可能需要自定义状态管理
LangGraph 在生产中
- 专为复杂、有状态应用构建
- 内置持久化和恢复
- 更适合多代理系统
- 显式的错误处理和重试逻辑
总结与建议
何时选择 LangChain
- 简单的线性工作流
- 快速原型和实验
- 需要广泛的第三方集成
- 团队刚接触 LLM 开发
- 标准用例(聊天机器人、问答、RAG)
何时选择 LangGraph
- 复杂的多步骤工作流
- 需要循环或迭代的应用程序
- 多代理系统
- 需要显式状态管理
- 长时间运行、可恢复的工作流
混合使用
值得注意的是,LangGraph 构建在 LangChain 之上,因此您可以同时使用两者。 许多应用程序使用 LangChain 进行基本操作(LLM 调用、检索器),同时使用 LangGraph 进行编排和状态管理。
未来趋势
两个框架都在积极开发和演进。LangGraph 正在成为 LangChain 生态系统中复杂应用的推荐方法。 未来预计会看到更多的融合和更好的工具支持。