Jerry's Blog

Back

这篇文章是 《Tool Call Dispatch:从 Normalize 到 Gateway Adapter 的统一分发设计》 的配套文章。上一篇主要说明 runtime 的执行流程,这一篇聚焦于代码结构中涉及的设计模式与架构模式。

围绕 dispatch_envelope_demo.py 这类 agent runtime / gateway 代码,常见的问题通常包括:

  • 这里是否可以归类为 Adapter
  • 运行时按 route 选择 adapter 是否属于 Strategy
  • TOOL_REGISTRYADAPTER_REGISTRY 是否属于 Registry
  • NormalizedToolCall 是否可以视为 Command
  • GatewaySession 更适合归类为普通数据对象,还是状态机

为了避免把经典设计模式、架构模式和一般数据对象混在一起,本文会按层次说明:

  1. 哪些结构可以比较明确地归入经典设计模式
  2. 哪些结构更适合归入架构模式或运行时模式
  3. 哪些对象不宜被直接归类为某种设计模式

一、先给结论:这份代码的主要模式组合#

如果先做一个整体概括,可以把这份代码的结构理解为:

Adapter + Strategy + Registry + Command 构成执行骨架,Anti-Corruption Layer + Pipeline + Session State 构成外围架构骨架。

对应关系可以先整理成一张表:

类别模式/结构对应情况
经典设计模式Adapter明确存在,是第二层 gateway 的核心
经典设计模式Strategy存在,主要体现为运行时选择 adapter / handler
经典设计模式Registry存在,而且分成两层
经典设计模式Command存在一个轻量版本
经典设计模式Null Object / Identity存在,用于 passthrough adapter
常见工程模式Simple Factory存在,用于 session 创建
架构模式Anti-Corruption Layer明确存在
架构模式Pipeline / Orchestrator明确存在
架构模式Session State / State Machine存在
架构模式Facade存在
运行时模式Event Log / Trace存在

同时需要补充一点:

并不是所有形式上类似模式的结构,都适合直接归类为 GoF 设计模式。

例如 AdapterRouteNormalizedToolCallGatewayTurnTraceEvent 这类对象,更适合优先视为 DTO / Value Object / State Snapshot

交互演示:模式分类图谱
点击任意模式,查看它在代码中的对应位置

二、为什么这份代码适合从模式视角分析#

设计模式分析最怕两种情况:

  • 职责边界不清晰
  • 同一层里混杂过多不同问题

这份 runtime 的特点是边界相对明确,可以拆成两层:

  1. 本地 host runtime
  2. provider gateway runtime

第一层主要处理:

  • 单个 tool envelope 的 normalize
  • dispatch key 到本地 handler 的查找
  • handler 输出到 tool result envelope 的包装

第二层主要处理:

  • provider 请求体的双向转换
  • provider-native tool calls 的提取
  • tool results 回填到下一轮请求历史
  • 多轮 loop 的 session 推进

在这种分层较清晰的前提下,模式就比较容易和实现一一对应。

交互演示:双层 Runtime 架构
点击每个组件,查看它的职责和涉及的设计模式

三、Adapter:第二层 gateway 的核心模式#

在这份代码中,最容易识别的一类模式是 Adapter

3.1 统一接口#

代码先定义了一个统一协议:

dispatch_envelope_demo.py
class ProviderAdapter(Protocol):
    def name(self) -> str: ...
    def route(self) -> AdapterRoute: ...
    def transform_request(self, body: Mapping[str, Any]) -> Dict[str, Any]: ...
    def transform_response(self, body: Mapping[str, Any]) -> Dict[str, Any]: ...
python

这意味着上层调用方只依赖两件事:

  • transform_request()
  • transform_response()

调用方不需要直接理解 Claude、OpenAI、Gemini 各自不同的请求和响应细节。

3.2 具体 adapter#

具体实现里:

  • ClaudeCodeOpenAIAdapter 负责 Anthropic Messages -> OpenAI Chat Completions
  • GeminiPassthroughAdapter 负责 Gemini 的透传路线
dispatch_envelope_demo.py
class ClaudeCodeOpenAIAdapter:
    def transform_request(...): ...
    def transform_response(...): ...


class GeminiPassthroughAdapter(IdentityAdapter):
    ...
python

从职责上看,这一层的作用就是:

把不同 provider 的不兼容协议,转换成上层统一可消费的接口。

这正是 Adapter 模式的典型场景。


四、Strategy:行为在运行时按 route 或 key 选择#

除了统一接口之外,系统还需要在运行时决定”具体采用哪一种行为”。这部分更适合用 Strategy 来描述。

4.1 adapter 选择#

get_adapter() 根据 route 选择具体 adapter:

dispatch_envelope_demo.py
def get_adapter(
    *,
    downstream_provider: Provider,
    downstream_format: ApiFormat,
    upstream_provider: Provider,
    upstream_format: ApiFormat,
) -> ProviderAdapter:
    ...
python

这里变化的是具体转换策略:

  • Claude -> OpenAI,对应一种策略
  • Gemini -> Gemini,对应另一种策略

4.2 handler 选择#

同样的结构也出现在本地 dispatch 层:

dispatch_envelope_demo.py
handler = active_registry.get(normalized.dispatch_key)
output = handler(**normalized.arguments)
python

不同的 dispatch_key 会绑定到不同 handler,例如:

  • functions.exec_command
  • functions.write_stdin
  • snapshot_before_edit
  • restore_snapshot

因此,这份代码里其实有两层 Strategy:

  1. 一层选择 adapter
  2. 一层选择 handler

五、Registry:两张注册表分别负责两层绑定#

Strategy 往往会和 Registry 配合出现,这份代码也是如此。

5.1 本地工具注册表#

第一张表是 TOOL_REGISTRY

dispatch_envelope_demo.py
TOOL_REGISTRY: Dict[str, Handler] = {
    "functions.exec_command": exec_command,
    "functions.write_stdin": write_stdin,
    "functions.apply_patch": apply_patch,
    ...
}
python

它解决的问题是:

工具名应该绑定到哪个本地函数。

5.2 adapter 注册表#

第二张表是 ADAPTER_REGISTRY

dispatch_envelope_demo.py
ADAPTER_REGISTRY = build_adapter_registry(REGISTERED_ADAPTERS)
python

它解决的问题是:

某条 provider route 应该绑定到哪个 adapter。

5.3 为什么这里更适合用 Registry#

这些逻辑当然也可以用条件分支实现,但 Registry 的优点在于:

  • 行为选择逻辑被单独抽离
  • 新增策略时不一定要改主流程
  • 测试时更容易替换绑定关系
  • 运行时绑定关系更明确

因此,这里的 Registry 不是普通字典,而是运行时绑定机制的一部分。


六、Command:NormalizedToolCall 可以视为轻量命令对象#

Command 的角度看,这里存在一个轻量版本的实现。

6.1 命令对象#

dispatch_envelope_demo.py
@dataclass
class NormalizedToolCall:
    source_format: str
    source_variant: str
    call_id: str
    dispatch_key: str
    arguments: Dict[str, Any]
    raw_envelope: Dict[str, Any]
python

这个对象保存的是一次调用所需的主要信息:

  • 调用目标:dispatch_key
  • 调用参数:arguments
  • 调用标识:call_id
  • 原始输入:raw_envelope

从结构上看,它可以被理解为一个”待执行命令”的载体。

6.2 调度器#

dispatch_tool_call() 则承担了调度器的角色:

dispatch_envelope_demo.py
normalized = normalize_tool_call(call_envelope)
handler = active_registry.get(normalized.dispatch_key)
output = handler(**normalized.arguments)
python

如果用 Command 的语言来描述:

  • NormalizedToolCall 类似 command object
  • TOOL_REGISTRY 提供 receiver lookup
  • dispatch_tool_call() 承担 invoker 的职责

它不是传统教材里那种完整的 OO Command 体系,但语义上与 Command 相当接近。


七、Null Object / Identity:透传也被建模成显式策略#

这份代码没有把”无转换”当作一个漏掉的特殊分支,而是显式建模成 IdentityAdapter

dispatch_envelope_demo.py
class IdentityAdapter:
    def transform_request(self, body: Mapping[str, Any]) -> Dict[str, Any]:
        return dict(body)

    def transform_response(self, body: Mapping[str, Any]) -> Dict[str, Any]:
        return dict(body)
python

然后由 GeminiPassthroughAdapter 复用:

dispatch_envelope_demo.py
class GeminiPassthroughAdapter(IdentityAdapter):
    ...
python

这里可以从两个角度理解:

  • 它接近 Null Object
  • 也可以理解成 Identity Strategy

核心点不在术语,而在设计选择本身:

“不做转换”也是一个需要被明确表达的策略。


八、Simple Factory:create_gateway_session()#

这份代码里没有复杂的工厂体系,但有一个比较清晰的 Simple Factory

dispatch_envelope_demo.py
def create_gateway_session(... ) -> GatewaySession:
    adapter = get_adapter(...)
    return GatewaySession(
        adapter=adapter,
        ...
    )
python

它把以下初始化细节集中起来:

  • route 到 adapter 的选择
  • request 的初始深拷贝
  • registry 注入
  • audit context 初始化
  • turn 列表初始化

这样调用方不需要自己组装完整的 GatewaySession(...),而是通过一个统一入口完成构建。


九、Anti-Corruption Layer:边界上的语义提纯#

如果把 Adapter 视为最明确的经典设计模式,那么 Anti-Corruption Layer 可以视为最明确的架构模式。

9.1 入口方向:normalize_tool_call()#

dispatch_envelope_demo.py
def normalize_tool_call(envelope: Mapping[str, Any]) -> NormalizedToolCall:
    ...
python

它的职责不是执行工具,而是:

  • 吸收 provider-native 结构差异
  • 提炼出内部统一使用的调用语义

例如外部世界可能是:

  • Codex 的 recipient_name + parameters
  • OpenAI Responses 的 name + arguments
  • OpenAI Chat 的 function.name + function.arguments
  • Claude 的 tool_use + input

进入系统内部以后,会被统一成:

  • dispatch_key
  • arguments
  • call_id

9.2 输出方向:build_tool_result_envelope()#

边界不仅有输入方向,也有输出方向:

dispatch_envelope_demo.py
def build_tool_result_envelope(...):
    ...
python

它负责把内部统一的 output: Dict[str, Any] 再包装回:

  • OpenAI role=tool
  • OpenAI function_call_output
  • Claude tool_result
  • Codex tool_result

因此,这里不是单向防腐,而是双向边界转换。


十、Pipeline / Orchestrator:流程被拆成稳定阶段#

这部分更适合归类为 PipelineOrchestrator,而不是某个特定的 GoF 模式。

10.1 第一层编排:dispatch_tool_call()#

dispatch_envelope_demo.py
normalized = normalize_tool_call(call_envelope)
handler = active_registry.get(normalized.dispatch_key)
output = handler(**normalized.arguments)
tool_result_envelope = build_tool_result_envelope(normalized, output)
updated_context = append_tool_interaction_to_context(...)
provider_native_context = build_provider_native_context_entries(...)
python

关键点在于:整个流程被拆成了稳定阶段。

  1. normalize
  2. lookup
  3. execute
  4. wrap result
  5. append generic context
  6. append provider-native context

10.2 第二层编排:GatewaySession.step()#

在 gateway loop 里,这种结构更明显:

dispatch_envelope_demo.py
upstream_request = self.adapter.transform_request(self.current_request)
upstream_response = dict(upstream_responder(...))
downstream_response = self.adapter.transform_response(upstream_response)
tool_calls = extract_tool_calls_from_provider_response(...)
assistant_text = extract_text_from_provider_response(...)
...
next_request = append_tool_results_to_provider_messages(...)
python

这对应的是一个比较清晰的流水线:

  • request transform
  • upstream call
  • response transform
  • extract tool calls
  • local dispatch
  • append back

因此,这里更适合用”编排器”或”流水线”来描述。

交互演示:dispatch_tool_call() 流水线
逐步查看 6 个阶段的执行细节

十一、Session State / State Machine:GatewaySession#

GatewaySession 在当前版本中承担了明确的状态管理职责:

dispatch_envelope_demo.py
@dataclass
class GatewaySession:
    ...
    current_request: Dict[str, Any]
    audit_context: List[Dict[str, Any]]
    turns: List[GatewayTurn]
    completed: bool = False
    stop_reason: Optional[str] = None
    final_response: Optional[Dict[str, Any]] = None
python

它维护的是一组 session state:

  • 当前请求
  • 审计上下文
  • 已执行轮次
  • 是否完成
  • 停止原因
  • 最终响应

step() 则推进一次状态转移:

  • 未完成 -> 进入下一次 step
  • 有 tool calls -> append back 后进入下一轮
  • 无 tool calls -> 标记 completed

从职责上看,这里已经比较接近一个轻量状态机。

交互演示:GatewaySession 状态转移
点击状态节点查看详细说明

十二、Facade:GatewayRuntime#

更高一层是 GatewayRuntime

dispatch_envelope_demo.py
class GatewayRuntime:
    def list_routes(self) -> List[Dict[str, Any]]: ...
    def get_adapter(...) -> ProviderAdapter: ...
    def create_session(...) -> int: ...
    def get_session(self, session_id: int) -> GatewaySession: ...
    def get_trace(self, session_id: int) -> List[Dict[str, Any]]: ...
    def step_session(self, session_id: int) -> Dict[str, Any]: ...
    def run_session(self, session_id: int, *, max_turns: int = 8) -> Dict[str, Any]: ...
python

它本身不处理底层协议细节,而是把多个子系统收口为一组更高层接口:

  • session 生命周期管理
  • adapter route 管理
  • trace 管理
  • step / run 入口管理

如果没有这一层,调用方需要自行组合:

  • get_adapter()
  • GatewaySession(...)
  • _record_trace(...)
  • step()
  • run()

因此,这里比较适合归类为 Facade


十三、Event Log / Trace:运行时事件记录#

TraceEvent_record_trace() 不属于经典设计模式,但在 runtime 中有明确作用。

dispatch_envelope_demo.py
@dataclass
class TraceEvent:
    session_id: int
    kind: str
    turn_index: Optional[int]
    payload: Dict[str, Any]
python

以及:

dispatch_envelope_demo.py
self._record_trace(
    session_id,
    kind="turn_started",
    turn_index=turn_index,
    payload={"request": ...},
)
python

它记录的不是最终状态快照,而是状态变化事件,例如:

  • session_created
  • turn_started
  • turn_completed
  • turn_failed
  • session_completed
  • session_stopped

这类结构的主要价值在于:

  • 便于调试
  • 便于回放
  • 便于审计
  • 便于做可视化

如果 runtime 继续演化,这部分也很自然会延伸为更完整的 observability / trace 体系。


十四、哪些结构不宜直接归类为经典设计模式#

在模式分析中,除了识别模式本身,也需要控制归类边界。

14.1 数据对象#

下面这些对象更适合首先视为数据对象:

  • AdapterRoute
  • NormalizedToolCall
  • GatewayTurn
  • TraceEvent

更准确的说法通常是:

  • DTO
  • Value Object
  • State Snapshot

它们的主要价值在于:

  • 明确边界
  • 降低隐式耦合
  • 提高可读性

14.2 dispatch_tool_call() 不太适合归类为 Template Method#

有时会有人把它归类为 Template Method,但这里并不完全符合该模式的标准定义。

Template Method 的典型形态通常是:

  • 父类定义算法骨架
  • 子类覆写可变步骤

而这里并没有通过继承机制来覆写步骤,因此更适合用 Pipeline OrchestratorWorkflow Coordinator 来描述。

14.3 normalizetransform 不应混为一类#

这也是边界上最容易混淆的地方:

  • normalize_tool_call():处理单个 envelope 的语义提纯
  • transform_request() / transform_response():处理整包 provider body 的协议转换

两者都在做结构改写,但职责并不相同。


十五、从实现角度看最关键的 4 个概念#

如果只保留最关键的 4 个概念,可以优先关注下面这些:

1. Adapter#

因为第二层 gateway 的核心就是 provider protocol adapter。

2. Registry#

因为这份设计的扩展性,主要依赖运行时绑定:

  • tool 通过 registry 绑定
  • adapter 也通过 registry 绑定

3. Command#

因为把 NormalizedToolCall 视为命令对象后,dispatch 层的职责关系会更容易理解。

4. Session State#

因为进入多轮 gateway loop 后,问题已经不只是单次函数调用,而是 session 状态推进。


十六、组合后的整体结构#

更重要的不是单个模式本身,而是这些结构在同一套 runtime 中的组合方式:

  1. normalize_tool_call() 先把异构输入压成统一命令对象
  2. TOOL_REGISTRY 把命令路由到具体 handler
  3. build_tool_result_envelope() 把内部结果翻回边界格式
  4. ProviderAdapter 对整包 request / response 做 provider 间双向转换
  5. GatewaySession 负责多轮 loop 的状态推进
  6. GatewayRuntime 再把 session 与 trace 管理收口

从整体上看,这形成的是一套分层明确的 runtime,而不是简单堆叠的转换逻辑。


十七、总结#

如果把问题表述为:

“这份代码中涉及哪些设计模式和架构模式?”

那么一个较准确的回答是:

这份代码通过 Adapter 和 Anti-Corruption Layer 处理边界异构,通过 Registry、Strategy、Command 和 Session State 维持清晰的执行模型。

换一种更中性的表述:

模式在这里不是目标本身,更重要的是用这些结构维持清晰的边界、稳定的扩展点和可追踪的执行流程。


References#

Tool Gateway Runtime:代码中涉及的设计模式与架构模式
https://jerry609.github.io/blog/tool-gateway-runtime-design-patterns
Author Jerry
Published at March 18, 2026
Comment seems to stuck. Try to refresh?✨