LangGraph 怎样收紧 LLM 的决策权

一、同一个问题，为什么会走出两条路

早期版本里，我让模型自己决定分析流程。给它工具、给它目标，让它规划下一步，这听起来很像 Agent。

后来我把同一个问题问了两次。第一次，它先做产业诊断，再查政策。第二次，它先查政策，再回头补诊断。两份结论都不离谱，却不能放在一起比较：一份把问题归因到产业结构，另一份归因到政策落地。结论的口径不同，后续也没法复盘为什么会不同。

聊天时这不算大问题。县域经济分析系统不一样，它的产出会进入决策材料。我需要的是：

  • 同类请求在相同规则和相同输入下走同一条业务路径；
  • 每个判断都能追到上游数据、执行节点和规则版本；
  • 数据不满足条件时，系统明确说明缺什么，而不是补一段看似合理的分析。

这三个要求把模型的自由度压得很低。我转向 LangGraph，原因就在这里。它没有自动消除模型的不确定性，但它提供了一个清晰的程序骨架，让我们能把不该交给模型的决定写进节点、边和状态里。

本文分两层写。前半部分让从未用过 LangGraph 的读者跑通一个小图，弄清它到底在执行什么；后半部分再复盘项目中的路由、数据门禁与人工复核。案例做过脱敏，字段和业务链是项目规则，不是 LangGraph 的固定写法。

二、先校准预期：读完能学会什么，学不会什么

如果你是 AI 初学者，读完并亲手敲完第三节，应该能做到：

  • 用 StateGraph 定义状态、节点和边；
  • 理解节点返回的是状态更新，不必原地修改整份状态；
  • 用条件边把流程分到两个出口；
  • 知道列表为什么需要 reducer，知道 START、END 和 compile() 分别做什么；
  • 把一个“模型决定下一步”的小流程改成“代码根据状态决定下一步”。

读完案例部分，你还会知道一套需要交付的 Agent 工作流通常有哪些部件：输入契约、数据证据、质量门禁、人工复核、执行轨迹和产物版本。

但别把这篇文章当成因果推断教程、统计建模教程或 LangGraph 全部 API 手册。这里的 readiness gate 只能决定“是否允许进入估计流程”，不能证明一个因果结论成立。模型选择、识别假设、对照组可比性和稳健性检验，仍然需要独立设计与审查。

三、先跑起来：一个不调用 LLM 的最小图

初学 LangGraph 最容易踩的坑，是第一段代码就接模型、工具、记忆和异步调用。报错时你不知道问题来自哪里。先从一个完全确定性的图开始，反而能看清框架在做什么。

安装依赖：

1
pip install -U langgraph

下面的程序接收一个问题。它先用普通 Python 规则打标签，再用条件边选择“分析”或“人工澄清”。没有 LLM，所以你每次运行都会得到同样的路径。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from operator import add
from typing import Annotated, Literal
from typing_extensions import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    question: str
    route: Literal["analysis", "clarify"]
    answer: str
    trace: Annotated[list[str], add]


def classify(state: State) -> dict:
    question = state["question"]
    if "产业" in question or "经济" in question:
        return {"route": "analysis", "trace": ["classify: analysis"]}
    return {"route": "clarify", "trace": ["classify: clarify"]}


def analyze(state: State) -> dict:
    return {
        "answer": "进入分析链。下一步可以读取指标、检查数据，再生成报告。",
        "trace": ["analyze: done"],
    }


def ask_for_clarification(state: State) -> dict:
    return {
        "answer": "暂时无法匹配业务链，请补充分析对象、时间范围和目标。",
        "trace": ["clarify: waiting for user"],
    }


def choose_route(state: State) -> Literal["analysis", "clarify"]:
    return state["route"]


builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("analyze", analyze)
builder.add_node("clarify", ask_for_clarification)
builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    choose_route,
    {"analysis": "analyze", "clarify": "clarify"},
)
builder.add_edge("analyze", END)
builder.add_edge("clarify", END)

app = builder.compile()

result = app.invoke({"question": "分析本县产业结构"})
print(result["answer"])
print(result["trace"])

预期输出的 trace 是：

1
["classify: analysis", "analyze: done"]

把问题改成“帮我想个标题”，就会进入 clarify。这段代码虽然小，却已经包含了 LangGraph 的主干：状态保存当前任务快照，节点计算状态更新，边安排下一步。

官方把图定义为 State、Node、Edge 的组合。节点既可以包着 LLM，也可以只是普通 Python 函数；条件边可以根据当前状态决定跳转。图需要在执行前 compile()，编译阶段会做基本的结构检查，并可在这里配置 checkpointer 等运行时能力。LangGraph Graph API

3.1 逐行理解这个例子

State 不是一个必须实例化的类，它是状态的类型契约。这里有四个键：

  • question 是入口输入；
  • route 是分类节点产出的决策；
  • answer 是终端节点产出的文字；
  • trace 记录每一步发生了什么。

注意 trace 的类型：Annotated[list[str], add]。add 是 reducer，意思是节点返回的新列表会追加到旧列表后面。没有它，后一个节点返回的 trace 会把前一个节点的记录整个覆盖掉。

LangGraph 默认对每个键采用“新值覆盖旧值”的规则。只有需要累积、合并或去重的键，才应该显式配置 reducer。不要看到列表就机械地用 add。比如 errors 常常需要追加，current_stage 则应该覆盖；若所有字段都做追加，状态会很快失去语义。

每个节点只返回自己更新的部分。例如 classify 返回 route 与一条 trace，而没有返回 question 或 answer。框架会把这个局部更新合进当前状态。原稿中“节点返回更新后完整状态”的说法过于宽泛，初学者很容易据此在节点里原地修改字典。实践中更推荐返回局部更新：输入和输出清楚，也更便于测试并发分支。

START 与 END 是两个特殊节点。前者表示图的入口，后者表示没有后继动作的终点。add_conditional_edges() 的第一个参数是来源节点，第二个参数是读取状态后返回分支名的函数，第三个参数把分支名映射到目标节点。三者不要混在一起理解。

3.2 什么时候才该把 LLM 放进节点

当任务需要归纳、改写、从证据中组织语言时，LLM 很合适。例如：

1
2
3
4
5
6
7
def draft_report(state: State) -> dict:
    prompt = f"""只根据以下材料写摘要，不要补充材料中没有的事实。

材料：{state['evidence_text']}
"""
    text = model.invoke(prompt).content
    return {"answer": text, "trace": ["draft_report: generated"]}

这段函数依然只是一个节点。LangGraph 不替你保证提示词、模型参数和外部数据相同，因此同一条图路径并不代表每次生成的自然语言完全一致。要追求可比性，至少还要记录模型名、模型版本、提示词版本、温度、检索结果版本和规则版本。

我自己的划分原则很朴素：存在明确判据的事情，先写代码；需要把多份材料压缩成易读文字的事情，再让模型做。路由、权限、金额计算、字段校验、状态迁移和是否允许发布，通常属于前一类。

四、State 才是工作流的接口

很多教程把重点放在 add_node() 与 add_edge() 上，真正决定系统能否维护的却是状态设计。状态相当于所有节点共同遵守的接口：上游承诺写出什么，下游才知道能读到什么。

项目中的状态按关注点拆成下面几块：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from typing_extensions import NotRequired, TypedDict


class CountyEconomyState(TypedDict):
    request: dict
    platform: NotRequired[dict]
    workflow: NotRequired[dict]
    roles: NotRequired[dict]
    data: NotRequired[dict]
    evidence: NotRequired[dict]
    analysis: NotRequired[dict]
    decisions: NotRequired[dict]
    draft: NotRequired[dict]
    review: NotRequired[dict]
    trace: NotRequired[dict]
    outputs: NotRequired[dict]

它不是聊天记录，也不是一个随手往里塞东西的大字典。每一块都有用途：

区域 写入者 下游使用者 应保存什么
request 接口层 全图 区域代码、周期、问题、用户选择
workflow intake 路由节点 工作流 ID、目录版本、当前步骤
data 数据节点 分析节点 原始指标、查询参数、数据版本
evidence 证据节点 写作与审查节点 来源、引用片段、缺口、适用范围
analysis 分析节点 草稿节点 可复算的中间结果与方法说明
review 门禁节点 发布节点 检查结果、问题清单、复核意见
trace 所有节点 运维人员 节点名、时间、输入摘要、规则版本

这个表里最容易被忽略的是 evidence.gaps。它专门记录“当前缺少什么”，例如“缺少 2025 年项目投资额”或“政策文本未给出实施范围”。如果缺口只存在于开发者脑中，生成节点很容易把空白补成顺口的句子；当缺口成为状态的一部分，门禁就能据此阻止正式结论。

4.1 状态字段要先写数据契约

只写 dict 能快速起步，但维护一段时间后会有两个问题：字段名漂移，和字段含义漂移。前者是 county_code、region_code 同时出现；后者更难查，比如 data.metrics 有时是原始数值，有时已经是同比增速。

业务稳定后，建议至少为关键区域定义较细的 TypedDict 或 Pydantic 模型：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from typing import Literal
from typing_extensions import NotRequired, TypedDict


class EvidenceItem(TypedDict):
    source_id: str
    source_url: str
    retrieved_at: str
    quote: str
    supports: list[str]


class EvidenceState(TypedDict):
    items: list[EvidenceItem]
    gaps: list[str]
    status: Literal["complete", "partial", "blocked"]
    boundary_notes: NotRequired[list[str]]

这不是为了把类型写得漂亮，而是为了让“证据来自哪里”“这段证据支持什么”“缺什么”成为可检查的数据。真正要对外发布的结构，最好再做运行时校验。TypedDict 只给编辑器和类型检查器看，运行时不会拦住错误数据。

4.2 Reducer 是并发与循环的隐含规则

Reducer 规定同一个状态键如何把旧值与新值合并。它平时不显眼，遇到并行分支或循环时却会直接决定数据是否丢失。官方文档也明确说明，未设置 reducer 的键会被新值覆盖；自定义 reducer 才会合并两侧的值。State 与 reducer

例如多个节点都记录执行日志，可以定义一个专用的追加函数：

1
2
3
4
5
6
7
8
9
10
from typing import Annotated
from typing_extensions import TypedDict


def append_trace(left: list[dict], right: list[dict]) -> list[dict]:
    return [*left, *right]


class TraceState(TypedDict):
    trace_steps: Annotated[list[dict], append_trace]

如果日志需要按 step_id 去重，append_trace 就不该简单拼接。Reducer 是业务规则的一部分，应当像其他规则一样测试。

五、Node：把副作用当成风险点

节点可以同步也可以异步。它们接收状态，完成计算或副作用，再返回更新。听上去简单，真正落地时最容易出事故的是“副作用”：写数据库、发消息、生成文件、扣费调用外部 API。

考虑一个导出报告的节点：

1
2
3
4
5
6
7
8
def export_report(state: CountyEconomyState) -> dict:
    report_id = state["platform"]["report_id"]
    draft = state["draft"]["content"]
    path = save_report(report_id, draft)  # 写文件或对象存储
    return {
        "outputs": {"draft_path": path},
        "trace": {"last_step": "export_report"},
    }

这段代码能跑，但它有一个工程问题：节点重试或从中断恢复时，save_report() 可能再执行一次。官方文档提示，使用 checkpointer 后恢复执行时，节点会从函数开头重新运行，而不是从函数中间继续；中断前的副作用需要具备幂等性。节点重执行与幂等性

所以副作用节点至少要做到其中一种：

  • 使用稳定的 report_id 或幂等键，重复写入覆盖同一个草稿版本；
  • 先查询是否已成功写入，再决定是否执行；
  • 把外部调用拆得更小，并把已完成的结果持久化；
  • 只有在所有质量门禁通过后，才把草稿提升为正式产物。

还有一个更基础的建议：节点名字写动作，不写角色。load_metrics、check_readiness、draft_report 比 data_agent、reviewer 更容易理解。角色可以存在于提示词或任务配置中，节点名应该告诉维护者它会产生什么状态变化。

六、路由：确定性不等于“只用关键词”

项目里最重要的决定之一，是不让 LLM 单独选择业务链。用户的问题进来后，系统先查一个版本化的 workflow_catalog。目录记录每条链的输入要求、可用工具、输出、人工复核策略和匹配规则。路由节点只负责根据目录写入 workflow_id。

最早的简化版是关键词匹配：

1
2
3
4
5
6
def find_workflows(question: str, catalog: list[dict]) -> list[dict]:
    normalized = question.strip().lower()
    return [
        item for item in catalog
        if any(word in normalized for word in item["trigger_words"])
    ]

它的优点是可解释，缺点也很明显。用户问“招商政策对产业结构有什么影响”，可能同时命中“招商”“政策”“产业”。直接取 matches[0] 并不确定，目录顺序一变，路径就变了。把 LLM 移出路由，并不代表路由天然正确。

更稳妥的做法是把匹配结果变成显式决策，并给歧义留出口：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from typing import Literal


def decide_workflow(question: str, catalog: list[dict]) -> dict:
    matches = find_workflows(question, catalog)

    if len(matches) == 1:
        return {
            "status": "matched",
            "workflow_id": matches[0]["workflow_id"],
            "candidates": [],
        }

    if len(matches) == 0:
        return {
            "status": "unknown",
            "workflow_id": None,
            "candidates": [],
        }

    return {
        "status": "ambiguous",
        "workflow_id": None,
        "candidates": [item["workflow_id"] for item in matches],
    }


def intake(state: CountyEconomyState) -> dict:
    decision = decide_workflow(
        state["request"]["question"],
        load_workflow_catalog(),
    )
    return {
        "workflow": decision,
        "decisions": {
            "requires_human_review": decision["status"] != "matched"
        },
    }


def route_after_intake(
    state: CountyEconomyState,
) -> Literal["dispatch", "clarify"]:
    return "dispatch" if state["workflow"]["status"] == "matched" else "clarify"

生产环境可以把关键词替换为规则表达式、表单选项、分类器分数或 LLM 的结构化分类结果。关键不在于“绝不使用 LLM”，而在于：

  1. 候选、分数、目录版本和最终决定都进入状态；
  2. 低置信度、并列和不支持的请求有固定出口；
  3. 如果 LLM 参与分类，它只能提供候选，代码仍负责阈值、权限和最终状态迁移；
  4. 用一组冻结的真实问题做回归测试，目录改动后比较路径是否变化。

这样才能说路径可复现。更准确地说，是在固定的目录版本、匹配实现、输入规范和外部依赖版本下，路径可以重放。LLM 输出、实时数据库查询和当前时间仍然会带来变化，别把“图是确定的”误写成“整个系统完全确定”。

七、图怎样表达一条真实业务链

县域经济分析不是一次问答。以项目招商为例，系统不应跳过诊断，直接给一串项目名。项目中的一条链大致是：

其他链可能在“运行监测”或“产业诊断”处结束，也可能在“瓶颈归因”后转去政策决策。共享前缀与多点分岔是图表达得很自然的一类结构：

图不是越细越好。一个实用的切分标准是：节点内的操作是否应该一起重试、一起观测、一起授权。如果“拉取指标”和“调用模型写报告”失败后的处理完全不同，它们应该是两个节点。反过来，连续的纯计算如果拆成十个节点，只会让状态在图里来回搬运，调试体验更差。

对于这个项目，我让多个业务链共享前置节点，条件边只读取已经形成的业务状态。条件函数本身不调用模型：

1
2
3
4
5
def after_monitoring(state: CountyEconomyState) -> str:
    workflow_id = state["workflow"]["workflow_id"]
    if workflow_id == "monitoring_report":
        return "expert_review"
    return "industry_diagnosis"

这让流程图能直接回答“某份报告为何经过这一步”。但它不替代业务设计。路由规则、共享节点的输入输出、每条链何时结束，仍要由懂业务的人确定。

八、证据链：不要只在提示词里写“请勿幻觉”

很多报告型 Agent 的证据管理停留在提示词：“请根据提供的材料回答，禁止编造。”这能改善输出，却不能证明输出引用了什么，也不能让程序发现材料缺失。

项目里我把证据做成独立对象。每条证据至少包含来源标识、抓取时间、原文片段、可支持的主张和适用边界。草稿节点读的是证据包，不是数据库里一大段未经整理的文本。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
def build_evidence_package(state: CountyEconomyState) -> dict:
    metrics = state["data"].get("metrics", [])
    gaps = []

    if not metrics:
        gaps.append("缺少核心指标数据")

    items = [
        {
            "source_id": item["source_id"],
            "source_url": item["source_url"],
            "retrieved_at": item["retrieved_at"],
            "quote": item["value_text"],
            "supports": ["运行监测"],
        }
        for item in metrics
    ]

    return {
        "evidence": {
            "items": items,
            "gaps": gaps,
            "status": "complete" if not gaps else "partial",
        }
    }

草稿生成时，把“允许写什么”和“必须保留什么不确定性”同时传进去：

1
2
3
4
5
只能根据 evidence.items 中的事实写作。
每个定量判断要附 source_id。
不得把 evidence.gaps 中缺失的数据补成结论。
若证据只支持相关性，不能使用“导致”“证明”等因果措辞。
把 evidence.boundary_notes 原样纳入“判断边界”小节。

提示词仍然会失效，所以它只是一层。下一层是程序检查，例如每个结论是否带来源、缺口非空时是否强制出现边界说明、草稿是否出现未经允许的数字。涉及语义一致性的检查可以再交给 LLM 评审，但必须让它输出结构化问题清单，并由代码决定是否发布。

九、Quality Gate：把硬标准写成代码

“请你严格审查这份报告是否合格”不适合承担硬门禁。模型可以帮助找遗漏，但不能作为唯一裁判。必备章节是否存在、上游报告是否已生成、预警数量与清单是否一致，这些都有明确判据，应该由确定性代码检查。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
REQUIRED_SECTIONS = {"结论", "证据来源", "判断边界"}


def review_outline(outline: str, package: dict) -> dict:
    issues = []

    missing_sections = [
        section for section in REQUIRED_SECTIONS if section not in outline
    ]
    for section in missing_sections:
        issues.append({"type": "missing_section", "message": f"缺少：{section}"})

    reports = package.get("upstream_reports", {})
    if not reports.get("monitoring_report_path"):
        issues.append({
            "type": "missing_upstream_report",
            "message": "缺少运行监测报告路径。",
        })

    context = package.get("planning_context", {})
    if context.get("warning_count", 0) > 0 and not context.get("warning_list"):
        issues.append({
            "type": "warning_incomplete",
            "message": "预警数量大于 0，但预警清单为空。",
        })

    if not context.get("boundary_notes"):
        issues.append({
            "type": "missing_boundary",
            "message": "缺少待核验事项或判断边界。",
        })

    return {"passed": not issues, "issues": issues}

门禁节点的职责不是把问题吞掉，而是把问题写回状态：

1
2
3
4
5
6
7
8
9
10
11
12
def quality_review(state: CountyEconomyState) -> dict:
    result = review_outline(
        state["draft"]["outline"],
        state["analysis"]["package"],
    )
    return {
        "review": {"planning_quality": result},
        "decisions": {
            "requires_human_review": not result["passed"],
            "publish_allowed": result["passed"],
        },
    }

这里有一个需要说清楚的边界。代码门禁适合检查可枚举的规则，不会自动判断“归因是否牵强”“政策建议是否真的回应了问题”。我会把质量控制拆成三层：

层级 适合检查什么 失败后的动作
数据契约 字段、类型、时间范围、单位 阻断并补数
确定性门禁 必备章节、引用、阈值、状态一致性 阻断、自动修订或人工复核
语义审查 论证是否跳步、建议是否可执行、措辞是否越界 产出问题清单，交人工或受限修订节点

把三层混成一个“AI 评审节点”，看起来省事，出了问题却很难定位。

十、Readiness Gate：允许估计，不等于允许下结论

模型分析链在跑回归、匹配或因果估计前有一道 readiness gate。它检查请求和数据是否具备最低条件。例如指标是否已选定、区域数量是否达到该模型族的最低要求、时间范围是否完整。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from typing import Literal


def evaluate_model_readiness(request: dict, dataset: dict) -> dict:
    missing = []
    county_codes = request.get("county_codes", [])
    metric_codes = request.get("metric_codes", [])
    periods = dataset.get("periods", [])

    if len(county_codes) < 2:
        missing.append("至少需要 2 个可比区域")
    if not metric_codes:
        missing.append("缺少待分析指标")
    if len(periods) < 2:
        missing.append("至少需要 2 个可比时期")

    status: Literal["ready", "needs_data"] = (
        "ready" if not missing else "needs_data"
    )
    return {
        "status": status,
        "missing_inputs": missing,
        "checks": {
            "county_count": len(county_codes),
            "metric_count": len(metric_codes),
            "period_count": len(periods),
        },
    }

门槛不过，条件边直接去 needs_data 或人工复核节点，不能继续产出正式结论。原因很直接：数据不足时，语言模型依然能写出流畅的因果叙述，而流畅并不增加证据。

不过，上面“2 个可比区域、2 个时期”只是示意性的准入条件，不是任何统计方法的充分条件。尤其是因果分析，至少还要单独说明处理变量、结果变量、识别假设、对照组、混杂因素、样本量与稳健性检验。没有这些，输出最多是待验证的分析假设，不能写成“政策 A 导致结果 B”。这条边界在报告中要明确展示给读者，而不是藏在系统内部。

十一、人工复核：业务暂停与 interrupt() 是两回事

原项目中的“暂停”是业务状态：任务写成 needs_human_review，人工补数或确认后，从原始请求重跑。它简单可靠，适合流程短、重跑成本可接受的场景。

LangGraph 还提供了另一种机制：在节点中调用 interrupt() 暂停图，并用 checkpointer 保存线程状态。恢复时通过相同的 thread_id 传入 Command(resume=...)。官方文档要求为中断配置持久化，并特别提醒恢复时节点会重新从头执行。Interrupts 与 Persistence

一个简化示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class ReviewState(TypedDict):
    draft: str
    approved: bool


def wait_for_reviewer(state: ReviewState) -> dict:
    decision = interrupt({"draft": state["draft"], "question": "是否发布？"})
    return {"approved": bool(decision["approved"])}


builder = StateGraph(ReviewState)
builder.add_node("review", wait_for_reviewer)
builder.add_edge(START, "review")
builder.add_edge("review", END)
app = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "report-20260403-001"}}
first = app.invoke({"draft": "待审草稿"}, config)
# 此处 first 带有中断信息，等待人工在界面上做决定
result = app.invoke(Command(resume={"approved": True}), config)

InMemorySaver 只适合演示和测试，进程退出后数据就没了。正式环境需要能跨进程保存状态的 checkpointer，并制定状态保留期限、访问权限和敏感字段脱敏策略。不要把原始业务数据、用户隐私和完整提示词不加筛选地塞进持久化状态或日志。

“暂停后恢复”也不是免费午餐。节点在 interrupt() 前做过的写入可能再次执行，所以把副作用放在中断后，或为它设计幂等键。人机协作的难点不只在 API，而在于谁有权批准、批准的是哪一版草稿、补充的数据如何审计，以及超时任务如何收尾。

十二、测试与观测：别等出问题才补

工作流不能只靠手动点几次。至少应覆盖下面四类测试：

测试 关注点 例子
节点单测 输入状态到局部更新的转换 缺指标时 build_evidence_package() 是否写入 gap
路由单测 分支是否稳定 “招商政策”同时命中时是否进入澄清而非随便选一条
图集成测试 节点顺序和终态 合法请求是否走到 publish_allowed=True
回归测试 规则变更的影响范围 更新目录后，历史样本的路径是否改变

一个路由测试可以很小：

1
2
3
4
5
6
7
8
9
10
def test_ambiguous_question_requires_review():
    catalog = [
        {"workflow_id": "policy", "trigger_words": ["政策"]},
        {"workflow_id": "investment", "trigger_words": ["招商"]},
    ]

    decision = decide_workflow("招商政策如何调整", catalog)

    assert decision["status"] == "ambiguous"
    assert decision["workflow_id"] is None

观测也不能只有 print()。每条 trace 至少要带任务 ID、节点名、开始和结束时间、输入摘要、输出摘要、规则版本、数据版本、错误码与重试次数。记录“摘要”而不是完整内容，既能排障，也能减少敏感数据泄漏。

对于模型节点，还建议记录模型名、温度、提示词模板版本、检索文档 ID 和 token 用量。这样当一份报告的表达突然改变时，你能判断是模型、提示词、数据还是路由规则发生了变化。

十三、几个容易写错的地方

1. 把状态当作可随意改写的全局变量

节点里直接 state.setdefault(...).append(...) 很方便，但会让输入和输出边界模糊，也不利于重试与测试。优先返回局部更新；确实需要复杂嵌套更新时，为那一块状态写清 reducer 或封装更新函数。

2. 以为加了 LangGraph 就自动可复现

图结构固定，只说明节点和边固定。实时数据、随机模型输出、网络重试、系统时间、外部工具版本都能改变结果。可复现是版本记录与输入快照共同达成的，不是框架单独提供的属性。

3. 用一个大节点包住整个 Agent Loop

把“检索、工具调用、分析、写作、审查”全放进一个节点，图表面上存在，实际还是黑箱。应按失败处理、权限边界、数据依赖和观测需求切分。也别走向另一个极端，纯计算拆得过细同样难维护。

4. 把 LLM 评审当硬门禁

模型适合找疑点，不适合独自批准发布。它可能漏掉一条简单规则，也可能把格式差异误判成问题。可枚举的规则交给代码；语义问题保留给模型和人。

5. 只记录最终报告，不记录中间依据

最终文本往往是最难审计的一层。要能解释“为什么得出这句话”，需要保存来源、数据快照、分析结果、门禁结果和路由版本。没有这些，trace 再长也只是日志。

十四、项目的当前边界与下一步

这套系统还有几处明确的妥协。

第一，它目前主要使用业务状态暂停和重跑，而不是原生 checkpoint 断点恢复。任务规模不大时这足够直接，代价是重复取数、重复生成，也无法保留人工在中间节点的精确修改。

第二，图内没有自动修订循环。质量门禁发现问题后，流程会标记人工复核，而不是自动把问题交给修订节点，再返回审查节点。后者可以减少重复劳动，但要先限定修订范围和最大循环次数，否则模型可能在错误方向上反复改写。

1
2
生成草稿 -> 质量审查 -> 通过 -> 发布
                    -> 不通过 -> 受限修订 -> 再审查

第三，产物还缺少完整的版本提升流程。理想情况下，每个节点的可重用结果都有版本；报告先写入草稿版本，门禁通过后才提升为正式版本；失败时保留上一个正式版本，并能用任务 ID 找回对应状态快照。

我不认为“所有 Agent 都该收紧到这个程度”。创意写作、开放式调研、头脑风暴需要探索空间，过早锁死路径反而会损失有用的分支。这里的原则只适用于结果要被比较、审计或据以决策的场景：自由度越高，越要明确它带来的风险由谁承担。

十五、结语

LangGraph 的 API 不多：状态、节点、边、编译、执行。难的部分从来不是记住函数名，而是回答这些业务问题：什么数据能进入状态，谁能修改它，哪些条件可以自动通过，哪些必须停下来找人，最终结论能追到哪份证据。

这也是我做完项目后对 Agent 的理解。Demo 往往靠让模型多试几步跑出来；需要交付的系统，则要先划清模型可以决定什么。LangGraph 给我的不是一套“更自主”的话术，而是一张能把这些边界写明白的图。

参考资料