Agent 架构与设计模式盲区梳理

概述

本篇梳理在学习 MiniClaude 项目过程中，围绕 Agent 核心架构、设计模式、消息格式、工具系统暴露的知识盲区。每条盲区包含：①原来的困惑/错误理解 ②正确解释 ③代码示例 ④延伸知识。

读者画像：有 Java 基础的 Python 新手，因此文中大量使用 Java 类比（如 Tool=员工工具箱扳手、EventHandler=监控室摄像头、EventBus=内部广播站）。

盲区清单（速查表）

逐条详解

A. ReAct 循环与工具调用

A1. ReAct 循环不是 Plan-Act-Observe 直线，add_assistant_message 不是 Observe

原来怎么理解的：把数据流图标注成 PLAN → OBSERVE → ACT，把 add_assistant_message（LLM 输出落账）这一步标成了 [OBSERVE]，认为 LLM 说话就是"观察"。 正确解释：经典 ReAct 是 Plan-Act-Observe 三段论，但 LLM 输出（add_assistant_message）只是记录 LLM 说了什么，不是 Observe。真正的 Observe 是 add_tool_result（工具结果落账）。正确的标注是：LLM 输出标 [LOG]，工具结果标 [OBSERVE]。流程是 PLAN → LOG → ACT → OBSERVE。

说明：源码 loop.py#L80 的注释把 add_assistant_message 标为 [observe]，本篇认为该标注不严谨并重新标注为 [LOG]——因为它记录的是 LLM 自己的输出，而不是工具返回的环境反馈，与 ReAct 论文里"Observe = 通过工具调用获取环境反馈"的原意不符。 代码示例：

# loop.py 简化版
response = await provider.chat(...)          # [PLAN] LLM 思考
context.add_assistant_message(response)      # [LOG]  记录 LLM 说了什么（不是 Observe！）
for tc in response.tool_calls:
    result = await invoke_tool(...)          # [ACT]  执行工具
    context.add_tool_result(tc.id, result)   # [OBSERVE] 真正的 Observe：工具结果落账

延伸：ReAct 论文原意是 Reasoning + Acting 交织，"Observe"指的是 Agent 通过工具调用获取环境反馈。把 LLM 自己的输出当成 Observe 是概念混淆——LLM 说话是"想"和"说"，工具返回才是"看到"。 出处：S1-Q6, S1-Q7

---

A2. tool_use block 四字段格式（type/id/name/input）

原来怎么理解的：看 loop.py 里的 tool_calls 不知道格式是什么，type、id、name、input 字段分别干嘛不清楚。 正确解释：遵循 Anthropic Claude API 格式。一个 tool_use block 包含四个字段：

• type：标识 block 类型（tool_use）
• id：唯一调用 ID，用于匹配工具调用和工具结果（配对）
• name：工具名（如 read_file）
• input：参数 dict（如 {"path": "config.yaml"}）

代码示例：

# 一个 tool_use block 长这样
{
    "type": "tool_use",
    "id": "toolu_01abc123",
    "name": "read_file",
    "input": {"path": "config.yaml"}
}
​
# loop.py 构造 assistant 消息时打包成 block
context.add_assistant_message(response)
# response.tool_calls 里就是这种 block 列表

延伸：id 字段是关键——LLM 一次可能调多个工具，每个工具返回结果时必须带上对应的 tool_use_id，否则 LLM 无法区分哪个结果对应哪个调用。Java 类比：像 RPC 调用的 correlationId。 出处：S0-Q10, S1-Q18

---

A3. block 概念（content 是块数组，三种类型）

原来怎么理解的：什么是 "block"？为什么添加的部分叫 block？ 正确解释：Claude API 中消息的 content 不是字符串，而是块数组，每个元素是一个 block。有三种类型：

• text：纯文本
• tool_use：调用工具
• tool_result：工具返回

loop.py 把 LLM 回复打包成 block 格式存入消息历史。 代码示例：

# 一条 assistant 消息的 content 是 list，里面是各种 block
{
    "role": "assistant",
    "content": [
        {"type": "text", "text": "我先读一下配置文件"},
        {"type": "tool_use", "id": "toolu_01", "name": "read_file", "input": {"path": "config.yaml"}}
    ]
}
​
# 一条 user 消息里的 tool_result block
{
    "role": "user",
    "content": [
        {"type": "tool_result", "tool_use_id": "toolu_01", "content": "..." }
    ]
}

延伸：这种"块数组"设计让一条消息可以混合多种内容（文本+多个工具调用），比"纯字符串"灵活得多。Java 类比：像 JSX 里一个组件可以包含多个子元素。 出处：S0-Q11, S1-Q18

---

A4. invoke_tool 四参数（registry/tool_call/bus/run_id）

原来怎么理解的：看 loop.py 第 67 行 invoke_tool 调用，不知道四个参数都是干什么的。 正确解释：invoke_tool(registry, tool_call, bus, run_id) 四参数职责：

• registry：工具注册表，用来查工具实例（按 name 找）
• tool_call：LLM 要调用的工具信息（含 name/input/id）
• bus：事件总线，用来发事件（ToolCallStarted/Finished）
• run_id：当前 run 的唯一 ID

代码示例：

# invocation.py 简化
async def invoke_tool(registry, tool_call, bus, run_id):
    tool = registry.get(tool_call.name)          # ① 用 registry 找工具
    await bus.publish(ToolCallStartedEvent(
        run_id=run_id,                            # ④ run_id 串事件
        tool_name=tool_call.name,                 # ② tool_call 提供信息
        params=tool_call.input,
    ))
    result = await tool.invoke(tool_call.input)   # ① 真正执行
    await bus.publish(ToolCallFinishedEvent(...)) # ③ bus 发事件
    return result

延伸：四个参数分别代表"找谁干"、"干什么"、"通知谁"、"哪次任务"。Java 类比：方法签名的依赖注入——把工具箱、任务单、广播器、任务编号都传进来，函数本身无状态。 出处：S0-Q12, S1-Q19

---

A5. run_id 全程不变（从 runner 生成到 run 结束）

原来怎么理解的：run_id 是自从这次 agent 启动就不变的？不确定。 正确解释：是。run_id 由 runs.py 的 new_run_id() 生成（格式如 20260625-143045-a1b2c3），runner.py 和 app.py 两处都调用这个函数，整个 agent run 期间不变，用于串联所有事件、工具调用和日志。所有事件都带这个 run_id，trace 文件、events.jsonl、目录名都用它。 代码示例：

# runs.py（run_id 的唯一生产者）
def new_run_id() -> str:
    ts = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
    suffix = uuid.uuid4().hex[:6]
    return f"{ts}-{suffix}"   # 如 "20260625-143045-a1b2c3"
​
# runner.py 与 app.py 各自调用 new_run_id() 生成一次，全程传递
run_id = new_run_id()
context = ExecutionContext(goal=goal, run_id=run_id)
# 后续 provider.chat、invoke_tool、bus.publish 全部带上同一个 run_id

延伸：时间戳确保可读+有序，随机后缀确保同一秒内不冲突。Java 类比：MDC（Mapped Diagnostic Context）里的 traceId，一次请求一个 ID 串起所有日志。 出处：S0-Q13, S1-Q19

---

A6. 原生 tool_use vs 文本描述工具（为什么不让 LLM 在文本里说"我要调 read_file"）

原来怎么理解的：不清楚为什么非要用原生 tool_use block，让 LLM 在文本里描述想调什么工具不行吗？ 正确解释：5 个原因：

0. 可靠性：结构化数据不用解析自然语言（文本解析脆弱）
1. 类型保证：JSON Schema 约束参数类型
2. 多工具：content 是 list，天然支持一次多个 tool_use block
3. 结果配对：tool_use_id 匹配调用和结果
4. 训练优化：Claude 被专门训练成原生支持 tool_use

代码示例：

# ❌ 文本描述方式（ReAct 论文时代的脆弱做法）
# LLM 输出: "Thought: 我需要读文件\nAction: read_file\nAction Input: config.yaml"
# 然后代码用正则解析 → 一旦 LLM 格式错了就崩
​
# ✅ 原生 tool_use 方式（现代 API）
# LLM 直接返回结构化 block
{"type": "tool_use", "id": "toolu_01", "name": "read_file", "input": {"path": "config.yaml"}}
# 不用解析，直接 dict["name"] / dict["input"] 取值

延伸：历史演进——从 ReAct 论文的文本解析（脆弱）到现代 API 原生支持。这反映了 LLM 工具调用从"提示工程"走向"协议化"。 出处：S1-Q69

---

A7. LLM 自我修复工具错误机制（错误塞回对话历史）

原来怎么理解的：工具错误不在 Loop 处理，由 invocation.py 转成 ToolResult(is_error=True) 塞回对话。但 LLM 怎么自我修复？ 正确解释：工具失败的结果作为 tool_result 塞回 messages（role 是 user，content 是错误消息）。LLM 下一轮看到这个"用户反馈"，会重新思考：可能调整参数重新调用，可能换一个工具，可能放弃这个方法。工具错误是可恢复的，LLM 错误是不可恢复的。 代码示例：

# invocation.py 兜底分支
except Exception as exc:
    error_class = "runtime_error"
    error_message = str(exc)
    # ↑ 经 _fail() 包装成 ToolResult(content=str(exc), is_error=True, error_type="runtime_error")
# content 只有 str(exc)，错误类型通过 error_type 字段单独传递，LLM 下一轮能看到
​
# 下一轮 LLM 看到 tool_result(is_error=True) 后可能这样想：
# "read_file 报 FileNotFoundError，可能路径错了，我先用 list_dir 看看目录结构"

延伸：错误信息越具体，修复越准确。所以 invocation.py 把异常消息（str(exc)）作为 content 塞回对话，错误分类（error_class="runtime_error"）单独存一个字段，方便重试逻辑判断。Java 类比：catch 块里把异常栈返回给调用方，而不是吞掉。这是 Agent 系统"鲁棒性"的核心来源——不用代码处理所有异常，让 LLM 像人一样看错误信息调整策略。 出处：S0-Q16, S1-Q21

---

B. 错误处理与分级

B1. CancelledError vs Exception 处理差异（取消向外抛，LLM 错误自己兜）

原来怎么理解的：看 loop.py 第 46-51 行 try/except，不知道为什么 CancelledError 和普通 Exception 处理方式不同。 正确解释：

• asyncio.CancelledError：任务被取消，raise 立即向上传播（不继续循环），让上层 runner.py 处理。尊重协程取消信号约定。
• 其他异常：LLM 出错了，标 context.mark_failed("llm_error") 然后 break 退出整个循环，由 AgentLoop 自己兜底。不是下一轮循环，而是终止整个 run。

代码示例：

# loop.py 简化
try:
    response = await provider.chat(...)
except asyncio.CancelledError:
    context.mark_failed("cancelled")
    raise              # ← 向外抛，让上层处理
except Exception as exc:
    context.mark_failed("llm_error")
    break              # ← 退出循环，整个 run 结束

延伸：Java 类比——CancelledError 像 InterruptedException（要尊重中断信号，不能吞），普通 Exception 像业务异常（记日志后退出当前流程）。关键区别：取消是"外部叫我停"，LLM 错是"我自己崩了"。 出处：S0-Q14, S1-Q20

---

B2. Python vs Java 中断机制差异（异常对象 vs 标志位）

原来怎么理解的：Python 处理中断需要像 Java 那样重设中断标志吗？ 正确解释：不需要。两者机制完全不同：

• Java：基于标志位。捕获 InterruptedException 后标志被清除，必须手动 Thread.currentThread().interrupt() 重设，否则上层感知不到。
• Python：基于异常对象。CancelledError 本身就是信号，直接 raise 传播即可，不需要"重设"。

代码示例：

# Python：直接 raise，信号自动传播
except asyncio.CancelledError:
    cleanup()
    raise   # ← 就这样，上层自动收到取消信号

# Java 对比：必须手动重设标志
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();  // ← 不写这行，上层就感知不到了
    throw e;
}

延伸：Python 的异常机制比 Java 的标志位更"显式"——异常是一个对象，可以携带信息，传播路径清晰。Java 的标志位是"隐式状态"，容易忘记重设导致 bug。 出处：S0-Q15

---

B3. 错误分级：可恢复 vs 致命（分界线是"LLM 能不能自己修"）

原来怎么理解的：错误分级的"可恢复 vs 致命"二分法在代码中怎么体现？ 正确解释：

• 致命错误（loop.py）：LLM 出错 → mark_failed + break / raise 终止整个 run。
• 可恢复错误（invocation.py）：工具执行出错 → 转成 ToolResult(is_error=True) 继续循环。
• 分界线：错误能不能让 LLM 自己修复。LLM 崩了没人修，工具错了 LLM 能看错误信息调整。

代码示例：

# 致命错误（loop.py）— LLM 出错，整个 run 终止
except Exception as exc:
    context.mark_failed("llm_error")  # 标记 run 失败
    break                              # 退出循环，run 结束

# 可恢复错误（invocation.py）— 工具出错，塞回对话继续
except Exception as exc:
    return ToolResult(is_error=True, content=str(exc))
    # ↑ 不终止 run，下一轮 LLM 看到错误自己调整

延伸：这个二分法是 Agent 系统的核心设计哲学。Java 类比：急诊分诊——TimeoutError 像外伤去外科，Exception 像急诊兜底。致命错误是"系统级故障"（LLM 挂了），可恢复错误是"业务级故障"（工具报错），层级不同处理方式不同。 出处：S0-Q17

---

C. 事件系统与设计模式

C1. EventBus 发布-订阅模式（进程内广播站）

原来怎么理解的：eventbus 是什么东西？bus.subscribe(h) 是什么操作？和 Kafka、Redis Pub/Sub 有什么区别？ 正确解释：EventBus 是"内部广播站"，实现了发布-订阅模式（Pub/Sub）。publish(event) = 广播一条消息；subscribe(handler) = 让某个处理函数登记关注。作用是解耦：事件产生者不知道谁会处理，只管发；消费者也不知道谁产生，只管订阅。

关键澄清：EventBus 是进程内通信，就是函数调用，不跨进程。Kafka/Redis 是跨机器跨进程的。MiniClaude 的进程间通信通过 TCP socket + JSON-RPC 实现，daemon 里有个特殊 handler 订阅 EventBus，收到事件后转 JSON 通过 socket 发给 client。

代码示例：

# events/bus.py 只有 21 行
class EventBus:
    def __init__(self):
        self._subscribers: list[Callable] = []

    def subscribe(self, handler):
        self._subscribers.append(handler)   # 登记

    async def publish(self, event):
        for handler in self._subscribers:   # 遍历
            await handler(event)            # 逐个调用

延伸：Java 类比——EventBus 像 Guava EventBus 或 Spring ApplicationEventPublisher，但更轻量（21 行 vs 几百行）。发布者不知道谁订阅，订阅者不知道谁发布，这是松耦合的核心。 出处：S1-Q27, S1-Q28, S1-Q43

---

C2. await handler(event) 本质是函数调用（没有网络、没有队列）

原来怎么理解的：await handler(event) 这行怎么就能发送消息？event 怎么传给每个 handler 函数？ 正确解释：就是普通函数调用——把 event 当参数传给 handler 函数。"发送消息" = "调用函数"，没有网络、没有队列、没有 broker。handler 是一个函数对象，handler(event) 就是调用它。EventBus 里的 list 存着函数引用，publish 时遍历调用。 代码示例：

# 这两行等价
await handler(event)
# 和
await some_async_function(event)  # ← 就是普通异步函数调用

# handler 可能是 StdoutPrinter.handle、EventWriter.write 等
# 它们都是 async def 定义的函数，被存进 _subscribers list

延伸：这是初学者最大的"去神秘化"——以为"事件系统"很高深，其实就是 for 循环 + 函数调用。"自动干活"不是神奇通知，而是 for 循环挨个调用。Java 类比：像 for (Listener l : listeners) l.onEvent(event);，没有任何魔法。 出处：S1-Q29

---

C3. Tool vs EventHandler 区别（扳手 vs 摄像头）

原来怎么理解的：handler 不是 agent 工具？是日志工具？ 正确解释：两者完全不同：

• Tool（agent 工具）：LLM 可以主动调用的能力（如 ReadFileTool），注册在 ToolRegistry，LLM 知道它们存在。
• EventHandler：被动接收系统事件的处理函数（如 StdoutPrinter、EventWriter），注册在 EventBus，LLM 完全不知道。

代码示例：

# Tool — LLM 主动调用
class ReadFileTool(BaseTool):
    name = "read_file"
    async def invoke(self, input): ...
# LLM 决定调它，参数 LLM 定

# EventHandler — 系统被动接收
class StdoutPrinter:
    async def handle(self, event):
        print(event)
# LLM 不知道它存在，事件来了自动被调用

延伸：Java 类比——Tool 像员工工具箱里的扳手（员工主动拿来用），EventHandler 像监控室的摄像头（被动记录发生的事，员工不知道自己在被拍）。两者注册位置不同（ToolRegistry vs EventBus），触发方式不同（LLM 调 vs 事件推），可见性不同（LLM 知道 vs LLM 不知道）。 出处：S1-Q32

---

C4. 依赖注入 provider = self._provider or AnthropicProvider(...)

原来怎么理解的：为什么需要 or？ 正确解释：Python 的短路或：传了 provider 就用传入的，没传（None）就自动创建默认的 AnthropicProvider。用于支持测试用 Mock、扩展用其他 LLM——这是依赖注入模式。99% 用户不传享受简单默认，高级用户传入自定义实现。 代码示例：

# runner.py
class AgentRunner:
    def __init__(self, config, provider=None, ...):
        self._provider = provider

    async def run(self, goal, run_id=None):
        provider = self._provider or AnthropicProvider(self._config.llm.default_model)
        #           ↑ 传了用传入的     ↑ 没传自动创建默认（接收 model 字符串）

# 测试时注入 Mock
runner = AgentRunner(config, provider=MockProvider())
# 生产时不传，用默认
runner = AgentRunner(config)

延伸：Java 类比——Spring 的 @Autowired(required=false) + @Bean 默认实现。Python 没有 IoC 容器，用短路或手动实现依赖注入。好处：解耦 + 可测试性，测试时不用改生产代码就能换 Mock。 出处：S1-Q33

---

C5. extra_handlers 扩展点设计（接口宽松、内部严格）

原来怎么理解的：extra_handlers 在两个地方用 or，起到什么作用？ 正确解释：两次 or 各有用途：

• 第一次 = None：避开 Python 可变默认参数共享大坑（函数定义时只创建一次 []，所有调用共享同一个列表）。
• 第二次 or []：把 None 兜底成空列表，让后面代码不用判断是不是 None。

这是"接口宽松、内部严格"的设计哲学——调用方传 None 也行、传 list 也行，内部统一当成 list 处理。

代码示例：

# 接口宽松：默认 None（不是 []）
def __init__(self, ..., extra_handlers=None):
    self._extra_handlers = extra_handlers or []
    #                     ↑ None → []

# 内部严格：后面统一当 list 用
for h in self._extra_handlers:
    bus.subscribe(h)
    # ↑ 不用判断 if self._extra_handlers:

延伸：整个项目里目前只有 CLI 模式传入 StdoutPrinter.handle（把事件格式化打印到终端）。这是扩展点设计——预留口子给未来，目前只有一个使用者。Java 类比：Spring 的 @ConditionalOnMissingBean，有默认实现但允许覆盖。 出处：S1-Q31, S1-Q34

---

C6. 可变默认参数共享大坑（field(default_factory=list)）

原来怎么理解的：messages: list = field(default_factory=list) 防止什么？field 是字段工厂的意思？ 正确解释：防止 Python 可变默认参数共享大坑——如果直接写 messages: list = []，函数/类定义时只创建一次 []，所有实例共享同一个列表，导致用户 A 的对话历史污染用户 B 的上下文。default_factory=list 每次创建实例时都调用一次 list() 生成新空列表。

field 不是"字段工厂"，是 @dataclass 的字段配置函数。default_factory 才是工厂。

代码示例：

# ❌ 大坑：所有实例共享同一个 list
@dataclass
class Bad:
    messages: list = []   # 定义时创建一次，所有实例共享！

a = Bad(); a.messages.append("A")
b = Bad()
print(b.messages)  # ['A'] ← 被污染了！

# ✅ 正确：每次创建新 list
@dataclass
class Good:
    messages: list = field(default_factory=list)
    #                          ↑ 每次实例化调一次 list()

a = Good(); a.messages.append("A")
b = Good()
print(b.messages)  # [] ← 干净的

延伸：这个坑在函数默认参数里也存在：def f(items=[]) 同样共享。规则：可变值（list/dict/set）必须用工厂，不可变值（int/str/None）可以直接用 =。Java 类比：Java 没这个坑因为对象默认值是 null，但 Python 的可变默认值是"共享单例"。 出处：S1-Q34, S1-Q48

---

C7. Provider Protocol 抽象（接口与实现分离）

原来怎么理解的：LLM 返回 tool_use: get_weather("北京")。Provider 内部解析。但 llm 返回消息还没有写逻辑？点进 chat 函数只看到参数列表和三个点。 正确解释：base.py 里的 LLMProvider 是 Protocol（接口），只有方法签名没有实现（三个点 ...）。真正的实现在 provider.py 的 AnthropicProvider 类里。Protocol 定义"应该有什么方法"，不关心具体实现。 代码示例：

# base.py — Protocol（接口）
class LLMProvider(Protocol):
    # 流式调用 LLM 并发布进度事件，返回完整响应
    async def chat(
        self,
        messages: list[dict[str, object]],
        tool_schemas: list[dict[str, object]],
        bus: EventBus,
        run_id: str,
        *,
        step: int = 0,
        system: str | None = None,
    ) -> LlmResponse: ...   # ← 只有签名，没有实现

# provider.py — 真正实现
class AnthropicProvider:
    def __init__(self, model: str, client: Any = None):
        self._client = client or anthropic.AsyncAnthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
        self._model = model

    async def chat(self, messages, tool_schemas, bus, run_id, *, step=0, system=None) -> LlmResponse:
        # 真正调 Anthropic API 的逻辑在这里
        async with self._client.messages.stream(**kwargs) as stream:
            ...

延伸：Protocol vs ABC——Protocol 更轻量（鸭子类型，不需要继承），ABC 更严格（继承检查）。Java 类比：Protocol 像 interface（但不需要显式 implements），ABC 像 abstract class。MiniClaude 的 BaseTool 是 ABC，LLMProvider 是 Protocol，按严格度选不同机制。 出处：S1-Q4, S1-Q57

---

C8. TracingProvider 装饰器模式（包装 inner provider 加 trace）

原来怎么理解的：TracingProvider 是什么设计模式？include_payload 是干什么的？ 正确解释：TracingProvider 是装饰器模式（Decorator）——包装一个 inner provider，在调用前后加 trace 记录，不改变接口。include_payload 控制 trace 记录里写不写 LLM 的完整输入输出内容（True 写完整 messages，False 只写摘要）。 代码示例：

# TracingProvider 包装 AnthropicProvider
class TracingProvider:
    def __init__(self, inner, trace, *, include_payload: bool = True):
        self._inner = inner           # ← 被包装的真实 provider
        self._trace = trace
        self._include_payload = include_payload

    async def chat(self, messages, *, tools, system, step):
        # 调用前记 trace
        self._trace.emit(direction="CORE→LLM", step=step,
                        data=messages if self._include_payload else {"count": len(messages)})
        # 委托给 inner
        response = await self._inner.chat(messages, tools=tools, system=system, step=step)
        # 调用后记 trace
        self._trace.emit(direction="LLM→CORE", step=step, ...)
        return response

# 组装：AnthropicProvider 被 TracingProvider 包着
provider = TracingProvider(AnthropicProvider(config), trace, include_payload=True)

延伸：装饰器模式的核心是"接口不变，功能增强"。调用方不知道拿到的是真实 provider 还是包装后的。Java 类比：Java IO 的 BufferedInputStream(new FileInputStream(...))，一层包一层。include_payload 开关像 Log4j 的 DEBUG vs INFO 级别。 出处：S3-Q6, S3-Q7

---

C9. Command-Query Separation：run() vs run_and_capture()

原来怎么理解的：run() 和 run_and_capture() 为什么多套一层？ 正确解释：两种调用场景需要不同返回值，遵循 Command-Query Separation (CQS) 原则——一个方法要么做事情（Command），要么回答问题（Query），不要两者兼顾。

run_and_capture() 是真正干活的，返回 RunOutcome。run() 是薄包装，丢弃返回值。 代码示例：

# run_and_capture() 真正干活，返回结果
async def run_and_capture(self, goal, run_id=None) -> RunOutcome:
    ...  # 完整逻辑
    return RunOutcome(status=..., result=...)

# run() 是薄包装，丢弃返回值（fire-and-forget）
async def run(self, goal, run_id=None) -> None:
    await self.run_and_capture(goal, run_id=run_id)
    # ← 返回值丢弃

延伸：Java 经典对照——Runnable.run() 返回 void（发射后不管）vs Callable.call() 返回值（发射后要结果）；Process.start() vs Process.waitFor()。这不是 GoF 23 个设计模式之一，是编程原则层面的设计。 出处：S3-Q5, S3-Q6

---

C10. S3 用到的 GoF 模式（Decorator/Observer/Factory Method）

原来怎么理解的：MiniClaude 用了什么设计模式？ 正确解释：S3 真正用到的 GoF 模式有三个：

• Decorator（装饰器）：TracingProvider 包装 AnthropicProvider，加 trace 不改接口
• Observer（观察者）：EventBus 的 publish-subscribe，发布者不知道谁订阅
• Factory Method（工厂方法）：_build_registry 创建工具注册表，封装创建逻辑

代码示例：

# Decorator — TracingProvider
provider = TracingProvider(AnthropicProvider(config), trace)
# 接口不变，加了 trace 功能

# Observer — EventBus
bus.subscribe(printer.handle)   # 订阅
await bus.publish(event)        # 发布，所有订阅者被调用

# Factory Method — _build_registry
def _build_registry() -> ToolRegistry:
    registry = ToolRegistry()
    registry.register(ReadFileTool())
    registry.register(BashTool())
    registry.register(TaskCreateTool(...))
    return registry
# 创建逻辑封装在工厂方法里，调用方不关心怎么建

延伸：设计模式不是"为了用而用"，是"解决问题自然出现"。MiniClaude 没硬套模式——需要解耦用 Observer，需要增强用 Decorator，需要封装创建用 Factory。Java 类比：Spring 里到处是这三种模式，但 Python 更轻量，很多时候用函数就能解决，不需要类层级。 出处：S3-Q6

---

D. 消息格式与数据流

D1. 消息 4 层嵌套结构（messages→message→content→block）

原来怎么理解的：整体消息格式的格式脉络详细讲解，搞晕了。 正确解释：4 层嵌套：

0. messages 是 list（对话历史）
1. 每条 message 是 dict（含 role + content）
2. content 是 list（block 列表，有时是字符串简化形式）
3. 每个 block 是 dict（type: text/tool_use/tool_result）

代码示例：

# 4 层嵌套全景
messages = [                          # 第1层: list
    {                                 # 第2层: message dict
        "role": "user",
        "content": "帮我读 config"     # 第3层: content（字符串简化形式）
    },
    {                                 # 第2层: message dict
        "role": "assistant",
        "content": [                  # 第3层: content (list)
            {                         # 第4层: block dict
                "type": "text",
                "text": "好的"
            },
            {                         # 第4层: block dict
                "type": "tool_use",
                "id": "toolu_01",
                "name": "read_file",
                "input": {"path": "config"}
            }
        ]
    }
]

延伸：Java 类比：像 JSON 树结构，层层嵌套。关键是 content 有两种形式（字符串或 list），这是 API 的"语法糖"——简单文本用字符串省事，复杂内容用 list 装 block。MiniClaude 的 ExecutionContext.messages 就是这个结构。 出处：S1-Q55

---

D2. role: user/assistant/system 约定

原来怎么理解的：消息有几种 role？分别什么时候用？ 正确解释：消息类型 3 种 role：

• user：用户输入 / 工具结果（所有外部信息都用 user）
• assistant：LLM 输出
• system：MiniClaude S1 没用（系统指令，单独字段传）

代码示例：

# user 消息（用户输入）
{"role": "user", "content": "帮我读 config"}

# assistant 消息（LLM 输出）
{"role": "assistant", "content": [{"type": "text", "text": "好的"}, ...]}

# user 消息（工具结果，也是 user role！）
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "...", "content": "..."}]}

延伸：Anthropic API 要求 user/assistant 交替，多个 tool_result 要合并放进同一条 user 消息。Java 类比：像聊天协议的消息类型枚举，但这里工具结果"借用" user role 是 API 设计的取舍。 出处：S1-Q53

---

D3. tool_result 用 user role 的原因（外部信息都用 user）

原来怎么理解的：工具调用消息是追加成 user 消息？为什么工具结果不算 assistant？ 正确解释：工具结果用 role: "user" 是 Anthropic API 约定——所有外部信息都用 user 角色，靠 content 是字符串还是 list 区分。合并是因为 Anthropic 要求 user/assistant 交替，多个 tool_result 要放进同一条 user 消息。 代码示例：

# 一个 LLM 回复（assistant）含 2 个 tool_use
{"role": "assistant", "content": [
    {"type": "tool_use", "id": "t1", "name": "read_file", "input": {...}},
    {"type": "tool_use", "id": "t2", "name": "list_dir", "input": {...}}
]}

# 下一轮必须是一条 user 消息，把 2 个 tool_result 合并进去
{"role": "user", "content": [
    {"type": "tool_result", "tool_use_id": "t1", "content": "文件内容..."},
    {"type": "tool_result", "tool_use_id": "t2", "content": "目录列表..."}
]}
# ↑ 不能拆成两条 user 消息，必须合并（交替规则）

延伸：为什么工具结果不是新 role（如 "tool"）？Anthropic 设计选择——保持 role 体系简单（只有 user/assistant/system），用 content 类型区分。OpenAI 的 API 用 role: "tool"，不同 API 设计不同。 出处：S1-Q53

---

D4. content 字符串 vs list 两种形式

原来怎么理解的：content 有时候是字符串有时候是 list，搞不清什么时候用哪种。 正确解释：

• 用户输入 content 是字符串（简化形式），因为通常就是纯文本
• LLM 输出和工具结果 content 必须是 list（标准形式），因为可能含多个 block

代码示例：

# 简化形式：用户输入（字符串）
{"role": "user", "content": "帮我读 config"}

# 标准形式：LLM 输出（list of block）
{"role": "assistant", "content": [
    {"type": "text", "text": "好的"},
    {"type": "tool_use", ...}
]}

# 标准形式：工具结果（list of block）
{"role": "user", "content": [
    {"type": "tool_result", ...}
]}

延伸：这是 API 的"语法糖"——简单场景用字符串省事，复杂场景用 list。MiniClaude 在 add_tool_result 里用 if 条件 and ... 判断要不要追加到现有 user 消息，就是因为合并逻辑依赖 content 是 list。 出处：S1-Q55

---

D5. Anthropic API 流式调用 stream.text_stream（逐 token 接收）

原来怎么理解的：provider.py 的 chat() 方法整体讲解，stream.text_stream 是什么？ 正确解释：chat() 有 7 个阶段：发布模型选定事件 → 构造 system prompt（缓存省钱）→ 构造 tools（缓存省钱）→ 打包 kwargs → 流式调用 Anthropic API 逐 token 收集 → 提取 token 统计 → 提取 tool_calls → 返回 LlmResponse。核心是流式调用：通过 stream.text_stream 逐 token 接收，每个 token 都发 LlmTokenEvent，同时收集到 text_parts 最后拼接。 代码示例：

# provider.py 简化
async with self._client.messages.stream(**kwargs) as stream:
    text_parts = []
    async for text in stream.text_stream:        # 逐 token
        text_parts.append(text)
        await bus.publish(LlmTokenEvent(run_id=run_id, token=text, ts=_now()))  # 每个 token 发事件
    final = await stream.get_final_message()      # 拿完整消息（含 tool_calls）

text = "".join(text_parts)                        # 拼接所有 token
return LlmResponse(text=text, tool_calls=final.content...)

延伸：SDK 有多个流——text_stream（只拿文本 token，实时显示用）、原始事件流（MiniClaude 没用）、get_final_message（拿完整结构化消息+tool_calls）。"".join(text_parts) 就是简单拼接，"".join 是空字符串分隔符拼接，性能比 + 好。Java 类比：像 SSE（Server-Sent Events）逐事件接收。 出处：S1-Q57, S1-Q63

---

D6. SDK 封装底层 HTTPS（看不见的网络调用）

原来怎么理解的：async with self._client.messages.stream(**kwargs) 这行哪调用了 LLM？没看见啊。 正确解释：就是这一行在调 LLM！ self._client 是 Anthropic SDK 客户端，messages.stream() 是 SDK 的流式调用方法。SDK 内部发了 HTTPS 请求。看不到是因为 SDK 封装了：拼 URL、设 headers、序列化 JSON、发 HTTPS、接收流，全是 SDK 内部做的。 代码示例：

# 这一行背后 SDK 做了：
async with self._client.messages.stream(**kwargs) as stream:
# SDK 内部：
# 1. 拼 URL: https://api.anthropic.com/v1/messages
# 2. 设 headers: Authorization: Bearer xxx, Content-Type: application/json
# 3. 序列化 kwargs 成 JSON
# 4. 发 HTTPS POST 请求
# 5. 接收 SSE 流式响应
# 6. 解析每个 chunk 成 text token
# 你只看到 stream.text_stream 拿 token，网络细节全隐藏

# 项目里 SDK 用在哪？
# - anthropic SDK：只在 llm/provider.py 用（调 Claude API）
# - pydantic：在 bus/events.py 等用（事件定义+校验+序列化）
# - python-dotenv：只在 config.py 用（加载 .env）
# 其他全是 Python 标准库

延伸：Java 类比——像用 OkHttp 或 Retrofit，你调一个方法，库帮你发 HTTP 请求。SDK 的价值就是"把复杂的网络协议封装成方法调用"。MiniClaude 运行时只依赖 5 个第三方库（pydantic、python-dotenv、anthropic、textual、httpx），非常轻量。 出处：S1-Q61, S1-Q62

---

D7. cache_control 缓存标记机制（标在末尾缓存整段）

原来怎么理解的：provider.py tools = list(tool_schemas) + 给最后一个加 cache_control，在干什么？标记在最后一个 tool 上，怎么缓存整个列表？ 正确解释：让 Anthropic API 把整个 tools 列表缓存 5 分钟——agent 多轮循环不重复付 token 费用。关键澄清：是"从请求开头到标记位置（含）全部缓存"，不是"往后"。所以标在最后一个 = 从第一个到最后一个全缓存 = 整个 tools 列表。 代码示例：

# provider.py 简化
tools = list(tool_schemas)              # ① 浅拷贝外层 list
last = dict(tools[-1])                  # ② 单独拷贝要改的那个 dict
last["cache_control"] = {"type": "ephemeral"}  # ③ 加缓存标记
tools[-1] = last                        # ④ 替换回去
# 现在 tools 最后一个有 cache_control，整段被缓存 5 分钟

# 多轮循环时，tools 内容不变 → 命中缓存 → 省 token 费用
# 第1轮: 全额付费 + 标记缓存
# 第2-5轮（5分钟内）: 缓存命中，tools 部分不重复计费

延伸：类比——标记像"整箱寄存"标签贴在末尾，覆盖整箱。system prompt 也用同样手法缓存。这是省钱的工程优化，agent 多轮循环时 tools 和 system 基本不变，缓存能省大量费用。 出处：S1-Q58, S1-Q60

---

D8. list() 浅拷贝 + 单独 dict() 拷贝要改的

原来怎么理解的：list() 只拷贝了外层 list，里层的 dict 还是共享的？Python 这什么玩意？ 正确解释：list() 是浅拷贝——只复制外层 list，里面 dict 共享。要改 dict 必须单独 dict() 拷贝，否则污染原数据。深拷贝用 copy.deepcopy 但太重，MiniClaude 只拷贝要改的那一个。性能+正确性兼顾。 代码示例：

# ❌ 直接改会污染原数据
tools = list(tool_schemas)          # 浅拷贝外层
tools[-1]["cache_control"] = {...}  # ← 改的是共享的 dict！原数据被污染

# ✅ 正确：单独拷贝要改的
tools = list(tool_schemas)          # ① 浅拷贝外层 list
last = dict(tools[-1])              # ② 单独拷贝要改的 dict
last["cache_control"] = {...}       # ③ 改拷贝，不影响原数据
tools[-1] = last                    # ④ 替换回去

# 三种拷贝对比
# list(x)      — 浅拷贝，外层新 list，内层共享
# dict(x)      — 浅拷贝，外层新 dict，内层共享
# copy.deepcopy(x) — 深拷贝，全部新建（太重，慎用）

延伸：Python 的拷贝语义是初学者大坑。Java 类比：Java 默认是引用赋值，要拷贝得显式 clone 或 new。Python 的 list() / dict() 像Java 的浅拷贝构造器，copy.deepcopy 像 Java 的深拷贝序列化。原则：只拷贝要改的，不深拷贝全部。 出处：S1-Q59

---

D9. EventPushEnvelope kind 字段（当前固定 'event'，event 字段是 dict）

原来怎么理解的：EventPushEnvelope 类的 kind 字段还能有除了 event 其他的？ 正确解释：kind 字段当前固定为 "event"，由 Literal["event"] 锁死，不是预留的扩展点——源码里没有 kind: "error"、kind: "system" 等其他取值。另一个易混淆点：event 字段类型是 dict[str, Any]（已经 model_dump() 过的序列化结果），不是 Event 模型对象。_dispatch 根据 kind 字符串分流，但目前只走 event 分支。 代码示例：

# envelope.py 实际定义
class EventPushEnvelope(BaseModel):
    kind: Literal["event"] = "event"   # Literal 锁死，不是预留扩展
    event: dict[str, Any]              # Event.model_dump() 的序列化结果，不是 Event 对象

# 客户端 _dispatch 根据 kind 分流
def _dispatch(msg):
    if msg.get("kind") == "event":
        on_event(msg["event"])      # 调事件回调（event 是 dict）

延伸：为什么 JSON-RPC 响应不用判别联合？设计选择——事件系统内部用 Pydantic 判别联合（discriminator="type"）因为事件类型多（20+ 种），结构各不相同。JSON-RPC 响应结构固定（{id, result} 或 {id, error}），没必要引入同样的复杂度。Java 类比：像协议头里固定取值的 version 字段。 出处：S2-Q48

---

D10. TraceRecord 字段含义（系统运行日志的一行）

原来怎么理解的：TraceRecord 的各个字段是什么意思？ 正确解释：TraceRecord 就是"系统运行日志里的一行"——记录了"什么时候、在哪个子系统、数据往哪流、具体内容是什么"。

代码示例：

# direction + layer 组合决定记录含义
# CLIENT→CORE + ipc = 客户端发来命令
# CORE→LLM + llm = daemon 调 LLM
# LLM→CORE + llm = LLM 返回响应
# CORE→CLIENT + ipc + push = 主动推送事件
# CORE + event = 内部事件

# data 为什么是开放 dict？
# 每层关注点不同——IPC 关心命令方法名，LLM 关心 token 用量，
# Event 关心事件类型。硬统一反而会让每层塞一堆用不到的字段

延伸：Java 类比——像 Log4j 的 MDC + 结构化日志。mini trace 命令就是把这些记录彩色回放，不同 direction 用不同颜色（青色=CLIENT→CORE，紫色=CORE→LLM 等）。 出处：S3-Q10

---

D11. IPC push 埋点只记摘要（去重避免体积翻倍）

原来怎么理解的：IPC 层 push 埋点为什么只记摘要不记完整内容？ 正确解释：因为完整内容已经在 EventBus 的 CORE 事件记录里了。同一个事件在 trace 里出现两次：

如果 push 也记完整内容，假设 3 个客户端订阅了同一个事件，同一个事件会在 trace 文件里出现 4 次，体积翻倍，且重复内容对排查问题毫无帮助。 代码示例：

# app.py — CORE 事件记录（完整）
trace.emit(layer="event", direction="CORE",
           data=event.model_dump())  # ← 完整事件 body

# ipc_broadcaster.py — IPC push 记录（摘要）
trace.emit(layer="ipc", direction="CORE→CLIENT", kind="push",
           data={"sub_id": sub.sub_id, "event_type": event_type})  # ← 只摘要

延伸：能排查什么问题？客户端说"我没收到事件" → 过滤 push 记录看有没有推送；某个事件推给了谁 → 过滤 kind="push" 看 sub_id 列表；推送延迟 → 对比 CORE 事件时间和 push 时间。这是"关注点分离"在日志设计上的体现。 出处：S3-Q11

---

D12. TracingProvider include_payload 开关（隐私+文件大小）

原来怎么理解的：TracingProvider 的 include_payload 字段是干什么的？在哪配置？ 正确解释：控制 trace 记录里写不写 LLM 的完整输入输出内容。

• True：写完整 messages 和 response
• False：只写摘要（消息数、token 用量、延迟）

为什么需要开关？ 隐私/安全（敏感信息不想写进文件）和文件大小（完整记录会让 daemon.jsonl 膨胀很快）。配置在 ~/.mini/config.toml： 代码示例：

# ~/.mini/config.toml
[trace]
enabled = true
file = "~/.mini/traces/daemon.jsonl"
include_llm_payload = true  # ← 这个开关

# 读取链路
# config.toml → TraceConfig.include_llm_payload → TracingProvider(include_payload=...)

# 环境变量覆盖（优先级最高）
# MINI_TRACE_INCLUDE_LLM_PAYLOAD=false

# 优先级：默认值(True) < config.toml < 环境变量

延伸：两个参数的作用——trace: TraceWriter 决定往哪写，include_payload: bool 决定写多细。Java 类比：Log4j 的 DEBUG 级别（完整请求体）vs INFO 级别（只打印摘要）。 出处：S3-Q7, S3-Q8

---

E. Task 系统（S3）

E1. Task 系统从"用户操控"变"Agent 认知工具"

原来怎么理解的：S3 任务系统的"建议性依赖"和"自主规划"怎么理解？ 正确解释：S3 的核心变化：任务系统从"用户通过命令操控"变成"Agent 自己脑子里的认知工具"。任务不是给用户操作的命令，而是 LLM 自己用来管理待办清单的工具。LLM 调 task_create 工具就像人往 checklist 写一条。 代码示例：

# S3 的 task 是 Agent 的"脑内待办清单"
# LLM 决定调用 task_create 工具
# tool_call: {name: "task_create", input: {subject: "分析目录", blocked_by: []}}
#                       ↑ LLM 自己决定的参数，不是用户传的

# TaskManager.create() 创建任务，写入文件
# 返回 ToolResult → LLM 继续推理

延伸：你的理解（已修正）——task 系统本质是"外部化认知"：把 LLM 脑子里记不住的待办清单、依赖关系、完成状态写到文件里，需要时读回来，让 LLM 像人类看 checklist 一样按部就班执行复杂任务。重要修正：LLM 不直接操作文件，它调工具，工具调 TaskManager，TaskManager 负责存储。LLM 看到的是工具返回的 JSON 结果，不是文件内容。 出处：S3-Q26

---

E2. blocked_by 建议性依赖（系统不强制检查）

原来怎么理解的：blocked_by 是"建议性依赖"是什么意思？不强制？ 正确解释：blocked_by 是"建议性依赖"：系统不强制检查，LLM 可能跳过依赖。但串行执行下后果不严重——最坏是工具报错，LLM 会自己纠错。并行执行才需要强制检查。 代码示例：

# Task 结构（model.py 实际定义，字段无默认值）
TaskStatus = Literal["pending", "in_progress", "completed"]

@dataclass
class Task:
    id: int
    subject: str
    description: str
    status: TaskStatus            # Literal 类型，不是普通 str
    blocked_by: list[int]
    created_at: str
    updated_at: str
# 注：上面列出全部字段；from_dict 时 status 缺省回退 "pending"，
# blocked_by 缺省回退 []，但 dataclass 字段本身没有默认值

# LLM 看到任务 #1 blocked_by=[2,3]
# 理论上应该先做 2 和 3
# 但如果 LLM 直接 update(1, status="in_progress")
# 系统不会拒绝——blocked_by 只是"建议"
# 最坏情况：任务1的工具报错（因为依赖没准备好）
# LLM 看到错误会自己退回去先做依赖

延伸：为什么不做强制检查？因为 Agent 是"自主"的——强制约束反而限制了 LLM 的灵活性。LLM 可能有理由跳过依赖（比如它已经知道结果）。这是"信任 LLM 智商"的设计哲学，和传统工作流引擎（如 Airflow）的强制 DAG 不同。 出处：S3-Q26

---

E3. TaskManager 极简设计原因（同步、纯文件、不发事件）

原来怎么理解的：TaskManager 为什么这么简单？ 正确解释：TaskManager 极简——同步、纯文件、不发事件。原因是：任务是 Agent 的认知状态（脑内待办清单），不是系统的运行时状态。两种状态不应耦合。如果 Task 改动发事件，会让事件流变得嘈杂，且任务状态变化不是"系统大事"，不需要广播。 代码示例：

# TaskManager 极简——没有 async，没有 bus，纯文件操作
class TaskManager:
    def __init__(self, tasks_dir: Path):
        self._dir = tasks_dir

    def create(self, subject, blocked_by=None) -> Task:
        task = Task(id=self._next_id(), subject=subject, blocked_by=blocked_by or [])
        self._save(task)   # 写文件
        return task
        # ← 没有 await bus.publish(TaskCreatedEvent)
        # ← 没有 async def

    def update(self, task_id, **kwargs) -> Task:
        task = self._load(task_id)
        # 更新字段
        self._save(task)
        return task

延伸：对比 EventBus 事件（RunStarted/ToolCallStarted 等系统级事件会广播），Task 变化是"脑内"的，不需要外界知道。Java 类比：人的思维过程不需要广播，只有行为（说话、做事）才需要被感知。这种"认知状态 vs 运行时状态"的区分是 Agent 设计的关键洞察。 出处：S3-Q26

---

E4. TaskCreateTool 参数由 LLM 决定（不是用户传的）

原来怎么理解的：TaskCreateTool 什么时候被调用？参数谁传的？ 正确解释：LLM（Claude）决定调用，参数是 LLM 推理出来的。参数不是用户传的，也不是代码写死的，是 LLM 自己决定的。LLM 看到工具 schema（subject、description、blocked_by），根据当前对话上下文决定传什么参数。 代码示例：

# 调用链
# ① LLM 决定要创建任务 → 返回 tool_call
#    {name: "task_create", input: {subject: "分析目录", blocked_by: []}}
#                       ↑ LLM 推理出来的，不是用户传的

# ② AgentLoop 检测到 stop_reason == "tool_use"
# ③ invoke_tool() → TaskCreateTool.invoke(params)
# ④ TaskManager.create() → 创建任务，写入文件
# ⑤ 返回 ToolResult → LLM 继续推理

# 工具 schema 告诉 LLM 有哪些参数
class TaskCreateTool(BaseTool):
    name = "task_create"
    input_schema = {
        "type": "object",
        "properties": {
            "subject": {"type": "string"},
            "blocked_by": {"type": "array", "items": {"type": "integer"}}
        },
        "required": ["subject"]
    }

延伸：这是 Agent "自主性"的体现——用户只给 goal（"帮我分析项目"），LLM 自己规划任务、决定参数、调用工具。Java 类比：像给员工一个目标，员工自己拆解任务清单。用户不指挥具体步骤，只看结果。 出处：S3-Q18

---

E5. add_blocked_by 动态追加依赖（渐进式规划）

原来怎么理解的：add_blocked_by 是什么？能给已有任务追加前置依赖？ 正确解释：对，就是给已有任务动态追加前置依赖。LLM 的规划不是一次性的——执行中可能发现新依赖，需要动态追加。这模拟了人类项目经理的渐进式规划。 代码示例：

# TaskUpdateTool 里
if add_blocked_by:
    task.blocked_by = list(set(task.blocked_by + add_blocked_by))
    #                ↑       ↑      ↑
    #                转回list  去重   拼接新旧

# 场景
# 时刻1: 创建A，blocked_by=[B, C]   → A依赖B和C
# 时刻2: 执行中发现还需要D
# 时刻3: update(A, add_blocked_by=[D]) → A现在依赖B、C、D

延伸：为什么用 set() 去重？防止 LLM 重复加同一个依赖。为什么 list() 转回？因为 blocked_by 字段是 list 类型。对应还有 remove_blocked_by：允许 LLM 主动取消依赖关系。这体现了 Agent 系统的"动态规划"能力——不像传统 DAG 一次画死，可以执行中调整。 出处：S3-Q19

---

E6. 多 Run 共享 EventBus/TraceWriter 的安全性

原来怎么理解的："共享的只有 EventBus 和 TraceWriter"什么意思？多 Run 并发安全吗？ 正确解释：同一个对象引用传给所有 Run。第1次调用和第2次调用拿到的是同一个 self._bus 和 self._trace。为什么能安全共享？

• EventBus：subscribe/publish 都是往列表里加/遍历回调，多个 Run 发布事件互不干扰
• TraceWriter：emit() 只做 put_nowait 往同一个队列放记录，队列是协程安全的

代码示例：

# app.py — 同一个 bus 和 trace 传给所有 Run
# runner 不在 CoreApp 里直接构造，而是由 SessionManager 内部创建
# （SessionManager.send_message 内部 new AgentRunner(self._config, bus=self._bus, trace=self._trace)）

# 多 Run 并发（S3 支持）
self._running_runs: set[asyncio.Task[Any]] = set()
run_task = asyncio.create_task(self._sessions.send_message(session.id, cmd.goal, run_id=run_id))
self._running_runs.add(run_task)
run_task.add_done_callback(self._running_runs.discard)  # 完成自动清理
return AgentRunResult(run_id=run_id)  # 立即返回，不等 run 跑完

# 为什么安全？
# 1. asyncio 单线程，没有 CPU 竞争
# 2. EventBus 的 list.append / for 遍历是原子操作
# 3. asyncio.Queue.put_nowait 是协程安全的
# 4. 每个 Run 有自己的 run_id，事件不会混淆

延伸：Java 类比——多个线程共享同一个 Logger 和 ExecutorService。关键前提是 asyncio 单线程模型——所有协程在同一个线程里，通过 await 让出控制权，不存在真正的并行，所以共享对象不需要锁。如果改成多线程或多进程，就得加锁了。 出处：S3-Q4

---

复习自检

• A1: ReAct 循环的 Plan/Act/Observe 三段，Observe 到底是哪一步？add_assistant_message 算 Observe 吗？
• A2: tool_use block 的四个字段是什么？id 字段的作用？
• A3: content 有几种 block 类型？为什么用块数组而不是字符串？
• A4: invoke_tool 的四个参数分别干什么？
• A5: run_id 在哪里生成？全程变不变？用来干什么？
• A6: 原生 tool_use 比文本描述工具好在哪？至少说出 3 点。
• A7: LLM 怎么自我修复工具错误？错误信息塞到哪里？
• B1: CancelledError 和普通 Exception 处理方式有什么不同？为什么？
• B2: Python 的 CancelledError 和 Java 的 InterruptedException 处理上有什么区别？
• B3: 错误分级的分界线是什么？
• C1: EventBus 是进程内还是跨进程？和 Kafka 区别？
• C2: await handler(event) 本质是什么？
• C3: Tool 和 EventHandler 的区别？用扳手和摄像头类比。
• C4: provider = self._provider or AnthropicProvider(...) 是什么模式？
• C5: extra_handlers 两次 or 各防什么？
• C6: 可变默认参数共享大坑是什么？怎么避免？
• C7: Protocol 和 ABC 的区别？
• C8: TracingProvider 是什么模式？include_payload 干什么？
• C9: run() 和 run_and_capture() 为什么分两个？遵循什么原则？
• C10: S3 用了哪三个 GoF 模式？
• D1: 消息的 4 层嵌套是什么？
• D2: 消息有几种 role？
• D3: tool_result 为什么用 user role？
• D4: content 什么时候是字符串什么时候是 list？
• D5: stream.text_stream 是什么？逐 token 干什么？
• D6: SDK 封装了哪些底层细节？
• D7: cache_control 标在最后一个 tool 上，怎么缓存整个列表？
• D8: list() 浅拷贝有什么坑？怎么解决？
• D9: EventPushEnvelope 的 kind 字段当前固定什么值？event 字段是 Event 对象还是 dict？
• D10: TraceRecord 的 direction + layer 组合表示什么？
• D11: IPC push 埋点为什么只记摘要？
• D12: include_payload 在哪配置？优先级链路？
• E1: Task 系统从"用户操控"变成什么？
• E2: blocked_by 为什么是"建议性"的？
• E3: TaskManager 为什么极简（同步、不发事件）？
• E4: TaskCreateTool 的参数谁决定？
• E5: add_blocked_by 怎么去重？为什么需要动态追加？
• E6: 多 Run 共享 EventBus/TraceWriter 为什么安全？

易错点总结

• A1 易错：把 add_assistant_message（LLM 输出落账）当成 Observe。正确：Observe 是 add_tool_result（工具结果落账）。
• B1 易错：以为 CancelledError 和 Exception 都是 break 退出循环。正确：CancelledError 是 raise 向外抛，Exception 才是 break。
• B2 易错：以为 Python 也要像 Java 那样重设中断标志。正确：Python 基于异常对象，直接 raise 即可。
• C2 易错：以为 await handler(event) 有网络/队列/broker。正确：就是普通函数调用。
• C3 易错：把 EventHandler 当成 agent 工具。正确：Tool 是 LLM 主动调的，EventHandler 是系统被动触发的，LLM 不知道 EventHandler 存在。
• C6 易错：field 不是"字段工厂"，是字段配置函数；default_factory 才是工厂。
• D3 易错：以为 tool_result 有独立 role（如 "tool"）。正确：Anthropic 用 user role 装 tool_result。
• D7 易错：以为 cache_control 标记"往后缓存"。正确：是"从开头到标记位置（含）全部缓存"。
• D8 易错：以为 list() 是深拷贝。正确：list() 是浅拷贝，内层 dict 共享，要改必须单独 dict() 拷贝。
• E1 易错：以为 LLM 直接操作文件。正确：LLM 调工具，工具调 TaskManager，TaskManager 存文件。LLM 看到的是工具返回的 JSON。
• E2 易错：以为 blocked_by 强制检查。正确：建议性依赖，系统不拦，靠 LLM 自纠。
• E6 易错：以为多 Run 共享需要加锁。正确：asyncio 单线程，共享对象天然安全（前提是不用多线程）。

未覆盖说明

本篇所有必含盲点均已在 学习记录.md 中找到对应问答并整理，无遗漏。S2 阶段的部分问答（如 SocketClient、ContextVar、Future 机制等）属于 IPC/网络层盲区，不在本篇"Agent 架构与设计模式"主题范围内，将在其他篇章梳理。