我的Agent凌晨2点还在跑任务

连续第三天了。我调好的多步骤工作流，数据抓取、清洗、分析、出图，跑得好好的。合上电脑睡觉，第二天打开对话，它一脸茫然地看着我，好像昨天的一切从未发生。

查了日志。没有报错，没有异常。它把所有事情重新生成了一遍——只是不记得自己昨晚停在哪一步。

这个问题的根源不在"模型不够聪明"，而是更扎心的事实：你的Agent从来没被设计成能活过今晚。

一、四阶段演进：Prompt → Context → Harness → Loop

这套框架我最初是当段子看的，直到自己踩了坑才服气：

• Prompt Engineering：把任务描述、示例、输出格式全塞进prompt。致命缺陷：任何预料外的输入直接崩。
• Context Engineering：把历史对话+中间结果全塞进上下文窗口。致命缺陷：token成本线性增长，迟早撞上窗口上限。
• Harness Engineering：加工具调用、结构化输出、错误捕获。框架有了，但Agent本质上还是一次性的。
• Loop Engineering：构建闭环——状态+记忆+反馈+重试+持久化。到这一步，Agent才算是"活"了。

注意，Loop Engineering不是对Prompt Engineering的否定，而是超越。Prompt仍然重要，但它只是引擎，你不能拿引擎当车开。

二、一个能过夜的Loop Agent（50行）

直接贴代码，先跑通再逐行理解。这是我试过最短的能持久化记忆的版本：

import json, os, time
from pathlib import Path
from datetime import datetime
from openai import OpenAI

client = OpenAI()  # 从环境变量读取 OPENAI_API_KEY

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_loop(task: str, keywords: list[str], max_steps: int = 5):
    mem = load_memory()
    state = load_state()
    step = state.get("step", 0)
    error_ctx = ""

    while step < max_steps:
        output = execute(task, mem, error_ctx)
        ok, msg = check(output, keywords)
        
        if ok:
            save_state("done", step + 1)
            mem["facts"][f"step_{step}"] = output
            save_memory(mem)
            print(f"[OK] Step {step} passed")
            break
        else:
            error_ctx = msg
            step += 1
            save_state("retry", step)
            mem["errors"].append({"step": step, "error": msg})
            save_memory(mem)
            print(f"[RETRY] Step {step}: {msg}")
            time.sleep(1)
    
    print("State file:", STATE_FILE.read_text())
    print("Memory file:", MEMORY_FILE.read_text())

# 用法示例
if __name__ == "__main__":
    run_loop(
        task="Fetch latest AI news, summarize in 3 bullets, save to file.",
        keywords=["AI", "bullet"],
        max_steps=3
    )

三、关键在哪

这个设计的核心是把"状态"和"记忆"落盘，而不是留在上下文里。每次执行前读进来，执行后写回去。这样即使进程被杀、机器重启、openai断连，Agent醒来还能从上次的step继续走。

说实话，我看到"50行搞定"这种标题本能地反感，但这个确实没吹牛。真正的工程点在于：把一次性对话变成可恢复的进程。

不过我也得泼盆冷水——state和memory分开存是合理的，但真实场景里你很快就会遇到并发写入、版本冲突、记忆污染这些新问题。这50行只是起点，不是终点。