LangGraph 编排实践

目标：把 Agent 最小可运行项目的手写 loop 改写成 LangGraph 状态图，看清节点、边、条件分支、循环各对应什么。选型对照见 RAG 与框架选型；概念见 Agent 核心概念与架构。

本篇只覆盖 Agent 编排。向量 RAG 拼装见 [LangChain RAG 实践](/docs/ai/RAG/LangChain RAG 实践)；不要和 GraphRAG 搞混——GraphRAG 是另一种索引，不是编排框架。终端侧可扩展 coding CLI 见 [Pi Agent Harness 实践](/docs/ai/Agent/Pi Agent Harness 实践)，和本篇不在同一层。

目标与验收

验收项	标准
对照	能说出 `for step in range(max_steps)` 对应图上哪条循环边
能跑	同一假 LLM + `add` 工具，问「3+5」仍走 tool → final
能终止	`max_steps` / `recursion_limit` 卡住死循环
能讲解	画出 State / Node / Edge / Conditional Edge

手写 loop 对照

手写 loop	LangGraph
`obs` / `decision` 变量	State（图上流转的共享状态）
`decide()` / `call_tool()`	Node（一步函数：读 state、返回补丁）
`for` 里顺序调用	Edge（固定下一跳）
`if type == "final"`	Conditional Edge（按 state 路由）
`max_steps`	状态里的计数 + `recursion_limit`

记忆口诀：LangGraph 把 Orchestrator 从 while 换成显式状态机；LLM 仍在某个节点里做决策，图只负责「下一步去哪」。

START
  → decide（假/真 LLM：选 tool 或 final）
       │
       ├─ tool  → call_tool → 回到 decide   （循环）
       └─ final → finish → END

建议目录

langgraph_agent/
  tools.py          # 与 mini_agent 相同
  fake_llm.py       # 与 mini_agent 相同
  graph.py          # StateGraph
  requirements.txt

requirements.txt 示例（版本以 LangGraph 文档为准）：

langgraph

状态与图

节点返回的是 state 补丁（dict），默认按 key 覆盖合并。对话历史若要「追加」而不是覆盖，需用 add_messages reducer——面试常考这一点。

# graph.py
from typing import Any, Literal, TypedDict

from langgraph.graph import END, START, StateGraph

from fake_llm import decide
from tools import call_tool


class AgentState(TypedDict):
    user: str
    last_obs: str | None
    result: str | None
    tool_name: str | None
    tool_args: dict[str, Any] | None
    steps: int


def decide_node(state: AgentState) -> dict:
    decision = decide(state["user"], state["last_obs"])
    print(f"step={state['steps']} decision={decision}")
    if decision["type"] == "final":
        return {
            "result": decision["content"],
            "tool_name": None,
            "tool_args": None,
            "steps": state["steps"] + 1,
        }
    return {
        "tool_name": decision["name"],
        "tool_args": decision["args"],
        "steps": state["steps"] + 1,
    }


def call_tool_node(state: AgentState) -> dict:
    obs = str(call_tool(state["tool_name"], state["tool_args"]))
    return {"last_obs": obs}


def finish_node(state: AgentState) -> dict:
    if state.get("result"):
        return {}
    return {"result": "达到最大步数，中止"}


def route(state: AgentState) -> Literal["call_tool", "finish"]:
    if state["steps"] >= 5:
        return "finish"
    if state.get("tool_name"):
        return "call_tool"
    return "finish"


def build_app():
    g = StateGraph(AgentState)
    g.add_node("decide", decide_node)
    g.add_node("call_tool", call_tool_node)
    g.add_node("finish", finish_node)
    g.add_edge(START, "decide")
    g.add_conditional_edges(
        "decide",
        route,
        {"call_tool": "call_tool", "finish": "finish"},
    )
    g.add_edge("call_tool", "decide")  # 观察后再规划
    g.add_edge("finish", END)
    return g.compile()


if __name__ == "__main__":
    app = build_app()
    out = app.invoke(
        {
            "user": "帮我算 3+5",
            "last_obs": None,
            "result": None,
            "tool_name": None,
            "tool_args": None,
            "steps": 0,
        },
        {"recursion_limit": 10},
    )
    print(out["result"])

运行：python graph.py，应与手写版一样看到一步 tool、一步 final。tools.py / fake_llm.py 直接复用最小项目。

recursion_limit 是图执行的硬天花板（节点跳转次数），和业务里的 max_steps 一起用：前者防图写错导致无限边，后者表达产品规则。

预置 ReAct（真 LLM）

有能 bind_tools 的模型时，不必手画 decide/tools 两节点，可用预置图：

from langgraph.prebuilt import create_react_agent

# model = ChatOpenAI(...)  # 需支持 tool calling
# app = create_react_agent(model, tools=[add])
# app.invoke({"messages": [{"role": "user", "content": "帮我算 3+5"}]})

自己画 StateGraph	`create_react_agent`
分支、拒答、检索打分、人审节点都要显式	标准 ReAct：模型 ↔ 工具循环
假 LLM / 规则路由也能跑	依赖真模型的 tool call
面试好画、好对照手写 loop	上线快，定制少时够用

先会手画，再把预置图当语法糖。需要「检索质量不够就改写 query」这类分支时，回到 add_conditional_edges。

检查点与人审（点到为止）

生产里图要能暂停再续，不是把整段对话塞进一次 invoke。

from langgraph.checkpoint.memory import InMemorySaver

memory = InMemorySaver()
# app = g.compile(checkpointer=memory)
# app.invoke(inputs, {"configurable": {"thread_id": "user-42"}})

能力	用途
`thread_id`	同一会话的 state 续跑（短期 / 工作记忆落在 checkpointer）
`interrupt_before=["call_tool"]`	调工具前停下，等人审
持久化 checkpointer	进程重启后仍能 resume；PoC 用内存即可

不可逆动作（转账、删数据、对外发送）把人审做成图上的中断，比只在 Prompt 里写「请谨慎」可靠。前后端事件、取消与超时未审见流式取消与人审；工具失败重试见工具调用与评测实践。跨会话画像、摘要与写入策略见 [Agent 记忆分层实践](/docs/ai/Agent/Agent 记忆分层实践)（checkpointer 只覆盖本 thread）。

和 RAG 怎么接

常见接法：检索当工具或当节点，图负责多步，RAG 负责知识。

decide
  ├─ retrieve（LangChain retriever / 自写检索）→ decide
  ├─ 其它业务工具 → decide
  └─ 有依据则 generate → END；无命中则拒答 → END

段落 FAQ：LangChain 向量链往往就够，不必上图。
要「检索 → 看结果 → 再搜或调 API」：图里挂 retriever 节点/工具。
全局主题 / 实体关系：检索后端可换成 GraphRAG，编排仍是 LangGraph。

常见坑

LangGraph = GraphRAG：前者编 Agent 循环；后者建图索引。名字都有 Graph，层不同。
State 默认为覆盖：messages 直接赋值会丢掉历史；追加要用 add_messages。
只靠模型自己停：循环边必须配 max_steps / recursion_limit / 人审，否则工具一抖就空转。
预置 ReAct 当万能：固定审批流、强路由、多拒答分支，手画图更清晰。
API 漂移：langgraph 与 langchain-* 拆包频繁，import 以官方文档为准。

实践要点

先跑通 Agent 最小可运行项目，再对照本篇把 while 拆成节点。
验收仍是：工具是否被调用、能否终止、轨迹是否可读——框架不改变这三项。
PoC 用假 LLM + 内存 checkpointer；评测与多工具见工具调用与评测实践；跨会话记忆见 [Agent 记忆分层实践](/docs/ai/Agent/Agent 记忆分层实践)。
私有知识接到图上时，检索实现见 [LangChain RAG 实践](/docs/ai/RAG/LangChain RAG 实践)。
面试挂钩：能画 StateGraph，并说明与手写 Orchestrator 的一一对应即可。

面试速记

LangGraph 是什么？ Agent 编排框架：用有状态图表达规划→行动→观察，而不是再写一层 while。
和 LangChain / GraphRAG？ LangChain 拼 RAG 链；GraphRAG 建图索引；LangGraph 编循环。常组合，不三选一。
四件套？ State、Node、Edge、Conditional Edge；循环 = 指回已有节点的边。
怎么防死循环？ 业务 max_steps + 运行时 recursion_limit + 必要人审。
何时手画、何时预置？ 标准 ReAct 用 create_react_agent；有显式分支/闸门就自己 StateGraph。

目标与验收​

手写 loop 对照​

建议目录​

状态与图​

预置 ReAct（真 LLM）​

检查点与人审（点到为止）​

和 RAG 怎么接​

常见坑​

实践要点​

面试速记​