把老历史浓缩成一条线：滚动摘要式上下文压缩的设计复盘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

┌─────────────────────────────────────────────┐
│  messages 列表                                │
│                                               │
│  ┌──────────────┐  ← 原系统提示（人格/角色设定）│
│  │ system (原)   │     永远不动，不参与压缩      │
│  ├──────────────┤                             │
│  │ 历史摘要      │  ← 上一轮压缩产生的摘要        │
│  │ (is_summary)  │     每次压缩会被新摘要替换     │
│  ├──────────────┤                             │
│  │ old messages  │  ← 滚出窗口的老对话           │
│  │ (要被压缩的)   │     压完就扔，只留浓缩版       │
│  ├──────────────┤                             │
│  │ recent × K    │  ← 最近 K 条原文              │
│  │ (保留不压)     │     模型对"刚说的话"不能断片   │
│  └──────────────┘                             │
└─────────────────────────────────────────────┘

1

[原system, 摘要(is_summary=True), recent×4]

1
2
3

def _estimate_tokens(text: str) -> int:
    """粗略估算 token 数（无外部依赖）：中文约 0.6 token/字，英文约 4 字符/token。"""
    return max(1, int(len(text) * 0.6))

1
2
3

def _fmt(messages) -> str:
    """消息列表 → 纯文本（喂给 LLM 做摘要的原料）。"""
    return "\n".join(f"{m['role']}: {m['content']}" for m in messages)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

def compress(messages: list[dict], llm, keep: int = 4) -> list[dict]:
    """滚动摘要版压缩。llm 需要提供 summarize(text) -> str 方法。"""
    # ① 三类消息分开：原系统提示 / 旧摘要 / 其余对话
    #    用 is_summary 标记区分"摘要"和"真实系统提示"，防止混为一谈
    sys_msgs = [m for m in messages
                if m.get("role") == "system" and not m.get("is_summary")]
    old_sums = [m for m in messages if m.get("is_summary")]
    others   = [m for m in messages
                if m.get("role") != "system" and not m.get("is_summary")]

    if len(others) <= keep:
        return messages          # 还没长到需要压，原样返回

    old, recent = others[:-keep], others[-keep:]

    # ② 滚动关键：旧摘要 + 新滚出窗口的原文 → 合并成一条新摘要
    #    这就是"写周报"：把上周周报和这周的事浓缩成一份新周报，旧的扔掉
    text = _fmt(old)
    if old_sums:
        text = f"已有摘要: {old_sums[-1]['content']}\n新增对话:\n{text}"

    # ③ 交给 LLM 浓缩。生产环境的提示词要强调：保留事实/决策/偏好、控制长度
    summary = llm.summarize(text)
    summary_msg = {"role": "system",
                   "content": f"历史摘要: {summary}",
                   "is_summary": True}
    return sys_msgs + [summary_msg] + recent

1
2
3
4
5
6

def maybe_compress(messages, llm, max_tokens: int = 3000, keep: int = 4):
    """自动触发：超阈值才压，没超原样返回。"""
    if (_estimate_tokens(_fmt(messages)) > max_tokens
            and len(messages) > keep + 1):
        return compress(messages, llm, keep=keep)
    return messages

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

# memory/compress.py
"""Day4 上下文压缩：老历史 → 一条滚动摘要，永远只有一条。"""

def _estimate_tokens(text: str) -> int:
    """粗略估算 token 数（无外部依赖）：中文约 0.6 token/字，英文约 4 字符/token。"""
    return max(1, int(len(text) * 0.6))

def _fmt(messages) -> str:
    """消息列表 → 纯文本（喂给 LLM 做摘要的原料）。"""
    return "\n".join(f"{m['role']}: {m['content']}" for m in messages)

def compress(messages: list[dict], llm, keep: int = 4) -> list[dict]:
    """滚动摘要版压缩。llm 需要提供 summarize(text) -> str 方法。"""
    # ① 三类消息分开：原系统提示 / 旧摘要 / 其余对话
    #    用 is_summary 标记区分"摘要"和"真实系统提示"，防止混为一谈
    sys_msgs = [m for m in messages
                if m.get("role") == "system" and not m.get("is_summary")]
    old_sums = [m for m in messages if m.get("is_summary")]
    others   = [m for m in messages
                if m.get("role") != "system" and not m.get("is_summary")]

    if len(others) <= keep:
        return messages          # 还没长到需要压，原样返回

    old, recent = others[:-keep], others[-keep:]

    # ② 滚动关键：旧摘要 + 新滚出窗口的原文 → 合并成一条新摘要
    #    这就是"写周报"：把上周周报和这周的事浓缩成一份新周报，旧的扔掉
    text = _fmt(old)
    if old_sums:
        text = f"已有摘要: {old_sums[-1]['content']}\n新增对话:\n{text}"

    # ③ 交给 LLM 浓缩。生产环境的提示词要强调：保留事实/决策/偏好、控制长度
    summary = llm.summarize(text)
    summary_msg = {"role": "system",
                   "content": f"历史摘要: {summary}",
                   "is_summary": True}
    return sys_msgs + [summary_msg] + recent

def maybe_compress(messages, llm, max_tokens: int = 3000, keep: int = 4):
    """自动触发：超阈值才压，没超原样返回。"""
    if (_estimate_tokens(_fmt(messages)) > max_tokens
            and len(messages) > keep + 1):
        return compress(messages, llm, keep=keep)
    return messages

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

def summarize(self, text: str) -> str:
    """将对话历史浓缩为摘要。"""
    prompt = f"""请将以下对话历史浓缩为一段摘要。
要求：
1. 保留关键事实、用户偏好、已做决策、未完成任务
2. 省略寒暄、重复内容、无信息量的对话
3. 控制在200字以内
4. 用第三人称客观陈述，不要编造未提及的信息

对话内容：
{text}"""
    return self.chat([{"role": "user", "content": prompt}])

1
2
3
4
5
6

# 第一次压缩后
[system(原), system(摘要), msg_5, msg_6, msg_7, msg_8]

# 又聊了4轮，触发第二次压缩
# 如果不区分摘要和普通消息，摘要会被当 old 再压一次
# 摘要的摘要 → 信息再砍一刀

1

总 token ≈ system(固定) + 摘要(有界) + recent×K(有界)