| Harness 组件 | 作用 | 对应系列文章 |
|---|---|---|
| 工具 + 执行循环 | 让模型能执行代码、读写文件 | 第一篇：工具 + 循环 |
| 记忆 + 规划 | 让模型能记住过去、分步完成复杂任务 | 第二篇：记忆与规划 |
| System Prompt + Rules + Skills + MCP | 注入知识、约束行为、扩展工具 | 第三篇：Rules、Skills、MCP |
| 子 Agent 生成 | 把任务委派给专门的子智能体 | 第四篇：SubAgent |
| 多 Agent 编排 | 持久团队、通信通道、生命周期管理 | 第五篇：Teams |
| 上下文压缩 / Compaction | 对抗 context window 限制 | 第六篇：上下文压缩 |
| 安全防线 + 执行钩子（Hook） | 黑名单、用户确认、输出截断、可插拔管道 | 第七篇：安全与权限 |
┌─────────────────────────────────────────────────────┐
│ Harness │
│ │
│ ┌─────────────┐ ┌──────────┐ ┌────────────────┐ │
│ │ Rules │ │ Skills │ │ MCP Tools │ │
│ │ (第三篇) │ │ (第三篇) │ │ (第三篇) │ │
│ └──────┬──────┘ └────┬─────┘ └───────┬────────┘ │
│ └──────────────┼────────────────┘ │
│ ▼ │
│ ┌──── System Prompt + 工具列表 ────┐ │
│ │ │ │
│ │ Memory (第二篇) │ │
│ │ Compaction (第六篇) │ │
│ │ │ │
│ └────────────┬──────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────┐ │
│ │ ┌─────────┐ │ │
│ │ │ Model │ │ ← 模型只管思考和决策 │
│ │ │ (裸模型) │ │ │
│ │ └─────────┘ │ │
│ └────────────┬────────────┘ │
│ ▼ │
│ ┌──── Hook 管道 (第七篇) ────┐ │
│ │ 黑名单 → 用户确认 → 执行 │ │
│ └────────────┬───────────────┘ │
│ ▼ │
│ ┌──── 工具执行层 (第一篇) ────┐ │
│ │ bash / read / write / edit │ │
│ └────────────┬─────────────────┘ │
│ ▼ │
│ ┌──── 协作层 (第四、五篇) ────┐ │
│ │ SubAgent / Teams / 通信 │ │
│ └──────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────┘Agent = LLM + 工具 + 循环定义
tools = [
{
"type": "function",
"function": {
"name": "execute_bash",
"description": "Execute a bash command on the system",
"parameters": {
"type": "object",
"properties": {
"command": {"type": "string", "description": "The bash command to execute"}
},
"required": ["command"]
}
}
},
# ... read_file, write_file 类似
]工具具体实现
def execute_bash(command):
try:
result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
return result.stdout + result.stderr
except Exception as e:
returnf"Error: {str(e)}"
def read_file(path):
try:
with open(path, 'r') as f:
return f.read()
except Exception as e:
returnf"Error: {str(e)}"
def write_file(path, content):
try:
with open(path, 'w') as f:
f.write(content)
returnf"Successfully wrote to {path}"
except Exception as e:
returnf"Error: {str(e)}"函数映射
available_functions = {
"execute_bash": execute_bash,
"read_file": read_file,
"write_file": write_file
}对应Claude Code的核心工具集。
base_tools = [
{"name": "read", "description": "Read file with line numbers", ...},
{"name": "write", "description": "Write content to file", ...},
{"name": "edit", "description": "Replace string in file", ...}, # 新增
{"name": "glob", "description": "Find files by pattern", ...}, # 新增
{"name": "grep", "description": "Search files for pattern", ...}, # 新增
{"name": "bash", "description": "Run shell command", ...},
{"name": "plan", "description": "Break down complex task", ...} # 新增
]edit： 旧串必须在文件中出现1次。
def edit(path, old_string, new_string):
try:
with open(path, 'r') as f:
content = f.read()
if content.count(old_string) != 1:
return f"Error: old_string must appear exactly once"
new_content = content.replace(old_string, new_string)
with open(path, 'w') as f:
f.write(new_content)
return f"Successfully edited {path}"
except Exception as e:
return f"Error: {str(e)}"read：行号 + 分页
def read(path, offset=None, limit=None):
try:
with open(path, 'r') as f:
lines = f.readlines()
start = offset if offset else0
end = start + limit if limit else len(lines)
numbered = [f"{i+1:4d}{line}"for i, line in enumerate(lines[start:end], start)]
return ''.join(numbered)
except Exception as e:
return f"Error: {str(e)}"流程图：AgentLoop -> 自主调用工具
用户任务
│
▼
┌──────────────────────────────────────────────────┐
│ Agent Loop │
│ │
│ ┌─────────┐ ┌──────────┐ ┌──────────────┐ │
│ │ 发送给 │───▶│ LLM 决策 │───▶│ 有tool_call? │ │
│ │ LLM │ │ │ └──────┬───────┘ │
│ └─────────┘ └──────────┘ │ │
│ ▲ Yes │ No │
│ │ ┌─────┴─────┐ │
│ │ ▼ ▼ │
│ ┌────┴────────┐ ┌──────────┐ 返回文本 │
│ │ 结果追加到 │◀─────────│ 执行工具 │ ──────▶ │
│ │ messages │ └──────────┘ 结束 │
│ └─────────────┘ │
└──────────────────────────────────────────────────┘核心代码：messages为短期记忆
def run_agent(user_message, max_iterations=5):
messages = [
{"role": "system", "content": "You are a helpful assistant that can interact with the system. Be concise."},
{"role": "user", "content": user_message}
]
for _ in range(max_iterations):
# Step 1: 把完整对话历史 + 工具列表发给 LLM
response = client.chat.completions.create(
model=os.environ.get("OPENAI_MODEL", "gpt-4o-mini"),
messages=messages,
tools=tools
)
message = response.choices[0].message
messages.append(message)
# Step 2: 如果 LLM 没有调用工具 → 任务完成，返回文本回答
if not message.tool_calls:
return message.content
# Step 3: 如果 LLM 要调用工具 → 逐个执行，把结果追加到对话历史
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
print(f"[Tool] {function_name}({function_args})")
function_response = available_functions[function_name](**function_args "function_name")
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": function_response
})
return"Max iterations reached"Agent = LLM + 工具 + 循环 + 记忆无持久记忆prompt中注入历史信息存储
加载
记忆注入
塞进System Prompt记忆存储
MEMORY_FILE = "agent_memory.md"
def save_memory(task, result):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
entry = f"\n## {timestamp}\n**Task:** {task}\n**Result:** {result}\n"
try:
with open(MEMORY_FILE, 'a') as f:
f.write(entry)
except:
pass示例：
## 2026-03-12 14:30:00
**Task:** 统计当前目录下的 Python 文件数量
**Result:** 当前目录下共有 42 个 Python 文件。
## 2026-03-12 15:00:00
**Task:** 创建一个 hello.py
**Result:** 已创建 hello.py，内容为打印 Hello World。记忆加载：滑动加载
def load_memory():
if not os.path.exists(MEMORY_FILE):
return""
try:
with open(MEMORY_FILE, 'r') as f:
content = f.read()
lines = content.split('\n')
return'\n'.join(lines[-50:]) if len(lines) > 50 else content
except:
return ""记忆注入：System Prompt
def run_agent_plus(task, use_plan=False):
memory = load_memory()
system_prompt = "You are a helpful assistant that can interact with the system. Be concise."
if memory:
system_prompt += f"\n\nPrevious context:\n{memory}"
messages = [{"role": "system", "content": system_prompt}]第 1 次运行 第 2 次运行
─────────── ───────────
用户: "创建 hello.py" 用户: "读取 hello.py 并加上注释"
│ │
▼ ▼
system prompt: system prompt:
"You are a helpful "You are a helpful
assistant..." assistant...
Previous context:
## 2026-03-12 14:30
Task: 创建 hello.py
Result: 已创建..."
│ │
▼ ▼
Agent 执行任务 Agent 执行任务
│ (知道之前创建过 hello.py)
▼ │
save_memory() ──写入──▶ agent_memory.md ◀─── save_memory()Agent = LLM + 工具 + 循环 + 记忆 + 规划先想再做多个子步骤，再单独去执行子步骤Break down the task into 3-5 simple, actionable steps. Return as JSON array of strings
def create_plan(task):
print("[Planning] Breaking down task...")
response = client.chat.completions.create(
model=os.environ.get("OPENAI_MODEL", "gpt-4o-mini"),
messages=[
{"role": "system", "content": "Break down the task into 3-5 simple, actionable steps. Return as JSON array of strings."},
{"role": "user", "content": f"Task: {task}"}
],
response_format={"type": "json_object"}
)
try:
plan_data = json.loads(response.choices[0].message.content)
steps = plan_data.get("steps", [task])
print(f"[Plan] {len(steps)} steps created")
for i, step in enumerate(steps, 1):
print(f" {i}. {step}")
return steps
except:
# plan失败
return [task] agent.py (ReAct) agent-plus.py (Plan-then-Execute)
思考 → 行动 → 观察 规划（全局思考）
↑ │ │
└─────────┘ 步骤1 → 步骤2 → 步骤3
(每步内部仍是 ReAct)[Planning] Breaking down task...
[Plan] 3 steps created
1. 使用 grep 递归搜索所有 TODO 注释
2. 整理搜索结果为 Markdown 清单格式
3. 将清单写入 todo.md
[Step 1/3] 使用 grep 递归搜索所有 TODO 注释
[Tool] execute_bash({"command": "grep -rn 'TODO' --include='*.py' ."})
→ ./app.py:23: # TODO: add error handling
→ ./utils.py:7: # TODO: refactor this function
→ ./main.py:45: # TODO: add logging
找到 3 处 TODO 注释。
[Step 2/3] 整理搜索结果为 Markdown 清单格式
（LLM 看到步骤 1 的 grep 结果，直接整理，无需再次搜索）
已整理为以下清单：
- app.py:23 - add error handling
- utils.py:7 - refactor this function
- main.py:45 - add logging
[Step 3/3] 将清单写入 todo.md
[Tool] write_file({"path": "todo.md", "content": "# TODO List\n\n..."})
已将 TODO 清单写入 todo.md。messages 作为上下文，由外部传入给到下一个步骤def run_agent_step(task, messages, max_iterations=5):
messages.append({"role": "user", "content": task})
actions = []
for _ in range(max_iterations):
response = client.chat.completions.create(
model=os.environ.get("OPENAI_MODEL", "gpt-4o-mini"),
messages=messages,
tools=tools
)
message = response.choices[0].message
messages.append(message)
ifnot message.tool_calls:
return message.content, actions, messages
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
print(f"[Tool] {function_name}({function_args})")
function_response = available_functions[function_name](**function_args "function_name")
actions.append({"tool": function_name, "args": function_args})
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": function_response})
return "Max iterations reached", actions, messages {
"name": "plan",
"description": "Break down complex task into steps and execute sequentially"
}防止在规划中再次规划，陷入死循环了。if function_name == "plan":
plan_mode = True
function_response = available_functions[function_name](**function_args "function_name")
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": function_response})
if current_plan:
results = []
for i, step in enumerate(current_plan, 1):
messages.append({"role": "user", "content": step})
# 关键：排除 plan
result, messages = run_agent_step(
messages,
[t for t in tools if t["function"]["name"] != "plan"]
)
results.append(result)
plan_mode = False
current_plan = []
return "\n".join(results), messages用户: "重构项目的测试框架"
│
▼
LLM 判断任务复杂 → 主动调用 plan 工具
│
▼
plan() 返回 4 个步骤
│
├── Step 1: 分析当前测试结构
│ └── run_agent_step() → [glob, read, grep]
│
├── Step 2: 创建新的测试目录
│ └── run_agent_step() → [bash, write]
│
├── Step 3: 迁移现有测试文件
│ └── run_agent_step() → [read, edit, write]
│
└── Step 4: 验证所有测试通过
└── run_agent_step() → [bash]system prompt = 基础指令 + Rules(项目规则) + Skills(技能描述) + Memory(历史记忆)
| 未解问题 | 解决方案 | 新概念 |
|---|---|---|
| 工具是硬编码的 | 外部配置文件动态加载工具 | MCP（Model Context Protocol） |
| 没有行为约束 | 声明式规则文件注入 prompt | Rules + Skills |
| 规划是被动的 | 把规划注册为 Agent 可自主调用的工具 | Plan-as-Tool |
背景
要求不同。写一次规则文件，永久生效。本质
项目级的SystemPrompt 扩展。声明式文件，定制Agent的行为边界。nanoAgent Rules实现
.agent/rules/文件夹下的markdown文件其他实现
CLAUDE.md.cursorrules.github/copilot-instructions.mdRules加载
RULES_DIR = ".agent/rules"
def load_rules():
rules = []
ifnot os.path.exists(RULES_DIR):
return ""
try:
for rule_file in Path(RULES_DIR).glob("*.md"):
with open(rule_file, 'r') as f:
rules.append(f"# {rule_file.stem}\n{f.read()}")
return "\n\n".join(rules) if rules else ""
except:
return ""Rules注入
if rules:
context_parts.append(f"\n# Rules\n{rules}") <!-- .agent/rules/code-style.md -->
- 使用 Python 3.10+ 语法
- 所有函数必须有 docstring
- 变量命名使用 snake_case
- 不要使用 print 调试，使用 logging 模块<!-- .agent/rules/safety.md -->
- 永远不要执行 rm -rf / 或类似的危险命令
- 修改文件前先备份
- 不要修改 .env 文件中的密钥核心作用
存储格式
Skills 加载
SKILLS_DIR = ".agent/skills"
def load_skills():
skills = []
ifnot os.path.exists(SKILLS_DIR):
return []
try:
for skill_file in Path(SKILLS_DIR).glob("*.json"):
with open(skill_file, 'r') as f:
skills.append(json.load(f))
return skills
except:
return []Skill 注入
if skills:
context_parts.append(
f"\n# Skills\n" + "\n".join(
[f"- {s['name']}: {s.get('description', '')}"for s in skills]
)
)| 维度 | Rules | Skills |
|---|---|---|
| 文件格式 | Markdown | JSON |
| 作用 | 约束行为（"不要做什么"） | 提供能力（"可以怎么做"） |
| 类比 | 公司规章制度 | 员工培训手册 |
| 注入方式 | 全文注入 | 名称 + 描述摘要 |
没有一个使用Tools声明来注册funtion calling，与其注册新函数，不如教LLM如何写代码。
| Skill | 有 scripts/ ？ | 有参考文档？ | 核心执行方式 |
|---|---|---|---|
| ✅ | ✅ | 脚本执行 + 参考文档 | |
| pptx | ✅ | ✅ | 脚本执行 + 参考文档 |
| xlsx | ✅ | ❌ | 脚本执行 |
| docx | ✅ | ✅ | 脚本执行 |
| webapp-testing | ✅ | ✅ | 脚本执行 |
| frontend-design | ❌ | ❌ | 纯 Prompt 注入 |
| brand-guidelines | ❌ | ❌ | 纯 Prompt 注入 |
| algorithmic-art | ❌ | ❌ | 纯 Prompt 注入 |
| doc-coauthoring | ❌ | ❌ | 纯 Prompt 注入 |
| internal-comms | ❌ | ❌ | 纯 Prompt 注入 |
| web-artifacts-builder | ❌ | ❌ | 纯 Prompt 注入 |
| canvas-design | ❌ | ❌ | 纯 Prompt + 资源 |
| theme-factory | ❌ | ❌ | 纯 Prompt + 资源 |
| slack-gif-creator | ❌ (有 core/) | ❌ | 库调用型 |
| mcp-builder | ✅ | ✅ | 参考文档 + 编排 |
| skill-creator | ✅ | ✅ | 编排型（含子 Agent） |
核心组件
SKILL.md → FsSkillRepository（扫描、解析）
→ Skill 对象（name, description, body, tools, resources）
→ SkillToolSet（6 个管理工具 + skill_run）
→ DynamicSkillToolSet（按需加载业务工具）
→ SkillsRequestProcessor（注入 system prompt）name + desc注入system prompt在沙箱中执行命令1. LLM 调用 skill_list()
→ 看到所有技能的 name + description（~30 Token/个）
2. LLM 调用 skill_load("pdf")
→ 触发 state_delta 写入，SKILL.md body 注入 system prompt
3. LLM 调用 skill_select_docs(docs=["forms.md"])
→ 按需加载详细文档
4. LLM 调用 skill_run(command="python3 scripts/xxx.py")
→ 在沙箱中执行命令Skill run的沙箱执行
1. 创建隔离工作空间 /tmp/ws_xxx/
2. 将技能目录复制到工作空间（增量哈希优化，不重复拷贝）
3. 自动注入环境变量：$WORK_DIR、$OUTPUT_DIR、$SKILLS_DIR
4. bash -lc "python3 scripts/xxx.py" 在隔离目录中执行
5. 收集 stdout + 指定的输出文件，返回给 LLM工作空间布局
/tmp/ws_session123/
├── skills/pdf/ ← 技能目录（只读保护）
│ ├── SKILL.md
│ ├── scripts/
│ ├── out/ → ../../out ← 符号链接
│ └── work/ → ../../work ← 符号链接
├── out/ ← $OUTPUT_DIR
├── work/ ← $WORK_DIR
└── runs/ ← 执行记录输入输出
注入
skill_run 这个工具加载技能后用skill_run执行command参数如何写 "python3 scripts/xxx.py ..."工具定义
System Prompt 注入
{
"name": "skill_run",
"description": "Run a command inside a skill workspace. Stages the entire skill directory and runs a single command.",
"parameters": {
"skill": {"type": "string", "description": "技能名称"},
"command": {"type": "string", "description": "要执行的 shell 命令"},
"output_files": {"type": "array", "description": "输出文件的 glob 模式"},
...
}
}instruction = f"""
Available skills:
{skill_instructions}
Tooling and workspace guidance:
- Skills run inside an isolated workspace...
- Prefer $SKILLS_DIR, $WORK_DIR, $OUTPUT_DIR... over hard-coded paths
- If a skill is not loaded, call skill_load
- When body and needed docs/tools are present, call skill_run or use tools directly
"""skill_load("pdf")：把SKILL.md的body文本，完整注入到SystemPromptcommand 具体怎么填写## 可用脚本
- `scripts/extract_form_structure.py` - 提取 PDF 表单结构
用法：python3 scripts/extract_form_structure.py <input_pdf>
- `scripts/fill_fillable_fields.py` - 填充 PDF 表单字段
用法：python3 scripts/fill_fillable_fields.py <input_pdf> <data_json>
输出到 $OUTPUT_DIR/filled_output.pdf
### 依赖安装
首次使用时运行：pip install -r scripts/requirements.txtSKILL.md 的 body
Body是给LLM看的使用手册，body质量是整个系统的关键。框架功能
手册加入SystemPromptskill_run这个万能执行器。后期考LLM
怎么组装命令、什么顺序调用，全靠LLM自己的理解能力。1. 看 system prompt 里的技能列表
→ 发现 "pdf" 技能匹配
2. 调 skill_load("pdf")
→ SKILL.md body 注入 system prompt
3. 阅读 body
→ 发现 scripts/extract_form_structure.py 可以分析表单结构
4. 拼装 command：
"python3 scripts/extract_form_structure.py input.pdf"
5. 调 skill_run(skill="pdf", command="python3 scripts/extract_form_structure.py input.pdf")
6. 读返回的 stdout
→ 知道表单有哪些字段
7. 再调 skill_run 执行填充脚本nanoAgent Skill 示例，和标准的claude code /openclaw 示例不一样，后续补充示例
{
"name": "docker-deploy",
"description": "Deploy application using Docker Compose. Steps: 1) Check Dockerfile exists, 2) Run docker-compose build, 3) Run docker-compose up -d, 4) Verify containers are running.",
"triggers": ["deploy", "docker", "container"]
}核心
Skill ：一段精心编写的 system prompt。价值
提供领域专业知识（设计原则、色彩理论）约束 LLM 行为（"永远不要用通用 AI 美学"）引导思维流程（"编码前先理解上下文"）代表
示例
frontend-design/
└── SKILL.md ← 唯一的内容文件SKILL.md
# 设计思考
在编码之前，先理解上下文并致力于一个 BOLD 美学方向：
- **目的**: 此界面解决什么问题？谁在使用它？
- **基调**: 选择一个极端：粗犷极简、复古未来主义、奢华精致...
- **差异化**: 什么让这个令人难忘？
## 前端美学指南
- 选择美观独特的字体，避免 Arial/Inter...
- 使用 CSS 变量保持一致性...
- **永远不要** 使用通用 AI 生成的美学...执行流程
用户: "帮我做一个科技感的落地页"
│
▼ LLM 匹配到 frontend-design，加载 SKILL.md
▼ SKILL.md body 注入 system message
▼ LLM 根据注入的美学指南，直接写 HTML/CSS/React
▼ 通过 write_to_file 输出代码
│
用户得到一个有设计感的落地页核心
SKILL.md 当教程，scripts/ 当工具箱价值(PDF示例)
SKILL.md，本质也是Promptscripts/代表
PDF 示例
pdf/
├── SKILL.md ← 使用手册 + 代码示例
├── forms.md ← 表单填写指南
├── reference.md ← 高级参考
└── scripts/
├── extract_form_structure.py
├── fill_fillable_fields.py
├── convert_pdf_to_images.py
└── ... (共 8 个脚本)执行流程
用户: "帮我填写这个 PDF 表单"
│
▼ LLM 加载 SKILL.md，了解 PDF 操作全景
▼ 发现需要填表 → 按需加载 forms.md
▼ 按照文档指引，调用预制脚本：
│
│ skill_run("python3 scripts/extract_form_structure.py input.pdf")
│ → 返回: "Found 12 fields..."
│
│ skill_run("python3 scripts/fill_fillable_fields.py input.pdf data.json")
│ → 返回: 填好的 PDF
│
用户得到填好的 PDF 表单SKILL.md代码示例和预制脚本的对比：
| SKILL.md 中的代码示例 | scripts/ 预制脚本 | |
|---|---|---|
| 用途 | 教 LLM 怎么写代码 | 直接执行的黑盒工具 |
| 谁执行 | LLM 自己写 + Agent 基础工具 | skill_run 在沙箱中执行 |
| 适合 | 简单一次性操作 | 复杂、需要验证的工作流 |
核心
自带python库，LLM 写代码可以 import 它。不执行预制脚本，而是现场编写自定义脚本，组合SKILL提供的库函数完成任务。价值
SKILL.md：API文档，教LLM如何使用函数。库函数：提供可调用的SDK代表
示例
core是一个python库
slack-gif-creator/
├── SKILL.md ← API 文档 + 使用示例
├── requirements.txt ← 依赖声明
└── core/ ← Python 库（不是 scripts/！）
├── gif_builder.py ← GIFBuilder 类
├── validators.py ← validate_gif, is_slack_ready
├── easing.py ← 缓动函数
└── frame_composer.py ← 帧生成辅助执行流程
用户: "做一个心跳动画的 Slack emoji GIF"
│
▼ LLM 加载 SKILL.md，理解 API
▼ LLM 自己编写一个完整脚本：
│
│ from core.gif_builder import GIFBuilder
│ from PIL import Image, ImageDraw
│ import math
│
│ builder = GIFBuilder(width=128, height=128, fps=10)
│ for i in range(20):
│ scale = 0.8 + 0.4 * abs(math.sin(i/19 * math.pi * 2))
│ frame = Image.new('RGB', (128, 128), (255, 240, 240))
│ # ... 绘制心形 + 缩放 ...
│ builder.add_frame(frame)
│ builder.save('heartbeat.gif', optimize_for_emoji=True)
│
▼ skill_run 执行这个 LLM 写的脚本
│ （core/ 已被复制到工作空间，可直接 import）
│
用户得到优化过的 Slack emoji GIF与脚本执行型的关键区别
| 脚本执行型（pdf） | 库调用型（slack-gif-creator） | |
|---|---|---|
| 文件结构 | scripts/xxx.py（独立可执行） | core/xxx.py（Python 模块） |
| 调用方式 | skill_run("python3 scripts/xxx.py") | LLM 自己写脚本 import core.xxx |
| LLM 角色 | 调用者（跑预制脚本） | 开发者（组合库函数写新代码） |
核心
SKILL.md 是路由表，详细文档按需加载。价值
SKILL.md：本身不包含详细操作步骤，它只告诉 LLM "你要做的事情，应该去读哪份文档"。定义多个详细文档，不用一次性全部加载，大量节省token。代表
示例
pptx/
├── SKILL.md ← 路由表（Quick Reference）
├── editing.md ← 编辑工作流（6.9KB）
├── pptxgenjs.md ← 从零创建（12.8KB）
└── scripts/
├── thumbnail.py
└── office/
├── unpack.py
└── soffice.py路由表
## Quick Reference
| Task | Guide |
|------|-------|
| Read/analyze content | `python -m markdown presentation.pptx` |
| Edit or create from template | Read [editing.md](editing.md) |
| Create from scratch | Read [pptxgenjs.md](pptxgenjs.md) |三层信息模型
第1层 description（~50词）
判断是否需要加载第2层 SKILL.md body（~200行）
Quick Reference 路由表 + 设计指南 + QA 流程注入 system message，LLM 知道大方向第3层 editing.md / pptxgenjs.md（按需）
详细操作步骤 + 代码示例skill_select_docs 按需加载第1层 description（~50词）
"Presentation creation, editing, and analysis..."
→ 始终在可用技能列表中，LLM 判断是否需要加载
第2层 SKILL.md body（~200行）
Quick Reference 路由表 + 设计指南 + QA 流程
→ 加载后注入 system message，LLM 知道大方向
第3层 editing.md / pptxgenjs.md（按需）
详细操作步骤 + 代码示例
→ LLM 通过 skill_select_docs 按需加载Token 节省效果
一次性全部加载
渐进加载（编辑任务）
核心原则
先粗筛再精选：面对海量数据/复杂问题，不要一步到位，应采用分层策略轻量信息粗筛：需要哪个 Skill？中等信息定方向：该读哪份文档？详细信息执行：具体怎么操作？| 注入层级 | 触发条件 | 典型 Token 消耗 |
|---|---|---|
| L0: 概览 | 始终注入 | ~30 Token/技能 |
| L1: SKILL.md body | skill_load 后 | ~500-2000 Token |
| L2: 详细文档 | skill_select_docs 后 | ~1000-5000 Token |
核心
SKILL.md 不是工具说明，而是一个完整的多阶段工作流编排方案价值
SKILL.md：定义了一条完整的多阶段流水线代表
示例
skill-creator/
├── SKILL.md ← 32KB 的超详细编排指南
├── agents/ ← 子 Agent 指令
│ ├── grader.md ← 评分 Agent
│ ├── comparator.md ← A/B 对比 Agent
│ └── analyzer.md ← 分析 Agent
├── scripts/ ← 自动化脚本
│ ├── aggregate_benchmark.py
│ ├── run_loop.py
│ ├── run_eval.py
│ └── package_skill.py
├── eval-viewer/ ← 评估结果查看器
└── references/ ← 参考文档SKILL.md 流水线
Capture Intent → Interview → Write SKILL.md → Run Tests
→ Evaluate → Improve → Repeat → PackageSKILL.md（用 write_to_file）.skill 文件选择
写好 SKILL.md 是一切的基础。
| 你的场景 | 推荐模式 | 你需要做的 |
|---|---|---|
| 教 LLM 遵循某种规范/风格 | 纯 Prompt 注入 | 写好 SKILL.md，全靠文本质量 |
| 需要 LLM 操作特定文件格式 | 脚本执行型 | 写预制脚本放 scripts/，SKILL.md 写使用说明 |
| 需要 LLM 灵活组合 API | 库调用型 | 写 Python 库放 core/，SKILL.md 当 API 文档 |
| 知识量大，不同任务需要不同文档 | 渐进加载型 | SKILL.md 做路由表，详细文档独立存放 |
| 需要 LLM 执行复杂多步骤工作流 | 编排型 | SKILL.md 定义流水线 + 脚本工具链 + 子 Agent |
对比
Skill 执行模式光谱
◄── 轻量 ────────────────────────────────── 重量 ──►
纯 Prompt 参考文档 库调用 脚本执行 编排
注入型 渐进加载型 型 型 型
────────── ────────── ────────── ────────── ──────────
frontend- pptx slack-gif- pdf skill-
design mcp-builder creator xlsx creator
────────── ────────── ────────── ────────── ──────────
仅注入 body 注入 + 注入 + 注入 + 注入 +
到 system 按需加载 LLM 写代码 预制脚本 多步骤
message docs import 库 skill_run 工作流| 框架机制 | 模式一 | 模式二 | 模式三 | 模式四 | 模式五 |
|---|---|---|---|---|---|
| skill_load → body 注入 | ✅ | ✅ | ✅ | ✅ | ✅ |
| skill_select_docs | ❌ | ✅ | ❌ | ✅ | ✅ |
| skill_run (预制脚本) | ❌ | ✅ | ❌ | ✅ | ✅ |
| skill_run (LLM 写的脚本) | ❌ | ❌ | ✅ | ❌ | ❌ |
| Tools: 声明 (function calling) | ❌ | ❌ | ❌ | ❌ | ❌ |
| 子 Agent 编排 | ❌ | ❌ | ❌ | ❌ | ✅ |
SKILL vs Funtion Call
确定性操作设计不确定性的内容无需注册funtion calling，仅需写好SKILL.md
看一眼SKILL.md，就能写出正确代码。不能灵活组合在 shell 里跑，skill_run 就能执行。不需要为每个操作都定义工具 Schema。不能通过写代码/Shell完成的操作，才需要注册funtion calling。都可以通过代码+脚本完成。执行力
SKILL.md body 质量 × (Agent 基础工具 + skill_run 沙箱能力)组件
灵魂，它决定LLM能否理解任务、选对方法、写对代码。手脚，它提供了在隔离沙箱中 执行任意命令的能力。可选配件。框架支持，但大多数场景不需要。all_tools = base_tools + mcp_tools
背景
硬编码ToolsMCP协议，配置可用的Tools即可。工具定义和工具实现。MCP
示例
通用标准协议
没有 MCP 的世界: 有 MCP 的世界:
Agent A Agent B MCP Server: Slack MCP Server: GitHub
├── Slack (自写) ├── Slack (自写) │ │
├── GitHub(自写) ├── Jira (自写) └───── 标准协议 ──┘
└── DB (自写) └── DB (自写) │
┌───────┼───────┐
每个 Agent 各写各的 Agent A Agent B Agent C
N × M 的工作量 工具实现一次，全部共享
N + M 的工作量配置文件：.agent/mcp.json
{
"mcpServers": {
"filesystem": {
"disabled": false,
"tools": [{
"name": "list_directory",
"description": "List contents of a directory with metadata",
"parameters": {
"type": "object",
"properties": {"path": {"type": "string"}},
"required": ["path"]
}
}]
},
"database": {
"disabled": true,
"tools": [...]
}
}
}MCP 配置读取
MCP_CONFIG = ".agent/mcp.json"
def load_mcp_tools():
ifnot os.path.exists(MCP_CONFIG):
return []
try:
with open(MCP_CONFIG, 'r') as f:
config = json.load(f)
mcp_tools = []
for server_name, server_config in config.get("mcpServers", {}).items():
if server_config.get("disabled", False):
continue
for tool in server_config.get("tools", []):
mcp_tools.append({"type": "function", "function": tool})
return mcp_tools
except:
return []def run_agent_claudecode(task, use_plan=False):
print("[Init] Loading ClaudeCode features...")
# 1. 从文件系统加载所有外部配置
memory = load_memory() # 历史记忆
rules = load_rules() # 行为规则
skills = load_skills() # 技能注册
mcp_tools = load_mcp_tools() # MCP 外部工具
# 2. 合并工具列表（基础工具 + MCP 工具）
all_tools = base_tools + mcp_tools
# 3. 构建 system prompt（基础指令 + Rules + Skills + Memory）
context_parts = ["You are a helpful assistant..."]
if rules:
context_parts.append(f"\n# Rules\n{rules}")
if skills:
context_parts.append(f"\n# Skills\n...")
if memory:
context_parts.append(f"\n# Previous Context\n{memory}")
messages = [{"role": "system", "content": "\n".join(context_parts)}]
# 4. 执行 ...┌─────────────────────── 文件系统 ───────────────────────┐
│ │
│ .agent/rules/*.md → load_rules() → system prompt │
│ .agent/skills/*.json → load_skills() → system prompt │
│ .agent/mcp.json → load_mcp_tools()→ tools 列表 │
│ agent_memory.md → load_memory() → system prompt │
│ │
└─────────────────────────────────────────────────────────┘
│
▼
┌──── Agent 运行时 ────┐
│ │
│ system prompt = │
│ 基础指令 │
│ + Rules │
│ + Skills │
│ + Memory │
│ │
│ tools = │
│ base_tools (7个) │
│ + mcp_tools (N个) │
│ │
└──────────────────────┘┌───────────────────────────────────────────────────────┐
│ Agent 架构全景 │
│ │
│ ┌──────────────┐ 第三篇：agent-claudecode.py │
│ │ Rules │ 行为约束层 ──── .agent/rules/ │
│ │ Skills │ 技能知识层 ──── .agent/skills/ │
│ │ MCP │ 工具扩展层 ──── .agent/mcp.json │
│ │ Plan Tool │ 自主规划层 ──── plan() 作为工具 │
│ ├──────────────┤ 第二篇：agent-plus.py │
│ │ Memory │ 持久记忆层 ──── agent_memory.md │
│ │ Planning │ 任务分解层 ──── create_plan() │
│ │ Multi-step │ 多步编排层 ──── 步骤间上下文共享 │
│ ├──────────────┤ 第一篇：agent.py │
│ │ LLM │ 推理决策层 ──── OpenAI API │
│ │ Tools │ 工具执行层 ──── bash/read/write │
│ │ Loop │ 核心循环层 ──── for + tool_calls │
│ └──────────────┘ │
└───────────────────────────────────────────────────────┘每一层都在回答一个关键问题：
Loop
自主运行？Tools
作用于世界？Memory
记住过去？Planning
应对复杂任务？Rules
遵守约束？Skills
领域知识？MCP
新工具？背景
找帮手、分工合作。主Agent
项目经理，把子任务委派给有不同专业身份的SubAgent。各管一块，互不干扰。SubAgent
一次性的、多次独立的、用完就扔的临时工。干净的上下文和专注的角色，而不是构建一个持久的团队。# 主 Agent: 协调者
"You are an orchestrator agent. You can delegate to sub-agents..."
# SubAgent: 专家
f"You are a {role}. Be concise and focused. Only do what is asked."生命周期
生成 → 接收任务 → 干活（可以调用工具）→ 返回结果摘要 → 消亡类比：但是没有人物名称等内容，有名称持久化记忆则为Teams。
之前（一个人干所有活）:
老板 → "小张，你把前端后端数据库全搞定"
小张（一个人扛所有）
- 写后端 API...
- 写前端页面...（等等，后端那个接口叫啥来着？）
- 建数据库表...（前端那个字段是什么格式？）
现在（项目经理 + 专人）:
老板 → 项目经理（主 Agent）
│
├── "后端用 FastAPI" → 后端工程师（SubAgent A）
├── "前端用 React" → 前端工程师（SubAgent B）
└── "验证能跑通" → 测试工程师（SubAgent C）
每个人只管自己的事，干完把结果交给项目经理汇总。SubAgent 调用链路
用户: "创建一个 TODO 应用，包含 Python 后端和 HTML 前端"
│
▼
主 Agent 的 run_agent() 循环启动
│
▼
(1) 代码把 messages + tools 列表发送给 LLM
tools 列表里包含: [read, write, edit, glob, grep, bash, subagent]
^^^^^^^^
LLM 看到了这个工具
│
▼
(2) LLM 分析任务，决定委派，返回:
{"tool_calls": [{"function": {"name": "subagent",
"arguments": {"role": "Python backend developer",
"task": "用 FastAPI 创建..."}}}]}
│
▼
(3) 核心循环中的通用调度代码执行:
fn = "subagent"
args = {"role": "Python backend developer", "task": "..."}
result = available_functions["subagent"](**args ""subagent"")
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
走到了我们写的 subagent() 函数！
│
▼
(4) subagent() 内部启动一个全新的 Agent 循环
- 独立的 system prompt: "You are a Python backend developer."
- 独立的 messages 列表
- 可以使用 read/write/edit/bash 等工具
- 循环结束后，返回结果文本
│
▼
(5) 结果返回给主 Agent，主 Agent 可能继续派出前端 SubAgent...SubAgent Tool 定义
{
"name": "subagent",
"description": "Delegate a task to a specialized sub-agent with its own role and independent context.",
"parameters": {
"type": "object",
"properties": {
"role": {"type": "string", "description": "The sub-agent's specialty, e.g. 'Python backend developer'"},
"task": {"type": "string", "description": "The specific task to delegate"}
},
"required": ["role", "task"]
}
}SubAgent Tool 实现
def subagent(role, task):
"""启动一个独立的 Agent 循环，拥有专属角色和独立上下文"""
print(f"\n[SubAgent:{role}] 开始: {task}")
# 关键 1：独立的 messages，独立的 system prompt
sub_messages = [
{"role": "system", "content": f"You are a {role}. Be concise and focused. Only do what is asked."},
{"role": "user", "content": task}
]
# 关键 2：排除 subagent 自身，防止无限递归
sub_tools = [t for t in tools if t["function"]["name"] != "subagent"]
# 关键 3：一个完整的 Agent 循环（和第一篇的核心循环一模一样）
for _ in range(10):
response = client.chat.completions.create(
model=MODEL, messages=sub_messages, tools=sub_tools
)
message = response.choices[0].message
sub_messages.append(message)
if not message.tool_calls:
print(f"[SubAgent:{role}] 完成")
return message.content
for tc in message.tool_calls:
fn = tc.function.name
args = json.loads(tc.function.arguments)
print(f" [SubAgent:{role}] {fn}({args})")
result = available_functions[fn](**args "fn")
sub_messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
return"SubAgent: max iterations reached"注册到路由表
available_functions["subagent"] = subagent| 维度 | Plan | SubAgent（本文） |
|---|---|---|
| 上下文 | 所有步骤共享 messages | 每个 SubAgent 独立 messages |
| 身份 | 同一个 Agent，同一个角色 | 每个 SubAgent 不同的专业角色 |
| 生命周期 | 步骤间 Agent 持续存在 | 生成 → 干活 → 返回摘要 → 消亡（一次性） |
| 跨次记忆 | 步骤 2 能看到 步骤 1的全部细节 | SubAgent B 看不到 SubAgent A 做了什么 |
| 适合 | 步骤之间有依赖 | 子任务之间相对独立 |
| 类比 | 一个人按步骤做事 | 叫了个跑腿临时工，干完就走 |
┌───────────────────────────────────────────────────────┐
│ Agent 架构全景 │
│ │
│ ┌──────────────┐ 第四篇 (本文) │
│ │ SubAgent │ 多智能体协作 ── subagent() 工具 │
│ ├──────────────┤ 第三篇 │
│ │ Rules │ 行为约束层 ──── .agent/rules/ │
│ │ Skills │ 技能知识层 ──── .agent/skills/ │
│ │ MCP │ 工具扩展层 ──── .agent/mcp.json │
│ ├──────────────┤ 第二篇 │
│ │ Memory │ 持久记忆层 ──── agent_memory.md │
│ │ Planning │ 任务分解层 ──── create_plan() │
│ ├──────────────┤ 第一篇 │
│ │ LLM │ 推理决策层 ──── OpenAI API │
│ │ Tools │ 工具执行层 ──── bash/read/write │
│ │ Loop │ 核心循环层 ──── for + tool_calls │
│ └──────────────┘ │
└───────────────────────────────────────────────────────┘| 篇 | 文件 | 核心主题 | 一句话总结 |
|---|---|---|---|
| 一 | agent.py (115行) | 工具 + 循环 | Agent 的最小本质——LLM 是大脑，代码是手脚 |
| 二 | agent-plus.py (182行) | 记忆 + 规划 | 时间维度——记住过去、规划未来 |
| 三 | agent-claudecode.py (265行) | Rules + Skills + MCP | 空间维度——扩展知识与工具 |
| 四 | agent-subagent.py (192行) ⭐新 | SubAgent | 协作维度——给 Agent 找帮手 |
背景
同一个人，被多次找到，而且他还记得上次做了什么。SubAgent
Teams Agent
| 问题 | SubAgent（临时工） | Teams Agent（正式员工） |
|---|---|---|
| 有名字吗？ | ❌ 只有一个临时角色描述 | ✅ 有名字（alice）、有固定角色 |
| 记得上次做了什么吗？ | ❌ 每次调用都失忆 | ✅ 多次交互之间记忆持续累积 |
| 能收到同事消息吗？ | ❌ 互相看不到 | ✅ 有收件箱，能收消息 |
| 什么时候消失？ | 函数返回就没了 | 团队解散才消失 |
TeamsAgent
name/role
inbox
messages
记得之前做过什么。class Agent:
def __init__(self, name, role):
# 身份：有名字，有角色
self.name = name
self.role = role
# 通信：收件箱
self.inbox = []
# 记忆：持久保持
self.messages = [
{"role": "system", "content": f"You are {name}, a {role}. Be concise and focused."}
] 对比SubAgent
# SubAgent（上文）—— 一个函数，用完就没
def subagent(role, task):
sub_messages = [...] # 局部变量，函数返回即消亡
for _ in range(10):
...
return result # 返回后 sub_messages 被垃圾回收，一切归零Chat：发消息
def chat(self, task):
# 第 1 步：如果 inbox 有新消息，先读取并消化
if self.inbox:
mail = "\n".join(f"[来自 {m['from']}]: {m['content']}"for m in self.inbox)
self.messages.append({"role": "user", "content": f"你收到了团队成员的消息:\n{mail}"})
resp = client.chat.completions.create(model=MODEL, messages=self.messages)
self.messages.append(resp.choices[0].message)
self.inbox.clear()
# 第 2 步：执行本次任务（和之前的 Agent 循环一样）
self.messages.append({"role": "user", "content": task})
for _ in range(10):
response = client.chat.completions.create(model=MODEL, messages=self.messages, tools=tools)
message = response.choices[0].message
self.messages.append(message)
ifnot message.tool_calls:
return message.content
for tc in message.tool_calls:
# ... 执行工具，追加结果（和第一篇完全一样）收消息
def receive(self, sender, message):
self.inbox.append({"from": sender, "content": message}) class Team:
def __init__(self):
self.agents = {} # name → Agent
def hire(self, name, role):
"""招募：创建一个持久 Agent"""
agent = Agent(name, role)
self.agents[name] = agent
return agent
def send(self, from_name, to_name, message):
"""点对点通信"""
self.agents[to_name].receive(from_name, message)
def broadcast(self, from_name, message):
"""广播：给团队所有其他人发消息"""
for name, agent in self.agents.items():
if name != from_name:
agent.receive(from_name, message)
def disband(self):
"""解散：所有 Agent 生命周期结束"""
self.agents.clear() 协作流程
def run_team(task):
team = Team()
# 第 1 阶段：组建团队
# LLM 自动拆分角色
members = plan_team(task)
for m in members:
team.hire(m["name"], m["role"])
# 第 2 阶段：逐个执行，每人干完广播成果
for m in members:
agent = team.agents[m["name"]]
result = agent.chat(m["task"])
team.broadcast(m["name"], f"我完成了任务。摘要: {result[:200]}")
# 第 3 阶段：最后一个成员做二次审查
reviewer = team.agents[members[-1]["name"]]
review = reviewer.chat("请根据团队成果做最终审查")
# 第 4 阶段：解散
team.disband()输入： 创建一个 TODO 应用，包含 Python 后端和 HTML 前端
[PM] 分析任务，组建团队...
[团队] 3 人:
1. alice — backend developer → 用 FastAPI 创建 TODO 后端 API
2. bob — frontend developer → 创建 HTML 前端页面
3. carol — test engineer → 验证前后端能正常工作
============================================================
第 1 阶段: 招募团队
============================================================
[创建] alice (backend developer)
[创建] bob (frontend developer)
[创建] carol (test engineer)
============================================================
第 2 阶段: 协作开发
============================================================
── [1/3] alice 开始工作 ──
[alice] write({"path": "app.py", ...})
[alice] → 已创建 app.py，包含 GET/POST/DELETE 三个接口...
[广播] alice → 全体: 我完成了任务。摘要: 已创建 app.py...
── [2/3] bob 开始工作 ──
（bob 的 inbox 里有 alice 的广播，他知道后端接口长什么样）
[bob] write({"path": "index.html", ...})
[bob] → 已创建 index.html，调用了 alice 定义的 API 接口...
[广播] bob → 全体: 我完成了任务。摘要: 已创建 index.html...
── [3/3] carol 开始工作 ──
（carol 的 inbox 里有 alice 和 bob 的广播）
[carol] read({"path": "app.py"})
[carol] read({"path": "index.html"})
[carol] bash({"command": "python -c 'import app; print(\"OK\")'"})
[carol] → 后端代码语法正确，前端页面已创建，接口调用地址匹配...
[广播] carol → 全体: 我完成了任务。摘要: 验证通过...
============================================================
第 3 阶段: carol 做最终审查
============================================================
（carol 被第二次调用 chat()，她还记得第一次测试的结果）
[carol] → 最终审查：后端 app.py 包含 3 个接口（GET/POST/DELETE），
前端 index.html 已正确引用后端地址，代码验证通过，可以交付。任务简单、互不相关用 SubAgent；需要协作、需要记忆用 Teams。| 场景 | 选 SubAgent | 选 Teams |
|---|---|---|
| 子任务之间完全独立 | ✅ 互不干扰，简单直接 | 没必要，杀鸡用牛刀 |
| 后续任务依赖前面的结果 | ❌ 看不到别人做了什么 | ✅ 通过通信通道传递信息 |
| 需要同一个人多次返工 | ❌ 每次都是新人，不记得 | ✅ 持久记忆，记得上次做了什么 |
| 需要测试 → 修 bug → 再测试 | ❌ 做不到 | ✅ 测试人员和开发都能被多次调用 |
| 篇 | 核心新增 | 一句话 |
|---|---|---|
| 一 | 工具 + 循环 | Agent 的最小本质 |
| 二 | 记忆 + 规划 | 记住过去，规划未来 |
| 三 | Rules + Skills + MCP | 扩展知识与工具 |
| 四 | SubAgent | 一次性临时工 |
| 五 | Agent 类 + Team 类 | 有记忆、有身份、能通信的正式团队 |
不断交互，迟早会爆。
第 1 轮: messages += [LLM的回复, 工具的返回结果]
第 2 轮: messages += [LLM的回复, 工具的返回结果]
第 3 轮: messages += [LLM的回复, 工具的返回结果]
...方法1：更大上下文模型
方法2：限制最大循环次数
方法3：截断工具返回结果
方法4：压缩旧的对话历史
记住要点，忘掉细节，保留现场。nanoAgent 的压缩是最朴素的实现。业界的方案更加精细。似乎ClaudeCode就很简单一个压缩Prompt
| 维度 | agent-compact.py | OpenClaw / Claude Code 等生产级实现 |
|---|---|---|
| 触发条件 | 消息条数超过固定阈值 | 基于 token 数精确计算，考虑模型的实际窗口大小 |
| 压缩方式 | 一次性把所有旧消息压缩成一段摘要 | 分层压缩：最近的保留原文，稍远的做摘要，更远的只保留关键事实 |
| 保留策略 | 固定保留最近 N 条 | 智能选择：保留包含文件路径、错误信息等关键消息 |
| 摘要质量 | 通用摘要 prompt | 针对 coding 场景优化的摘要 prompt，确保保留文件路径、代码片段、决策原因 |
压缩前的 messages（30 条，快爆了）:
┌────────┐
│ system │ ← 永远保留
├────────┤
│ user │ ← 最初的任务
│ assist │ ← LLM 调用了 bash
│ tool │ ← bash 输出了 200 行文件列表
│ assist │ ← LLM 调用了 read_file
│ tool │ ← 文件内容 500 行 ─┐
│ assist │ ← LLM 决定统计行数 │
│ tool │ ← 统计结果 │ 这些旧消息
│ assist │ ← LLM 调用了 grep │ 交给 LLM 做摘要
│ tool │ ← grep 结果 300 行 │
│ ... │ ← 更多历史 ─┘
│ assist │ ← LLM 准备写文件 ─┐
│ tool │ ← 写入成功 │ 最近 6 条
│ assist │ ← LLM 调用 read 验证 │ 保留原样
│ tool │ ← 文件内容 │ （不压缩）
│ assist │ ← LLM 准备做最后总结 │
│ user │ ← 当前操作 ─┘
└────────┘
↓ compact_messages() ↓
压缩后的 messages（9 条，清爽了）:
┌────────┐
│ system │ ← 永远保留（不动）
├────────┤
│ user │ ← "之前的对话摘要：找到了 42 个 Python 文件，
│ │ 统计了行数，最长的是 utils.py (350行)..."
│ assist │ ← "明白了，我继续。"
├────────┤
│ assist │ ← LLM 准备写文件 ─┐
│ tool │ ← 写入成功 │ 最近 6 条
│ assist │ ← LLM 调用 read 验证 │ 完整保留
│ tool │ ← 文件内容 │
│ assist │ ← LLM 准备做最后总结 │
│ user │ ← 当前操作 ─┘
└────────┘Summarize the following conversation history. Keep all important facts, file paths, command results, and decisions. Be concise but don't lose critical details.
关键代码
COMPACT_THRESHOLD = 20 # 超过 20 条就压缩
KEEP_RECENT = 6 # 保留最近 6 条不压缩
def compact_messages(messages):
if len(messages) <= COMPACT_THRESHOLD:
return messages # 没超阈值，不压缩
# system prompt 永远保留
system_msg = messages[0]
# 旧消息 → 要被压缩
old_messages = messages[1:-KEEP_RECENT]
# 最近的消息 → 保留原样
recent_messages = messages[-KEEP_RECENT:]
# 把旧消息拼成文本
old_text = ""
for msg in old_messages:
role = msg.get("role", "unknown") if isinstance(msg, dict) else getattr(msg, "role", "unknown")
content = msg.get("content", "") if isinstance(msg, dict) else getattr(msg, "content", "")
if content:
old_text += f"[{role}]: {content}\n"
# 调用 LLM 生成摘要
summary_response = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": "Summarize the following conversation history. Keep all important facts, file paths, command results, and decisions. Be concise but don't lose critical details."},
{"role": "user", "content": old_text}
]
)
summary = summary_response.choices[0].message.content
# 重新组装
return [
system_msg,
{"role": "user", "content": f"[Previous conversation summary]: {summary}"},
{"role": "assistant", "content": "Understood. I have the context from our previous conversation. Let me continue."},
*recent_messages
]| 篇 | 核心主题 | 解决什么问题 |
|---|---|---|
| 一 | 工具 + 循环 | Agent 如何自主工作 |
| 二 | 记忆 + 规划 | Agent 如何记住过去、规划未来 |
| 三 | Rules + Skills + MCP | Agent 如何扩展知识和工具 |
| 四 | SubAgent | Agent 如何临时找帮手 |
| 五 | Teams | Agent 如何组建持久团队 |
| 六 | 上下文压缩 | Agent 如何在有限窗口内持续工作 |
LLM 可能生成 rm -rf /等命令。
LLM 输出一条命令
│
▼
防线 1: 命令黑名单
│ "rm -rf /" → �� 直接拦截，不问用户
│ "ls -la" → ✅ 通过
▼
防线 2: 用户确认
│ "find . -name '*.py'" → 用户看到后按 Y 放行
│ → 用户按 N 跳过
│ → 用户按 Q 终止 Agent
▼
防线 3: 输出截断
│ 命令输出 10000 行 → 截断为首尾各 2500 字符
│ 命令输出 10 行 → 原样返回
▼
结果返回给 LLM核心是正则匹配，只能拦阻已知危险，拦不住所有的危险。
DANGEROUS_PATTERNS = [
r'\brm\s+(-[a-zA-Z]*f[a-zA-Z]*\s+|.*--no-preserve-root)', # rm -rf
r'\brm\s+(-[a-zA-Z]*r[a-zA-Z]*\s+)?/', # rm /
r'\bmkfs\b', # 格式化磁盘
r'\bdd\s+.*of\s*=\s*/dev/', # 覆写磁盘
r'>\s*/dev/sd[a-z]', # 重定向到磁盘设备
r'\bchmod\s+(-R\s+)?777\s+/', # chmod 777 /
r':\(\)\s*\{', # fork bomb
r'\bcurl\b.*\|\s*(ba)?sh', # curl | bash
r'\bwget\b.*\|\s*(ba)?sh', # wget | bash
r'\bshutdown\b', # 关机
r'\breboot\b', # 重启
]
def is_dangerous(command):
for pattern in DANGEROUS_PATTERNS:
if re.search(pattern, command):
return True, pattern
return False, Nonedef ask_user_confirmation(tool_name, args):
if AUTO_APPROVE:
return True
print(f"\n┌─ 确认执行 ─────────────────────────────")
print(f"│ 工具: {tool_name}")
for key, value in args.items():
print(f"│ {key}: {str(value)[:200]}")
print(f"└────────────────────────────────────────")
while True:
answer = input("[Y]执行 / [N]跳过 / [Q]终止 Agent > ").strip().lower()
if answer in ('y', 'yes', ''):
return True
elif answer in ('n', 'no'):
return False
elif answer in ('q', 'quit'):
sys.exit(0)并非所有动作都需要确认
在 agent-safe.py 中，三个工具（bash、read_file、write_file）都会触发确认。
但在实际产品中，确认策略可以更精细：
read_file 通常是安全的——只读不写，可以默认放行
write_file 要看路径——写入项目目录内的放行，写入 /etc/ 的要确认
bash 最危险——每次都确认，或者用白名单模式（只允许 ls、grep、cat 等安全命令免确认）
--auto 参数可以跳过所有确认，用于信任场景（比如在 Docker 容器里运行）。
解决问题
cat /var/log/syslog，返回了 10MB 的日志。截断策略
MAX_OUTPUT_LENGTH = 5000
def truncate_output(text):
if len(text) <= MAX_OUTPUT_LENGTH:
return text
half = MAX_OUTPUT_LENGTH // 2
return (
text[:half]
+ f"\n\n... [输出过长，已截断。原始 {len(text)} 字符，保留首尾各 {half} 字符] ...\n\n"
+ text[-half:]
)def execute_bash(command):
# 防线 1: 黑名单
dangerous, pattern = is_dangerous(command)
if dangerous:
return f"�� 命令被拦截: {command}"
# 防线 2: 用户确认
ifnot ask_user_confirmation("execute_bash", {"command": command}):
return "用户跳过了此命令。"
# 执行
try:
result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
output = result.stdout + result.stderr
except subprocess.TimeoutExpired:
output = "Error: 命令执行超时（30秒）"
except Exception as e:
output = f"Error: {str(e)}"
# 防线 3: 输出截断
return truncate_output(output)Hook 管道的execute_bash
def execute_bash(command):
is_dangerous(command) # 检查 1：黑名单
ask_user_confirmation(...) # 检查 2：用户确认
result = subprocess.run(...) # 实际执行
truncate_output(result) # 后处理：截断# 定义 Hook 管道
before_hooks = [check_blacklist, ask_confirmation, log_command]
after_hooks = [truncate_output, log_result]
# 通用的工具执行函数
def execute_tool(name, args):
# 执行前：依次过所有 before hook
for hook in before_hooks:
blocked, msg = hook(name, args)
if blocked:
return msg # 任何一个 hook 可以拦截
# 实际执行
result = available_functions[name](**args "name")
# 执行后：依次过所有 after hook
for hook in after_hooks:
result = hook(name, result)
return result