8.1. 从零搭建一个能记住一切的客服智能体

你刚刚接到一个紧急任务：将公司试用期的 AI 客服从“金鱼记忆”的对话机器人，升级为能记住用户偏好、历史订单，甚至在服务器宕机后自行恢复的正式员工。产品经理撂下一句话：“它必须比最细心的金牌客服还有记性。”

时间是东八区周五下午 15:30，距离下周一演示还有 72 小时。

本章，我们不谈抽象理论，而是用 LangChain 灵活的工具编排能力，搭配 Letta 白盒式的持久化记忆框架，再将一切状态落地到 PostgreSQL。最终交付一个功能完整、能记住一切的客服智能体。

你需要什么

• Python 环境：Python 3.10 及以上
• PostgreSQL 实例：本地或远程均可，需开启 pgvector 扩展（用于语义搜索）
• OpenAI API 密钥：作为核心 LLM 推理后端
• Letta 服务：我们使用自托管（Self-hosted）模式，确保数据不外泄
• 预计时间：约 45–60 分钟完成搭建与首次对话

最终成果

一个拥有三层记忆的客服智能体：

1. 热记忆（Redis 或进程内存）：缓存当前对话上下文，毫秒级响应；
2. 冷记忆（PostgreSQL）：存储历史订单、对话摘要、用户画像，持久化跨会话状态；
3. 语义记忆（pgvector）：对历史对话做向量化，支持“我上次说的那个退货问题”这类模糊查询。

在下周一的演示中，它会记住“王先生昨天投诉过物流延迟，并偏好微信通知”，在宕机重启后，第一句话就能说：“王先生您好，我们看到您昨天反馈的物流问题已解决，您现在还需要微信通知退款进度吗？”

步骤一：需求分析与记忆结构设计

普通客服对话中，LLM 的上下文窗口一旦溢出，早期信息就会丢失。我们需要提前规划：代理到底要记住什么？

定义记忆条目

注意：Letta 的“内存块（memory blocks）”机制天然支持这种结构。你创建一个标注为 human 的内存块，代理就会自动在对话中更新用户画像。

我们从三个维度入手：

1. 热记忆：LangChain 原生 ConversationBufferWindowMemory 维护最近的对话轮次；
2. 冷记忆：Letta 代理在每一次回复后，自动调用 memory_replace 将更新后的 human 块回写到 PostgreSQL 的 memories 表；
3. 语义搜索：利用 pgvector 索引历史对话摘要，当用户提到模糊线索时，从向量库中检索最相关的 3 条记忆条目。

步骤二：环境搭建

启动 Letta 自托管服务

Letta 官方提供了易于部署的 Docker 镜像，我们选择“自托管 + PostgreSQL”模式，这样所有数据都在你自己的基础设施上。

# 拉取 Letta 服务镜像（截至当前调研资料，v0.5.x 为推荐版本）
docker pull letta/letta:latest

# 启动，注入数据库连接字符串和 OpenAI 密钥
docker run -d \
  --name letta-server \
  -p 8283:8283 \
  -e LETTA_PG_URI=postgresql://user:password@host:5432/letta_db \
  -e OPENAI_API_KEY=sk-xxxxx \
  letta/letta:latest

预期结果：访问 http://localhost:8283/docs 可见 Letta 的 Swagger API 文档。

初始化 Python 项目

mkdir memory-agent && cd memory-agent
python -m venv venv && source venv/bin/activate
pip install letta-client langchain langchain-openai pgvector psycopg2-binary redis

配置环境变量：

export LETTA_BASE_URL="http://localhost:8283"
export LETTA_API_KEY="letta"  # 自托管下可设置简单密钥，等于密码
export OPENAI_API_KEY="sk-xxxxx"

步骤三：实现记忆管理层

我们不希望业务代码直接操作数据库，设计一个统一的 MemoryManager 类，封装三个核心记忆操作。

3.1 创建冷记忆管理器（PostgreSQL + Letta）

在 Letta 中，创建代理时需定义 memory_blocks。这是智能体“知道自己是谁”和“知道用户是谁”的起点。

from letta_client import Letta

client = Letta(base_url="http://localhost:8283", token="letta")

# 创建客服代理，注入初始记忆
agent = client.agents.create(
    name="金牌客服小美",
    memory_blocks=[
        {
            "label": "persona",         # 代理对自己的认知
            "value": "你是小美，一位耐心、细心的客服专家。你总是用中文回复，语气温和。"
        },
        {
            "label": "human",           # 代理对用户的认知（初始为空，后续自动更新）
            "value": "用户姓名: 未知; 会员等级: 无; 历史投诉: 无; 偏好: 无。"
        }
    ],
    model="openai/gpt-4o-mini"          # 可根据成本切换为 gpt-3.5-turbo
)
print(f"代理 ID: {agent.id}")

预期结果：返回代理 ID，Letta 会自动在 PostgreSQL 的 agents 表中创建记录。

3.2 编写统一的记忆 API

class MemoryManager:
    """统一封装三层记忆的读写"""
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        # 热记忆由 LangChain 管理，这里仅作示例
        self.hot_cache = {}   # 实际项目中可用 Redis 替代

    def remember_user(self, user_id: str, field: str, value: str):
        """更新用户画像，写入 Letta 冷记忆"""
        # 读取当前 human 块内容
        state = client.agents.get(self.agent_id)
        human_block = next(b for b in state.memory.blocks if b.label == "human")
        current_data = human_block.value

        # 简化的字符串解析，生产环境请用 JSON
        if f"{field}:" in current_data:
            current_data = current_data.replace(f"{field}: ...", f"{field}: {value}")

        # 调用 Letta memory_replace 工具（由代理内部自动调用）
        # 这里通过直接发送带有更新意图的消息来触发更新
        client.agents.messages.create(
            agent_id=self.agent_id,
            messages=[{"role": "user", "content": f"请更新 {user_id} 的 {field} 为 {value}"}],
            use_tool=True   # 允许代理自动选择 memory_replace
        )

    def recall_user_context(self, user_id: str) -> dict:
        """查询用户完整记忆"""
        # 实际实现中可从 PostgreSQL 直接查询 memories 表
        # 这里返回热缓存或 Letta agent 状态中的 human 块
        state = client.agents.get(self.agent_id)
        human_block = next(b for b in state.memory.blocks if b.label == "human")
        return {"raw": human_block.value}

踩坑经验：Letta 自动调用 memory_replace 时可能“过度修正”，比如将“用户喜欢红色”改为“用户喜欢蓝色”源于一次误听。建议在生产环境增加人工审核步骤或日志监控。

预期结果：MemoryManager 实例化后，可以调用 remember_user 来显式更新用户信息。

步骤四：构建客服对话逻辑

现在整合 LangChain 工具封装和 Letta 代理的持续对话能力。

4.1 LangChain 工具抽象

为了让代理能查询订单或执行退款，我们用 LangChain 将模拟的业务 API 封装成工具。

from langchain.tools import tool

@tool
def query_order(order_id: str) -> str:
    """查询订单详情，输入订单 ID"""
    # 模拟数据库查询
    orders = {
        "ORD123": "商品: 蓝牙耳机; 状态: 已签收; 物流: 顺丰 SF12345",
        "ORD456": "商品: 机械键盘; 状态: 运输中; 物流: 圆通 YT7890"
    }
    return orders.get(order_id, "未找到该订单")

@tool
def request_refund(order_id: str, reason: str) -> str:
    """发起退款，输入订单 ID 和退款原因"""
    # 模拟退款流程
    return f"已为订单 {order_id} 发起退款申请，原因: {reason}，预计 3 个工作日内完成。"

4.2 记忆驱动的对话流程

LangChain 负责工具调用路由，Letta 负责持久记忆，二者通过 MemoryManager 连接。

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# Letta 代理 ID
AGENT_ID = agent.id
mem_mgr = MemoryManager(AGENT_ID)

# LangChain 设置
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
tools = [query_order, request_refund]

# 记忆：保留最近 5 轮对话
memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    return_messages=True,
    k=5
)

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是小美。从 Letta 记忆中提取用户信息，如果缺少就礼貌询问。
当前用户记忆: {user_memory}"""),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

agent_chain = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent_chain, tools=tools, memory=memory)

def chat(user_id: str, user_message: str):
    # 1. 从 Letta 冷记忆中加载用户画像
    user_profile = mem_mgr.recall_user_context(user_id)["raw"]

    # 2. 执行对话
    response = agent_executor.invoke({
        "input": user_message,
        "user_memory": user_profile
    })

    # 3. 将本轮对话写回 Letta，让代理自动更新 human 块
    mem_mgr.remember_user(user_id, "last_query", user_message)
    return response["output"]

预期结果：首次调用 chat("user_101", "我的订单 ORD123 到哪了？") 时，代理会先看到 human 块中用户信息为“未知”，并自动调用 memory_replace 将姓名更新为“user_101”。

步骤五：测试记忆持久性与灾难恢复

测试一：跨会话记忆

# 会话 1：用户投诉
print(chat("user_101", "我昨天说的那个蓝牙耳机还是没到，我要退款！"))

# 手动模拟“服务器重启”——将内存缓存清空
memory.clear()
mem_mgr.hot_cache = {}

# 会话 2：用户再次进来，代理应记得昨天的投诉
print(chat("user_101", "我的退款处理得怎么样了？"))

预期输出：

会话1: "user_101 您好，看到您对订单 ORD123 的物流有疑问，已为您发起退款。"
会话2: "您昨天提交的退款申请正在处理中，预计 2 个工作日内完成。"  # 基于 Letta 持久化的 previous complaint 字段

测试二：宕机后自动恢复

如果在生产环境，重启后只需重新连接 Letta 服务，从 PostgreSQL 加载代理状态，无需任何额外操作。这就是 Letta 将整个代理视为“持久服务”的优势。

回顾

现在，你手中的智能体已具备金牌客服的核心记忆能力。它不只是“看似聪明”，而是真正将用户的生活细节、抱怨和偏好刻进了长期记忆，且能在任何意外重启后自动恢复。

行动清单

1. 根据你的业务，扩展 memory_blocks，在 human 块中增加 订单偏好、投诉历史 等结构化字段；
2. 为 MemoryManager 增加人工审核接口，防止代理自动更新记忆时产生误差；
3. 用 redis 替换 hot_cache 字典，缓存最近 N 轮对话和用户画像；
4. 部署 pgvector 索引，将历史对话摘要向量化，开通“语义搜索记忆”能力。

下一章预告

你的客服智能体已经能在崩溃后优雅恢复，但如果遇到一个复杂的审批流程——可能需要暂停、等待人工干预，甚至中途挂起数小时后继续——该怎么办？

下一章 《用 LangGraph 实现带记忆断点恢复的工作流》，我们将把 LangGraph 的持久化检查点与 Letta 的长期记忆结合，构建一个可随时暂停、随时恢复的多步审批智能体。你将看到“挂起”不再是中断，而仅仅是工作流中的一个状态。