实用笔记：MiniMax智能体团队最佳实践——AI智能体即一个while循环。
《实用笔记》操作指南：MiniMax智能体团队最佳实践——AI智能体即循环结构；为采用该模式的团队提供的代码模板、检查项及可直接插入的代码片段。
可将此内容视为《MiniMax Agent Team Best Practice：AI智能体即循环结构，可靠的智能体则是状态机》中理念的面向操作人员的重构版本：包含清晰的阶段划分、有序的代码模块以及能在交接时保留的恢复说明。 将“概览”阶段视为可度量的基准最为有效。在扩大范围之前，先记录一份最佳案例、一个故障实例以及回滚说明。 应将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，杜绝默许的半完成状态。
# 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."))
# 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)
# 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")
操作检查清单
在处理操作检查清单阶段时，首先需明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。
应优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。
在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次计费相同的LLM调用。
锁定依赖版本的编号，并记录用于运行演示的镜像摘要。可重复性比经验知识更为重要。
将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，拒绝默许的半完成状态。
在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次计费相同的LLM调用。
在推广该技术栈之前，应先冻结版本，为关键路径生成标准记录，并明确回滚步骤。共享环境需要设置速率限制、租户验证机制，以及明确的密钥轮换负责人。与其展示花哨的一次性演示，不如注重扎实的可靠性。
e49728a65412的批量处理说明：不要将提供商密钥放入代码仓库，为每个会话设置令牌使用上限，并将记录存储在评估测试用例旁边，以便后续模型更换时保持可比性。
针对强化安全性的第0阶段，在修改代码之前需明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任模块，而非复杂的流程链。
强化措施细节 0/726：为该笔记记录运行时间、错误类型以及代币消耗情况，然后依据固定的问题清单而非个人经验来判断是否保留该变更。
在处理强化措施笔记的第一阶段时，首先写下合约的必要输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改更加规范。 在功能结果旁记录时间以及代币或查询成本。提前明确成本有助于避免从演示环境过渡到共享环境时出现意外费用。
强化措施细节 1/726：为该笔记记录运行时间、错误类型以及代币消耗情况，然后依据固定的问题清单而非个人经验来判断是否保留该变更。
在将加固措施视为可测量的表面时，第二阶段的效果最佳。在扩大范围之前，需记录一份理想的运行日志、一个故障案例以及回滚说明。同时记录正常流程与恢复流程。重试机制、人工审核环节以及错误处理都属于产品本身的功能，而非后续需要补充的内容。
加固细节2/726：针对该措施需测量处理时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
对于加固措施的三阶段，应在修改代码之前明确输入参数、负责该步骤的人员以及完成标准。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。应将此阶段视为输入参数与验证后输出结果之间的契约，为相关成果命名、定义成功检测标准，并拒绝默许的半完成状态。
强化措施细节 3/726：为该记录测量运行时间、错误类型以及代币消耗情况，然后依据固定的问题清单而非个人经验来判断是否保留该变更。
在处理强化措施笔记的第4阶段时，首先写下合约的必要输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改更加规范。 应将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。
强化措施细节 4/726：为该记录测量运行时间、错误类型以及代币消耗情况，然后依据固定的问题清单而非个人经验来判断是否保留该变更。
在将加固措施视为可测量的表面时，第5阶段的效果最佳。在扩大范围之前，需记录一份理想的测试结果、一个故障案例以及回滚说明。 相比庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障原因应能指向单一责任主体，而非复杂的流程链。
加固细节5/726：需为该措施测量执行时间、错误类型以及令牌消耗情况，然后依据固定的问题清单而非个人经验来判断是否保留该变更。
在实施强化措施的第6阶段，应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及代币或查询成本。提前了解这些成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。
强化措施细节6/726：需测量该步骤的耗时、错误类型以及代币消耗情况，然后依据固定的评估标准而非主观判断来决定是否保留该更改。
在处理强化措施的第7阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合要求。
同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的一部分，而非后续才添加的完善措施。
强化措施细节7/726：需测量该阶段的执行时间、错误类型以及令牌消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该变更。
将强化措施的第8阶段视为一个可量化的目标面，效果最佳。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。
把这一阶段视为输入参数与验证后输出结果之间的契约。为相关文档命名，明确成功判定标准，杜绝无声的半完成状态。
强化措施细节 8/726：为该记录测量运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
在强化措施的第9阶段，应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。配置应置于应用程序代码之外，环境文件、密钥存储以及功能标志应集中存放于一个操作人员可以审核的位置，无需查看整个系统结构。
强化措施细节 9/726：为该记录测量运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
在处理强化措施的第10阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议使用小型、可测试的单元，而非冗长的脚本。当某个步骤失败时，故障应能指向单一责任点，而非复杂的流程链。
强化措施细节10/726：需测量该步骤的运行时间、错误类型以及令牌消耗情况，然后根据固定的评估标准而非主观感受来决定是否保留该修改。