TT Lab
开始
学习 学习路径 课程

AI 智能体 — 是图不是模型

节点负责干活,状态负责记住

在 TT Lab 中继续学习

一句话总结

在智能体中,事后修改代价最高的决定是状态设计。而且状态的一半不是键的列表,而是合并规则(Reducer)。

为什么需要它

把智能体写成图之后,最先遇到的讨厌的 bug,就是经过了好几个节点,走过的路径却只剩下最后一个。

class State(TypedDict):
    trace: list          # 리듀서가 없다

def intake(state):  return {"trace": ["intake"]}
def lookup(state):  return {"trace": ["lookup"]}
def finish(state):  return {"trace": ["finish"]}

依次经过三个节点之后打开 trace 看,是 ["finish"]。不是 ["intake", "lookup", "finish"]。

不会报错。所以这个 bug 会让人先怀疑日志代码,一边问着“为什么没有留下日志”,半天就这样耗掉了。真正的原因不在日志,而在于那个键没有合并规则。

工作原理

LangGraph 的状态,是每个键对应一个通道的结构。节点不是返回整个状态,而是返回只包含想修改的键的字典,再由图按通道逐个应用。没有动过的键保持原样——所以哪怕某个节点只返回 {"answer": ...},question 也不会消失。

应用的方法就是 Reducer。Graph API overview 规定的规则只有两条。

Reducer 是接收 (지금까지의 값, 노드가 돌려준 값)(占位符依次为到目前为止的值与节点返回的值)两个参数并返回新值的普通函数。没有什么特别的。所以可以把“合并规则”集中写在代码的一个地方,那个地方就是状态定义。

def merge_sources(old, new):
    out = list(old or [])
    for item in new or []:
        if item not in out:
            out.append(item)
    return out

class State(TypedDict, total=False):
    trace: Annotated[list, operator.add]       # 이어 붙인다
    sources: Annotated[list, merge_sources]    # 중복 없이 이어 붙인다
    used_calls: Annotated[int, operator.add]   # 더한다

这里有一个实测确认的陷阱。不能直接把内置函数作为 Reducer。写成 Annotated[int, max],在编译图时会以 ValueError: no signature found for builtin max 崩溃。这是因为 LangGraph 会去查看 Reducer 的签名,而 Python 内置函数没有可查看的签名。像 def keep_max(old, new): return new if new > old else old 这样包一层就行了。

状态会增长——这就是代价

加上 Reducer 之后,又会遇到反方向的问题。trace、sources、messages 都会无休止地增长。

增长本身没关系,问题在于 checkpointer 会整个保存状态。一次运行中节点执行了 30 次,就会生成 30 个检查点,每一个都带着那个时刻的完整状态。状态大了十倍,保存量也就是十倍。

所以对会增长的键,要把上限写在 Reducer 里。

def keep_recent(old, new):
    return (list(old or []) + list(new or []))[-3:]

这样一来,节点直接写就行,上限只在一个地方守住。这比在每个节点里分散地写“太长就截断”要好——分散的话,只要漏掉一处,就会悄悄泄漏出去。

分别确定入口和出口

状态中会混进不应让外部看到的东西。用户上传的原文、中间笔记、内部评分之类。用 StateGraph(State, input=InputState, output=OutputState) 分别给出输入和输出 schema,节点能看到完整状态,外部只接收输出 schema 中的键。

如果不这样做,就得在界面一侧去删“响应里为什么附带了内部笔记”。删除的代码每增加一个新键就会落后一步,最终还是会泄漏。把对外输出的内容以白名单方式确定,总是更安全。

在现场相遇的样子

第一,走过的路径只剩一格。就是上面看到的那样。漏掉了 Reducer,症状表现为“没有留下日志”。

第二,同一个来源被写了二十遍。反复循环的智能体不断引用同一份文档,sources 就会被同样的值填满。用去重的 Reducer 取代 operator.add,问题当场解决。

第三,预算分散在各个节点里。如果统计调用次数的代码分布在三个节点里,改其中一处时,其余的就会落后。把它设为 used_calls: Annotated[int, operator.add],节点只返回 {"used_calls": 1},统计规则就只留在一个地方。

第四,检查点变大,恢复变慢。把原文整个放进状态、运行几十轮的图就会这样。原文放在外部(文件、存储),状态里只放指向它的值,会更好。

实际工作中真正重要的事

下一项实验要做什么

一步步扩展 /root/work/agstate/state.py。首先亲手复现没有 Reducer 的键被覆盖的情形,接着依次加上接续拼接的 Reducer、去重的 Reducer、只保留最大值的 Reducer、截取窗口的 Reducer。最后把输入和输出 schema 分开,收窄对外输出的键,并把这样决定的理由记录下来。评分器会真正导入你写的模块并运行,每次用不同的名称和数字直接检验 Reducer。