ChatGPT Token 管理与优化：从成本控制到高效调用

1次阅读

没有评论

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

在大型语言模型应用中，Token 的消耗直接关系到 API 调用成本。尤其在以下场景中，这个问题尤为突出：

长对话场景 ：连续对话需要携带历史上下文，每次请求的 Token 数量会不断累积
高频调用场景 ：需要频繁与 API 交互的应用，即使单次调用 Token 不多，总量也会非常可观
内容生成长度不可控 ：当需要生成长文本时，输出的 Token 数量难以精确预测

根据 OpenAI 的定价模型，API 费用是按照 Token 数量计费的，这意味着无效或冗余的 Token 会直接增加使用成本。

OpenAI 使用 Byte Pair Encoding(BPE) 算法进行分词，这种算法：

将常见字符序列合并为单个 Token
对罕见词汇会拆分成多个子词 Token
中英文混合文本可能产生不同的 Token 分布

理解这一点很重要，因为同样的文本内容，采用不同的表达方式可能会产生不同数量的 Token。

核心思想：根据上下文重要性动态裁剪历史对话，保留关键信息。实现要点：

分析对话轮次的重要性权重
优先保留用户最近的输入和系统关键回复
设置 Token 数量阈值作为截断触发条件

对于重复或相似的查询，可以：

建立语义相似度匹配机制
对高频问题缓存标准回答
设计合理的缓存失效策略

将多个独立请求合并为一个批量请求：

减少 API 调用次数
共享相同的系统提示词
适用于可并行处理的场景

def dynamic_truncation(conversation_history, max_tokens=1024):
    """
    动态截断对话历史以控制 Token 数量
    :param conversation_history: 对话历史列表，格式 [{'role':'user','content':'...'},...]
    :param max_tokens: 允许的最大 Token 数
    :return: 截断后的对话历史
    """
    current_tokens = estimate_token_count(conversation_history)

    # 如果未超限则直接返回
    if current_tokens <= max_tokens:
        return conversation_history

    # 计算需要移除的 Token 数量
    excess_tokens = current_tokens - max_tokens

    # 从最早的对话开始移除，但保留最后两轮
    truncated_history = conversation_history.copy()
    removed_tokens = 0

    # 保留最后两轮对话
    preserve_last = 2
    to_remove = len(truncated_history) - preserve_last

    for i in range(to_remove):
        if removed_tokens >= excess_tokens:
            break

        # 估计当前轮次的 Token 数
        turn_tokens = estimate_token_count([truncated_history[0]])
        removed_tokens += turn_tokens

        # 移除最早的一轮
        truncated_history.pop(0)

    return truncated_history

# 辅助函数：估算 Token 数量（简化版，实际应使用 tiktoken 库）def estimate_token_count(messages):
    # 这里应该调用 tiktoken 进行精确计算
    return sum(len(msg['content'].split()) for msg in messages)

import asyncio
import aiohttp

async def batch_request(api_key, prompts):
    """
    异步批量请求 ChatGPT API
    :param api_key: OpenAI API 密钥
    :param prompts: 提示词列表
    :return: 响应列表
    """headers = {'Authorization': f'Bearer {api_key}','Content-Type':'application/json'
    }

    # 准备批量请求数据
    requests_data = [{
        'model': 'gpt-3.5-turbo',
        'messages': [{'role': 'user', 'content': prompt}],
        'max_tokens': 150
    } for prompt in prompts]

    async with aiohttp.ClientSession() as session:
        tasks = []
        for data in requests_data:
            task = session.post(
                'https://api.openai.com/v1/chat/completions',
                json=data,
                headers=headers
            )
            tasks.append(task)

        responses = await asyncio.gather(*tasks)
        return [await resp.json() for resp in responses]

基于实际测试数据（模拟场景）：