《AI Agent 开发入门》6.2 上下文裁剪与摘要

# minagent/compact.py
SUMMARY_SYSTEM = (
    "你是对话摘要器。把较早的对话压缩成不超过 120 字的要点，"
    "保留已确认的决定、关键数字与未决问题，不要引入原文没有的新信息。"
)

from __future__ import annotations
from typing import Callable

from minagent.types import Message
from minagent.budget import count_messages


def needs_compaction(messages: list[Message], budget: int) -> bool:
    return count_messages(messages) > budget


def render_for_summary(messages: list[Message]) -> str:
    return "\n".join(f"{m.role}: {m.content}" for m in messages)


def make_model_summarizer(model) -> Callable[[list[Message]], str]:
    """把「一段被丢弃的对话」交给模型压成摘要文本。"""
    def summarize(dropped: list[Message]) -> str:
        prompt = [
            Message(role="system", content=SUMMARY_SYSTEM),
            Message(role="user", content=render_for_summary(dropped)),
        ]
        return model.complete(prompt, tools=[]).message.content
    return summarize

def compact(messages: list[Message], budget: int, keep_recent: int = 4,
            summarizer: Callable[[list[Message]], str] | None = None
            ) -> tuple[list[Message], list[Message]]:
    """保留 system + 最近 keep_recent 条，更早的消息压成一条摘要。

    返回 (压缩后的消息, 被压缩的原始消息)。未超预算时原样返回、第二项为空。
    """
    if not needs_compaction(messages, budget):
        return messages, []
    system = [m for m in messages if m.role == "system"]
    convo = [m for m in messages if m.role != "system"]
    if len(convo) <= keep_recent:
        return messages, []
    old, recent = convo[:-keep_recent], convo[-keep_recent:]
    if summarizer is None:
        text = f"较早 {len(old)} 条对话已省略（未接摘要器）。"
    else:
        text = summarizer(old)
    note = Message(role="user", content=f"[历史摘要] {text}")
    return system + [note] + recent, old

# run62.py
import sys
sys.path.insert(0, "/tmp/ab6work")
from minagent.types import Message, ModelResponse
from minagent.models import ScriptedModel
from minagent.budget import count_messages
from minagent.compact import compact, make_model_summarizer, needs_compaction

convo = [Message(role="system", content="你是会议助理，负责跟踪任务进度。")]
for i in range(1, 7):
    convo.append(Message(role="user", content=f"第{i}轮：请汇总里程碑 {i} 的状态与风险。"))
    convo.append(Message(role="assistant",
                         content=f"第{i}轮结论：里程碑 {i} 完成度 60%，风险是接口联调。"))
convo.append(Message(role="user", content="现在把整体进度压成 3 条要点。"))

budget = 150
print(f"压缩前：{len(convo)} 条，约 {count_messages(convo)} tokens；预算 {budget}")
print(f"需要压缩吗？{needs_compaction(convo, budget)}\n")

summary = "前 6 轮里程碑均完成度约 60%，共同风险是接口联调，尚未关闭。"
summarizer = make_model_summarizer(
    ScriptedModel([ModelResponse(message=Message(role="assistant", content=summary))])
)

new, old = compact(convo, budget, keep_recent=4, summarizer=summarizer)
print(f"被压缩 {len(old)} 条 → 1 条摘要")
print(f"压缩后：{len(new)} 条，约 {count_messages(new)} tokens\n")
for m in new:
    print(f"  {m.role:9} {m.content[:40]}")

压缩前：14 条，约 264 tokens；预算 150
需要压缩吗？True

被压缩 9 条 → 1 条摘要
压缩后：6 条，约 124 tokens

  system    你是会议助理，负责跟踪任务进度。
  user      [历史摘要] 前 6 轮里程碑均完成度约 60%，共同风险是接口联调，尚未关闭。
  assistant 第5轮结论：里程碑 5 完成度 60%，风险是接口联调。
  user      第6轮：请汇总里程碑 6 的状态与风险。
  assistant 第6轮结论：里程碑 6 完成度 60%，风险是接口联调。
  user      现在把整体进度压成 3 条要点。

SUMMARY_MIN_TOKENS = 80


def plan_action(dropped: list[Message], has_summarizer: bool) -> str:
    """决定对一段旧消息是「摘要」还是「直接丢」。"""
    if not dropped:
        return "none"
    if count_messages(dropped) < SUMMARY_MIN_TOKENS:
        return "drop"
    return "summarize" if has_summarizer else "drop"

阈值： 80 tokens
短对话：约 4 tokens -> 有摘要器 drop；无摘要器 drop
长对话：约 114 tokens -> 有摘要器 summarize；无摘要器 drop