Agent Loop + Graph Engineer：从状态爆炸到显式状态机的工程解法

原创

用户12339161

发布于 2026-10-13 16:26:11

550

核心结论： Agent Loop 失稳的根源不是模型能力不足，而是隐式状态爆炸——每一步的上下文、决策、工具结果全部堆叠在同一个消息数组里，导致路由条件无法收敛。Graph Engineer 的解法是把隐式 Loop 改写为显式状态机：节点定义工作性质，边定义转移条件，状态字段约束读写边界。

一、失稳的根因：隐式状态爆炸

考虑一个典型的裸 Loop：

def naive_loop(task, tools, max_steps=20):
    messages = [{"role": "user", "content": task}]
    for _ in range(max_steps):
        resp = llm.chat(messages, tools=tools)
        if not resp.tool_calls:
            return resp.content
        for tc in resp.tool_calls:
            messages.append({"role": "tool",
                             "content": execute(tc)})
    return "未完成"

这段代码的问题在 messages。信息收集的输出、推理的结论、工具报错、验证结果全部追加到同一个数组。到第 N 步时，模型看到的是全部历史——它无法判断"当前处于哪个阶段"，于是可能重复收集、跳过验证、或在推理阶段执行动作。

状态没有被显式建模，所以路由条件没有依据。

二、显式状态机：Graph 的本质

Graph Engineer 把 Loop 拆成三层结构：节点按工作性质分类，边由状态字段驱动，执行器负责调

from dataclasses import dataclass, field
from enum import Enum
from typing import Callable

class Kind(Enum):
    COLLECT, REASON, ACT, VERIFY = range(4)

@dataclass
class Node:
    id: str
    kind: Kind
    prompt: str
    tools: list[str] = field(default_factory=list)
    edges: dict[str, str | None] = field(default_factory=dict)
    max_iter: int = 5

@dataclass
class State:
    task: str
    data: dict = field(default_factory=dict)
    trace: list = field(default_factory=list)

State 只保留结构化字段，不保留原始对话。每个节点读写自己负责的字段：

READS = {
    Kind.COLLECT: [],
    Kind.REASON:  ["data.collected"],
    Kind.ACT:     ["data.hypotheses"],
    Kind.VERIFY:  ["data.action_result"],
}
WRITES = {
    Kind.COLLECT: "data.collected",
    Kind.REASON:  "data.hypotheses",
    Kind.ACT:     "data.action_result",
    Kind.VERIFY:  "data.verified",
}

三、执行器：Graph 管流转，Loop 管迭代

class Engine:
    def __init__(self, nodes: dict[str, Node], entry: str,
                 llm, tools, perm, max_steps=20):
        self.nodes, self.entry = nodes, entry
        self.llm, self.tools, self.perm = llm, tools, perm
        self.max_steps = max_steps

    def run(self, task: str) -> State:
        st = State(task=task)
        cur = self.entry
        for _ in range(self.max_steps):
            node = self.nodes[cur]
            out = self._step(node, st)
            st.trace.append((node.id, out))
            cond = self._route(node, out)
            cur = node.edges.get(cond)
            if cur is None:
                break
        return st

    def _step(self, node: Node, st: State) -> dict:
        for _ in range(node.max_iter):           # 节点内有界 Loop
            ctx = self._slice(node, st)           # 只读所需字段
            act = self.llm.decide(node.prompt, ctx, node.tools)
            if act.kind == "done":
                return self._write(node, st, act.payload)
            if not self.perm.allow(act.tool, act.args):
                return {"error": "denied"}
            result = self.tools[act.tool](**act.args)
            st.data.setdefault("_scratch", []).append(result)
        return {"error": "node_not_converged"}

    def _slice(self, node: Node, st: State) -> dict:
        out = {}
        for path in READS[node.kind]:
            keys = path.split(".")
            v = st.data
            for k in keys[1:]:
                v = v.get(k, {}) if isinstance(v, dict) else {}
            out[path] = v
        return out

    def _write(self, node: Node, st: State, payload) -> dict:
        key = WRITES[node.kind].split(".")[-1]
        st.data[key] = payload
        return payload

    def _route(self, node: Node, out: dict) -> str:
        if isinstance(out, dict) and out.get("error"):
            return "error"
        if node.kind == Kind.VERIFY:
            return "done" if out.get("ok") else "retry"
        return "ok"

三个关键机制：_slice 让每个节点只看到自己需要的字段；_step 内部的 max_iter 限制单节点迭代次数；_route 根据结构化输出决定出边，而不是让模型自由决定。

四、一张真实的状态图

graph = {
    "collect": Node("collect", Kind.COLLECT,
        "读取数据源，输出结构化概况",
        tools=["read", "query"],
        edges={"ok": "reason", "error": None}),
    "reason": Node("reason", Kind.REASON,
        "基于概况列 3 个假设",
        edges={"ok": "act"}),
    "act": Node("act", Kind.ACT,
        "对每个假设执行验证查询",
        tools=["query", "corr"],
        edges={"ok": "verify"}),
    "verify": Node("verify", Kind.VERIFY,
        "确认结论是否回答问题",
        edges={"done": None, "retry": "reason"}),  # 回边有上限
}

verify → reason 是一条回边，但受两个上限约束：节点 max_iter 和执行器 max_steps。跑满未收敛则退出并报告 node_not_converged，而不是无限循环。

五、工程要点

第一，状态即协议。 State 的字段定义就是节点间的接口契约。新增节点时先想清楚它读什么、写什么，而不是"它能干什么"。

第二，节点粒度按工作性质分。 "读文件"和"读数据库"同属 COLLECT，放一个节点；"判断异常"和"提假设"同属 REASON，放一个节点。按性质分，节点数才可控。

第三，回边必须有界。 任何 X → ... → X 的路径都需要次数上限。上限不是保护模型，是保护你的 token 账单。

第四，可观测性来自 trace。 每次执行的 (节点ID, 输出) 记录在 state.trace，回溯"为什么走这条边"时看 trace，不看对话历史。

Graph Engineer 不追求让 Agent 更聪明，它让 Agent 的每一步都可预测、可回溯、可中断。在一个会烧钱的系统里，可预测比聪明重要得多。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

reverse-engineering