Transport ABC 架构详解

一句话解释

Transport ABC 就是 Hermes Agent 的"多语言翻译官"——让同一个 AIAgent 能同时跟 Anthropic / OpenAI / AWS Bedrock / Codex 四种不同的 AI 服务商对话，业务代码不用改。

1. 为什么需要 Transport 层？

Hermes Agent 要跟多个 AI 服务商（Provider）打交道，每个 Provider 的请求格式和响应格式都不一样：

如果没有 Transport 层，run_agent.py 里会充满这样的代码：

if api_mode == "anthropic_messages":
    # 30行格式转换
    payload = {"model": model, "system": system, "messages": anthropic_format, ...}
elif api_mode == "chat_completions":
    # 25行格式转换
    payload = {"model": model, "messages": openai_format, ...}
elif api_mode == "bedrock_converse":
    # 40行格式转换
    payload = {...}

每新增一个 Provider，就要到核心代码里加一套 if-elif，重复、难维护、容易出错。

2. Transport 层是如何工作的？

架构图

                      AIAgent（大脑）
                          │
                          ▼
                ┌─────────────────┐
                │ NormalizedResponse │  ← 统一响应格式
                │  content          │
                │  tool_calls       │
                │  finish_reason   │
                │  usage           │
                └────────▲─────────┘
                         │ normalize_response()
           ┌─────────────┼─────────────┐
           │             │             │
      ┌────┴────┐  ┌─────┴────┐  ┌───┴───┐
      │Anthropic │  │ Bedrock  │  │ Codex  │
      │Transport │  │Transport │  │Transport│
      └────┬────┘  └─────┬────┘  └───┬───┘
           │             │             │
      convert_       convert_     convert_
      messages()     messages()   messages()

两个阶段

请求阶段（你 → AI）：把统一格式的消息，翻译成目标 AI 服务商能看懂的语言。

transport = get_transport(api_mode)      # 拿到对应的翻译器
kwargs = transport.build_kwargs(model, messages, tools)  # 格式转换
response = client.messages.create(**kwargs)               # 调用 AI

响应阶段（AI → 你）：把 AI 的原生响应，翻译回统一格式。

result = transport.normalize_response(response)  # 翻译回 NormalizedResponse

上层 AIAgent 只认识 NormalizedResponse，不用管底下用的是哪个 AI 服务商。

3. 核心组件

3.1 ProviderTransport（抽象基类）

文件：agent/transports/base.py

class ProviderTransport(ABC):
    @property
    @abstractmethod
    def api_mode(self) -> str:
        """返回字符串标识，如 'anthropic_messages'"""

    @abstractmethod
    def convert_messages(self, messages, **kwargs) -> Any:
        """把 OpenAI 格式转成 Provider 原生格式"""

    @abstractmethod
    def convert_tools(self, tools) -> Any:
        """把 OpenAI tool schema 转成 Provider 格式"""

    @abstractmethod
    def build_kwargs(self, model, messages, tools, **params) -> Dict:
        """组装完整的 API 调用参数"""

    @abstractmethod
    def normalize_response(self, response) -> NormalizedResponse:
        """把 Provider 响应归一化为统一格式"""

关键约束：Transport 只管"格式转换"，不管 client 构建、streaming、凭证刷新、重试逻辑——那些都在上层 AIAgent 里。

3.2 NormalizedResponse（统一响应）

文件：agent/transports/types.py

@dataclass
class NormalizedResponse:
    content: Optional[str]           # AI 的文本回复
    tool_calls: Optional[List[ToolCall]]  # 工具调用列表
    finish_reason: str               # 结束原因：stop / tool_calls / length
    reasoning: Optional[str] = None  # 思考过程（如有）
    usage: Optional[Usage] = None    # token 用量
    provider_data: Optional[Dict] = None  # Provider 特有数据

所有 Provider 的响应最终都被归一化为这个结构。Provider 特有的元数据（如 Anthropic 的 reasoning_details、Codex 的 codex_reasoning_items）存入 provider_data，仅在需要时由协议感知代码读取。

3.3 注册机制

文件：agent/transports/__init__.py

_REGISTRY: dict = {}

def register_transport(api_mode: str, transport_cls: type) -> None:
    _REGISTRY[api_mode] = transport_cls

def get_transport(api_mode: str):
    if not _REGISTRY:
        _discover_transports()   # 懒加载，发现所有 transport
    return _REGISTRY.get(api_mode)  # 返回 None 表示未注册

_discover_transports() 按需 import 各个 transport 模块，触发模块末尾的 register_transport() 调用，完成注册。

3.4 四个具体实现

4. 实际例子：发一个工具调用请求

你写的业务代码（不变）

messages = [{"role": "user", "content": "查一下北京天气"}]
tools = [{"type": "function", "function": {"name": "get_weather", ...}}]
transport = get_transport("anthropic_messages")
kwargs = transport.build_kwargs(model, messages, tools)
response = client.messages.create(**kwargs)
result = transport.normalize_response(response)
# result 是 NormalizedResponse，业务代码无需感知用的是哪个 Provider

AnthropicTransport 内部做了什么？

请求转换：

• convert_messages：把 OpenAI 的 {"role": "system", "content": "..."} 拆成 (system_prompt, messages) 元组
• convert_tools：把 OpenAI tool schema 转成 Anthropic 的 input_schema 格式
• build_kwargs：组装 model/max_tokens/thinking/tool_choice 等参数

响应转换：

• normalize_response：把 Anthropic 的 content blocks（[{"type": "text", "text": "..."}]）解析为 content 字符串；把 {"type": "tool_use", ...} 解析为 ToolCall 对象

5. 这个设计跟我们有什么关系？

你现在用的是 MiniMax TTS。假设以后：

• 要同时接入 MiniMax + 火山引擎两个 TTS
• 要在不同的 TTS 服务商之间动态切换
• 要换掉 MiniMax 换成别的服务商

有了 Transport 层，你的业务代码不用动——只需要新增或切换一个 Transport，调用方式完全一样。

就像你们的五金门店系统：不管供应商怎么换，送进仓库的货统一贴上内部条码，仓库管理系统只认这个条码，不关心供应商是谁。

6. 现有批评与局限性

Transport ABC 架构并非完美，以下是一些真实存在的批评：

6.1 过度设计（部分成立）

ChatCompletionsTransport.convert_tools() 的实现是：

def convert_tools(self, tools) -> Any:
    return tools  # 恒等映射，什么都没做

为这样的 identity 函数专门建立一层 ABC 抽象，对于单 Provider 场景确实是过度工程。

6.2 类型安全不完整

def convert_messages(self, messages, **kwargs) -> Any:  # 中间全是 Any
def normalize_response(self, response) -> NormalizedResponse:  # 只有出口有类型

中间转换过程全是 Any，真正的类型安全只存在于最终出口。

6.3 注册机制非真正插件化

_discover_transports() 用硬编码 import 链，新增 Provider 必须修改这个函数，否则静默失败返回 None。

6.4 16 个 Provider 实际只用一个 Transport

ChatCompletionsTransport 的注释承认覆盖 16 个 OpenAI 兼容 Provider，但内部充斥着 is_openrouter/is_nvidia 等 flag——ABC 并没有带来真正的运行时多态，if-else 只是从 run_agent.py 移到了 chat_completions.py 里。

7. 总结

一句话总结：Transport ABC = Hermes Agent 的"多语言翻译官"，让同一个 AIAgent 能同时跟多种 AI 服务商对话，业务代码永远写一套，不用管底下换没换 Provider。

相关文件

• agent/transports/base.py — ProviderTransport ABC 定义
• agent/transports/types.py — NormalizedResponse / ToolCall / Usage 定义
• agent/transports/__init__.py — 注册与发现机制
• agent/transports/anthropic.py — Anthropic 实现
• agent/transports/bedrock.py — Bedrock 实现
• agent/transports/chat_completions.py — OpenAI 兼容实现
• agent/transports/codex.py — Codex 实现