OpenHarness源码研究-4-AgentLoop对话引擎与工具系统

OpenHarness源码研究-4-AgentLoop对话引擎与工具系统从debug说起第2篇我们写了oh -p 你是谁的 debug 配置。断点打在run_print_mode()里跟着调用栈一路往下走能看到这样一条链路run_print_mode() ui/app.py └─ build_runtime(...) ui/runtime.py └─ _resolve_api_client_from_settings() └─ QueryEngine(...) └─ start_runtime(bundle) └─ handle_line(bundle, 你是谁, ...) ui/runtime.py └─ engine.submit_message(你是谁) engine/query_engine.py └─ run_query(context, messages) engine/query.py第3篇分析了 API Client 层_resolve_api_client_from_settings那段这篇讲剩下的消息模型、Agent Loop、工具系统、Runtime 组装、System Prompt。消息模型不同 LLM 的 API 格式不一样但引擎内部需要一套统一的消息表示。这就是engine/messages.py的职责。# engine/messages.py 第14-61行 class TextBlock(BaseModel): type: Literal[text] text text: str class ImageBlock(BaseModel): type: Literal[image] image media_type: str data: str # base64编码 class ToolUseBlock(BaseModel): type: Literal[tool_use] tool_use id: str # 工具调用的唯一ID name: str # 工具名 input: dict[str, Any] # 参数 class ToolResultBlock(BaseModel): type: Literal[tool_result] tool_result tool_use_id: str # 关联到哪个tool_use content: str # 执行结果文本 is_error: bool False # 是否执行出错 ContentBlock TextBlock | ImageBlock | ToolUseBlock | ToolResultBlock class ConversationMessage(BaseModel): role: Literal[user, assistant] content: list[ContentBlock]四个 ContentBlock 覆盖了对话中的所有信息类型。role 只有 user 和 assistant 两种——system prompt 不在这里它作为独立参数传给 API。这套模型是 Anthropic Messages API 的原生格式。OpenAICompatibleClient 把它翻译成 OpenAI 格式第3篇分析过AnthropicApiClient 直接序列化# engine/messages.py 第92-97行 def to_api_param(self) - dict[str, Any]: return { role: self.role, content: [serialize_content_block(block) for block in self.content], }消息模型决定了整个引擎的设计。如果把 system prompt 也当作一条消息、把 tool 也当作一种 role那代码复杂度会成倍增加——OpenAI 就是这么做的而 Anthropic 的设计更简洁。OpenHarness 选择了 Anthropic 格式作为内部规范翻译工作全部丢给 OpenAICompatibleClient。AgentLoopwhile循环tool_use检测engine/query.py的run_query()是整个项目的核心。它做的事说起来简单用户发一句话 → 模型回复可能带 tool_use → 有 tool_use执行工具把结果喂回去 → 模型再回复 → 还有 tool_use继续循环 → 没有了结束用代码表达# engine/query.py 第88-143行简化 turn_count 0 while context.max_turns is None or turn_count context.max_turns: turn_count 1 # 上下文压缩检查 messages, was_compacted await auto_compact_if_needed(...) # 调模型 async for event in context.api_client.stream_message(request): if isinstance(event, ApiTextDeltaEvent): yield AssistantTextDelta(textevent.text) # 逐字流式输出 elif isinstance(event, ApiMessageCompleteEvent): final_message event.message # 模型说完了 messages.append(final_message) # 没有工具调用 → 结束 if not final_message.tool_uses: return # 有工具调用 → 执行 if len(tool_calls) 1: result await _execute_tool_call(...) # 单工具顺序执行 tool_results [result] else: results await asyncio.gather(*[_run(tc) for tc in tool_calls]) # 多工具并发 tool_results list(results) # 把工具结果作为新的user消息追加 messages.append(ConversationMessage(roleuser, contenttool_results)) # → 继续循环几个关键设计决策1. 单工具顺序 vs 多工具并发的选择如果模型一次返回了多个 tool_use比如同时读3个文件用asyncio.gather并发执行。只返回一个时走顺序路径。这个分支不是无谓的优化——并发时需要等所有工具都完成才能通知 UI 更新顺序时则不需要等。两种场景的 UI 事件发送时序不同。2. max_turns 的硬截断# engine/query.py 第42-45行 class MaxTurnsExceeded(RuntimeError): def __init__(self, max_turns: int) - None: super().__init__(fExceeded maximum turn limit ({max_turns}))默认 200 轮。一次turn 模型回复 可能的一次工具调用。200 轮对于大多数任务绰绰有余但如果不设限一个死循环的工具调用就能烧掉大量 token。这个截断是防御性的。3. auto-compact上下文太长怎么办在每轮循环开始时检查# engine/query.py 第91-97行 messages, was_compacted await auto_compact_if_needed( messages, api_clientcontext.api_client, modelcontext.model, system_promptcontext.system_prompt, statecompact_state, )上下文压缩分两步先尝试 microcompact把旧工具结果的 content 清掉换成占位文本成本极低如果还不够再做 full compact调 LLM 对旧消息做摘要成本较高但有意义。阈值是AUTOCOMPACT_BUFFER_TOKENS 13000给模型预留的输出空间。流式事件-引擎怎么通知UI引擎在执行过程中产生 6 种事件# engine/stream_events.py AssistantTextDelta → 模型正在逐字输出这是下一个字 AssistantTurnComplete → 模型说完了附完整消息token用量 ToolExecutionStarted → 开始执行工具告诉UI显示了 ToolExecutionCompleted → 工具执行完毕附输出 ErrorEvent → 出错了 StatusEvent → 临时状态通知重试中之类UI 层只消费这些事件不关心事件是怎么产生的。run_print_mode()和run_repl()处理同一套事件只是渲染方式不同。这个设计和前端框架里的状态管理是一个思路——引擎是 store事件是 actionUI 是 view。工具系统-AI操控电脑的接口tools/base.py定义了工具的契约class BaseTool(ABC): name: str description: str input_model: type[BaseModel] # Pydantic模型自动生成JSON Schema async def execute(self, arguments, context) - ToolResult: 执行工具 def is_read_only(self, arguments) - bool: 是否只读。权限检查用 def to_api_schema(self) - dict: 转为API要求的JSON Schema格式每个工具就是实现这 5 个东西。以文件读取为例# tools/file_read_tool.py class FileReadTool(BaseTool): name read_file description Read a text file from the local repository. input_model FileReadToolInput # path offset limit def is_read_only(self, arguments): return True # 读文件是只读操作 async def execute(self, arguments, context): # 读文件、编号行号、返回is_read_only是关键——它直接关联权限系统。读操作自动放行写操作触发权限检查。Bash 工具更复杂一些。它的is_read_only默认返回 False没法静态判断一个 shell 命令是不是只读所以总是需要权限确认除非用户在 full_auto 模式下# tools/bash_tool.py 第30-79行 async def execute(self, arguments, context): process await create_shell_subprocess(arguments.command, cwdcwd, ...) stdout, stderr await asyncio.wait_for(process.communicate(), timeout...) # 超时则 kill 进程 # 输出截断到 12000 字符 return ToolResult(outputtext, is_errorprocess.returncode ! 0, ...)AgentTool 更有意思——工具里调子 Agent# tools/agent_tool.py 第43-94行 async def execute(self, arguments, context): executor registry.get_executor(subprocess) config TeammateSpawnConfig( namearguments.subagent_type, promptarguments.prompt, cwdstr(context.cwd), ... ) result await executor.spawn(config) return ToolResult(outputfSpawned agent {result.agent_id} ...)工具不再是简单的文件操作而是可以递归地启动新的 Agent 进程。这就从工具调用升级到了多 Agent 协作。Runtime组装-build_runtime的12步ui/runtime.py的build_runtime()把所有零件装到一起# ui/runtime.py 第163-301行 async def build_runtime(cwd, model, ...): # 1. 合并 CLI 覆盖到 settings settings load_settings().merge_cli_overrides(...) # 2. 加载 plugins plugins load_plugins(settings, cwd, ...) # 3. 创建 API Client resolved_api_client _resolve_api_client_from_settings(settings) # 4. 连接 MCP servers mcp_manager McpClientManager(load_mcp_server_configs(...)) await mcp_manager.connect_all() # 5. 创建 ToolRegistry内置工具 MCP 工具 tool_registry create_default_tool_registry(mcp_manager) # 6. 初始化 AppStateUI 状态 app_state AppStateStore(AppState(...)) # 7. 加载 Hooks hook_executor HookExecutor(hook_reloader.current_registry(), ...) # 8. 构建 System Prompt system_prompt_text build_runtime_system_prompt(settings, ...) # 9. 创建 QueryEngine engine QueryEngine(api_client..., tool_registry..., ...) # 10. 恢复历史消息如果有 if restore_messages: engine.load_messages(restored) # 11. 注册 slash 命令 commands create_default_command_registry(...) # 12. 打包成 RuntimeBundle 返回 return RuntimeBundle(api_client..., engine..., ...)这 12 步的依赖关系是单向的settings 在最前面engine 在最后面。RuntimeBundle只是一个 dataclass把所有东西打成一个包。后续handle_line()从这个包里取东西用。handle_line()是交互的核心——它判断用户输入是 slash 命令还是普通对话# ui/runtime.py 第428-567行 async def handle_line(bundle, line, ...): parsed bundle.commands.lookup(line) if parsed is not None: # slash命令 → 走 CommandHandler result await command.handler(args, context) # 可能触发 prompt 提交或 pending continuation else: # 普通对话 → 直接送 engine async for event in bundle.engine.submit_message(line): await render_event(event) # UI层渲染每个事件 bundle.session_backend.save_snapshot(...) # 自动存档SystemPrompt-AI看到的第一段话System Prompt 不是一段写死的文本而是在build_runtime_system_prompt()里动态拼装的# prompts/context.py 第46-120行 def build_runtime_system_prompt(settings, cwd, ...): sections [ build_system_prompt(), # 基础promptBASE_SYSTEM_PROMPT 环境信息 fEffort: {settings.effort}, # 推理设置 ] # Skills列表 skills load_skill_registry(cwd, ...) if skills: sections.append(skills_section) # CLAUDE.md项目指令 claude_md load_claude_md_prompt(cwd) if claude_md: sections.append(claude_md) # Issue/PR上下文如果有 if issue_file.exists(): sections.append(issue_content) # 记忆系统 if settings.memory.enabled: memory_section load_memory_prompt(cwd) sections.append(memory_section) relevant find_relevant_memories(query, cwd) sections.append(relevant_section) return \n\n.join(sections)基础 prompt 本身就包含了环境信息——OS、Shell、日期、Git 分支等由get_environment_info()自动探测。这样 AI 不用问你是什么系统就能直接给出正确的命令。CLAUDE.md是放在项目根目录的一个文件用户在里面写项目约定和偏好。它会被原样注入 System Prompt。这和 GitHub Copilot 的.github/copilot-instructions.md是一个思路。总结Agent Loop 的核心就是一个 while 循环模型说一句 → 有 tool_use 就执行 → 结果喂回去 → 继续单工具顺序执行多工具asyncio.gather并发。分支的原因不是性能是 UI 事件时序不同max_turns200 是防御性的硬截断auto-compact 在每轮循环前检查是否需要压缩上下文BaseTool 的 5 个契约方法构成了工具系统的基础is_read_only直接关联权限RuntimeBundle 是会话的零件包12 步装配顺序是单向依赖System Prompt 是动态拼装的base 环境 skills CLAUDE.md 记忆写到最后