09. 用 LangGraph 构建进阶 Agent

学习目标

理解简单 Agent loop 的边界。
认识 LangGraph 的 state、node 和 edge。
知道什么时候需要显式工作流、条件路由、持久执行或 human-in-loop。
学会在 create_agent 和 LangGraph 之间做选择。

为什么需要 LangGraph

create_agent 适合快速构建“模型决定是否调用工具，再给出答案”的循环。它隐藏了很多流程细节，适合入门和常见工具调用场景。

当流程变复杂时，隐藏细节会变成限制。例如你可能需要先起草答案，再审核，再根据评分决定是否重试；或者工具调用后必须等待人工确认；或者执行过程需要持久化，程序重启后还能继续。这些需求不只是“让模型多想一步”，而是需要明确的工作流结构。

LangGraph 提供显式图结构：state 保存流程数据，node 执行一步计算，edge 决定下一步去哪。这样你可以把复杂 Agent 拆成可理解、可测试、可观测的步骤。

最小图

运行示例：

bash

uv run python examples/09_langgraph_agent.py

这个脚本定义了一个最小图：

START 进入 draft_answer。
draft_answer 根据问题生成简短草稿，并写入 state["draft"]。
polish_answer 把草稿润色成两个清晰要点，并写入 state["final"]。
polish_answer 结束后到达 END。

这里的 state 是结构化的 TypedDict，包含 question、draft 和 final。每个 node 只更新自己负责的字段。图的好处是流程清楚：你能看到每一步输入、输出和边界，而不是把所有逻辑塞进一个长 prompt。

LangGraph 还可以做条件路由。例如根据模型评分决定走“直接回答”还是“重新检索”，根据工具结果决定“继续执行”还是“请求人工确认”。在生产场景里，它也常用于 durable execution 和 human-in-loop：流程状态可以保存下来，等待外部事件或人工审批后继续。

选择规则

用 create_agent：工具少、流程短、主要让模型自己决定是否调用工具。
用 LangGraph：步骤固定、需要显式 state、需要条件路由、需要人工介入、需要重试和恢复、需要清楚观测每个节点。
先用 create_agent：原型阶段不确定流程是否稳定，可以先快速验证。
迁移到 LangGraph：当调试开始依赖“它到底走了哪一步”，或者业务要求每一步可控时，就应该显式建图。

不要因为 LangGraph 更强就默认使用它。简单任务用简单抽象，复杂流程再引入图。

代码阅读重点

阅读 examples/09_langgraph_agent.py 时，重点看四个元素：

AgentState：定义图在节点之间传递什么状态。
draft_answer()：第一个节点，生成草稿。
polish_answer()：第二个节点，基于草稿生成最终回答。
builder.add_edge(...)：定义节点执行顺序。

这个示例故意没有工具调用和复杂条件分支。它的目标是让你先看懂 LangGraph 的基本形状：状态进来，节点处理，边决定下一步，最终得到新状态。

从线性图到复杂图

最小图是线性的：

text

START -> draft_answer -> polish_answer -> END

复杂图会增加：

条件边：根据状态决定走哪条路径。
重试节点：失败后重新执行某一步。
人工节点：等待人类审核或补充信息。
持久化：中断后恢复到上次状态。
多模型节点：不同节点使用不同模型或不同工具。

不要一开始就构建复杂图。先把流程画出来，再决定哪些步骤需要模型，哪些步骤应该由确定性代码完成。

实验练习

在 polish_answer 前增加一个 critique_answer 节点。
给 state 增加一个 critique 字段。
把最终回答格式从“两条 bullet”改成“三步建议”。

自测问题

LangGraph 的 state 和聊天 memory 是同一个概念吗？
为什么固定流程更适合用图表达？
什么情况下 create_agent 比 LangGraph 更简单、更合适？

深入理解：图不是为了复杂，而是为了显式

LangGraph 的价值不在于“看起来高级”，而在于把隐含流程显式化。

在简单 Agent 中，下一步通常由模型隐式决定。你只能通过日志观察模型到底做了什么。当流程变复杂时，这种隐式控制会带来三个问题：

难以保证必须执行的步骤一定执行。
难以恢复中断后的状态。
难以解释为什么走了某条路径。

图结构把流程拆成节点和边。每个节点负责一件事，每条边说明下一步怎么走。这样你可以单独测试节点，也可以观察每次运行经过了哪些节点。

State 设计原则

State 应该保存流程真正需要的结构化信息，而不是把所有内容塞进一个字符串。

好的 state 字段：

question: 用户原始问题。
draft: 草稿。
retrieved_docs: 检索结果。
approval_status: 人工审核状态。
retry_count: 重试次数。

不好的 state 字段：

everything: 所有内容混在一起。
context: 无法区分来源和用途的大字符串。
result: 每个节点都写这个字段，互相覆盖。

State 字段越清晰，节点边界越清晰，调试越容易。

LangGraph 反模式

为两步简单任务引入图，增加无意义复杂度。
节点太大，一个节点里做模型调用、检索、工具执行和格式化。
state 字段没有结构，所有节点读写同一个字符串。
条件边过多，但没有日志解释为什么走某条边。
没有失败路径，任何节点异常都会终止整个流程。

LangGraph 更适合严肃流程，但它不会自动让系统可靠。可靠性来自清楚的 state、可测试节点、明确边界和良好的错误处理。

迁移策略

如果你已经有一个 create_agent 原型，可以按下面方式迁移：

先保留原 Agent，把它作为图中的一个节点。
把确定性前置步骤抽成节点，例如权限校验、输入清洗、检索。
把后置步骤抽成节点，例如格式化、审核、保存结果。
再考虑条件路由，例如检索不足时走澄清问题。
最后加入持久化和 human-in-the-loop。

不要一次性重写。先把最痛的流程控制问题显式化。

完整示例

python

from __future__ import annotations

import sys
from pathlib import Path
from typing import TypedDict

sys.path.append(str(Path(__file__).resolve().parents[1]))

from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph

from examples.common.config import load_config, ollama_client_kwargs
from examples.common.ollama_check import assert_model_available, assert_ollama_running


DEFAULT_QUESTION = "什么时候应该使用 LangGraph？"


def _friendly_error(exc: RuntimeError, model: str) -> str:
    message = str(exc)
    if message.startswith("Cannot reach Ollama."):
        return "无法连接 Ollama。请先打开 Ollama 应用，或运行 `ollama serve`，并检查 OLLAMA_BASE_URL。"
    if message.startswith("Cannot read Ollama model list"):
        return "无法读取 Ollama 模型列表。请检查 Ollama 是否正常运行，以及 OLLAMA_BASE_URL 是否正确。"
    if message.startswith("Ollama model list"):
        return "Ollama 返回的模型列表格式异常。请检查 OLLAMA_BASE_URL 是否指向 Ollama。"
    if message.startswith("Ollama model `"):
        return f"当前配置的 Ollama 模型未安装。请先运行 `ollama pull {model}`。"
    return message


class AgentState(TypedDict):
    question: str
    draft: str
    final: str


def draft_answer(state: AgentState, model: ChatOllama) -> dict[str, str]:
    response = model.invoke(
        f"请用中文为这个问题起草一个简短回答，不超过 120 字：{state['question']}"
    )
    return {"draft": str(response.content)}


def polish_answer(state: AgentState, model: ChatOllama) -> dict[str, str]:
    response = model.invoke(
        "请把下面的草稿润色成两个清晰的中文要点。只输出两个 bullet：\n\n"
        f"{state['draft']}"
    )
    return {"final": str(response.content)}


def build_graph(model: ChatOllama):
    graph = StateGraph(AgentState)
    graph.add_node("draft_answer", lambda state: draft_answer(state, model))
    graph.add_node("polish_answer", lambda state: polish_answer(state, model))
    graph.add_edge(START, "draft_answer")
    graph.add_edge("draft_answer", "polish_answer")
    graph.add_edge("polish_answer", END)
    return graph.compile()


def main() -> None:
    config = load_config()
    try:
        assert_ollama_running(config.ollama_base_url)
        assert_model_available(config.ollama_base_url, config.ollama_model)
    except RuntimeError as exc:
        print(f"错误：{_friendly_error(exc, config.ollama_model)}")
        return

    model = ChatOllama(
        model=config.ollama_model,
        base_url=config.ollama_base_url,
        client_kwargs=ollama_client_kwargs(),
        temperature=0,
    )

    question = input(f"问题（直接回车使用默认问题：{DEFAULT_QUESTION}）：").strip()
    if not question:
        question = DEFAULT_QUESTION

    try:
        app = build_graph(model)
        result = app.invoke({"question": question, "draft": "", "final": ""})
    except Exception as exc:
        print(f"错误：LangGraph 执行失败：{exc}")
        return

    print(f"\nAI：{result.get('final', '')}")


if __name__ == "__main__":
    main()

09. 用 LangGraph 构建进阶 Agent ​

学习目标 ​

为什么需要 LangGraph ​

最小图 ​

选择规则 ​

代码阅读重点 ​

从线性图到复杂图 ​

实验练习 ​

自测问题 ​

深入理解：图不是为了复杂，而是为了显式 ​

State 设计原则 ​

LangGraph 反模式 ​

迁移策略 ​

完整示例 ​