【AI进阶】从 LangChain 到 LangGraph:手把手教你构建“自动纠错”的多智能体工作流
在大模型应用开发(LLM App Development)的早期,LangChain几乎是开发者的首选,它像胶水一样把模型、提示词和数据连接在一起。但随着业务越来越复杂,我们开始需要构建Agent(智能体)——也就是那些能自主思考、循环执行、甚至自我纠错的系统。这时候,线性的“链(Chain)”就不够用了。我们需要“图(Graph)”和“循环(Loop)”。LangGraph因此应运而生。
前言
在大模型应用开发(LLM App Development)的早期,LangChain 几乎是开发者的首选,它像胶水一样把模型、提示词和数据连接在一起。但随着业务越来越复杂,我们开始需要构建 Agent(智能体)——也就是那些能自主思考、循环执行、甚至自我纠错的系统。
这时候,线性的“链(Chain)”就不够用了。我们需要“图(Graph)”和“循环(Loop)”。LangGraph 因此应运而生。
本文将带你快速厘清 LangChain 与 LangGraph 的区别,并通过一个经典的**“开发者-测试者(Coder-Tester)”多智能体协作案例**,展示如何用 LangGraph 构建一个能自动修复 Bug 的编程助手。
一、 核心概念:LangChain vs LangGraph
在开始写代码之前,我们需要先理解为什么要从 Chain 升级到 Graph。
1. LangChain:线性的流水线
LangChain 的核心是 DAG(有向无环图)。任务通常是线性的:
输入 -> 步骤A -> 步骤B -> 步骤C -> 输出
它非常适合 RAG(检索增强生成)或简单的问答机器人,因为流程是确定的,只有向前走,没有回头路。
2. LangGraph:有“记忆”的循环
LangGraph 是建立在 LangChain 之上的扩展。它的核心特性是 State(状态) 和 Cycle(循环)。
输入 -> 思考 -> 执行 -> 观察结果 -> (如果不满意) -> 回到思考 -> ...
它允许我们将流程定义为图中的节点(Nodes)和边(Edges),并且可以根据条件(Conditional Edges)决定是继续下一步,还是回退重试。这正是构建能够自我纠错的 Agent 所需的基础。
|
特性 |
LangChain (LCEL) |
LangGraph |
|
结构 |
线性链 (DAG) |
图结构 (支持循环 Cycle) |
|
状态管理 |
较弱,主要靠传参 |
强,全局状态 (State) 共享 |
|
适用场景 |
简单问答、数据处理流水线 |
复杂 Agent、多步推理、自我纠错 |
二、 实战案例:构建“自动纠错”编程助手
为了演示 LangGraph 的强大,我们来构建一个多智能体协作系统。
场景描述
我们需要两个 AI 角色配合完成任务:
- Coder(开发者):负责编写代码。
- Tester(测试者):负责运行代码并反馈结果。
工作流逻辑:
Coder 写代码 -> Tester 运行 -> 如果报错,把错误信息扔回给 Coder 重写 -> 直到测试通过或达到最大重试次数。
三、 代码实现 (Python)
1. 环境准备
首先,你需要安装必要的库:
pip install langgraph langchain langchain_openai
2. 定义“共享状态” (The State)
这是多智能体协作的“黑板”,所有节点都能读写这里的信息。
from typing import TypedDict, List, Annotated
import operator
class AgentState(TypedDict):
messages: List[str] # 记录用户的原始需求
code: str # 存放当前生成的代码
test_output: str # 存放测试运行结果或报错信息
iterations: int # 记录重试次数,防止死循环
3. 定义节点 (The Nodes)
我们定义两个函数,分别代表 Coder 和 Tester 的逻辑。
Coder 节点:负责写代码和改代码
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o") # 或者其他你喜欢的模型
def coder_node(state: AgentState):
task = state["messages"][-1]
error = state.get("test_output", "")
current_iter = state.get("iterations", 0)
# 构建 Prompt:如果有报错,要求根据报错修复
if error and error != "Success":
prompt = f"""
任务:{task}
上次运行代码报错如下:
{error}
请根据报错信息,重写修复后的 Python 代码。只返回代码,不要其他废话。
"""
else:
prompt = f"任务:{task}\n\n请编写 Python 代码解决该问题。只返回代码。"
# 调用模型 (此处简化了代码提取逻辑)
response = llm.invoke(prompt)
code = response.content.strip().replace("```python", "").replace("```", "")
print(f"\n[Coder] 第 {current_iter + 1} 次编写代码...")
return {
"code": code,
"iterations": current_iter + 1
}
Tester 节点:负责“质检”
def tester_node(state: AgentState):
code = state["code"]
print("[Tester] 正在运行测试...")
try:
# 警告:生产环境请使用沙箱执行代码 (如 e2b 或 docker)
# 这里为了演示,使用 exec 模拟运行
exec_globals = {}
exec(code, exec_globals)
return {"test_output": "Success"}
except Exception as e:
return {"test_output": f"Error: {str(e)}"}
4. 定义控制流 (The Router)
这是 LangGraph 的核心,决定图的走向。
from langgraph.graph import END
def router(state: AgentState):
result = state["test_output"]
iterations = state["iterations"]
# 1. 成功则结束
if result == "Success":
print(">>> 测试通过,任务完成!")
return "end"
# 2. 超过3次重试强制结束
if iterations >= 3:
print(">>> 重试次数过多,任务失败。")
return "end"
# 3. 失败则回退给 Coder
print(f">>> 检测到错误,路由回 Coder 进行修复...")
return "retry"
5. 组装图 (Build Graph)
最后,把积木搭起来。
from langgraph.graph import StateGraph
# 初始化图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("coder", coder_node)
workflow.add_node("tester", tester_node)
# 设置起点
workflow.set_entry_point("coder")
# 添加普通边:Coder 写完必然去 Tester
workflow.add_edge("coder", "tester")
# 添加条件边:Tester 之后,由 router 决定去向
workflow.add_conditional_edges(
"tester", # 上一个节点
router, # 路由函数
{ # 映射关系
"end": END,
"retry": "coder"
}
)
# 编译应用
app = workflow.compile()
6. 运行测试
# 这是一个会故意出错的任务(斐波那契数列,容易出现递归深度错误或语法问题)
initial_state = {
"messages": ["写一个计算斐波那契数列第100项的函数,确保打印结果"],
"iterations": 0,
"test_output": ""
}
# 启动图
for s in app.stream(initial_state):
pass
# stream 会逐步输出每个节点的状态,这里略过打印细节
四、 运行过程模拟
当你运行上述代码时,可能会看到这样的日志流:
- [Coder] 第一次写代码,使用了简单的递归
fib(n-1) + fib(n-2)。 - [Tester] 运行代码,因为计算第100项,导致计算超时或极其缓慢(或者如果模型写错语法,直接报错)。
- [Router] 捕获到问题,决定 retry。
- [Coder] 收到反馈:“运行超时/报错”。思考后改为动态规划或迭代法。
- [Tester] 再次运行。
- [Router] 发现结果 "Success",路由至 END。
总结
通过这个案例,我们可以看到 LangGraph 相比 LangChain 的质变:
- 容错性:错误不再是程序的终点,而是下一次迭代的输入。
- 状态流转:多智能体之间通过共享状态(State)实现了无缝的上下文传递。
更多推荐


所有评论(0)