从 Prompt 到 Loop:可落地的长链 Agent 工程实战指南
从 Prompt 到 Loop:可落地的长链 Agent 工程实战指南
2026 年,AI 开发的核心范式正在发生一次根本性跃迁——从"写好提示词"到"设计循环系统"。本文将带你从零理解 Loop Engineering 的核心理念,并给出可直接运行的 Python 实现代码。
目录
- 一、为什么需要 Loop Engineering
- 二、从 Prompt 到 Loop 的进化路径
- 三、Agent Loop 的核心机制
- 四、六大要素详解
- 五、五大核心模块
- 六、从零实现一个 Agent Loop(Python 完整代码)
- 七、进阶:多 Agent 协作 Loop
- 八、设计你的第一个 Loop:最小可行版本
- 九、常见陷阱与避坑指南
- 十、总结与行动清单
一、为什么需要 Loop Engineering
2026 年 6 月,Google 工程师 Addy Osmani 系统整理了一个新概念——Loop Engineering(循环工程)。Anthropic Claude Code 负责人 Boris Cherny 和开发者 Peter Steinberger 也分别在公开场合提出了类似理念。这个概念在 AI 编程社区迅速传播,因为它精准地解决了一个痛点:
Prompt Engineering 只管"怎么问",Loop Engineering 管的是"怎么让 AI 持续干活直到完成"。
你有没有遇到过这些场景?
- 让 AI 写代码,写完了不跑测试,你得手动复制结果再问它修
- 让 AI 分析数据,分析到一半上下文超了,之前的结果全丢了
- 让 AI 做一个多步骤任务,每一步都需要你手动触发下一步
- AI 的输出质量不稳定,你得反复提示"再检查一遍"
这些问题的根源不是 Prompt 写得不好,而是缺少一个自动化的循环系统。Loop Engineering 就是解决这个问题的工程方法论。
二、从 Prompt 到 Loop 的进化路径
过去两年,AI 开发的重心经历了四个阶段:
| 阶段 | 关注点 | 典型问题 | 代表工具 |
|---|---|---|---|
| Prompt | 怎么问 AI | "为什么 AI 不听我的?" | ChatGPT、Few-shot |
| Context | 给 AI 什么信息 | "上下文太长怎么办?" | RAG、长上下文窗口 |
| Harness | 如何组织 AI 的能力 | "怎么让 AI 用工具?" | LangChain、Function Calling |
| Loop | 如何让 AI 持续创造结果 | "怎么让 AI 自己跑完?" | Claude Code、Codex、Cursor |
每一个阶段都不是替代前一个,而是在前者基础上增加了一层抽象。Loop 是目前最高的抽象层级——它把 Prompt、Context、Harness 全部封装进一个自动运行的循环里。
核心转变:你不再是"提示词工程师",而是"循环系统设计师"。你定义目标和规则,Agent 自己决定怎么执行。
三、Agent Loop 的核心机制
3.1 Perceive - Reason - Act - Observe
每个 Agent 在执行任务时,天然内置了一个"内循环":
+---------------------------------------------+
| |
| +-----------+ +-----------+ |
| | Perceive |----->| Reason | |
| | 感知 | | 推理 | |
| +-----------+ +-----+-----+ |
| ^ | |
| | v |
| +-----+------+ +-----------+ |
| | Observe |<-----| Act | |
| | 观察 | | 行动 | |
| +------------+ +-----------+ |
| |
+---------------------------------------------+
与传统 while 循环的本质区别:
传统 while 循环执行的是确定性指令序列——每次迭代做的事情完全一样。
Agent Loop 执行的是目标导向的推理序列——每次迭代的具体行动由 Agent 根据上下文自主决策。你写的是 while not goal_achieved,但循环体里的内容是 Agent 自己"想"出来的。
# 传统循环:确定性
while not done:
process_item(queue.get()) # 每次执行完全相同的逻辑
# Agent Loop:目标导向
while not goal_achieved:
observation = perceive(environment)
action = agent.reason(observation, goal) # Agent 自主决策下一步
result = execute(action)
environment.update(result)
3.2 内循环 vs 外循环
理解 Loop Engineering 需要区分两个层次:
- 内循环(Inner Loop):Agent 单次执行中的 Perceive - Reason - Act - Observe。这是 LLM 调用本身自带的。
- 外循环(Outer Loop):你设计的工程系统——包含目标设定、评估机制、记忆持久化、停止条件。这是 Loop Engineering 的核心。
大多数人在做的是优化内循环(更好的 Prompt、更好的 Context)。Loop Engineering 关注的是外循环的设计。
四、六大要素详解
一个完整的 Agent Loop 由六个要素构成。缺少任何一个,循环都无法可靠运行。
4.1 目标设定(Goal)
反面示例:
"帮我优化代码"
这个目标太模糊。Agent 不知道优化到什么程度算完成,会无限迭代。
正面示例:
"将
process_data()函数的测试覆盖率从 72% 提升到 95%,并且所有现有测试仍然通过。"
目标必须满足三个条件:
- 可验证:Agent 自己能判断是否达成(运行测试套件,检查覆盖率数字)
- 具体:有明确的衡量标准(95%,不是"更高")
- 边界清晰:知道什么时候该停(达到了就停,不需要"再好一点")
落地建议:为每个 Goal 写一个验证函数。如果 Agent 自己跑不了验证,这个 Goal 就不合格。
def validate_goal(result: dict) -> bool:
"""Goal 验证函数,Agent 每轮迭代后调用"""
coverage = result.get("test_coverage", 0)
all_passing = result.get("all_tests_pass", False)
return coverage >= 95.0 and all_passing
4.2 上下文管理(Context)
Loop 中的上下文不是静态的,而是随迭代动态更新。这是长链 Agent 最容易出问题的地方。
核心策略:三层上下文
| 层级 | 内容 | 生命周期 | 存储位置 |
|---|---|---|---|
| 工作记忆 | 当前迭代的输入输出 | 单次迭代 | LLM 上下文窗口 |
| 短期记忆 | 最近 N 次迭代的摘要 | 跨迭代,有容量上限 | 本地文件 / 内存 |
| 长期记忆 | 关键决策、已完成任务、学到的模式 | 永久 | 磁盘文件 / 数据库 |
class ContextManager:
def __init__(self, max_short_term=10):
self.working_memory = "" # 当前迭代
self.short_term = [] # 最近 N 次摘要
self.long_term_path = "memory.md"
self.max_short_term = max_short_term
def update(self, iteration_result: str):
"""每轮迭代后更新上下文"""
self.working_memory = iteration_result
summary = self._compress(iteration_result)
self.short_term.append(summary)
if len(self.short_term) > self.max_short_term:
self._persist_to_long_term(self.short_term.pop(0))
def _compress(self, text: str) -> str:
"""用 LLM 压缩为 1-2 句摘要"""
return text[:200] + "..."
def _persist_to_long_term(self, summary: str):
with open(self.long_term_path, "a") as f:
f.write(f"\n[{datetime.now()}] {summary}")
def build_prompt_context(self) -> str:
"""构建注入 LLM 的上下文"""
long_term = ""
if os.path.exists(self.long_term_path):
with open(self.long_term_path) as f:
long_term = f.read()[-2000:]
return f"""
## 长期记忆
{long_term}
## 近期迭代摘要
{chr(10).join(self.short_term[-5:])}
## 当前状态
{self.working_memory}
"""
关键原则:模型在每次运行之间会忘掉一切。记忆必须存在磁盘上,而不是依赖上下文窗口。智能体会忘,但代码仓库不会。
4.3 工具接入(Tools / Plugins & Connectors)
Agent 需要读取实际环境信息,也要把结果写回环境。典型的工具接入:
TOOLS = {
# 代码相关
"read_file": lambda path: open(path).read(),
"write_file": lambda path, content: open(path, 'w').write(content),
"run_command": lambda cmd: subprocess.run(cmd, shell=True, capture_output=True, text=True),
"run_tests": lambda: subprocess.run(["pytest", "-v"], capture_output=True, text=True),
# Git 相关
"git_diff": lambda: subprocess.run(["git", "diff"], capture_output=True, text=True).stdout,
"git_commit": lambda msg: subprocess.run(["git", "commit", "-am", msg], capture_output=True, text=True),
# 外部服务
"search_web": lambda q: requests.get(f"https://api.search.brave.com/res/v1/web/search?q={q}").json(),
}
设计原则:
- 最小权限:第一个 Loop 只给读权限,不给写权限
- 操作可逆:每次写操作前保存快照,失败时能回滚
- 日志完整:每个工具调用的输入输出全部记录
4.4 产出评估(Evaluation)
这是 Loop Engineering 与简单循环脚本的本质区别。没有评估机制,循环只会无限执行,无法收敛。
三种评估方式:
class Evaluator:
def evaluate(self, result: dict) -> tuple:
"""返回 (是否通过, 反馈信息)"""
# 方式一:客观评估 -- 运行测试
test_result = subprocess.run(["pytest", "-x"], capture_output=True, text=True)
if test_result.returncode != 0:
return False, f"测试失败:\n{test_result.stdout[-500:]}"
# 方式二:主观评估 -- LLM 打分
score = self._llm_score(result["output"])
if score < 7:
return False, f"质量评分 {score}/10,需要改进"
# 方式三:混合评估 -- 对比 diff 范围
diff = subprocess.run(["git", "diff", "--stat"], capture_output=True, text=True).stdout
changed_lines = self._count_changed_lines(diff)
if changed_lines > 500:
return False, f"变更范围过大({changed_lines} 行),请拆分"
return True, "评估通过"
4.5 停止条件(Stop Condition)
停止条件的设计直接影响 Token 消耗和结果质量的平衡。
class StopCondition:
def __init__(self, max_iterations=10, max_tokens=100000, max_time=600):
self.max_iterations = max_iterations
self.max_tokens = max_tokens
self.max_time = max_time
self.start_time = time.time()
self.iteration = 0
self.total_tokens = 0
def should_stop(self, goal_achieved: bool, tokens_used: int) -> tuple:
self.iteration += 1
self.total_tokens += tokens_used
if goal_achieved:
return True, "目标达成"
if self.iteration >= self.max_iterations:
return True, f"达到最大迭代次数 ({self.max_iterations})"
if self.total_tokens >= self.max_tokens:
return True, f"达到 Token 上限 ({self.max_tokens})"
if time.time() - self.start_time > self.max_time:
return True, f"超时 ({self.max_time}s)"
return False, ""
最佳实践:设置一个"优雅退出"机制。即使达到上限没完成,也要把当前进度保存到记忆文件,下次可以从断点继续。
4.6 记忆持久化(Memory)
一个 Markdown 文件、一个 Linear 看板、甚至一个 JSON 文件——任何能活在单次对话之外的地方都行。
# 最简实现:TODO.md 文件
def save_progress(task: str, status: str, next_steps: list):
with open("TODO.md", "a") as f:
f.write(f"\n## [{datetime.now().strftime('%Y-%m-%d %H:%M')}] {task}\n")
f.write(f"状态: {status}\n")
f.write(f"下一步:\n")
for step in next_steps:
f.write(f"- [ ] {step}\n")
# 恢复进度
def load_progress() -> str:
if os.path.exists("TODO.md"):
return open("TODO.md").read()
return "无历史记录"
五、五大核心模块
在 Claude Code 和 Codex 等工具中,这五个模块已经内置。理解它们有助于你设计自己的 Loop。
5.1 定时自动化(Automations)
循环的心跳。按时间表自动触发,不需人工介入。
# 示例:每天早上 8 点运行 CI 分析
schedule: "0 8 * * 1-5"
task: |
Read yesterday's CI failure logs.
Categorize failures by root cause.
Write prioritized TODO.md.
Do NOT edit source files.
5.2 工作树(Worktrees)
每个任务进入独立 worktree,Agent 在隔离环境中尝试修复。这是并行执行的基础——多个 Agent 可以同时在不同 worktree 上工作,互不干扰。
# 创建隔离工作环境
git worktree add ../task-fix-auth -b fix/auth-bug
cd ../task-fix-auth
# Agent 在这里工作,不影响主分支
5.3 Skills(技能系统)
别再每个 session 给 Agent 重新解释一遍项目背景。一个 SKILL.md 文件搞定:
# 项目技能文件
## 技术栈
- Python 3.12 + FastAPI
- pytest 测试框架
- Docker 部署
## 代码规范
- 函数命名:snake_case
- 类命名:PascalCase
- 所有公开函数必须有 docstring
## 测试要求
- 新功能必须有对应测试
- 测试覆盖率不低于 80%
5.4 插件与连接器
把 Agent 接入已有工具链——GitHub、Linear、Slack、数据库。
connectors = {
"github": GitHubConnector(token="ghp_xxx"),
"linear": LinearConnector(api_key="lin_api_xxx"),
"slack": SlackConnector(webhook_url="https://hooks.slack.com/xxx"),
"db": DatabaseConnector(connection_string="postgresql://..."),
}
5.5 子 Agent(Sub-Agents)
让开发者敢放开监管的最大底气。一个 Agent 执行,另一个 Agent 验证——就像代码审查一样。
# 执行 Agent 写代码
code = executor_agent.run("实现用户认证模块")
# 验证 Agent 独立审查
review = reviewer_agent.run(f"审查以下代码的安全性和质量:\n{code}")
if review["approved"]:
commit(code)
else:
# 把审查反馈回传给执行 Agent,继续迭代
executor_agent.run(f"根据审查意见修改:\n{review['feedback']}")
六、从零实现一个 Agent Loop(Python 完整代码)
下面是一个可直接运行的最小 Agent Loop 实现。约 150 行 Python,包含完整的六大要素。
"""
Agent Loop 最小实现
用法: python agent_loop.py "你的目标描述"
"""
import os
import json
import time
import subprocess
from datetime import datetime
from openai import OpenAI
client = OpenAI() # 或换成你的 API 端点
# ============ 配置 ============
MODEL = "gpt-4o" # 或你的模型
MAX_ITERATIONS = 15
MAX_TOKENS_BUDGET = 200_000
# ============ 工具定义 ============
TOOLS = [
{
"type": "function",
"function": {
"name": "run_command",
"description": "执行 shell 命令并返回输出",
"parameters": {
"type": "object",
"properties": {
"command": {"type": "string", "description": "要执行的命令"}
},
"required": ["command"]
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取文件内容",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "文件路径"}
},
"required": ["path"]
}
}
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "写入文件(覆盖)",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "文件路径"},
"content": {"type": "string", "description": "文件内容"}
},
"required": ["path", "content"]
}
}
},
{
"type": "function",
"function": {
"name": "goal_achieved",
"description": "声明目标已达成,结束循环",
"parameters": {
"type": "object",
"properties": {
"summary": {"type": "string", "description": "完成总结"}
},
"required": ["summary"]
}
}
},
{
"type": "function",
"function": {
"name": "request_help",
"description": "遇到无法自行解决的问题,请求人工介入",
"parameters": {
"type": "object",
"properties": {
"issue": {"type": "string", "description": "问题描述"}
},
"required": ["issue"]
}
}
}
]
# ============ 工具执行 ============
def execute_tool(name, args):
try:
if name == "run_command":
result = subprocess.run(
args["command"], shell=True,
capture_output=True, text=True, timeout=30
)
output = result.stdout + result.stderr
return output[:3000] if output else "(无输出)"
elif name == "read_file":
with open(args["path"]) as f:
return f.read()[:5000]
elif name == "write_file":
os.makedirs(os.path.dirname(args["path"]) or ".", exist_ok=True)
with open(args["path"], "w") as f:
f.write(args["content"])
return f"已写入 {args['path']}"
elif name == "goal_achieved":
return f"GOAL_ACHIEVED: {args['summary']}"
elif name == "request_help":
return f"REQUEST_HELP: {args['issue']}"
else:
return f"未知工具: {name}"
except Exception as e:
return f"工具执行错误: {e}"
# ============ 记忆系统 ============
MEMORY_FILE = "agent_memory.md"
def save_memory(entry):
with open(MEMORY_FILE, "a") as f:
f.write(f"\n[{datetime.now().strftime('%H:%M:%S')}] {entry}\n")
def load_memory():
if os.path.exists(MEMORY_FILE):
content = open(MEMORY_FILE).read()
return content[-3000:] if len(content) > 3000 else content
return ""
# ============ Agent Loop 主循环 ============
def agent_loop(goal):
print(f"\n{'='*60}")
print(f"目标: {goal}")
print(f"{'='*60}\n")
if os.path.exists(MEMORY_FILE):
os.remove(MEMORY_FILE)
messages = [
{
"role": "system",
"content": (
"你是一个自主执行任务的 Agent。你的工作方式是一个循环:\n"
"1. 分析当前状态和目标之间的差距\n"
"2. 选择最合适的工具来缩小差距\n"
"3. 执行并观察结果\n"
"4. 重复,直到目标达成\n\n"
"规则:\n"
"- 每次只做一个小步骤\n"
"- 如果遇到错误,分析原因并尝试修复\n"
"- 如果 3 次尝试都失败了,用 request_help 请求人工介入\n"
"- 目标达成时,用 goal_achieved 工具声明完成\n"
f"\n当前目标: {goal}"
)
}
]
total_tokens = 0
for iteration in range(1, MAX_ITERATIONS + 1):
print(f"\n--- 迭代 {iteration}/{MAX_ITERATIONS} ---")
memory = load_memory()
if memory and iteration > 1:
messages.append({
"role": "system",
"content": f"## 历史记忆\n{memory}"
})
response = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS,
tool_choice="auto",
)
msg = response.choices[0].message
total_tokens += response.usage.total_tokens
if not msg.tool_calls:
print(f"Agent: {msg.content}")
messages.append(msg)
messages.append({
"role": "user",
"content": "继续执行,或者如果你认为已经完成了,使用 goal_achieved 工具。"
})
continue
messages.append(msg)
for tool_call in msg.tool_calls:
fn_name = tool_call.function.name
fn_args = json.loads(tool_call.function.arguments)
print(f"调用: {fn_name}({json.dumps(fn_args, ensure_ascii=False)[:100]})")
result = execute_tool(fn_name, fn_args)
print(f"结果: {result[:200]}")
save_memory(f"{fn_name}: {result[:300]}")
if result.startswith("GOAL_ACHIEVED"):
print(f"\n目标达成! 总 Token: {total_tokens}")
return {"status": "completed", "tokens": total_tokens, "summary": result}
if result.startswith("REQUEST_HELP"):
print(f"\n需要人工介入: {result}")
return {"status": "needs_help", "tokens": total_tokens, "issue": result}
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
if total_tokens >= MAX_TOKENS_BUDGET:
print(f"\n达到 Token 上限 ({MAX_TOKENS_BUDGET})")
return {"status": "budget_exceeded", "tokens": total_tokens}
print(f"\n达到最大迭代次数 ({MAX_ITERATIONS})")
return {"status": "max_iterations", "tokens": total_tokens}
if __name__ == "__main__":
import sys
if len(sys.argv) < 2:
print('用法: python agent_loop.py "你的目标描述"')
sys.exit(1)
goal = sys.argv[1]
result = agent_loop(goal)
print(f"\n最终结果: {json.dumps(result, ensure_ascii=False, indent=2)}")
使用方式:
# 安装依赖
pip install openai
# 设置 API Key
export OPENAI_API_KEY="sk-xxx"
# 运行
python agent_loop.py "分析当前目录的 Python 项目,找出所有未使用的 import 并清理"
python agent_loop.py "将 README.md 翻译成英文,并保存为 README_EN.md"
python agent_loop.py "检查当前目录下所有 .py 文件的类型注解,为缺少注解的函数添加注解"
七、进阶:多 Agent 协作 Loop
当单个 Agent Loop 不够用时,可以引入多 Agent 协作。核心模式有两种:
7.1 执行者 + 审查者(Executor + Reviewer)
def collaborative_loop(goal):
"""双 Agent 协作:一个执行,一个审查"""
feedback = ""
for iteration in range(MAX_ITERATIONS):
# 执行者:完成任务
executor_result = executor_agent.run(goal, context=feedback)
# 审查者:独立评估
review = reviewer_agent.run(
f"审查以下结果是否达成目标 '{goal}':\n{executor_result}"
)
if review["approved"]:
return executor_result
# 审查不通过,把反馈传给执行者
feedback = review["feedback"]
return "达到最大迭代次数"
7.2 分治模式(Decompose + Parallel)
def decompose_and_conquer(goal):
"""分解任务,并行执行"""
# 规划 Agent:拆分任务
subtasks = planner_agent.run(f"将以下目标拆分为可独立执行的子任务:\n{goal}")
# 并行执行子任务
results = []
for task in subtasks:
result = agent_loop(task) # 每个子任务一个独立 Loop
results.append(result)
# 汇总 Agent:整合结果
final = synthesizer_agent.run(f"整合以下结果:\n{json.dumps(results)}")
return final
八、设计你的第一个 Loop:最小可行版本
别上来就搞全自动。 先从"只读不写"开始。
Phase 1:观察者 Loop(零风险)
# 每天早上帮你整理问题清单,不碰任何代码
loop = AgentLoop(
goal="读取昨天的 CI 日志和 GitHub Issues,按优先级写入 TODO.md",
tools=["read_file", "run_command"], # 只给读权限
write_tools=[], # 不给写权限
schedule="0 8 * * 1-5", # 工作日早 8 点
)
这个 Loop 已经很有价值:你每天到公司打开 TODO.md 就知道今天该做什么,而不是花半小时翻 CI 日志。
Phase 2:受控执行者 Loop
确认 Phase 1 稳定后,逐步放开写权限:
loop = AgentLoop(
goal="修复 TODO.md 中标记为 P0 的 bug",
tools=["read_file", "write_file", "run_command"],
constraints=[
"每次修改前先 git stash",
"单次变更不超过 50 行",
"修改后必须运行测试",
"不开 PR,只提交到本地分支",
],
)
Phase 3:全自动 Loop
等 Phase 2 跑了一个月没出大问题,再考虑全自动:
loop = AgentLoop(
goal="持续改进代码库质量",
tools=["read_file", "write_file", "run_command", "git_commit", "open_pr"],
schedule="0 */4 * * *", # 每 4 小时
safety=[
"PR 必须通过 CI 才能合并",
"每次 PR 变更不超过 200 行",
"关键文件(config, auth)需要人工审批",
],
)
九、常见陷阱与避坑指南
陷阱 1:目标太模糊
- "帮我优化这个项目" -> 失败
- "将 API 响应时间从 500ms 降到 200ms 以下" -> 成功
陷阱 2:没有评估机制
循环只执行,不检查结果 -> 无限循环。每轮迭代后必须运行验证(测试、lint、性能基准)。
陷阱 3:上下文无限增长
把所有历史对话塞进上下文 -> 上下文溢出。使用三层记忆结构:工作记忆 + 短期摘要 + 长期持久化。
陷阱 4:权限过大
第一天就给 Agent 全部工具权限 -> 风险失控。渐进式放开:只读 -> 受控写 -> 全自动。
陷阱 5:没有人工介入点
Agent 遇到困难时无限重试 -> 浪费 Token。设置 request_help 工具,3 次失败自动升级。
陷阱 6:忽略 Token 成本
不设预算,一个 Loop 消耗 100 万 Token -> 破产。设定 Token 上限,溢出时优雅退出并保存进度。
陷阱 7:记忆只在内存里
重启后丢失所有进度 -> 从头来过。关键状态写入磁盘文件,支持断点续传。
十、总结与行动清单
Loop Engineering 的核心主张:与其反复手工调试提示词,不如设计一套让 Agent 自主迭代的循环系统。
你的第一步行动清单
- 今天:用上面的
agent_loop.py跑一个只读任务(分析日志、整理 TODO) - 这周:为你的项目写一个
SKILL.md,记录技术栈和代码规范 - 这月:设计一个 Phase 2 的受控执行者 Loop,每天自动修复简单 bug
- 三个月后:评估是否升级到 Phase 3 全自动
核心公式
好的 Agent Loop = 清晰的目标 + 可靠的评估 + 适当的权限 + 持久的记忆
记住:你不是在写更好的 Prompt,你是在设计一个系统。 Prompt 是一次性的对话,Loop 是持续运行的工程。
本文部分内容参考自菜鸟教程《Loop Engineering》及 Addy Osmani、Boris Cherny 等人的公开分享。