为什么传统链式调用（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 的主力框架之一。