痛点:你花了一个下午调优你的代理。第二天早上,它却盯着你发呆——仿佛昨天的一切从未发生。
你将学到:四阶段演进(提示词 → 上下文 → 框架 → 循环),以及一个可运行的 50 行循环代理实现持久记忆。
0. 前置条件
- Python ≥ 3.10
-
pip install openai(openai ≥ 1.0.0) - OpenAI API Key
- 操作系统:macOS / Linux / Windows WSL
目标:复制粘贴代码,运行并看到一个不会遗忘的循环代理。
1. 痛点:为什么你的代理一夜之间就忘得一干二净?
凌晨 2 点,你终于让多步骤工作流正常运行。代理按照你精心设计的提示词——数据获取、清洗、分析、可视化。你满意地合上笔记本电脑。
第二天早上,你满怀期待地打开对话——代理却盯着你发呆,仿佛这一切从未发生。
你检查日志。没有错误。没有异常。代理重新生成了所有内容——它只是“忘记”昨天停在哪里了。
这不是玩笑。这是每一位认真做代理的开发者都经历过的噩梦。根本原因不是“模型不够聪明”。而是一个更根本的事实:你的代理从设计之初就没打算活过一夜。
2. 四阶段演进:提示词 → 上下文 → 框架 → 循环
为了理解这一点,让我们使用一个简单的演进框架:
| 阶段 | 你做了什么 | 致命缺陷 |
|---|---|---|
| 提示词工程 | 将任务描述、示例、格式写入提示词 | 任何意外输入都会导致输出崩溃 |
| 上下文工程 | 将历史记录 + 中间结果塞进上下文窗口 | Token 成本线性增长,最终触及窗口上限 |
| 框架工程 | 添加工具调用、结构化输出、错误捕获 | 框架已搭建,但代理仍然是“一次性”的 |
| 循环工程 | 构建闭环:状态 + 记忆 + 反馈 + 重试 + 持久化 | 真正的工程——代理开始“活”起来 |
循环工程不是对提示词工程的否定——而是超越。提示词仍然重要。但它是引擎,你不能像开汽车一样驾驶引擎。
3. 最小可运行循环代理(50 行)
下面是完整的、可复制粘贴的、可运行的循环代理。先运行它,然后逐行理解。
import json, os, time
from pathlib import Path
from datetime import datetime
from openai import OpenAI
client = OpenAI() # reads OPENAI_API_KEY from env
STATE_FILE = Path("./agent_state.json")
MEMORY_FILE = Path("./agent_memory.json")
MAX_RETRIES = 3
def load_memory() -> dict:
if MEMORY_FILE.exists():
return json.loads(MEMORY_FILE.read_text())
return {"facts": {}, "errors": []}
def save_memory(mem: dict):
MEMORY_FILE.write_text(json.dumps(mem, indent=2, ensure_ascii=False))
def load_state() -> dict:
if STATE_FILE.exists():
return json.loads(STATE_FILE.read_text())
return {"state": "idle", "step": 0}
def save_state(state: str, step: int):
STATE_FILE.write_text(json.dumps({
"state": state, "step": step,
"updated_at": datetime.now().isoformat()
}, indent=2, ensure_ascii=False))
def execute(task: str, memory: dict, error_ctx: str = "") -> str:
system = f"You are a task-execution agent. Known facts: {json.dumps(memory.get('facts', {}), ensure_ascii=False)}"
if error_ctx:
system += f"\nLast error: {error_ctx}\nPlease fix."
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "system", "content": system},
{"role": "user", "content": task}]
)
return resp.choices[0].message.content
def check(output: str, keywords: list[str]) -> tuple[bool, str]:
missing = [kw for kw in keywords if kw not in output]
if missing:
return False, f"Missing: {missing}"
if len(output) < 20:
return False, "Output too short"
return True, ""
def run(task: str, keywords: list[str]):
memory = load_memory()
save_state("running", 0)
for i in range(1, MAX_RETRIES + 1):
error = memory["errors"][-1]["reason"] if memory["errors"] else ""
output = execute(task, memory, error)
ok, reason = check(output, keywords)
if ok:
save_state("done", i)
memory["facts"][task[:30]] = output[:100]
save_memory(memory)
return f"OK on attempt {i}:\n{output}"
else:
save_state("retrying", i)
memory["errors"].append(
{"task": task, "reason": reason, "attempt": i}
)
save_memory(memory)
print(f"Retry {i} failed: {reason}")
time.sleep(1)
save_state("failed", MAX_RETRIES)
return f"All {MAX_RETRIES} attempts failed"
if __name__ == "__main__":
result = run(
task="List 3 Python web frameworks and their features",
keywords=["Flask", "Django", "FastAPI"]
)
print(result)
Enter fullscreen mode Exit fullscreen mode
复制到 loop_agent.py 并运行。
4. 验证
# 1. 安装
pip install openai
# 2. 设置 API 密钥
export OPENAI_API_KEY="sk-..."
# 3. 首次运行
python loop_agent.py
Enter fullscreen mode Exit fullscreen mode
预期输出:
OK on attempt 1:
The 3 mainstream Python web frameworks:
1. **Flask**: lightweight micro-framework...
2. **Django**: full-stack, batteries included...
3. **FastAPI**: modern async, auto OpenAPI docs...
Enter fullscreen mode Exit fullscreen mode
现在验证持久化——关闭终端,重新打开,再次运行:
# 4. 检查状态文件
cat agent_state.json
Enter fullscreen mode Exit fullscreen mode
预期结果:
{
"state": "done",
"step": 1,
"updated_at": "2026-07-01T12:00:00.000000"
}
Enter fullscreen mode Exit fullscreen mode
# 5. 检查记忆文件
cat agent_memory.json
Enter fullscreen mode Exit fullscreen mode
预期结果:facts 字典包含你的任务和输出。
# 6. 再次运行——代理会自动加载记忆
python loop_agent.py
Enter fullscreen mode Exit fullscreen mode
代理会从 agent_memory.json 读取 facts,并将其作为已知上下文传入。这就是“隔夜记住”的机制。
5. 代码中映射的六大组件
| 组件 | 代码位置 | 描述 |
|---|---|---|
| 记忆存储 |
load_memory() / save_memory()
|
将记忆持久化到 JSON |
| 状态机 |
load_state() / save_state()
|
将状态持久化到 JSON |
| 执行器 | execute() |
调用 OpenAI API |
| 检查器 | check() |
验证关键词是否存在 |
| 任务调度器 | run() |
重试循环 + 状态转换 |
| 护栏 | MAX_RETRIES = 3 |
重试限制 |
6. 常见错误
错误 1:ModuleNotFoundError: No module named 'openai'
pip install openai
Enter fullscreen mode Exit fullscreen mode
错误 2:openai.AuthenticationError: 401
你没有设置 API 密钥,或密钥无效。
export OPENAI_API_KEY="sk-..."
Enter fullscreen mode Exit fullscreen mode
错误 3:代理输出缺少 Flask/Django/FastAPI
这是正常的!大模型并不总是完美遵循指令。这正是循环代理的价值所在——检查器检测到缺少关键词,拒绝输出,并自动重试。你会看到:
Retry 1 failed: Missing: ['Flask', 'Django', 'FastAPI']
Retry 2 failed: Missing: ['Django']
OK on attempt 3:
...
Enter fullscreen mode Exit fullscreen mode
错误 4:API 超时或速率限制
gpt-4o-mini 非常便宜(约 $0.00015/次调用)。如果遇到速率限制,请增加 time.sleep(1) 或检查你的 OpenAI 控制台。
7. 从 50 行代码开始的下一步
现在你拥有了一个完整的可运行循环代理。它使用普通 JSON 文件进行持久化——这正是让你能用手触摸每个组件的原因。
扩展它:
- 将 ChromaDB 替换为 JSON → 向量记忆
- 接入 Celery → 真正的异步任务队列
- 添加飞书/Slack webhook → 完成通知
- 添加结构化日志 + Trace ID → 可观测性
循环工程不是你安装的包。它是架构思维的转变。从这 50 行开始,你已站在第四阶段的门槛。
下一篇文章:你的代理需要多少记忆?——记忆存储选择指南
关于作者:吴记(无记)——专注于代理工程、循环工程和数字化转型的 AI 与数字化实践者。实用、手把手教程——跟着做就能跑起来。
0 Comments
Log in to join the conversation.No comments yet. Be the first to share your thoughts.