【203篇系列】043 Agent 执行引擎
1 概念
Agent 执行引擎(AEE, Agent Execution Engine)是是一个协调大模型和外部工具的中间层系统。
定义: Agent执行引擎 = 一个能自动协调大模型与外部工具/资源进行多轮交互的系统
简单说,它把“用户提问 → 大模型思考 → 调用工具 → 获取结果 → 继续对话”这个循环自动化了。
┌─────────────────────────────────────────┐
│ Agent 执行引擎 │
├─────────────────────────────────────────┤
│ 1. 接收用户输入 │
│ 2. 构造 Prompt + 工具描述 → 大模型 │
│ 3. 解析模型输出(思考/工具调用) │
│ 4. 执行工具调用 → 获取结果 │
│ 5. 结果回填 → 继续对话 │
│ 6. 重复 2-5 直到任务完成 │
└─────────────────────────────────────────┘
2 理解
在继续更具体的内容之前,我想先归纳一下我对于 Agent的期待,继而进一步讨论其架构与实现。
Agent的两个成功的具体例子:Opencode 与 Openclaw,一个用于编程辅助,一个是电脑助手。
先谈opencode, 它用于编程辅助。使用时在对于的目录执行cli打开,然后接入一个好用点的大模型(glm4.7,minimax2.5),然后通过一些规范的输入(md)和一些结构设计,就可以让大模型非常稳定高效的帮我们把代码写好。简单来说: 抽象设计 + 规范 = 高效开发
然后是openclaw, 它是电脑助手。在你想整理文件时,通过语言告诉它,它就可以直接替你进行操作。在需要打开浏览器进行搜索等资料收集任务时,它又可以/需要通过浏览器插件来帮你执行一整套的操作。特别是你在需要持续完成一些任务时,它能够记得你之前做过什么,然后比较省心。我简单总结一下:多种技能 + 记忆 = 什么都能干
当然事实上我发现目前openclaw不那么完美(当然也一直在迭代和扩展),在辅助安装软件啥的,code buddy 要好的多。
总体上,让我们感到惊艳的地方是这些工具展示出了可以独立、高效完成具有一定复杂性的任务,这使得它们可以在更高层次上解放我们的生产力:一些具有重复性、细节性的工作可以让他们承担了。
这次的变化首先是由大模型所引发的,先是国外的几个,当时我还嫌麻烦,后来是从glm4.7出现,我觉得进度就大幅加快了。
这些大模型最大的变化是更擅长反思、类逻辑思维能力大幅提升了,我觉得这会带来:
- 1 整体规划能力变强。比如解析、重构复杂的项目能看出来,这样从基础上就提升了质量。
- 2 任务完成度变高。原来的模型在犯错时,反思容易陷入死循环,这样会导致任务完成度不高(越复杂,环节越多,越容易断链)。
- 3 细节填充度高。如果说写函数的细致程度能用分辨率类比,原来可能是720p,现在至少是2k了。
这些变化意味着,通过这些新的大模型,有可能构造出新的工具。这些工具的体验提升应该是巨大的。应该具备一些特征:
- 1 更快更靠谱 - 模式匹配
- 2 更懂你 - 图方法
- 3 更聪明 - 态势感知
- 4 能进化 - 强化学习
3 调研
一些通用的结构如下:
3.1 推理与规划层
- 1 任务分解器负责将复杂的用户请求拆解为多个可管理的子任务。
任务分解这个我之前在opencode里做过,可以让大模型按照图的方式来构建任务分解。
- 2 推理引擎是执行引擎的核心动力源,负责在每个执行步骤中进行思考和决策。推理引擎需要根据当前状态、可用工具、历史执行记录等信息,决定下一步应该采取什么行动
到这里我觉得特别像动态规划。
3.2 工具与资源管理层
-
1 工具注册与发现系统维护着一个可用的工具列表,每个工具都有其功能描述、参数规范和使用示例。当Agent需要执行特定任务时,它可以从工具库中选择最合适的工具。工具的注册需要遵循一定的规范,包括工具名称、功能描述、输入输出格式、调用示例等信息。
-
2 工具调用执行器负责实际调用外部工具或服务。这包括参数验证、请求构造、HTTP调用、响应处理、错误处理等完整流程。执行器需要处理各种可能的异常情况,如网络超时、服务不可用、参数错误等,并确保调用结果的正确传递。
-
3 API网关与认证管理处理与外部服务交互的安全性问题,包括API密钥管理、访问令牌刷新、请求签名等。对于需要认证的外部服务,执行引擎需要安全地存储和使用凭证信息,并能够处理认证过期等场景。
前面做的MCP和调用,主要就是这一层的任务。
3.3 记忆与状态管理层
记忆与状态管理层是Agent能够维护上下文一致性和持续执行任务的关键。这一层需要处理多种类型的信息:
-
1 短期记忆也称为工作记忆或上下文窗口,负责存储当前任务执行过程中的临时信息。这包括对话历史、最近的思考过程、已执行的步骤、当前可用的变量值等。短期记忆的容量受到模型上下文窗口大小的限制,现代大模型的上下文窗口已经从几千 tokens 扩展到数十万 tokens,这为Agent提供了更大的工作空间。
-
2 长期记忆存储Agent的持久化知识,包括用户偏好、历史交互总结、领域知识、工具使用经验等。长期记忆通常存储在外部数据库或向量存储中,通过检索增强生成(RAG)技术来获取相关信息。有效的长期记忆管理使Agent能够从历史经验中学习,并在后续交互中表现得更加智能。
-
3 向量存储与检索是现代Agent架构的重要组成部分,用于高效地存储和检索非结构化信息。通过将文本转换为向量嵌入,Agent可以快速找到与当前任务相关的历史信息、领域知识或参考文档。
3.4 编排与工作流层
编排与工作流层负责协调各个组件的运行,处理多步骤任务的执行流程。这一层的主要能力包括:
-
1 任务编排器管理任务的执行流程,包括任务的启动、暂停、恢复、终止等状态转换。对于复杂的多步骤任务,编排器需要维护任务状态、处理依赖关系、管理并行执行、管理异常流程等。
-
2 执行上下文管理维护整个Agent执行会话的状态信息,包括当前任务进度、已使用的资源、累计的成本、执行日志等。这些信息对于调试、性能优化和成本控制都很重要。
-
3 安全与权限控制确保Agent的操作在安全的范围内进行,包括输入验证、输出过滤、敏感信息处理、权限检查等。这对于在生产环境中部署Agent系统至关重要。
4 思路
我们期待的工具/助手:具有逻辑思维能力,可以解决问题
我觉得要从动态规划的角度来思考:让大模型自主探索,找到方案,在一次次的迭代中逐渐找到最优解。
核心价值:解决问题
我们需要工具帮我们解决问题,或者是一个简单的QA:上海现在的天气? 或者是一个复杂的问题:我想去上海旅游,给我推荐(需要查目标地点和当前地点的交通方式、目标地天气、酒店住宿或景点开放等多种信息)
我们仍然可以把这个过程假想为在对话框中一问一答的形式,问题的提出与解决都反映在对话中。但因为对话可承载的内容是无限的,所以首选需要对这种问答进行规范化(使之可以通用描述),从而可以进行有效的评估。
对话通用描述和评估
对话问答是回合制的,但总体上可以按照 T0,T1… 这样的时间序列进行描述。
回合数也是不定的,有的是因为问题本身的复杂度决定,有的则是问题在解决过程中是否顺利。
对于工具的要求,通常是在有限步内使得问题得到足够多的解决百分比。
用户可能作出提问、回答和评价;工具主要是回答,也可以适当的作出提问,以及通知。
知识图:用户在提出某个问题时,可以将之匹配到某个知识子图,提问和解答的过程就是图的路径遍历。
状态图:理论上,用户的满意度是由状态决定的,这应该是一个高维空间描述(也许可以用向量表示);更简单一点,每个时序可观察的部分是不固定的,但是我们假设隐含状态(极不满意、不满意、一般、满意、极为满意)则是可以用来描述每一个时刻的隐含状态的。
所以描述可以是:T时刻用户作出了提问(回答和评价是某个提问下的子动作),Agent在T+1 时刻给了回复,用户在T+2 时刻进行评价(或隐含评价)。
评价,特别是隐含评价特别有用,这就是Rewards。
5 单步引擎
如果要做完整的AEE工程量会比较大,考虑只做(3.1和3.2),这样的引擎是没有记忆和状态的,但应该能满足一半以上的需求。
其中3.2已经在AndybotFS里讨论过了,通过MCP可以继续拓展,比如命令行、浏览器等。主要做的调整是失败重试,以及切换到coding大模型(具有thinking,也就是解决错误的能力)
而3.1 我觉得可以通过skills 取巧。假设我们没有任何的前提,在用户提出问题后,3.1应该由大模型理解后产生一个蓝图,然后执行蓝图尝试解决问题。也许中间还要犯几次错误,逐步调整,才会成功。最终我们能得到一个正确的、可被执行的蓝图帮我们解决问题。这是比较典型的学习过程:探索(试错)-纠偏 - 达到目标。到这里,人的大脑会形成“经验”,下一次碰到相同(或者相似)的场景时,几乎就可以一次成功,不必再试错了。这个其实就是skill:可能无法做结构迁移学习,但是可以用来做模板复用。
所以我的单步引擎打算基于 claude code定义的skill规范做简单实现
5.1 Claude Code Skill
在 Claude Code 体系里:
Skill = 可被模型选择调用的结构化能力单元
它具有:
-
1 可发现(discoverable)
-
2 可选择(selectable)
-
3 可参数化(parameterizable)
-
4 可执行(executable)
-
5 可组合(composable)
本质上它是:
结构化、声明式、可调用的能力模块
5.1.1 核心组成
- 1 元数据(Metadata)
形式:
name: create_fastapi_service
description: Scaffold a FastAPI project with Docker support
tags:
- backend
- python
- fastapi
作用:
- 给 LLM 用于决策
- 用于检索与匹配
这部分要尽量剪短,而且未来也可以采用二级分类(来扩大数百倍的skill容量);甚至还可以通过es来实现几乎无限大的技能检索。
- 2 输入 Schema(参数定义)
inputs:
project_name:
type: string
required: true
use_database:
type: boolean
default: false
这是强约束部分。
Skill 不是自由文本调用,而是:结构化参数调用
这个倒是符合一贯的概念,从LangChain时代开始的好经验。我回头仔细看了下claude code skill ,发现这部分的约束可以更弱化一些,这样写skill会比较容易,因为后续的function calling(MCP) 都做了参数说明,只要大模型稍微强点应该就可以。(我用豆包的Lite,发现在语义转参数上是的确有问题)
- 3 Execution Blueprint(执行流程)
可以是:
- shell steps
- file edits
- tool calls
- MCP 调用
- 内部函数
steps:
- run: poetry new {{project_name}}
- write_file: docker-compose.yml
- if: use_database
run: add postgres service
这里,我会将多种可能的工具/资源全部统一为MCP,这样更简单且规范。
- 4 Failure Handling(可选)
Claude Code 倾向于:
-
执行失败 → 交给模型修正
-
不是 skill 内部自修
Skill 本身通常不内置复杂 retry 逻辑。
我觉得对于单步引擎来说,通过模型进行有限度的重试是足够的,skill本身是经验,不应该有结构性错误,只可能有局部细节的的typo错误。按这个假设,如果失败应该让大模型尝试重构skill,而不是一直重试。
5.1.2 执行时流程
体验上,skill的调用(如果触发)应该是一个单次输入,n次确认(n可以是0),但是流程连贯的工作流。
User Query
↓
LLM 判断是否需要 skill
↓
LLM 生成 skill 调用 + 参数
↓
Runtime 执行 blueprint
↓
结果返回 LLM
注意:
Skill 的选择权在 LLM,不在 runtime。
这是关键。
5.1.3 注册Skills
Skills 从文件的角度来看只是一个个文件,大模型是不“知道”其存在的,而要让大模型能够知道并调度,可以有三种方式:
- 1 Prompt 注入:实现很简单,这个也是之前我用的最多的方式
You have access to the following skills:
1. create_fastapi_service
Description: Scaffold a FastAPI project
Parameters:
- project_name (string, required)
- use_db (boolean)
2. analyze_csv
Description: Analyze a CSV file
Parameters:
- file_path (string, required)
每次请求调用要求模型
If a skill is appropriate, return:
{
"skill": "...",
"arguments": {...}
}
这种方式会有很多流程需要自己拼凑,特别是当skill多了会有大量的token需要被占用。但是应该也能行得通。
- 2 function calling
这应该算是一个标准做法,好处是更结构化,且很多流程已经自动化了。缺点仍然是需要大量的前置嵌入。
- 3 动态检索
通过类似动态规划的思想,增加有限轮次的交互,从而极大减少前置嵌入。大模型不必一开始就记住大量的skills,而是通过有限轮次的交互,在极大的空间内自由检索。
6 单步引擎实验
我先让大模型帮我写了一个skill
name: test_file_cp
description: |
查看下载文件夹最近7天的文件,并在 /tmp 下创建临时文件夹进行备份拷贝。
执行流程:
1. 扫描 ~/Downloads 目录,筛选最近7天修改的文件
2. 在 /tmp 下创建以时间戳命名的备份文件夹
3. 将筛选出的文件拷贝到备份文件夹
tags:
- file
- backup
- download
- copy
inputs:
download_dir:
type: string
default: "~/Downloads"
description: 下载文件夹路径,默认为 ~/Downloads
days:
type: integer
default: 7
description: 筛选最近几天的文件,默认7天
execution:
- step: 1
description: 计算时间范围
action: compute
output: time_range
formula: |
current_date = now()
min_date = current_date - timedelta(days={{days}})
min_mtime = min_date.strftime("%Y-%m-%d")
- step: 2
description: 列出下载文件夹中最近7天的文件
tool: andybot_fs.list_directory
arguments:
path: "{{download_dir}}"
pattern: "*"
recursive: false
min_mtime: "{{min_mtime}}"
output: file_list
- step: 3
description: 创建临时备份文件夹
action: compute
output: backup_dir
formula: |
timestamp = now().strftime("%Y%m%d_%H%M%S")
backup_dir = f"/tmp/download_backup_{timestamp}"
- step: 4
description: 创建备份目录
tool: andybot_fs.create_directory
arguments:
path: "{{backup_dir}}"
condition: "{{file_list.count}} > 0"
- step: 5
description: 循环拷贝文件到备份目录
tool: andybot_fs.copy_file
arguments:
src: "{{item.path}}"
dst: "{{backup_dir}}/{{item.name}}"
loop: "{{file_list.items}}"
condition: "{{item.type}} == 'file'"
- step: 6
description: 返回执行结果
action: output
result:
success: true
backup_dir: "{{backup_dir}}"
files_copied: "{{file_list.count}}"
files: "{{file_list.items | map(attribute='name') | list}}"
然后我又让大模型按照这个要求帮我写了测试脚本
参考agent_runtime.py 这个文件里的方式,帮我实现skill的调用,写在 call_skill.py这个文件里。然后进行测试。
我的要求:
- 1 整个体验是通过 messages的方式,用户提出问题,在经过一系列处理后返回结果
- 2 skill 以function calling 的方式植入
- 3 用户的问题会自然触发大模型启用这个技能
- 4 这个技能又会按照既定顺序调用mcp的若干工具
- 5 大模型完成后汇报报告结果
一些测试截图



几个测试的结论:
- 1 有瑕疵,但是可以证明假设是可行的
- 2 流程贯通: 有了基础的MCP - [针对具体的应用问题] - > 解决具体问题的skill -[技能加载]-> LLM -[根据用户问题判断]-[Y]-> 触发skill -[有序调用工具链] -> 最终结果
最需要做的不是写代码,而是根据具体的业务应用来构造MCP,而MCP本身的构造也不需要我们写代码,而是我们知道做的方式。然后根据不同的业务问题,让大模型帮我们写skill摸索后固化下来,这样就会有越来越多的流程可以被自动化。
说一下瑕疵,大模型在根据语义给到最小筛选时间过程是有问题的
...
# action: compute - 计算变量
if action == "compute":
output_var = step.get("output")
# 计算时间范围
days = self.variables.get("days", 7)
min_date = datetime.now() - timedelta(days=days)
min_mtime = min_date.strftime("%Y-%m-%d")
self.variables["min_mtime"] = min_mtime
# 计算备份目录
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_dir = f"/tmp/download_backup_{timestamp}"
self.variables["backup_dir"] = backup_dir
return {"computed": True, "min_mtime": min_mtime, "backup_dir": backup_dir}
这主要是因为大模型本身不擅长这类逻辑推理,而且大模型在运行时没有时间概念的。针对这种问题,最好是再做一个mcp来处理时间(andybot_time),因为要用于筛选。而临时文件夹倒是完全无所谓,甚至可以随机取一个名字,本身是不用计算的。
总结
接下来就需要做一个真正可用的单步引擎,这需要:
- 1 将一些必要的组件规范化。比如大模型方面,目前用豆包Lite肯定是不够的,需要把pro,还有我其他各个coding plan的接口都调试好(有更好的规划和反思能力)
- 2 将单步引擎的结构稳定下来,各个循环步骤的状态,失败时的反思重试等功能要加上。
- 3 不断扩大mcp和skill的范围,将生态建立起来。
更多推荐



所有评论(0)