首页
学习
活动
专区
圈层
工具
发布
社区首页 >专栏 >大模型 Agent 的三个状态断层:计划漂移、记忆污染与恢复非幂等

大模型 Agent 的三个状态断层:计划漂移、记忆污染与恢复非幂等

原创
作者头像
IT大佬 jzit-top
发布于 2026-11-09 18:25:37
发布于 2026-11-09 18:25:37
310
举报

把大模型接进智能体,表面上只是"多了一个会调工具的 LLM",但真正改变工程性质的,是系统从此有了跨轮次的状态。普通 LLM 应用是无状态的:一次请求、一次响应、结束。Agent 则要维护计划、记忆、执行进度,这些状态在多个模型调用之间传递,任何一处被污染或错位,都不会立刻报错,只会在几步之后表现为"它怎么突然做了件莫名其妙的事"。下面三个断层分别落在计划、记忆和恢复这三个状态边界上,每一个都配有可运行的检查代码。

断层一:规划器生成的计划,与执行器的动作空间不对齐

让模型先"制定计划"再"逐步执行",是 Agent 架构中最常见的模式。问题在于,规划器输出的是自然语言步骤,执行器能调用的却是有限的工具集合。两者之间没有强制对齐机制:计划里写"查询用户最近的订单并汇总",执行器只有 query_order(order_id),没有"汇总"工具;计划里写"调用搜索接口",实际注册的工具叫 web_search。这些偏差不会被解析器拦住,因为计划本身就是一段文本,它"看起来"永远合法。

工程上的第一道防线是在计划生成之后、执行之前,做一次基于工具注册表的可执行性校验:

代码语言:javascript
复制
import json

def validate_plan(plan_steps: list[dict], registry: dict) -> dict:
    """plan_steps: [{"step": 1, "tool": "query_order", "args_hint": "..."}]"""
    errors, normalized = [], []
    for step in plan_steps:
        tool = step.get("tool")
        if tool not in registry:
            errors.append({
                "step": step.get("step"),
                "error": "unknown_tool",
                "requested": tool,
                "available": sorted(registry.keys()),
            })
            continue
        required = registry[tool]["required_args"]
        hint = step.get("args_hint", "")
        missing = [a for a in required if a not in hint]
        if missing:
            errors.append({
                "step": step.get("step"),
                "error": "missing_args",
                "tool": tool,
                "missing": missing,
            })
            continue
        normalized.append({"step": step["step"], "tool": tool})
    return {"ok": not errors, "errors": errors, "plan": normalized}

REGISTRY = {
    "query_order": {"required_args": ["order_id"]},
    "web_search":  {"required_args": ["query"]},
}

关键在于:校验失败时把 available 工具列表和 missing 参数名一起返回给规划器,让它重新生成计划,而不是让执行器在运行时抛一个 KeyError。规划器看见"你要的工具不存在,现有工具是这些",比看见"执行失败"更能自我修正。如果框架没有这一层,模型会持续生成不可执行的计划,执行器持续失败,循环在默认步数上限里耗尽预算,最后返回一个半成品。

断层二:长期记忆的写入缺少门禁,模型的自述被当成事实

Agent 的记忆系统通常分三层:工作记忆(当前上下文)、短期记忆(本轮对话历史)、长期记忆(跨会话持久化)。出问题最多的是长期记忆的写入路径。一个常见实现是每轮结束后让模型"总结本轮要点"并写入向量库,下一轮检索注入。这个流程本身没错,但它默认了一个假设:模型的总结是忠实的。实际上,模型很容易把自己的推测、承诺和意图写进总结,下一轮再检索出来时,这些内容已经变成了"已确认事实"。

一个可操作的缓解是在写入前区分事实性记忆和意图性记忆,事实性记忆必须带来源,意图性记忆不允许被当作前提注入:

代码语言:javascript
复制
from pydantic import BaseModel, Field, field_validator
from typing import Literal

class MemoryItem(BaseModel):
    kind: Literal["fact", "intent"]
    content: str
    source_turn: int
    evidence: str | None = None

    @field_validator("evidence")
    @classmethod
    def fact_needs_evidence(cls, v, info):
        if info.data.get("kind") == "fact" and not v:
            raise ValueError("fact memory must carry evidence")
        return v

def write_memory(item: MemoryItem, store: list) -> dict:
    if item.kind == "intent":
        # 意图性记忆标记为未验证,检索时降权或不注入
        store.append({**item.model_dump(), "verified": False})
        return {"written": True, "weight": 0.3}
    store.append({**item.model_dump(), "verified": True})
    return {"written": True, "weight": 1.0}

evidence 字段强制事实性记忆附带来源(原始用户发言的引用、工具返回的 ID),这让"模型说用户要退款"和"用户确实说要退款"在数据结构上可区分。verified: False 的意图性记忆在检索时可以降权,避免模型的自我承诺在后续轮次里被当成约束条件。

断层三:崩溃恢复重放整个循环,副作用被重复执行

Agent 的执行循环可能在中途崩溃:模型超时、工具异常、进程重启。恢复时如果简单地"从头再跑一遍",已经执行过的副作用(发邮件、下单、写数据库)会被重复执行。这类问题在测试环境几乎不会暴露,因为测试环境的工具通常返回固定值、失败率低;在生产环境里,它表现为"用户收到两封一样的邮件"或"订单被创建了两次"。

正确的恢复模型是检查点 + 幂等键,而不是重放整个循环。每个成功执行的副作用步骤都要落一个检查点,恢复时跳过已完成的步骤,未完成的步骤用同一个幂等键重试:

代码语言:javascript
复制
import hashlib, json, pathlib

class Checkpoint:
    def __init__(self, path: str):
        self.path = pathlib.Path(path)
        self.state = (json.loads(self.path.read_text())
                      if self.path.exists() else {"done": [], "idem": {}})

    def _key(self, step_id: str, args: dict) -> str:
        raw = f"{step_id}:{json.dumps(args, sort_keys=True)}"
        return hashlib.sha256(raw.encode()).hexdigest()[:16]

    def is_done(self, step_id: str) -> bool:
        return step_id in self.state["done"]

    def idem_key(self, step_id: str, args: dict) -> str:
        key = self._key(step_id, args)
        self.state["idem"][step_id] = key
        return key

    def mark_done(self, step_id: str) -> None:
        self.state["done"].append(step_id)
        self.path.write_text(json.dumps(self.state, ensure_ascii=False))

def run_step(step_id: str, args: dict, tool_fn, ckpt: Checkpoint):
    if ckpt.is_done(step_id):
        return {"skipped": True, "step_id": step_id}
    key = ckpt.idem_key(step_id, args)
    result = tool_fn(**args, idempotency_key=key)  # 下游按 key 去重
    ckpt.mark_done(step_id)
    return {"skipped": False, "step_id": step_id, "result": result}

mark_done 必须在工具成功返回之后执行,顺序反过来会让"检查点已写、副作用未发生"成为一个新的不一致状态。idempotency_key 由步骤 ID 和参数共同决定,意味着同一逻辑步骤用不同参数重试会被视为不同的操作——这是正确的,因为参数变了,语义就变了。真正需要防止的是同一参数下的重复执行,那由下游系统按 key 去重来保证。

这三个断层共享一个底层逻辑:Agent 的状态不是模型状态,是系统状态。计划要经过可执行性校验才能进入执行,记忆要区分事实与意图才能写入长期库,恢复要依赖检查点和幂等键而不是重放循环。模型在这些环节里负责"提出",系统负责"验证和执行"。把验证层去掉,模型会以很高的置信度驱动一条不可执行、不可追溯、不可恢复的路径——而且全程不报错。

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

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

目录
  • 断层一:规划器生成的计划,与执行器的动作空间不对齐
  • 断层二:长期记忆的写入缺少门禁,模型的自述被当成事实
  • 断层三:崩溃恢复重放整个循环,副作用被重复执行
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档