为什么长时间运行的智能体需要状态机：领导者、执行者和验证者的角色
单个智能体在处理长任务时会出现的四种故障，以及由确定性引擎驱动的领导者、执行者和验证者团队如何解决这些问题，还有何时应跳过该机制。
来源: MiniMax Agent Team Best Practice: An AI Agent Is a While Loop. A Reliable One Is a State Machine.
从聊天机器人发展到智能代理需要的是while循环，而非某种研究突破。这确实是事实，但也是大多数解释的终点以及生产环境中诸多问题的起点。更有意义的问题是：为何在处理时长超过十分钟的任务时，这种循环会悄然失效，以及何种结构能够修复它。MiniMax就此主题发布了两份详尽的参考资料：一是开源的Mini-Agent教程，其中展示了基础循环结构；二是他们智能代理团队的技术报告，阐述了大规模应用中出现的故障以及为此开发的运行时解决方案。这两份资料都构成了本文的核心基础，其设计模式也可应用于任何技术栈。
一个循环，从头到尾
将一个智能体简化后，仍会剩下四个部分：模型、消息列表、工具注册表以及围绕这些组件的循环结构。Mini-Agent的实现方式如下：首先检查历史记录是否接近上下文容量限制，若是则对其进行压缩；接着将消息和工具定义发送给模型。如果模型不需要任何工具，循环就会结束；否则它会运行所需的工具，将其输出添加到消息中，然后重复该过程。以下是为所有兼容OpenAI的端点准备的完整版本，其中包含了两个实际工具以及用于总结内容的机制：
# naive_loop.py -- a single-agent tool loop,
# Mini-Agent style: summarize if needed ->
# llm.generate(messages, tools) -> execute
# tool_calls -> append results -> repeat.
import json
import subprocess
from openai import OpenAI
client = OpenAI()
MODEL = "gpt-4o-mini"
MAX_HISTORY = 40 # crude context budget
def read_file(path: str) -> str:
with open(path) as f:
return f.read()[:4000]
def run_cmd(cmd: str) -> str:
p = subprocess.run(
cmd, shell=True, timeout=30,
capture_output=True, text=True)
out = p.stdout + p.stderr
return f"exit={p.returncode}\n{out[-2000:]}"
TOOLS = {"read_file": read_file,
"run_cmd": run_cmd}
def schema(name, desc, arg):
return {
"type": "function",
"function": {
"name": name,
"description": desc,
"parameters": {
"type": "object",
"properties": {
arg: {"type": "string"},
},
"required": [arg],
},
},
}
SCHEMAS = [
schema("read_file",
"Read a local text file.", "path"),
schema("run_cmd",
"Run a shell command.", "cmd"),
]
def summarize(msgs):
# Mini-Agent compresses old turns with an
# LLM summary call when the context nears
# its limit; hard truncation is the v0.
if len(msgs) <= MAX_HISTORY:
return msgs
return [msgs[0]] + msgs[-MAX_HISTORY:]
def run(task: str, max_steps: int = 20):
msgs = [{"role": "user", "content": task}]
for _ in range(max_steps):
msgs = summarize(msgs)
r = client.chat.completions.create(
model=MODEL,
messages=msgs,
tools=SCHEMAS,
)
msg = r.choices[0].message
msgs.append(msg)
if not msg.tool_calls:
return msg.content # model says done
for call in msg.tool_calls:
fn = TOOLS[call.function.name]
args = json.loads(
call.function.arguments)
try:
out = fn(**args)
except Exception as e:
out = f"tool error: {e}"
msgs.append({
"role": "tool",
"tool_call_id": call.id,
"content": out,
})
return "hit max_steps -- gave up"
if __name__ == "__main__":
print(run("Count the .py files here, then "
"summarize naive_loop.py."))
该循环有两个退出条件，且两者都至关重要。成功由模型自行判断，表现为不再进行工具调用；而安全则通过一个简单的计数器，即步骤限制来体现。正是这种由模型独自决定工作何时完成的不对称性，最先成为问题的根源。
单次循环在处理长时间任务时的局限
将同样的循环应用于两小时长的任务而非两分钟的任务时，故障便开始出现。MiniMax团队记录下了他们不断遇到的问题，但这些都不是漏洞，而是单一上下文环境下的固有特性，因此增加处理时间或优化提示语都无法解决这些问题。
首先是上下文焦虑问题：在未被要求时，代理就会停止工作。在完成七个子任务后，它只完成了三个，声称已经处理完1、2、3号任务，然后询问是否继续。模型对于长任务何时能结束的判断往往不够准确，而直接训练这种判断能力在成本和算法资源上都非常高昂。简单的循环结构则将停止决策的任务交给了最不可靠的部分来处理。
其次是性能下降问题：随着上下文规模的增加，代理的智能程度会降低。在第五分钟时它表现还很出色，但到第六十五分钟时就像一个心不在焉的实习生。一旦出现偏差，比如研究任务混入了营销文案中，就会影响之后的所有输出，因为后续的生成结果都是基于这种被污染的历史数据来进行的。
第三，由于法官同时也是参赛者，因此没有任何机制来检查工作质量。那个生成最终结果的单一智能体会在产生错误的同一环境中对自己进行评分。这并非不诚实，而是正在检查模型自己刚刚构建的场景。
第四是沉默问题。聊天式界面会让用户期望即使在处理需要一小时的任务时也能在几秒内得到回应，但隐藏在模型调用中的循环无法表明它已收到请求并会反馈结果。根据MiniMax的统计，“为什么我的智能体没有回复”这一问题引发的用户反馈比其他任何问题都多。
解决方案：三种角色与确定性引擎
MiniMax以结构化方式给出答案。三种角色由普通的确定性代码协调，而非依赖巧妙的提示，该代码被称为团队引擎。一个任务生命周期对应一个会话，引擎会推动每个会话经历生成、验证和完成三种状态，一旦验证失败就会再次启动生成器。
- 领导者将用户的请求拆解为结构化的计划，决定哪些部分同时执行，在几秒内向用户回复，其余部分则在后台处理，最后合并各部分的结果。
- 工作节点负责处理子任务，每个节点都在独立的上下文中运行，拥有自己的工具、内存及所需的输出格式。这种分离很重要，因为工作节点仅能看到一个子任务，而无法看到两小时积累的冗余数据。
该引擎的核心部分小到可以直接写出来。此版本无需API密钥，因为虚拟角色可以替代LLM调用，且无需修改状态机即可更换：
# team_engine.py -- the loop promoted to a
# state machine. One task lifecycle is one
# Session; deterministic code owns every
# transition: producing -> verifying -> done.
from dataclasses import dataclass, field
PRODUCING = "producing"
VERIFYING = "verifying"
DONE = "done"
FAILED = "failed"
@dataclass
class Session:
goal: str
state: str = PRODUCING
attempts: int = 0
artifact: str = ""
reviews: list = field(default_factory=list)
def run_session(s, worker, verifier,
max_retries: int = 3):
while s.state not in (DONE, FAILED):
if s.state == PRODUCING:
s.attempts += 1
s.artifact = worker(
s.goal, s.reviews)
s.state = VERIFYING
elif s.state == VERIFYING:
ok, report = verifier(
s.goal, s.artifact)
s.reviews.append(report)
if ok:
s.state = DONE
elif s.attempts >= max_retries:
s.state = FAILED
else:
# Reject: wake the worker with
# the review log in context.
s.state = PRODUCING
return s
def run_team(goals, worker, verifier):
# Leader's job: split the goal, run the
# sessions (a thread pool in real use),
# then merge N artifacts into 1 result.
done, failed = [], []
for g in goals:
s = run_session(Session(g),
worker, verifier)
if s.state == DONE:
done.append(s.artifact)
else:
failed.append(s.goal)
return done, failed
if __name__ == "__main__":
# Stub roles so the engine runs anywhere;
# swap in LLM calls for the real thing.
def worker(goal, reviews):
v = len(reviews) + 1
return f"draft v{v}: {goal}"
def verifier(goal, artifact):
ok = "v2" in artifact
return ok, f"review {artifact!r}: {ok}"
done, failed = run_team(
["intro", "benchmarks", "faq"],
worker, verifier)
print("delivered:", done)
print("failed:", failed)
运行后，每个会话的第一个版本会被拒绝，工作进程会在其上下文中携带审核日志继续运行，而第二个版本则可以通过。将其与简单的循环对比，可见几处差异。
- 停止机制：单个循环会在模型停止发出工具调用（模糊处理）或达到步数上限时结束，而该引擎则通过代码中定义的明确完成状态和失败状态以及重试次数限制来控制流程。
- 验证方式：在循环中，生成结果的同一上下文会负责对其进行验证，而该团队则使用基于命令的独立上下文中的对抗性验证器来进行验证。
- 并行性：循环是在单一顺序上下文中运行的，而该团队则在多个并行会话中运行工作进程，由领导者将多个结果合并为一个。
- 持久性：循环会在总结之前保留消息列表，而该团队则会保存会话状态、审核记录、生成结果以及每个智能体的内存数据。
Leader并不局限于那种生成子任务并返回文本的单一调用方式。在MiniMax的设计中，它能读取引擎的实时状态，并可以向正在处理或验证中的会话发送额外提示，从而实现主动推送与按需查询两种功能。这同时也解决了沉默问题：Leader会在几秒内给出响应，随后用户就能在开始、阻塞、需要决策以及完成等各个节点收到相应报告。
验证器的定位
只有当判定结果来自模型外部时，自我评分的缺陷才会消失。对于代码而言，MiniMax会通过专用框架来控制其执行过程。代码存储在分支中，在沙箱环境中运行，修改以差异形式呈现，测试可以重复进行，审查会留下记录，失败情况也可以重新模拟。最重要的是：停止条件必须与某种可从外部确定且可观察的因素相关联，比如测试、代码检查工具以及构建过程，而绝不能依赖于模型自身判断是否完成的工作状态。
# grounded_verifier.py -- evidence over vibes.
# The verdict comes from exit codes and logs,
# never from the model's self-assessment.
import subprocess
import sys
PY = sys.executable
# pytest exit codes: 0 = all passed,
# 1 = failures, 5 = no tests collected
# (acceptable for a docs-only change).
PYTEST_OK = (0, 5)
CHECKS = [
("diff", ["git", "diff", "--stat"], (0,)),
("build", [PY, "-m", "compileall",
"-q", "."], (0,)),
("tests", [PY, "-m", "pytest", "-q",
"--maxfail", "1"], PYTEST_OK),
]
def run_check(name, cmd, ok_codes, cwd):
p = subprocess.run(
cmd, cwd=cwd,
capture_output=True, text=True)
tail = (p.stdout + p.stderr)[-2000:]
return {
"check": name,
"cmd": " ".join(cmd),
"exit_code": p.returncode,
"ok": p.returncode in ok_codes,
"output_tail": tail,
}
def verify(repo: str):
evidence = []
for name, cmd, ok_codes in CHECKS:
e = run_check(name, cmd, ok_codes, repo)
evidence.append(e)
if not e["ok"]:
# Reject with logs attached: the
# worker retries against real
# errors, not "please improve".
return False, evidence
return True, evidence
if __name__ == "__main__":
ok, ev = verify(".")
for e in ev:
print(f"{e['check']:>6} exit "
f"{e['exit_code']} ok={e['ok']}")
print("verdict:", "pass" if ok else "fail")
证据列表同时充当审查记录，属于可重复追溯、可问责的痕迹，用以区分真正可交付的结果与仅看似合理的成果。MiniMax进一步细分了审查环节：测试人员通过运行测试、压缩故障日志并添加最简化的复现步骤来生成外部证据；而审查人员则提出不同的问题，不是关注测试是否通过，而是判断代码是否应如此修改，他们会审视抽象边界、新出现的依赖关系、被扩大的权限以及泄露到日志中的敏感信息。审查人员可以与通用型、安全型及领域型等不同专业的审查者并行工作。
组建团队的成本及何时无需组建
团队并非免费的并行处理方式。一旦各成员开始协作，就会出现三种成本，而更大的上下文窗口也无法消除这些成本。
交接成本：相同的信息需要在不同处理节点之间重新整理，因此研究页面会变成写作者的概要，而该概要又会作为格式检查的输入。解决办法是传递可读文件，即通过路径传递附带简短摘要的文件，而非原始的上下文数据。
共享成本：每一轮中，所有工作者都需要为每个全局共享的标记付费。合理的架构能让每个处理节点独立管理内存，允许它们通过命令行中断相互沟通，并提供一个共享白板，供处理节点在需要时查看。
第三种成本是聚合成本。收集十个部分草稿很简单，但要将它们整合为包含一致事实、引用和风格的单一文件，却是负责人的最艰巨任务，这需要一个合并步骤，而非增加更多人手。
共识也有其代价。智能体之间的自由辩论所消耗的代币可能是独立自我修正所需代币的数倍，却仍无法提升准确性。其价值在于结构：有限的会话时长、明确的输出结果，以及有权拒绝结果的验证机制。
当任务规模较小、可在本地完成且验证成本较低时，无需使用团队协作，因为单次循环的成本更低。而在工作可清晰划分、错误代价高昂且需要独立验证来证明模型无法自圆其说时，才应使用团队协作。一个好的判断标准是：只有当某个角色能覆盖其他角色无法处理的故障模式时才添加该角色，否则协作带来的额外开销毫无意义。
应首先构建什么
从最简单且可信赖的状态机开始设计。每个会话都需设定目标、生成结果物、记录审查日志以及设定重试次数，同时每一步状态转换都必须有明确说明。将验证器置于由测试、代码检查工具、构建过程或类似自动化检查机制所提供的结果支撑之下。并行处理机制可稍后添加：并行性属于优化手段，而正确性边界则是架构层面的要求。需让领导节点的合并步骤保持可见，让各环节的输出结果物可供检查，并记录每个会话结束的原因。
最终产生的并非自主思维，而是一个受控的工作流程：模型提出建议，工具负责执行，普通代码则判断任务是否完成。在采用框架之前，应先手动构建基础循环结构。长期任务往往会因焦虑、偏离目标、自我评估问题以及沉默状态而破坏其结构，而这些问题无论怎样提示都无法解决。应将多智能体系统视为一个具备状态管理、事件处理、权限控制、可观测性以及可恢复机制的运行环境，而非提示语中的角色扮演。在组建团队之前，就要做好任务交接、数据共享与整合的规划，绝不能因为一个拼写错误就匆忙组建团队。
相关阅读
- 实用笔记：Claude Code Agent Teams——构建团队的实用指南 — 详细介绍了如何运用合同、检查机制以及即插即用的代码模块来构建采用该模式的团队。
- 为何智能体日志无法满足第12条要求：决策记录、因果关系与来源追溯 — 追踪记录并不能等同于责任认定。本文介绍如何利用Semantica将智能体的决策记录为具有因果关联、哈希链式来源追溯以及确定性策略规则的持久化对象。