Agent 凌晨两点还在跑任务却不记得昨晚进度

连续三天了，我配置的多步骤工作流运行得非常顺畅，涵盖了数据抓取、清洗、分析到出图的全过程。但奇怪的是，我合上电脑睡觉，第二天再次打开对话，Agent 却像失忆一样，完全不记得昨晚发生了什么。

翻看日志发现并没有报错或异常，它只是把所有任务重新生成了一遍，完全忘记了昨晚停留在哪一步。

Agent 失忆背后的设计缺陷

这其实揭露了一个扎心的事实：问题不在于模型不够聪明，而在于你的 Agent 从设计之初就没能“活过今晚”。

一、从 Prompt 到 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 分开存储虽合理，但很快会面临并发写入、版本冲突以及记忆污染等更复杂的问题。