04. Agent 核心循环：让模型回答变成下一轮上下文

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

配套练习：下载 04_agent_core_loop.py。

系列导航

返回路线图 · 上一篇：03. Protocol 与多态：用同一套插座更换三类组件 · 下一篇：05. Environment 与退出协议：把 action 变成真实副作用

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

• src/minisweagent/agents/default.py:38-188
• tests/agents/test_default.py

本节目标：关闭源码后，能够写出 run -> step -> query -> execute_actions；能够在纸上推演两轮后的 messages；能够解释成功、超限、格式错误和意外异常为何走不同分支。

1. 先用 JavaScript 建立心智模型

1.1 DefaultAgent 是同步状态机

如果用 JavaScript 重写核心结构，大致是：

class DefaultAgent {
  constructor(model, env, config) {
    this.model = model;
    this.env = env;
    this.config = AgentConfig.parse(config);
    this.messages = [];
    this.cost = 0;
    this.nCalls = 0;
  }

  run(task) {
    this.messages = [makeSystemMessage(), makeUserMessage(task)];

    while (true) {
      try {
        this.step();
        this.nConsecutiveFormatErrors = 0;
      } catch (error) {
        if (error instanceof AgentFlowSignal) {
          this.messages.push(...error.messages);
        } else {
          this.recordUnexpectedError(error);
          throw error;
        }
      } finally {
        this.save();
      }

      if (this.messages.at(-1)?.role === "exit") break;
    }

    return this.messages.at(-1)?.extra ?? {};
  }

  step() {
    const modelMessage = this.query();
    return this.executeActions(modelMessage);
  }
}

Python 版本当前全部同步执行，没有 Promise、async 或 await。Model 网络调用和 Environment 命令执行都会阻塞当前线程。

1.2 关键对象的 JS 类比

1.3 先记住全仓核心句

return self.execute_actions(self.query())

Python 与 JS 一样先计算函数参数，所以它不表示“同时做两件事”，而是：

message = self.query()
return self.execute_actions(message)

一次普通 step 会追加两类消息：query() 追加 assistant，execute_actions() 追加 observation。

2. 5W2H

3. 完整控制流

注意：真正终止条件是最后一条消息的 role == "exit"，不是“发生过异常”。一个不携带 exit 消息的 InterruptAgentFlow 只会给下一轮增加上下文。

4. 先认清五种数据

messages 会保存 model message 和 observation，而不是只保存用户聊天文本。它既是下一轮输入，也是 trajectory 的核心内容。

5. 按路线图顺序逐行读 default.py:38-188

第 38-40 行：定义 class 与构造器合同

class DefaultAgent:
    def __init__(self, model: Model, env: Environment, *, config_class: type = AgentConfig, **kwargs):

• model、env 是被注入的两个策略对象。
• 单独的 * 表示它后面的 config_class 只能按名称传入，不能作为第三个位置参数。
• config_class: type 表示这里接收 class object，不是 config 实例。
• **kwargs 收集剩余具名参数，对应 JS 的 options object。

第 41 行：验证配置

self.config = config_class(**kwargs)

近似 AgentConfig.parse(options)。Pydantic 检查必填模板、默认限制和 Path 类型；错误会在 Agent 构造阶段直接抛出。

第 42-50 行：所有实例状态

前导 _ 只是内部使用约定。特别注意：计时从构造 Agent 开始，而不是从 run() 开始。

第 52-64 行：合并模板变量

Agent config
  < Environment vars
  < Model vars
  < 实时统计
  < extra_template_vars
  < 本次 kwargs

recursive_merge 后传入的 dict 优先级更高：

• 第 54 行把 Pydantic config 转成普通 dict。
• 第 55-56 行让 Environment、Model 分别贡献变量。
• 第 57-61 行加入调用次数、费用和整数秒耗时。
• 第 62 行加入由 run() 保存的 task 等变量。
• 第 63 行让本次调用参数拥有最高优先级。

因此自定义变量可以覆盖前面的同名 config；这既灵活，也意味着命名冲突会静默覆盖。

第 66-67 行：严格渲染 Jinja

return Template(template, undefined=StrictUndefined).render(**self.get_template_vars())

相当于 Nunjucks 渲染，但 StrictUndefined 会让拼错或缺失变量立刻报错，不会偷偷输出空字符串。每次调用都会重新构造 Template。

第 69-72 行：唯一的历史追加入口

• *messages 对应 JS rest parameter ...messages。
• logger.debug(messages) 记录的是本次传入的 tuple。
• self.messages.extend(messages) 对应 array.push(...items)。
• 返回 list(messages)，只包含本次新增项，不是完整历史。

这里没有复制 message dict；外部若之后修改同一 dict，历史也会跟着变。

第 74-86 行：把意外异常写成 exit

handle_uncaught_exception(e) 调用当前 Model 的 format_message()，生成 provider 对应的 exit 消息：

• content 是 str(e)。
• exit_status 是异常 class 名。
• submission 为空。
• 同时保存异常文本和当前 traceback。

它只负责记录，不负责吞掉异常；run() 稍后还会 raise。

第 88-95 行：run() 初始化一轮任务

第 90 行：

self.extra_template_vars |= {"task": task, **kwargs}

|= 近似 Object.assign(existing, {task, ...kwargs})。同名 key 被新值覆盖，旧的未提及 key 会保留。

第 91 行只清空 messages。第 92-95 行依次：

1. 渲染 system template。
2. 用 Model 包装 system 消息。
3. 渲染 instance template，其中可读取 task。
4. 用 Model 包装 user 消息。
5. 一次 add_messages 加入两条。

这段初始化在后面的 try/finally 外面；若模板缺变量，异常不会被转换为 exit，也不会自动 checkpoint。

第 96-99 行：正常循环

• while True 是无限循环，退出由第 120-121 行控制。
• 一次 self.step() 必须完整返回，才会把连续格式错误计数重置为 0。
• “Model 成功返回”还不算 clean step；Environment 执行和 observation 格式化也必须完成。

第 100-112 行：FormatError 子状态机

FormatError 继承 InterruptAgentFlow，所以必须先捕获更具体的子类。

1. 每次错误先 += 1。
2. 0 < max <= current 表示启用限制且已达到阈值。
3. 未达到时只加入异常携带的纠错消息，下一轮让模型重试。
4. 达到时先加入纠错消息，再加入 RepeatedFormatError exit。

max_consecutive_format_errors=0 表示关闭这一限制。

第 113-114 行：其他控制流信号

Submitted、LimitsExceeded、TimeExceeded、UserInterruption 都会走这里。Agent 不根据异常 class 再写条件；它只把 e.messages 展开并加入历史。

这就是异常携带消息的价值：深层 Environment 可以直接把控制权交还最外层循环。

第 115-119 行：意外异常与 finally

• 普通 Exception 先写入异常 exit。
• 裸 raise 重新抛出当前异常，保留原 traceback；它不是新建异常。
• finally 无论正常、可恢复错误、完成还是意外异常都会执行 save()。

如果 handle_uncaught_exception() 或 save() 自己失败，新异常可能掩盖最初的错误。

第 120-122 行：真正退出条件和返回值

if self.messages[-1].get("role") == "exit":
    break
return self.messages[-1].get("extra", {})

普通文本/tool-call 模型的最终统一边界都是带 role="exit" 的消息。run() 返回的是最后一条消息的 extra，不是完整 Agent、messages 或 content。

第 124-126 行：核心 step()

return self.execute_actions(self.query())

真实求值顺序：query() 完整结束后，返回 message 才进入 execute_actions(message)。任一阶段抛异常，后面的阶段不会执行。

第 128-137 行：step/cost 软限制

if 0 < step_limit <= n_calls or 0 < cost_limit <= cost:

• 第一个 0 < 让 0 表示禁用；负数也会被视为禁用。
• 检查发生在下一次模型调用之前。
• step_limit=1 时，第一次 query 看到 n_calls=0，允许调用；第二次看到 1 <= 1 才退出。
• cost 也可能由一次调用从限额下方直接花到上方，甚至该轮 action 已执行；下一轮才停止。

达到任一限制就抛出携带 LimitsExceeded exit 的异常。

第 138-145 行：wall-time 软限制

同样只在 query 前检查。它不会中断正在等待的模型，也不会杀掉正在执行的命令；Environment 自己的 command timeout 是另一套机制。

int(time.time() - _start_time) 会截断小数秒。测试明确证明：wall limit 为 1 秒时，当前 sleep 2 仍先执行完，第二轮才得到 TimeExceeded。

第 146-150 行：调用 Model 并记录 assistant

顺序不能读反：

1. n_calls += 1，失败调用也计数。
2. 把完整 self.messages 交给 Model。
3. 正常返回后，从 message.extra.cost 累加实例费用；缺失按 0。
4. 将 assistant/provider response 加入历史。
5. 返回同一个 message 给 action 阶段。

若 Model 抛 FormatError，n_calls 已增加，但第 148-150 行不会执行。

第 152-155 行：action -> output -> observation

第 154 行列表推导式等价于：

outputs = []
for action in message.get("extra", {}).get("actions", []):
    outputs.append(self.env.execute(action))

actions 按顺序同步执行，不是并发。缺少 extra 或 actions 时使用空列表。

第 155 行让 Model 决定 observation 的 provider 格式，再用 add_messages(*items) 展开加入历史。这就是同一 Agent 支持 user observation、tool message 和 function-call output 的原因。

若 Environment 抛 Submitted，代码会在 formatter 之前跳走，所以提交 action 没有 observation。

第 157-178 行：生成 trajectory dict

• 第 159 行为空历史提供 {}，避免 messages[-1] 越界。
• 第 160 行读取最后消息 extra。
• 第 163-166 行保存实例 cost 与 API 调用数。
• 第 168 行 model_dump(mode="json") 把 Path 等值转为 JSON 兼容类型。
• 第 169 行记录实际运行 class 的完整 dotted path；子类会记录子类名。
• 第 171-176 行保存版本、最终状态、完整消息和 trajectory schema 版本。
• 第 178 行合并 Agent、Model、Environment 与额外 dict。

合并优先级为：

agent_data < model.serialize() < env.serialize() < extra_dicts

benchmark 和保存测试依赖最后的 extra dict 能覆盖 exit_status；不要随便改顺序。

第 180-188 行：可选写盘

1. 无论 path 是否为 None，先调用 serialize()。
2. 有 path 才创建父目录。
3. json.dumps(..., indent=2) 生成文本。
4. Path.write_text() 覆盖写入文件。
5. 始终返回 data dict。

因此 save(None) 的意思是“只序列化，不写文件”，不是“什么都不做”。

6. 两轮成功任务的 messages 推演

对应 tests/agents/test_default.py:333-361：

第 4 与第 5 项之间没有 observation：Environment 在发现完成标记后直接抛 Submitted，execute_actions() 尚未来得及调用 formatter。

7. 用 JS 测试经验读 tests/agents/test_default.py

第 1-16 行：测试依赖

• pytest 对应 Jest/Vitest 测试运行器。
• yaml 读取真实 Agent 配置。
• DefaultAgent + LocalEnvironment 走真实控制循环和受控本机命令。
• 三个 Deterministic Model 不请求 API，只按顺序返回 fixtures。

这份测试没有 patch Model API；它通过注入固定实现取得确定性。

第 18-53 行：跨消息协议的读取 adapter

get_text()：

• content 为 None -> 空字符串。
• content 为 str -> 直接返回。
• content 为非空 list -> 取第一个 item 的 text。
• 其他形状 -> 空字符串。

get_observation_text()：Responses API 的 observation 使用顶层 output，其余复用 get_text()。

is_assistant_message() 与 is_observation_message() 是测试侧 type guards，把三种 provider envelope 归一化。它们让业务断言关注“这是 assistant/observation”，而不是绑定某个 API 的字段布局。

第 56-74 行：两个配置 fixture

@pytest.fixture 近似 Jest 的 beforeEach + dependency injection。测试函数只要声明同名参数，pytest 就先调用 fixture 并注入结果。

• default_config 读取旧式文本 action 模板。
• toolcall_config 读取原生 tool-call 模板。
• 两者都只返回 YAML 的 agent section。

with open(...) 是自动关闭文件的上下文管理器。按当前仓库风格更常见的写法会是 Path.read_text()。

第 77-113 行：三个固定模型 builder

统一输入都是：

list[tuple[str, list[dict]]]

近似 TS 的 Array<[string, Action[]]>。

• make_text_model 用 list comprehension 直接生成普通输出。
• make_tc_model 用 enumerate() 同时拿索引和值，为每个 action 创建稳定 tool_call_id 和 OpenAI tool-call envelope。
• make_response_api_model 创建 Responses API action ids，再复用 make_response_api_output()。

Agent 最终统一读取 extra.actions；原始 envelope 主要用于消息历史和 observation 对应关系。

第 116-124 行：一次写测试，跑三种 Model

@pytest.fixture(params=["text", "toolcall", "response_api"])

近似 describe.each(["text", "toolcall", "response_api"])。request.param 是当前参数；fixture 返回 (factory, config) tuple。

前 15 个测试都依赖 model_factory，所以各跑 3 次；最后 2 个 FormatError 测试只跑一次：

15 * 3 + 2 = 47 cases

第 130-150 行：成功结束

第一条固定响应执行普通 echo，产生 observation；第二条让 Environment 输出完成标记并抛 Submitted。断言证明：

• 最终状态是 Submitted。
• 标记后面的文本成为 submission。
• 两轮恰好调用模型两次。

第 153-183 行：step 与 cost 限制

test_step_limit_enforcement 设置 step_limit=1，证明第一次调用允许、第二轮 query 前停止，n_calls == 1。

test_cost_limit_enforcement 的固定消息 cost 默认是 1.0，而限制是 0.5。第一轮仍会完成，下一轮才退出。这验证的是软限制，不是预先保留预算。

语法 **{**config, "step_limit": 1} 有两层展开：内层构造覆盖后的新 dict，外层再把它展开为 Python 具名参数。

第 185-228 行：Environment command timeout

这里不是 Agent wall-time：LocalEnvironment(timeout=1) 捕获子进程超时并返回错误 output，Agent 把它格式化成 observation，所以第二轮仍能恢复并提交。

第二个测试先 echo 再 sleep，证明 TimeoutExpired 携带的 partial stdout 没有丢失。

第 231-253 行：多轮循环

三个普通 action 各产生 observation，第四个 action 提交。cost limit 被提高到 5，最终断言 n_calls == 4。

第 256-282 行：配置和 task 模板

测试覆盖 system/instance template、step/cost 设置，并检查：

• messages[0] 是自定义 system 文本。
• run("Test custom config") 的 task 出现在 messages[1]。

第 285-306 行：实时模型统计模板变量

测试手动加入 system/user，再直接调两次 agent.query()，不执行 Environment。每条固定消息 cost=1，因此模板渲染为 Calls: 2, Cost: 2.0。

第 309-330 行：timestamps

用 list comprehension 筛出 assistant/provider response，要求每条都有 timestamp；已有 timestamp 的消息必须是 float。

测试名称说 assistant 和 observation 都应有 timestamp，但当前断言没有明确要求每条 observation 一定包含 timestamp，这是一个覆盖缺口。

第 333-362 行：完整消息历史

这是本课最值得手推的测试：它验证六条消息数量和 system/user/assistant/observation/assistant 的位置，最终 exit 已由长度与返回状态间接证明。

第 364-384 行：单次 step()

测试先手动放入两条上下文，记录长度，然后只调用 agent.step()。正常 action 后恰好多两条：assistant + observation。这是核心等式最直接的证据。

第 387-406 行：observation 内容

三条模型响应中，前两条普通命令的 stdout 依次进入 observation；第三条提交直接产生 exit。因此 observation 数量是 2，不是 3。

第 409-439 行：Agent wall-time

• limit=1 并不会杀掉已经运行的 sleep 2。
• 命令完成后，第二轮 query 前抛 TimeExceeded。
• elapsed_seconds 是 int，wall_time_limit_seconds 也进入模板变量。

第 442-459 行：空 actions

首轮 actions=[] 时：

• query 仍计数并加入 assistant。
• Environment 不执行任何内容。
• formatter 通常返回空 observation list。
• 最后一条不是 exit，所以继续下一轮。

第 462-477 行：可控的 FormatError 模型

_FlakyToolcallModel 继承固定模型并重写 query。前导 _ 表示只供本测试模块使用。

• 每次先移动索引。
• fixture 标有 _format_error 时抛带纠错消息的 FormatError。
• 否则返回正常输出。

这里故意模拟真实模型回答被截断或没有 tool call 的情况。

第 480-492 行：连续错误终止

预置五个错误，但阈值为 2。第二次错误后 Agent 加入 RepeatedFormatError exit，因此只消耗两次模型调用。

第 495-511 行：成功 step 重置错误计数

序列是：

error -> clean step -> error -> submit

两个错误之间有完整成功 step，计数被清零，所以不会被当成连续两次，最终正常 Submitted。

8. 这些代码在哪里被使用

直接构造和运行

• run/hello_world.py:32,37：直接构造 DefaultAgent 并调用 run。
• agents/__init__.py:9：工厂短名 default 的目标类。
• run/mini.py:99-102：通过三个工厂装配；默认实际常是 InteractiveAgent，但它复用本循环。
• run/benchmarks/swebench_single.py:90,96：单 benchmark 装配和运行。
• run/benchmarks/swebench.py:154,164、programbench.py:101,115：运行并补充保存结果。

子类扩展点

• InteractiveAgent 继承 DefaultAgent，覆盖 add_messages/query/step/execute_actions，再通过 super() 回到这里。
• ProgressTrackingAgent.step() 先更新进度，再调用 super().step()。
• cookbook 中的自定义 Agent 也通过重写这些小方法扩展，而不是复制 run 循环。

测试与 trajectory 消费

• tests/agents/test_default.py 验证循环行为。
• tests/run/test_save.py 验证 Agent/Model/Environment 类型和合并后的 trajectory。
• inspector、benchmark 输出和恢复逻辑依赖 trajectory_format 与 messages 结构。

9. 可维护性、性能与测试审查

高价值风险

1. Agent 实例的复用语义不清晰。 run() 只重置 messages，不重置 n_calls、cost、开始时间、格式错误计数，也会保留旧 extra template keys。官方路径都是构造一次、运行一次；若支持复用，需要先定义累计预算还是每任务独立。
2. FormatError 的实例费用可能漏记。 n_calls 在 Model 前增加，但 cost 只在正常返回后增加。真实 LitellmModel 已计算并加入全局费用后，action 解析仍可能抛 FormatError；结果是全局统计计费、Agent instance_cost 和 cost_limit 却低估。修复涉及 Model 异常 payload、所有 provider 和测试。
3. 多 action 中途退出会丢前序 observation。 默认实现先收集完整 outputs 再统一格式化；后一个 action 抛 Submitted/异常时，前一个 output 尚未写进 history。InteractiveAgent 已用 try/finally 保留 partial outputs；统一实现时必须保留确认和续任务逻辑。
4. 初始化和错误记录仍可能失败。 初始模板渲染在 try 外；unexpected error 的 model formatter 或 finally 中 serializer/write 失败，都可能让轨迹缺失或掩盖主要异常。
5. trajectory 可能含敏感配置和原始响应。 Model/Environment serialize 会合并完整配置；保存文件不应默认当作可公开日志。

持久化性能

• 每轮即使 path=None，也会调用三个 serializer。
• 有 output_path 时，每轮重新 JSON 编码并覆盖不断增长的完整 messages，累计写入量接近 O(n²)。
• 覆盖写不是原子的；进程在 write 中断可能留下截断 JSON。
• serialize() 返回的 messages 是原 list 引用，不是隔离快照；Agent 后续追加会改变此前返回 data 看到的列表。

可考虑原子临时文件、降低 checkpoint 频率或事件追加日志。风险是 inspector、恢复机制、benchmark 和既有 trajectory schema 都依赖当前完整快照。

测试质量与缺口

• 最大优点：三种 Model 格式共享同一批 15 个行为测试，确实验证了多态边界。
• 47 passed 约需 13 秒，因为大多数测试实际启动本机 shell，timeout/wall-time 还会 sleep。可用 MemoryEnvironment 承担纯 Agent 单测，只保留 sentinel、timeout 和 partial-output 的少量 LocalEnvironment 集成测试。
• cost 测试没有断言实际 overshoot；timestamp 测试没有强制每条 observation 带时间。
• 当前文件没有直接覆盖 unexpected Exception、保存失败、同一 Agent 二次 run、多 action partial output。
• make_tc_model 手拼 JSON arguments；复杂引号/换行可能生成无效 JSON。若开始验证原始 tool_calls，应改用 json.dumps()。

10. 无 API、无 shell 小练习

练习脚本已作为本文附件提供。

它使用真实 DefaultAgent、固定响应 Model 和纯内存 Environment。Environment 只记录字符串；收到 finish 时抛 Submitted，不调用 subprocess。

先预测四行输出，再运行：

MSWEA_SILENT_STARTUP=1 PYTHONPATH=src .venv/bin/python \
  exercises/04_agent_core_loop.py

核心代码：

class MemoryEnvironment:
    def execute(self, action: dict, cwd: str = "") -> dict:
        command = action["command"]
        self.actions.append(command)
        if command == "finish":
            raise Submitted(
                {
                    "role": "exit",
                    "content": "offline complete",
                    "extra": {"exit_status": "Submitted", "submission": "offline complete"},
                }
            )
        return {"output": f"recorded: {command}", "returncode": 0, "exception_info": ""}

已验证输出：

{'exit_status': 'Submitted', 'submission': 'offline complete'}
['read files', 'finish']
['system', 'user', 'assistant', 'user', 'assistant', 'exit']
2 0.5

动手改两次：

1. 把 step_limit=3 改成 1。先预测 info、actions、角色序列和 n_calls，再运行确认限制发生在第二次 query 之前。
2. 把 finish 分支改成返回普通 output，不抛 Submitted。解释为什么下一轮会耗尽 DeterministicModel.outputs 并抛 IndexError。

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

1. step() 只有一行，为什么一次普通 step 会向 messages 新增两条消息？请按真实求值顺序说出方法调用链。
2. 两个固定模型输出、step_limit=1 时，为什么结果是 LimitsExceeded 且 n_calls == 1，而不是调用模型两次？
3. 为什么提交轮的历史是 assistant -> exit，中间没有 observation？