Agent架构下如何实现省Token优化：从原理到工程实践

1次阅读

共计 2453 个字符，预计需要花费 7 分钟才能阅读完成。

最近在对接某电商客服系统时发现，一个日均处理 5000 次咨询的 Agent，每天消耗的 Token 量高达 200 万。按主流 API 价格计算，这意味着每月近 3000 元的纯文本交互成本。更严重的是，当遇到复杂问题时，长对话场景的 Token 累积可能导致响应速度下降 40% 以上。

传统方案会直接截断历史消息，但我们采用 LLM 实时生成摘要的方式。关键是在第 N 轮对话时，将前 N - 1 轮内容压缩为保持关键信息的摘要：

def generate_summary(current_summary, new_dialogue, model='gpt-3.5-turbo'):
    prompt = f''' 请将以下对话内容压缩为保留核心信息的摘要（不超过 100 字）：当前摘要：{current_summary}
    新增对话：{new_dialogue}'''

    try:
        response = openai.ChatCompletion.create(
            model=model,
            messages=[{'role': 'user', 'content': prompt}],
            temperature=0.3  # 降低随机性保证摘要稳定性
        )
        return response.choices[0].message.content
    except Exception as e:
        logging.error(f'摘要生成失败: {str(e)}')
        return current_summary  # 降级方案

收益：在测试中，将 20 轮对话上下文从平均 1800 Token 压缩到 300 Token，节省 83% 消耗。需注意避免过度压缩导致信息失真（后文会详细说明）。

通过 Faiss 建立向量索引库，当新请求到来时，先检索历史相似问答：

import faiss
import numpy as np

class SemanticCache:
    def __init__(self, dim=768):
        self.index = faiss.IndexFlatIP(dim)
        self.cache = {}

    def add(self, text, embedding, response):
        idx = self.index.ntotal
        self.index.add(np.array([embedding]).astype('float32'))
        self.cache[idx] = response

    def search(self, embedding, threshold=0.85):
        D, I = self.index.search(np.array([embedding]).astype('float32'), 1)
        return self.cache[I[0][0]] if D[0][0] >= threshold else None

实施要点 ：
– 使用 sentence-transformers 生成 Embedding
– 相似度阈值建议从 0.8 开始逐步调优
– 定期清理过期缓存（建议 TTL 设置为 24 小时）

实测效果 ：在商品咨询场景中，约 35% 的重复问题命中缓存，平均减少 2.1 次 API 调用。

当检测到后续生成内容价值低于阈值时主动终止：

def dynamic_stopping(prompt, max_tokens=200, confidence_threshold=0.7):
    result = ''
    for chunk in openai.ChatCompletion.create(
        model='gpt-3.5-turbo',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
        max_tokens=max_tokens
    ):
        # 监控每个 token 的概率值
        logprob = chunk.choices[0].logprobs 
        current_conf = np.exp(np.mean(logprob)) if logprob else 0

        result += chunk.choices[0].delta.get('content', '')

        # 连续 3 个 token 置信度低于阈值则停止
        if current_conf < confidence_threshold:
            if hasattr(dynamic_stopping, '_low_conf_count'):
                dynamic_stopping._low_conf_count += 1
                if dynamic_stopping._low_conf_count >= 3:
                    break
            else:
                dynamic_stopping._low_conf_count = 1
        else:
            dynamic_stopping._low_conf_count = 0

    return result

调优建议 ：
– 通过验证集计算不同阈值下的准确率 / 节省率曲线
– 对话场景建议 0.65-0.75，知识问答建议 0.7-0.8