Letta：给 AI 智能体装上长期记忆与自我进化能力的开源平台

一、问题：一个被"金鱼记忆"拖垮的客服助手

去年我给一家 B2B SaaS 公司做内部支持助手。场景不复杂：3400 家企业客户，累计 12 万篇产品文档、Release Note 和历史工单。一线支持同学在飞书里问问题，助手给答案。

第一版是标准 RAG：向量检索 top-k 拼接 + 无状态 LLM。上线两周，吐槽集中在三件事：客户第二次来问，助手又要他从头描述一遍环境；同一客户第三次问同一个问题，给的答案和第一次不一样；客户明确说过"我们不用 SSO"，助手下一轮还在推 SSO 方案。

说白了，它没有记忆。上下文窗口不等于记忆——塞进去的是"这一轮的相关片段"，不是"这个客户是谁、踩过什么坑"。

二、需求拆解

• 数据：12 万篇文档（约 210 万 chunk）作为公共知识；3400 家客户各自的历史工单、环境信息、偏好要严格隔离，A 客户的信息绝不能出现在 B 客户的会话里。
• 性能：支持同学在对话里等，P95 必须压到 3 秒以内；上下文越长成本越高，不能靠无脑塞 20k token 换效果。
• 成本：预算卡死在每月 ¥8000 的 GPU 摊销内（模型自建，数据不出内网），没有空间做"多轮全量重放"。
• 部署约束：客户工单含商业合同细节，必须私有化部署；只有推理走内网 vLLM，任何托管 SaaS 都不接受。

三、方案设计：为什么是 Letta

当时横向看了三条路。

方案 A：LangChain + Redis 自研记忆层。可控，但"什么时候写记忆、写什么、旧记忆怎么裁剪"全得自己写 prompt 和调度逻辑。我估过工作量，光记忆读写策略的调优就得两三个月，而且这本质是在重造轮子。

方案 B：Dify / Coze 这类平台。开箱快，但 memory 是会话级 KV，做不了"智能体主动决定记什么"。而且私有化版本对多租户 agent 的管理粒度不够。

方案 C：Letta。它把"有状态智能体"当成一等公民：每个 agent 有独立的 memory_blocks（常驻上下文，智能体自己可以读写），有 archival memory（长尾检索），有工具调用循环，还能把整个 agent 序列化成 .af 文件持久化。Apache 2.0，自托管，24.8k star，社区活跃度够。

关键取舍：我没有把 12 万篇公共文档灌进 archival memory。用 client.agents.passages.create 灌 210 万 chunk 到 3400 个 agent 里，存储和同步都是灾难。我的做法是——公共知识走已有的向量检索服务，包装成一个 custom tool 挂给 agent；archival memory 只放每个客户自己的历史工单。职责清晰，也让归档检索的召回率更好控。

四、落地实现

步骤 1：起服务与数据准备

# docker-compose.yml（核心部分）
services:
  letta:
    image: letta/letta:latest
    ports: ["8283:8283"]
    environment:
      - OPENAI_API_KEY=sk-internal-vllm
      - OPENAI_BASE_URL=http://vllm.internal:8000/v1
      - LETTA_PG_URI=postgresql://letta:pass@postgres:5432/letta
    volumes:
      - ./data:/app/data
    depends_on: [postgres]
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: letta
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: letta
    volumes: ["pgdata:/var/lib/postgresql/data"]

docker compose up -d
curl -s localhost:8283/v1/health   # {"status":"ok"}

步骤 2：把知识库包成一个 custom tool

# tools/kb_search.py —— Letta 的自定义工具要求函数自包含，import 写在函数里
def kb_search(query: str, top_k: int = 8) -> str:
    """检索内部产品知识库，返回最相关的文档片段。query 用中文自然语言。"""
    import requests
    r = requests.post(
        "http://kb.internal/search",
        json={"q": query, "k": top_k, "rerank": True},
        timeout=5,
    )
    hits = r.json()["hits"]
    return "

---

".join(f"[{h['doc']}] {h['text']}" for h in hits)

# register_tool.py
from letta_client import Letta

client = Letta(base_url="http://localhost:8283")
kb_tool = client.tools.create(source_code=open("tools/kb_search.py").read())
print(kb_tool.id)

步骤 3：定义 agent 模板并批量建租户

PERSONA = """你是 XX 产品的技术支持助手。回答必须基于 kb_search 的结果，不要臆测。
如果客户档案里写了他们的环境约束（如"不用 SSO"），方案必须尊重这些约束。
遇到无法确认的问题，明确说不知道并建议升级工单。"""

def build_agent(tenant):
    return client.agents.create(
        name=f"cs-{tenant['id']}",
        model="openai/qwen2.5-72b-instruct",
        embedding="openai/bge-m3",
        memory_blocks=[
            {"label": "persona", "value": PERSONA, "limit": 2000},
            {"label": "customer_profile", "value": tenant["profile"], "limit": 3000},
        ],
        tool_ids=[kb_tool.id],
        include_base_tools=True,   # 自带 archival_memory_insert / search
    )

limit 是关键配置。第一版我没设，agent 把 customer_profile 越写越长，三周后单次请求 prompt 冲到 26k token。加上硬上限后，写超了就会被截断并触发它自己重写。

步骤 4：灌历史工单 + 接线

from concurrent.futures import ThreadPoolExecutor

def backfill(agent_id, tickets):
    for t in tickets:
        client.agents.passages.create(
            agent_id=agent_id,
            text=f"[{t['date']}] {t['title']}
{t['resolution']}",
            tags=[t["status"], t["product"]],
        )

with ThreadPoolExecutor(max_workers=16) as ex:
    for tenant in tenants:
        agent = build_agent(tenant)
        ex.submit(backfill, agent.id, tenant["tickets"])

3400 个 agent、平均每家 23 条历史工单，约 6 分钟跑完。对外只暴露一个薄网关，按 tenant_id 路由到对应 agent，流式回吐：

for chunk in client.agents.messages.create_stream(
    agent_id=agent_id, messages=[{"role": "user", "content": question}]
):
    yield chunk

步骤 5：部署与运维

# 每天 03:00 导出 agent 快照，agent 本身就是可持久化的资产
0 3 * * * curl -s "$LETTA/v1/agents/$AGENT_ID/export" -o /backup/$AGENT_ID.af

五、效果与数据

上线 6 周后，取 300 条真实提问做人工抽检（双人盲评，判定"答案可直接采用且不违反客户约束"）：

延迟下降不是模型变快了，而是 prompt 短了 71%。这是记忆系统最容易被忽略的收益：记对了，就不用每轮重讲一遍故事。

六、踩过的坑

坑 1：记忆块被写爆，成本反向飙升

现象：上线第 18 天，监控报 prompt token 中位数从 4.8k 涨到 26k，单轮成本翻了 4 倍。排查：把某个 agent 的 message 历史拉出来看，发现它在每轮对话后都往 customer_profile 里追加一句"用户又问了 XX"，三周堆了 8000 字。解决：给 block 加 limit；在 persona 里明确写"只有当客户透露了新的长期约束（环境、合规、预算）时才更新档案，一次性问题不写"；把易膨胀的字段拆成独立 block，各自限额。

坑 2：中文检索召回塌了

现象：agent 明明有 archival memory，但问"上个月那个 OOM 报错怎么解的"，它只会说"我需要更多信息"，从不检索历史工单。排查：直接调底层检索接口，发现团队内部黑话（"双跑"、"灰度池"）在默认 embedding 下相似度极低。解决：换自建 bge-m3 做 embedding，同时在 tool 描述里加了一句"当客户提到历史问题、报错码、之前发生过的事时，先调用 archival_memory_search"。工具描述就是 prompt，这点别省。

坑 3：多租户串记忆

现象：测试环境一个客户看到了另一家的产品版本号。排查：早期为了省事，测试用的是同一个 agent，靠 tags 区分租户。解决：硬性改成一个租户一个 agent，网关层按 tenant_id 强制路由，并在 CI 里加了一条断言：任意两个 agent 的 customer_profile 内容不得有交集文本。这类隔离必须靠架构，不能靠约定。

七、复盘与扩展

做对的：一是没把公共知识塞进 archival memory，职责分离让两边都好调；二是给每个租户独立 agent，隔离干净、快照可迁移；三是给记忆块设硬上限，这是成本可控的前提。

可以更好的：第一版应该在 persona 里就把"什么值得记"写清楚，而不是等记忆爆了再补；另外我一开始没接 Letta 的睡眠时智能体（后台异步整理记忆）能力，现在还在用同步写记忆，长会话下偶尔会拖慢响应，这是下一步要补的。

可扩展方向：把 3400 个客户 agent 聚合成一个"团队 agent"，让它汇总共性问题反哺知识库——客户的集体记忆其实就是最好的 FAQ 来源；另外可以按客户的产品版本给 agent 打 tag，做版本化的记忆迁移，客户升级后旧记忆不会变成误导。