这篇文章是 《Tool Call Dispatch:从 Normalize 到 Gateway Adapter 的统一分发设计》 的配套文章。上一篇主要说明 runtime 的执行流程,这一篇聚焦于代码结构中涉及的设计模式与架构模式。
围绕 dispatch_envelope_demo.py 这类 agent runtime / gateway 代码,常见的问题通常包括:
- 这里是否可以归类为
Adapter - 运行时按 route 选择 adapter 是否属于
Strategy TOOL_REGISTRY和ADAPTER_REGISTRY是否属于RegistryNormalizedToolCall是否可以视为CommandGatewaySession更适合归类为普通数据对象,还是状态机
为了避免把经典设计模式、架构模式和一般数据对象混在一起,本文会按层次说明:
- 哪些结构可以比较明确地归入经典设计模式
- 哪些结构更适合归入架构模式或运行时模式
- 哪些对象不宜被直接归类为某种设计模式
一、先给结论:这份代码的主要模式组合#
如果先做一个整体概括,可以把这份代码的结构理解为:
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 设计模式。
例如 AdapterRoute、NormalizedToolCall、GatewayTurn、TraceEvent 这类对象,更适合优先视为 DTO / Value Object / State Snapshot。
二、为什么这份代码适合从模式视角分析#
设计模式分析最怕两种情况:
- 职责边界不清晰
- 同一层里混杂过多不同问题
这份 runtime 的特点是边界相对明确,可以拆成两层:
- 本地 host runtime
- provider gateway runtime
第一层主要处理:
- 单个 tool envelope 的 normalize
- dispatch key 到本地 handler 的查找
- handler 输出到 tool result envelope 的包装
第二层主要处理:
- provider 请求体的双向转换
- provider-native tool calls 的提取
- tool results 回填到下一轮请求历史
- 多轮 loop 的 session 推进
在这种分层较清晰的前提下,模式就比较容易和实现一一对应。
三、Adapter:第二层 gateway 的核心模式#
在这份代码中,最容易识别的一类模式是 Adapter。
3.1 统一接口#
代码先定义了一个统一协议:
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 CompletionsGeminiPassthroughAdapter负责 Gemini 的透传路线
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:
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 层:
handler = active_registry.get(normalized.dispatch_key)
output = handler(**normalized.arguments)python不同的 dispatch_key 会绑定到不同 handler,例如:
functions.exec_commandfunctions.write_stdinsnapshot_before_editrestore_snapshot
因此,这份代码里其实有两层 Strategy:
- 一层选择 adapter
- 一层选择 handler
五、Registry:两张注册表分别负责两层绑定#
Strategy 往往会和 Registry 配合出现,这份代码也是如此。
5.1 本地工具注册表#
第一张表是 TOOL_REGISTRY:
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:
ADAPTER_REGISTRY = build_adapter_registry(REGISTERED_ADAPTERS)python它解决的问题是:
某条 provider route 应该绑定到哪个 adapter。
5.3 为什么这里更适合用 Registry#
这些逻辑当然也可以用条件分支实现,但 Registry 的优点在于:
- 行为选择逻辑被单独抽离
- 新增策略时不一定要改主流程
- 测试时更容易替换绑定关系
- 运行时绑定关系更明确
因此,这里的 Registry 不是普通字典,而是运行时绑定机制的一部分。
六、Command:NormalizedToolCall 可以视为轻量命令对象#
从 Command 的角度看,这里存在一个轻量版本的实现。
6.1 命令对象#
@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() 则承担了调度器的角色:
normalized = normalize_tool_call(call_envelope)
handler = active_registry.get(normalized.dispatch_key)
output = handler(**normalized.arguments)python如果用 Command 的语言来描述:
NormalizedToolCall类似 command objectTOOL_REGISTRY提供 receiver lookupdispatch_tool_call()承担 invoker 的职责
它不是传统教材里那种完整的 OO Command 体系,但语义上与 Command 相当接近。
七、Null Object / Identity:透传也被建模成显式策略#
这份代码没有把”无转换”当作一个漏掉的特殊分支,而是显式建模成 IdentityAdapter:
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 复用:
class GeminiPassthroughAdapter(IdentityAdapter):
...python这里可以从两个角度理解:
- 它接近
Null Object - 也可以理解成
Identity Strategy
核心点不在术语,而在设计选择本身:
“不做转换”也是一个需要被明确表达的策略。
八、Simple Factory:create_gateway_session()#
这份代码里没有复杂的工厂体系,但有一个比较清晰的 Simple Factory:
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()#
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_keyargumentscall_id
9.2 输出方向:build_tool_result_envelope()#
边界不仅有输入方向,也有输出方向:
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:流程被拆成稳定阶段#
这部分更适合归类为 Pipeline 或 Orchestrator,而不是某个特定的 GoF 模式。
10.1 第一层编排:dispatch_tool_call()#
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关键点在于:整个流程被拆成了稳定阶段。
- normalize
- lookup
- execute
- wrap result
- append generic context
- append provider-native context
10.2 第二层编排:GatewaySession.step()#
在 gateway loop 里,这种结构更明显:
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
因此,这里更适合用”编排器”或”流水线”来描述。
十一、Session State / State Machine:GatewaySession#
GatewaySession 在当前版本中承担了明确的状态管理职责:
@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]] = Nonepython它维护的是一组 session state:
- 当前请求
- 审计上下文
- 已执行轮次
- 是否完成
- 停止原因
- 最终响应
step() 则推进一次状态转移:
- 未完成 -> 进入下一次 step
- 有 tool calls -> append back 后进入下一轮
- 无 tool calls -> 标记 completed
从职责上看,这里已经比较接近一个轻量状态机。
十二、Facade:GatewayRuntime#
更高一层是 GatewayRuntime:
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 中有明确作用。
@dataclass
class TraceEvent:
session_id: int
kind: str
turn_index: Optional[int]
payload: Dict[str, Any]python以及:
self._record_trace(
session_id,
kind="turn_started",
turn_index=turn_index,
payload={"request": ...},
)python它记录的不是最终状态快照,而是状态变化事件,例如:
session_createdturn_startedturn_completedturn_failedsession_completedsession_stopped
这类结构的主要价值在于:
- 便于调试
- 便于回放
- 便于审计
- 便于做可视化
如果 runtime 继续演化,这部分也很自然会延伸为更完整的 observability / trace 体系。
十四、哪些结构不宜直接归类为经典设计模式#
在模式分析中,除了识别模式本身,也需要控制归类边界。
14.1 数据对象#
下面这些对象更适合首先视为数据对象:
AdapterRouteNormalizedToolCallGatewayTurnTraceEvent
更准确的说法通常是:
- DTO
- Value Object
- State Snapshot
它们的主要价值在于:
- 明确边界
- 降低隐式耦合
- 提高可读性
14.2 dispatch_tool_call() 不太适合归类为 Template Method#
有时会有人把它归类为 Template Method,但这里并不完全符合该模式的标准定义。
Template Method 的典型形态通常是:
- 父类定义算法骨架
- 子类覆写可变步骤
而这里并没有通过继承机制来覆写步骤,因此更适合用 Pipeline Orchestrator 或 Workflow Coordinator 来描述。
14.3 normalize 与 transform 不应混为一类#
这也是边界上最容易混淆的地方:
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 中的组合方式:
normalize_tool_call()先把异构输入压成统一命令对象TOOL_REGISTRY把命令路由到具体 handlerbuild_tool_result_envelope()把内部结果翻回边界格式ProviderAdapter对整包 request / response 做 provider 间双向转换GatewaySession负责多轮 loop 的状态推进GatewayRuntime再把 session 与 trace 管理收口
从整体上看,这形成的是一套分层明确的 runtime,而不是简单堆叠的转换逻辑。
十七、总结#
如果把问题表述为:
“这份代码中涉及哪些设计模式和架构模式?”
那么一个较准确的回答是:
这份代码通过 Adapter 和 Anti-Corruption Layer 处理边界异构,通过 Registry、Strategy、Command 和 Session State 维持清晰的执行模型。
换一种更中性的表述:
模式在这里不是目标本身,更重要的是用这些结构维持清晰的边界、稳定的扩展点和可追踪的执行流程。