第四章 Subagents (子智能体)
s01 > s02 > s03 > [ s04 ] s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12
“本专栏基于开源项目
learn-claude-code的官方文档。原文档非常硬核，为了方便像我一样的新手小白理解，我对文档进行了逐行精读，并加入了很多中文注释、大白话解释和踩坑记录。希望这套‘咀嚼版’教程能帮你推开 AI Agent 开发的大门。”
"大任务拆小, 每个小任务干净的上下文" -- 子智能体用独立 messages[], 不污染主对话。
一、问题-脏数据污染上下文
智能体工作越久, messages 数组越胖。每次读文件、跑命令的输出都永久留在上下文里。"这个项目用什么测试框架?" 可能要读 5 个文件, 但父智能体只需要一个词: "pytest。"
为什么会这样：在 LLM（大语言模型）的对话中，你和 AI 的所有聊天记录并不是像微信那样存在服务器里，而是每一次如果你想让 AI 记得前面的话，你必须把之前的每一句对话都打包发给它。
二、解决方案
Parent agent Subagent
+------------------+ +------------------+
| messages=[...] | | messages=[] | <-- fresh
| | dispatch | |
| tool: task | ----------> | while tool_use: |
| prompt="..." | | call tools |
| | summary | append results |
| result = "..." | <---------- | return last text |
+------------------+ +------------------+
Parent context stays clean. Subagent context is discarded.
核心思想就是 “外包”
把那些需要大量翻找资料、产生大量中间过程数据的脏活累活，丢给一个临时的子智能体去做。子智能体做完后，只把最终精华结论传回来，然后立刻销毁，从而保持主智能体（Parent Agent）的大脑（上下文）永远清爽、干净、高效。
三、工作原理
- 父智能体与子智能体的系统提示词对两者的角色进行了隔离，父智能体主要是拆解问题，子智能体主要是干活（跑代码、读文件），最重要的是干完活之后要写总结报告。
SYSTEM = f"You are a coding agent at {WORKDIR}. Use the task tool to delegate exploration or subtasks."
SUBAGENT_SYSTEM = f"You are a coding subagent at {WORKDIR}. Complete the given task, then summarize your findings."
- 父智能体有一个
task工具。子智能体拥有除task外的所有基础工具 (禁止递归生成)。
PARENT_TOOLS = CHILD_TOOLS + [
{
"name": "task",
"description": "Spawn a subagent with fresh context.",
"input_schema": {
"type": "object",
"properties": {
"prompt": {
"type": "string"}},
"required": ["prompt"],
}},
]
- 子智能体以
messages=[]启动, 运行自己的循环。只有最终文本返回给父智能体。
def run_subagent(prompt: str) -> str:
sub_messages = [{
"role": "user", "content": prompt}]
for _ in range(30): # safety limit
response = client.messages.create(
model=MODEL, system=SUBAGENT_SYSTEM,
messages=sub_messages,
tools=CHILD_TOOLS, max_tokens=8000,
)
sub_messages.append({
"role": "assistant",
"content": response.content})
if response.stop_reason != "tool_use":
break
results = []
for block in response.content:
if block.type == "tool_use":
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": str(output)[:50000]})
sub_messages.append({
"role": "user", "content": results})
return "".join(
b.text for b in response.content if hasattr(b, "text")
) or "(no summary)"
子智能体可能跑了 30+ 次工具调用, 但整个消息历史直接丢弃。父智能体收到的只是一段摘要文本, 作为普通 tool_result 返回。
四、相对 s03 的变更
| 组件 | 之前 (s03) | 之后 (s04) |
|---|---|---|
| Tools | 5 | 5 (基础) + task (仅父端) |
| 上下文 | 单一共享 | 父 + 子隔离 |
| Subagent | 无 | run_subagent() 函数 |
| 返回值 | 不适用 | 仅摘要文本 |
五、试一试
cd learn-claude-code
python agents/s04_subagent.py
试试这些 prompt (英文 prompt 对 LLM 效果更好, 也可以用中文):
Use a subtask to find what testing framework this project usesDelegate: read all .py files and summarize what each one doesUse a task to create a new module, then verify it from here
六、完整代码
#!/usr/bin/env python3
import os
import subprocess
from pathlib import Path
from anthropic import Anthropic
from dotenv import load_dotenv
load_dotenv(override=True)
if os.getenv("ANTHROPIC_BASE_URL"):
os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)
WORKDIR = Path.cwd()
client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))
MODEL = os.environ["MODEL_ID"]
SYSTEM = f"You are a coding agent at {WORKDIR}. Use the task tool to delegate exploration or subtasks."
SUBAGENT_SYSTEM = f"You are a coding subagent at {WORKDIR}. Complete the given task, then summarize your findings."
# -- Tool implementations shared by parent and child --
def safe_path(p: str) -> Path:
path = (WORKDIR / p).resolve()
if not path.is_relative_to(WORKDIR):
raise ValueError(f"Path escapes workspace: {p}")
return path
def run_bash(command: str) -> str:
dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
if any(d in command for d in dangerous):
return "Error: Dangerous command blocked"
try:
r = subprocess.run(command, shell=True, cwd=WORKDIR,
capture_output=True, text=True, timeout=120)
out = (r.stdout + r.stderr).strip()
return out[:50000] if out else "(no output)"
except subprocess.TimeoutExpired:
return "Error: Timeout (120s)"
def run_read(path: str, limit: int = None) -> str:
try:
lines = safe_path(path).read_text().splitlines()
if limit and limit < len(lines):
lines = lines[:limit] + [f"... ({len(lines) - limit} more)"]
return "\n".join(lines)[:50000]
except Exception as e:
return f"Error: {e}"
def run_write(path: str, content: str) -> str:
try:
fp = safe_path(path)
fp.parent.mkdir(parents=True, exist_ok=True)
fp.write_text(content)
return f"Wrote {len(content)} bytes"
except Exception as e:
return f"Error: {e}"
def run_edit(path: str, old_text: str, new_text: str) -> str:
try:
fp = safe_path(path)
content = fp.read_text()
if old_text not in content:
return f"Error: Text not found in {path}"
fp.write_text(content.replace(old_text, new_text, 1))
return f"Edited {path}"
except Exception as e:
return f"Error: {e}"
TOOL_HANDLERS = {
"bash": lambda **kw: run_bash(kw["command"]),
"read_file": lambda **kw: run_read(kw["path"], kw.get("limit")),
"write_file": lambda **kw: run_write(kw["path"], kw["content"]),
"edit_file": lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),
}
# Child gets all base tools except task (no recursive spawning)
CHILD_TOOLS = [
{
"name": "bash", "description": "Run a shell command.",
"input_schema": {
"type": "object", "properties": {
"command": {
"type": "string"}}, "required": ["command"]}},
{
"name": "read_file", "description": "Read file contents.",
"input_schema": {
"type": "object", "properties": {
"path": {
"type": "string"}, "limit": {
"type": "integer"}}, "required": ["path"]}},
{
"name": "write_file", "description": "Write content to file.",
"input_schema": {
"type": "object", "properties": {
"path": {
"type": "string"}, "content": {
"type": "string"}}, "required": ["path", "content"]}},
{
"name": "edit_file", "description": "Replace exact text in file.",
"input_schema": {
"type": "object", "properties": {
"path": {
"type": "string"}, "old_text": {
"type": "string"}, "new_text": {
"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
]
# -- Subagent: fresh context, filtered tools, summary-only return --
def run_subagent(prompt: str) -> str:
sub_messages = [{
"role": "user", "content": prompt}] # fresh context
for _ in range(30): # safety limit
response = client.messages.create(
model=MODEL, system=SUBAGENT_SYSTEM, messages=sub_messages,
tools=CHILD_TOOLS, max_tokens=8000,
)
sub_messages.append({
"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
break
results = []
for block in response.content:
if block.type == "tool_use":
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
results.append({
"type": "tool_result", "tool_use_id": block.id, "content": str(output)[:50000]})
sub_messages.append({
"role": "user", "content": results})
# Only the final text returns to the parent -- child context is discarded
return "".join(b.text for b in response.content if hasattr(b, "text")) or "(no summary)"
# -- Parent tools: base tools + task dispatcher --
PARENT_TOOLS = CHILD_TOOLS + [
{
"name": "task", "description": "Spawn a subagent with fresh context. It shares the filesystem but not conversation history.",
"input_schema": {
"type": "object", "properties": {
"prompt": {
"type": "string"}, "description": {
"type": "string", "description": "Short description of the task"}}, "required": ["prompt"]}},
]
def agent_loop(messages: list):
while True:
response = client.messages.create(
model=MODEL, system=SYSTEM, messages=messages,
tools=PARENT_TOOLS, max_tokens=8000,
)
messages.append({
"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
return
results = []
for block in response.content:
if block.type == "tool_use":
if block.name == "task":
desc = block.input.get("description", "subtask")
print(f"> task ({desc}): {block.input['prompt'][:80]}")
output = run_subagent(block.input["prompt"])
else:
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
print(f" {str(output)[:200]}")
results.append({
"type": "tool_result", "tool_use_id": block.id, "content": str(output)})
messages.append({
"role": "user", "content": results})
if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms04 >> \033[0m")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):
break
history.append({
"role": "user", "content": query})
agent_loop(history)
response_content = history[-1]["content"]
if isinstance(response_content, list):
for block in response_content:
if hasattr(block, "text"):
print(block.text)
print()
