OpenHarness源码研究-3-codex配置到输出对话

OpenHarness源码研究-3-codex配置到输出对话

契机

第2篇末尾提到，asyncio.run启动了REPL循环。但一次对话是怎么发生的？先要回答三个问题：

1. 用哪个模型？→ Provider Registry 怎么管理42个供应商
2. 怎么跟模型通信？→ 不同API格式的差异如何抹平
3. 凭什么四种后端共用一个引擎？→ Protocol 的策略模式

正好以DeepSeek为例串起来——它的API天然兼容OpenAI格式，而且国内开发者用得最多。

运行DeepSeek例子

# 申请 DeepSeek API Key：platform.deepseek.com # 方式一：直接命令行 oh -p "你是谁" --api-format openai \ --base-url https://api.deepseek.com/v1 \ --model deepseek-chat \ -k "你的DEEPSEEK_API_KEY" # 方式二：创建持久化 Profile oh provider add deepseek \ --label "DeepSeek" \ --provider deepseek \ --api-format openai \ --auth-source openai_api_key \ --model deepseek-chat \ --base-url https://api.deepseek.com/v1 oh provider use deepseek oh -p "用Python写冒泡排序"

ProviderRegistry-模型的通讯录

在api/registry.py中，每个供应商是一个声明式的数据结构：

# api/registry.py 第157-169行 ProviderSpec( name="deepseek", keywords=("deepseek",), env_key="DEEPSEEK_API_KEY", display_name="DeepSeek", backend_type="openai_compat", # ← 最关键：决定了走哪个Client default_base_url="https://api.deepseek.com/v1", detect_by_base_keyword="deepseek", )

backend_type只有三种值，对应三种Client实现：

另外还有一个特殊的CodexApiClient（Codex订阅），它不走backend_type而是直接按provider == "openai_codex"判断。

注册了 ProviderSpec ≠ 实现了专门的Client。DeepSeek 用的就是通用的OpenAICompatibleClient，零额外代码。

自动检测三级优先级：

1. api_key 前缀 → "sk-or-v1-xxx" → OpenRouter 2. base_url 关键字 → "api.deepseek.com" → DeepSeek 3. model名关键字 → "deepseek-chat" → DeepSeek

所以用户不需要手动指定--provider，输个--model deepseek-chat --base-url ...就能自动推断。

统一的Protocol-四种Client的共同契约

整个适配层只有一个接口：

# api/client.py 第79-83行 class SupportsStreamingMessages(Protocol): async def stream_message(self, request: ApiMessageRequest) -> AsyncIterator[ApiStreamEvent]: """Yield streamed events for the request."""

不是抽象类，不是继承——是Protocol（结构化子类型）。任何实现了stream_message这个方法的对象都能被引擎使用。

在build_runtime()中根据配置选择具体实现：

# ui/runtime.py 第113-160行 def _resolve_api_client_from_settings(settings) -> SupportsStreamingMessages: if settings.api_format == "copilot": return CopilotClient(model=copilot_model) if settings.provider == "openai_codex": return CodexApiClient(auth_token=..., base_url=...) if settings.api_format == "openai": return OpenAICompatibleClient(api_key=..., base_url=...) return AnthropicApiClient(api_key=..., base_url=...) # 默认

这是策略模式，但没有继承、没有抽象类、没有注册表。QueryEngine 不关心后端是谁：

# engine/query_engine.py 第21-22行 class QueryEngine: def __init__(self, *, api_client: SupportsStreamingMessages, ...):

为什么不用 ABC？如果用class BaseClient(ABC)，所有 Client 必须显式继承。但 CodexApiClient 用的是 httpx 裸 HTTP，CopilotClient 内部复用 AnthropicClient——强制继承只会制造不必要的耦合。Protocol 是"如果你长得像鸭子，那你就是鸭子"。

输入输出也完全统一：

ApiMessageRequest: model + messages + system_prompt + max_tokens + tools ApiStreamEvent: ApiTextDeltaEvent | ApiMessageCompleteEvent | ApiRetryEvent

QueryEngine 看到的是：stream_message(request) → events。差异全部封装在 Client 内部。

OpenAI兼容客户端-两种API之间的翻译官

这是覆盖范围最广的 Client。引擎内部说的是 Anthropic Messages API 格式，但 DeepSeek/Qwen/GPT 说的是 OpenAI Chat Completions 格式。OpenAICompatibleClient负责翻译。

消息格式转换（api/openai_client.py第78-123行）：

Anthropic（引擎内部）→ OpenAI（发给DeepSeek） system: "你是助手" → {"role": "system", "content": "你是助手"} user: "帮我写代码" → {"role": "user", "content": "帮我写代码"} assistant: tool_use block → {"role": "assistant", "tool_calls": [{ "function": {"name": ..., "arguments": ...}}]} user: tool_result block → {"role": "tool", "tool_call_id": "xxx", "content": "..."} ← 每个结果一条独立消息！

最关键的差异是 tool_result：Anthropic 里它是 user 消息 content 数组中的一个 block，OpenAI 里它是一条独立的role: "tool"消息。搞错了模型会直接忽略工具结果。

Tool Schema 转换：

# Anthropic 格式 {"name": "read_file", "input_schema": {...}} # ↓ 变为 ↓ # OpenAI 格式 {"type": "function", "function": {"name": "read_file", "parameters": {...}}}

input_schema→parameters改名，外加{"type": "function", "function": {...}}包裹。

流式响应的增量拼接— OpenAI 的流式响应是零散的 delta，需要手动累加tool_calls和reasoning_content，不像 Anthropic SDK 已经帮你拼好了。这让_stream_once()从 AnthropicClient 的 40 行膨胀到 110 行。

Thinking 模型兼容 —DeepSeek 有 thinking 模型。OpenAI 兼容客户端对它做了特殊处理：

# api/openai_client.py 第163-168行 reasoning = getattr(msg, "_reasoning", None) if reasoning: openai_msg["reasoning_content"] = reasoning # 回放推理内容 elif tool_uses: openai_msg["reasoning_content"] = "" # 即使为空也必须带这个字段

思考模型在调用工具时，即使没有推理内容也必须返回空reasoning_content，否则 API 拒绝请求。这是踩坑踩出来的。

Token 限制字段兼容：

# api/openai_client.py 第40-53行 _MAX_COMPLETION_TOKEN_MODEL_PREFIXES = ("gpt-5", "o1", "o3", "o4") # GPT-5/o系列 → max_completion_tokens # 其他模型（包括DeepSeek）→ max_tokens

四种Client横向对比

AnthropicApiClient

引擎的消息格式本身就是 Anthropic 格式的，不需要任何转换。但 OAuth 模式有两个额外操作：

# api/client.py 第207-228行 if self._claude_oauth: params["system"] = f"{attribution}\n{params['system']}" # 注入归因头 params["betas"] = claude_oauth_betas() # 开启OAuth beta params["metadata"] = {"user_id": json.dumps({...})} # 设备/会话元数据

绑 Claude Code 订阅时必须的参数——告诉 Anthropic"这是个合法订阅用户"。

还有 token 刷新：每次stream_message前检查 token 是否过期，过期就重建整个AsyncAnthropic实例，避免请求中途失效的竞态条件。

OpenAICompatibleClient

承担最重的翻译工作：消息格式 + Tool Schema + 流式增量拼接 + thinking 模型 + token 字段。覆盖了 40+ 个注册 Provider 中的绝大多数。

CodexApiClient

不依赖任何 SDK，httpx裸 HTTP 直连 ChatGPT 后端。自己解析 JWT 拿 account_id，自己写 SSE 解析器（按行解析data:前缀），自己组装chatgpt-account-id等特殊 header。因为 Codex 用的是/codex/responses端点，不是标准/v1/chat/completions——没有官方 SDK，只能裸调。

CopilotClient

底层复用AnthropicApiClient，只是认证换成 GitHub OAuth 设备码流。内部持有一个配置好的 AnthropicApiClient 实例，把 Copilot token 适配成 Anthropic 格式即可。

重试机制的统一模式

四者重试逻辑殊途同归：最多 3 次、指数退避、429/5xx/网络错误重试、401/403 不重试。AnthropicClient 额外尊重服务端的Retry-After响应头，并给退避时间加 25% 随机抖动——防止大量并发客户端在同一瞬间同时重试（所谓的"thundering herd"问题）。

为什么不用langchain

*如果要新增供应商（比如智谱 GLM）：

1. 在registry.py加一条ProviderSpec
2. API 是 OpenAI 兼容的 →不需要写任何新代码
3. API 格式特殊 → 实现一个stream_message方法即可
   新增行为不修改现有代码，完美符合开闭原则。
   对比 langchain：你必须继承BaseLLM，实现_generate、_stream、_llm_type等一串抽象方法，还附赠了你可能不需要的 prompt 模板和 output parser。
   OpenHarness 的选择：用标准库的 Protocol 替代第三方框架的抽象类。减少依赖、提高透明度、降低调试难度。*

总结

• Provider Registry 以声明式数据结构管理 42 个供应商，三级自动检测，backend_type决定走哪个 Client
• OpenAICompatibleClient 承担了最重的翻译工作，覆盖绝大多数供应商（包括 DeepSeek）
• SupportsStreamingMessagesProtocol 是整个适配层的唯一契约——策略模式 + 鸭子类型，不需要继承
• 四种 Client 共享统一的重试模式（3 次、指数退避、错误分类），各有特殊处理
• 认证刷新只发生在 OAuth 场景，API Key 模式保持简单

写到最后