03. Protocol 与多态：用同一套插座更换三类组件

本系列基于 mini-SWE-agent 源码学习整理；源码版本固定为 Asuhe404/mini-swe-agent@388da74，便于长期核对实现。

配套练习：下载 03_protocol_polymorphism.py。

系列导航

返回路线图 · 上一篇：02. 最小装配与入口：从命令找到 Agent · 下一篇：04. Agent 核心循环：让模型回答变成下一轮上下文

适合：零 Python 基础、已有 JavaScript 基础。
重点文件：

• src/minisweagent/__init__.py
• src/minisweagent/models/test_models.py
• src/minisweagent/agents/__init__.py
• src/minisweagent/models/__init__.py
• src/minisweagent/environments/__init__.py

本节目标：能解释 DeterministicModel 为什么没有继承 Model 仍可替换真实模型；能从 YAML 中的类名追到具体实例；能分清静态类型提示与运行时保障。

1. 先用 JavaScript / TypeScript 建立心智模型

先记住一句话：

Protocol 规定“对象要会什么”，具体类负责“怎样做到”，工厂负责“这次选谁”。

1.1 Protocol 最像 TypeScript interface

Python：

class Model(Protocol):
    config: Any

    def query(self, messages: list[dict[str, str]], **kwargs) -> dict: ...
    def format_message(self, **kwargs) -> dict: ...
    def format_observation_messages(self, message: dict, outputs: list[dict], template_vars: dict | None = None) -> list[dict]: ...
    def get_template_vars(self, **kwargs) -> dict[str, Any]: ...
    def serialize(self) -> dict: ...

近似 TypeScript：

interface Model {
  config: unknown;
  query(messages: Message[], options?: object): ModelMessage;
  formatMessage(options: object): Message;
  formatObservationMessages(message: ModelMessage, outputs: Output[], templateVars?: object): Message[];
  getTemplateVars(options?: object): Record<string, unknown>;
  serialize(): Record<string, unknown>;
}

TypeScript 中，一个 class 即使不写 implements Model，只要结构匹配，也能赋给 Model 类型。Python Protocol 同样采用结构类型：具体类不必继承 Protocol。

1.2 Python 类型标注也不会自动做运行时检查

def use_model(model: Model) -> None:
    model.serialize()

这近似 TS 编译后的 JavaScript：运行时只会尝试调用 serialize()。传入错误对象时，Python 不会因为参数标了 Model 就在函数入口拒绝它；通常要等真正调用缺失方法时才抛 AttributeError。

本项目的 Protocol 也没有加 @runtime_checkable，所以：

isinstance(model, Model)

不会返回 True/False，而会抛 TypeError。不要把 Protocol 当成 JS 的 instanceof 或 Java 的运行时 interface。

1.3 三个工厂最像 index.ts + registry + dynamic import()

const registry = {
  default: ["minisweagent.agents.default", "DefaultAgent"],
};

async function resolve(spec: string) {
  const fullPath = registry[spec] ?? spec;
  const [moduleName, exportName] = splitAtLastDot(fullPath);
  const module = await import(moduleName);
  return module[exportName];
}

Python 的三个 __init__.py 就像三个包的 index.ts。它们不是自动扫描目录的插件注册系统，而是：

1. 用 dict 保存常用短名。
2. 把短名变成 module.ClassName。
3. 动态 import 模块并取出 class。
4. 把配置用 **config 交给 class 构造实例。

1.4 三层概念不要混在一起

get_model_class(...) 返回的是类本身；get_model(...) 返回的才是实例。对应 JS 中 DeterministicModel 与 new DeterministicModel(...) 的差别。

2. 5W2H

3. 一张图看清装配与多态

关键方向：Protocol 不负责“制造”对象，工厂也不负责“证明”对象满足 Protocol。它们分别解决静态合同和运行时选择两个问题。

4. 逐行读顶层 minisweagent/__init__.py

真实路径是 src/minisweagent/__init__.py。

第 1-9 行：模块职责声明

docstring 说这个文件同时提供路径、版本和核心 Protocol。第 7-8 行已经点明设计：Protocol 主要服务静态类型检查；正常使用依赖结构类型和鸭子类型。

第 11 行：版本号

__version__ = "2.4.5"

双下划线并不代表真正私有。打包配置会读取该属性，启动横幅也会显示它。

第 13-21 行：两组不同职责的 import

• Any、Protocol 只服务类型合同。
• os、Path、dotenv、platformdirs、Console、logger 服务包初始化。

因此这个文件不是纯类型声明文件；只要 import minisweagent，下面的启动副作用也会发生。

第 23 行：计算安装包目录

package_dir = Path(__file__).resolve().parent

__file__ 是当前模块文件路径，近似 Node ESM 的 import.meta.url；resolve().parent 得到 src/minisweagent 或安装后的包目录。

第 26-28 行：确定并创建全局配置目录

第 26 行优先使用 MSWEA_GLOBAL_CONFIG_DIR，否则使用当前平台的用户配置目录。a or b 对应 JS 的 a || b，这里空字符串也会落到后者。

第 27 行在 import 时就执行 mkdir(parents=True, exist_ok=True)；即使设置了静默启动，也不会跳过建目录。第 28 行得到目录里的 .env 路径。

第 30-36 行：启动输出与 dotenv

• 没有 MSWEA_SILENT_STARTUP 时，Rich 打印版本和配置路径。
• 第 36 行始终尝试加载全局 .env。
• 静默变量只控制打印，不等于“无初始化副作用”。

第 39-40 行：Protocol 区域标记

注释再次强调：不做静态类型检查时可以忽略这些声明，但调用方依赖的方法形状仍然真实存在。

第 43-59 行：Model 合同

class Model(Protocol):

Protocol 是结构化接口基类。这里没有 @runtime_checkable，也没有抽象方法强制。

语法桥：

• self 对应 this。
• list[dict[str, str]] 对应 Array<Record<string, string>>。
• **kwargs 收集额外具名参数，最接近一个开放的 options object。
• dict | None = None 对应 Record<string, unknown> | null，默认 null。
• 行末 ... 是 Ellipsis 占位，表示这里只声明形状；不是 throw new Error()。

第 61-70 行：Environment 合同

Agent 不关心 action 最终由本机 shell、Docker、Singularity 还是远端服务执行，只关心 execute() 返回约定形状的 dict，或抛出约定的流程异常。

第 73-80 行：Agent 合同

path: Path | None 允许值为 None，但这里没有写 = None，所以调用时仍必须传这个位置参数。这是“可空”与“可省略”的区别。

第 83-92 行：公开名称列表

__all__ 只控制 from minisweagent import * 会导入哪些名字，近似集中列出 public exports。它不是访问控制：没有列入的名字仍可被显式 import。

5. DeterministicModel：不继承也能替换的证据

src/minisweagent/models/test_models.py 不是 pytest 测试文件。它是随源码发布的离线模型替身；真正测试它的是 tests/models/test_test_models.py。

第 1-13 行：依赖

• logging、time 用于模拟警告、等待和时间戳。
• Any 用于模板变量返回类型。
• Pydantic BaseModel 验证配置。
• 三个 formatter 分别服务文本消息、Chat Completions tool call 和 Responses API tool call。
• GLOBAL_MODEL_STATS 让离线模型也走与真实模型相同的全局计数路径。

第 16-28 行：制造普通固定响应

make_output(content, actions, cost=1.0) 返回：

{
    "role": "assistant",
    "content": content,
    "extra": {
        "actions": actions,
        "cost": cost,
        "timestamp": time.time(),
    },
}

它像 JS test fixture builder。content 是展示文本，Agent 真正执行的是 extra.actions，两者不会在这里互相解析或校验。

第 31-44 行：制造 Chat Completions tool-call 响应

它在普通响应上增加顶层 tool_calls，并允许 content 为 None。actions 是项目解析后的统一动作；tool_calls 是供应商原始消息格式。二者同时保留，便于 Agent 执行和 trajectory 还原。

第 47-72 行：制造 Responses API 响应

第 54 行先建 output_items = []。第 55-58 行有文本才追加 type="message"；第 59-67 行把每个 action 转成 type="function_call"；第 68-72 行返回完整 response。

这里 for action in actions 对应 actions.forEach(...)。注意第 65 行手工拼 JSON arguments；命令中若含引号、反斜杠或换行，可能产生无效 JSON，这是后面的维护风险。

第 75-87 行：测试专用控制 action

_process_test_actions() 扫描 actions：

• 有 {"raise": exception} 就直接抛该异常。
• command 以 /sleep 开头就等待指定秒数并要求 query() 取下一条响应。
• command 以 /warning 开头就记录 warning，再取下一条响应。
• 都没有则返回 False。

前导下划线 _ 只是“内部使用”的命名约定，不是真私有。

第 90-101 行：普通固定模型的配置 schema

DeterministicModelConfig(BaseModel) 近似 Zod schema + 已解析配置对象：

• outputs 必填，保存将按顺序返回的消息。
• model_name 默认 deterministic。
• cost_per_call 控制全局模拟成本。
• observation_template 是 Jinja 模板，把 return code、异常和 stdout 格式化给模型。
• multimodal_regex="" 表示默认关闭多模态展开。

括号中相邻的多个字符串字面量会自动拼接，不需要 +。

第 104-108 行：构造 DeterministicModel

class DeterministicModel:

类名后没有 (Model)，这就是结构类型的直接证据。

第 107 行把 **kwargs 交给 Pydantic 校验；第 108 行从 -1 开始索引，因为第一次查询会先 += 1，变成列表第一个索引 0。

第 110-116 行：按顺序返回固定响应

1. current_index += 1。
2. 取 outputs[current_index]，这是 O(1)。
3. 若响应含 sleep/warning 控制 action，就递归调用自己读取下一条。
4. 记录一次全局模拟调用成本。
5. 返回预置 dict。

messages 参数故意没有参与计算，所以同一 outputs 总会产生同一顺序，测试因此可复现。列表用完后没有自定义错误，下一次调用会抛 IndexError。

还有两个容易混淆的 cost：make_output(..., cost=...) 写进消息，供 Agent 统计；cost_per_call 交给 GLOBAL_MODEL_STATS。练习里两者都设为 0 才能明确表达“零模拟成本”。

第 118-143 行：补齐剩余 Model 形状

• format_message() 把 kwargs 交给多模态展开工具；普通文本时近似原样返回 dict。
• format_observation_messages() 用文本 observation formatter 处理环境 outputs。
• get_template_vars() 用 model_dump() 把 Pydantic config 转成 dict。
• serialize() 输出模型 config 和完整类路径，例如 minisweagent.models.test_models.DeterministicModel。

现在逐项对照 Model Protocol：config 加五个方法全部存在，所以 DefaultAgent 可以使用它；不需要继承关系。

第 146-201 行：Chat Completions tool-call 变体

DeterministicToolcallModelConfig 与普通配置形状接近。DeterministicToolcallModel 的索引、query、模板变量和序列化逻辑也相同；真正不同的是第 177-188 行：它从原模型消息取出 actions，再生成带 role="tool" 和 tool_call_id 的 observation。

同一个 DefaultAgent 不需要知道这种差异，因为差异被封装在 Model 的 format_observation_messages() 里。这就是多态带来的价值。

第 204-269 行：Responses API 变体

这个版本仍满足同一 Protocol，但 provider 消息外形不同：

• format_message() 把字符串 content 包成 [{"type": "input_text", "text": ...}]。
• 有 extra 才加入消息，if extra 会把空 dict 也视为“不加入”。
• observation 变成 type="function_call_output"。
• query、模板变量、序列化仍维持同一公开方法名。

三个类的内部 wire format 不同，公开形状相同，因此 Agent 可以在它们之间多态分派。

6. 逐行读 Agent 工厂 agents/__init__.py

第 1-6 行：包出口与依赖

这个 __init__.py 类似 agents/index.ts。copy.deepcopy 最接近 structuredClone；importlib 提供动态 import；三个 Protocol 只用于类型标注。

第 8-11 行：短名表

_AGENT_MAPPING = {
    "default": "minisweagent.agents.default.DefaultAgent",
    "interactive": "minisweagent.agents.interactive.InteractiveAgent",
}

下划线说明它是包内实现细节。这里只注册便捷 alias；用户仍可直接提供任意完整类路径。

第 14-22 行：字符串解析成 class

• -> type[Agent] 表示返回“满足 Agent 形状的 class object”，不是 Agent 实例。
• 第 15 行短名命中就取完整路径；未命中就把原字符串当完整路径。
• rsplit(".", 1) 只从最右边拆一次，得到模块名和类名。
• import_module(module_name) 动态加载模块。
• getattr(module, class_name) 对应 JS 的 module[className]。
• 格式错误、模块不存在、类名不存在被统一转换成带可用映射的 ValueError。

这里的返回类型标注不会验证 getattr() 取出的东西真是 class，也不会验证它满足 Agent Protocol。

第 25-28 行：class 构造成实例

1. 接收已经构造好的 Model、Environment 和 agent config。
2. deepcopy(config)，避免后续 pop 修改调用者原 dict 或共享的嵌套数据。
3. pop("agent_class", default_type) 同时读出选择器并从副本删除它。
4. 解析 class，调用 agent_class(model, env, **config)。

因此 agent_class 不会被误传给 DefaultAgent 的配置 schema。配置值优先于 default_type；两者都为空时会尝试解析空字符串并报 Unknown agent type。

主 CLI 在 run/mini.py:101 明确提供 default_type="interactive"。

7. 逐行读 Environment 工厂 environments/__init__.py

第 1-16 行：包出口、依赖与短名

结构与 Agent 工厂相同，映射了 Docker、Singularity、Local、SWE-ReX、Bubblewrap 和 ConTree 等实现。映射值是字符串，而不是启动时直接 import 的 class，因此未选中的可选后端不会仅为注册表而被加载。

第 19-27 行：短名或完整路径变成 class

流程完全同构：mapping fallback -> rsplit -> import_module -> getattr -> 统一 ValueError。

第 30-33 行：构造实例

• 深拷贝 config。
• pop("environment_class", default_type) 取出选择器。
• 解析并用剩余 **config 构造 Environment。

主 CLI 在 run/mini.py:100 给默认值 local；ProgramBench 使用 docker。工厂自己仍没有隐含默认类型。

8. 逐行读 Model 工厂 models/__init__.py

模型工厂比另外两个多了全局用量统计和“模型名”解析。model_name 是供应商模型标识，如某个 GPT/Claude 名称；model_class 是用哪个 Python 适配器。二者不是一回事。

第 1-10 行：依赖

除 deepcopy/importlib 外，还使用 os 读环境变量、threading 保护全局统计，并导入 Model 作为返回类型。

第 13-23 行：初始化全局统计器

构造器把成本和次数设为 0，创建 Lock，并在 import 时读取两个全局限制。只要设置了正限制且未静默，就打印限制。

第 25-39 行：计数、限制与只读 property

• with self._lock 近似进入并自动释放 mutex 的临界区。
• 每次 add(cost) 累加 cost 和调用数。
• 超限时抛 RuntimeError。
• @property 让 stats.cost / stats.n_calls 看起来像字段，内部仍通过方法读取。

第 30 行的调用上限判断使用 _n_calls + 1；限制为 N 时，第 N 次 add() 已完成后便会抛错，实际只有前 N-1 次结果能正常返回。现有该类测试只覆盖启动打印，没有覆盖限制边界。

第 42 行：import 时创建单例

GLOBAL_MODEL_STATS = GlobalModelStats()

所有模型实例共享这个进程级对象。环境变量在这次 import 时读取；之后修改环境变量不会自动刷新已有单例。

第 45-62 行：get_model() 构造实例

1. 第 47 行先解析最终 model_name。
2. config is None 时改为空 dict。
3. 深拷贝 config。
4. 把最终模型名写回副本，保证构造器总能收到它。
5. pop("model_class", "") 后解析 Python 实现类。
6. 模型名含 Anthropic/Claude 关键词且用户未指定时，补默认 cache control。
7. model_class(**config) 构造并返回实例。

显式自定义类若不接受自动补入的 model_name 或 set_cache_control，会在构造时失败。

第 65-75 行：模型名优先级

优先级是：

函数参数 input_model_name
    > config["model_name"]
    > 环境变量 MSWEA_MODEL_NAME
    > ValueError

第 71、73 行的 := 是赋值表达式，近似 JS 在条件里先赋值再判断。只配置 model_class="deterministic" 并不能替代模型名；三处都没有模型名时，代码在选 class 之前就会失败。

第 78-89 行：Model class 短名表

表中既有真实 API adapter，也有：

"deterministic": "minisweagent.models.test_models.DeterministicModel"

因此无 API 测试模型也能走与生产模型相同的工厂装配路径。

第 92-108 行：显式类选择

只有 model_class 非空时才走短名/完整路径解析。注释中的“根据 model_name 选择最佳类”容易让人误会；当前实现并没有按模型名分派不同 Python class。

第 110-113 行：默认类

没有显式 model_class 时，总是懒加载并返回 LitellmModel。这个函数的 model_name 参数在当前分支没有参与判断。

9. 三个工厂放在一起比较

共同点：

• 配置中的选择器优先于调用者传的默认类型。
• 都接受短名或完整 dotted path。
• 都不会在运行时验证 Protocol。
• 都用 deepcopy 防止工厂内部改坏原配置。
• 动态模块首次导入有开销，之后由 Python 缓存。

不同点：Model 的 model_name 与 model_class 分离，而且没有 class 时有 Litellm 默认实现。

10. 这些合同在哪里被真正使用

Model

• agents/default.py:39 接收 model: Model。
• 同文件 56 调 get_template_vars，76/93/94 调 format_message，147 调 query，155 调 format_observation_messages，178 调 serialize。
• models/__init__.py:45 把工厂结果标为 Model。
• models/extra/roulette.py:25,59 把子模型选择结果标为 Model。

Environment

• agents/default.py:39 接收 env: Environment。
• 同文件 55 调 get_template_vars，154 调 execute，178 调 serialize。
• environments/__init__.py:19,30 标注 class 与实例工厂返回值。
• run/benchmarks/swebench.py:79,91 标注并直接执行环境。

ContreeEnvironment 是当前唯一显式写 class ContreeEnvironment(Environment) 的环境；其他环境不继承 Protocol 仍可使用。这说明显式继承不是项目要求。

Agent

• agents/__init__.py:14,25 标注工厂返回的类与实例。
• run/mini.py:101-102 构造后调用 run()。
• run/benchmarks/swebench_single.py:90,96 同样构造并运行。
• batch runner 还会调用 Agent 的 save()。

DeterministicModel

• tests/agents/test_default.py、test_interactive.py 用三种固定模型测试同一 Agent 控制流。
• tests/run/test_local.py、test_run_hello_world.py、test_swebench*.py 替换真实模型，让入口测试不请求 API。
• tests/run/test_save.py 验证它能像真实模型一样序列化进 trajectory。
• tests/models/test_test_models.py 才是专门验证这份源码替身的 pytest 文件。

11. 显式 Protocol 之外还有隐式 dict 协议

Protocol 只声明参数大致是 dict，却没有声明 key。真正跨模块的数据合同还包括：

model message
  role
  content
  extra.actions[]
  extra.cost

action
  command
  tool_call_id?        # tool-call 模型需要

environment output
  output
  returncode
  exception_info

这些 key 缺失时，错误可能在另一个模块很远的位置才出现。TypedDict 能像 TS interface 一样描述 dict key，但三种 provider wire format 不同，若强行统一成一个巨大 schema，反而会削弱这个仓库追求的简单与可替换性。

12. 可维护性与性能审查

高价值风险

1. 工厂不验证导入结果。 getattr() 可能拿到函数或缺方法的 class；错误会延迟到构造或 Agent 调用时。@runtime_checkable 只能浅查成员存在，不能验证完整签名，因此不是完整解决方案。
2. 动态 import 会掩盖模块内部错误。 三个 resolver 把 ImportError 都包装为 Unknown type；目标模块存在但内部依赖缺失时，错误提示可能误导排障。
3. Protocol 与实际消息类型偏离。 list[dict[str, str]] 无法准确表达 content=None、list content、tool calls 和嵌套 extra。可用 TypedDict/union 收紧，但会波及所有模型实现、formatter、Agent 和测试。
4. 已有实现暴露出合同漂移。 models/extra/roulette.py 的 RouletteModel 有 query/get_template_vars/serialize，却没有 Protocol 要求的 format_message/format_observation_messages；直接交给 DefaultAgent 会触发 AttributeError。修复不能只随便转发，因为一次 query 选择的子模型必须与该轮消息/observation formatter 一致。
5. Responses 测试 fixture 手拼 JSON。 make_response_api_output() 应考虑用 json.dumps({"command": ...})，否则复杂命令可能生成无效 arguments。修改会影响相关 fixtures 和断言。
6. 全局调用上限有边界疑点。 _n_calls + 1 使限制 N 在第 N 次结果返回前抛错；若修复，要同时定义“允许 N 次”还是“准备第 N+1 次才阻止”，并把检查放在真正 API 调用之前还是之后。

是否要立即抽象重复代码？

三个 resolver 和三个 Deterministic Model 有重复。可以抽公共 helper/base class，但当前每段都很短，且仓库明确奖励最小、直接的实现。只有当共同逻辑继续增长或已发生同步遗漏时，抽象才真正降低维护成本。

性能判断

• deepcopy 是 O(配置大小)，YAML 配置很小，相比模型请求和容器启动可忽略。
• 字符串动态 import 只在首次加载模块时有明显工作，后续命中 sys.modules 缓存。
• DeterministicModel 取下一条输出是 O(1)，内存是 O(预置响应总大小)。
• 连续大量 /sleep 或 /warning 通过递归跳过响应，理论上可能触发递归深度；测试 fixture 很短，现实影响低，循环实现会更稳。
• 顶层 import minisweagent 会建目录、可能打印并读取 dotenv；这比 Protocol 本身更值得注意，但每个进程通常只发生一次。

13. 无 API 小练习

练习脚本已作为本文附件提供。它只动态选择 class 并读取预置响应：不联网、不调用真实模型，也不执行 action 中的 shell 文本。

先不要运行，预测七行输出，再执行：

MSWEA_SILENT_STARTUP=1 PYTHONPATH=src .venv/bin/python \
  exercises/03_protocol_polymorphism.py

代码：

from minisweagent import Model
from minisweagent.agents import get_agent_class
from minisweagent.environments import get_environment_class
from minisweagent.models import get_model, get_model_class
from minisweagent.models.test_models import make_output


def inspect_model(model: Model) -> None:
    user_message = model.format_message(role="user", content="hello")
    answer = model.query([user_message])

    print(f"answer: {answer['content']}")
    print(f"model name: {model.get_template_vars()['model_name']}")
    print(f"config class: {type(model.config).__name__}")
    print(f"inherits Model: {Model in type(model).__mro__}")


print(f"agent alias: {get_agent_class('default').__name__}")
print(f"environment alias: {get_environment_class('local').__name__}")
print(f"model alias: {get_model_class('unused', 'deterministic').__name__}")

offline_model: Model = get_model(
    "offline-lesson",
    {
        "model_class": "deterministic",
        "outputs": [make_output("fixed answer", [{"command": "echo offline"}], cost=0.0)],
        "cost_per_call": 0.0,
    },
)
inspect_model(offline_model)

已验证输出：

agent alias: DefaultAgent
environment alias: LocalEnvironment
model alias: DeterministicModel
answer: fixed answer
model name: offline-lesson
config class: DeterministicModelConfig
inherits Model: False

动手改三次：

1. 把 "deterministic" 改成完整路径 "minisweagent.models.test_models.DeterministicModel"，确认结果不变。
2. 在 inspect_model() 末尾再调用一次 model.query([])，解释为什么只有一条预置 output 时会抛 IndexError。
3. 另写一个没有继承 Model、但提供相同成员的 class；再删掉它的 serialize()，观察“类型标注本身”和“实际方法调用”分别在什么时候暴露问题。

14. 三道检查题（请先回答，不要查答案）

1. DeterministicModel 没有继承 Model，为什么仍能传给 model: Model？此时执行 isinstance(model, Model) 又为什么会抛 TypeError？
2. 配置中的 agent_class: interactive 从字符串变成实例会经历哪四步？为什么 agent_class 本身不会传给 InteractiveAgent 构造器？
3. 若没有函数参数、config 的 model_name 和 MSWEA_MODEL_NAME，只写 model_class: deterministic 能否构造离线模型？代码会在哪个函数、哪一步失败？