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关键字 → 模型名不依赖注册顺序。同样的模式出现在 Hookhooks/schemas.py的 HookDefinition、Skill声明式 markdown、Pluginmanifest 文件。适用场景你需要管理大量同类配置项、配置需要被不同子系统按不同维度查询。分层覆盖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 TUIbackend_only模式引擎层一行代码都不需要改。适用场景你需要解耦数据生产和消费、或者同一个数据流有多个消费方。原子文件写入Mailboxswarm/mailbox.py和记忆系统都用了同样的模式# 先写 .tmp再 rename tmp_path.write_text(payload, encodingutf-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全部是frozenTrue。不可变数据消除了数据在传递过程中被意外修改的可能性。这在异步代码中尤其重要——你不会想知道一个协程在 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 数据类模块内部高内聚模块间通过事件解耦写到最后