让智能体通过自身经验进化的记忆循环

最终成果：一个能主动记录失败、提炼教训并持续提升决策质量的 Letta 智能体
你需要：已安装 Letta SDK（1.0 系列）、Python 3.10+、一个 OpenAI API 密钥（或兼容的 LLM 端点）
预计耗时：45 分钟

2025 年末，Letta 从 MemGPT 的学术原型进化为面向工程实践的 1.0 版本，带来了稳定的 Agent Loop 和可编程记忆块。当我们构建面向真实业务的智能体时，最大的痛点不再是“它能否记住上一次对话”，而是 “它能否从上次搞砸的事情里学到东西，下次不再犯同样的错”。这一章，我们将用三种机制——失败反思钩子、经验蒸馏、自我进化基准测试——把记忆循环从“被动记录”升级为“主动进化”。

1. 你需要什么

安装命令：

pip install letta>=1.0.0
export OPENAI_API_KEY="sk-..."

踩坑经验：Letta 1.0 不再使用旧版 MemGPT 的 Tool 基类来定义工具，而是直接采用普通 Python 函数。如果你看到旧教程里用 class MyTool(Tool)，请跳过——现在的写法简单得多，函数签名加上类型注解即可。

2. 最终成果：一个会“反思”和“蒸馏”的客服智能体

我们将模拟一个电商客服智能体，它能调用后台接口查询订单、退款、记录备注。智能体一开始可能犯低级错误——比如在退款前忘记检查订单状态。我们的目标是：

1. 捕获每次工具调用的失败，自动生成结构化反思，写入记忆块（Memory Block）。
2. 定期触发经验蒸馏，将反思总结为简短的“长期策略”（Long‑term Policy），检索式注入后续决策。
3. 用量化基准测试，对比第 1 次、第 10 次、第 50 次执行的成功率与步骤效率。

最终你会看到一条可被观察、可被信任的上扬曲线。

3. 步骤说明

3.1 搭建基础智能体并定义工具

我们先创建一个智能体，赋予它三个工具。注意：以下函数就是 Letta 1.0 的自定义工具。

# tools.py
import uuid
from datetime import datetime

# 模拟后端数据库
orders_db = {
    "ORD-1001": {"status": "delivered", "amount": 299},
    "ORD-1002": {"status": "pending", "amount": 149},
}

def query_order(order_id: str) -> dict:
    """查询订单状态和金额。"""
    return orders_db.get(order_id, {"error": "Order not found"})

def refund_order(order_id: str, amount: float) -> dict:
    """发起退款，要求订单必须是 delivered 状态。"""
    order = orders_db.get(order_id)
    if not order:
        return {"error": "Order not found"}
    if order["status"] != "delivered":
        return {"error": f"Refund not allowed: order status is {order['status']}"}
    # 执行退款...
    return {"refund_id": str(uuid.uuid4()), "amount": amount, "at": datetime.now().isoformat()}

def add_note(order_id: str, note: str) -> dict:
    """为订单添加备注。"""
    if order_id not in orders_db:
        return {"error": "Order not found"}
    # 记录备注...
    return {"status": "noted", "order_id": order_id, "note": note}

注册为智能体的工具集——只需将函数放入列表即可。

3.2 初始化 Letta 智能体

Letta 1.0 的核心概念是记忆块（Memory Block），它们是可持久化在上下文窗口中的 JSON 片段。我们为本次实验预置两个块：

• user_profile：存放用户偏好和短期任务
• lessons_learned：存放反思与长期策略

# agent_setup.py
from letta import create_client
from letta.schemas.memory import Memory
import tools  # 我们定义的工具模块

client = create_client()  # 默认使用 OPENAI_API_KEY

agent_state = client.create_agent(
    name="reflective_agent",
    tools=[tools.query_order, tools.refund_order, tools.add_note],   # 普通函数直接传入
    memory=Memory(
        memory_blocks=[
            {"label": "user_profile", "value": "{}"},   # 初始为空
            {"label": "lessons_learned", "value": "[]"}  # 初始为空数组
        ]
    ),
    instructions="你是一个电商客服。在处理退款前，必须先查询订单状态。",
    model="gpt-4o-mini",
    embedding="openai/text-embedding-3-small",
)
agent_id = agent_state.id

踩坑经验：create_agent 时须明确 instructions，这里我们手工注入了一条规矩——“退款前先查单”，但智能体一开始不一定遵守。这正是我们要观测的进化起点。

预取结果：Agent 创建成功，memory_blocks 中出现了两个可编辑的 JSON 字段。你可以在 API 后台看到它们的实时内容。

3.3 失败反思钩子的实现

目标：捕获工具调用异常，自动生成事后报告并写入 lessons_learned。

在 Letta 1.0 中，工具执行的错误会以异常的形式抛回给 Agent Loop。我们可以用 client.send_message 驱动对话，然后解析返回的 let_agent_event 事件，检查是否有 MESSAGE_TYPE_TOOL_CALL_FAILED。

下面是一个轮询循环，模拟连续的客服对话，并在每次失败后追加反思。

# run_with_reflection.py
import json
from letta import LettaAgentEvent

def append_lesson(client, agent_id, lesson: dict):
    """将一条教训追加到 lessons_learned 块中。"""
    state = client.get_agent(agent_id)
    for block in state.memory.memory_blocks:
        if block.label == "lessons_learned":
            current_lessons = json.loads(block.value) if block.value else []
            current_lessons.append(lesson)
            block.value = json.dumps(current_lessons, ensure_ascii=False)
    client.update_agent(agent_id, state)   # 提交更新

def run_episode(client, agent_id, user_message: str):
    """执行一次用户交互，自动反思失败。"""
    response, events = client.send_message(
        agent_id=agent_id,
        message=user_message,
        return_events=True,    # 要求返回事件流
    )
    # 解析事件
    for event in events:
        if isinstance(event, LettaAgentEvent) and event.message_type == "tool_call_failed":
            failure = event.data
            # 生成反思内容（此处用模拟，生产中可让LLM总结）
            lesson = {
                "timestamp": failure.get("timestamp"),
                "tool": failure.get("tool_name"),
                "error": failure.get("error"),
                "reflection": f"下次调用 {failure.get('tool_name')} 前，应检查前置条件：{failure.get('error')}",
            }
            append_lesson(client, agent_id, lesson)
            print(f"🧠 教训已记录: {lesson['reflection']}")
    return response

预期结果：当用户说“直接给 ORD-1002 退款 149”时，智能体可能遗忘查询订单，直接调用 refund_order——而该订单状态为 pending，触发 tool_call_failed。我们的钩子会捕获这个事件，向 lessons_learned 写入一条教训。

踩坑经验：return_events=True 是关键参数，默认不返回底层事件。如果忘记开启，你将完全失去对工具调用成败的可观测性。

3.4 经验蒸馏：将短期教训固化为长期策略

反思钩子产生的“教训”逐条增长，但上下文窗口有限。我们必须定期蒸馏这些片段，生成可被检索的指导原则，存放在另一个记忆块（或同一块的 policies 字段）。

蒸馏的触发条件可以设为：lessons_learned 中积累了 3 条以上，或距上次蒸馏超过 1 小时。我们用定时任务模拟：

# distill.py
from datetime import datetime

def distill_lessons(client, agent_id):
    state = client.get_agent(agent_id)
    lessons_block = next(b for b in state.memory.memory_blocks if b.label == "lessons_learned")
    lessons = json.loads(lessons_block.value) if lessons_block.value else []

    if len(lessons) < 3:
        print("⏳ 教训不足，跳过蒸馏")
        return

    # 构造提示词，让LLM提炼原则
    prompt = f"以下是智能体过去的失败记录：\n{json.dumps(lessons, indent=2, ensure_ascii=False)}\n请提炼出不超过3条具体、可操作的原则，用英文bullet point列出。"
    principles = client.call_model(prompt)  # 利用Letta内置的模型调用

    # 存入 user_profile 的 “policies” 字段（或在 lessons_learned 新增字段）
    profile_block = next(b for b in state.memory.memory_blocks if b.label == "user_profile")
    profile = json.loads(profile_block.value) if profile_block.value else {}
    profile["policies"] = profile.get("policies", "") + "\n" + principles
    profile_block.value = json.dumps(profile, ensure_ascii=False)

    # 清空已蒸馏的教训，避免重复
    lessons_block.value = json.dumps([], ensure_ascii=False)
    client.update_agent(agent_id, state)
    print(f"✨ 长期原则已更新:\n{principles}")

预期结果：智能体的 user_profile 中多出一段类似这样的内容：

- Always query the order status before initiating a refund.
- If an order is not found, confirm the order ID with the user before retrying.

此后，每次对话开始时，Letta 会自动将这些原则注入上下文窗口（因为 user_profile 是核心记忆块），引导模型优先遵循。

踩坑经验：蒸馏不一定要清空教训，可以保留最近 2～3 条作为短期记忆。但长期原则必须单独存放，确保跨会话持久性。

3.5 构建自我进化基准测试

现在，我们设计一套可复现的任务集，量化智能体在 10 次独立运行中的进步。

测试套件（5 个典型客户请求）：

1. "请帮我退款 ORD-1001 的 299 元" （合法退款，但需先查询）
2. "ORD-1002 的订单状态是什么？如果不好就退款" （状态为 pending，应拒绝）
3. "给 ORD-9999 添加备注：客户要求加急" （订单不存在，应报错）
4. "我不记得单号，但我叫张伟，帮我查最近的订单" （边界情况）
5. "立刻退款 ORD-1002 149 元" （必须被阻止）

度量指标：

测试脚本基本结构：

# benchmark.py
from run_with_reflection import run_episode
from distill import distill_lessons

def run_benchmark(client, agent_id, rounds=10):
    for r in range(rounds):
        print(f"\n===== Round {r+1} =====")
        for task in task_suite:
            run_episode(client, agent_id, task)
        # 每轮结束自动蒸馏一次
        distill_lessons(client, agent_id)
        # 打印当前统计……（此处省略统计代码）

预期结果：前 2 轮中，智能体频繁触发失败，步骤浪费度高。第 5 轮后，成功率从约 40% 提升至 85% 以上，并且日志中出现 policy: always query order status... 等引用。

踩坑经验：测试轮次不要设为连续调用，中间可以稍作停顿（如 time.sleep(2)），防止 API 限流。另外，使用固定 seed 保证可重复性。

4. 回顾

我们做了什么：

1. 创建带自定义工具的智能体（普通函数直传），并初始化 lessons_learned 记忆块。
2. 实现失败反思钩子，解析 tool_call_failed 事件，自动记录结构化教训。
3. 实现经验蒸馏，定期将教训浓缩为长期原则，注入上下文。
4. 搭建基准测试，用 5 个任务持续观测成功率与步骤效率的提升。

完整代码约 150 行，从零到可运行约 45 分钟。核心收获是：记忆循环的进化力，来自对“失败时刻”的显式编码与周期性提炼，而非模型的随机涌现。

5. 行动清单

1. 安装 Letta SDK，复用本章的工具函数和 append_lesson。
2. 开启 return_events=True，确认你能捕获 tool_call_failed。
3. 自定义蒸馏提示词，适配你业务中的“关键前置条件”。
4. 用一套固定测试问题，跑至少 10 轮，记录成功率曲线。
5. 将蒸馏出的原则写入 instructions 或 user_profile，观察行为转变。

6. 下一步

当你拥有一个能从错误中进化的智能体时，新的挑战悄然浮现：上下文窗口溢出——这是生产环境中最常见的静默故障。它会如何一步步劣化你的智能体、以及如何识别早期信号？下一章我们将深入解剖这个无声的杀手。