AI Agent执行功能三件套:工作记忆、任务规划与元认知监控架构

发布时间:2026/10/24 4:30:37
AI Agent执行功能三件套:工作记忆、任务规划与元认知监控架构

这两年 AI Agent 的话题有多热,相信不用我多说。但我见过太多“看起来很聪明”的 Demo:让它订个会议室能连续调用十几个工具,看起来非常丝滑;可一旦任务步骤超过五步,或者中间某个工具返回了异常,它就开始丢三落四、原地打转。问题出在哪儿?很多团队把 Agent 做成了一条“提示词 + 工具”的直线流水线,缺的不是模型能力,而是人脑里那套叫“执行功能”的管理机制。

执行功能是认知科学里的概念,简单说就是大脑负责“计划、组织、记住目标、抑制冲动”的那套管理机制。放到 AI Agent 里对应三件事:工作记忆(Working Memory)让 Agent 始终记得“当前做到哪了、还剩哪些关键约束”;任务规划(Task Planning)把模糊目标拆成可以逐步验证的子任务;元认知监控(Metacognitive Monitoring)给 Agent 装一双“回头看”的眼睛,让它知道自己做得对不对、要不要换方案。这篇文章就是我最近做了一个带执行功能的 Agent 实验项目之后的整理,方向就是标题里那套架构:基于工作记忆的任务规划与元认知监控架构。

内容偏向工程实现,会讲清楚每一块为什么要这么设计、数据结构怎么建、主循环怎么写、踩了哪些坑。适合两类人:一类是自己做 Agent 应用、总觉得现有框架不够可控的开发者;另一类是研究智能体架构、想弄清楚“除了给大模型更多上下文之外还能怎么提升稳定性”的学生或研究者。如果你正好在做多步骤工具调用的 Agent,这篇应该能给你不少可以直接照搬的设计思路。

1. 为什么 AI Agent 需要“执行功能”这堂课

1.1 现有 Agent 的通病:能说会道,但不会“做事”

最近一年我看了不少 Agent 开源项目,自己也上手改过几个,发现一个非常普遍的现象:Agent 的“口头能力”远强于“办事能力”。你让它一步步说怎么做,它头头是道;你让它真的调工具、查资料、写文件、做校验,拉通一个五步以上的完整流程,它就露馅了。

我印象很深的一次测试,是让一个 Agent 去做“整理一份智能体技术选型的调研报告”。它一开始表现很好,调用搜索、抓取页面、收藏链接,看起来忙得不亦乐乎。但跑了几分钟之后我发现,它一直在收集资料,从没想过要产出文档。原因是它的提示词里虽然写了“最终输出一份报告”,但中间没有任何机制把“输出报告”这个最终目标固定下来,每个子步骤完成之后,模型很快就忘了自己最开始要干什么。这就是典型的“目标漂移”。

还有一种常见毛病是“重复执行”。工具调用失败之后,Agent 不会去分析失败原因,只会用同一套参数再试一遍,甚至试三遍、五遍。有的 Agent 还会在一个错误路径上反复绕圈,每一步看起来都有产出,但离最终目标越来越远。这些问题其实都不是模型不够聪明,而是系统里缺少一套“执行管理”机制,也就是执行功能。它负责把“记住目标、拆解计划、监控进度、发现偏差、及时止损”这些事变成系统的一部分,而不是指望模型在每一轮对话里靠自觉完成。

1.2 执行功能的三件套:工作记忆、任务规划、元认知

我在设计这套架构时,把执行功能拆成了三个可以独立实现、又互相依赖的模块,刚好对应认知科学里执行功能的核心组成。

第一个是工作记忆。它不是单纯的“记忆存储”,而是当前任务正在使用的信息缓冲区。比如你做饭时,脑中会同时记住“锅里的菜还要炖五分钟”“盐已经放过了”“等下要关火”,这些信息不用记在长期记忆里,做完这顿饭就可以忘掉。对 Agent 来说,工作记忆就是一份动态更新的便利贴,记录目标、当前子任务、关键约束、最近结果和反思结论。它决定了每一轮模型看到什么上下文。

第二个是任务规划。人做复杂事之前会先在脑子里过一遍顺序:先烧水,再切菜,再下锅。Agent 也一样,拿到一个模糊目标后,先拆成若干子任务,标清依赖关系,再逐个执行。没有规划模块的 Agent 只能靠模型现场发挥,给它多少上下文它就发挥到多少,完全没有“长期施工图”。

第三个是元认知监控。这个听起来玄乎,其实就是一个自检循环。每一步执行完,系统都要回答几个问题:这一步成功了吗?置信度有多高?结果和目标还一致吗?要不要继续往下走,还是应该调整计划、回滚重来?这就是为什么叫“元认知”——它不是认知任务本身,而是对认知过程的监测和调控。

三个模块的分工可以类比成一个成熟的实习生:工作记忆是手里的便利贴,任务规划是待办清单,元认知监控是每完成一项就划掉一项、定期复盘进度的习惯。三者配合起来,Agent 才像一个有执行力的助手,而不是一个只会聊天的模型接口。

1.3 这套架构能解决什么问题、谁最需要它

这套架构不是给所有 Agent 都需要的。如果你只是做一个单轮问答、或者固定三步以内的工具调用,强行套上三元架构反而增加延迟和复杂度。真正需要它的是下面这类场景:一次任务要连续调用 5 个以上工具,任务中间可能失败、可能返回异常、可能需要根据中间结果调整方向,而且最后必须产出一个明确交付物。

典型场景包括:自动生成研究报告(搜索 + 抓取 + 分析 + 写作)、自动化运维操作(检查状态、定位问题、执行修复、验证结果)、数据处理流水线(读数据、清洗、建模、导出报表),以及任何需要长期多步规划的智能体应用。我做实验时是在一个完全虚构的“模拟项目 X”环境里跑的,但设计思路可以直接移植到真实业务上。

这套架构的收益并不仅仅在“成功率”上。它更大的价值是让 Agent 的行为变得可追踪、可解释、可干预。因为工作记忆里记录了每个决策依据,监控器会输出每个步骤的置信度和决策理由,所以当 Agent 出问题时,你打开日志就能知道它是在哪一步、因为什么原因开始跑偏的,而不是面对一个黑盒抓瞎。这一点对生产环境来说,比多跑通几个任务重要得多。

2. 架构总览:三个模块如何各司其职又咬合在一起

2.1 主循环:从目标到执行的完整数据流

这套架构的运行主体是一个主循环,我把它设计成下面这样的顺序流程,每一步都有明确的输入和输出,不依赖模型临场发挥。

第一,用户给出目标,这个目标首先被写进工作记忆,作为最高优先级的 goal 条目。此时 Agent 可能会有很多额外信息,比如用户偏好、环境限制、可用的工具列表,也一并写入工作记忆里的 constraint 和 state 条目。

第二,规划器读取当前工作记忆,把目标拆成一棵任务树。任务树的每个节点都有一个 id、一段描述、依赖关系和明确的完成标准(success_criteria)。这个完成标准非常关键,后面监控器就是靠它来判断“做没做成”的。

第三,执行器从任务队列里取出一个“当前可执行”的任务,调用对应的工具或代码模块,拿到原始结果。这里有一个容易被忽略的点:执行器返回的结果不能只是字符串,要带结构化信息,包括返回码、摘要、耗时、置信度等。

第四,结果写回工作记忆。执行摘要存成 result 条目,任务状态从 active 改成 done 或 failed。如果执行失败,失败原因也要写进去,方便后续反思。

第五,元认知监控器读取刚写入的工作记忆,结合任务完成度和置信度给出决策。决策只有四种:继续执行下一个任务、重新规划剩余任务、回滚到更早的干净状态、停下来问用户。监控器的输出是整个循环的刹车和方向盘,没有它,前两步只是机械执行而已。

第六,根据监控器的决策回到第三步继续迭代,或者进入“反思通道”,把反思结果写回工作记忆后再重跑一遍。如此循环,直到工作记忆里的 goal 条目变成 done,或者需要用户干预为止。

这套主循环看起来简单,但它把 Agent 的每一次行为都变成了系统里可观测的状态转移。你不会再看到模型漫无目的地发挥,因为每一步都被工作记忆和监控器约束住了。

2.2 为什么不能靠无限加长上下文来硬扛

你可能想问:为什么不直接把所有历史记录一股脑塞进上下文,让大模型自己根据长文本去规划?这个思路确实有很多团队在用,我也试过,但实践下来有几个绕不开的问题。

首先是成本问题。每次请求都把全部历史包含进去,token 消耗是线性增长的,跑一个长任务的钱足够你买好几顿午饭。其次是注意力稀释问题。上下文越长,模型越容易“抓不住重点”,尤其是你真正想让它记住的那条目标和几个关键约束,很容易淹没在大量历史对话里。我在实验里对比过,同样是让 Agent 执行一个 8 步任务,把完整历史塞进上下文时,它反而在第三步就开始忘了原始目标;而用一个精炼的工作记忆去喂它,目标漂移的情况少很多。

最要命的是“不可回滚”。如果所有历史都在上下文里,一旦某一步执行错了,你很难把状态退回到出错之前。模型可能已经在后续对话里被污染,产生了一堆基于错误前提的推理。工作记忆则天然支持回滚,因为每一步的中间结果都以结构化的方式存在外部模块里,你想回退到哪一步,直接把对应的状态快照恢复就行,完全不影响模型本身的参数。

所以我才把工作记忆单独拎出来做模块,而不是继续依赖上下文窗口。上下文窗口仍然是模型单次读取的缓冲地带,但装什么进去、装多少、不装什么,由工作记忆来决定。

2.3 分层边界与模块通信方式

为了让这套架构不乱成一锅粥,我在实现时严格分了层。最底下是感知层,负责接受用户输入、工具返回值、环境状态;中间是认知层,也就是工作记忆、任务规划、元认知监控这三个核心模块;最上面是执行层,负责实际调用工具、运行代码、调 API。

层与层之间通过结构化消息通信。比如执行器返回的不是一段散装字符串,而是一个包含 returncode、summary、confidence、raw_output_path 的对象;规划器输出的也不是自然语言草稿,而是带有 JSON Schema 校验过的任务树。这样做的好处有两个:一是监控器可以直接读取结构化字段做判断,而不需要再对文本做一轮解析;二是在出问题的时候,整个通信链路都是可回放、可追溯的,你能精确知道是哪一层出现了逻辑错误。

我在做“模拟项目 X”的早期版本时,偷懒用过“把所有信息拼成一段话直接丢给模型”,后来改造成结构化消息之后,整个系统的可观测性提升了一个量级,排查问题的时间至少缩短了一半。

3. 工作记忆模块:给 Agent 一张永远带在身上的便利贴

3.1 工作记忆到底是什么,不是什么

工作记忆这个概念经常被人误解成长期数据库,或者干脆等同于上下文窗口,其实都不是。长期数据库是用来存“以后可能用得上”的信息的,比如知识库、历史对话记录;上下文窗口是模型单次能看到的 token 范围;而工作记忆,是“当前正在处理的事情所需要的信息子集”,它强调的是主动维护、容量有限、随时刷新。

我习惯把它理解成桌面上那张便利贴。你不可能把整本笔记本都贴在桌面上,你只会贴当前这个任务需要提醒自己的三四件事。其他东西放在抽屉里(长期记忆),想查的时候再拿出来。工作记忆也是一样,它不应该塞下所有历史,而只应该维持当前目标和行动所需的少量关键信息。

这个区别决定了设计方向。长期记忆要考虑检索效率和存储量,工作记忆要考虑的是“什么该留下、什么该遗忘”。如果工作记忆里塞了太多信息,它就会被噪声淹没;如果少了,模型就容易丢失上下文。所以工作记忆模块的核心不是存储技术,而是信息筛选和维护策略。

3.2 数据结构设计:一条可追溯的 WM Entry

在实现里,我把工作记忆设计成一张带优先级的临时记录表,每条记录称为一个 WM Entry。每个 Entry 都有明确的类型和状态,监控器和规划器可以直接通过类型筛选它们关心的条目。

我用的是这样一组字段,你可以直接把这个结构搬到你自己的项目里。

字段名类型含义
eidstring条目唯一 ID,例如 wm_001
etypestring条目类型,goal / subtask / state / constraint / result / reflection
contentstring条目的核心内容,尽量精简
statusstringactive / done / failed / discarded
confidencefloat该条目的置信度,0 到 1,由执行器或监控器写入
priorityint优先级,1 到 5,越高越重要
created_atfloat创建时间戳
last_accessfloat最近访问时间戳,用于刷新和淘汰策略
sourcestring条目来源,planner / executor / monitor / user
extradict扩展字段,比如 success_criteria、工具名、失败原因

举个例子,规划器生成一个子任务后,工作记忆里可能出现这样一个条目:eid 是 task_003,etype 是 subtask,content 是“调用搜索接口获取 TOP10 行业报告”,status 是 active,priority 是 4,extra 里记录了 success_criteria 是“返回至少 5 条有效链接且标题与目标相关”。

关键设计点在于,每个条目都要带置信度和来源。这样监控器在判断时不用只靠模型空口自评,它可以调取某个 result 条目的 confidence 字段,看看那个内容到底是谁、在什么条件下写入的。置信度低的内容,在组装提示词时会排在更靠后的位置,甚至干脆不进入当轮上下文。

3.3 遗忘与刷新:容量控制是设计题,不是后遗症

工作记忆最容易被忽略的是遗忘机制。很多人一开始只顾着往里写,写着写着就变成一个变相的完整日志,失去了“当前焦点”的意义。我在早期版本里就犯过这个错,结果工作记忆里堆满了十几个 result 条目,模型反而分不清哪个是最新状态。

后来我加了三层控制。

第一层是容量上限。我给工作记忆设置了 max_entries 参数,默认 50 条。超过上限后,新增条目会触发淘汰逻辑:优先淘汰 status 为 done 或 discarded、且 priority 较低、最近没被访问过的条目。这模拟了人脑“做完了就翻篇”的机制。

第二层是刷新。每一条被读取的 Entry,last_access 都会更新。这样在淘汰时,最近用不上的旧条目会更早被清掉,而仍在关键链路上的条目会被保留。刷新策略不需要很复杂,只要保证“最近正在用的信息不被误删”就行。

第三层是衰减。我并不是简单保留所有条目,而是会在每轮主循环结束时,对 result 和 reflection 类条目做一次置信度衰减,比如乘以 0.95。如果后续执行结果和这条记录矛盾,它的置信度会进一步下降。这样模型在组装上下文时,会自然更信任更新、更一致的信息。

工作记忆还需要支持快照和恢复。我会在每次执行器运行之前,把当前工作记忆序列化成 JSON 存一份,文件名带上时间戳。一旦监控器决定回滚,直接把对应快照重新加载进来,再让模型从那个时间点继续。这个功能在调试时价值巨大,因为它让你能无限重放、反复观察 Agent 在某一步之后是如何决策的。

4. 任务规划模块:把模糊目标变成可执行的检查清单

4.1 规划器怎么拆任务:LLM 生成 + 规则校验

规划器的职责,是接收一个模糊目标和工作记忆快照,输出一棵结构清晰的任务树。在这里我选择让 LLM 负责生成拆解草案,因为它的语义理解能力强,能处理各种领域的目标;但我不让 LLM 的生成结果直接进入执行器,而是先过一道规则校验。

我会在调用 LLM 时,要求它严格按 JSON Schema 输出任务树。Schema 大概是这样的:一个 goal 字段,加一个 subtasks 数组,每个子任务包含 id、description、dependencies、success_criteria。dependencies 用来描述任务之间的先后关系,success_criteria 是后面判断成败的依据。之所以必须带 success_criteria,是因为执行器不能只“做了事”就算成功,必须有一个可验证的完成标准。

生成之后,规则校验会检查几件事:每个子任务是否有非空的 success_criteria;description 里是否明确提到要调用哪个工具或写哪个文件;子任务之间是否形成了 DAG(有向无环图),避免循环依赖;以及目标本身是否出现在任务树的根上。如果校验不过,我不会让模型重写整个方案,而是把校验错误信息回传,让它只修正有问题的部分,这样省 token,结果也更稳定。

我在这套架构里反复强调的是:规划器不是生成一次就完了。它需要根据执行反馈反复调整剩余任务。所以规划器的另一个输入,是当前已经完成和失败的任务列表,这样它就知道哪些可以保留、哪些需要重新设计。

4.2 任务表达与执行循环:从任务树到任务队列

任务树生成之后,如何驱动执行?我采用的是“依赖解析 + 就绪队列”的方式。每个子任务记录它依赖哪些前置任务,只有前置任务全部处于 done 状态,这个子任务才会进入可执行队列。执行器每次从队列头部取一个任务,执行完更新工作记忆,再重新计算就绪队列。

子任务的粒度是规划器里一个拿捏点。太粗的子任务一步做不完,内部其实还藏着多个隐含步骤,出错了不好定位;太细的子任务数量爆炸,规划开销高,也容易让模型失去全局视角。我自己的参考标准是:一个子任务应该能在一到两次工具调用内完成,并且有非常明确的结束信号。

比如目标是“整理一份智能体技术选型的调研报告”,规划器生成的子任务可能长这样:收集资料、提取关键选型标准、生成对比表、撰写报告正文、校验引用来源、导出最终文档。每个子任务的 success_criteria 包括“链接数量不少于 5”“表格包含至少 4 个选型维度”“报告正文引用数量不少于 3”这类可量化条件。这样的粒度,让监控器在每一步都能清楚判断“这步到底成没成”。

执行循环是这套架构里最简单、也最容易被写歪的部分。我在实现时只用了一个 while 循环,每次从就绪队列里拿任务,交给执行器,然后把结果写回工作记忆,交给监控器评估。如果监控器说继续,就进入下一个任务;如果它说要重规划,就回到规划器生成新的剩余任务列表。整个循环看起来很简单,但正因为简单,才不容易出错。

4.3 什么时候触发重规划:动态调整的四个信号

规划不是一次定死,执行过程中百分之百会遇到需要调整的情况。我在监控器里预设了四个触发重规划的信号,你也可以根据自己的场景增减。

第一个是连续失败。某个子任务连续失败达到重试上限(我默认 2 次),说明很可能不是工具临时抖动,而是这个方案本身就不可行。这时候继续重试没有意义,必须让规划器重新设计这个子任务或调整后续依赖。

第二个是新信息冲突。执行过程中发现一个之前没料到的约束,比如工具返回的数据结构和预期不一致,或者用户临时增加了需求。这类新信息写进工作记忆后,如果与现有 goal 或 constraint 冲突,监控器就会触发重规划。

第三个是前置假设不成立。任务树是基于初始假设生成的,如果执行结果证明某个关键假设错了,比如“目标网站无法访问”,那么后面所有依赖于它的任务都需要重新设计。

第四个是置信度持续偏低。我不想等到任务失败才反应,如果连续几个子任务返回的置信度都低于阈值,说明 Agent 已经进入了一个不稳定的区域,应该主动停下来重新校准路径。

重规划有一个容易踩的坑:不要让规划器从零开始。更好的做法是,把已完成的任务、已获得的关键结果、当前阻塞点一起传给规划器,让它只重新设计失败点之后的剩余任务。这样既保留已付出的努力,也避免模型在重规划时“推翻重来”导致资源浪费。

5. 元认知监控模块:给 Agent 装上仪表盘和后视镜

5.1 监控什么:不只是看“成没成功”

元认知监控模块是三个模块里最容易被低估的。很多人觉得监控就是判断任务成没成功,成功了继续,失败了重试,这就够了。但真正跑起来会发现,很多失败并不是非黑即白的。工具返回了结果,但结果可能是空列表、可能是错误格式、可能压根不相关。所以监控器至少要盯四类信息。

第一类是进度指标。当前完成了多少个子任务,剩余任务之间有没有依赖阻塞,整个任务是否还走在关键路径上。如果 Agent 在一个分支任务上花掉了大量调用次数,却迟迟不影响主线推进,监控器就要警惕“跑偏”。

第二类是执行质量。每个子任务结果的置信度,工具调用的耗时,失败重试的次数。这些指标能反映当前方案是否健康。

第三类是状态一致性。工作记忆里是否存在互相矛盾的条目。比如一个 result 说“接口返回正常”,但另一个 constraint 写了“该接口存在认证问题”,这显然不一致。监控器需要能识别这类冲突,或者至少把冲突标记出来。

第四类是资源使用。调用次数是不是异常增长?token 预算是不是快超了?反思循环是不是反复在同一块打转?这些都是监控器应该统计的“成本信号”。

在实际实现里,我并不是让监控器每一步都调用一次完整 LLM 来思考,那样太慢了。我的策略是:先用规则和工具返回值做快速判断,只有快速判断无法确定的时候,才启动一次轻量级的 LLM 评估。比如工具返回码非零,直接判负,不需要模型参与;返回结果通过了 success_criteria 里的数量校验,直接判过。这样既快又稳。

5.2 置信度自评与反思循环

置信度自评是元认知的核心抓手。做法是,在每次执行器返回结果时,额外附一个 0 到 1 的 confidence 值,表示这次执行结果在多大程度上达到了预期。这个值可以由执行器直接给出,也可以由监控器用一个轻量 LLM 调用来判断。我在实验里发现,单靠模型自评往往偏乐观,所以更好的办法是结合规则校验一起打分。

比如一个搜索任务,要求返回 5 条有效链接。如果模型自评 0.95,但工具实际返回了 3 条,我就会把置信度直接调低到 0.5,因为这明显没达到 success_criteria。置信度低于阈值的条目,会进入反思通道。

反思通道是一个结构化的反思循环。监控器会把目标和当前状态、执行步骤、失败结果、失败可能原因、下一步建议组合成一段反思 Prompt,让模型输出一个包含“结论”和“建议动作”的 JSON,动作包括重试、回滚、修改计划、询问用户。反思结果会作为 reflection 条目写回工作记忆,并保留 2 到 3 轮,防止模型在下一步又做出和反思结论矛盾的动作。

反思循环最大的危险是“越改越乱”。如果模型每次都基于错误前提做反思,反思结果会不断互相矛盾。所以我强制要求反思 Prompt 里带上历史反思记录,并加一条提示:“如果本次反思结论与上次反思相同,说明需要改变方向而不是重复执行。”这个细节帮我在实验里减少了很多无效循环。

5.3 “继续还是停下”的门控机制

监控器的最终输出,是一个决策,我把它设计成四选一。continue 表示当前状态健康,进入下一个任务;replan 表示当前完成任务方案有问题,需要规划器调整剩余计划;rollback 表示当前状态已经被污染,需要回滚到更早的快照重新执行;ask_user 表示目标歧义或缺少关键信息,需要停下来问用户,不要自作主张。

这个门控机制的意义,是给 Agent 装上一个“刹车”。很多 Agent 之所以表现得失控,就是因为没有任何机制在“跑错方向”的时候踩刹车。有了门控之后,我发现 Agent 的很多错误行为早早就被拦截了,不会等到最后交付一个完全跑偏的结果。

为了安全,我还会给门控加一个“全局最大重试次数”保护。比如整个任务最多只允许重试 10 次,超过之后强制转为 ask_user。这个保护在调试中很有用,避免 Agent 在无人看管时无限烧掉 token。

5.4 监控信号速查表

下面这张表是我在实现中总结的监控信号和处理动作,基本可以覆盖大多数多步骤工具调用场景。

触发信号可能原因处理动作
连续 2 次子任务失败方案与工具不匹配触发 replan,重新设计失败点之后的任务
置信度持续低于 0.6上下文缺失或目标存在歧义启动反思,必要时询问用户
结果与工作记忆中的约束冲突中间状态污染回滚到最近的干净快照
工具返回空数据或重复内容接口变化或搜索策略局限切换工具或调整搜索参数
反思结论连续两次相同模型陷入思维死循环强制改变方案,禁止重复建议
整个任务重试次数超过全局上限目标本身不可行或环境问题停住并请求用户决策

这张表不是死的,我会根据场景不断补充。它的价值在于,让“监控”从一句空话变成了可枚举、可测试、可扩展的具体规则。

6. 实操过程与核心实现:一个最小可运行的 Agent 骨架

6.1 工作记忆与任务数据类的代码实现

理论讲再多,不如直接看代码。下面是我在实验项目里用的一个最小实现骨架,去掉了业务细节,保留了核心结构。你可以直接抄走,替换成自己的工具调用逻辑。

# wm.py from dataclasses import dataclass, field from typing import Dict, List import time @dataclass class WmEntry: eid: str etype: str # goal / subtask / state / constraint / result / reflection content: str status: str = "active" # active / done / failed / discarded confidence: float = 0.0 priority: int = 3 created_at: float = field(default_factory=time.time) last_access: float = field(default_factory=time.time) source: str = "planner" # planner / executor / monitor / user extra: Dict = field(default_factory=dict) class WorkingMemory: def __init__(self, max_entries=50): self.entries: Dict[str, WmEntry] = {} self.max_entries = max_entries def add(self, entry: WmEntry): entry.last_access = time.time() self.entries[entry.eid] = entry self._evict_if_needed() def get(self, eid: str): e = self.entries.get(eid) if e: e.last_access = time.time() return e def update(self, eid: str, **kwargs): e = self.entries.get(eid) if e: for k, v in kwargs.items(): setattr(e, k, v) e.last_access = time.time() def active_subtasks(self) -> List[WmEntry]: return [e for e in self.entries.values() if e.etype == "subtask" and e.status == "active"] def to_prompt(self, max_chars=4000) -> str: # 按优先级降序、最近访问时间降序排序,截取前 N 条 items = sorted(self.entries.values(), key=lambda x: (x.priority, x.last_access), reverse=True) parts = [] used = 0 for it in items: s = f"[{it.etype}|{it.status}] {it.content}" if used + len(s) > max_chars: break parts.append(s) used += len(s) return "\n".join(parts) def snapshot(self) -> str: import json data = {eid: e.__dict__ for eid, e in self.entries.items()} return json.dumps(data, ensure_ascii=False) def _evict_if_needed(self): if len(self.entries) <= self.max_entries: return candidates = [e for e in self.entries.values() if e.status in ("done", "discarded")] candidates.sort(key=lambda x: x.priority) while len(self.entries) > self.max_entries and candidates: victim = candidates.pop(0) self.entries.pop(victim.eid, None)

这个 WorkingMemory 类看起来不复杂,但注意几个细节。to_prompt 方法直接决定了模型每轮看到什么,所以它必须做信息筛选,而不是无脑 dump。_evict_if_needed 只淘汰已完成或已丢弃的低优先级条目,active 的任务永远不会在容量满时被误删,这是安全底线的考虑。

6.2 规划器、监控器与主循环的代码实现

接下来是规划器和监控器的实现。这里的 LLM 调用我做了抽象,你接任何大模型服务时,只要把 json_schema 参数换成对应接口的写法就行。

# agent_core.py from typing import List, Dict, Any class Planner: def __init__(self, llm): self.llm = llm def plan(self, goal: str, context: str) -> List[Dict[str, Any]]: # 示意:要求模型输出 JSON 任务树,并强制带 success_criteria task_tree = self.llm.chat( messages=[ {"role": "system", "content": "你是任务规划器,只输出 JSON 任务树。"}, {"role": "user", "content": f"目标:{goal}\n当前状态:{context}"} ], json_schema={ "type": "object", "properties": { "subtasks": { "type": "array", "items": { "type": "object", "properties": { "id": {"type": "string"}, "description": {"type": "string"}, "dependencies": { "type": "array", "items": {"type": "string"} }, "success_criteria": {"type": "string"} }, "required": ["id", "description", "success_criteria"] } } } } ) return task_tree["subtasks"] class MetaMonitor: def __init__(self, confidence_threshold=0.6, max_retries=2): self.confidence_threshold = confidence_threshold self.max_retries = max_retries self.failed_counts: Dict[str, int] = {} self.global_retries = 0 def evaluate(self, task: Dict[str, Any], result: Any, wm) -> str: # 规则优先:工具返回码非零直接判负 if result.returncode != 0: self.failed_counts[task["id"]] = self.failed_counts.get(task["id"], 0) + 1 self.global_retries += 1 if self.global_retries > 10: return "ask_user" if self.failed_counts[task["id"]] >= self.max_retries: return "replan" return "retry" if result.confidence < self.confidence_threshold: return "reflect" wm.add(WmEntry(eid=f"result_{task['id']}", etype="result", content=result.summary, status="done", confidence=result.confidence, source="executor")) return "continue"

主循环的写法是这样的。这里我简化了依赖解析,假设任务已经按拓扑顺序排列好,实际项目里需要根据 dependencies 字段做就绪队列管理。

def run_agent(goal: str, planner: Planner, executor, monitor: MetaMonitor, wm: WorkingMemory): wm.add(WmEntry(eid="goal", etype="goal", content=goal, priority=5, source="user")) task_list = planner.plan(goal, wm.to_prompt()) for task in task_list: wm.add(WmEntry(eid=task["id"], etype="subtask", content=task["description"], priority=4, extra={"criteria": task["success_criteria"]})) for attempt in range(monitor.max_retries + 1): result = executor.execute(task) decision = monitor.evaluate(task, result, wm) if decision == "continue": wm.update(task["id"], status="done") break if decision == "replan": new_tasks = planner.plan( goal, wm.to_prompt() + "\n注意:之前的方案失败,需要调整剩余任务。" ) task_list = new_tasks break if decision == "ask_user": print("需要用户介入:", result.error_summary) return # retry / reflect 时继续循环,让执行器再次尝试

这段代码故意写得朴素,就是为了让结构清晰。真实项目中,我会把依赖解析、快照保存、重试延迟、日志记录都加进去,但核心骨架就是上面这几句话。你拿到之后,替换自己的 executor 和 llm 接口,就能跑起一个带工作记忆和元认知的最小 Agent。

6.3 关键参数配置建议

参数调优是我花了不少时间的地方,一开始全凭感觉,后来总结出一套还算稳的默认值。下面这张表是实验中的推荐配置,你可以当作起点,再根据任务类型调整。

参数建议值说明
confidence_threshold0.6 - 0.7低于该值触发反思或回滚,太高容易频繁打扰,太低失去监控意义
max_retries2单个子任务的最大重试次数
wm.max_entries30 - 50工作记忆容量上限,超出后淘汰低优先级已完成条目
to_prompt 截断长度3000 - 5000 字符保证模型每轮看到的上下文精炼,不被历史堆满
规划器 JSON 模式开启强制结构化输出,减少解析错误
反思保留轮数2 - 3保留最近几轮反思结论,防止模型反复横跳

置信度阈值这个参数最敏感。设成 0.5 太低,模型经常带着低质量结果继续往下跑;设成 0.9 太高,Agent 频繁反思,任务迟迟推进不下去。0.6 到 0.7 之间是个比较稳的区间,适合大多数工具调用场景。如果你跑的任务本身充满不确定性,比如开放式调研,可以适当降到 0.55;如果是代码生成这种对正确性要求高的任务,建议到 0.7 以上。

6.4 模拟实验:有监控和无监控的差别

我在“模拟项目 X”里跑过一次对比实验,用的是一套 8 步的“资料收集、信息分析、报告生成”任务,每个配置跑 5 次。实验规模不大,不算严谨的 benchmark,但趋势非常明显。

无监控版的 Agent,就是单纯“LLM + 工具调用”,没有工作记忆、没有规划器、没有元认知。它在 5 次运行里成功了 2 次,平均工具调用次数是 45 次,失败原因包括:忘了最终要产出文档、重复执行同一个搜索请求、在错误路径上反复尝试。

有监控版的 Agent,直接用上面这套架构,成功 4 次,平均工具调用次数 32 次。失败的那 1 次,是工具返回了一个完全不符合 schema 的格式,重试耗尽后没有及时触发 replan。后续我在规则校验里加了一条“如果工具返回格式异常,直接跳过重试进入 replan”,这个问题也解决了。

这个对比说明,监控和规划带来的提升不只是“多成功几次”,更重要的是减少了无效调用。少了 13 次调用,意味着成本更低、速度更快、系统更可控。如果你的 Agent 已经开始运行业务,这套架构省下的 token 费用大概率能覆盖开发成本。

7. 常见问题与排查技巧实录

7.1 我踩过的四个坑

第一个坑,工作记忆塞太满。我最早实现时觉得,既然是记忆,那能记多少记多少,结果 to_prompt 里堆了几十条 result,模型反而抓不住重点。后来我把 result 条目全部改成“摘要 + 存储路径”,完整结果写到外部日志,工作记忆里只留一句话。效果立刻提升,上下文变干净,目标漂移几乎消失。

第二个坑,反思循环自我 PUA。有一次实验里,Agent 因为连续失败进入了反思通道,结果模型越反思越悲观,把原本还能用的中间结果也否定了,最后把一份好好的报告改成了一堆负面评价。解决方法是两层:一是每次反思必须附带至少一个“下一步具体行动”,禁止输出纯粹的批评;二是保留最近 2 到 3 轮反思记录,如果新一轮反思结论和上一轮一样,强制要求改变行动方向。

第三个坑,任务粒度不一致。我发现规划器生成的任务树经常出现“一个任务里藏了五个步骤”的情况,出错之后监控器不知道具体是哪一步出了问题。后来我在规划器提示词里明确写了任务粒度的标准:“一个子任务必须能通过一次工具调用或一段短代码完成,且 success_criteria 必须包含可量化的检查点。”加上这个约束之后,任务质量稳定很多。

第四个坑,置信度自评不可靠。模型给自己的结果打分普遍偏高,尤其是开放式任务。我不再把 LLM 的置信度当作唯一标准,而是用工具返回码、字段数量、关键字匹配这些规则去校准。比如搜索接口返回空数组,不管模型自评 0.95,我直接把置信度置成 0。校准之后,元认知监控的决策明显更准。

7.2 问题排查速查表

这套架构调试起来其实不麻烦,只要把关键日志打全。下面是我整理的排查表,基本都是日志里能一眼看出来的症状。

症状可能环节排查思路修复动作
Agent 重复执行同一个任务工作记忆检查子任务状态是否及时更新为 done执行成功后立刻更新状态并写入 result
上下文像滚雪球一样越来越长工作记忆观察 to_prompt 是否包含过多 result 和细节限制 result 数量,只保留摘要
越反思越离谱,结论互相矛盾元认知监控查看 reflection 条目历史保留反思轮数上限,增加回滚快照
置信度全线飘高,监控不起作用元认知监控检查置信度来源是否过于依赖模型自评引入工具返回值、规则校验校准置信度
失败后直接瘫痪,不继续也不重规划任务规划检查主线循环是否有 replan 分支添加连续失败计数和 replan 触发
重规划结果和原来的方案几乎一样任务规划检查 planner 是否拿到了新增的失败信息在重规划输入中明确加入失败原因和阻塞点

这六条是我实际排查中使用频率最高的。你遇到诡异行为时,先看日志里的决策链,再看工作记忆快照,基本都能定位到是哪个环节出了逻辑问题。

7.3 几条实战经验总结

最后分享几条我反复用了很久的经验,属于那种“不写在文档里、但每次都能救命”的东西。

第一,先跑通人工监控版,再自动化。我不建议一上来就实现完整的监控器。你可以先只保留工作记忆和规划器,执行器每跑几步打印一次当前状态,由你自己扮演元认知,观察哪些信号真正能预测失败。等你找到规律了,再把这些信号写进监控器的规则里。这样比凭空设计参数靠谱得多。

第二,工作记忆从设计那天就要留好“删除”和“遗忘”的口子。哪怕第一版只有一个 add 方法,也要把 update、delete、snapshot 这些接口先定义出来。后面加遗忘策略时,你会发现这些基础接口缺一不可。别到时候为了加遗忘策略,回头重构整个数据结构。

第三,所有关键动作打结构化日志。我用的格式是 JSON Lines,每行包含 timestamp、task_id、action、confidence、trigger、wm_snapshot 路径。排查问题的时候,直接把日志丢进命令行按 task_id 过滤,就能完整重放一个任务的决策链。没有这套日志,所谓“可解释性”就是空谈。

第四,参数一次只改一个。调 confidence_threshold 的时候就不要同时改 max_entries,改完跑同样的任务集,把成功率和平均调用次数记下来。如果每次都同时动三四个参数,出了效果提升你也不知道是哪个参数起的作用,出了问题更没法定位。

最后再分享一个实际操作中的小技巧

这套架构跑通之后,我最大的体会是:别贪心。我见过很多人一上来就做超级复杂的记忆系统,工作记忆、情景记忆、语义记忆全上,结果 Agent 根本跑不动,调试也痛苦。我自己做完这套“工作记忆 + 任务规划 + 元认知监控”的三件套之后,第一件事就是把工作记忆的条目数量砍到 30,把反思轮数限制在 2,跑通一个真正的 5 步以上任务,再慢慢加复杂度。

现在每接到一个新 Agent 需求,我都会先问一句:这个任务超过五个工具调用了吗?如果没超过,我根本不上监控模块,直接用最朴素的提示词解决。如果超过了,再把这套架构搬出来。做 Agent 的乐趣在于,你永远能找到一个让它在具体任务里更可控的细节,而这个细节往往不在模型参数里,而在系统设计上。