From dcd7a83cd0f819f2ec9b4d1e380dda634f8c3ded Mon Sep 17 00:00:00 2001 From: Vink <97326507+Vink567@users.noreply.github.com> Date: Mon, 29 Jun 2026 12:49:26 +0800 Subject: [PATCH] Add Day 10-14 tutorial docs --- docs/day-10-subagents.md | 938 +++++++++++++++++++++++++++++ docs/day-11-context-cost.md | 974 +++++++++++++++++++++++++++++++ docs/day-12-agent-coordinator.md | 801 +++++++++++++++++++++++++ docs/day-13-worktree-demo.md | 704 ++++++++++++++++++++++ docs/day-14-mcp-toolsearch.md | 882 ++++++++++++++++++++++++++++ 5 files changed, 4299 insertions(+) create mode 100644 docs/day-10-subagents.md create mode 100644 docs/day-11-context-cost.md create mode 100644 docs/day-12-agent-coordinator.md create mode 100644 docs/day-13-worktree-demo.md create mode 100644 docs/day-14-mcp-toolsearch.md diff --git a/docs/day-10-subagents.md b/docs/day-10-subagents.md new file mode 100644 index 0000000..7273c2b --- /dev/null +++ b/docs/day-10-subagents.md @@ -0,0 +1,938 @@ +# Day 10：Subagents + +Day 9 我们把“按需知识”接进了 CLI：模型知道有哪些 skill，用到时再 `skill_load`，用户也可以用 `/skill` 把某份流程绑定到一轮任务。 + +但 skill 仍然是在主 Agent 的同一条上下文里工作。读一个大文件、跑一组测试、做一次 review，所有中间过程都会回到主会话里。任务稍微复杂一点，主 Agent 的上下文就会被工具噪声淹掉。 + +今天做 Subagents：主 Agent 可以把一件子任务委派给一个专门的子 Agent。子 Agent 有自己的 system prompt、自己的 fresh messages、自己的工具面；跑完之后只把结论作为 summary 还给主 Agent。 + +跑完之后你会看到： + +- `/agents` 能列出 `.agent/agents/*.md` 里的子 Agent 模板。 +- 模型可以调用 `agent(agent_name, task)` 启动一个同步子 Agent。 +- 子 Agent 的完整过程落到 `.agent/sessions/.../sub-*.jsonl`，主会话只收到一条 summary。 +- 子 Agent 不继承父 messages，但共享 cwd、权限模式和 ESC 中断信号。 +- 子 Agent 的工具池会按模板 `allowed_tools` 收敛，并且永远拿不到 `agent` / `subagent_list`，避免无限递归委派。 + +代码约 360 行，新增代码约 230 行。 + +今天分三版： + +1. v1 做 agent 模板注册、`/agents`、`subagent_list` 和同步 `agent()`。 +2. v2 做 sidechain transcript，让完整子过程落盘，父会话只保留 summary。 +3. v3 做工具收敛和递归拦截，把 Day 9 的 `allowed_tools` 规则复用到子 Agent。 + +## 起手：今天的起点 + +从 Day 9 的 `agent-code` 项目继续改。先准备三个子 Agent 模板： + +```bash +mkdir -p .agent/agents +cat > .agent/agents/code-reviewer.md <<'EOF' +--- +name: code-reviewer +description: Review code changes and report concrete risks, bugs, and missing tests. +allowed_tools: [read_file, grep, git_status, git_diff] +--- + +You are a focused code-review subagent. + +Review only the task you were given. +Prioritize bugs, risky behavior changes, missing tests, and unclear failure modes. +Return a concise summary with findings first. +Do not edit files. +EOF + +cat > .agent/agents/test-writer.md <<'EOF' +--- +name: test-writer +description: Inspect a target module and propose or add focused pytest coverage. +allowed_tools: [read_file, grep, file_write, file_edit] +--- + +You are a focused test-writing subagent. + +Read the relevant implementation and existing tests before writing. +Prefer the smallest test that proves the behavior. +If you edit files, explain what the test covers in the final summary. +EOF + +cat > .agent/agents/debugger.md <<'EOF' +--- +name: debugger +description: Debug a failing command by reading errors, inspecting code, and identifying the smallest fix. +allowed_tools: [read_file, grep, bash] +--- + +You are a focused debugging subagent. + +Start from the failing command or error text. +Inspect only the files needed to explain the failure. +Return the likely root cause and the smallest next fix. +Do not edit files unless the parent task explicitly asks for that. +EOF +``` + +现在项目里多了： + +```txt +.agent/ + agents/ + code-reviewer.md + debugger.md + test-writer.md +``` + +这三个文件和 Day 9 的 `SKILL.md` 很像，也有 `name`、`description`、`allowed_tools`。区别是：skill body 是给主 Agent 临时加载的工作流；agent body 是子 Agent 自己的 system prompt。 + +今天的核心边界先画清楚： + +```mermaid +flowchart TD + A["主 Agent messages"] --> B["tool_call: agent(name, task)"] + B --> C["harness 读取 .agent/agents/name.md"] + C --> D["创建子 Agent fresh messages"] + D --> E["子 Agent 独立跑工具循环"] + E --> F["子 transcript 落 sub-*.jsonl"] + F --> G["summary 作为父 tool_result 回填"] + G --> H["主 Agent 继续思考"] +``` + +注意最关键的一点：子 Agent 不继承父会话 messages。它只收到一条 `task` user message。这样子任务不会把父会话里的讨论、试错和工具结果全背进去。 + +## v1：能发现，也能同步启动 + +先把子 Agent 当成一种本地模板： + +- `.agent/agents/.md` 是模板文件。 +- frontmatter 里的 `name + description` 进入 ``，让模型知道可以委派给谁。 +- body 不进主 Agent prompt，只在真正启动子 Agent 时作为子 system prompt。 + +这和 Day 9 skills 的“目录常驻、正文按需”是同一个思路。不同的是，`agent()` 不是把正文回填给主模型，而是直接启动一条新的 Agent Loop。 + +### 1.1 新增 `agent_code/subagents.py` + +新建 `agent_code/subagents.py`： + +```python +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +from .skills import _parse_allowed_tools, _split_frontmatter, _unquote + + +@dataclass(frozen=True) +class AgentTemplate: + name: str + description: str + allowed_tools: list[str] | None + system_prompt: str + path: Path + + +class AgentRegistry: + def __init__(self, cwd: Path) -> None: + self.cwd = cwd + self.agents_dir = cwd / ".agent" / "agents" + self.warnings: list[str] = [] + + def list(self) -> list[AgentTemplate]: + agents: list[AgentTemplate] = [] + if not self.agents_dir.exists(): + return agents + for path in sorted(self.agents_dir.glob("*.md")): + agent = self._load_file(path) + if agent is not None: + agents.append(agent) + return agents + + def load(self, name: str) -> AgentTemplate | None: + for agent in self.list(): + if agent.name == name: + return agent + return None + + def render_list(self) -> str: + agents = self.list() + if not agents: + return "(no agents found)" + return "\n".join(f"{agent.name} {agent.description}" for agent in agents) + + def render_available_agents(self) -> str: + agents = self.list() + if not agents: + return "" + lines = [""] + lines.extend(f"- {agent.name}: {agent.description}" for agent in agents) + lines.append("") + return "\n".join(lines) + + def _load_file(self, path: Path) -> AgentTemplate | None: + try: + text = path.read_text(encoding="utf-8") + except OSError as exc: + self.warnings.append(f"{path}: {exc}") + return None + + fields, body = _split_frontmatter(text) + name = _unquote(fields.get("name", path.stem)).strip() + description = _unquote(fields.get("description", "")).strip() + if not name or not description: + self.warnings.append(f"{path}: missing name or description") + return None + + return AgentTemplate( + name=name, + description=description, + allowed_tools=_parse_allowed_tools(fields.get("allowed_tools")), + system_prompt=body, + path=path, + ) +``` + +这里复用了 Day 9 的 frontmatter parser。字段语义也保持一致： + +```txt +字段缺失 不收敛工具，继承默认工具面 +allowed_tools: [] 纯文本子 Agent，禁止工具 +allowed_tools: [a,b] 子 Agent 只能用 a / b +``` + +v1 先把字段读出来，v3 再让它变成真实工具边界。 + +### 1.2 把可用 agent 目录注入 system prompt + +打开 `agent_code/agent.py`，找到 `build_system_prompt()` 里注入 `available_skills` 的那段。 + +在 `available_skills` 之后、`output_style` 之前追加： + +```python + from .subagents import AgentRegistry + + available_agents = AgentRegistry(cwd).render_available_agents() + if available_agents: + # 这里只放子 Agent 目录，不放子 Agent system prompt。 + parts.append(available_agents) +``` + +这一段最终顺序大概是： + +```txt +core prompt +AGENT.md +project-memory +available-skills +available-agents +output-style +``` + +`available-agents` 只是一张目录卡片。模型知道有 `code-reviewer`、`test-writer`、`debugger`，但看不到它们完整 prompt。真正 spawn 时才加载 body。 + +### 1.3 新增 `/agents` + +打开 `agent_code/slash.py`，在 `_cmd_skills` 后面新增： + +```python +def _cmd_agents(_args: list[str], ctx: SlashContext) -> SlashResult: + from .subagents import AgentRegistry + + registry = AgentRegistry(ctx.cwd) + message = registry.render_list() + if registry.warnings: + message += "\n\nwarnings:\n" + "\n".join(f"- {w}" for w in registry.warnings) + return SlashResult(handled=True, message=message) +``` + +再在底部注册，放在 `/skills` 附近： + +```python +register("agents", "列出本地 .agent/agents 里的子 Agent 模板", _cmd_agents) +``` + +先跑一个本地验证： + +```bash +$ uv run agent-code "/agents" +code-reviewer Review code changes and report concrete risks, bugs, and missing tests. +debugger Debug a failing command by reading errors, inspecting code, and identifying the smallest fix. +test-writer Inspect a target module and propose or add focused pytest coverage. +``` + +这条命令和 `/skills` 一样，不进入模型，只读本地模板目录。 + +### 1.4 给子 Agent 一条 runner + +现在要让 `agent()` 工具真的能启动子 Agent。先加一个 runner，v1 只做同步执行，不落子 transcript。 + +新建 `agent_code/subagent_runner.py`： + +```python +from __future__ import annotations + +from pathlib import Path + +from .agent import build_system_prompt, run_agent +from .model import create_provider +from .runtime import RuntimeState +from .subagents import AgentRegistry, AgentTemplate +from .tools import ToolContext, ToolRegistry, default_tools + + +RECURSIVE_AGENT_TOOLS = frozenset({"agent", "subagent_list"}) + + +def build_subagent_system_prompt(cwd: Path, state: RuntimeState, template: AgentTemplate) -> str: + base = build_system_prompt(cwd, state) + return ( + f"{base}\n\n" + f"\n" + f"{template.system_prompt}\n" + f"\n\n" + "You are running as an isolated subagent. Work only on the task you receive. " + "Return a concise final summary for the parent agent." + ) + + +def _child_allowed_tool_names(template: AgentTemplate, tools: ToolRegistry) -> list[str]: + all_names = [tool.name for tool in tools.list() if tool.name not in RECURSIVE_AGENT_TOOLS] + if template.allowed_tools is None: + return all_names + return [name for name in template.allowed_tools if name in all_names] + + +def _child_state(parent_state: RuntimeState | None) -> RuntimeState: + parent_state = parent_state or RuntimeState() + state = RuntimeState( + permission_mode=parent_state.permission_mode, + model=parent_state.model, + provider=parent_state.provider, + base_url=parent_state.base_url, + ) + # ESC 应该能让父 turn 和正在跑的子 turn 都在步间停下来。 + state.abort_event = parent_state.abort_event + return state + + +def run_subagent(agent_name: str, task: str, ctx: ToolContext, max_steps: int = 6) -> str: + registry = AgentRegistry(ctx.cwd) + template = registry.load(agent_name) + if template is None: + return f"error: agent not found: {agent_name}" + + state = _child_state(ctx.runtime_state) + all_tools = default_tools() + allowed_names = _child_allowed_tool_names(template, all_tools) + state.skill_allowed_tools = allowed_names + child_tools = all_tools.filtered(allowed_names) + + provider = create_provider(state.provider, state.model, state.base_url) + system_prompt = build_subagent_system_prompt(ctx.cwd, state, template) + result = run_agent( + task, + provider, + child_tools, + max_steps=max_steps, + cwd=ctx.cwd, + state=state, + session=None, + system_prompt=system_prompt, + ) + return result.final or "(subagent returned no final text)" +``` + +这里有三个边界值得停一下： + +- `messages` 是 fresh 的：`run_agent()` 没有拿父 session，所以它从 `task` 这一条 user message 冷启动。 +- `state` 不是父对象本身：子 Agent 共享权限模式、模型和 abort 信号，但不共享父 todo、output-style、临时 skill 白名单。 +- 工具池先剔除 `agent` / `subagent_list`：哪怕模板写错，子 Agent 也不能再启动孙 Agent。 + +不过这段代码引用了 `RuntimeState.base_url`。Day 9 的 `RuntimeState` 还没有这个字段，先补上。 + +打开 `agent_code/runtime.py`，在 `provider` 后面加一行： + +```python + provider: str = "anthropic" + base_url: str | None = None +``` + +再打开 `agent_code/cli.py`，两处创建 `RuntimeState` 的地方都把 `base_url` 传进去。 + +`run_once()` 里： + +```python + state = RuntimeState( + permission_mode=permission_mode, + model=model, + provider=provider_name, + base_url=base_url, + ) +``` + +交互模式里： + +```python + state = RuntimeState(permission_mode=permission_mode, model=model, provider=provider, base_url=base_url) +``` + +`run_turn()` 里也把 provider 创建改成读状态： + +```python + turn_provider = create_provider(state.provider, state.model, state.base_url) +``` + +这样子 Agent 和主 Agent 会使用同一套 provider/model/base URL。否则你用 `--base-url` 起了主 CLI，子 Agent 却会退回环境变量默认值，很难排查。 + +### 1.5 注册 `subagent_list` 和 `agent` + +打开 `agent_code/tools.py`，在 `skill_load` 后面新增两个工具函数： + +```python +def subagent_list(args: dict[str, Any], ctx: ToolContext) -> str: + """给模型看的子 Agent 目录；和 /agents 共用同一份 registry。""" + from .subagents import AgentRegistry + + return AgentRegistry(ctx.cwd).render_list() + + +def agent(args: dict[str, Any], ctx: ToolContext) -> str: + """同步启动一个子 Agent。完整隔离边界在 subagent_runner.py。""" + from .subagent_runner import run_subagent + + agent_name = str(args.get("agent_name", "")).strip() + task = str(args.get("task", "")).strip() + if not agent_name: + return "error: missing required argument 'agent_name'" + if not task: + return "error: missing required argument 'task'" + max_steps = max(1, min(int(args.get("max_steps", 6)), 12)) + return run_subagent(agent_name, task, ctx, max_steps=max_steps) +``` + +再在 `default_tools()` 里，放在 `skill_load` 注册后面： + +```python + registry.register( + Tool( + name="subagent_list", + description="List available local subagent templates with their descriptions.", + run=subagent_list, + parameters={"type": "object", "properties": {}, "required": []}, + is_read_only=True, + ) + ) + registry.register( + Tool( + name="agent", + description=( + "Run a focused subagent by template name. Use it for isolated review, " + "debugging, or test-writing tasks. The subagent receives only the task, " + "not the parent conversation." + ), + run=agent, + parameters={ + "type": "object", + "properties": { + "agent_name": {"type": "string", "description": "Subagent template name."}, + "task": {"type": "string", "description": "Focused task for the subagent."}, + "max_steps": { + "type": "integer", + "description": "Maximum subagent loop steps, default 6.", + "default": 6, + }, + }, + "required": ["agent_name", "task"], + }, + is_read_only=False, + ) + ) +``` + +最后让权限层认识这两个工具。打开 `agent_code/permissions.py`： + +把 `subagent_list` 加进 `_READONLY_TOOLS`： + +```python + "skill_list", "skill_load", "subagent_list", +``` + +把 `agent` 加进 `_LOW_RISK_WRITES`： + +```python + "todo_write", "enter_plan_mode", "exit_plan_mode", + "agent", +``` + +`agent` 本身只是启动子 loop，不直接写文件。子 loop 里的 `file_write`、`file_edit`、`bash` 仍然会走自己的权限判断，所以这里可以放行委派动作本身。 + +### 1.6 跑验证 + +先用本地结构检查，不依赖真实模型： + +```bash +$ uv run python - <<'PY' +from pathlib import Path +from agent_code.subagents import AgentRegistry +from agent_code.tools import ToolContext, default_tools +from agent_code.runtime import RuntimeState + +cwd = Path.cwd() +registry = AgentRegistry(cwd) +print([a.name for a in registry.list()]) + +tools = default_tools() +print(tools.get("subagent_list") is not None) +print(tools.get("agent") is not None) + +ctx = ToolContext(cwd=cwd, runtime_state=RuntimeState(provider="mock", model="mock")) +print(tools.get("subagent_list").run({}, ctx).splitlines()[0]) +PY +['code-reviewer', 'debugger', 'test-writer'] +True +True +code-reviewer Review code changes and report concrete risks, bugs, and missing tests. +``` + +再跑真实模型链路： + +```bash +$ uv run agent-code --max-steps 8 "先调用 subagent_list，然后用 agent 工具让 code-reviewer 看一下当前 git diff，只返回最重要的风险" +Agent Code +cwd: /your/project +provider: anthropic model: deepseek-v4-pro + +tool_call: subagent_list {} +tool_call: agent {'agent_name': 'code-reviewer', 'task': 'Review the current git diff and report the most important risks.'} +final: ... +``` + +你看到的父 trace 里只有 `tool_call: agent ...` 和最终回答。子 Agent 里面具体调了 `git_diff`、`read_file` 还是 `grep`，v1 暂时不会落盘。下一版补上。 + +## v2：完整过程落子 transcript，父会话只看 summary + +v1 已经能同步启动子 Agent，但还有个问题：子 Agent 的中间过程不落盘。出了问题你没法复盘它到底读了什么、用了什么工具、为什么给出那个 summary。 + +我们要把子过程写进 sidechain transcript： + +```txt +父 session: + user: 请 review + assistant: tool_use agent(...) + user: tool_result "" + assistant: final + +子 session: + user: Review the current git diff... + assistant: tool_use git_diff + user: tool_result ... + assistant: final "" +``` + +父会话只保留 summary，是为了降噪；子 transcript 另存，是为了可查。 + +### 2.1 `Session.create()` 支持 prefix + +打开 `agent_code/session.py`，把 `create()` 方法改成： + +```python + @classmethod + def create(cls, cwd: Path, prefix: str = "") -> "Session": + """新建会话：生成 12 位 hex session_id，创建空 JSONL 文件。""" + sid = prefix + uuid.uuid4().hex[:12] + file_path = _sessions_dir(cwd) / f"{sid}.jsonl" + file_path.touch() + return cls(cwd=cwd, session_id=sid, file_path=file_path, resumed=False) +``` + +主会话仍然用： + +```python +Session.create(resolved_cwd) +``` + +子会话会用： + +```python +Session.create(ctx.cwd, prefix="sub-") +``` + +路径仍然沿用 Day 6 的 session 目录： + +```txt +.agent/sessions//sub-xxxxxxxxxxxx.jsonl +``` + +### 2.2 `run_subagent()` 带上子 session + +打开 `agent_code/subagent_runner.py`，顶部加一行： + +```python +from .session import Session +``` + +然后把 `run_agent(... session=None ...)` 那段改成： + +```python + session = Session.create(ctx.cwd, prefix="sub-") + result = run_agent( + task, + provider, + child_tools, + max_steps=max_steps, + cwd=ctx.cwd, + state=state, + session=session, + system_prompt=system_prompt, + ) + return result.final or "(subagent returned no final text)" +``` + +这一步没有把子 transcript 路径塞回父 tool_result。父会话仍然只收到 summary。你要查细节时，直接看 `.agent/sessions/.../sub-*.jsonl`。 + +为什么不把完整 transcript 放进父 tool_result？因为那会破坏 subagent 的意义：主 Agent 又被子过程的工具结果塞满了。 + +### 2.3 跑验证 + +先删掉旧的 sub session，方便看这次生成了什么： + +```bash +rm -f .agent/sessions/*/sub-*.jsonl +``` + +然后跑一次子 Agent： + +```bash +$ uv run agent-code --max-steps 8 "用 agent 工具让 code-reviewer 总结当前项目顶层结构里最值得注意的一点" +Agent Code +cwd: /your/project +provider: anthropic model: deepseek-v4-pro + +tool_call: agent {'agent_name': 'code-reviewer', 'task': 'Summarize the most important thing to notice about the project top-level structure.'} +final: ... +``` + +看子 transcript： + +```bash +$ ls .agent/sessions/*/sub-*.jsonl +.agent/sessions/.../sub-a1b2c3d4e5f6.jsonl + +$ head -n 3 .agent/sessions/*/sub-*.jsonl +{"role":"user","content":"Summarize the most important thing to notice about the project top-level structure.","timestamp":"..."} +... +``` + +父输出没有打印子过程细节，但磁盘上有完整 JSONL。这个就是今天的降噪边界。 + +## v3：工具收敛和递归拦截 + +现在模板里的 `allowed_tools` 已经被 runner 读到了，但我们要明确验证它真的生效。 + +Day 10 沿用 Day 9 的双保险： + +1. 给子模型看的工具 schema 先过滤。 +2. 权限层再用同一份白名单兜底。 + +同时，子 Agent 永远不能调用 `agent` / `subagent_list`。这不是靠 prompt 提醒，而是在工具池里直接剔除。 + +### 3.1 看懂 `run_subagent()` 里的三层过滤 + +回到 `agent_code/subagent_runner.py` 这一段： + +```python + all_tools = default_tools() + allowed_names = _child_allowed_tool_names(template, all_tools) + state.skill_allowed_tools = allowed_names + child_tools = all_tools.filtered(allowed_names) +``` + +这里同时做了两件事： + +- `child_tools` 是给 provider 的 schema 面。模型正常看不到越界工具。 +- `state.skill_allowed_tools` 是权限兜底。即使模型从旧上下文或幻觉里发出越界 tool call，`decide_permission()` 也会 deny。 + +而 `_child_allowed_tool_names()` 里先把递归工具剔掉： + +```python +all_names = [tool.name for tool in tools.list() if tool.name not in RECURSIVE_AGENT_TOOLS] +``` + +所以模板就算写成这样： + +```yaml +allowed_tools: [agent, read_file] +``` + +实际给子 Agent 的也只会剩下： + +```txt +read_file +``` + +### 3.2 为什么要禁递归 + +如果子 Agent 能继续调 `agent()`，模型很容易生成这种链： + +```txt +main -> agent(code-reviewer) +code-reviewer -> agent(debugger) +debugger -> agent(test-writer) +... +``` + +这会让成本、transcript 归属和权限白名单都变得不可解释。今天的目标不是多 Agent 社会，而是“一次性委派，跑完回 summary”。 + +多 Agent 协作会放到 Day 12 的 coordinator。今天先把单层委派边界锁死。 + +### 3.3 跑两个验证 + +第一个验证 `code-reviewer` 的工具面。它只允许： + +```txt +read_file, grep, git_status, git_diff +``` + +直接用本地 Python 看过滤结果： + +```bash +$ uv run python - <<'PY' +from pathlib import Path +from agent_code.subagents import AgentRegistry +from agent_code.subagent_runner import _child_allowed_tool_names +from agent_code.tools import default_tools + +cwd = Path.cwd() +template = AgentRegistry(cwd).load("code-reviewer") +names = _child_allowed_tool_names(template, default_tools()) +print(names) +print("agent" in names) +print("subagent_list" in names) +PY +['read_file', 'grep', 'git_status', 'git_diff'] +False +False +``` + +第二个验证权限兜底。即使子 Agent 试图调用 `file_edit`，也会被白名单挡住： + +```bash +$ uv run python - <<'PY' +from pathlib import Path +from agent_code.permissions import PermissionRequest, decide_permission + +decision = decide_permission(PermissionRequest( + tool_name="file_edit", + args={"file_path": "agent_code/agent.py"}, + mode="default", + cwd=Path.cwd(), + allowed_tools=["read_file", "grep", "git_status", "git_diff"], +)) +print(decision.behavior) +print(decision.message) +PY +deny +skill allowed_tools does not allow file_edit +``` + +消息里仍然写着 `skill allowed_tools`，因为 Day 9 的字段名就是这么接进权限层的。你可以把文案改成更通用的 `allowed_tools does not allow ...`，但不影响今天的行为。 + +最后跑一个真实任务： + +```bash +$ uv run agent-code --max-steps 10 "请用 agent 工具让 code-reviewer review 当前 diff。主回答只给它的 summary。" +Agent Code +cwd: /your/project +provider: anthropic model: deepseek-v4-pro + +tool_call: agent {'agent_name': 'code-reviewer', 'task': 'Review the current diff. Return concrete risks first.'} +final: ... +``` + +如果模型没有主动调用 `agent`，把 prompt 写得更硬一点： + +```bash +uv run agent-code --max-steps 10 "必须调用 agent 工具，agent_name=code-reviewer，task=Review the current git diff and summarize findings." +``` + +今天要验证的是 harness 边界，不是考模型自觉。 + +### 3.4 还有一种自动 fork + +今天做的是手动 subagent：主模型明确调用 `agent()`，子 Agent 跑完，把 summary 作为工具结果还回来。 + +真实长会话里还会有另一类自动 fork。比如 compact、记忆抽取、进度摘要这些维护任务，不一定是用户或模型显式委派，而是 harness 自己开一条 cache-safe 子循环去整理上下文。那类 fork 追求的是不破坏主请求的缓存前缀、少引入工具副作用，结果通常服务 compact 或 memory，不作为普通 `agent()` 工具结果回填。 + +我们今天不实现它。先把用户可见的一次性委派讲清楚：fresh task、隔离 transcript、summary 回填。Day 11 做上下文和成本时，再把这种自动维护循环接回来。 + +## 收尾：今天改了哪些文件 + +今天新增两个文件： + +```txt +agent_code/subagents.py +agent_code/subagent_runner.py +``` + +今天改了七个已有文件： + +```txt +agent_code/runtime.py RuntimeState 加 base_url，子 Agent 复用 provider 配置 +agent_code/agent.py system prompt 注入 available-agents +agent_code/slash.py 新增 /agents +agent_code/tools.py 新增 subagent_list / agent 两个工具 +agent_code/permissions.py 放行 subagent_list / agent，并沿用 allowed_tools 兜底 +agent_code/session.py Session.create(prefix="sub-") 支持子 transcript +agent_code/cli.py RuntimeState 记录 base_url，run_turn 按 state.base_url 建 provider +``` + +如果你想做一次完整手动验证，按这个顺序： + +```bash +$ uv run agent-code "/agents" +code-reviewer Review code changes and report concrete risks, bugs, and missing tests. +debugger Debug a failing command by reading errors, inspecting code, and identifying the smallest fix. +test-writer Inspect a target module and propose or add focused pytest coverage. + +$ uv run python - <<'PY' +from pathlib import Path +from agent_code.subagents import AgentRegistry +from agent_code.subagent_runner import _child_allowed_tool_names +from agent_code.tools import default_tools + +template = AgentRegistry(Path.cwd()).load("code-reviewer") +print(_child_allowed_tool_names(template, default_tools())) +PY +['read_file', 'grep', 'git_status', 'git_diff'] + +$ uv run agent-code --max-steps 10 "必须调用 agent 工具，agent_name=code-reviewer，task=Review the current git diff and summarize findings." +... +``` + +跑完后检查： + +```bash +ls .agent/sessions/*/sub-*.jsonl +``` + +能看到 `sub-*.jsonl`，说明 sidechain transcript 已经落盘。 + +## 手动 trace 一遍 + +### 路径一：`/agents` + +```txt +用户输入：/agents +1. cli.py / interactive.py 先走 dispatch_slash，不进入模型。 +2. slash.py 调 AgentRegistry(ctx.cwd).render_list()。 +3. AgentRegistry 扫 .agent/agents/*.md。 +4. 终端打印 name + description。 +5. 本轮结束，没有 LLM call，也没有 session 新消息。 +``` + +### 路径二：模型自己委派子 Agent + +```txt +用户输入：请让 code-reviewer review 当前 diff +1. build_system_prompt 注入 目录卡片。 +2. provider.complete 看到 agent / subagent_list 两个工具。 +3. 主模型发 tool_call: agent {"agent_name":"code-reviewer","task":"..."}。 +4. tools.py 调 run_subagent()。 +5. subagent_runner 读取 code-reviewer.md，创建 child RuntimeState。 +6. child messages 从 [{"role":"user","content":task}] 冷启动，不继承父 messages。 +7. child tools 按 allowed_tools 过滤，并剔除 agent / subagent_list。 +8. child run_agent 同步跑完，完整 messages 落 sub-*.jsonl。 +9. run_subagent 返回 child final summary。 +10. 父 Agent 把 summary 当作普通 tool_result 回填，再继续回答。 +``` + +### 路径三：子 Agent 试图越界用工具 + +```txt +1. code-reviewer 模板 allowed_tools = [read_file, grep, git_status, git_diff]。 +2. subagent_runner 把 provider 可见工具过滤到这四个。 +3. 如果模型仍然发出 file_edit，execute_one_tool_call 会构造 PermissionRequest。 +4. PermissionRequest.allowed_tools 不包含 file_edit。 +5. decide_permission 返回 deny。 +6. 子 transcript 记录这次 deny，父会话仍只拿最终 summary。 +``` + +## 今天有了什么 + +- **Agent 模板注册**：`.agent/agents/*.md` 变成可发现的本地子 Agent 模板。 +- **``**：主 Agent 只常驻 name + description，不吞子 prompt 正文。 +- **同步 `agent()` 工具**：主 Agent 可以把一个任务委派给子 Agent，并等待 summary 回来。 +- **fresh 子上下文**：子 Agent 不继承父 messages，只收到自己的 task。 +- **sidechain transcript**：子过程完整落 `sub-*.jsonl`，父会话只保留 summary。 +- **工具收敛双保险**：子工具池先过滤，权限层再兜底。 +- **递归拦截**：子 Agent 永远拿不到 `agent` / `subagent_list`。 + +## 常见问题 + +### `/agents` 显示 `(no agents found)` + +确认你在项目根目录运行，且模板路径是： + +```txt +.agent/agents/code-reviewer.md +``` + +Day 10 用的是 `.md` 文件，不是 `.agent/agents/code-reviewer/AGENT.md` 目录。 + +### `agent` 工具返回 `agent not found` + +真正匹配的是 frontmatter 里的 `name`，不是文件名。检查： + +```yaml +--- +name: code-reviewer +description: ... +--- +``` + +### 子 Agent 没有用到你想要的工具 + +先看模板里的 `allowed_tools`。如果 `code-reviewer` 只允许 `git_diff` / `read_file` / `grep`，它就看不到 `bash` 和 `file_edit`。 + +要让 `debugger` 能跑命令，就把任务交给 `debugger`，或者在对应模板里加 `bash`。 + +### 为什么父会话里看不到子 Agent 的工具细节 + +这是有意的。父会话只需要 summary，否则 subagent 就失去了降噪价值。 + +细节在： + +```txt +.agent/sessions//sub-*.jsonl +``` + +### 子 Agent 能不能后台跑 + +今天不做。Day 10 的 `agent()` 是同步工具：父 Agent 等它跑完再继续。 + +后台 task、`task_output`、`task_stop` 会放到 Day 12。现在先把“隔离上下文 + summary 回填 + transcript 可查”这条边界跑稳。 + +### 为什么不让子 Agent 继续调用 `agent` + +递归委派会让成本、权限和 transcript 归属都变得不可控。今天只做单层委派。多个 Agent 协作会在 Day 12 用 coordinator 重新设计。 + +## 课后挑战 + +1. **更友好的 transcript 提示**：让 `agent()` 的 summary 后面附一行本地 transcript 路径，但不要把完整 transcript 放回父会话。 +2. **模板校验**：给 `/agents` 加 `--verbose`，列出缺少 `name` / `description` 的坏模板。 +3. **只读子 Agent 快捷模板**：写一个 `doc-reader`，`allowed_tools: [read_file, grep, project_tree]`，专门做文档归纳。 +4. **子 Agent 最大步数策略**：按模板加 `max_steps` frontmatter，避免每次都由工具参数传。 +5. **更通用的权限文案**：把 `skill allowed_tools does not allow ...` 改成 `allowed_tools does not allow ...`，让 skill 和 subagent 共用同一条错误信息。 + +## 思考题 + +1. **子 Agent 为什么不继承父 messages？** 提示：父会话里哪些内容对子任务有用，哪些只是噪声？ +2. **summary 回填父会话，完整 transcript 另存，这个边界解决了什么问题？** 如果把子 Agent 所有消息都塞回父会话，会发生什么？ +3. **子 Agent 共享 cwd 和权限模式，但不共享父 `RuntimeState` 整个对象。** 这避免了哪些状态污染？ +4. **为什么要先过滤工具 schema，再在权限层兜底？** 只做其中一层分别会漏掉什么？ + +## 下一天 + +今天我们把复杂子任务隔离出去了：主 Agent 发起委派，子 Agent fresh loop 跑完，只把 summary 回来。 + +下一天会处理更大的上下文问题：长会话会爆 token，工具结果会撑满窗口，compact 不能只靠“消息太多就截掉”。Day 11 做 Context + Cost，让 harness 开始主动管理上下文层级、token 预算和压缩策略。 diff --git a/docs/day-11-context-cost.md b/docs/day-11-context-cost.md new file mode 100644 index 0000000..d0ae85a --- /dev/null +++ b/docs/day-11-context-cost.md @@ -0,0 +1,974 @@ +# Day 11：Context + Cost，上下文终于归 harness 管 + +Day 10 以后，主 Agent 已经能把复杂子任务委派给 subagent。主会话轻了很多，但另一个问题变得更明显：长会话总会爆上下文。 + +前面我们在 Day 6 做过一个很粗的 compact：`messages` 超过 40 条就压一下。它能救急，但解释不了这几个问题： + +- 到底是谁把上下文撑大的？ +- 工具结果太长时，是删掉，还是留预览？ +- compact 前能不能先备份？ +- prompt 太长报错时，harness 能不能自己压缩后重试？ + +今天让 harness 接管上下文和成本。跑完之后你会看到： + +- `/cost` 能显示本轮 token / 美元成本，并按 model、tool 粗略归因。 +- `/context` 能把上下文拆成 `Pinned / Working / Compressed` 三层。 +- micro compact 只替换旧 `tool_result` 正文，不破坏 `tool_use_id` 配对。 +- auto compact 按 token/USD 阈值触发，压缩前备份 transcript。 +- `/compact --dry-run` 先预览，`/compact --apply` 再执行。 +- 大工具结果落到 `.agent/tool-results/`，上下文里只留 preview 和路径。 +- 429/529、prompt 太长这类错误有一个最小 recovery 流程。 + +代码约 900 行，新增约 520 行。Day 11 是重型天，版本会多一点，但每一版都能单独验证。 + +今天分七段： + +1. v0 做 `CostTracker` 和 `/cost`。 +2. v1 做三层 `/context`。 +3. v2 做 micro compact。 +4. v3 做 auto compact + transcript 备份。 +5. v4 做 `compact()` 工具和 `/compact --dry-run|--apply`。 +6. 收尾 a 做工具结果预算和溢出落盘。 +7. 收尾 b 做最小 recovery。 + +## 起手：今天的起点 + +Day 10 的 `agent-code` 已经有这些东西： + +```txt +agent.py run_agent / build_system_prompt / tool loop +model.py AnthropicProvider / ModelResponse +session.py JSONL session +tools.py ToolRegistry + file/bash/web/skill/subagent 工具 +compact_basic.py Day 6 的简化 compact +slash.py /context /compact 目前还是轻量占位 +``` + +今天不改工具能力本身，改的是“工具和模型调用产生的信息怎么被计量、压缩、备份和恢复”。 + +先把今天新增文件列出来： + +```txt +agent_code/cost_prices.py +agent_code/cost.py +agent_code/context.py +agent_code/transcript.py +agent_code/token_budget.py +agent_code/compactor.py +agent_code/recovery.py +agent_code/tool_results.py +``` + +它们的职责要分清： + +- `cost.py` 只管 usage 计费和归因。 +- `context.py` 只管 Pinned / Working / Compressed 分类和估算。 +- `compactor.py` 只管 micro / auto / manual compact。 +- `tool_results.py` 只管大结果落盘和 preview。 +- `recovery.py` 只管 provider 调用失败后怎么补救。 + +## v0：先把成本看见 + +auto compact 不能靠感觉。第一步是让每次模型调用都有 usage，能累计到 session 里。 + +### 0.1 `ModelResponse` 加 usage + +打开 `agent_code/model.py`，新增一个 dataclass： + +```python +@dataclass +class Usage: + input_tokens: int = 0 + output_tokens: int = 0 + cache_read_input_tokens: int = 0 + cache_creation_input_tokens: int = 0 +``` + +然后给 `ModelResponse` 加字段： + +```python +@dataclass +class ModelResponse: + text: str | None = None + tool_calls: list[ToolCall] | None = None + assistant_content: list[dict[str, Any]] | None = None + stop_reason: str = "end_turn" + usage: Usage | None = None +``` + +`MockProvider` 可以不返回 usage。真实 provider 里，在 `response = self.client.messages.create(...)` 后解析： + +```python +usage = Usage( + input_tokens=getattr(response.usage, "input_tokens", 0), + output_tokens=getattr(response.usage, "output_tokens", 0), + cache_read_input_tokens=getattr(response.usage, "cache_read_input_tokens", 0), + cache_creation_input_tokens=getattr(response.usage, "cache_creation_input_tokens", 0), +) +``` + +最后返回 `ModelResponse(..., usage=usage)`。 + +如果某个兼容 endpoint 没有 usage，`usage=None`。后面 `CostTracker` 会退回 `chars/4` 估算。 + +### 0.2 新增价格表 + +新建 `agent_code/cost_prices.py`： + +```python +from __future__ import annotations + + +# 单位：每 100 万 token 美元。教学版只放常见模型，未知模型按 deepseek 默认价估算。 +MODEL_PRICES_USD_PER_MTOKENS: dict[str, tuple[float, float]] = { + "deepseek-v4-pro": (2.0, 8.0), + "deepseek-v4-flash": (0.27, 1.10), + "claude-sonnet-4-5": (3.0, 15.0), + "claude-haiku-4-5": (0.80, 4.0), +} + + +def price_for_model(model: str) -> tuple[float, float]: + if model in MODEL_PRICES_USD_PER_MTOKENS: + return MODEL_PRICES_USD_PER_MTOKENS[model] + if "flash" in model: + return MODEL_PRICES_USD_PER_MTOKENS["deepseek-v4-flash"] + return MODEL_PRICES_USD_PER_MTOKENS["deepseek-v4-pro"] +``` + +这不是账单级精确计费，只是让你看到趋势：哪个模型贵、哪类工具结果让下一轮变贵。 + +### 0.3 新增 `agent_code/cost.py` + +```python +from __future__ import annotations + +from dataclasses import dataclass, field + +from .cost_prices import price_for_model +from .model import Usage + + +@dataclass +class CostBucket: + input_tokens: int = 0 + output_tokens: int = 0 + usd: float = 0.0 + + def add(self, input_tokens: int, output_tokens: int, usd: float) -> None: + self.input_tokens += input_tokens + self.output_tokens += output_tokens + self.usd += usd + + +@dataclass +class CostTracker: + total: CostBucket = field(default_factory=CostBucket) + by_model: dict[str, CostBucket] = field(default_factory=dict) + by_tool: dict[str, CostBucket] = field(default_factory=dict) + + def record( + self, + model: str, + usage: Usage | None, + fallback_chars: int, + previous_tools: list[str], + ) -> None: + if usage is None: + input_tokens = max(1, fallback_chars // 4) + output_tokens = 0 + else: + input_tokens = usage.input_tokens + usage.cache_read_input_tokens + usage.cache_creation_input_tokens + output_tokens = usage.output_tokens + + in_price, out_price = price_for_model(model) + usd = (input_tokens / 1_000_000) * in_price + (output_tokens / 1_000_000) * out_price + + self.total.add(input_tokens, output_tokens, usd) + self.by_model.setdefault(model, CostBucket()).add(input_tokens, output_tokens, usd) + + names = previous_tools or ["_initial"] + share_input = input_tokens // len(names) + share_output = output_tokens // len(names) + share_usd = usd / len(names) + for name in names: + self.by_tool.setdefault(name, CostBucket()).add(share_input, share_output, share_usd) + + def render(self) -> str: + lines = [ + f"total: {self.total.input_tokens} input / {self.total.output_tokens} output / ${self.total.usd:.4f}", + "", + "by model:", + ] + for model, bucket in sorted(self.by_model.items()): + lines.append(f" {model}: {bucket.input_tokens}+{bucket.output_tokens} tokens / ${bucket.usd:.4f}") + lines.append("") + lines.append("by tool (rough API-round attribution):") + for tool, bucket in sorted(self.by_tool.items(), key=lambda item: item[1].usd, reverse=True): + lines.append(f" {tool}: {bucket.input_tokens}+{bucket.output_tokens} tokens / ${bucket.usd:.4f}") + return "\n".join(lines) +``` + +`by_tool` 是粗归因。规则是：一次模型调用的成本，平摊给上一轮所有 `tool_use`。因为工具结果是在下一次模型调用时才真正进入上下文。 + +### 0.4 `RuntimeState` 挂上 cost + +打开 `agent_code/runtime.py`，在 `RuntimeState` 里新增： + +```python + cost_tracker: "CostTracker | None" = None +``` + +为了避免循环 import，可以在文件顶部加： + +```python +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .cost import CostTracker +``` + +然后在 `cli.py` 创建 `RuntimeState` 后初始化： + +```python +from .cost import CostTracker + +state.cost_tracker = CostTracker() +``` + +one-shot 和交互模式两处都要做。 + +### 0.5 `run_agent()` 记录 usage + +打开 `agent_code/agent.py`，在 loop 外加： + +```python +last_round_tool_names: list[str] = [] +``` + +每次 provider 返回后立刻记录： + +```python +if state.cost_tracker is not None: + fallback_chars = sum(len(str(m.get("content", ""))) for m in messages) + state.cost_tracker.record( + model=state.model, + usage=response.usage, + fallback_chars=fallback_chars, + previous_tools=last_round_tool_names, + ) +``` + +当这一轮有工具调用时，更新： + +```python +last_round_tool_names = [call.name for call in response.tool_calls or []] +``` + +当没有工具调用、准备 final 时，可以清空： + +```python +last_round_tool_names = [] +``` + +### 0.6 `/cost` + +打开 `agent_code/slash.py`，新增： + +```python +def _cmd_cost(_args: list[str], ctx: SlashContext) -> SlashResult: + if ctx.state is None or ctx.state.cost_tracker is None: + return SlashResult(handled=True, message="cost: no tracker for this run") + return SlashResult(handled=True, message=ctx.state.cost_tracker.render()) +``` + +底部注册： + +```python +register("cost", "显示 token / USD 成本估算", _cmd_cost) +``` + +跑一下： + +```bash +$ uv run agent-code +> 今天几号？请用 system_date 工具回答 +tool_call: system_date {} +final: ... +> /cost +total: 1234 input / 120 output / $0.0034 + +by model: + deepseek-v4-pro: 1234+120 tokens / $0.0034 + +by tool (rough API-round attribution): + _initial: ... + system_date: ... +``` + +数字会不一样。只要能看到 total、by model、by tool，v0 就通了。 + +## v1：把上下文拆成三层 + +现在有成本了，下一步是把上下文分层。我们不再只说“messages 太多”，而是问：哪些必须 pinned？哪些是最近 working？哪些可以 compressed？ + +### 1.1 新增 `agent_code/context.py` + +```python +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +def estimate_tokens(value: object) -> int: + return max(1, len(str(value)) // 4) + + +@dataclass +class ContextLayer: + name: str + tokens: int + items: int + + +@dataclass +class ContextReport: + pinned: ContextLayer + working: ContextLayer + compressed: ContextLayer + + @property + def total_tokens(self) -> int: + return self.pinned.tokens + self.working.tokens + self.compressed.tokens + + def render(self, cost_text: str = "") -> str: + lines = [ + f"Pinned: {self.pinned.tokens} tokens / {self.pinned.items} items", + f"Working: {self.working.tokens} tokens / {self.working.items} items", + f"Compressed: {self.compressed.tokens} tokens / {self.compressed.items} items", + f"Total: {self.total_tokens} tokens", + ] + if cost_text: + lines.extend(["", cost_text]) + return "\n".join(lines) + + +def analyze_context(system_prompt: str, messages: list[dict[str, Any]], todo_count: int = 0) -> ContextReport: + pinned_tokens = estimate_tokens(system_prompt) + estimate_tokens(todo_count) + working_messages = messages[-12:] + older_messages = messages[:-12] + + compressed_items = [ + msg for msg in older_messages + if "" in str(msg.get("content", "")) or "[Previous:" in str(msg.get("content", "")) + ] + compressed_tokens = sum(estimate_tokens(msg) for msg in compressed_items) + working_tokens = sum(estimate_tokens(msg) for msg in working_messages) + + return ContextReport( + pinned=ContextLayer("Pinned", pinned_tokens, 1 + int(todo_count > 0)), + working=ContextLayer("Working", working_tokens, len(working_messages)), + compressed=ContextLayer("Compressed", compressed_tokens, len(compressed_items)), + ) +``` + +这个分类是教学版的。它的价值不是 tokenizer 精确，而是让你知道：系统规则、项目记忆、当前 todo 属于 pinned；最近消息属于 working；摘要和旧工具占位属于 compressed。 + +### 1.2 `/context` 增强 + +`/context` 需要能看到当前 session messages。最简单做法是在 `RuntimeState` 里挂一份最近 messages： + +```python + last_messages: list[dict] = field(default_factory=list) + last_system_prompt: str = "" +``` + +`run_agent()` 每次进入 loop 前保存： + +```python +state.last_system_prompt = system_prompt or "" +``` + +每次 messages 改动后更新： + +```python +state.last_messages = messages +``` + +然后改 `slash.py` 的 `_cmd_context`： + +```python +def _cmd_context(_args: list[str], ctx: SlashContext) -> SlashResult: + if ctx.state is None: + session = ctx.session_id or "(none)" + return SlashResult(handled=True, message=f"cwd: {ctx.cwd}\nsession: {session}") + + from .context import analyze_context + + report = analyze_context( + ctx.state.last_system_prompt, + ctx.state.last_messages, + todo_count=len(ctx.state.todo_store), + ) + cost_line = "" + if ctx.state.cost_tracker is not None: + cost_line = f"Cost: ${ctx.state.cost_tracker.total.usd:.4f}" + return SlashResult(handled=True, message=report.render(cost_line)) +``` + +跑一下： + +```bash +> /context +Pinned: 820 tokens / 1 items +Working: 2100 tokens / 8 items +Compressed: 0 tokens / 0 items +Total: 2920 tokens + +Cost: $0.0041 +``` + +到这里，上下文第一次变成了可观察对象。 + +## v2：micro compact 只压旧工具结果 + +长会话里最肥的通常不是用户问题，而是工具 observation：`grep` 一扫几十行、`bash` 一跑几千字、`web_fetch` 一抓整篇文档。 + +micro compact 的原则是：不删消息、不破坏工具配对，只把较老的工具结果正文替换成占位。 + +### 2.1 在 `compactor.py` 里写 micro + +新建 `agent_code/compactor.py`： + +```python +from __future__ import annotations + +from copy import deepcopy +from typing import Any + + +def _tool_name_by_id(messages: list[dict[str, Any]]) -> dict[str, str]: + names: dict[str, str] = {} + for msg in messages: + if msg.get("role") != "assistant": + continue + content = msg.get("content", []) + if not isinstance(content, list): + continue + for block in content: + if isinstance(block, dict) and block.get("type") == "tool_use": + names[str(block.get("id"))] = str(block.get("name")) + return names + + +def micro_compact(messages: list[dict[str, Any]], keep_recent: int = 12) -> list[dict[str, Any]]: + if len(messages) <= keep_recent: + return messages + + result = deepcopy(messages) + tool_names = _tool_name_by_id(result) + cutoff = max(0, len(result) - keep_recent) + + for msg in result[:cutoff]: + content = msg.get("content") + if not isinstance(content, list): + continue + for block in content: + if not isinstance(block, dict) or block.get("type") != "tool_result": + continue + tool_use_id = str(block.get("tool_use_id", "")) + tool_name = tool_names.get(tool_use_id, "tool") + block["content"] = f"[Previous: used {tool_name}]" + return result +``` + +注意这里没有删除任何 `tool_result` block，也没有改 `tool_use_id`。Anthropic Messages API 需要 `tool_use` 和 `tool_result` 配对，micro compact 不能破坏这条协议。 + +### 2.2 `run_agent()` 里先 micro + +在 `agent.py` 的 loop 顶部，把 Day 6 的： + +```python +if len(messages) > 40: + messages = compact(messages, keep=8) +``` + +先替换成： + +```python +from .compactor import micro_compact + +messages = micro_compact(messages, keep_recent=12) +``` + +这个版本只做轻压，不做 LLM 摘要。 + +跑一个本地验证： + +```bash +$ uv run python - <<'PY' +from agent_code.compactor import micro_compact + +messages = [ + {"role":"assistant","content":[{"type":"tool_use","id":"u1","name":"grep","input":{}}]}, + {"role":"user","content":[{"type":"tool_result","tool_use_id":"u1","content":"x" * 5000}]}, +] + [{"role":"user","content":"keep"} for _ in range(12)] + +out = micro_compact(messages, keep_recent=12) +print(out[1]["content"][0]["content"]) +print(out[1]["content"][0]["tool_use_id"]) +PY +[Previous: used grep] +u1 +``` + +## v3：auto compact 要先备份 + +micro compact 只适合旧工具结果。会话继续变长时，还是需要把旧对话压成摘要。这个动作风险更大，所以先备份。 + +### 3.1 transcript 备份 + +新建 `agent_code/transcript.py`： + +```python +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + + +def backup_transcript(cwd: Path, messages: list[dict[str, Any]]) -> Path: + directory = cwd / ".agent" / "transcripts" + directory.mkdir(parents=True, exist_ok=True) + stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + path = directory / f"transcript_{stamp}.jsonl" + with path.open("w", encoding="utf-8") as f: + for msg in messages: + f.write(json.dumps(msg, ensure_ascii=False, separators=(",", ":")) + "\n") + return path +``` + +### 3.2 token / USD 阈值 + +新建 `agent_code/token_budget.py`： + +```python +from __future__ import annotations + + +class TokenBudget: + def __init__(self, max_tokens: int = 60_000, max_usd: float = 0.25) -> None: + self.max_tokens = max_tokens + self.max_usd = max_usd + + def should_compact(self, tokens: int, usd: float) -> bool: + return tokens >= self.max_tokens or usd >= self.max_usd +``` + +教学版阈值可以偏低，方便你制造 demo。真实项目可以把它放到配置文件。 + +### 3.3 `auto_compact()` + +在 `compactor.py` 里追加： + +```python +from pathlib import Path + +from .context import analyze_context +from .model import ModelProvider +from .transcript import backup_transcript + + +def _summary_prompt(messages: list[dict[str, Any]]) -> str: + return ( + "Summarize this conversation for continuing a coding-agent session. " + "Keep user goals, decisions, file changes, open questions, and tool outcomes. " + "Do not include irrelevant raw logs.\n\n" + f"{messages}" + ) + + +def auto_compact( + cwd: Path, + messages: list[dict[str, Any]], + provider: ModelProvider, + system_prompt: str, + threshold_tokens: int, +) -> tuple[list[dict[str, Any]], str | None]: + report = analyze_context(system_prompt, messages) + if report.total_tokens < threshold_tokens: + return messages, None + + backup_path = backup_transcript(cwd, messages) + old = messages[:-8] + recent = messages[-8:] + response = provider.complete( + [{"role": "user", "content": _summary_prompt(old)}], + tools=[], + system="You compact coding-agent transcripts into short continuation summaries.", + ) + summary = response.text or "(compact summary unavailable)" + compacted = [ + { + "role": "user", + "content": f"\n{summary}\n", + } + ] + return compacted + recent, str(backup_path) +``` + +然后在 `run_agent()` 里，provider call 前加： + +```python +from .compactor import auto_compact, micro_compact + +messages = micro_compact(messages, keep_recent=12) +messages, backup_path = auto_compact( + resolved_cwd, + messages, + provider, + system_prompt or "", + threshold_tokens=60_000, +) +if backup_path: + console.print(f"[dim]compacted: backup={backup_path}[/dim]") +``` + +跑验证时可以把阈值临时改成 500，造一个长 messages，确认 `.agent/transcripts/transcript_*.jsonl` 出现。 + +## v4：`compact()` 工具和 `/compact` + +auto compact 是 harness 自己触发。手动 compact 有两个入口： + +- 用户：`/compact --dry-run` / `/compact --apply` +- 模型：`compact()` + +### 4.1 `manual_compact_plan()` + +在 `compactor.py` 里追加： + +```python +def manual_compact_plan(messages: list[dict[str, Any]], system_prompt: str) -> str: + report = analyze_context(system_prompt, messages) + return "\n".join([ + "compact dry-run:", + f"- pinned stays: {report.pinned.tokens} tokens", + "- working keeps the latest 8-12 messages", + "- older tool_result content may be replaced by [Previous: used ]", + "- old conversation will be summarized into ", + "- a transcript backup will be written before apply", + ]) +``` + +### 4.2 `compact()` 工具 + +打开 `tools.py`，加一个函数： + +```python +def compact_tool(args: dict[str, Any], ctx: ToolContext) -> str: + """模型主动请求 compact。真正压缩在 agent.py turn boundary 里处理。""" + state = ctx.runtime_state + if state is None: + return "error: no runtime state" + state.compact_requested = True + return "compact requested; the harness will compact at the next safe boundary" +``` + +`RuntimeState` 里新增： + +```python + compact_requested: bool = False +``` + +注册工具： + +```python +registry.register( + Tool( + name="compact", + description="Request the harness to compact old context at the next safe boundary.", + run=compact_tool, + parameters={"type": "object", "properties": {}, "required": []}, + is_read_only=False, + ) +) +``` + +在 `run_agent()` loop 顶部检查： + +```python +if state.compact_requested: + state.compact_requested = False + messages, backup_path = auto_compact(resolved_cwd, messages, provider, system_prompt or "", threshold_tokens=1) + if backup_path: + console.print(f"[dim]manual compact: backup={backup_path}[/dim]") +``` + +### 4.3 `/compact` + +打开 `slash.py`，把 `_cmd_compact` 改成： + +```python +def _cmd_compact(args: list[str], ctx: SlashContext) -> SlashResult: + if ctx.state is None: + return SlashResult(handled=True, message="compact 需要交互 shell") + + from .compactor import manual_compact_plan + + if not args or args[0] == "--dry-run": + return SlashResult( + handled=True, + message=manual_compact_plan(ctx.state.last_messages, ctx.state.last_system_prompt), + ) + + if args[0] == "--apply": + ctx.state.compact_requested = True + return SlashResult(handled=True, message="compact scheduled for the next safe boundary") + + return SlashResult(handled=True, message="用法: /compact --dry-run | /compact --apply") +``` + +跑验证： + +```bash +> /compact --dry-run +compact dry-run: +- pinned stays: ... +- working keeps ... +... +> /compact --apply +compact scheduled for the next safe boundary +``` + +## 收尾 a：工具结果预算 + 溢出落盘 + +一轮 `bash` 或 `web_fetch` 可能返回几万字。上下文里不应该塞全文，但全文也不能丢。 + +### a.1 新增 `agent_code/tool_results.py` + +```python +from __future__ import annotations + +from pathlib import Path + + +MAX_TOOL_RESULT_CHARS = 8_000 +PREVIEW_CHARS = 1_000 + + +def persist_if_large(cwd: Path, tool_call_id: str, content: str) -> str: + if len(content) <= MAX_TOOL_RESULT_CHARS: + return content + + directory = cwd / ".agent" / "tool-results" + directory.mkdir(parents=True, exist_ok=True) + path = directory / f"{tool_call_id}.txt" + path.write_text(content, encoding="utf-8") + head = content[:PREVIEW_CHARS] + tail = content[-PREVIEW_CHARS:] + return ( + f"[large tool result stored]\n" + f"Full output: {path}\n\n" + f"--- head ---\n{head}\n\n" + f"--- tail ---\n{tail}" + ) +``` + +### a.2 接进 `execute_one_tool_call()` + +在 `agent.py` 里，`result = tools.run(call, ctx)` 后面加： + +```python +from .tool_results import persist_if_large + +result.content = persist_if_large(ctx.cwd, result.tool_call_id, result.content) +``` + +这样模型仍然拿到合法 `tool_result`，但上下文只有预览和路径。如果它需要全文，可以再 `read_file` 那个路径。 + +跑验证： + +```bash +$ uv run python - <<'PY' +from pathlib import Path +from agent_code.tool_results import persist_if_large + +text = "x" * 9000 +out = persist_if_large(Path.cwd(), "call_big", text) +print("Full output:" in out) +print((Path.cwd() / ".agent" / "tool-results" / "call_big.txt").exists()) +PY +True +True +``` + +`/compact --apply` 后可以再做孤儿文件清理：扫描 session JSONL 里还引用哪些 `tool_call_id`，删除 `.agent/tool-results/` 下没被引用的文件。第一版可以先保留文件，课后挑战再做精确清理。 + +## 收尾 b：最小 recovery + +最后补一个很小的恢复层。它不做生产级 API SDK，只处理三类常见问题： + +- 429/529：退避后重试。 +- prompt too long：触发 compact 后重试一次。 +- max_tokens 不够：提示模型续写或提高输出上限。 + +新建 `agent_code/recovery.py`： + +```python +from __future__ import annotations + +import time +from typing import Callable, TypeVar + +T = TypeVar("T") + + +def is_capacity_error(exc: Exception) -> bool: + text = str(exc).lower() + return "429" in text or "529" in text or "overloaded" in text or "rate limit" in text + + +def is_prompt_too_long(exc: Exception) -> bool: + text = str(exc).lower() + return "prompt" in text and ("too long" in text or "context" in text) + + +def with_recovery(call: Callable[[], T], on_prompt_too_long: Callable[[], None] | None = None) -> T: + for attempt in range(3): + try: + return call() + except Exception as exc: + if is_capacity_error(exc) and attempt < 2: + time.sleep(0.5 * (2 ** attempt)) + continue + if is_prompt_too_long(exc) and on_prompt_too_long is not None: + on_prompt_too_long() + on_prompt_too_long = None + continue + raise + return call() +``` + +在 `agent.py` 里把 provider call 包起来： + +```python +from .recovery import with_recovery + +response = with_recovery( + lambda: provider.complete(messages, tools=visible_tools.list(), system=system_prompt), + on_prompt_too_long=lambda: setattr(state, "compact_requested", True), +) +``` + +这只是最小闭环。OAuth、provider fallback、prompt cache break、无人值守长 retry 都不进主线。 + +## 收尾：今天改了哪些文件 + +今天新增八个文件： + +```txt +agent_code/cost_prices.py +agent_code/cost.py +agent_code/context.py +agent_code/transcript.py +agent_code/token_budget.py +agent_code/compactor.py +agent_code/tool_results.py +agent_code/recovery.py +``` + +今天改了六个已有文件： + +```txt +agent_code/model.py ModelResponse 增 usage +agent_code/runtime.py 挂 cost/context/compact 状态 +agent_code/agent.py 记录 usage、micro/auto compact、overflow、recovery +agent_code/tools.py 新增 compact 工具 +agent_code/slash.py /cost、/context、/compact +agent_code/cli.py 初始化 CostTracker +``` + +## 手动 trace 一遍 + +### 路径一：一次带工具调用的成本归因 + +```txt +1. 用户问“今天几号”。 +2. 第一轮模型看到 system_date，发 tool_use。 +3. harness 执行 system_date。 +4. 第二轮模型读取 tool_result，返回 final。 +5. CostTracker 把第二轮 usage 平摊给上一轮工具 system_date。 +6. /cost 显示 by_tool: system_date。 +``` + +### 路径二：micro compact + +```txt +1. 老消息里有 tool_result 大文本。 +2. micro_compact 找到对应 tool_use_id。 +3. 用 tool_use_id 映射回工具名。 +4. 把 content 替换成 [Previous: used grep]。 +5. tool_result block 和 tool_use_id 仍然保留。 +``` + +### 路径三：大工具结果落盘 + +```txt +1. bash 返回 9000 字。 +2. persist_if_large 写 .agent/tool-results/.txt。 +3. tool_result.content 变成 head/tail preview + 文件路径。 +4. 模型需要全文时，再调用 read_file 读取该路径。 +``` + +## 今天有了什么 + +- **成本可见**：`/cost` 能看到 total、by model、by tool 的粗估。 +- **三层上下文**：`/context` 把上下文拆成 Pinned / Working / Compressed。 +- **micro compact**：旧工具结果变占位，不破坏 tool_use/tool_result 配对。 +- **auto/manual compact**：阈值触发或用户触发，压缩前写 transcript 备份。 +- **工具结果预算**：大结果落盘，上下文里只留预览。 +- **recovery**：服务忙和 prompt 太长有最小补救路径。 + +## 常见问题 + +### `/cost` 的 by-tool 准吗？ + +它是趋势工具，不是账单。一次模型调用的成本按上一轮工具平摊，所以只能说明“哪些工具结果大概率让上下文变贵”。 + +### 为什么 micro compact 不删消息？ + +因为真实工具协议要求 `tool_use` 和 `tool_result` 配对。删掉旧 `tool_result` 可能让下一轮请求直接报错。 + +### compact 后是不是历史没了？ + +主上下文里只剩摘要，但 compact 前已经备份到： + +```txt +.agent/transcripts/transcript_.jsonl +``` + +### 大工具结果为什么不直接截断？ + +直接截断会丢信息。落盘后，上下文短了，全文还可以通过 `read_file` 找回来。 + +## 课后挑战 + +1. 把价格表移到 `.agent/settings.json`。 +2. 给 `/context` 加 inline/deferred 工具数量，为 Day 14 铺路。 +3. `/compact --apply` 后清理未被 session 引用的 `.agent/tool-results/*.txt`。 +4. 给 `web_fetch` 和 `bash` 分别设置不同的结果上限。 +5. 用真实 tokenizer 替换 `chars/4` 估算。 + +## 思考题 + +1. **为什么 by-tool cost 要归到“上一轮工具”，而不是当前这一轮？** 提示：工具结果什么时候进入模型上下文？ +2. **Pinned / Working / Compressed 三层里，哪一层最危险，不能随便压？** +3. **micro compact 为什么必须保留 `tool_use_id`？** +4. **工具结果落盘后，模型还能如何拿到全文？这对上下文预算有什么好处？** + +## 下一天 + +今天 harness 开始管理上下文和成本。下一天我们让多个有身份的 Agent 协作：建 team、发消息、调度 teammate，再把后台任务统一成可查、可停、可回流的 task runtime。 diff --git a/docs/day-12-agent-coordinator.md b/docs/day-12-agent-coordinator.md new file mode 100644 index 0000000..0558670 --- /dev/null +++ b/docs/day-12-agent-coordinator.md @@ -0,0 +1,801 @@ +# Day 12：Agent Coordinator + +Day 10 的 `agent()` 是一次性委派：主 Agent 把一个任务交给子 Agent，等它跑完，再拿 summary。 + +今天往前走一步：如果一个任务需要多个角色协作呢？比如 lead 负责拆解，reviewer 看风险，tester 跑验证，docs-writer 写说明。它们不应该都挤在一个 prompt 里，也不应该互相抢当前 turn。 + +今天做 Agent Coordinator：用 team、inbox、coordinator loop 和 task runtime，把“多个有身份的 Agent”变成 harness 里可观察、可调度、可停止的系统。 + +跑完之后你会看到： + +- `team_create` 能创建一个内存团队，`/team` 能列出成员。 +- `send_message(recipient, content)` 只写 inbox，不打断对方当前 turn。 +- coordinator 用 round-robin 把 pending message 派给 teammate，再把 summary 回给 lead。 +- Day 5 的后台 bash 会登记成 `TaskState`，可用 `task_output` 查、`task_stop` 停。 +- worker 完成后用 `` 回流，不在工具执行一半时改 messages。 + +代码约 620 行，新增约 380 行。 + +今天分四版： + +1. v1：`team_create` / `team_delete` / `/team`。 +2. v2：`send_message` 和 inbox。 +3. v3：coordinator 调度循环。 +4. v4：`task_output` / `task_stop`，把后台 bash 接进统一 task runtime。 + +## 起手：今天的起点 + +Day 12 复用前面的能力： + +```txt +Day 5 background bash: .bg/.out / .err / pid +Day 8 RuntimeState: 共享运行态、worker 线程、type-ahead +Day 10 subagent runner: fresh task + template prompt + summary +Day 11 context/cost: 不扩展，只尊重安全 turn 边界 +``` + +今天新增四个模块： + +```txt +agent_code/inbox.py +agent_code/teammate.py +agent_code/coordinator.py +agent_code/tasks.py +``` + +先把边界画出来： + +```mermaid +flowchart TD + A["lead / main agent"] --> B["send_message"] + B --> C["Inbox"] + C --> D["Coordinator drain"] + D --> E["teammate runner"] + E --> F["subagent summary"] + F --> G["lead inbox"] + H["background bash"] --> I["TaskRegistry"] + I --> J["task_output / task_stop"] + I --> K["task notification"] +``` + +今天最重要的原则：**消息和任务完成通知都只在安全边界注入**。不要在某个工具执行到一半时突然往 messages 里插东西。 + +## v1：先让 team 可见 + +团队先放内存里，不持久化磁盘。我们只要能创建、删除、查看，就足够进入调度问题。 + +### 1.1 新增 `agent_code/teammate.py` + +```python +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class Teammate: + name: str + agent_template: str + role: str + + +@dataclass +class Team: + name: str + members: list[Teammate] + + def render(self) -> str: + lines = [f"team: {self.name}"] + for member in self.members: + lines.append(f" - {member.name}: {member.agent_template} ({member.role})") + return "\n".join(lines) +``` + +`agent_template` 指向 Day 10 的 `.agent/agents/.md`，`role` 是这个 teammate 在团队里的职责。 + +### 1.2 `RuntimeState` 加 team registry + +打开 `agent_code/runtime.py`： + +```python + teams: dict[str, "Team"] = field(default_factory=dict) + active_team: str | None = None +``` + +同样为了避免循环 import，顶部用 `TYPE_CHECKING`： + +```python +if TYPE_CHECKING: + from .teammate import Team +``` + +### 1.3 新增 team 工具 + +打开 `agent_code/tools.py`，加三个函数： + +```python +def team_create(args: dict[str, Any], ctx: ToolContext) -> str: + from .teammate import Team, Teammate + + state = ctx.runtime_state + if state is None: + return "error: no runtime state" + name = str(args.get("name", "")).strip() + raw_members = args.get("members", []) + if not name: + return "error: missing required argument 'name'" + if not isinstance(raw_members, list) or not raw_members: + return "error: members must be a non-empty list" + + members: list[Teammate] = [] + for item in raw_members: + members.append( + Teammate( + name=str(item.get("name", "")).strip(), + agent_template=str(item.get("agent_template", "")).strip(), + role=str(item.get("role", "")).strip(), + ) + ) + if any(not m.name or not m.agent_template for m in members): + return "error: each member needs name and agent_template" + + state.teams[name] = Team(name=name, members=members) + state.active_team = name + return state.teams[name].render() + + +def team_delete(args: dict[str, Any], ctx: ToolContext) -> str: + state = ctx.runtime_state + if state is None: + return "error: no runtime state" + name = str(args.get("name", "")).strip() + if not name: + return "error: missing required argument 'name'" + state.teams.pop(name, None) + if state.active_team == name: + state.active_team = None + return f"team deleted: {name}" + + +def team_read(args: dict[str, Any], ctx: ToolContext) -> str: + state = ctx.runtime_state + if state is None or not state.teams: + return "(no team)" + name = str(args.get("name") or state.active_team or "") + team = state.teams.get(name) + if team is None: + return f"team not found: {name}" + return team.render() +``` + +注册： + +```python +registry.register(Tool( + name="team_create", + description="Create an in-memory team. Members point to local subagent templates.", + run=team_create, + parameters={ + "type": "object", + "properties": { + "name": {"type": "string"}, + "members": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "agent_template": {"type": "string"}, + "role": {"type": "string"}, + }, + "required": ["name", "agent_template", "role"], + }, + }, + }, + "required": ["name", "members"], + }, +)) +registry.register(Tool( + name="team_delete", + description="Delete an in-memory team.", + run=team_delete, + parameters={ + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }, +)) +registry.register(Tool( + name="team_read", + description="Read the active team.", + run=team_read, + parameters={"type": "object", "properties": {}, "required": []}, + is_read_only=True, +)) +``` + +把 `team_read` 加进 `_READONLY_TOOLS`，把 `team_create` / `team_delete` 加进 `_LOW_RISK_WRITES`。 + +### 1.4 `/team` + +打开 `slash.py`： + +```python +def _cmd_team(_args: list[str], ctx: SlashContext) -> SlashResult: + if ctx.state is None or not ctx.state.teams: + return SlashResult(handled=True, message="(no team)") + if ctx.state.active_team and ctx.state.active_team in ctx.state.teams: + return SlashResult(handled=True, message=ctx.state.teams[ctx.state.active_team].render()) + return SlashResult(handled=True, message="\n\n".join(team.render() for team in ctx.state.teams.values())) +``` + +注册： + +```python +register("team", "查看当前团队", _cmd_team) +``` + +跑验证： + +```bash +$ uv run agent-code +> 建一个团队 dev-team，成员 reviewer 使用 code-reviewer，tester 使用 debugger +tool_call: team_create {...} +final: ... +> /team +team: dev-team + - reviewer: code-reviewer (review code changes) + - tester: debugger (run failing checks) +``` + +模型给的 member 文案可能不同。关键是 `/team` 能看到内存 team。 + +## v2：`send_message` 只写 inbox + +现在有 teammate 名字了，但还不能通信。先做 inbox。 + +`send_message` 不能同步跑 recipient。原因很简单：recipient 可能正在自己的 Agent Loop 里；中途把消息塞进去，会破坏 turn 边界。 + +### 2.1 新增 `agent_code/inbox.py` + +```python +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone + + +@dataclass +class InboxMessage: + sender: str + recipient: str + content: str + timestamp: str + + +@dataclass +class Inbox: + messages: dict[str, list[InboxMessage]] = field(default_factory=dict) + + def send(self, sender: str, recipient: str, content: str) -> None: + msg = InboxMessage( + sender=sender, + recipient=recipient, + content=content, + timestamp=datetime.now(timezone.utc).isoformat(), + ) + self.messages.setdefault(recipient, []).append(msg) + + def drain(self, recipient: str) -> list[InboxMessage]: + pending = self.messages.get(recipient, []) + self.messages[recipient] = [] + return pending + + def has_pending(self) -> bool: + return any(bool(items) for items in self.messages.values()) + + +def format_messages(messages: list[InboxMessage]) -> str: + if not messages: + return "" + lines = [""] + for msg in messages: + lines.append(f'') + lines.append(msg.content) + lines.append("") + lines.append("") + return "\n".join(lines) +``` + +### 2.2 `RuntimeState` 挂 inbox + +```python + inbox: "Inbox | None" = None + current_role: str = "lead" +``` + +创建 `RuntimeState` 后初始化： + +```python +from .inbox import Inbox + +state.inbox = Inbox() +``` + +### 2.3 `send_message` 工具 + +打开 `tools.py`： + +```python +def send_message(args: dict[str, Any], ctx: ToolContext) -> str: + state = ctx.runtime_state + if state is None or state.inbox is None: + return "error: no inbox" + recipient = str(args.get("recipient", "")).strip() + content = str(args.get("content", "")).strip() + if not recipient: + return "error: missing required argument 'recipient'" + if not content: + return "error: missing required argument 'content'" + sender = getattr(state, "current_role", "lead") + state.inbox.send(sender, recipient, content) + return f"message queued for {recipient}" +``` + +注册： + +```python +registry.register(Tool( + name="send_message", + description="Send an asynchronous message to a teammate inbox. Does not run the recipient immediately.", + run=send_message, + parameters={ + "type": "object", + "properties": { + "recipient": {"type": "string"}, + "content": {"type": "string"}, + }, + "required": ["recipient", "content"], + }, +)) +``` + +把 `send_message` 加进 `_LOW_RISK_WRITES`。 + +### 2.4 drain 时机 + +`send_message` 只写 inbox。谁来读？在 teammate 下一次 model call 前 drain。 + +Day 12 主 Agent 自己也可以 drain lead inbox。打开 `agent.py`，provider call 前加： + +```python +from .inbox import format_messages + +if state.inbox is not None: + pending = state.inbox.drain(getattr(state, "current_role", "lead")) + if pending: + messages.append({"role": "user", "content": format_messages(pending)}) +``` + +这一步要放在 provider.complete 前的安全边界，不要放在工具函数内部。 + +跑验证： + +```bash +> 给 reviewer 发消息，让它稍后 review 当前 diff +tool_call: send_message {'recipient': 'reviewer', 'content': 'Review the current diff.'} +final: ... +``` + +这时 reviewer 不会立刻跑。消息只是进了 inbox。 + +## v3：Coordinator 调度循环 + +现在可以建 team、发消息。下一步让 lead 负责调度。 + +教学版先做 round-robin，不做并行 swarm。我们要先看清三个东西：队列、调度权、停止条件。 + +### 3.1 新增 `agent_code/coordinator.py` + +```python +from __future__ import annotations + +from dataclasses import dataclass + +from .inbox import format_messages +from .runtime import RuntimeState +from .subagent_runner import run_subagent +from .tools import ToolContext + + +@dataclass +class Coordinator: + state: RuntimeState + ctx: ToolContext + max_rounds: int = 3 + + def run(self, team_name: str) -> str: + team = self.state.teams.get(team_name) + if team is None: + return f"team not found: {team_name}" + if self.state.inbox is None: + return "error: no inbox" + + summaries: list[str] = [] + for round_index in range(self.max_rounds): + progressed = False + for teammate in team.members: + pending = self.state.inbox.drain(teammate.name) + if not pending: + continue + progressed = True + task = ( + f"You are {teammate.name}, role: {teammate.role}.\n\n" + f"{format_messages(pending)}\n\n" + "Respond with a concise summary for the lead." + ) + old_role = self.state.current_role + self.state.current_role = teammate.name + try: + summary = run_subagent(teammate.agent_template, task, self.ctx, max_steps=6) + finally: + self.state.current_role = old_role + self.state.inbox.send(teammate.name, "lead", summary) + summaries.append(f"{teammate.name}: {summary}") + if not progressed: + break + return "\n".join(summaries) or "(no pending teammate messages)" +``` + +teammate 复用 Day 10 的 subagent runner：fresh task、template system prompt、summary 回 lead。 + +### 3.2 `team_run` 工具 + +路线图只列了 `team_create/team_delete/send_message`，但 v3 需要一个显式入口让模型触发调度。我们把它叫 `team_run`，也可以放在 `/team run`，这里用工具更贴近 Agent Loop。 + +```python +def team_run(args: dict[str, Any], ctx: ToolContext) -> str: + from .coordinator import Coordinator + + state = ctx.runtime_state + if state is None: + return "error: no runtime state" + name = str(args.get("name") or state.active_team or "").strip() + max_rounds = max(1, min(int(args.get("max_rounds", 3)), 8)) + return Coordinator(state, ctx, max_rounds=max_rounds).run(name) +``` + +注册： + +```python +registry.register(Tool( + name="team_run", + description="Run the active team coordinator loop over pending inbox messages.", + run=team_run, + parameters={ + "type": "object", + "properties": { + "name": {"type": "string"}, + "max_rounds": {"type": "integer", "default": 3}, + }, + "required": [], + }, +)) +``` + +把 `team_run` 加到 `_LOW_RISK_WRITES`。它会启动 subagent，子任务里的写/命令仍走权限。 + +### 3.3 跑验证 + +```bash +> 创建 dev-team，成员 reviewer(code-reviewer) 和 tester(debugger) +... +> send_message 给 reviewer：看当前 git diff 的风险；给 tester：说明应该跑什么验证 +tool_call: send_message {'recipient': 'reviewer', ...} +tool_call: send_message {'recipient': 'tester', ...} +final: ... +> 调用 team_run 运行 dev-team +tool_call: team_run {'name': 'dev-team', 'max_rounds': 2} +final: reviewer: ... +tester: ... +``` + +如果模型不主动调 `team_run`，用明确 prompt： + +```bash +uv run agent-code "必须调用 team_run，name=dev-team，max_rounds=2" +``` + +## v4：后台任务统一成 TaskRuntime + +Day 5 的后台 bash 已经能跑，但它只返回 `.bg/.out` 和 pid。今天把它接成 `TaskState`，让模型用 `task_output` 和 `task_stop` 统一管理。 + +### 4.1 新增 `agent_code/tasks.py` + +```python +from __future__ import annotations + +import os +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path + + +@dataclass +class TaskState: + id: str + type: str + status: str + output_path: Path + stop_handle: int | None = None + started_at: float = field(default_factory=time.time) + ended_at: float | None = None + + +@dataclass +class TaskRegistry: + tasks: dict[str, TaskState] = field(default_factory=dict) + notifications: list[str] = field(default_factory=list) + + def register_bash(self, output_path: Path, pid: int) -> TaskState: + task = TaskState( + id="task-" + uuid.uuid4().hex[:8], + type="bash", + status="running", + output_path=output_path, + stop_handle=pid, + ) + self.tasks[task.id] = task + return task + + def finish(self, task_id: str, status: str) -> None: + task = self.tasks.get(task_id) + if task is None: + return + task.status = status + task.ended_at = time.time() + self.notifications.append( + f"\ntask_id: {task.id}\nstatus: {task.status}\noutput_path: {task.output_path}\n" + ) + + def output(self, task_id: str, block: bool = False, timeout: float | None = None) -> str: + task = self.tasks.get(task_id) + if task is None: + return f"error: task not found: {task_id}" + deadline = time.time() + (timeout or 0) + while block and task.status == "running" and timeout is not None and time.time() < deadline: + time.sleep(0.2) + text = task.output_path.read_text(encoding="utf-8", errors="replace") if task.output_path.exists() else "" + preview = text[-4000:] if len(text) > 4000 else text + return f"task: {task.id}\nstatus: {task.status}\noutput_path: {task.output_path}\n\n{preview}" + + def stop(self, task_id: str) -> str: + task = self.tasks.get(task_id) + if task is None: + return f"error: task not found: {task_id}" + if task.type == "bash" and task.stop_handle: + try: + os.kill(task.stop_handle, 15) + except OSError as exc: + return f"error: {exc}" + task.status = "stopped" + task.ended_at = time.time() + return f"stopped: {task.id}" +``` + +### 4.2 RuntimeState 挂 task registry + +```python + tasks: "TaskRegistry | None" = None +``` + +CLI 初始化： + +```python +from .tasks import TaskRegistry + +state.tasks = TaskRegistry() +``` + +### 4.3 background bash 登记 task + +打开 `tools.py` 的 `bash()`。`background=True` 分支现在返回 `.bg` 信息。改成： + +```python +if background: + from .bg_manager import start_background + + result = start_background(command, ctx.cwd) + state = ctx.runtime_state + if state is not None and state.tasks is not None: + task = state.tasks.register_bash(Path(result["output_file"]), int(result["pid"])) + return ( + f"task_id: {task.id}\n" + f"status: {task.status}\n" + f"output_path: {task.output_path}\n" + f"pid: {task.stop_handle}" + ) + return ... +``` + +最小版还不能自动知道进程什么时候结束。可以在 `bg_manager` 的 wait 线程里回调 `state.tasks.finish(task.id, "done")`，或者 v4 先让 `task_output` 读取文件并显示 `running`。如果要完成回流，就把 wait callback 接上。 + +### 4.4 `task_output` / `task_stop` + +`tools.py`： + +```python +def task_output(args: dict[str, Any], ctx: ToolContext) -> str: + state = ctx.runtime_state + if state is None or state.tasks is None: + return "error: no task registry" + task_id = str(args.get("task_id", "")).strip() + block = bool(args.get("block", False)) + timeout = args.get("timeout") + return state.tasks.output(task_id, block=block, timeout=float(timeout) if timeout else None) + + +def task_stop(args: dict[str, Any], ctx: ToolContext) -> str: + state = ctx.runtime_state + if state is None or state.tasks is None: + return "error: no task registry" + task_id = str(args.get("task_id", "")).strip() + return state.tasks.stop(task_id) +``` + +注册： + +```python +registry.register(Tool( + name="task_output", + description="Read output from a background task.", + run=task_output, + parameters={ + "type": "object", + "properties": { + "task_id": {"type": "string"}, + "block": {"type": "boolean", "default": False}, + "timeout": {"type": "number"}, + }, + "required": ["task_id"], + }, + is_read_only=True, +)) +registry.register(Tool( + name="task_stop", + description="Stop a running background task.", + run=task_stop, + parameters={ + "type": "object", + "properties": {"task_id": {"type": "string"}}, + "required": ["task_id"], + }, +)) +``` + +权限：`task_output` 只读，`task_stop` 放 `_LOW_RISK_WRITES` 或默认 ask。教学版建议 `task_stop` 走 ask，避免误杀进程。 + +### 4.5 task notification 回流 + +在 `agent.py` provider call 前，加一个安全边界： + +```python +if state.tasks is not None and state.tasks.notifications: + notices = "\n\n".join(state.tasks.notifications) + state.tasks.notifications.clear() + messages.append({"role": "user", "content": notices}) +``` + +worker 线程结束时只写 notification 队列，不直接改当前 messages。下一轮安全边界再注入。 + +跑验证： + +```bash +> 用 bash 后台执行 python -c "import time; time.sleep(2); print('done')"，然后返回 task_id +tool_call: bash {'command': 'python -c "...', 'background': True} +final: task_id: task-... +> 查询这个 task 的输出 +tool_call: task_output {'task_id': 'task-...', 'block': True, 'timeout': 5} +final: ... done +``` + +## 收尾：今天改了哪些文件 + +今天新增四个文件： + +```txt +agent_code/inbox.py +agent_code/teammate.py +agent_code/coordinator.py +agent_code/tasks.py +``` + +今天改了六个已有文件： + +```txt +agent_code/runtime.py team / inbox / tasks 状态 +agent_code/tools.py team_* / send_message / team_run / task_* 工具 +agent_code/permissions.py 工具权限分类 +agent_code/slash.py /team +agent_code/agent.py inbox drain + task notification 安全注入 +agent_code/cli.py 初始化 Inbox / TaskRegistry +``` + +## 手动 trace 一遍 + +### 路径一：send_message + +```txt +1. 主模型调用 send_message(recipient="reviewer")。 +2. 工具只写 state.inbox，不启动 reviewer。 +3. 当前 turn 继续正常结束。 +4. coordinator 或 reviewer 下一轮启动前 drain inbox。 +5. pending 消息变成 注入 teammate user prompt。 +``` + +### 路径二：coordinator + +```txt +1. lead 创建 team。 +2. lead 给 reviewer/tester 发消息。 +3. team_run 读取 active team。 +4. round-robin 遍历成员，谁有 pending 就调用 Day 10 subagent runner。 +5. teammate summary 发回 lead inbox。 +6. 达到 max_rounds 或没有 pending 后停止。 +``` + +### 路径三：background bash task + +```txt +1. bash(background=True) 启动进程。 +2. TaskRegistry 生成 task_id，记录 pid/output_path。 +3. 模型后续用 task_output(task_id) 查输出。 +4. 需要停止时用 task_stop(task_id)。 +5. 完成通知排进 notifications，在下一轮安全边界回流。 +``` + +## 今天有了什么 + +- **Team registry**：团队和成员可见。 +- **Inbox**：消息异步投递，不打断当前 turn。 +- **Coordinator**：lead 用 round-robin 调度 teammate。 +- **Teammate runner**：复用 Day 10 subagent，不把完整 transcript 灌回主会话。 +- **Task runtime**：后台 bash 有统一 `task_id`，可查、可停、可回流。 + +## 常见问题 + +### 为什么 `send_message` 不直接运行 recipient？ + +因为 recipient 可能正在自己的回合里。同步插消息会破坏工具调用配对，也会让调度顺序不可解释。 + +### 为什么 coordinator 不并行？ + +教学版先用 round-robin 讲清楚队列、调度和停止条件。真正并行 worker 会引入锁、取消、输出回流和 UI 问题，适合后续扩展。 + +### `task_output` 和 TodoWrite 是一回事吗？ + +不是。TodoWrite 是模型规划任务列表；`task_output` / `task_stop` 是运行时后台进程 IO。一个管计划，一个管进程。 + +### task notification 为什么不立刻进 messages？ + +worker 线程不能在主 Agent 的 tool batch 中间改 messages。notification 要排队，在下一轮 provider call 前注入。 + +## 课后挑战 + +1. 给 `/team` 增加 `create/delete/run` 子命令。 +2. 把 inbox 持久化到 `.agent/inbox.jsonl`。 +3. 让 `team_run` 支持并行 teammate。 +4. 给 TaskRegistry 加 `/tasks` 列表。 +5. 把异步 subagent worker 也登记成 `TaskState(type="subagent")`。 + +## 思考题 + +1. **为什么 inbox drain 要放在下一轮 LLM call 前？** +2. **coordinator 的停止条件为什么必须有 `max_rounds`？** +3. **`send_message` 和 `agent()` 的区别是什么？一个是投递，一个是调用。** +4. **后台任务完成后，为什么要用 notification 队列回流？** + +## 下一天 + +今天我们让多个 Agent 能协作，也让后台任务有了统一 IO。下一天解决另一个实际问题：Agent 直接改主分支太危险。Day 13 会把当前工作切进 git worktree，在隔离目录里修 bug、跑测试，再合回主分支。 diff --git a/docs/day-13-worktree-demo.md b/docs/day-13-worktree-demo.md new file mode 100644 index 0000000..895e8ea --- /dev/null +++ b/docs/day-13-worktree-demo.md @@ -0,0 +1,704 @@ +# Day 13：Worktree 隔离 + 端到端 Demo 1.0 + +前 12 天，`agent-code` 已经能读代码、改文件、跑命令、做计划、用 skill、启动 subagent、协调 teammate。 + +但还有一个现实问题：Agent 直接改主分支很危险。它可以改错文件，可以跑到一半被中断，也可能在你已有未提交改动时制造一堆冲突。 + +今天做 Worktree 隔离：让 Agent 进入一个独立 git worktree，在那里修 bug、跑测试。确认没问题后，再 merge 回主目录。 + +跑完之后你会看到： + +- `enter_worktree(branch)` 会先检查主目录是否干净，再创建 `.worktrees/`。 +- harness 的 cwd 会真的切到 worktree，文件工具、bash、subagent 都跟着新 cwd 走。 +- `worktree-state` 会写进 session JSONL，`--resume` 后还能回到隔离目录。 +- `exit_worktree("merge")` 成功后合回主分支并清理 worktree；冲突时保留现场。 +- v3 会串起一个端到端 demo：plan → subagent → file_edit → pytest → merge。 + +代码约 520 行，新增约 300 行。 + +今天分三版： + +1. v1：`enter_worktree`，创建隔离目录并切 cwd。 +2. v2：`exit_worktree`，支持 `merge / discard / keep`。 +3. v3：端到端修复 `examples/buggy-python-project` 的 failing test。 + +## 起手：今天的起点 + +Day 12 的 `agent-code` 已经有 `RuntimeState`，但当前 cwd 仍然多半是 `cli.py` 里的局部变量： + +```txt +resolved_cwd = cwd.resolve() +run_agent(..., cwd=resolved_cwd) +``` + +这在 worktree 里会卡住：工具里就算创建了 `.worktrees/fix-bug`，下一轮 `run_agent()` 还是用旧 `resolved_cwd`。 + +所以 Day 13 的第一件事不是写 git 命令，而是把 cwd 提升到运行态： + +```txt +RuntimeState.cwd +RuntimeState.original_cwd +RuntimeState.worktree +``` + +只要 `ToolContext.cwd` 对了，`read_file`、`file_edit`、`bash`、subagent 都不用各自改路径。 + +## v1：进入 worktree，不只是返回一段文本 + +`enter_worktree` 要做四件事： + +1. 主目录必须干净。 +2. 分支名必须 sanitize。 +3. `git worktree add` 创建隔离目录。 +4. 更新 `RuntimeState.cwd`，让下一轮工具真的在 worktree 里跑。 + +### 1.1 新增 `agent_code/worktree.py` + +```python +from __future__ import annotations + +import re +import subprocess +from dataclasses import dataclass +from pathlib import Path + + +@dataclass +class WorktreeState: + original_cwd: Path + worktree_path: Path + branch: str + from_branch: str + + def to_dict(self) -> dict[str, str]: + return { + "original_cwd": str(self.original_cwd), + "worktree_path": str(self.worktree_path), + "branch": self.branch, + "from_branch": self.from_branch, + } + + @classmethod + def from_dict(cls, data: dict[str, str]) -> "WorktreeState": + return cls( + original_cwd=Path(data["original_cwd"]), + worktree_path=Path(data["worktree_path"]), + branch=data["branch"], + from_branch=data["from_branch"], + ) + + +def _git(cwd: Path, *args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["git", *args], + cwd=cwd, + text=True, + capture_output=True, + timeout=60, + ) + + +def sanitize_branch(raw: str) -> str: + value = raw.strip().replace(" ", "-") + value = re.sub(r"[^a-zA-Z0-9._-]", "-", value) + value = re.sub(r"-+", "-", value).strip(".-") + if not value: + raise ValueError("branch name is empty after sanitize") + if value in (".", "..") or ".." in value or "/" in value or "\\" in value: + raise ValueError(f"unsafe branch name: {raw}") + return value[:64] + + +def ensure_clean_worktree(cwd: Path) -> None: + proc = _git(cwd, "status", "--porcelain") + if proc.returncode != 0: + raise RuntimeError(proc.stderr.strip() or "git status failed") + if proc.stdout.strip(): + raise RuntimeError("working tree is dirty; commit or stash before enter_worktree") + + +def enter_worktree(branch: str, from_branch: str, state, session) -> str: + if state.worktree is not None: + return f"error: already in worktree: {state.worktree.worktree_path}" + + original = state.cwd + ensure_clean_worktree(original) + safe_branch = sanitize_branch(branch) + base = original / ".worktrees" + base.mkdir(exist_ok=True) + path = base / safe_branch + + proc = _git(original, "worktree", "add", "-b", safe_branch, str(path), from_branch) + if proc.returncode != 0: + return f"error: {proc.stderr.strip()}" + + state.worktree = WorktreeState( + original_cwd=original, + worktree_path=path, + branch=safe_branch, + from_branch=from_branch, + ) + state.cwd = path + if session is not None: + session.append_worktree_state(state.worktree) + return f"entered worktree: {path}\nbranch: {safe_branch}" +``` + +进入前检查 dirty tree 是刻意严格。主目录有未提交改动时，再 merge worktree 分支会很难判断冲突来自哪里。 + +### 1.2 `RuntimeState` 加 cwd/worktree + +打开 `agent_code/runtime.py`： + +```python +from pathlib import Path + +if TYPE_CHECKING: + from .worktree import WorktreeState +``` + +`RuntimeState` 里新增： + +```python + cwd: Path = field(default_factory=Path.cwd) + original_cwd: Path = field(default_factory=Path.cwd) + worktree: "WorktreeState | None" = None +``` + +### 1.3 `cli.py` 改成读 `state.cwd` + +交互模式创建 state 后： + +```python +state = RuntimeState(...) +state.cwd = resolved_cwd +state.original_cwd = resolved_cwd +``` + +`run_turn()` 改成： + +```python +run_agent( + line, + turn_provider, + tools, + max_steps=max_steps, + cwd=state.cwd, + state=state, + session=session, + system_prompt=system_prompt, +) +``` + +`make_slash_context()` 也改： + +```python +cwd=state.cwd +``` + +one-shot `run_once()` 同样设置： + +```python +state.cwd = cwd +state.original_cwd = cwd +``` + +### 1.4 session 记录 worktree-state + +打开 `agent_code/session.py`，`history` 解析时先跳过非消息行： + +```python +if "role" not in data or "content" not in data: + continue +``` + +新增两个方法： + +```python +def append_worktree_state(self, state) -> None: + now = datetime.now(timezone.utc).isoformat() + payload = state.to_dict() if state is not None else None + with open(self.file_path, "a", encoding="utf-8") as f: + f.write(json.dumps( + {"type": "worktree-state", "worktree": payload, "timestamp": now}, + ensure_ascii=False, + separators=(",", ":"), + ) + "\n") + + +def load_worktree_state(self): + from .worktree import WorktreeState + + latest = None + if not self.file_path.exists(): + return None + for line in self.file_path.read_text(encoding="utf-8").splitlines(): + try: + data = json.loads(line) + except json.JSONDecodeError: + continue + if data.get("type") == "worktree-state": + latest = data.get("worktree") + if latest is None: + return None + state = WorktreeState.from_dict(latest) + if not state.worktree_path.exists(): + self.append_worktree_state(None) + return None + return state +``` + +resume 时，在 `cli.py` 创建 `RuntimeState` 后： + +```python +if session is not None: + restored = session.load_worktree_state() + if restored is not None: + state.worktree = restored + state.original_cwd = restored.original_cwd + state.cwd = restored.worktree_path +``` + +路径不存在时 fail-closed：回到原 cwd，不假装还在隔离目录。 + +### 1.5 注册 `enter_worktree` + +打开 `tools.py`： + +```python +def enter_worktree_tool(args: dict[str, Any], ctx: ToolContext) -> str: + from .worktree import enter_worktree + + state = ctx.runtime_state + if state is None: + return "error: no runtime state" + branch = str(args.get("branch", "")).strip() + from_branch = str(args.get("from_branch", "HEAD")).strip() or "HEAD" + if not branch: + return "error: missing required argument 'branch'" + session = getattr(state, "session", None) + return enter_worktree(branch, from_branch, state, session) +``` + +为了让工具能拿到 session，`cli.py` 创建 state 后加： + +```python +state.session = session +``` + +注册： + +```python +registry.register(Tool( + name="enter_worktree", + description="Create a git worktree and switch the harness cwd into it.", + run=enter_worktree_tool, + parameters={ + "type": "object", + "properties": { + "branch": {"type": "string"}, + "from_branch": {"type": "string", "default": "HEAD"}, + }, + "required": ["branch"], + }, +)) +``` + +权限里把 `enter_worktree` 放到 ask 或 low-risk write。建议默认 ask，因为它会跑 git 命令并创建目录。 + +跑验证： + +```bash +$ git status --porcelain +# 必须没有输出 + +$ uv run agent-code --permission-mode acceptEdits "调用 enter_worktree，branch=fix-demo" +tool_call: enter_worktree {'branch': 'fix-demo'} +final: entered worktree: .../.worktrees/fix-demo +``` + +再问： + +```bash +> 用 bash 跑 pwd +tool_call: bash {'command': 'pwd'} +final: .../.worktrees/fix-demo +``` + +如果 `pwd` 仍然是主目录，说明 `run_turn()` 还在用旧 `resolved_cwd`，没有切到 `state.cwd`。 + +## v2：退出 worktree，支持 merge / discard / keep + +进入只是半边。退出时要明确三种动作： + +```txt +keep 切回主目录，保留 worktree 和分支 +discard 删除 worktree 和分支，需要强确认 +merge 合回主分支，成功后清理；冲突时保留现场 +``` + +### 2.1 `exit_worktree` + +在 `worktree.py` 里追加： + +```python +def _remove_worktree(original: Path, wt: WorktreeState, force: bool = False) -> str | None: + args = ["worktree", "remove"] + if force: + args.append("--force") + args.append(str(wt.worktree_path)) + proc = _git(original, *args) + if proc.returncode != 0: + return proc.stderr.strip() + proc = _git(original, "branch", "-D", wt.branch) + if proc.returncode != 0: + return proc.stderr.strip() + return None + + +def exit_worktree(action: str, state, session) -> str: + wt = state.worktree + if wt is None: + return "error: not in a worktree" + action = action.strip() + original = wt.original_cwd + + if action == "keep": + state.cwd = original + state.worktree = None + if session is not None: + session.append_worktree_state(None) + return f"left worktree and kept branch: {wt.branch}" + + if action == "discard": + err = _remove_worktree(original, wt, force=True) + if err: + return f"error: {err}" + state.cwd = original + state.worktree = None + if session is not None: + session.append_worktree_state(None) + return f"discarded worktree: {wt.branch}" + + if action == "merge": + proc = _git(original, "merge", "--no-ff", wt.branch) + if proc.returncode != 0: + return ( + "error: merge failed; worktree kept for manual resolution\n" + + (proc.stderr.strip() or proc.stdout.strip()) + ) + err = _remove_worktree(original, wt, force=False) + if err: + return f"merged but cleanup failed: {err}" + state.cwd = original + state.worktree = None + if session is not None: + session.append_worktree_state(None) + return f"merged and removed worktree: {wt.branch}" + + return "error: action must be merge, discard, or keep" +``` + +merge 冲突时不清理 worktree。现场留着，人才知道去哪里处理。 + +### 2.2 注册 `exit_worktree` + +`tools.py`： + +```python +def exit_worktree_tool(args: dict[str, Any], ctx: ToolContext) -> str: + from .worktree import exit_worktree + + state = ctx.runtime_state + if state is None: + return "error: no runtime state" + action = str(args.get("action", "")).strip() + if action not in ("merge", "discard", "keep"): + return "error: action must be merge, discard, or keep" + session = getattr(state, "session", None) + return exit_worktree(action, state, session) +``` + +注册： + +```python +registry.register(Tool( + name="exit_worktree", + description="Exit the active git worktree. action is merge, discard, or keep.", + run=exit_worktree_tool, + parameters={ + "type": "object", + "properties": { + "action": {"type": "string", "enum": ["merge", "discard", "keep"]}, + }, + "required": ["action"], + }, +)) +``` + +权限建议： + +- `keep` 可以 allow。 +- `merge` 需要 ask。 +- `discard` 需要强确认。 + +教学版最小做法：`exit_worktree` 统一走 ask，确认 UI 里显示 action。 + +### 2.3 跑验证 + +```bash +> 调用 enter_worktree branch=try-keep +... +> 调用 exit_worktree action=keep +tool_call: exit_worktree {'action': 'keep'} +final: left worktree and kept branch: try-keep +> 用 bash 跑 pwd +final: /your/main/project +``` + +discard 验证要小心，它会删除 branch： + +```bash +> 调用 enter_worktree branch=try-discard +> 调用 exit_worktree action=discard +``` + +merge 验证留到 v3 的 demo。 + +## v3：端到端修一个 failing test + +现在用一个小项目把前 12 天串起来。先准备 demo。 + +### 3.1 创建 buggy project + +在你的 `agent-code` 项目里执行： + +```bash +mkdir -p examples/buggy-python-project/src/buggy_calc examples/buggy-python-project/tests +cat > examples/buggy-python-project/pyproject.toml <<'EOF' +[project] +name = "buggy-python-project" +version = "0.1.0" +requires-python = ">=3.10" + +[tool.pytest.ini_options] +pythonpath = ["src"] +EOF + +cat > examples/buggy-python-project/src/buggy_calc/__init__.py <<'EOF' +from .calculator import divide + +__all__ = ["divide"] +EOF + +cat > examples/buggy-python-project/src/buggy_calc/calculator.py <<'EOF' +def divide(a: int, b: int) -> float: + return a / (b + 1) +EOF + +cat > examples/buggy-python-project/tests/test_calculator.py <<'EOF' +from buggy_calc import divide + + +def test_divide(): + assert divide(6, 2) == 3 +EOF + +cat > examples/buggy-python-project/README.md <<'EOF' +# Buggy Python Project + +Run: + +```bash +uv run pytest +``` + +Expected fix: `divide(6, 2)` should return `3`. +EOF +``` + +验证它确实失败： + +```bash +$ cd examples/buggy-python-project +$ uv add --dev pytest +$ uv run pytest +FAILED tests/test_calculator.py::test_divide +``` + +回到主项目根目录： + +```bash +cd ../.. +git status --short +``` + +如果这些 demo 文件还没提交，`enter_worktree` 会因为 dirty tree 拒绝。你可以先提交 demo 素材，或者在自己的练习仓库里做。 + +### 3.2 端到端任务 prompt + +启动： + +```bash +uv run agent-code --permission-mode plan +``` + +输入： + +```txt +在隔离 worktree 里修复 examples/buggy-python-project 的 failing test。 +流程必须是： +1. 调用 enter_worktree，branch=fix-buggy-divide +2. 先用 plan mode 给出计划，等我批准 +3. 用 debugger subagent 分析失败原因 +4. 修复代码 +5. 在 examples/buggy-python-project 里跑 uv run pytest +6. 通过后调用 exit_worktree action=merge +``` + +预期 trace 大概是： + +```txt +tool_call: enter_worktree {'branch': 'fix-buggy-divide'} +tool_call: exit_plan_mode {'plan_summary': '...'} +# 用户批准 +tool_call: agent {'agent_name': 'debugger', ...} +tool_call: read_file {'path': 'examples/buggy-python-project/src/buggy_calc/calculator.py'} +tool_call: file_edit {'file_path': 'examples/.../calculator.py', ...} +tool_call: bash {'command': 'cd examples/buggy-python-project && uv run pytest'} +tool_call: exit_worktree {'action': 'merge'} +final: ... +``` + +关键不是 trace 一模一样，而是这几个验收： + +```txt +1. enter 后 pwd 在 .worktrees/fix-buggy-divide。 +2. pytest 在 worktree 里从失败变通过。 +3. merge 后主目录的 calculator.py 已修复。 +4. .worktrees/fix-buggy-divide 被清理。 +5. git status 能看见 merge commit 或合并后的干净状态。 +``` + +### 3.3 为什么所有工具会自动跟着 worktree + +因为前面几天所有工具都只看 `ToolContext.cwd`： + +```txt +read_file -> resolve_in_cwd(ctx.cwd, path) +file_edit -> resolve_in_cwd(ctx.cwd, path) +bash -> subprocess.run(..., cwd=ctx.cwd) +agent -> subagent_runner(..., cwd=ctx.cwd) +``` + +Day 13 只要把 `state.cwd` 切到 worktree，再让 `run_agent(..., cwd=state.cwd)`，整个工具面就会跟着走。 + +## 收尾：今天改了哪些文件 + +今天新增一个文件： + +```txt +agent_code/worktree.py +``` + +今天改了五个已有文件： + +```txt +agent_code/runtime.py cwd / original_cwd / worktree +agent_code/session.py worktree-state 元数据 +agent_code/cli.py run_agent 和 slash context 改读 state.cwd +agent_code/tools.py enter_worktree / exit_worktree +agent_code/permissions.py worktree 工具审批 +``` + +教程里还让你在练习项目里创建： + +```txt +examples/buggy-python-project/ +``` + +## 手动 trace 一遍 + +### 路径一：enter_worktree + +```txt +1. 模型调用 enter_worktree(branch="fix-demo")。 +2. worktree.py 检查 git status --porcelain。 +3. sanitize branch。 +4. git worktree add -b fix-demo .worktrees/fix-demo HEAD。 +5. RuntimeState.cwd = .worktrees/fix-demo。 +6. session 写 worktree-state。 +7. 下一轮所有工具从新 cwd 开始。 +``` + +### 路径二：resume + +```txt +1. session JSONL 里有最后一条 worktree-state。 +2. --resume 读取它。 +3. 如果 worktree_path 还存在，state.cwd 恢复到 worktree。 +4. 如果路径不存在，写回 null，回主 cwd。 +``` + +### 路径三：merge + +```txt +1. exit_worktree(action="merge") 在主目录执行 git merge --no-ff branch。 +2. 成功后 git worktree remove path。 +3. 删除临时 branch。 +4. state.cwd 回 original_cwd。 +5. session 写 worktree-state=null。 +6. 冲突时不清理，返回错误，让人手动处理。 +``` + +## 今天有了什么 + +- **会话级 worktree**：一次只允许一个 active worktree。 +- **cwd 运行态**：工具不是各自改路径，而是统一读 `RuntimeState.cwd`。 +- **dirty tree 预检**：进入前主目录必须干净。 +- **session 恢复**：`worktree-state` 让中断后的会话回到隔离目录。 +- **merge/discard/keep**：退出动作明确，冲突不清现场。 +- **端到端 demo**：前 12 天能力第一次串成修 bug 流程。 + +## 常见问题 + +### `enter_worktree` 报 working tree is dirty + +先运行： + +```bash +git status --short +``` + +把已有改动 commit 或 stash。Day 13 故意要求主目录干净，这样 merge 时风险更小。 + +### 进入 worktree 后工具还是读主目录 + +检查 `cli.py`：`run_agent(..., cwd=state.cwd)` 和 `SlashContext(cwd=state.cwd)` 是否都改了。只改工具函数不够，下一轮仍会用旧闭包变量。 + +### `exit_worktree("merge")` 冲突了怎么办 + +不要删除 worktree。它会保留 `.worktrees/` 和分支。你可以手动解决冲突，再决定 merge 或 keep。 + +### 为什么不是每个 subagent 自己开 worktree + +今天做的是会话级 worktree。父 Agent、subagent、bash 都在同一个隔离目录里工作。每个 subagent 再开 worktree 会让路径和 merge 归属变复杂，先不做。 + +## 课后挑战 + +1. `enter_worktree` 支持复用已有 branch。 +2. 给 `exit_worktree("discard")` 做强确认短语，例如必须输入 branch 名。 +3. `/context` 显示当前是否在 worktree。 +4. merge 前自动跑一条用户配置的验证命令。 +5. worktree 目录不存在时，在 resume 提示用户恢复或丢弃状态。 + +## 思考题 + +1. **为什么 worktree 切换要放进 `RuntimeState.cwd`，而不是让每个工具自己判断？** +2. **进入前 dirty tree 预检解决了什么问题？** +3. **`merge` 冲突时为什么不能清理 worktree？** +4. **`Session.history` 为什么要跳过 `worktree-state` 元数据行？** + +## 下一天 + +今天把修改隔离到了 git worktree。最后一天接 MCP：工具不再只能写在 `tools.py` 里，而是可以从外部 server 动态接入。工具一多，还要靠 ToolSearch 按需发现，而不是把所有 schema 都塞进首轮 prompt。 diff --git a/docs/day-14-mcp-toolsearch.md b/docs/day-14-mcp-toolsearch.md new file mode 100644 index 0000000..3159e5d --- /dev/null +++ b/docs/day-14-mcp-toolsearch.md @@ -0,0 +1,882 @@ +# Day 14：MCP 生态 + ToolSearch + +前 13 天，我们的工具都写在 `agent_code/tools.py` 里。这样适合教学起步，但真实代码 Agent 不可能把所有能力都内置进去。 + +今天做最后一块：MCP。外部 server 可以通过标准协议把工具交给 harness；工具多到塞不进 prompt 时，再用 ToolSearch 按需发现。 + +跑完之后你会看到： + +- `agent-code` 能用 stdio 启动一个 MCP server。 +- 最小 JSON-RPC 流程是 `initialize -> initialized -> tools/list -> tools/call`。 +- MCP 工具会被命名成 `mcp____`，像普通工具一样回填 `tool_result`。 +- `.mcp.json` 决定项目工具池，换工具只改配置。 +- MCP 工具默认 deferred，不把完整 schema 全塞进首轮 prompt。 +- `tool_search(query)` 会把匹配的 deferred 工具“解锁”为可调用工具。 + +代码约 760 行，新增约 480 行。 + +今天分五版： + +1. v1：stdio MCP client 骨架，连 echo server 并列工具。 +2. v2：把 MCP 工具接入 ToolRegistry。 +3. v3：`.mcp.json` 配置发现和生命周期。 +4. v4：ToolSearch + deferred tools。 +5. v5：git/sqlite 端到端 2.0。 + +## 起手：今天的起点 + +Day 14 新增这些模块： + +```txt +agent_code/mcp/protocol.py +agent_code/mcp/client.py +agent_code/mcp/config.py +agent_code/mcp/registry.py +agent_code/tool_pool.py +agent_code/tools/tool_search.py +``` + +今天也需要两个 demo server。教程里会让你创建： + +```txt +examples/mcp-echo-server/server.py +examples/mcp-sqlite-server/server.py +``` + +先记住一条边界：MCP server 不是模型，MCP tool 也不是模型自己执行。它只是把工具 schema 和调用入口交给 harness。模型仍然只发 `tool_use`；真正 `tools/call` 的还是 Python harness。 + +## v1：先跑通 stdio JSON-RPC + +第一版不接 Agent Loop。我们只证明：能启动一个 server，发 `initialize`，再发 `tools/list`。 + +### 1.1 创建 echo MCP server + +在项目根目录执行： + +```bash +mkdir -p examples/mcp-echo-server +cat > examples/mcp-echo-server/server.py <<'EOF' +from __future__ import annotations + +import json +import sys + + +def send(payload: dict) -> None: + sys.stdout.write(json.dumps(payload, separators=(",", ":")) + "\n") + sys.stdout.flush() + + +def handle(req: dict) -> None: + method = req.get("method") + req_id = req.get("id") + + if method == "initialize": + send({ + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "echo", "version": "0.1.0"}, + }, + }) + return + + if method == "tools/list": + send({ + "jsonrpc": "2.0", + "id": req_id, + "result": { + "tools": [{ + "name": "echo", + "description": "Return the input text.", + "inputSchema": { + "type": "object", + "properties": {"text": {"type": "string"}}, + "required": ["text"], + }, + "annotations": {"readOnlyHint": True}, + }] + }, + }) + return + + if method == "tools/call": + params = req.get("params", {}) + text = params.get("arguments", {}).get("text", "") + send({ + "jsonrpc": "2.0", + "id": req_id, + "result": {"content": [{"type": "text", "text": str(text)}]}, + }) + return + + if req_id is not None: + send({"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": f"unknown method: {method}"}}) + + +for line in sys.stdin: + if not line.strip(): + continue + msg = json.loads(line) + # initialized 是 notification，没有 id，不需要响应 + if msg.get("method") == "notifications/initialized": + continue + handle(msg) +EOF +``` + +这个 server 用一行 JSON 作为一条 JSON-RPC message。够教学，生产里可以换正式 MCP SDK。 + +### 1.2 新增 `agent_code/mcp/protocol.py` + +```python +from __future__ import annotations + +import json +import subprocess +import threading +from typing import Any + + +class JsonRpcPeer: + def __init__(self, proc: subprocess.Popen[str], timeout: float = 10.0) -> None: + self.proc = proc + self.timeout = timeout + self._next_id = 1 + self._lock = threading.Lock() + + def request(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]: + with self._lock: + req_id = self._next_id + self._next_id += 1 + payload = {"jsonrpc": "2.0", "id": req_id, "method": method} + if params is not None: + payload["params"] = params + self._write(payload) + while True: + line = self.proc.stdout.readline() + if not line: + raise RuntimeError("MCP server closed stdout") + msg = json.loads(line) + if msg.get("id") != req_id: + continue + if "error" in msg: + raise RuntimeError(msg["error"].get("message", msg["error"])) + return msg.get("result", {}) + + def notify(self, method: str, params: dict[str, Any] | None = None) -> None: + payload = {"jsonrpc": "2.0", "method": method} + if params is not None: + payload["params"] = params + self._write(payload) + + def _write(self, payload: dict[str, Any]) -> None: + self.proc.stdin.write(json.dumps(payload, separators=(",", ":")) + "\n") + self.proc.stdin.flush() +``` + +### 1.3 新增 `agent_code/mcp/client.py` + +```python +from __future__ import annotations + +import os +import subprocess +from dataclasses import dataclass, field +from typing import Any + +from .protocol import JsonRpcPeer + + +@dataclass +class McpServerConfig: + command: str + args: list[str] = field(default_factory=list) + env: dict[str, str] = field(default_factory=dict) + + +class McpClient: + def __init__(self, name: str, config: McpServerConfig) -> None: + self.name = name + self.config = config + self.proc: subprocess.Popen[str] | None = None + self.peer: JsonRpcPeer | None = None + + def connect(self) -> list[dict[str, Any]]: + env = os.environ.copy() + env.update(self.config.env) + self.proc = subprocess.Popen( + [self.config.command, *self.config.args], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=env, + ) + self.peer = JsonRpcPeer(self.proc) + self.peer.request("initialize", {"clientInfo": {"name": "agent-code", "version": "0.1.0"}}) + self.peer.notify("notifications/initialized") + return self.list_tools() + + def list_tools(self) -> list[dict[str, Any]]: + assert self.peer is not None + result = self.peer.request("tools/list") + return result.get("tools", []) + + def call_tool(self, name: str, arguments: dict[str, Any]) -> str: + assert self.peer is not None + result = self.peer.request("tools/call", {"name": name, "arguments": arguments}) + parts: list[str] = [] + for block in result.get("content", []): + if block.get("type") == "text": + parts.append(str(block.get("text", ""))) + return "\n".join(parts) + + def shutdown(self) -> None: + if self.proc is None: + return + self.proc.terminate() + try: + self.proc.wait(timeout=1) + except subprocess.TimeoutExpired: + self.proc.kill() +``` + +### 1.4 跑最小连接 + +```bash +$ uv run python - <<'PY' +from agent_code.mcp.client import McpClient, McpServerConfig + +client = McpClient("echo", McpServerConfig("python", ["examples/mcp-echo-server/server.py"])) +tools = client.connect() +print(f"[mcp:echo] connected, tools={[t['name'] for t in tools]}") +client.shutdown() +PY +[mcp:echo] connected, tools=['echo'] +``` + +v1 到这里就够了：stdio、JSON-RPC、initialize、tools/list 跑通。 + +## v2：把 MCP 工具接进工具池 + +现在要把 echo server 的 `echo` 变成模型能调用的工具名： + +```txt +mcp__echo__echo +``` + +注意：模型看到的是 `mcp__echo__echo`，但发给 server 的仍然是原始工具名 `echo`。 + +### 2.1 新增 `agent_code/mcp/registry.py` + +```python +from __future__ import annotations + +import re +from typing import Any + +from .client import McpClient +from ..model import ToolCall, ToolResult +from ..tools import Tool, ToolContext + + +def normalize_part(value: str) -> str: + return re.sub(r"[^a-zA-Z0-9_]+", "_", value).strip("_") + + +def build_mcp_tool_name(server: str, tool: str) -> str: + return f"mcp__{normalize_part(server)}__{normalize_part(tool)}" + + +def parse_mcp_tool_name(name: str) -> tuple[str, str] | None: + if not name.startswith("mcp__"): + return None + parts = name.split("__", 2) + if len(parts) != 3: + return None + return parts[1], parts[2] + + +class McpRegistry: + def __init__(self) -> None: + self.clients: dict[str, McpClient] = {} + self.tool_to_raw: dict[str, tuple[str, str]] = {} + + def register_client(self, name: str, client: McpClient, schemas: list[dict[str, Any]]) -> list[Tool]: + self.clients[name] = client + tools: list[Tool] = [] + for schema in schemas: + raw_name = schema["name"] + tool_name = build_mcp_tool_name(name, raw_name) + self.tool_to_raw[tool_name] = (name, raw_name) + read_only = bool(schema.get("annotations", {}).get("readOnlyHint", False)) + tools.append(Tool( + name=tool_name, + description=schema.get("description", ""), + run=self._make_runner(tool_name), + parameters=schema.get("inputSchema", {"type": "object", "properties": {}, "required": []}), + is_read_only=read_only, + )) + return tools + + def _make_runner(self, tool_name: str): + def run(args: dict[str, Any], ctx: ToolContext) -> str: + server, raw_tool = self.tool_to_raw[tool_name] + return self.clients[server].call_tool(raw_tool, args) + return run + + def shutdown_all(self) -> None: + for client in self.clients.values(): + client.shutdown() +``` + +### 2.2 把 MCP tool 注册进 `ToolRegistry` + +先手动接 echo： + +```python +from agent_code.mcp.client import McpClient, McpServerConfig +from agent_code.mcp.registry import McpRegistry +from agent_code.tools import default_tools + +tools = default_tools() +mcp = McpRegistry() +client = McpClient("echo", McpServerConfig("python", ["examples/mcp-echo-server/server.py"])) +schemas = client.connect() +for tool in mcp.register_client("echo", client, schemas): + tools.register(tool) +``` + +真实接入时，这段会放进 CLI 启动流程；v2 先用 Python 验证： + +```bash +$ uv run python - <<'PY' +from pathlib import Path +from agent_code.mcp.client import McpClient, McpServerConfig +from agent_code.mcp.registry import McpRegistry +from agent_code.model import ToolCall +from agent_code.tools import ToolContext, default_tools + +tools = default_tools() +mcp = McpRegistry() +client = McpClient("echo", McpServerConfig("python", ["examples/mcp-echo-server/server.py"])) +for tool in mcp.register_client("echo", client, client.connect()): + tools.register(tool) + +result = tools.run(ToolCall("call_1", "mcp__echo__echo", {"text": "hello mcp"}), ToolContext(cwd=Path.cwd())) +print(result.content) +mcp.shutdown_all() +PY +hello mcp +``` + +到这里，MCP 工具对 Agent Loop 来说就是普通工具了。 + +## v3：`.mcp.json` 决定工具池 + +手动写连接代码不现实。项目应该用 `.mcp.json` 声明工具池。 + +### 3.1 新增配置文件 + +项目根目录新建： + +```json +{ + "mcpServers": { + "echo": { + "command": "python", + "args": ["examples/mcp-echo-server/server.py"], + "env": {} + } + } +} +``` + +用户级配置放： + +```txt +~/.config/agent-code/mcp.json +``` + +项目级覆盖用户级。 + +### 3.2 新增 `agent_code/mcp/config.py` + +```python +from __future__ import annotations + +import json +from pathlib import Path + +from .client import McpServerConfig + + +def _load(path: Path) -> dict[str, McpServerConfig]: + if not path.exists(): + return {} + data = json.loads(path.read_text(encoding="utf-8")) + servers: dict[str, McpServerConfig] = {} + for name, raw in data.get("mcpServers", {}).items(): + servers[name] = McpServerConfig( + command=raw["command"], + args=list(raw.get("args", [])), + env=dict(raw.get("env", {})), + ) + return servers + + +def load_mcp_config(cwd: Path) -> dict[str, McpServerConfig]: + user = _load(Path.home() / ".config" / "agent-code" / "mcp.json") + project = _load(cwd / ".mcp.json") + merged = dict(user) + merged.update(project) + return merged +``` + +### 3.3 CLI 启动 connect，退出 shutdown + +在 `cli.py` 里： + +```python +from .mcp.config import load_mcp_config +from .mcp.client import McpClient +from .mcp.registry import McpRegistry + + +def connect_mcp_tools(cwd: Path, tools) -> McpRegistry: + registry = McpRegistry() + for name, config in load_mcp_config(cwd).items(): + client = McpClient(name, config) + schemas = client.connect() + for tool in registry.register_client(name, client, schemas): + tools.register(tool) + console.print(f"[dim][mcp:{name}] connected, tools={[s['name'] for s in schemas]}[/dim]") + return registry +``` + +交互模式： + +```python +tools = default_tools() +mcp_registry = connect_mcp_tools(state.cwd, tools) +try: + run_interactive_shell(...) +finally: + mcp_registry.shutdown_all() +``` + +one-shot 也一样，`run_once()` 里创建 tools 后 connect，结束后 shutdown。 + +跑验证： + +```bash +$ uv run agent-code --provider mock "用 echo 工具说 hi" +[mcp:echo] connected, tools=['echo'] +... +``` + +真实模型验证： + +```bash +$ uv run agent-code "必须调用 mcp__echo__echo，参数 text=hello" +tool_call: mcp__echo__echo {'text': 'hello'} +final: hello +``` + +## v4：ToolSearch，不要把所有 schema 常驻 prompt + +如果你接了 5 个 MCP server，每个 20 个工具，把全部 schema 都塞进首轮请求，上下文会被工具描述挤掉。 + +所以 MCP 工具默认 deferred：首轮只告诉模型有哪些名字；模型需要时先调用 `tool_search(query)`，命中的工具再进入可调用集合。 + +### 4.1 `RuntimeState` 记录 discovered + +```python + deferred_tools: dict[str, str] = field(default_factory=dict) # name -> description + discovered_tool_names: set[str] = field(default_factory=set) +``` + +### 4.2 新增 `agent_code/tool_pool.py` + +```python +from __future__ import annotations + +from .tools import ToolRegistry + + +class ToolPool: + def __init__(self, registry: ToolRegistry, deferred_names: set[str]) -> None: + self.registry = registry + self.deferred_names = deferred_names + + def visible_registry(self, discovered: set[str]) -> ToolRegistry: + visible = ToolRegistry() + for tool in self.registry.list(): + if tool.name in self.deferred_names and tool.name not in discovered: + continue + visible.register(tool) + return visible + + def render_deferred(self, discovered: set[str]) -> str: + lines = [""] + for tool in self.registry.list(): + if tool.name in self.deferred_names and tool.name not in discovered: + lines.append(f"- {tool.name}: {tool.description}") + lines.append("") + return "\n".join(lines) if len(lines) > 2 else "" +``` + +MCP 工具放进 `deferred_names`，内置工具不放。`tool_search` 自己也不能 defer。 + +### 4.3 新增 `tool_search` + +新建 `agent_code/tools/tool_search.py`： + +```python +from __future__ import annotations + +from typing import Any + +from ..tools import ToolContext + + +def tool_search(args: dict[str, Any], ctx: ToolContext) -> str: + state = ctx.runtime_state + if state is None: + return "error: no runtime state" + query = str(args.get("query", "")).lower().strip() + if not query: + return "error: missing required argument 'query'" + + matches: list[str] = [] + for name, description in state.deferred_tools.items(): + haystack = f"{name} {description}".lower() + if query in haystack or any(part in haystack for part in query.split()): + matches.append(name) + if not matches: + return "(no matching deferred tools)" + + state.discovered_tool_names.update(matches) + lines = ["Discovered tools:"] + for name in matches[:10]: + lines.append(f"- {name}: {state.deferred_tools.get(name, '')}") + return "\n".join(lines) +``` + +注册到内置 tools： + +```python +registry.register(Tool( + name="tool_search", + description="Search deferred tools by keyword and make matching tools available in later turns.", + run=tool_search, + parameters={ + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, + is_read_only=False, +)) +``` + +### 4.4 provider 只看 visible tools + +在 `agent.py` provider call 前： + +```python +pool = getattr(state, "tool_pool", None) +if pool is not None: + visible_tools = pool.visible_registry(state.discovered_tool_names) +else: + visible_tools = tools +visible_tools = visible_tools.filtered(state.skill_allowed_tools) +``` + +system prompt 里追加 deferred 列表： + +```python +if state is not None and getattr(state, "tool_pool", None) is not None: + deferred = state.tool_pool.render_deferred(state.discovered_tool_names) + if deferred: + parts.append(deferred) +``` + +CLI connect MCP 后： + +```python +deferred_names = set() +for mcp_tool in mcp_tools: + tools.register(mcp_tool) + deferred_names.add(mcp_tool.name) + state.deferred_tools[mcp_tool.name] = mcp_tool.description +state.tool_pool = ToolPool(tools, deferred_names) +``` + +### 4.5 `/context` 显示工具数量 + +在 Day 11 的 `/context` 里追加： + +```python +if ctx.state and getattr(ctx.state, "tool_pool", None): + total_deferred = len(ctx.state.deferred_tools) + discovered = len(ctx.state.discovered_tool_names) + message += f"\nInline tools discovered: {discovered}\nDeferred tools: {total_deferred - discovered}" +``` + +跑验证： + +```bash +> /context +Inline tools discovered: 0 +Deferred tools: 1 + +> 请先 tool_search 搜 echo，再调用 MCP echo 工具说 hello +tool_call: tool_search {'query': 'echo'} +tool_call: mcp__echo__echo {'text': 'hello'} +final: hello +``` + +如果模型第一轮直接调用 `mcp__echo__echo`，provider 看不到这个 schema，正常不会成功。它应该先 `tool_search`。 + +## v5：端到端 2.0，git/sqlite 都从 MCP 来 + +最后证明一件事：工具池可以换，harness 不用改。 + +### 5.1 SQLite MCP server + +准备一个最小 sqlite server： + +```bash +mkdir -p examples/mcp-sqlite-server +cat > examples/mcp-sqlite-server/server.py <<'EOF' +from __future__ import annotations + +import json +import sqlite3 +import sys +from pathlib import Path + +DB = Path(__file__).with_name("demo.db") + + +def ensure_db() -> None: + conn = sqlite3.connect(DB) + conn.execute("create table if not exists notes(id integer primary key, title text)") + conn.execute("insert or ignore into notes(id, title) values(1, 'hello from sqlite mcp')") + conn.commit() + conn.close() + + +def send(payload: dict) -> None: + sys.stdout.write(json.dumps(payload, separators=(",", ":")) + "\n") + sys.stdout.flush() + + +def result(req_id, text: str) -> None: + send({"jsonrpc": "2.0", "id": req_id, "result": {"content": [{"type": "text", "text": text}]}}) + + +ensure_db() + +for line in sys.stdin: + msg = json.loads(line) + method = msg.get("method") + req_id = msg.get("id") + if method == "notifications/initialized": + continue + if method == "initialize": + send({"jsonrpc": "2.0", "id": req_id, "result": {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": "sqlite", "version": "0.1.0"}}}) + elif method == "tools/list": + send({"jsonrpc": "2.0", "id": req_id, "result": {"tools": [{ + "name": "query", + "description": "Run a read-only SQL query against the demo database.", + "inputSchema": {"type": "object", "properties": {"sql": {"type": "string"}}, "required": ["sql"]}, + "annotations": {"readOnlyHint": True} + }]}}) + elif method == "tools/call": + sql = msg.get("params", {}).get("arguments", {}).get("sql", "") + if not sql.strip().lower().startswith("select"): + result(req_id, "error: only SELECT is allowed") + continue + conn = sqlite3.connect(DB) + rows = conn.execute(sql).fetchall() + conn.close() + result(req_id, "\n".join(str(row) for row in rows)) +EOF +``` + +上面这段故意只允许 `SELECT`，因为 Day 14 不是数据库权限系统教程。 + +### 5.2 `.mcp.json` 接 sqlite 和 git + +```json +{ + "mcpServers": { + "sqlite": { + "command": "python", + "args": ["examples/mcp-sqlite-server/server.py"], + "env": {} + }, + "git": { + "command": "uvx", + "args": ["mcp-server-git"], + "env": {} + } + } +} +``` + +git MCP 的工具名要以实际 `tools/list` 为准。你可能会看到类似： + +```txt +mcp__git__git_status +mcp__git__git_diff +``` + +不要凭记忆写死，先看启动日志或让模型 `tool_search("git diff")`。 + +### 5.3 跑端到端 + +```bash +$ uv run agent-code +[mcp:sqlite] connected, tools=['query'] +[mcp:git] connected, tools=[...] + +> 先 tool_search 搜 sqlite，然后查 notes 表 +tool_call: tool_search {'query': 'sqlite'} +tool_call: mcp__sqlite__query {'sql': 'select * from notes'} +final: notes 表里有一条记录：... + +> 先 tool_search 搜 git diff，然后用 git MCP 看当前 diff +tool_call: tool_search {'query': 'git diff'} +tool_call: mcp__git__... +final: ... +``` + +这就是 Day 14 的最终论点：内置工具不再是边界。换工具池，只改 `.mcp.json`。 + +## 收尾：今天改了哪些文件 + +今天新增六个模块： + +```txt +agent_code/mcp/protocol.py +agent_code/mcp/client.py +agent_code/mcp/config.py +agent_code/mcp/registry.py +agent_code/tool_pool.py +agent_code/tools/tool_search.py +``` + +今天改了五个已有文件： + +```txt +agent_code/cli.py 启动 connect MCP，退出 shutdown +agent_code/agent.py provider tools 改走 ToolPool visible_registry +agent_code/tools.py 注册 tool_search +agent_code/runtime.py deferred/discovered/tool_pool 状态 +agent_code/slash.py /context 显示 inline/deferred 工具数 +``` + +教程里还让你创建： + +```txt +examples/mcp-echo-server/server.py +examples/mcp-sqlite-server/server.py +.mcp.json +``` + +## 手动 trace 一遍 + +### 路径一：MCP 连接 + +```txt +1. CLI 读取 .mcp.json。 +2. 对每个 server 启动子进程。 +3. 发送 initialize。 +4. 发送 initialized notification。 +5. 发送 tools/list。 +6. 把 server 工具转换成 mcp__server__tool。 +``` + +### 路径二：MCP 工具调用 + +```txt +1. 模型发 tool_use: mcp__echo__echo。 +2. registry 解析出 server=echo, raw_tool=echo。 +3. client 发送 tools/call 给 echo server。 +4. server 返回 content。 +5. harness 包成普通 tool_result 回填模型。 +``` + +### 路径三：ToolSearch + +```txt +1. MCP 工具默认 deferred，只在 prompt 里露出名字列表。 +2. 模型调用 tool_search("sqlite")。 +3. harness 把 mcp__sqlite__query 加入 discovered_tool_names。 +4. 下一轮 provider tools 里包含 mcp__sqlite__query 的完整 schema。 +5. 模型调用 query。 +``` + +## 今天有了什么 + +- **MCP stdio client**：最小 JSON-RPC 协议跑通。 +- **动态工具注入**：MCP tool 变成 `mcp__server__tool`。 +- **配置驱动工具池**：`.mcp.json` 决定项目有哪些外部工具。 +- **ToolSearch**：MCP 工具默认 deferred，用到时再解锁 schema。 +- **端到端 2.0**：git/sqlite 能力从 MCP server 进来，不再写死在 `tools.py`。 + +## 常见问题 + +### MCP server 启动后没响应 + +先手动跑： + +```bash +python examples/mcp-echo-server/server.py +``` + +确认它不会主动打印非 JSON 日志到 stdout。stdio MCP 的 stdout 必须留给 JSON-RPC；日志应该写 stderr。 + +### `.mcp.json` 配了但没加载 + +确认你启动 `agent-code` 的 cwd 就是 `.mcp.json` 所在项目根。教学版只查当前 cwd，不向父目录递归查。 + +### 为什么 MCP 工具默认 deferred？ + +MCP server 数量没有上限。全部 schema 常驻会把代码上下文挤掉。deferred 让模型先看名字，需要时再 `tool_search`。 + +### `tool_search` 能不能也 deferred？ + +不能。它是打开 deferred 工具库的钥匙。如果它自己也被藏起来，模型就没有入口发现其它工具。 + +### 为什么不做 OAuth / HTTP / elicitation？ + +那些是生产连接层。Day 14 先讲工具协议闭环：stdio、tools/list、tools/call、ToolSearch。授权和表单交互可以放扩展篇。 + +## 课后挑战 + +1. 支持从父目录向上查找 `.mcp.json`。 +2. 给 MCP server 增加启动超时和健康检查。 +3. 把 `resources/list` 接成只读资源工具。 +4. 给 `tool_search` 加简单 BM25 排序，而不是关键词包含。 +5. 给 MCP 工具加权限：readOnlyHint 自动 allow，destructiveHint 自动 ask。 + +## 思考题 + +1. **为什么模型看到的是 `mcp__server__tool`，但 server 收到的是原始 tool name？** +2. **`.mcp.json` 放项目里，比用户全局配置多解决了什么问题？** +3. **MCP 默认 deferred，内置工具默认 inline，这个差异背后的上下文预算逻辑是什么？** +4. **如果没有 ToolSearch，接入 100 个 MCP 工具会发生什么？** + +## 下一步 + +到这里，14 天主线完成了。 + +你已经从一个回声 CLI，一步步搭出一个教学版代码 Agent harness：模型 provider、工具调用、文件和 Web 工具、安全编辑、bash 权限、session/memory、slash/hooks/cron、交互 shell、Plan Mode、skills、subagents、context/cost、coordinator、worktree、MCP 和 ToolSearch。 + +它不是完整复刻任何生产 CLI，但它覆盖了大部分可教学的核心骨架。真正重要的是：你现在知道一个大模型是怎么被 harness 变成“能读代码、改文件、跑命令、管理上下文、接外部工具”的代码 Agent。