s3：Planning Tool（规划工具）

s05：Planning Tool（规划工具）

解决的问题

---

多轮对话后，Agent 开始出现注意力漂移（忘记最初目标、在细节里打转）。todo_write 把计划显式写进上下文，供 Agent 随时参考，极大提升了处理复杂、多步任务的能力和稳定性。

三重机制

---

1. todo_write 规划工具：一个带有全局状态的工具，把任务清单持久化。
2. Nag Reminder 催办机制：引入主动的、基于规则的干预——在 agent_loop 中维护 rounds_since_todo 计数器，达到阈值就注入提醒，且带双重重置逻辑。
3. 高度结构化与校验的输入：todo_write 的输入不是简单字符串，而是一个严格定义了 status 状态机的对象数组。

工具系统扩展

---

新增 todo_write 这样一个与文件操作完全不同的、有全局状态的工具，只需三步（与 s02 的开闭原则一致）：实现 run_todo_write、在 TOOLS 列表注册、在 TOOL_HANDLERS 添加映射。核心循环 agent_loop 除了催办逻辑，其主体分发代码完全不变。

# todo_write 工具定义
{"name": "todo_write",
"description": "Create and manage a task list ...",
     "input_schema": {
         "type": "object",
         "properties": {
             "todos": {
                 "type": "array",
                 "items": {
                     "type": "object",
                     "properties": {
                         "content": {"type": "string"},
                         "status": {"type": "string", "enum": ["pending", "in_progress", "completed"]},
                     },
                 },
             },
         },
     },
}

status 状态机

状态值	含义
pending	待处理
in_progress	进行中
completed	已完成

规划工具的实现与校验

---

_normalize_todos(todos)

健壮的输入解析：它试图解析模型可能返回的 JSON 字符串或 Python 字面量，并对每一个任务的格式、必要字段、状态值进行严格校验。这保证了 CURRENT_TODOS 状态的纯净和一致性。

# 输入数据的清理与校验守卫：输出严格，接收宽容
def _normalize_todos(todos):
    # 输入宽容，极大提高了工具调用的成功率
    if isinstance(todos, str):
        try:
            # 检查是否是 JSON
            todos = json.loads(todos)
        except json.JSONDecodeError:
            try:
                # 失败尝试 ast 解析
                todos = ast.literal_eval(todos)
            except (SyntaxError, ValueError):
                return None, "Error: todos must be a list or JSON array string"
    # 检查是否是 list
    if not isinstance(todos, list):
        return None, "Error: todos must be a list"
    # 输出严格
    for i, t in enumerate(todos):
        # 字典类型判断
        if not isinstance(t, dict):
            return None, f"Error: todos[{i}] must be an object"
        # 每条必须确保 content 和 status 存在
        if "content" not in t or "status" not in t:
            return None, f"Error: todos[{i}] missing 'content' or 'status'"
        # 且 status 只能是 pending、in_progress、completed 三者之一
        if t["status"] not in ("pending", "in_progress", "completed"):
            return None, f"Error: todos[{i}] has invalid status '{t['status']}'"
    return todos, None

设计哲学：接收宽容，输出严格。模型偶尔会返回 JSON 字符串或 Python 字面量而非规范对象，_normalize_todos 都先试着解析；但一旦解析出来，就按最严的标准校验字段和状态值，确保存入 CURRENT_TODOS 的数据永远合法。

run_todo_write(todos)

def run_todo_write(todos: list) -> str:
    # 全局的、可变的状态，之前所有工具都是无状态的
    global CURRENT_TODOS
    # 通过 _normalize_todos 的校验
    todos, error = _normalize_todos(todos)
    if error:
        return error
    # 更新 todos 任务列表
    CURRENT_TODOS = todos
    # ... 格式化打印到终端 ...
    return f"Updated {len(CURRENT_TODOS)} tasks"

催办机制：Nag Reminder

---

双重重置逻辑：

1. 催办后重置：催办消息发出后，计数器立即归零，防止连续多轮发送催办信息，造成信息冗余。
2. 任务更新后重置：当模型主动使用了 todo_write 后，计数器也立即归零。

def agent_loop(messages: list):
    rounds_since_todo = 0  # 1. 初始化计数器
    while True:
        # 2. 检查计数器，计数达到阈值 3，系统注入一条不可见的用户消息，迫使模型停下当前工作来更新计划
        if rounds_since_todo >= 3 and messages:
            messages.append({"role": "user", "content": "Update your todos."})
            # 催办消息发布后，计数器归零，防止连续多轮发送催办信息
            rounds_since_todo = 0
        # ... LLM 调用 ...
        # 3. 每轮工具调用后，计数器递增
        rounds_since_todo += 1
        # ... 工具执行循环 ...
        for block in response.content:
            if block.type != "tool_use":
                continue
            # ... Hook 和工具执行 ...
            # 4. 关键复位点：如果调用了 todo_write，则重置计数器
            if block.name == "todo_write":
                rounds_since_todo = 0
        messages.append({"role": "user", "content": results})

SYSTEM 提示词演进

---

💡 提示词从 s01 的 "Act, don't explain"，演进为 "plan your steps... Update status as you go"。这反映了 Agent 心智模型的进化：它不再是一个只会执行单步命令的工具，而是一个被期望先规划、后执行、再更新的自主实体。这句提示词是连接全局状态（CURRENT_TODOS）和 LLM 行为模式的桥梁。

SYSTEM = (
    f"You are a coding agent at {WORKDIR}. "
    "Before starting any multi-step task, use todo_write to plan your steps. "
    "Update status as you go."
)

本模块知识点

---

• todo_write 引入全局可变状态 CURRENT_TODOS，把计划显式写进上下文，抑制注意力漂移。
• 输入严格结构化：status 状态机（pending / in_progress / completed），_normalize_todos 宽容输入、严格校验。
• Nag Reminder：rounds_since_todo 计数器双重重置（催办后 / 模型主动更新 todo），强制规划。
• SYSTEM 提示词从 "Act, don't explain" 演进为 "plan your steps... Update status as you go"。

---

s06：Subagent（子代理模式）

核心思想

---

当一个任务过于复杂时，主 Agent 不自己处理所有细节，而是生成一个（spawn）子 Agent，给它：

• 一个明确、独立的任务描述；
• 一个全新的消息历史 messages=[]（与主对话隔离）。

解决的问题

---

• 上下文窗口污染：多步任务产生大量工具调用和输出，塞进主对话历史会迅速耗尽上下文窗口，导致模型遗忘核心目标。
• 注意力分散：强迫主 Agent 关注所有细节会降低其规划和推理能力。将子任务隔离，让主 Agent 只关心子任务的最终结果摘要，保持其思路清晰。

关键设计

---

子 Agent 的整个思考和执行过程（中间的 messages）在完成任务后会被直接丢弃（DISCARDED），只有最终的文本摘要被返回给主 Agent。

能力受限的子代理

• 防止递归：子 Agent 的工具列表（SUB_TOOLS）是主 Agent 工具列表的一个子集。最关键的是，它不包含 task 工具。这从架构上禁止了子 Agent 再创建孙代理（sub-subagent），避免无限递归和失控的资源消耗。
• 最小权限原则：子 Agent 只获得完成其特定任务所需的最小工具集（读写文件、执行命令、查找文件）。

安全护栏与循环限制

• 硬性循环限制：子 Agent 的循环被限制在 30 轮（for _ in range(30)）以内。
• 共享钩子系统：子代理在执行工具时，会触发同样的 PreToolUse 和 PostToolUse 钩子。这意味着在 s04 中为主 Agent 设定的权限检查（如命令黑名单）和日志记录，自动且一致地应用到了子代理身上，没有安全盲区。

优化的结果提取

extract_text() 辅助函数和 spawn_subagent 最后的回退查找逻辑，共同确保即使在非正常结束（如达到 30 轮限制）的情况下，也能尽最大努力从子代理的消息历史中提取出有意义的文本摘要，返回给主 Agent，而不是一个空的结果。

主代理 vs 子代理

---

维度	主 Agent	子 Agent
消息历史	共享完整对话历史	全新 messages=[]，任务结束即丢弃
工具集	完整（含 task、todo_write、load_skill）	子集（不含 task，防递归）
循环上限	无硬上限（受主循环控制）	for _ in range(30) 硬性 30 轮
钩子	触发 PreToolUse/PostToolUse 等	同样触发（权限策略自动生效）
返回值	直接响应用户	仅返回最终文本摘要

子代理工具与系统提示词

---

SYSTEM = (
    f"You are a coding agent at {WORKDIR}. "
    "Complete the task you were given, then return a concise summary. "
    # 不再委派任务
    "Do not delegate further."
)
# 能力受限的工具集，确保子代理的行为符合预期边界。
# NO "task" tool：与 sysprompt 中的不再委派任务形成双重保障
SUB_TOOLS = [
    # ... 包含 bash, read_file, write_file, edit_file, glob ...
]
# 指向了相同的底层函数，但由于工具列表的缺失，子代理即便想调用 `task` 或 `todo_write` 也无从下手
SUB_HANDLERS = {
    "bash": run_bash, "read_file": run_read, # ... 映射到相同的实现函数
}

子代理生成函数：spawn_subagent(description: str) -> str

---

def spawn_subagent(description: str) -> str:
    # 上下文隔离：全新的消息历史，只包含父代理给的任务描述
    messages = [{"role": "user", "content": description}]
    # 安全护栏：硬性循环限制，子代理最多只能循环 30 轮就会强制退出
    for _ in range(30):
        response = client.messages.create(
            model=MODEL,
            # 使用独立的 SYSTEM 提示
            system=SUB_SYSTEM,
            messages=messages,
            # 使用受限的工具集
            tools=SUB_TOOLS,
            max_tokens=8000,
        )
        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":
                # 架构亮点：子代理同样触发钩子，父代理权限策略自动生效
                blocked = trigger_hooks("PreToolUse", block)
                if blocked:
                    results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(blocked),
                    })
                    continue
                handler = SUB_HANDLERS.get(block.name)
                output = handler(**block.input) if handler else f"Unknown: {block.name}"
                trigger_hooks("PostToolUse", block, output)
                print(f"  \033[90m[sub] {block.name}: {str(output)[:100]}\033[0m")
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})
    # 结果提取：只返回摘要，丢弃中间过程
    result = extract_text(messages[-1]["content"])
    if not result:
        # 回退策略：若最后一条消息不是文本（如达到循环上限）则向前查找
        for msg in reversed(messages):
            if msg["role"] == "assistant":
                result = extract_text(msg["content"])
                if result:
                    break
        if not result:
            result = "Subagent stopped after 30 turns without final answer."
    # 只返回最终的摘要字符串，子代理完整的 `messages` 列表（包含所有中间推理和工具调用记录）随着函数返回而**被垃圾回收**。
    return result

extract_text：从内容块提取纯文本

---

因为 Anthropic API 返回的 content 是一个内容块（content block）列表，包含文本块、工具调用块、工具结果块等。extract_text 从混合列表中干净提取所有纯文本内容，拼接成一个字符串。

response.content = [
    ContentBlock(type="text", text="我来帮你查找文件..."),
    ContentBlock(type="tool_use", id="tool_001", name="bash", input={"command": "ls *.py"}),
]

def extract_text(content) -> str:
    """Extract text from message content blocks."""
    if not isinstance(content, list):
        return str(content)
    return "\n".join(getattr(b, "text", "") for b in content if getattr(b, "type", None) == "text")

主 Agent 工具集成

---

父 Agent 通过一个名为 task 的工具来调用子代理：

# 添加 task 任务
TOOLS.append({
    "name": "task",
    "description": "Launch a subagent to handle a complex subtask. Returns only the final conclusion.",
    "input_schema": {"type": "object", "properties": {"description": {"type": "string"}}, "required": ["description"]},
})
TOOL_HANDLERS["task"] = spawn_subagent

本模块知识点

---

• 复杂任务隔离：spawn 子 Agent，给独立任务描述 + 全新 messages=[]，中间过程直接丢弃，只回传最终摘要。
• 防递归：SUB_TOOLS 不含 task 工具，架构上禁止孙代理。
• 最小权限 + 共享钩子：子代理工具是子集，且触发同样的 PreToolUse / PostToolUse 钩子，权限策略自动生效。
• 硬性循环上限 30 轮；extract_text + 回退查找确保非正常结束时也能提取摘要。