Graph Engineering：在确定性边界里放置 Agent

图工程不是给 Agent 画一张漂亮的流程图，而是把系统已知的业务顺序、权限边界和失败路径写成可执行结构。语言模型擅长在信息不完整时提出假设、归纳内容、选择下一步；它不擅长天然遵守每一个固定流程。把两者混在一段自由提示词中，系统会在最需要稳定的地方变得随机。

LangGraph 的基本模型很直观：节点做事，边决定下一步，状态承载已知事实。节点可以是普通 Python 函数、一次模型调用、工具调用，甚至是内部自带循环的完整 Agent。边既可以固定，也可以依状态进行条件跳转。

先写状态，再写节点

图最容易失控的原因，是把全部信息塞进聊天消息。更好的做法是区分：可被模型阅读的消息、业务流程需要的结构化字段、以及存放在文件或数据库中的大产物。下面是一个内部知识库问答的最小状态：

from typing import Annotated, TypedDict
from operator import add

class SearchState(TypedDict):
    question: str
    route: str                 # github | docs | chat
    evidence: Annotated[list[str], add]
    answer: str
    retries: int

route、retries 这类字段是流程事实，不应该从自然语言中猜出来；evidence 使用 reducer 汇聚并行检索的结果。状态越明确，节点的输入输出就越容易测试，也越容易在暂停后恢复。

让代码决定硬边界，让模型处理软判断

以“回答一条内部技术问题”为例，分类、检索、合成看起来都是模型任务，但它们的风险并不相同。是否允许查生产数据库、是否必须经过人工审批、答案是否缺少证据，应该由代码控制。模型可以给出路由建议，条件边再把这个建议限制在允许集合中。

from langgraph.graph import StateGraph, START, END

def classify(state: SearchState):
    route = classify_with_llm(state["question"])
    return {"route": route if route in {"github", "docs", "chat"} else "docs"}

def retrieve_docs(state: SearchState):
    return {"evidence": search_docs(state["question"])}

def synthesize(state: SearchState):
    return {"answer": answer_with_citations(state["question"], state["evidence"])}

def next_after_classify(state: SearchState):
    return {"github": "github", "docs": "docs", "chat": "chat"}[state["route"]]

graph = StateGraph(SearchState)
graph.add_node("classify", classify)
graph.add_node("docs", retrieve_docs)
graph.add_node("synthesize", synthesize)
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", next_after_classify)
graph.add_edge("docs", "synthesize")
graph.add_edge("synthesize", END)
app = graph.compile()

示例省略了 GitHub 与聊天节点，但图的意图已经明确：允许模型决定“哪类信息更相关”，不允许它凭空跳过检索后直接输出答案。

生产图通常不是 DAG

真实 Agent 必须重试、等待用户补充、修订不合格答案、重复调用工具直到证据足够。因此它通常带环：验证失败回到检索，工具限流后等待重试，审批节点暂停后由外部事件恢复。把图理解为有限状态机，比把它理解成一次性流水线更接近生产现实。

def verify(state: SearchState):
    ok = has_direct_evidence(state["answer"], state["evidence"])
    return {"retries": state["retries"] + (0 if ok else 1), "verified": ok}

def after_verify(state):
    if state["verified"]:
        return END
    if state["retries"] >= 2:
        return "needs_human"
    return "docs"

环必须配上退出条件：重试上限、超时、预算和人工接管。没有上限的“再试一次”不是韧性，而是一个会安静烧钱的死循环。

动态分发让结构保持弹性

图并不要求预先写死每一条边。长文档分析、批量代码审查和多源检索常需要按输入大小动态生成 worker：先切分任务，再并发处理，最后汇聚。已知的是“会分发并汇总”，未知的是“需要几个 worker”。这种场景适合 LangGraph 的动态 Send，而不是预设十个闲置节点。

固定结构与动态分发并不矛盾：分类、权限和汇聚策略仍可确定；具体查哪些仓库、拆成多少段则由运行时状态决定。这样既保留了业务路径的可预测性，也不牺牲面对复杂输入时的弹性。

什么时候不要强行画图

如果任务本身高度开放，例如通用深度研究、探索未知代码库、跨多个模糊目标的规划，预先规定所有路径反而会把 Agent 锁死。此时更适合给一个具备文件系统、工具、记忆和验证能力的 Harness，让计划、委派与上下文管理在运行中形成。

图适合“我们已经知道哪些阶段必须发生”；Harness 适合“我们只知道目标，还不知道会经过哪些阶段”。两者并不是对立关系：一个图节点可以运行一个 Agent，一个 Agent 的工具循环也可以被图包在明确的审批和交付边界中。可靠系统的关键，是把确定性放在规则与风险处，把自主性放在探索与推理处。

检查点让暂停、恢复和回放成为能力

一张会运行很久的图不能只存在于内存里。节点执行完后把状态写入 checkpoint，系统就能在审批、限流、人工补充信息或进程故障后，从最近的安全点恢复，而不是从头再跑。更重要的是，checkpoint 提供了调试入口：我们可以比较某次异常运行在分类、检索和验证节点分别看到了什么，而不是只看最终回答。

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "ticket-1842"}}
app.invoke({"question": "支付回调为什么重复？", "retries": 0, "evidence": []}, config)

# 中断后可使用相同 thread_id 从最近状态继续
app.invoke({"human_note": "只允许查 staging 日志"}, config)

生产环境会把 checkpoint 存到持久数据库，并为状态定义版本号与迁移策略。不能只保存消息列表：审批结果、任务 ID、外部资源引用、已经执行过的写操作、预算和重试数都需要持久化，否则恢复后的图可能重复发消息、重复扣费或越过已完成步骤。

把人工介入写成一个节点

“遇到危险时问人”如果只是提示词，很容易在某个边缘路径被遗忘。图模型允许把审批建成明确节点：节点产出待确认的计划和影响范围，图暂停；外部 UI 或 webhook 写入同意、拒绝或修改意见，条件边再进入执行或结束。这样人机协作是状态迁移的一部分，而不是临时插话。

def prepare_deploy(state):
    return {"proposal": make_release_plan(state), "status": "awaiting_approval"}

def after_approval(state):
    if state.get("approval") == "approved":
        return "deploy"
    return END

审批节点应携带最小必要信息：将执行的操作、目标环境、影响对象、回滚方式、证据与过期时间。没有这些信息的“是否继续？”只会把判断负担转回人。

先为边写测试

图中的 bug 常常不在单个节点，而在错误的迁移：空检索结果仍进入合成、已被拒绝的审批仍能部署、验证失败后无限回到同一节点。测试应覆盖状态和边，而不只覆盖模型输出。模型节点可用固定 fixture 或 mock 代替，验证“给定状态必然走向哪条边”。

def test_empty_evidence_returns_to_retrieve():
    state = {"answer": "可能是缓存", "evidence": [], "retries": 0, "verified": False}
    assert after_verify(state) == "docs"

def test_second_failure_requires_human():
    state = {"verified": False, "retries": 2}
    assert after_verify(state) == "needs_human"

模型输出仍需评测，但图测试先保证系统不会把一次不可靠输出扩大为不可靠的外部动作。对于关键节点，可以记录输入摘要、选中的边、工具耗时、错误类别和状态大小；这些轨迹会暴露哪些节点反复重试、哪些路由经常误判、哪些上下文导致成本飙升。

图的粒度由可观察性决定

节点过粗时，“研究一下并写 PR”失败后很难定位是检索、设计还是提交问题；节点过细时，状态和边会淹没业务逻辑。一个实用准则是：节点应对应一个可命名、可观测、可重试的工作单元。能单独设置超时、权限、模型、缓存或人工审批的边界，通常值得成为节点。

开始时不需要绘制全公司的万能超级图。选一个流程稳定、失败成本可控的场景，例如工单分类→资料检索→草稿生成→人工发送，先把状态字段和异常边做实。图的价值不在形式，而在于让每次路径选择都能被检查、重放和改进。

当图开始变复杂时，先重看状态是否混入了职责，再重看边是否承担了业务语义。清楚的状态与少量有名字的迁移，通常比更多模型提示更能降低系统的不确定性。

还应为每个节点标记拥有者、输入来源、可写资源和失败后的恢复动作。这样图不只是运行图，也是团队讨论责任边界的共同语言；新增一个模型节点时，能立刻看出它会读到什么、会影响什么、由谁验证。