在一个客服 Agent 中，用户第 50 次对话时提到了第 3 次对话的偏好，Agent 该如何记得？

这个问题其实在问两件事：如何存得久，以及如何找得准。如果是我来设计这个客服 Agent 的记忆系统，会围绕“用索引撬动精准回忆”这个思路来落。

🧩 先把“记忆”拆成两部分¶

事实性偏好：比如“我只接受短信通知”“我不喜欢电话沟通”。这种偏好很稳定，一旦确立就要能跨会话复用。
情境线索：第 3 次对话的上下文（比如当时在投诉物流），它能帮助 Agent 理解偏好是在什么场景下产生的，避免断章取义。

所以，我们需要的不是一个单纯的日志归档，而是一套索引驱动、持久化、可更新的用户记忆模块。

🔍 存储设计：让第 50 次对话能“找到”第 3 次¶

我一般会用三层协作：

为什么这样分？第 3 次对话的偏好一旦被提取出来，就应该写入用户画像表的 preferences JSONB 字段里，比如：

{
  "notification_channel": "sms",
  "contact_frequency": "low",
  "source_session": 3
}

这样，第 50 次对话不需要去翻第 3 次的海量原始文本，只需要查出这个字段就够了。这是保证召回速度的核心。

但有时用户会在第 50 次对话里说：“我记得上次我提到过不喜欢电话，还是改成邮件吧。” 这里“上次”可能指很久以前，Agent 需要理解这次变更的上下文，所以还得有手段回到原始对话摘要。

💾 存储与检索的代码骨架¶

import json
import chromadb
import redis
from datetime import datetime
from sentence_transformers import SentenceTransformer

class CustomerMemory:
    def __init__(self, db_conn, redis_conn):
        self.db = db_conn                       # PostgreSQL 连接
        self.redis = redis_conn                 # Redis 连接
        self.chroma = chromadb.Client()
        self.collection = self.chroma.create_collection("sessions")
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')

    def store_session_summary(self, user_id, session_id, summary, preferences_extracted):
        """每次对话结束调用：存摘要、向量化，并更新画像"""
        # 1. 向量库存储（用于未来模糊回忆）
        vec = self.encoder.encode(summary).tolist()
        self.collection.add(
            embeddings=[vec],
            documents=[summary],
            metadatas=[{
                "user_id": user_id,
                "session_id": session_id,
                "timestamp": datetime.now().isoformat()
            }],
            ids=[f"{user_id}_{session_id}"]
        )
        # 2. 更新用户画像里的偏好字段
        self.db.execute(
            "UPDATE user_profiles SET preferences = preferences || %s WHERE user_id = %s",
            (json.dumps(preferences_extracted), user_id)
        )
        # 3. 缓存到 Redis（热数据）
        self.redis.set(f"user:{user_id}:prefs", json.dumps(preferences_extracted))

    def recall_preference(self, user_id, query_context):
        """第50次对话时，用户说了某句话，检索相关历史偏好"""
        # 先看 Redis 热缓存
        cached = self.redis.get(f"user:{user_id}:prefs")
        if cached:
            prefs = json.loads(cached)
            # 如果 query_context 恰好命中某个偏好类别，直接返回
            if any(kw in query_context for kw in prefs.keys()):
                return prefs

        # 没命中或需要上下文，走向量检索
        q_vec = self.encoder.encode(query_context).tolist()
        results = self.collection.query(
            query_embeddings=[q_vec],
            where={"user_id": user_id},
            n_results=3
        )
        # 从结果里提取相关摘要，供 LLM 判断
        summaries = results.get("documents", [[]])[0]
        return {"related_summaries": summaries, "preferences_from_db": self.db.get_preferences(user_id)}

⚡ 第 50 次对话时的实际流程¶

当用户说出：“能不能像上次那样，别给我打电话，发短信就行了”时：

意图识别：检测到“偏好召回”意图。
快速检查：从 Redis / 用户画像表里拿 notification_channel，如果是 sms，直接回复“当前设置已经是不打电话，需要调整吗？”
深度检索（如有必要）：如果用户说“上次我明明说过”，Agent 会用“短信电话偏好”做语义向量检索，带上 user_id 过滤，抓出第 3 次对话的摘要，交给 LLM 生成自然解释：“您在 4 月 12 日的沟通中提到过这个偏好，我们一直按此执行。”

🧠 一个容易出错的坑：偏好的动态演化¶

用户可能在第 3 次说“永远别打电话”，第 30 次又说“紧急情况可以打电话”。如果只是简单追加记录，最终 Agent 会看到两条矛盾信息。

解法：偏好存储必须支持“覆盖更新”和“冲突标记”。在画像表里用 JSONB 的路径更新，同时写入一条变更事件到日志表，便于追溯演变过程。如果冲突明显，可以主动询问用户确认。

在一个客服 Agent 中，用户第 50 次对话时提到了第 3 次对话的偏好，Agent 该如何 记得 ？