为什么传统链式调用(Chains)不够用?
当 AI 任务变得复杂(例如:编写代码 -> 运行单元测试 -> 报错分析 -> 修复重试),传统的顺序链(DAG)无法处理循环图结构和状态回滚。
LangGraph 将任务建模为有向图(Graph),每个节点是一个 Agent 或 Tool,边决定流转条件,状态(State)贯穿全局并在每一步保留快照。
1. 核心状态图构建
from typing import TypedDict, Annotated, Sequence
import operator
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
messages: Annotated[Sequence[str], operator.add]
iteration: int
test_passed: bool
def coder_node(state: AgentState):
print(f"Coder: 正在生成第 {state['iteration']} 版代码...")
return {"messages": ["代码已生成"], "iteration": state["iteration"] + 1}
def test_runner_node(state: AgentState):
# 模拟测试运行
passed = state["iteration"] >= 2
return {"test_passed": passed, "messages": [f"测试结果: {'PASS' if passed else 'FAIL'}"]}
def should_continue(state: AgentState):
if state["test_passed"]:
return "end"
if state["iteration"] > 3:
return "end"
return "coder"
# 构建图结构
workflow = StateGraph(AgentState)
workflow.add_node("coder", coder_node)
workflow.add_node("test_runner", test_runner_node)
workflow.set_entry_point("coder")
workflow.add_edge("coder", "test_runner")
workflow.add_conditional_edges("test_runner", should_continue, {
"coder": "coder",
"end": END
})
app = workflow.compile()
2. 人在回路(Human-in-the-Loop)中断机制
在关键生产操作(例如合并 PR、数据库迁移)前,可以通过 interrupt_before 参数挂起执行:
app = workflow.compile(interrupt_before=["deploy_node"])
# 执行到 deploy_node 前挂起,等待人工审核确认后再继续恢复状态
LangGraph 极大提升了智能体流水线的可控性与确定性,是目前工业级复杂 Agent 的主力框架之一。