在线咨询

专属客服在线解答，提供专业解决方案

工单支持

专业技术支持团队，随时响应服务需求

首页博客正文

语音智能体的跨会话记忆：持久化与隐私删除

2026-07-27

开发者教程

max_history 只管当前这条命的事——会话一停，智能体就把你忘干净了。想让用户下次打进来还被记得，得在停止前把记忆取出来存好，新会话创建时再注入回去。

这件事技术上不难，难在做得安全。所以下面会连带处理用户隔离、数量限制和删除接口。少了它们，跨会话记忆就是个隐私事故等着发生。

跑完这一套，用户第二次进来时智能体还记得上次聊过什么；而调用删除接口之后，它会干净地忘掉。

> 开始之前　会话创建和停止的部分沿用《Python + FastAPI 搭建中文语音智能体》，先跑通那一篇。

一. 架构与准备工作

会话 A ── get_history（停止前）──→ 清洗 → SQLite（HMAC 用户键）
                                                  │
会话 B ← system_messages 注入 ← 最近 20 条记忆 ←─┘
用户删除请求 ── 鉴权 ──→ DELETE memory

生产环境的 userId 必须从已验证的登录态取，绝不能相信浏览器提交上来的昵称。这是后面所有安全设计的地基。

二. 环境变量

AGORA_APP_ID=REPLACE_WITH_SHENGWANG_APP_ID
AGORA_APP_CERTIFICATE=REPLACE_WITH_PRIMARY_CERTIFICATE
DEEPSEEK_API_KEY=REPLACE_WITH_DEEPSEEK_KEY
MINIMAX_API_KEY=REPLACE_WITH_MINIMAX_KEY
MEMORY_DB_PATH=./data/memory.db
MEMORY_MAX_TURNS=20
MEMORY_USER_PEPPER=REPLACE_WITH_RANDOM_SECRET
MEMORY_ADMIN_TOKEN=REPLACE_WITH_RANDOM_ADMIN_TOKEN

三. 用户隔离与 SQLite 存储

import hashlib, hmac, json, os, sqlite3, time
from typing import Any

MAX_TURNS = int(os.getenv("MEMORY_MAX_TURNS", "20"))

def pseudonymous_key(user_id: str) -> str:
    pepper = os.environ["MEMORY_USER_PEPPER"]
    if not user_id.strip():
        raise ValueError("user_id 不能为空")
    return hmac.new(pepper.encode(), user_id.strip().encode(),
                    hashlib.sha256).hexdigest()

def connect() -> sqlite3.Connection:
    connection = sqlite3.connect(
        os.getenv("MEMORY_DB_PATH", "memory.db"), check_same_thread=False)
    connection.execute(
        "CREATE TABLE IF NOT EXISTS memories ("
        "user_key TEXT PRIMARY KEY, turns_json TEXT NOT NULL, updated_at REAL NOT NULL)")
    connection.commit()
    return connection

def sanitize_turns(turns: list[dict[str, Any]]) -> list[dict[str, str]]:
    result = []
    for turn in turns:
        role, content = str(turn.get("role", "")), str(turn.get("content", "")).strip()
        if role in {"user", "assistant"} and content:
            result.append({"role": role, "content": content[:2000]})
    return result

def load_memory(connection, user_key: str) -> list[dict[str, str]]:
    row = connection.execute(
        "SELECT turns_json FROM memories WHERE user_key=?", (user_key,)).fetchone()
    return json.loads(row[0]) if row else []

def save_memory(connection, user_key: str, turns: list[dict[str, Any]]) -> None:
    merged = (load_memory(connection, user_key) + sanitize_turns(turns))[-MAX_TURNS:]
    connection.execute(
        "INSERT INTO memories(user_key, turns_json, updated_at) VALUES (?, ?, ?) "
        "ON CONFLICT(user_key) DO UPDATE SET "
        "turns_json=excluded.turns_json, updated_at=excluded.updated_at",
        (user_key, json.dumps(merged, ensure_ascii=False), time.time()))
    connection.commit()

数据库里只存 HMAC 之后的稳定用户键，手机号、昵称、业务账号一律不落库。这样即使数据库泄露，拿到的也只是一堆哈希值。

代价是 pepper 变成了关键资产：一旦更换，所有旧用户键都对不上，记忆等于全丢。所以要有安全的备份和轮换方案，别随手写在代码里。

四. 在下一次会话注入记忆

def memory_system_message(turns: list[dict[str, str]]):
    if not turns:
        return None
    lines = "\n".join(f"- {t['role']}: {t['content']}" for t in turns)
    return {"role": "system", "content": (
        "这是同一位用户过去明确同意保存的对话摘要。请自然利用，"
        "不要逐字复述，也不要声称记得未提供的信息：\n" + lines)}

user_key = pseudonymous_key(authenticated_user_id)
connection = connect()
try:
    turns = load_memory(connection, user_key)
finally:
    connection.close()

system_messages = [{"role": "system", "content":
    "你是温暖、简洁的中文语音助手。只有在相关时才使用历史记忆。"}]
memory_message = memory_system_message(turns)
if memory_message:
    system_messages.append(memory_message)

把 system_messages 传给 DeepSeekLLM，其余 ASR、TTS 和会话创建的代码跟快速开始完全一样。

启动请求要带上稳定的业务用户 ID。下面这段只是说明请求契约长什么样。生产环境里 userId 应该由后端从登录态覆盖，不能采信前端传的值：

await fetch(`${apiBase}/startAgent`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    channelName: config.channel_name,
    rtcUid: Number(config.agent_uid),
    userUid: Number(config.uid),
    userId: authenticatedUser.id,
  }),
});

五. 停止前取出记忆

async def capture_then_stop(session, user_key: str) -> None:
    try:
        history = await session.get_history()
        contents = getattr(history, "contents", None) or []
        turns = [{"role": getattr(item, "role", ""),
                  "content": getattr(item, "content", "")}
                 for item in contents]
        connection = connect()
        try:
            save_memory(connection, user_key, turns)
        finally:
            connection.close()
    except Exception:
        # 生产环境记录脱敏错误指标；记忆失败不能阻止释放 RTC 会话。
        logger.exception("保存跨会话记忆失败")
    finally:
        await session.stop()

顺序不能反：get_history() 需要智能体还在运行，先 stop 再取就什么都没有了。这是这篇教程最容易出错的一步。

另外，短期记忆里除了角色和内容，还带着轮次、时间戳、打断状态等扩展字段。本文只持久化必要的 role/content，存得越少风险越小。

六. 提供删除接口

from fastapi import Header, HTTPException

@app.post("/deleteMemory")
async def delete_memory_api(body: DeleteRequest,
                            authorization: str | None = Header(None)):
    expected = os.getenv("MEMORY_ADMIN_TOKEN")
    if not expected or authorization != f"Bearer {expected}":
        raise HTTPException(401, "invalid credential")
    key = pseudonymous_key(body.userId)
    connection = connect()
    try:
        connection.execute("DELETE FROM memories WHERE user_key=?", (key,))
        connection.commit()
    finally:
        connection.close()
    return {"code": 0, "message": "deleted"}

七. 运行与验证

cd code/cross-session-memory
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env.local
uvicorn app:app --host 0.0.0.0 --port 8000

用同一个业务用户 ID 完整走三次：

第一次会话说「我更喜欢 Python」，正常停止。
第二次会话问「你还记得我的技术偏好吗」，应该能召回。
调用删除接口，再开第三次会话，这次智能体不应该还记得。

第 3 步别省。能记住只完成了一半，能按用户要求彻底忘掉才算做完。

八. 故障排查

第二次没记忆：检查第一次是不是先成功调了 get_history() 再 stop，以及两次用的业务用户 ID 和 pepper 是否一致。
记忆串到别的用户：绝对不能用客户端可以随便改的昵称当数据库主键。所有查询都必须带上服务端认证出来的租户和用户范围。
原始对话不要无限期保存。上生产前把这条链路补全：摘要压缩、用户同意、保留期限、审计记录、删除闭环。这不只是合规要求，也是用户愿意继续用下去的前提。

九. 下一步

记忆存下来了，接着要管的是什么内容不该被说出去。

《语音智能体内容过滤：在进 TTS 之前拦住它》

在声网，连接无限可能

想进一步了解「对话式 AI 与实时互动」？欢迎注册，开启探索之旅。

注册体验

本博客为技术交流与平台行业信息分享平台，内容仅供交流参考，文章内容不代表本公司立场和观点，亦不构成任何出版或销售行为。