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

系列导航

返回路线图 · 下一篇：02. 最小装配与入口：从命令找到 Agent

适合：零 Python 基础、已有 JavaScript 基础。
本节文件：

  • src/minisweagent/run/hello_world.py
  • src/minisweagent/exceptions.py
  • src/minisweagent/utils/serialize.py

本节目标不是学完 Python，而是能回答：对象怎样装配、配置怎样覆盖、流程信号怎样携带消息。

1. 先看 JavaScript 等价物

1.1 hello_world.py：应用装配入口

Python 原意可以先理解成下面的 JS 伪代码：

import { Command } from "commander";
import YAML from "yaml";
import { readFileSync } from "node:fs";

const app = new Command();

function main({ task, modelName = process.env.MSWEA_MODEL_NAME }) {
  const agent = new DefaultAgent(
    new LitellmModel({ modelName }),
    new LocalEnvironment(),
    YAML.parse(readFileSync("config/default.yaml", "utf8")).agent,
  );
  agent.run(task);
  return agent;
}

if (isDirectlyExecuted(import.meta.url)) {
  app.parse();
}

它是 composition root：选择一个 Model、一个 Environment、一个 Agent，把配置注入，然后运行。

1.2 exceptions.py：带 payload 的流程信号

class InterruptAgentFlow extends Error {
  constructor(...messages) {
    super();
    this.messages = messages;
  }
}

class Submitted extends InterruptAgentFlow {}
class LimitsExceeded extends InterruptAgentFlow {}
class TimeExceeded extends LimitsExceeded {}
class UserInterruption extends InterruptAgentFlow {}
class FormatError extends InterruptAgentFlow {}

这里的异常不全代表程序 bug。它们更像 Redux action 或状态机 event，只是用 throw 快速跳出当前深层调用栈，同时携带要加入对话历史的 message。

1.3 serialize.py：递归版 object spread

const UNSET = Symbol("UNSET");

function recursiveMerge(...objects) {
  const result = {};
  for (const object of objects) {
    if (object == null) continue;
    for (const [key, value] of Object.entries(object)) {
      if (value === UNSET) continue;
      if (isPlainObject(result[key]) && isPlainObject(value)) {
        result[key] = recursiveMerge(result[key], value);
      } else if (isPlainObject(value)) {
        result[key] = recursiveMerge(value);
      } else {
        result[key] = value;
      }
    }
  }
  return result;
}

它与 {...base, ...override} 的区别是：嵌套 object 也会继续合并；array 不合并，后值直接替换前值。

2. 5W2H

问题回答
What一个最小运行入口、一组流程信号、一个配置深合并函数。
Why把“选择组件”“中断循环”“覆盖配置”分离，核心 Agent 不需要知道所有实现细节。
WhoRun Script 装配；Agent 捕获信号；CLI、Agent、Environment 用 recursive_merge。
When启动时装配；运行中完成/超限/格式错误时抛信号；读取 YAML/CLI 参数后合并配置。
Whererun/hello_world.py、exceptions.py、utils/serialize.py。
How依赖注入、异常继承与捕获、从左到右的递归覆盖。
How much三个文件共 97 行；合并复杂度约为 O(访问到的 key 数)，空间约为新建 dict 的大小。

3. 总流程

流程图 1流程图 1

4. 逐行读 run/hello_world.py

第 1-3 行：模块 docstring

"""This is the simplest possible example ..."""

文件开头的三引号字符串是 module docstring，类似模块级 JSDoc。Python 会把它保存为模块的 __doc__。

第 5-7 行：标准库 import

import logging
import os
from pathlib import Path
  • import logging 类似 import * as logging from ...。
  • os.getenv("NAME") 对应 process.env.NAME。
  • from pathlib import Path 对应 named import；Path 是跨平台路径对象。

第 9-10 行：第三方依赖

import typer
import yaml
  • Typer 类似 Commander/Yargs，但它从函数签名和类型注解生成 CLI。
  • yaml.safe_load(text) 把 YAML 字符串解析成 Python dict/list/scalar，类似 YAML.parse(text)。

第 12-15 行：项目内 import

from minisweagent import package_dir
from minisweagent.agents.default import DefaultAgent
from minisweagent.environments.local import LocalEnvironment
from minisweagent.models.litellm_model import LitellmModel

这四行已经暴露架构：路径常量 + Agent + Environment + Model。from x import y 只把 y 绑定到当前模块作用域。

第 17 行：创建 CLI 应用

app = typer.Typer()

类似 const program = new Command()。此时只创建命令容器，还没有运行。

第 20 行：decorator 注册命令

@app.command()

decorator 可以先理解成：定义完 main 后，再做一次 app.command()(main)。它类似 NestJS decorator；不是函数调用的主体。

第 21 行：定义函数

def main(

def 对应 function。Python 用缩进定义函数体，没有 {}。

第 22 行：必填 task 选项

task: str = typer.Option(..., "-t", "--task", ...)
  • task: str 是类型注解，类似 task: string。
  • 这里的 ... 是 Python 的 Ellipsis 单例，Typer 用它表示“必填”，不是 JS rest operator。
  • "-t"、"--task" 是 CLI 短名/长名。
  • prompt=True 表示没有传值时可以询问用户。

第 23-29 行：模型名选项

model_name: str = typer.Option(os.getenv("MSWEA_MODEL_NAME"), ...)

默认来源是环境变量；没有值时 Typer 用 prompt= 文本询问。要注意：类型注解表达“Typer 正常调用时应给字符串”，Python 运行时不会仅凭注解强制类型。

另一个与 JS 默认参数不同的点：这里的 typer.Option(...) 和 os.getenv(...) 在模块 import、函数被定义时就求值，不会在每次直接调用 main() 时重新求值。正常 CLI 调用会由 Typer 解析并注入最终字符串。

第 30 行：返回类型

) -> DefaultAgent:

对应 TS 的 ): DefaultAgent。它帮助 IDE/type checker，不会自动验证实际 return。

第 31 行：设置全局日志

logging.basicConfig(level=logging.DEBUG)

类似初始化全局 logger。DEBUG 会输出比较详细的运行日志。

第 32-36 行：核心装配

agent = DefaultAgent(
    LitellmModel(model_name=model_name),
    LocalEnvironment(),
    **yaml.safe_load(Path(package_dir / "config" / "default.yaml").read_text())["agent"],
)

从内向外读：

  1. package_dir / "config" / "default.yaml"：Path 用 / 运算符拼路径，不是做除法；外层 Path(...) 再包装一次已有 Path，本例中不会改变结果。
  2. .read_text()：读取整个文件，返回 str。
  3. yaml.safe_load(...)：YAML 转成 dict。
  4. ["agent"]：取顶层 agent 配置。
  5. **dict：把 dict 展开成命名参数，近似 JS 的 options object spread。
  6. 先构造 Model 和 Environment，再注入 DefaultAgent。

如果 YAML 得到 {"system_template": "...", "cost_limit": 0}，那么：

DefaultAgent(model, env, **config)

等价于：

DefaultAgent(model, env, system_template="...", cost_limit=0)

第 37-38 行：运行并返回实例

agent.run(task)
return agent

run 执行任务；返回整个 agent 而不是只返回结果，方便调用者检查 agent.messages、成本和调用次数。类似返回一个已经运行过、仍保留状态的 controller 实例。

第 41-42 行：直接执行保护

if __name__ == "__main__":
    app()
  • 直接运行文件时，__name__ == "__main__"，启动 CLI。
  • 被其他模块 import 时，__name__ 是模块名，不自动启动。
  • 类似 CommonJS 的 if (require.main === module)。

5. 逐行读 exceptions.py

第 1 行：继承内置异常

class InterruptAgentFlow(Exception):

class Child(Parent) 对应 class Child extends Parent。继承 Exception 后，它可以被 raise 和 except 使用。

第 2 行：类 docstring

描述设计意图：中断当前 Agent 调用路径，并把 messages 交给外层循环。

第 4 行：可变数量参数

def __init__(self, *messages: dict):
  • __init__ 对应 constructor。
  • self 对应 this，但 Python 必须显式写出来。
  • *messages 对应 JS ...messages rest parameter。
  • 调用 Submitted(msg1, msg2) 后，messages 是 tuple，近似只读语义的 array。

第 5 行：保存 payload

self.messages = messages

把 rest 参数 tuple 存到实例，外层 except ... as e 就能访问 e.messages。

第 6 行：初始化父类

super().__init__()

对应 super()。这里没有传错误文本，所以 str(e) 通常是空字符串；真正信息放在 messages。

第 9-26 行：只用类型区分信号

class Submitted(InterruptAgentFlow): ...
class LimitsExceeded(InterruptAgentFlow): ...
class TimeExceeded(LimitsExceeded): ...
class UserInterruption(InterruptAgentFlow): ...
class FormatError(InterruptAgentFlow): ...

这些类没有自己的方法，类体里的 docstring 已经是一个合法 statement。价值在“类型标签”：

  • except FormatError 可以只处理格式错误。
  • except LimitsExceeded 同时能接住 LimitsExceeded 和子类 TimeExceeded。
  • except InterruptAgentFlow 能接住整个家族。
  • TimeExceeded 继承 LimitsExceeded，表达“时间超限也是一种限制超限”。

异常匹配从具体到宽泛，因此 DefaultAgent.run 先捕获 FormatError，再捕获 InterruptAgentFlow。

6. 逐行读 utils/serialize.py

第 1 行：导入类型占位符

from typing import Any

Any 类似 TS 的 any，只用于类型标注。

第 3 行：创建唯一 sentinel

UNSET = object()

object() 创建一个没有业务含义、但身份唯一的对象，类似 const UNSET = Symbol("UNSET")。

为什么不用 None？因为 None 可能是用户明确要设置的值；UNSET 表示“这次不要覆盖已有值”。

第 6 行：函数签名

def recursive_merge(*dictionaries: dict | None) -> dict:
  • *dictionaries：接收任意数量参数，函数内是 tuple。
  • dict | None：每个参数可以是 dict 或 None，对应 TS Record<string, unknown> | null。
  • -> dict：返回 dict。

第 7-12 行：行为契约

后传入的 dict 优先；嵌套 dict 递归合并；UNSET 跳过。

第 13-14 行：零参数边界

if not dictionaries:
    return {}

空 tuple 是 falsy。对应 if (dictionaries.length === 0) return {}。

第 15 行：新结果对象

result: dict[str, Any] = {}

对应 const result = {}，类型近似 Record<string, any>。它不直接修改输入 dict。

第 16-18 行：从左到右遍历配置层

for d in dictionaries:
    if d is None:
        continue
  • for ... in 对应 for ... of。
  • is None 是身份判断；检查 None 时是 Python 推荐写法。
  • continue 跳到下一个 dict。

第 19 行：遍历 key/value

for key, value in d.items():

tuple unpacking 对应 for (const [key, value] of Object.entries(d))。

第 20-21 行：忽略 UNSET

if value is UNSET:
    continue

这里必须用 is 检查同一个 sentinel，近似 JS 的 value === UNSET。

第 22-23 行：两个嵌套 dict 相遇

if key in result and isinstance(result[key], dict) and isinstance(value, dict):
    result[key] = recursive_merge(result[key], value)
  • key in result 对应 key in result。
  • isinstance(x, dict) 对应运行时类型守卫。
  • 两边都是 dict 才深合并，否则后值整体覆盖。

第 24-26 行：第一次遇到嵌套 dict

elif isinstance(value, dict):
    result[key] = recursive_merge(value)

看似可以直接 result[key] = value，但递归一次能过滤这个新 dict 内部更深层的 UNSET。这正是第 25 行注释解释的逻辑难点。

第 27-28 行：标量和 list 直接覆盖

else:
    result[key] = value

字符串、数字、bool、None、list 等都直接赋值。list 不会像 dict 一样合并。

第 29 行：返回新 dict

return result

输入 dict 的容器没有被修改；但 list 和其他可变非 dict 值会被引用复用，这不是完整 deep clone。

手算一次

base = {
    "agent": {"cost_limit": 3.0, "mode": "confirm"},
    "tags": ["base"],
}
override = {
    "agent": {"cost_limit": 1.0, "mode": UNSET},
    "tags": ["cli"],
}

结果：

{
    "agent": {"cost_limit": 1.0, "mode": "confirm"},
    "tags": ["cli"],
}

cost_limit 被后值覆盖，mode=UNSET 被跳过，list 整体替换。

7. 三份代码在哪里被使用

hello_world.main

  • 文档建议用 python src/minisweagent/run/hello_world.py 运行。
  • tests/run/test_run_hello_world.py:23 做端到端验证。
  • 它不是 mini console script；正式 mini 入口是 run/mini.py。

异常家族

  • Submitted：Local、Docker、Singularity、Bubblewrap、SWE-ReX、ConTree 环境检测完成标记后抛出；InteractiveAgent 可先询问是否真的结束。
  • LimitsExceeded / TimeExceeded：DefaultAgent.query 检查调用数、费用、墙钟时间时抛出；InteractiveAgent 对两者采取不同处理。
  • UserInterruption：InteractiveAgent 在用户拒绝命令、切换模式或追加任务时抛出。
  • FormatError：text/tool-call/Responses action parser 在模型回答不符合协议时抛出；DefaultAgent 把纠错 message 放回下一轮。
  • InterruptAgentFlow：DefaultAgent.run 的统一兜底捕获点。

recursive_merge / UNSET

  • run/mini.py 和三个 benchmark 入口：合并 YAML 与 CLI 覆盖。
  • DefaultAgent.get_template_vars：合并 Agent、Model、Environment 和运行时模板变量。
  • DefaultAgent.serialize：合并 Agent、Model、Environment 的轨迹数据。
  • 所有主要 Environment：合并配置、平台信息、环境变量和额外模板变量。
  • UNSET 主要用于 CLI：用户没传某项时不要覆盖 YAML；显式传 0 或 False 时仍允许覆盖。

8. 无 API 小练习

目标：一次看到 **config 装配、UNSET 合并、Submitted 流程信号。这里先写三个极小的替身类，不访问 API，也不执行 shell。

MSWEA_SILENT_STARTUP=1 PYTHONPATH=src .venv/bin/python - <<'PY'
from minisweagent.exceptions import Submitted
from minisweagent.utils.serialize import UNSET, recursive_merge


class FakeModel:
    def query(self, task: str) -> dict:
        return {"role": "assistant", "content": f"plan: {task}"}


class FakeEnvironment:
    def execute(self, message: dict) -> None:
        raise Submitted({"role": "exit", "content": message["content"]})


class TinyAgent:
    def __init__(self, model: FakeModel, env: FakeEnvironment, **config):
        self.model = model
        self.env = env
        self.config = config

    def run(self, task: str) -> dict:
        try:
            self.env.execute(self.model.query(task))
        except Submitted as signal:
            return signal.messages[0]


config = recursive_merge(
    {"agent": {"step_limit": 3, "label": "default", "output_path": "result.json"}},
    {"agent": {"step_limit": 1, "label": UNSET, "output_path": None}},
)
agent = TinyAgent(FakeModel(), FakeEnvironment(), **config["agent"])

print(config)
print(agent.config)
print(agent.run("read code"))
PY

已验证输出：

{'agent': {'step_limit': 1, 'label': 'default', 'output_path': None}}
{'step_limit': 1, 'label': 'default', 'output_path': None}
{'role': 'exit', 'content': 'plan: read code'}

练习里的 f"plan: {task}" 是 f-string，对应 JS template literal `plan: ${task}`。

动手改三次：

  1. 运行前先手算三次 print；再对照真实输出。
  2. 把第二层的 step_limit 改成 UNSET：旧值 3 应该保留，而不是让 key 消失。
  3. 给 Submitted(...) 再传一个 dict，并把 return signal.messages[0] 暂时改成 return signal.messages，观察它是长度为 2 的 tuple。

9. 三道检查题（先不要看答案）

  1. hello_world.py:35 中，read_text -> safe_load -> ["agent"] -> ** 四步各产生什么值？为什么 ** 不能展开 list？
  2. 预测 recursive_merge({"a": 1, "b": [1]}, {"a": UNSET, "b": [2], "c": None) 的结果，并解释为什么 UNSET 不能直接换成 None。
  3. TimeExceeded 为什么继承 LimitsExceeded 而不是直接继承 InterruptAgentFlow？except LimitsExceeded 和 except InterruptAgentFlow 分别能捕获哪些类型？

10. 可维护性与性能审查

hello_world.py

  • Typer 的 Option(...) 同时充当 CLI 元数据和 Python 默认值；绕过 Typer 直接调用 main() 且省略参数时，参数可能是 OptionInfo 而不是普通字符串。更清晰的结构是 CLI 函数只解析参数，再调用一个纯 build_agent(task, model_name) 函数。
  • 现有 hello-world 测试直接调用 main(task=...)，同时替换了 LitellmModel，所以没有暴露默认 model_name 可能是 OptionInfo 的问题。
  • logging.basicConfig 修改全局日志配置，作为示例脚本合理，作为库函数副作用较重。
  • LocalEnvironment 会在宿主机用 shell 执行模型命令；真实使用时这是安全边界。
  • 第 35 行很密集，但符合仓库“保持代码最小”的风格。拆分会更易调试，却违背项目偏好的简洁装配写法。

改动风险：拆分 main 可能影响文档、Typer 命令签名和直接调用它的测试；更换默认 Environment 会改变 hello-world 示例语义。

exceptions.py

  • super().__init__() 没有错误文本，因此普通日志中的 str(e) 为空；调试依赖 messages。可以传摘要，但会改变序列化/日志表现。
  • messages 实际是 tuple，属性没有显式写成 tuple[dict, ...]；更精确的类型能帮助自定义实现。
  • 用异常做正常控制流有认知成本，但能让深层 Environment/Parser 直接跳回 Agent 循环，保持每个类很短。

改动风险：改成 return status 需要同时修改所有 Environment、所有 action parser、DefaultAgent、InteractiveAgent 和相关测试，是协议级变更。

serialize.py

  • 时间复杂度约 O(遍历到的 key 数)，普通配置规模下可以忽略；极深 dict 才可能碰到 Python 递归深度限制。
  • 它只复制 dict 容器，list 等可变值仍与输入共享引用。改成 deepcopy 更符合“完全独立副本”的直觉，但更慢，也可能破坏 UNSET 身份或无法复制第三方对象。
  • 现有“不修改输入”测试比较调用前后的相等性，没有修改返回值中的嵌套 list 来检查别名，因此不会捕获这类共享引用。
  • UNSET 本身不可 JSON 序列化；当前递归会过滤 dict value 中的 UNSET，但放在 list 内不会过滤。
  • 只含 UNSET 的嵌套 dict 会留下空父节点，例如 {"run": {}}；测试明确接受这个行为。

改动风险：任何 merge 语义变化都会影响 CLI 覆盖优先级、模板变量、轨迹序列化和 benchmark 配置，应先扩充 tests/utils/test_serialize.py 再改。