多轮对话缓存：增量存储与检索历史轮次 KV

Key: session:<session_id> Value: { "meta": { "created_at": "2025-01-01T00:00:00Z", "updated_at": "..." }, "turns": [ { "turn_id": "turn:1", "role": "user", "content": "你好，请帮我规划一次旅行。" }, { "turn_id": "turn:2", "role": "assistant", "content": "好的，请问您想去哪里？" } ] }

### 为什么要用 turn_id

每个轮次都分配一个唯一 ID（例如 `turn:递增序号` 或 UUID），这使我们能够：

- 精确更新某一条消息（如加入评分、用户反馈）。
- 增量插入新轮次时避免重复。
- 在检索时按 ID 范围切片，实现「跳过前 N 轮」或「仅取最近 M 轮」。

## 增量写入实现

增量写入的核心操作是：**读取当前历史 → 追加新消息 → 写回存储**。为了保证原子性，在生产环境中通常使用 Lua 脚本或事务。

下面用一个 Python 伪代码示例展示基于 Redis 的增量写入逻辑：

```python
import json
import redis

r = redis.Redis()

def append_message(session_id: str, role: str, content: str):
    key = f"session:{session_id}"
    # 1. 读取已有数据
    data = r.get(key)
    if data:
        session = json.loads(data)
    else:
        session = {"meta": {}, "turns": []}
    
    # 2. 生成新的 turn
    new_turn_id = f"turn:{len(session['turns']) + 1}"
    new_turn = {
        "turn_id": new_turn_id,
        "role": role,
        "content": content
    }
    
    # 3. 追加到轮次列表
    session["turns"].append(new_turn)
    session["meta"]["updated_at"] = "..."  # 更新元数据
    
    # 4. 写回 Redis
    r.set(key, json.dumps(session))
    return new_turn_id

如果并发不高，以上逻辑已足够。当存在大量并发时，可以改用 Redis List 结构天然支持原子追加：

RPUSH session:<session_id>:turns '{"role":"user","content":"..."}'

这样无需读取全部历史，只做一次追加操作就能完成增量写入，性能极高。

检索历史轮次

检索通常分为两种需求：获取完整上下文和获取最近的N轮。

获取完整上下文

直接拉取整个 turns 列表即可。但在轮次过长时需要配合上下文窗口裁剪，避免超出模型最大 token 限制。

def get_full_history(session_id: str):
    key = f"session:{session_id}"
    data = r.get(key)
    if not data:
        return []
    session = json.loads(data)
    return session["turns"]

获取最近的 N 轮

利用 Redis List 的负索引范围读取：

LRANGE session:<session_id>:turns -N -1

如果使用字符串存储，则在内存中切片：

def get_recent_turns(session_id: str, n: int):
    turns = get_full_history(session_id)
    return turns[-n:] if n > 0 else []

按轮次 ID 范围检索

在需要回溯特定片段时，可以存储 turn_id 到内容的映射，或者直接在列表中遍历筛选。更高效的方式是为轮次单独建立 Sorted Set（按时间戳或序号排序），然后按分页取出。

上下文窗口管理策略

多轮对话缓存需要和 token 预算 协同工作。你可以实现一个函数，从最新轮次向前累加，直到总 token 数接近上限：

def trim_context(turns, max_tokens, token_counter):
    selected = []
    total = 0
    for turn in reversed(turns):
        turn_tokens = token_counter(turn["content"])
        if total + turn_tokens > max_tokens:
            break
        selected.insert(0, turn)
        total += turn_tokens
    return selected