智能体主循环:generate → 执行工具 → 回灌
30 秒导读: gptme 的「自主」不是魔法,而是一个三层嵌套的循环。用户说一句话,agent 可能自己连跑好几步;每一步先让模型生成回复,把回复里的工具跑掉,再把工具输出回灌进 对话历史——只要模型还想调工具,就自动再生成一步,直到它不再调工具才停下、把控制权还给你。 本章只讲这台发动机,全部代码集中在
gptme/chat.py。
本章是 gptme 文档 的核心一章。它只回答一个问题:「让 AI 自己动手干活」这句话, 在代码里到底是怎么转起来的? 工具怎么被解析和执行(→02)、消息与上下文 怎么组装(→04)、钩子本身怎么定义(→05), 都不在这里展开。
1. 这是什么(零基础也能懂)
先建立一个直觉:什么叫「agent 自己跑多步」
想象你在终端里对一个 AI 助手说:「帮我看看这个项目为什么测试挂了。」
一个只会聊天的模型只能回你一段话:「你可以先跑 pytest 看看。」然后就停了,等你手动去跑、
再手动把结果贴回来。
而一个 agent 会这样:它先说「我先跑一下测试」——然后真的自己去跑了 pytest,拿到报错,
不用你插手,接着说「看到了,是 foo.py 第 12 行的问题,我去看一眼」——又自己打开了文件,
读完再说「我改一下」——自己改了,再自己重跑测试确认通过。整个过程你只说了开头那一句。
这就是 agent 主循环干的事:把「模型生成一段话」和「真的去执行那段话里的动作」这两件事, 串成一个能自动往前滚的循环。
一句话定义
gptme/chat.py 是 gptme 的中央控制循环:它反复地「让模型生成回复 → 执行回复里的工具 →
把工具结果回灌进历史 → 再让模型生成」,直到本轮没有工具要跑,才回到用户。
用起来什么样
下面是一次真实交互的形态(agent 在一句用户输入后,自己跑了两步):
User: 这个仓库有多少个 Python 文件?
Assistant: 我数一下。
```shell
find . -name '*.py' | wc -l
```
System: 42 ← 工具输出被"回灌"进历史,循环没有停,自动再生成
Assistant: 一共 42 个 Python 文件。 ← 这一步没有工具了,循环停下,把控制权还给你
User: ← 轮到你说下一句
注意:用户只打了一次字,agent 却生成了两次(中间夹着一次工具执行)。「什么时候该继续、 什么时候该停」不是模型用特殊 token 声明的,而是这台循环根据「上一条回复里还有没有可运行的工具」 自己判断的——这是本章最关键的机制。
一句话直觉/类比
把主循环想成一个记账员盯着一台传送带:模型往传送带上放一段回复,记账员检查里面有没有 「待办工具」;有就执行、把结果贴回账本(历史),再让模型看着更新后的账本继续写;直到某段回复 里一个待办工具都没有,记账员才合上账本,喊「下一位」。
2. 顶层全景(它大概怎么转)
三层嵌套的循环
主循环不是一个 while,而是三层,由外到内一层比一层「小」。看懂这张图,本章就懂了一半:
chat() ← 会话级:一次进程,做初始化、锁模型、装历史、收尾
└─ _run_chat_loop() ← 轮次级:一个 while,反复取"下一条输入"
每次拿一条输入(队列里的 / 用户敲的)
└─ _process_message_conversation() ← 步骤级:一个 while,自主跑多步
└─ step() → step() → step() … ← 每步 = 生成一次 + 执行工具一次
↑______________________|
只要上一步回复里"还有可运行的工具"就再来一步
一句话读这张图:「轮次」是「你说一句话」,「步骤」是「agent 为了回应这句话,自己动手的每一下」。 一轮里可以有很多步,一步里可以有很多个工具。
一步(step)内部的数据流
最内层的 step() 是发动机的活塞。它做的事恰好就是标题那三个动词:
prepare_messages reply() execute_msg()
历史 ──▶ 裁剪/组装上下文 ──▶ 模型生成一段回复 ──▶ 解析回复里的工具、逐个执行
▲ │ │
│ ▼ ▼
└────────────── 回灌:回复 + 工具输出都 append 回历史 ◀───────┘
生成(generate)→ 执行(execute tools)→ 回灌(feed back)——回灌完历史变长了,
外 层循环若判定「还有工具要跑」,就拿着这份更长的历史再进一次 step()。 自主推进就来自这个回环。
部件一句话职责
| 部件 | 干什么 | 位置(gptme/chat.py) |
|---|---|---|
chat() | 会话入口:初始化、锁定模型/流式、装载历史、切工作目录、收尾钩子 | chat.py:47 |
_run_chat_loop() | 轮次循环:统一「队列输入」和「用户输入」,拦截 /命令,处理中断 | chat.py:176 |
_process_message_conversation() | 步骤循环:自主跑多步,判断何时停,后台自动命名 | chat.py:333 |
step() | 单步:组装上下文 → 生成回复 → 执行工具,yield 出每条消息 | chat.py:520 |
_get_user_input() / _should_prompt_for_input() | 决定「该问用户」还是「直接替用户续跑」(崩溃恢复) | chat.py:483 / chat.py:449 |
_drain_external_prompt_queue() | 把磁盘上的持久排队 prompt 并进内存队列 | chat.py:317 |
3. 核心原理(逐个机制,由浅入深)
3.1 会话启动:chat() 把桌子摆好
它要解决的小问题: 循环真正转起来之前,得先确定用哪个模型、能不能流式、历史从哪来、 工作目录在哪、以及——如果我是被别的 agent 内联调用的,别把人家的输出格式搞乱。
做了哪些事(按代码顺序):
-
锁定输出格式,并保存调用者的格式。
_prev_output_format = get_output_format(), 再set_output_format(...);整个函数体包在try/finally里,finally里还原成调用者的格式 (chat.py:87-89、chat.py:170-173)。这一步专为「subagent 内联 chat」设计,见 §3.6。 -
初始化 + 触发会话开始钩子。
init(...)装配模型/工具/确认策略(chat.py:93);随后触发HookType.SESSION_START,钩子可以往initial_msgs里追加消息(如重放 todo 状态) (chat.py:96-104)。 -
确定模型、按能力关掉流式。 若模型不支持流式而调用方要了流式,就静默降级
stream = False(chat.py:116-122)——这样后面step()不必再操心。 -
装载历史、切到工作目录。
LogManager.load(logdir, initial_msgs=..., create=True)把历史 读进内存(chat.py:126);os.chdir(workspace)(chat.py:134)让工具的相对路径落在正确的项目里。 -
把 prompt_msgs 变成一个队列。
prompt_queue = list(prompt_msgs)(chat.py:146)——这是关键的 统一动作:命令行-传入的多条初始 prompt、和后面用户交互敲的输入,都走同一条队列/输入路径, 见 §3.2。 -
进主循环 / 收尾。 调
_run_chat_loop(...)(chat.py:149)。若循环里抛出SessionCompleteException(autonomous 模式下 agent 主动宣告完工),chat()捕获它、 触发SESSION_END钩子后干净退出(chat.py:160-169)。
真实实现(模型能力降级流式):
# gptme/chat.py:116 —— 不支持流式的模型,自动关掉 stream
if not modelmeta.supports_streaming and stream:
logger.info("Disabled streaming for '%s/%s' model (not supported)", ...)
stream = False
这段的价值:把「能力判断」收在入口做一次,内层
step()拿到的stream已经是「真能用的」值。
3.2 轮次循环:一条队列统一「排队的」和「敲进来的」
它要解决的小问题: 输入有两个来源——(a) 启动时传入的 prompt、钩子塞回来的消息,(b) 交互时 用户实时敲的字。如果两套代码分别处理会很乱。gptme 的解法是:优先吃队列,队列空了才问用户。
_run_chat_loop() 是一个 while True(chat.py:189)。每圈开头先把磁盘上的持久排队 prompt
并进内存队列(_drain_external_prompt_queue,chat.py:190),然后二选一:
每圈:
┌─ prompt_queue 非空? ──是──▶ msg = 队列.pop(0) ← 排队的输入(初始 prompt / 钩子回灌)
│ append 进历史
│ 是 /命令 就执行并 continue
│ 触发 TURN_PRE 钩子
│ _process_message_conversation() ← 进步骤循环
│
└─ 队列空? ──否交互──▶ break(非交互模式:prompt 跑完就退出)
└─交互──▶ _get_user_input() ← 问用户要下一句
几个要点:
-
/命令拦截。 若消息是用户角色且execute_cmd(msg, manager)返回真(说明这是个被消费掉的 斜杠命令,如/undo、/model),就continue,不进模型(chat.py:201、chat.py:257)。 -
TURN_PRE只在「一次真正的用户提问」时触发一次。 在消息 append 之后、且命令处理已经放行 之后,才触发(chat.py:204-210、chat.py:263-268)——保证一轮只触发一次,而不是每步。 -
非交互模式的退出条件。 若
interactive=False且队列已空,直接break(chat.py:230-232)—— 这正是脚本化/CI 里「把 prompt 喂完就退出」的行为。 -
持久队列的意义。
_drain_external_prompt_queue(chat.py:317)从manager.logdir拉取别的进程 (如 web server)排进来的 prompt,受MAX_PROMPT_QUEUE_SIZE(=100)容量约束——这让「外部往一个 正在跑的会话追加指令」成为可能。
3.3 步骤循环:一次请求如何自主跑多步(本章核心)
它要解决的小问题: 用户问一句,agent 可能需要「查文件 → 读内容 → 改代码 → 跑测试」好几下才能 真正答上。谁来决定「还要不要再来一下」?
答案: _process_message_conversation() 里的 while True(chat.py:355)。它每圈调一次 step(),
step() 会把「模型回复 + 所有工具输出 」作为一串消息 yield 出来;循环把它们 append 回历史,然后
检查最后一条 assistant 回复里还有没有「可运行的工具」——有就再转一圈,没有就停。
判停的真实代码——这是整台发动机的「油门」:
# gptme/chat.py:428 —— 上一条 assistant 回复里还有可运行工具吗?
last_content = next(
(m.content for m in reversed(manager.log) if m.role == "assistant"),
"",
)
has_runnable = any(
tooluse.is_runnable for tooluse in ToolUse.iter_from_content(last_content)
)
if not has_runnable:
break
读法:
ToolUse.iter_from_content(定义在gptme/tools/base.py:837)把回复正文里的工具调用块 全部解析出来,只要有一个is_runnable(tools/base.py:796),就再来一步;一个都没有,break。 停止条件是「结构性」的——看回复里有没有工具,而不是让模型说「我说完了」。 这就是为什么 gptme 里的 agent 会「一直干到没活可干」为止。
这一步的完整生命周期(每圈):
STEP_PRE 钩子 ──▶ step() 生成+执行 ──▶ 把 yield 的消息 append 回历史
│
┌── ──────────────────────────────────┤
▼ ▼ ▼ ▼
用户角色消息 有 DECLINED_CONTENT 后台自动命名 step_count+1
且是/命令? (用户拒绝执行)? (前几条回复时) 达到 MAX_STEPS?
→ return → break 回用户 → 起后台线程 → 追加系统消息 break
│
▼
检查 has_runnable(见上)
有→再来一圈 / 无→break
最后:触发 TURN_POST 钩子
逐个说清这几道「岔口」:
-
用户拒绝执行(
DECLINED_CONTENT)。 若某条工具在确认环节被用户回了「n」,step()会产出一条 内容为DECLINED_CONTENT的消息;循环检测到就break,把控制 权还给用户,行为等同于 Ctrl+C (chat.py:393-396)。 -
后台自动命名。 在会话最初几条 assistant 回复期间(
1 <= assistant_count <= MAX_ASSISTANT_MSGS_FOR_NAMING),起一个 daemon 线程调try_auto_name给会话取个显示名, 用copy.deepcopy拷贝消息避免与主线程竞争(chat.py:398-416)。放后台是为了不阻塞主循环。 -
步数上限
GPTME_MAX_STEPS。 从环境变量读一个整数上限(解析失败则忽略并告警,chat.py:344-353);每步step_count += 1,到顶就追加一条系统消息并break(chat.py:418-426)。 这是防「agent 无限自嗨」的安全阀。 -
TURN_POST钩子。 整个步骤循环结束后触发一次(chat.py:441-446)——预提交检查、自动 commit 等都挂在这里(不再写死在循环里)。
3.4 单步内幕:step() 怎么「生成一次 + 执行一次」
它要解决的小问题: 把「组装上下文 → 调模型 → 跑工具」这三件事封成一个可复用、可被 list()
一次性收集的生成器。
step() 是个 Generator[Message, ...](chat.py:520)。核心四步:
-
组装/裁剪上下文。
msgs = prepare_messages(log.messages, workspace)(chat.py:546)——必要时做 上下文压缩(细节见 04)。 -
结构化工具模式下才传 tools。 只有
tool_format == "tool"(即用模型原生 function-calling)时, 才把可运行工具列表传给模型;markdown/XML 模式下tools保持None,靠提示词告诉模型怎么写工具块 (chat.py:548-550):# gptme/chat.py:548tools = Noneif tool_format == "tool":tools = [t for t in get_tools() if t.is_runnable] -
生成回复。
reply(msgs, model, stream, tools, ...)(chat.py:554)拿到msg_response; 随后触发GENERATION_POST钩子(如 TTS 朗读,chat.py:566-573)。 -
产出回复、再执行工具。 先
yield msg_response,紧接着yield from execute_msg(msg_response, log=log, workspace=workspace)(chat.py:576-578)——execute_msg(gptme/tools/__init__.py:316)解析并逐个跑掉回复里的工具,把每条工具输出也yield出来。外层list(step(...))因此一次性收齐「回复 + 全部工具输出」,再统一 append 回历史。
关键顺序:先 yield 回复,后 yield 工具输出。这保证历史里「assistant 说要干嘛」永远排在 「system 工具结果」前面,下一步
step()看到的是一段自洽、有因有果的对话。
3.5 中断与崩溃恢复:两处 KeyboardInterrupt、一个「续跑」判断
它要解决的小问题: 用户随时可能 Ctrl+C;进程也可能崩了重启、留下一份「最后一条是用户消息、 还没来得及回复」的历史。循环要都能优雅接住。
两处捕获 Ctrl+C,语义不同:
| 位置 | 何时 | 处理 |
|---|---|---|
_run_chat_loop 外层(chat.py:301) | 空闲/等输入时按 Ctrl+C | 追加 INTERRUPT_CONTENT、清空 prompt_queue、continue |
_process_message_conversation 内(chat.py:377) | 生成/跑工具中途按 Ctrl+C | 追加 INTERRUPT_CONTENT、break 出步骤循环回用户 |
崩溃恢复 / 「直接续跑」: 队列空且交互时,_get_user_input() 先问
_should_prompt_for_input(log)(chat.py:449)。若历史最后一条是用户消息(说明上次没答完),
它返回 None 表示「别问了,直接替用户生成」——于是 _run_chat_loop 不加新消息,直接调
_process_message_conversation 把回复补上(chat.py:237-248)。_should_prompt_for_input 还兼顾
「历史为空/最后是 assistant/刚被中断或拒绝/最后一条 pinned/整个历史没有任何 user」等边界
(chat.py:474-480)。
3.6 输出格式栈:让 subagent 内联调用不污染父会话
它要解决的小问题: gptme 的 subagent 是内联实现的——父 chat() 会直接再调一次 chat()。
如果子调用把全局输出格式无脑重置成 "text",父会话的 JSON 模式就被冲掉了。
解法是一个「保存—设置—还原」的栈式约定(不是显式栈,而是靠每层函数各自的局部变量 +
finally 天然形成后进先出):
父 chat() : _prev = "json"; set("json") … 调子 chat() … finally set(_prev="json")
└─ 子 chat(): _prev = "json"; set("text") … finally set(_prev="json")
每层进来先 _prev_output_format = get_output_format(),退出时在 finally 里 set_output_format( _prev_output_format)(chat.py:87-89、chat.py:170-173)。子调用退出时还原的是「它进来时看到的父格式」,
而不是硬编码的 "text"——父会话的格式因此毫发无损。这段代码注释把动机说得很直白:
「so nested chat() calls (inline subagents) don't clobber the parent's JSON mode when they exit.」
3.7 循环级钩子:让「继续/停止」也能被插件左右
它要解决的小问题: 有时希望循环在一轮结束后自动再喂一句(如 auto-reply、自评自纠),而不是 停下来等人。这不该写死在循环里。
_run_chat_loop 每轮末尾触发 HookType.LOOP_CONTINUE(chat.py:281-299):钩子若返回消息,就把它们
追加进 prompt_queue(受 MAX_PROMPT_QUEUE_SIZE 上限保护,超了丢弃并告警),然后 continue——
下一圈自然会把这些消息当输入处理。「要不要继续」于是变成一个可插拔的决策点,而循环主体只管
「队列里有就接着跑」。钩子系统本身见 05。
4. 巧妙之处(可借鉴的技术)
-
停止条件「结构化」而非「声明式」。 不靠模型输出特殊结束 token,而是解析回复里「还有没有可运行 的工具」来决定继续/停止(
chat.py:428-437)。模型只要「不再写工具块」就等于说完了——鲁棒、 与具体模型无关。 -
一条队列统一两种输入源。 初始 prompt、钩子回灌、外部持久排队,全部
pop(0)自同一个prompt_queue;队列空了才问用户(chat.py:146、chat.py:194)。控制流因此只有一条主干。 -
list(step(...))把「流式生成器」一次性物化。step()是生成器(能流式),但步骤循环用list(...)收齐再统一 append(chat.py:367)——既享受流式渲染,又保证「回复+工具输出」作为一个 原子单位落进历史。 -
自动命名放 daemon 线程 + deepcopy。 取名是锦上添花,绝不能卡住主循环,也不能与主线程共享可变 历史,于是后台线程 + 深拷贝(
chat.py:407-416)。 -
finally还原输出格式而非硬重置。 用「保存调用者值」支持任意深度的内联 subagent 嵌套 (chat.py:170-173),这是把「全局状态」做成「可重入」的经典手法。
5. 边界与局限
-
本章不覆盖工具解析/执行的内部。
execute_msg/ToolUse.iter_from_content/is_runnable这里只当黑盒用,真正怎么解析 markdown/XML/结构化工具块、怎么确认与沙箱,见 02。 -
上下文如何裁剪不在这里。
prepare_messages(chat.py:546)只被调用,压缩策略见 04。 -
钩子只被「触发」,不在此定义。 本章列出触发点(SESSION_START/TURN_PRE/STEP_PRE/GENERATION_POST/ TURN_POST/LOOP_CONTINUE/SESSION_END),但它们的注册与实现见 05;类型定义在
gptme/hooks/types.py:66。 -
无限循环的唯一硬护栏是
GPTME_MAX_STEPS。 不设该环境变量时,理论上只要模型持续产出工具, 步骤循环就会一直跑(chat.py:344、chat.py:418-426)——交互模式下靠用户 Ctrl+C 兜底, 自动化场景务必设上限。 -
prompt_user/prompt_input标了# pragma: no cover。 这两处真正读终端输入的函数 (chat.py:584、chat.py:606)不进单测,行为要以真实运行为准。
6. 横向对比
同为 coding agent,主循环的「停止判定」各有取舍:
- gptme:看回复里有没有可运行工具来决定继续/停止(
chat.py:428-437),模型无需显式声明结束。 - 许多基于原生 function-calling 的 agent:靠「模型这次没发起 tool_call」作为停止信号——本质相近, 但 gptme 在 markdown/XML 工具模式下也统一成同一套判定,不依赖模型原生工具协议。
更细的 shelf 级对比见 ai-agent-reference 总库文档的「agent 主循环」原理条目。
7. 代码地图(导航索引)
全部符号位于克隆内
gptme/chat.py(除标注外),as-ofsourceCommit。用符号名grep比行号更抗漂移。
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 会话入口 / 初始化 / 收尾 | gptme/chat.py | chat |
| 输出格式保存与还原(subagent 安全) | gptme/chat.py | chat(_prev_output_format) |
| 轮次循环 / 队列与用户输入统一 | gptme/chat.py | _run_chat_loop |
| 持久排队 prompt 并入内存队列 | gptme/chat.py | _drain_external_prompt_queue |
| 步骤循环 / 多步自主推进 / 判停 | gptme/chat.py | _process_message_conversation |
| 单步:生成 + 执行工具 | gptme/chat.py | step |
| 是否该问用户(崩溃恢复) | gptme/chat.py | _should_prompt_for_input / _get_user_input |
| 读终端输入 | gptme/chat.py | prompt_user / prompt_input |
| 判「回复里还有没有工具」 | gptme/tools/base.py | ToolUse.iter_from_content / ToolUse.is_runnable |
| 执行一条消息里的工具 | gptme/tools/__init__.py | execute_msg |
| 上下文组装/裁剪 | gptme/logmanager/manager.py | prepare_messages |
| 历史装载与追加 | gptme/logmanager/manager.py | LogManager.load / LogManager.append |
| 钩子类型定义 | gptme/hooks/types.py | HookType |
| 钩子触发 | gptme/hooks/registry.py | trigger_hook |
| 循环相关常量 | gptme/constants.py | MAX_PROMPT_QUEUE_SIZE / DECLINED_CONTENT / INTERRUPT_CONTENT |