多轮对话机器人的Token优化解决方案

摘要：多轮对话场景下，上下文Token数随轮次指数级增长是制约对话质量的核心瓶颈。本文提出一种四层递进式Token压缩架构——从工具结果预算、结构化记忆、滑动窗口微压缩到应急模型摘要，逐层降级，在保留业务关键信息的前提下最大化释放上下文空间。同时给出Token估算、消息计数、历史污染阻断等关键工程细节的实现方案。

Token优化的核心思想不是"怎么用好LLM压缩上下文"，而是"怎么在不动用LLM的情况下，把Token压下去"。LLM摘要只是兜底，不是主力。

---

一、问题定义

在多轮对话系统中，每一轮交互都会将新的用户输入、模型回复、工具调用结果追加到消息列表中。当对话轮次超过一定阈值后，消息列表的Token消耗会呈现非线性增长，主要原因有三：

1. 工具返回膨胀：RAG检索、数据库查询等工具可能返回数千字符的长文本，直接注入上下文会瞬间吃掉大量Token预算。
2. 历史消息堆积：每轮对话至少包含用户消息+助手消息，若涉及工具调用还需追加AIMessage(tool_calls)和多条ToolMessage，10轮后消息列表可达30条以上。
3. 无效信息污染：安全拦截、意图拒绝等系统回复若进入历史，会干扰后续轮次的模型判断，形成"错误历史→模型跟随错误→再次拒绝"的恶性循环。

如果不做系统性优化，要么被迫使用更昂贵的长上下文模型，要么在上下文截断时丢失关键业务信息。

---

二、四层Token压缩防御架构

整体思路是从无损压缩到有损压缩，从局部裁剪到全局摘要，优先保留业务关键实体，逐步丢弃次要细节。

L1：工具结果预算与持久化缓存

工具层是Token膨胀的首要来源。设计两层约束：

单条截断：任何单条工具返回超过TOOL_RETURN_MAX_LEN（800字符）时，执行截断并生成缓存键：

def budget_tool_result(self, content: str, tool_name: str) -> str:
    if len(content) <= self.tool_result_max_len:
        return content
    content_hash = hashlib.md5(content.encode()).hexdigest()
    cache_key = f"{tool_name}:{content_hash}"
    if cache_key in self._tool_cache:
        preview = content[:self.tool_result_max_len]
        return (f"[工具结果已缓存，键:{cache_key}]\n"
                f"预览({self.tool_result_max_len}字):\n{preview}\n"
                f"如需完整数据，可引用缓存键 '{cache_key}'")
    self._tool_cache[cache_key] = content
    # 返回预览 + 缓存键引用

总预算控制：遍历历史消息中所有ToolMessage，若工具结果总字符超过TOOL_RESULT_BUDGET（12,000字符），按内容长度降序逐条压缩，直到总字符低于预算。

缓存机制的价值：同一轮或后续轮次中，若模型再次需要相同工具的超长返回结果，无需重复传输完整文本，只需引用缓存键即可。

L2：结构化记忆（硬保留实体槽位）

纯文本摘要的问题在于会丢失关键业务信息。客服场景中，用户的预算、关注产品、痛点一旦丢失，后续推荐和答疑将完全失焦。

设计SessionState作为硬保留层：

@dataclass
class SessionState:
    user_intent: str = ""        # 询价/售后/安装/投诉
    target_product: str = ""   # 关注产品型号
    budget_hint: int = 0       # 预算线索
    pain_points: Set[str] = field(default_factory=set)  # 用户痛点
    order_mentioned: str = ""  # 订单号/设备ID

每轮对话结束后，通过规则从用户文本中抽取实体并更新状态。这些实体在build_messages时以系统消息形式注入上下文：

state_msg = SystemMessage(
    content=f"【当前用户画像】{self.state.to_prompt()}\n"
            f"【历史摘要】{self.conv_summary}"
)

"硬保留"的含义：即使L3滑动窗口把历史对话压缩到只剩最近4轮，即使L4模型摘要把早期对话压缩成一句话，这些实体信息依然通过state_msg存在，永远不会丢失。

L3：滑动窗口与微压缩

在结构化记忆之上，对自由文本历史做三层压缩：

微压缩（micro_compact）：针对工具调用链的特殊处理。一条工具调用通常包含AIMessage(tool_calls)和多条ToolMessage，早期轮次的这类消息链可以压缩为单条占位提示：

if isinstance(msg, AIMessage) and getattr(msg, 'tool_calls', None):
    tool_call_ids = {tc['id'] for tc in msg.tool_calls}
    # 找到关联的ToolMessage并跳过
    # 用占位提示替代整个调用链
    compacted.append(SystemMessage(
        content=f"[Earlier tool call '{tool_name}' compacted. Re-run if needed.]"
    ))

snip_compact：当消息总数超过12条时，中间部分直接省略，只保留头部4条和尾部6条，中间用占位提示衔接：

head = messages[:4]
tail = messages[-6:]
snipped_count = len(messages) - len(head) - len(tail)
placeholder = SystemMessage(
    content=f"[对话中间 {snipped_count} 条消息已省略，关键状态：{self.state.to_prompt()}]"
)
return head + [placeholder] + tail

滑动窗口：在微压缩和snip之后，若消息数仍超过keep_recent * 2（默认8条），直接丢弃更早的消息，只保留最近N轮。

L4：应急模型摘要

当前三层仍无法将Token压到阈值以下时，触发LLM生成增量摘要。

循环合并机制：不是每次只合并最早一轮，而是持续合并直到Token低于触发阈值：

while token_count > self.summary_trigger and len(self.dialog_context) >= 2:
    old_user = self.dialog_context[0]
    old_ai = self.dialog_context[1]
    # 跳过非用户-助手对话轮（如孤立的ToolMessage）
    if not isinstance(old_user, HumanMessage) or not isinstance(old_ai, AIMessage):
        self.dialog_context = self.dialog_context[1:]
        continue
    self.conv_summary = await self._generate_summary(
        self.conv_summary, old_user.content, old_ai.content
    )
    self.dialog_context = self.dialog_context[2:]
    # 重新计算token，继续循环
    token_count = sum(self._count_msg_tokens(m) for m in temp_msgs)

兜底截断前快速摘要：若循环摘要后仍超过硬上限token_threshold，执行滑动窗口兜底截断。被截断的部分不做完全丢弃，而是提取前200字符快速追加到conv_summary：

dropped = self.dialog_context[:-keep]
if dropped:
    dropped_text = "\n".join([
        f"{'用户' if isinstance(m, HumanMessage) else '助手'}:{m.content[:100]}"
        for m in dropped
    ])
    self.conv_summary = f"{self.conv_summary}\n[历史摘要]早期{dropped_text[:200]}...".strip()
self.dialog_context = self.dialog_context[-keep:]

---

三、关键工程细节

3.1 Token估算策略

精确计数优先使用tiktoken（gpt-4编码器），未安装时降级为字符估算。降级估算的系数必须校准：

def _count_tokens_approx(text: str) -> int:
    cn = len(re.findall(r'[一-鿿]', text))
    other = len(text) - cn
    # 中文约1.3字/token，英文约4字符/token（0.25/token）
    return int(cn * 1.3 + other * 0.25)

3.2 消息级Token计数（含tool_calls开销）

AIMessage.tool_calls本身也消耗Token（JSON schema + 参数），在计数时必须纳入：

def _count_msg_tokens(self, msg: BaseMessage) -> int:
    base = _count_tokens_approx(msg.content if hasattr(msg, 'content') else str(msg))
    if isinstance(msg, AIMessage) and getattr(msg, 'tool_calls', None):
        for tc in msg.tool_calls:
            base += 50  # schema固定开销
            args_str = json.dumps(tc.get('args', ...), ensure_ascii=False)
            base += _count_tokens_approx(args_str)
    return base

3.3 拒绝回复不入历史

安全拦截或意图拒绝的回复若进入历史，会污染后续轮次的模型判断。检测拒绝特征后，只保留用户原始问题，不写入助手回复：

is_rejection = any(k in final_text for k in [
    "我仅提供扫地机器人相关咨询",
    "麻烦调整下问题", "内容违规", "无法解答"
])
if not is_rejection:
    await self.ctx.after_turn(user_msg, ai_msg, tool_calls, system_prompt=...)
else:
    self.ctx.dialog_context.append(user_msg)
    # 不写入ai_msg，阻断污染循环

3.4 预算数值对齐

TOOL_RESULT_BUDGET必须与TOKEN_THRESHOLD对齐。若工具结果总预算设为20万字符，而Token阈值仅6000，该预算永远不会触发。按校准后的估算公式，12,000字符与6,000Token阈值大致匹配。

---

四、效果与总结

四层架构的触发优先级为 L1 → L2 → L3 → L4，从无损压缩逐步过渡到有损压缩：

层级	机制	信息损失
L1	工具结果截断+缓存	无（可引用缓存键恢复）
L2	结构化记忆	无（实体硬保留）
L3	滑动窗口+微压缩	低（工具链压缩为占位提示）
L4	模型摘要	中（自由文本压缩为摘要）

该方案的核心价值在于分层决策：优先用工程手段（截断、缓存、窗口）解决问题，只在必要时调用LLM做智能摘要，既控制了成本，又确保了业务关键信息不丢失。