OpenHarness源码研究-6-架构全景与设计模式总结
前言
把前5篇的东西串起来，看整个项目靠什么设计模式撑起来的
完整数据流
一次oh -p "帮我改个bug"从头到尾经过的路径：
┌─────────────────────────────────────────────────────────────────────┐ │ CLI 层 │ │ cli.py:main() │ │ 解析参数 → typer.Option 声明的全部 flag │ │ └─ run_print_mode("帮我改个bug") │ └──────────────────────────────┬──────────────────────────────────────┘ │ ┌──────────────────────────────▼──────────────────────────────────────┐ │ Runtime 装配层 │ │ ui/runtime.py:build_runtime() │ │ load_settings() → 四层覆盖合并 │ │ _resolve_api_client_from_settings() → 根据配置选 Client │ │ load_plugins() → 扫描并加载插件 │ │ McpClientManager.connect_all() → 连接全部 MCP Server │ │ create_default_tool_registry() → 内置工具 + MCP 工具注册 │ │ build_runtime_system_prompt() → 动态拼装 System Prompt │ │ QueryEngine(...) → 创建引擎实例 │ │ → 打包为 RuntimeBundle │ └──────────────────────────────┬──────────────────────────────────────┘ │ ┌──────────────────────────────▼──────────────────────────────────────┐ │ 输入分发层 │ │ ui/runtime.py:handle_line() │ │ bundle.commands.lookup(line) │ │ ├─ slash命令 → CommandHandler │ │ └─ 普通对话 → engine.submit_message() │ └──────────────────────────────┬──────────────────────────────────────┘ │ ┌──────────────────────────────▼──────────────────────────────────────┐ │ Agent Loop (核心) │ │ engine/query.py:run_query() │ │ while turn_count < max_turns: │ │ ├─ auto_compact_if_needed() ← 上下文压缩检查 │ │ ├─ api_client.stream_message() ← 调 LLM │ │ │ └─ [AnthropicApiClient | OpenAICompatibleClient | ...] │ │ │ └─ HTTP/SSE → ApiStreamEvent │ │ ├─ yield StreamEvent ← 通知 UI 层 │ │ ├─ IF 无 tool_use → return │ │ └─ IF 有 tool_use → _execute_tool_call() │ │ ├─ hook_executor.execute(PRE_TOOL_USE) ← Hook 拦截 │ │ ├─ permission_checker.evaluate() ← 权限决策链 │ │ ├─ tool.execute() ← 真正执行 │ │ └─ hook_executor.execute(POST_TOOL_USE) ← Hook 审计 │ │ → 工具结果作为新 user 消息 → 继续循环 │ └──────────────────────────────┬──────────────────────────────────────┘ │ ┌──────────────────────────────▼──────────────────────────────────────┐ │ UI 渲染层 │ │ ui/app.py: 消费 StreamEvent │ │ AssistantTextDelta → 逐字打印到终端 │ │ ToolExecutionStarted/Completed → 显示工具状态 │ │ AssistantTurnComplete → 换行 + 存档 │ │ ErrorEvent / StatusEvent → stderr 输出 │ └─────────────────────────────────────────────────────────────────────┘这是一个典型的管道架构——数据单向流动，每层有自己的职责，层与层之间通过定义好的接口通信。
六个核心设计模式
Protocol-策略模式
第3篇分析过的。SupportsStreamingMessages是个 Protocol（结构化子类型），四种 Client 谁也不继承谁，但都能被 QueryEngine 使用。
# 新增供应商的成本 # 如果 API 是 OpenAI 兼容的 → 零代码，只加 ProviderSpec # 如果 API 格式特殊 → 实现 stream_message() 一个方法对比传统的 ABC + Factory 方案：ABC 要求显式继承，Factory 要求显式注册。Protocol 把这两步都省了。
适用场景：你有多种后端实现、它们之间没有共享代码、你不想引入依赖框架。
声明式注册表
api/registry.py的 PROVIDERS 元组是声明式的极致——42个 Provider，每个是一个 dataclass 实例，没有任何函数调用。检测逻辑是独立的三层扫描（key前缀 → URL关键字 → 模型名），不依赖注册顺序。
同样的模式出现在 Hook（hooks/schemas.py的 HookDefinition）、Skill（声明式 markdown）、Plugin（manifest 文件）。
适用场景：你需要管理大量同类配置项、配置需要被不同子系统按不同维度查询。
分层覆盖
Settings 的四层覆盖：cli arg → env var → settings.json → default。
这不是什么新奇模式，但实现上的细节值得注意——merge_cli_overrides()只覆盖非 None 的值，意味着用户可以不传大部分参数，只传需要覆盖的那一个。
适用场景：你的应用有多个配置来源，需要明确的优先级。
决策链
PermissionChecker 的 evaluate() 是一个顺序决策链。每个环节只检查一件事，拦截了就返回，不拦截就往下走。
敏感路径 → 黑名单 → 白名单 → 路径规则 → 命令规则 → 权限模式这种写法比一个巨大的 if-elif-else 函数清晰得多。新增一个检查条件只需要在链中插入一个新步骤，不修改现有逻辑。
适用场景：你需要做多重条件判断，每层条件来源不同、优先级明确。
事件总线
引擎产生的 6 种 StreamEvent 是狭义的事件总线。更广义的——Hook 系统也是事件驱动（4 个 HookEvent → 3 种 Handler）。
两个事件系统的共同点：生产者不关心消费者是谁，消费者不关心事件是谁产生的。这让 UI 层可以从 Textual TUI 换成 React TUI（backend_only模式），引擎层一行代码都不需要改。
适用场景：你需要解耦数据生产和消费、或者同一个数据流有多个消费方。
原子文件写入
Mailbox（swarm/mailbox.py）和记忆系统都用了同样的模式：
# 先写 .tmp，再 rename tmp_path.write_text(payload, encoding="utf-8") os.replace(tmp_path, final_path) # 原子操作os.replace在 POSIX 系统上是原子的——读取方要么看到旧文件，要么看到新文件，绝不会看到写了一半的残缺文件。
Mailbox 还加了一层文件锁（exclusive_file_lock），防止并发写入冲突。这在多 Agent 协作场景下是必须的。
适用场景：多进程/多 Agent 并发读写同一文件系统，需要保证数据一致性。
工程启发
减少依赖，用标准库
OpenHarness 没有引入 langchain、没有用任何 LLM 抽象框架。它的 API 层全部是直接封装原生 SDK 或裸 HTTP。代价是 4 个 Client 类加起来约 1300 行代码，收益是不会被第三方框架的版本升级绑架，也不会有"这个框架不支持某 API 的某个特性"的困境。
对于个人项目而言，引入一个依赖的决策门槛应该很高。尤其是当依赖做的事情你可以用 200 行代码自己写完的时候。
数据类优于字典
整个项目大量使用 dataclass 和 Pydantic BaseModel 做数据传递。从ApiMessageRequest到TeammateSpawnConfig，没有看到裸 dict 在模块间传递。
dict 的问题是：没有类型提示，没有字段校验，拼错了 key 要到运行时才发现。dataclass 的成本几乎为零，但能让 IDE 帮你检查字段名。
不可变数据
ApiMessageRequest、StreamEvent、ResolvedAuth全部是frozen=True。不可变数据消除了数据在传递过程中被意外修改的可能性。这在异步代码中尤其重要——你不会想知道一个协程在 await 期间，另一个协程改了你手上的对象。
延迟导入
cli.py 的函数体内部到处是from openharness.xxx import ...。这是 CLI 工具的常见优化——oh --help不需要加载 anthropic SDK 和所有工具实现。启动时间从可能的好几秒降到毫秒级。
错误分类
API Client 层的错误不是一把抓的Exception，而是分类为AuthenticationFailure、RateLimitFailure、RequestFailure。重试逻辑据此决定是否重试——认证错误不重试（没用），限流错误要等一等再重试，网络错误可以立即重试。
这个思路可以用在任何需要调外部 API 的项目里。哪怕你只分了"可重试"和"不可重试"两类，也比统一重试所有错误要好。
总结
- 数据流是单向管道：CLI → Runtime → Engine → API Client → LLM → Tool → 循环
- 六个核心模式中，Protocol 策略模式和决策链是最有实用价值的两个
- 减少依赖、用 dataclass 代替 dict、不可变数据、延迟导入、错误分类——这些不是创新，但组合在一起构成了工程上的可靠性
- 197 个文件但耦合度低的本质原因是：层间接口清晰（Protocol + 数据类），模块内部高内聚，模块间通过事件解耦
