buddyMe 架构深度解析：从 AgentMain 九步构建到任务拆解执行

admin · 2026-05-13 · architecture · agent · pipeline · 深度解析 · 任务拆解

buddyMe 架构深度解析：从 AgentMain 构建到任务拆解执行

01 项目全貌

buddyMe 是一个多模型智能体框架，核心源码在 buddyMe/ 目录下，采用分层解耦设计：

agent_moudle/ 编排层：AgentMain + TaskPipeline + TaskRunner

agent.py — 生命周期管理 + 依赖注入入口
task_pipeline.py — 子任务 Prompt 构建 + LLM 工具调用循环
task_runner.py — 三阶段管线编排器
todo_manager.py — LLM 任务拆解（plan_task）

anthropic_standard/ 适配层：多 provider LLM 客户端

basic_anthropic_client.py — OpenAI 兼容客户端 + 重试
unified_client.py — 自动选协议（OpenAI / Anthropic）
basic_anthropic_tool.py — BaseTool ABC + ToolExecutor

tool_moudle/ 工具层：7 个内置工具

bash_tool.py — Bash, ReadFile, WriteFile, EditFile, Grep, Glob
baidu_search_tool.py — 千帆 AI 搜索
invoke_skill_tool.py — Skill 桥接

context/ 上下文层

prompt_builder.py — System Prompt 分层组装
brain_loader.py — Brain 文件加载器
compressor.py — 中段对话压缩

memory/ 记忆层

store.py — MemoryStore：评分 / 衰减 / 整合
manager.py — MemoryManager 门面
extractor.py — LLM 结构化记忆提取

session/ 会话层

state.py — SessionState 共享可变状态
subtasks_manager.py — subtask_results.json 读写
conversation_logger.py — 对话日志持久化

cmd_library/ 命令层

builtin/ — /help, /model, /memory, /log ...

initspace/brain/ 人格层：SOUL → IDENTITY → AGENT

skill_library/skills/ 技能库：30+ 可热加载技能

整个系统的设计原则是：依赖注入，反向通道归零。TaskPipeline 和 TaskRunner 不持有 AgentMain 引用，所有依赖通过构造函数注入。两者共享同一个 SessionState dataclass，各自往里写 used_tools、token_in/out、written_files。

02 AgentMain 九步依赖注入

AgentMain.__init__() 在 agent.py:47-171 中完成全部组件的创建和注入。按执行顺序拆解为九步：

Step 1 — 基础配置

设定 max_steps=11（每个子任务最多 11 轮工具调用）、max_messages_length=20（消息历史上限）。创建 MessageHistory(20)。

Step 2 — LLM 客户端

通过 basic_llm.create_client() 工厂创建主力模型客户端（如 DeepSeek、GLM）和子任务客户端 "sub_agent_code_plan"。
子任务 token 预算自动设为主力的 1/4：_sub_agent_max_token = _agent_max_token // 4。

Step 3 — 目录系统

三级目录：_PACKAGE_DIR（源码模板只读）、_DATA_DIR（~/.buddyme/ 用户数据）、_WORKSPACE_DIR（当前工作目录，文件输出目标）。

Step 4 — 工具层

创建 ToolExecutor，注册 6 个基础工具（BashTool, ReadFileTool, WriteFileTool, EditFileTool, GrepTool, GlobTool）+ 1 个 InvokeSkillTool（桥接 SkillLoader）。

Step 5 — System Prompt 构建

PromptBuilder 加载 brain 目录下的 SOUL → IDENTITY → AGENT 文件，拼接工具 schema 和 Skill 元数据，生成完整的 system prompt。详见下一节。

Step 6 — 记忆 + 对话日志

MemoryStore 管理 USER.md 的读写和评分衰减；ConversationLogger 按 JSON 持久化对话记录；SubtasksManager 管理子任务间数据桥。

Step 7 — 上下文压缩器

ContextCompressor(keep_first=2, keep_last=5)：当子任务执行中消息超过 20 条时，保留首 2 条 + 尾 5 条，中段调 LLM 生成结构化摘要替代。

Step 8 — 任务执行引擎

创建 SessionState 共享状态 → 注入 15 个参数构建 TaskPipeline → 注入 10 个参数构建 TaskRunner。三者共享同一个 _state 引用。

Step 9 — 命令系统

CommandRegistry 注册 /help、/model、/memory、/log、/skill、/reset、/exit 等斜杠命令。在进入 LLM 之前拦截，节省 token。

注入关系一览

AgentMain ├── _client (主力 LLM) ─── 注入 ──→ TaskPipeline ├── _sub_client (子任务 LLM) ─── 注入 ──→ TaskPipeline ├── _executor (7 个工具) ─── 注入 ──→ TaskPipeline ├── _skill_loader ─── 注入 ──→ TaskPipeline + TaskRunner ├── _memory_store ─── 注入 ──→ TaskPipeline ├── _compressor ─── 注入 ──→ TaskPipeline ├── _state (SessionState) ─── 共享 ──→ TaskPipeline + TaskRunner ├── _prompt_builder ─── 注入 ──→ TaskPipeline ├── _subtask_mgr ─── 注入 ──→ TaskRunner ├── conv_logger ─── 注入 ──→ TaskRunner └── _pipeline ─── 注入 ──→ TaskRunner 关键约束：TaskPipeline 和 TaskRunner 不持有 AgentMain 引用，只持有注入的具体依赖。两者通过 SessionState 共享可变状态。

03 System Prompt 的分层组装

System prompt 是 LLM 行为的"宪法"。PromptBuilder.build() 按固定层次组装，由 prompt_builder.py:37-67 实现：

System Prompt 组装顺序 L0 SOUL.md 人格内核 — 专业、务实、高效；结论先行、结构化表达 L1 IDENTITY.md 角色身份 — 专业编程与个人助手；5 大服务标准 L2 AGENT.md 工作规则 — 工具优先级、安全边界、记忆管理 Memory 用户记忆 — 从 USER.md 提取的活跃记忆片段 Skills 技能元数据 — 30+ 技能的名称+简介（~100 tokens/技能） Tools 工具指南 — 7 个工具的 schema + 调用规范

每层职责清晰：SOUL 几乎不变，IDENTITY 切换角色时替换，AGENT 定义操作合同。子任务执行时，还会额外注入 SUB_AGENT.md（安全规则 + 执行规范），并通过 .format(max_steps=11, max_output=98304) 填入运行时参数。

04 三阶段执行管线

用户输入进入 AgentMain.invoke() 后，经过命令拦截，交由 TaskRunner.run() 执行。这是一个三阶段管线：

三阶段管线

Phase 1 — 上下文构建 + 任务分类

_build_conversation_context() 读取 memory_summary.md 获取近期摘要。
is_simple_task() 检查关键词（"创建""制作""html" 等）和文本长度，简单任务走 run_simple() 快速路径。

Phase 2 — LLM 任务规划

plan_task() 调用主力 LLM，将用户需求按 [SEARCH]、[CREATE]、[EDIT]、[VERIFY]、[SKILL:xxx] 标签拆解为最多 8 个执行步骤。
结果写入 subtask_results.json，全部 status="pending"。

Phase 3 — 逐个执行子任务

对每个子任务：classify_subtask() 判断类型 → build_subtask_system/user() 构建专属 Prompt → pick_tools() 按类型筛选工具 → run_sub_task() 进入 LLM+工具循环。
全部完成后 _merge_results() 合并输出。

子任务的三种类型

类型	判定条件	可用工具	行为约束
`research`	不含 VERIFY/BUILD 关键词	全部 7 个工具	只输出搜索/整理结果，不创建文件
`build`	含 CREATE/EDIT 等关键词	6 个（无 Bash）	骨架 → 填充，优先使用前置结果
`verify`（end_task）	最后一个子任务	4 个（read/edit/write/skill）	禁止搜索，只能读+修

run_sub_task 内部循环

每个子任务最多执行 11 轮（max_steps），每轮流程：

run_sub_task 循环（最多 11 轮） ┌─── Step N ──────────────────────────────────────────┐ │ │ │ _client.chat(messages, tools=tool_schemas) │ │ │ │ │ ▼ │ │ 解析 response: │ │ ├── text → 追加到 full_text │ │ └── tool_calls → 逐个执行 │ │ │ │ │ ├── 搜索次数检查 (max 5) │ │ ├── _executor.execute(name, input) │ │ ├── 结果追加到 messages │ │ └── 消息长度检查 → 可能触发压缩 │ │ │ │ 无 tool_calls → 返回 full_text ✅ │ │ 达到搜索上限 → 强制总结返回 🛑 │ └──────────────────────────────────────────────────────────────┘

05 实战：故宫网页导览的完整旅程

让我们用一个具体例子走完全流程。用户输入：

用户输入

"帮我做一个北京故宫的网页导览，包含景点介绍、开放时间、交通路线"

阶段 0 — 命令拦截

invoke() 首先调用 cmd_registry.dispatch()。输入不是 / 开头，返回 None，继续执行。

阶段 1 — 任务分类

_build_conversation_context() 读取 memory_summary.md，首次对话无历史，返回空字符串。

is_simple_task() 检查关键词：命中"制作"和"网页"等 → 返回 False，判定为复杂任务。

阶段 2 — LLM 任务规划

plan_task() 发送规划 prompt 给 LLM，要求按标签拆解。LLM 返回：

LLM 生成的 7 步计划

SEARCH 搜索故宫各景点介绍、历史背景、开放时间、门票价格
SEARCH 搜索故宫交通路线、地铁公交、自驾停车信息
CREATE 创建 HTML 骨架，包含页面结构、导航栏、各板块容器
EDIT 填充景点介绍板块（太和殿、乾清宫、御花园等）
EDIT 填充开放时间、门票、交通路线板块
EDIT 添加 CSS 样式（中式配色、响应式布局）
VERIFY 读取最终文件，检查完整性、链接有效性

SubtasksManager.create() 将这 7 步写入 subtask_results.json，第一条标记 "start_task"，最后一条标记 "end_task"。

阶段 3 — 逐个执行子任务

子任务 1/7 — 搜索景点信息 research

Prompt 构建

System: "当前子任务: [SEARCH] 搜索故宫各景点介绍... 只完成这一个子任务。只输出搜索/整理结果，不要创建文件。"
User: "请完成以下任务: [SEARCH] 搜索故宫各景点介绍... 背景信息: 帮我做一个北京故宫的网页导览..."
工具: 全部 7 个

LLM + 工具循环

Step 1: LLM → baidu_search("故宫景点介绍太和殿乾清宫") → 返回搜索结果
Step 2: LLM → baidu_search("故宫开放时间门票价格 2026") → 补充信息
Step 3: LLM 不再调工具 → 返回文本总结
结果：故宫主要景点包括太和殿(皇帝登基)、乾清宫... 开放时间: 旺季8:30-17:00...

subtask_mgr.update() 将结果写入 JSON：status="completed"。

子任务 2/7 — 搜索交通路线 research

这次 build_subtask_user() 检测到 idx > 0，自动调用 subtask_mgr.read_completed() 读取子任务 1 的结果，作为"前置子任务已完成的结果"注入。LLM 看到已有景点信息，专注搜索交通。

子任务 3/7 — 创建 HTML 骨架 build

Build 类型的关键差异

System Prompt 包含：
【文件构建策略】先用 write_file 创建骨架 → 再用 edit_file 填充
【已生成的文件】(暂无)
【前置子任务结果】子任务1+2 的搜索结果全文

工具集：read_file, write_file, edit_file, grep, glob, baidu_search, invoke_skill

LLM 执行 write_file("index.html", "<!DOCTYPE html>...骨架...")。写入成功后，系统追加提示 "[系统] 文件已成功写入，当前子任务已完成。"

子任务 4-6/7 — 填充内容 + 样式 build

后续 build 子任务的 System Prompt 中，【已生成的文件】更新为 - index.html。LLM 用 edit_file 逐步填充景点介绍、交通信息到 HTML 对应位置，用 write_file 创建 style.css。

子任务 7/7 — 最终验证 end_task

End Task 的关键约束

System Prompt：
"当前子任务: 最终验证与修复"
"1. 用 read_file 读取已生成的文件"
"2. 检查结构完整性"
"3. 发现问题用 edit_file 修复"
"5. 禁止重新搜索"

工具集（仅 4 个）：read_file, edit_file, write_file, invoke_skill
没有 baidu_search，没有 grep！只能读 + 改。

LLM 执行 read_file("index.html") → 检查 HTML 闭合、内容完整性 → 如有问题 edit_file 修复 → 输出验证结果。

阶段 4 — 结果合并

_merge_results() 取最后一个子任务（验证）的结果作为主体，前 6 个子任务转为摘要列表追加：

验证完成，index.html 结构完整，包含导航栏、5 个景点介绍... --- 已完成所有子任务： - [SEARCH] 搜索故宫各景点介绍...: 已完成 - [SEARCH] 搜索故宫交通路线...: 已完成 - [CREATE] 创建 HTML 骨架...: 已完成 - [EDIT] 填充景点介绍板块...: 已完成 - [EDIT] 填充开放时间、交通...: 已完成 - [EDIT] 添加 CSS 样式...: 已完成

阶段 5 — 收尾

回到 AgentMain.invoke()，完成三件事：

1. 文件列表

_state.written_files → ["index.html", "style.css"]
追加输出: "项目已生成到:\n - index.html\n - style.css"

2. 对话日志

conv_logger.log(query, result, model="deepseek", tool_calls=_state.used_tools, extra={execute_cost_time, used_skills, subtask_count=7, episode})

3. 记忆同步

_memory.sync_turn(user_input, result[:500]) — 更新相关记忆的活跃时间

06 数据在子任务间的流转

子任务之间是隔离的——每个子任务都有独立的 messages 列表和 LLM 对话上下文。那么前一个子任务的成果如何传递给下一个？答案在 subtask_results.json：

数据流转机制 subtask_results.json 是子任务间的唯一数据桥梁写入时机：每个子任务完成后 → subtask_mgr.update(task_text, result) → 将 status 改为 "completed"，存入 result 文本读取时机：下一个子任务构建 Prompt 时 → subtask_mgr.read_completed() → 遍历 JSON，拼接所有 status="completed" 的条目 → 格式: "【[SEARCH] 搜索故宫景点...】\n故宫主要景点包括..." 注入位置： System Prompt → "【前置子任务结果】" 段落（build 类型） User Prompt → "前置子任务已完成的结果" 段落（research 类型）约束：非 end_task 的结果会被截断到 max_result_len end_task 结果保留全文（作为最终输出） LLM 被明确禁止修改此文件

除了 subtask_results.json，还有一条隐式数据通道：SessionState.written_files 列表。每当 LLM 调用 write_file 成功时，文件路径被追加到这个列表，后续子任务的 System Prompt 中【已生成的文件】段落就会更新。

07 安全、压缩与记忆

安全机制

安全不是事后补丁，而是贯穿整个管线的约束：

层级	约束	实现位置
工具级	BashTool 拦截危险命令模式	`bash_tool.py`
Prompt 级	SUB_AGENT.md：禁止修改 subtask_results.json	`SUB_AGENT.md`
执行级	end_task 关闭搜索和 grep 权限	`pick_tools()`
搜索级	每子任务最多 5 次搜索	`max_search_calls=5`
人格级	AGENT.md：破坏性操作需确认	`AGENT.md §6`

上下文压缩

单个子任务可能经历 11 轮工具调用，积累大量消息。ContextCompressor 在消息数超过 max_messages_length=20 时触发：

压缩策略

原始消息序列（可能 30+ 条）： [0] system prompt ← 保留 (keep_first=2) [1] user message ← 保留 [2] assistant reply ← 压缩区 [3] tool result ← 压缩区 [4] assistant reply ← 压缩区 ... [25] tool result ← 压缩区 [26] assistant reply ← 保留 (keep_last=5) [27] tool result ← 保留 [28] assistant reply ← 保留 [29] tool result ← 保留 [30] assistant reply ← 保留压缩区 → 调用 _sub_client 生成结构化摘要： [Key Decisions Made] ... [Files Created/Modified] ... [Important Findings] ... [Pending Items] ...

记忆系统

记忆分两层：短期（memory_summary.md，近 5 日对话摘要）和长期（USER.md，用户画像与偏好）。

记忆生命周期 1. 提取 MemoryExtractor 从对话日志中提取结构化信息 ↓ 2. 去重 SequenceMatcher 相似度 > 0.8 → 跳过更新 ↓ 3. 评分综合分 = 相关性(0.4) + 重要性(0.3) + 新鲜度(0.3) ↓ 4. 衰减得分 < 0.4 → 归档；得分 < 0.2 → 清理 ↓ 5. 整合含"上次""曾经"等关键词的碎片 → 合并到"历史摘要"

08 总结

buddyMe 的架构可以总结为三条主线：

1. 依赖注入，反向通道归零

AgentMain 是唯一的全知对象，它将具体依赖注入 TaskPipeline 和 TaskRunner，后者只知道自己需要的组件，不持有 AgentMain 引用。SessionState dataclass 作为共享可变状态的唯一通道。

2. 三阶段管线 + JSON 数据桥

分类 → 规划 → 逐个执行。subtask_results.json 是子任务间唯一的数据流转通道：每个子任务完成后写入结果，下一个子任务构建 Prompt 时读取并注入。end_task 关闭搜索权限，确保最终阶段只做验证。

3. 分层 System Prompt + 上下文压缩

SOUL → IDENTITY → AGENT 三层人格模板，加上记忆、技能、工具的动态注入。长对话中通过 ContextCompressor 保留首尾关键消息、压缩中段为结构化摘要，保证 LLM 在有限 token 内始终拥有最关键的上下文。

完整源码见 buddyMe/agent_moudle/agent.py、task_pipeline.py、task_runner.py。本文基于 buddyMe v0.1.35，核心架构在可见的未来会保持稳定。

附：完整时序图

用户: "帮我做一个北京故宫的网页导览..." │ ▼ AgentMain.invoke() ├── cmd_registry.dispatch() → 不是命令 ├── _state.reset_episode() ├── _memory.prefetch("故宫网页导览") → 记忆文本注入 system prompt │ ▼ TaskRunner.run() │ ├── plan_task() → LLM 拆解为 7 步 │ ├── _subtask_mgr.create(7步) → 写入 subtask_results.json │ │ │ ├── [1/7 SEARCH] 景点信息 → baidu_search ×2 → 文本结果 → update JSON │ ├── [2/7 SEARCH] 交通路线 → baidu_search ×2 → 文本结果 → update JSON │ ├── [3/7 CREATE] HTML骨架 → write_file("index.html") → update JSON │ ├── [4/7 EDIT] 填充景点 → edit_file ×3 → update JSON │ ├── [5/7 EDIT] 填充交通 → edit_file ×2 → update JSON │ ├── [6/7 EDIT] 填充样式 → write_file("style.css") → update JSON │ └── [7/7 VERIFY] 最终检查 → read_file ×2 → 无问题 → 验证通过 │ ▼ AgentMain.invoke() 收尾 ├── _merge_results → 返回最终文本 ├── conv_logger.log() → 持久化对话 ├── _memory.sync_turn() → 记忆同步 └── return "验证完成，index.html 结构完整..."