大模型Token优化与上下文管理完全教程

从原理到实战，系统掌握大模型应用中的Token优化、上下文窗口管理和成本控制策略。

目录

1. Token基础概念与计费原理
2. Token计算方法与估算工具
3. Prompt压缩技术
4. 上下文窗口管理策略
5. 长文本处理：分块、摘要与检索
6. 上下文缓存（Prompt Caching）
7. 多轮对话Token控制
8. Token优化工具链
9. 成本监控与告警
10. 企业级Token管理方案
11. 各模型Token限制对比
12. 最佳实践总结

1. Token基础概念与计费原理

1.1 什么是Token

Token是大语言模型（LLM）处理文本的最小单位。它不等同于"字"或"词"——模型在训练时通过分词器（Tokenizer）将文本切分为Token序列。不同模型使用不同的分词策略，因此同一段文本在不同模型中的Token数可能不同。

常见Token化示例（以GPT系列为例）：

关键点： 中文文本的Token消耗通常比同等语义的英文高出50%-100%，因为大多数主流模型的分词器以英文语料为主训练。

1.2 Token计费模型

LLM API的计费通常按以下维度计算：

总费用 = (输入Token数 × 输入单价) + (输出Token数 × 输出单价)

为什么输入和输出价格不同？

• 输入（Prompt Tokens）：可以被预计算（KV Cache），计算成本相对固定
• 输出（Completion Tokens）：逐Token生成，每个Token都需要完整的前向推理，计算成本与输出长度线性相关

典型定价对比（2025年参考）：

实际成本陷阱：很多开发者只关注输入Token优化，忽略了输出Token才是成本大头。一个典型的对话场景中，输出Token费用往往是输入的3-5倍。

1.3 Token计费的隐藏成本

除了直接的API调用费用，Token相关的隐藏成本还包括：

• 上下文膨胀：多轮对话中历史消息的重复计费
• 系统提示（System Prompt）：每轮对话都要重新发送，容易被忽略
• 格式开销：JSON结构化输出中的引号、括号、字段名都是Token
• 重试成本：API调用失败后的重试会重复消耗Token
• 工具调用：Function Calling的工具描述也会占用上下文Token

2. Token计算方法与估算工具

2.1 主流分词器实现

不同模型家族使用不同的分词器，了解它们有助于精确估算Token数。

OpenAI系列（tiktoken）：

import tiktoken

# GPT-4o / GPT-4o-mini 使用 o200k_base
enc = tiktoken.encoding_for_model("gpt-4o")
text = "大语言模型正在改变软件开发的方式"
tokens = enc.encode(text)
print(f"Token数量: {len(tokens)}")  # 通常 10-14 个Token
print(f"Token列表: {[enc.decode([t]) for t in tokens]}")

Claude系列（Anthropic Tokenizer）：

# Anthropic 提供了在线tokenizer工具
# 也可使用 approximation 方法
def estimate_claude_tokens(text: str) -> int:
    """粗略估算Claude模型的Token数"""
    # 英文：约 1 token / 4 字符
    # 中文：约 1 token / 1.5 字符
    chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
    other_chars = len(text) - chinese_chars
    return int(chinese_chars / 1.5 + other_chars / 4)

通用估算函数：

import re

def estimate_tokens(text: str, model: str = "gpt-4") -> int:
    """
    通用Token估算（无需依赖特定分词器库）
    精度约 ±15%，适合快速预算估算
    """
    # 统计中文字符
    chinese = len(re.findall(r'[\u4e00-\u9fff\u3400-\u4dbf]', text))
    # 统计英文单词
    words = len(re.findall(r'[a-zA-Z]+', text))
    # 统计数字
    numbers = len(re.findall(r'\d+', text))
    # 统计标点和特殊字符
    specials = len(re.findall(r'[^\w\s]', text))
    
    if 'gpt-4' in model or 'gpt-3.5' in model:
        return int(chinese * 1.5 + words * 1.3 + numbers * 1.2 + specials * 0.5)
    elif 'claude' in model:
        return int(chinese * 1.3 + words * 1.25 + numbers * 1.1 + specials * 0.5)
    else:
        return int(chinese * 1.5 + words * 1.3 + numbers * 1.2 + specials * 0.5)

# 使用示例
prompt = """你是一个专业的技术写作者。请根据以下要求撰写一篇关于微服务架构的技术博客：
1. 介绍微服务的核心概念
2. 对比单体架构和微服务架构
3. 给出实际的部署建议"""

print(f"预估Token数: {estimate_tokens(prompt, 'gpt-4')}")

2.2 API返回的Token统计

大多数LLM API在响应中直接返回Token使用量，这是最准确的数据源：

from openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "解释量子计算的基本原理"}],
    max_tokens=500
)

usage = response.usage
print(f"输入Token: {usage.prompt_tokens}")
print(f"输出Token: {usage.completion_tokens}")
print(f"总计Token: {usage.total_tokens}")
# GPT-4o 还会返回 cached_tokens（如果有缓存命中）
if hasattr(usage, 'prompt_tokens_details'):
    print(f"缓存命中: {usage.prompt_tokens_details.cached_tokens}")

3. Prompt压缩技术

3.1 指令压缩

核心思想：在不损失语义的前提下，用更少的Token表达相同的意思。

压缩前（约120 tokens）：

你是一个非常有经验的高级软件工程师，你在代码审查方面有超过10年的丰富经验。
请你仔细地、全面地审查以下代码，找出其中存在的所有潜在问题，包括但不限于：
代码质量问题、安全漏洞、性能问题、可维护性问题等。请用结构化的方式输出你的审查结果。

压缩后（约45 tokens）：

作为资深工程师审查代码。检查质量、安全、性能、可维护性。结构化输出。

压缩技巧：

3.2 上下文压缩

def compress_context(messages: list, max_tokens: int = 4000) -> list:
    """
    智能压缩对话历史，保留关键信息
    
    策略：
    1. 保留系统消息（不压缩）
    2. 保留最近N轮对话
    3. 中间轮次用摘要替代
    """
    if not messages:
        return messages
    
    system_msgs = [m for m in messages if m['role'] == 'system']
    non_system = [m for m in messages if m['role'] != 'system']
    
    if estimate_total_tokens(non_system) <= max_tokens:
        return messages
    
    # 保留最近 4 轮对话
    recent = non_system[-8:]  # 4轮 = 8条消息
    
    # 将更早的消息压缩为摘要
    older = non_system[:-8]
    if older:
        summary = summarize_messages(older)
        summary_msg = {
            "role": "system",
            "content": f"[历史对话摘要] {summary}"
        }
        return system_msgs + [summary_msg] + recent
    
    return system_msgs + recent

def summarize_messages(messages: list) -> str:
    """用LLM生成对话摘要（或使用简单规则提取关键信息）"""
    # 实际实现中可以调用LLM进行摘要
    # 这里用简单规则演示
    topics = []
    for msg in messages:
        if msg['role'] == 'user':
            # 提取用户消息的关键句（取前50字）
            topics.append(msg['content'][:50])
    return "；".join(topics[-5:])  # 最多保留5个关键点

3.3 结构化Prompt模板

使用模板化的方式减少重复Token消耗：

class CompactPromptTemplate:
    """Token友好的Prompt模板系统"""
    
    def __init__(self):
        self.templates = {}
    
    def register(self, name: str, template: str):
        """注册模板，使用变量占位符"""
        self.templates[name] = template
    
    def render(self, name: str, **kwargs) -> str:
        """渲染模板并计算Token"""
        template = self.templates[name]
        result = template.format(**kwargs)
        token_count = estimate_tokens(result)
        return result, token_count

# 使用示例
tmpl = CompactPromptTemplate()
tmpl.register("code_review", "审查{lang}代码。关注{concerns}。{format}")

result, tokens = tmpl.render(
    "code_review",
    lang="Python",
    concerns="安全、性能",
    format="Markdown表格输出"
)
print(f"渲染结果: {result}")
print(f"Token数: {tokens}")
# 输出: "审查Python代码。关注安全、性能。Markdown表格输出"
# Token数: ~25

4. 上下文窗口管理策略

4.1 上下文窗口的构成

一个典型的LLM请求中，上下文窗口的使用分布如下：

┌─────────────────────────────────────────────┐
│              上下文窗口 (Context Window)       │
├──────────┬──────────┬──────────┬─────────────┤
│ System   │ History  │ Current  │  Reserved   │
│ Prompt   │ Messages │ Query    │  for Output │
├──────────┼──────────┼──────────┼─────────────┤
│  ~500 tk │  变化     │ ~200 tk  │  ~2000 tk   │
└──────────┴──────────┴──────────┴─────────────┘

预算分配建议：

4.2 滑动窗口策略

from collections import deque
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class MessageWindow:
    """
    滑动窗口消息管理器
    
    核心思想：维持一个固定大小的消息窗口，
    当超出限制时，丢弃最早的消息（保留系统消息）。
    """
    max_tokens: int = 8000
    system_messages: list = field(default_factory=list)
    _messages: deque = field(default_factory=deque)
    _current_tokens: int = 0
    
    def set_system(self, messages: list):
        """设置系统消息（不参与滑动）"""
        self.system_messages = messages
    
    def add_message(self, role: str, content: str) -> bool:
        """添加消息，自动管理窗口大小"""
        new_tokens = estimate_tokens(content)
        
        # 检查是否需要淘汰旧消息
        while (self._current_tokens + new_tokens > self.max_tokens 
               and len(self._messages) > 0):
            removed = self._messages.popleft()
            self._current_tokens -= estimate_tokens(removed['content'])
        
        self._messages.append({"role": role, "content": content})
        self._current_tokens += new_tokens
        return True
    
    def get_messages(self) -> list:
        """获取完整的上下文消息列表"""
        return self.system_messages + list(self._messages)
    
    @property
    def current_tokens(self) -> int:
        return self._current_tokens

# 使用示例
window = MessageWindow(max_tokens=4000)
window.set_system([{"role": "system", "content": "你是技术助手"}])

# 模拟多轮对话
for i in range(20):
    window.add_message("user", f"这是第{i+1}个问题，请详细解答。" * 3)
    window.add_message("assistant", f"这是第{i+1}个回答。" * 5)

print(f"当前Token使用: {window.current_tokens}")
print(f"消息数量: {len(window.get_messages())}")

4.3 优先级淘汰策略

不是所有历史消息的重要性相同。基于优先级的淘汰策略能保留更有价值的上下文：

import heapq
from enum import IntEnum

class MessagePriority(IntEnum):
    SYSTEM = 100      # 系统消息，永不淘汰
    KEY_FACTS = 90     # 关键事实（如用户名、偏好）
    RECENT = 80        # 最近的消息
    DECISION = 70      # 包含决策的消息
    CONTEXT = 50       # 一般上下文
    FILLER = 20        # 闲聊、寒暄

@dataclass
class PrioritizedMessage:
    priority: MessagePriority
    index: int
    message: dict
    
    def __lt__(self, other):
        return self.priority < other.priority

class PriorityWindowManager:
    """基于优先级的上下文窗口管理"""
    
    def __init__(self, max_tokens: int = 8000):
        self.max_tokens = max_tokens
        self.messages: list[PrioritizedMessage] = []
    
    def add(self, role: str, content: str, 
            priority: MessagePriority = MessagePriority.CONTEXT):
        msg = PrioritizedMessage(
            priority=priority,
            index=len(self.messages),
            message={"role": role, "content": content}
        )
        self.messages.append(msg)
    
    def build_context(self) -> list[dict]:
        """构建满足Token限制的最优上下文"""
        total_tokens = 0
        selected = []
        
        # 按优先级排序（高优先级在前），同优先级按时间排序
        sorted_msgs = sorted(self.messages, 
                           key=lambda m: (-m.priority, m.index))
        
        for pm in sorted_msgs:
            msg_tokens = estimate_tokens(pm.message['content'])
            if total_tokens + msg_tokens <= self.max_tokens:
                selected.append(pm)
                total_tokens += msg_tokens
        
        # 恢复时间顺序
        selected.sort(key=lambda m: m.index)
        return [pm.message for pm in selected]

5. 长文本处理：分块、摘要与检索

5.1 文本分块（Chunking）

当输入文本超过上下文窗口时，需要将其切分为可管理的块：

import re
from typing import Generator

def chunk_text(text: str, chunk_size: int = 2000, 
               overlap: int = 200) -> Generator[str, None, None]:
    """
    智能文本分块，按自然边界切分
    
    优先级：段落 > 句子 > 词边界 > 字符
    """
    # 按段落分割
    paragraphs = re.split(r'\n\s*\n', text)
    
    current_chunk = ""
    
    for para in paragraphs:
        para = para.strip()
        if not para:
            continue
        
        # 如果当前块加上新段落不超限
        if estimate_tokens(current_chunk + para) <= chunk_size:
            current_chunk += ("\n\n" + para if current_chunk else para)
        else:
            # 输出当前块
            if current_chunk:
                yield current_chunk
            
            # 如果单个段落就超限，按句子切分
            if estimate_tokens(para) > chunk_size:
                sentences = re.split(r'([。！？.!?])', para)
                current_chunk = ""
                for i in range(0, len(sentences), 2):
                    sent = sentences[i]
                    if i + 1 < len(sentences):
                        sent += sentences[i + 1]
                    
                    if estimate_tokens(current_chunk + sent) <= chunk_size:
                        current_chunk += sent
                    else:
                        if current_chunk:
                            yield current_chunk
                        current_chunk = sent
            else:
                current_chunk = para
    
    if current_chunk:
        yield current_chunk

# 使用示例
long_text = open("technical_doc.md").read()
for i, chunk in enumerate(chunk_text(long_text, chunk_size=1500)):
    tokens = estimate_tokens(chunk)
    print(f"块 {i+1}: {tokens} tokens, {len(chunk)} 字符")

5.2 Map-Reduce摘要

处理超长文档的经典策略——先分块摘要，再合并总结：

from openai import OpenAI

client = OpenAI()

def map_reduce_summarize(text: str, model: str = "gpt-4o-mini") -> str:
    """
    Map-Reduce 长文档摘要
    
    流程：
    1. 将文档分块
    2. 对每个块生成摘要 (Map)
    3. 合并所有摘要生成最终总结 (Reduce)
    """
    # Step 1: 分块
    chunks = list(chunk_text(text, chunk_size=3000))
    
    if len(chunks) == 1:
        # 文档够短，直接摘要
        return _summarize_chunk(chunks[0], model)
    
    # Step 2: Map - 对每个块生成摘要
    chunk_summaries = []
    for i, chunk in enumerate(chunks):
        print(f"摘要处理中... 块 {i+1}/{len(chunks)}")
        summary = _summarize_chunk(chunk, model)
        chunk_summaries.append(summary)
    
    # Step 3: Reduce - 合并所有摘要
    combined = "\n\n".join([
        f"## 第{i+1}部分摘要\n{s}" 
        for i, s in enumerate(chunk_summaries)
    ])
    
    final_summary = client.chat.completions.create(
        model=model,
        messages=[{
            "role": "user",
            "content": f"请基于以下各部分摘要，生成一份完整的文档总结（500字以内）：\n\n{combined}"
        }],
        max_tokens=1000
    )
    
    return final_summary.choices[0].message.content

def _summarize_chunk(chunk: str, model: str) -> str:
    response = client.chat.completions.create(
        model=model,
        messages=[{
            "role": "user",
            "content": f"请用200字以内总结以下内容的关键信息：\n\n{chunk}"
        }],
        max_tokens=400
    )
    return response.choices[0].message.content

5.3 RAG（检索增强生成）

RAG是处理长文档和知识库的标准方案：

from dataclasses import dataclass
import numpy as np

@dataclass
class Document:
    content: str
    metadata: dict
    embedding: list[float] = None

class SimpleRAG:
    """简化的RAG实现，展示核心流程"""
    
    def __init__(self, llm_client):
        self.documents: list[Document] = []
        self.client = llm_client
    
    def add_documents(self, texts: list[str], metadata_list: list[dict] = None):
        """添加文档到知识库"""
        for i, text in enumerate(texts):
            chunks = list(chunk_text(text, chunk_size=1000))
            for chunk in chunks:
                meta = metadata_list[i] if metadata_list else {}
                embedding = self._get_embedding(chunk)
                self.documents.append(Document(
                    content=chunk,
                    metadata=meta,
                    embedding=embedding
                ))
    
    def query(self, question: str, top_k: int = 3) -> str:
        """检索相关文档并生成回答"""
        # 1. 获取问题的向量
        q_embedding = self._get_embedding(question)
        
        # 2. 计算相似度并排序
        scored = []
        for doc in self.documents:
            similarity = self._cosine_similarity(q_embedding, doc.embedding)
            scored.append((similarity, doc))
        scored.sort(key=lambda x: -x[0])
        
        # 3. 取Top-K相关文档
        context = "\n\n---\n\n".join([
            doc.content for _, doc in scored[:top_k]
        ])
        
        # 4. 生成回答
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "基于提供的上下文回答问题。如果上下文中没有相关信息，请明确说明。"},
                {"role": "user", "content": f"上下文：\n{context}\n\n问题：{question}"}
            ]
        )
        return response.choices[0].message.content
    
    def _get_embedding(self, text: str) -> list[float]:
        response = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding
    
    @staticmethod
    def _cosine_similarity(a: list[float], b: list[float]) -> float:
        a, b = np.array(a), np.array(b)
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

6. 上下文缓存（Prompt Caching）

6.1 Prompt Caching原理

Prompt Caching是2024年以来各厂商推出的重要优化特性。其核心思想是：当多个请求共享相同的前缀（如System Prompt）时，可以缓存该前缀的计算结果（KV Cache），避免重复计算。

工作流程：

请求1: [System Prompt + 问题A]  → 完整计算
请求2: [System Prompt + 问题B]  → 命中缓存前缀，只计算问题B部分
请求3: [System Prompt + 问题C]  → 命中缓存前缀，只计算问题C部分

6.2 各平台缓存策略对比

6.3 最大化缓存命中率

# 好的做法：固定前缀，变化后缀
def build_messages_with_cache(
    system_prompt: str,      # 固定，适合缓存
    context: str,            # 可缓存（如知识库内容）
    user_query: str          # 每次变化
) -> list[dict]:
    """构建消息列表，优化缓存命中"""
    return [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": f"参考文档：\n{context}"},
        {"role": "user", "content": user_query}
    ]

# 坏的做法：每次随机插入不同内容
# 这会导致前缀不同，无法命中缓存
def bad_message_builder(system_prompt, random_context, query):
    return [
        {"role": "system", "content": f"[时间: {time.now()}] {system_prompt}"},  # ❌ 时间戳破坏缓存
        {"role": "user", "content": random_context},
        {"role": "user", "content": query}
    ]

缓存优化要点：

1. 保持前缀稳定：System Prompt和静态上下文放在消息列表最前面
2. 避免动态元素：不要在前缀中插入时间戳、随机ID等变化内容
3. 批量处理：将多个相似请求攒批处理，共享缓存前缀
4. 显式控制：在支持的平台上使用显式缓存API，而非依赖自动缓存

# Anthropic 显式缓存示例
import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "你是一个技术文档助手...",  # 大段系统提示
            "cache_control": {"type": "ephemeral"}  # 显式标记缓存
        }
    ],
    messages=[{"role": "user", "content": "你的问题"}]
)

# 检查缓存命中
print(f"缓存创建: {response.usage.cache_creation_input_tokens}")
print(f"缓存读取: {response.usage.cache_read_input_tokens}")

7. 多轮对话Token控制

7.1 对话Token预算模型

from dataclasses import dataclass, field
from typing import Optional

@dataclass 
class ConversationBudget:
    """
    多轮对话Token预算管理
    
    动态分配每轮对话可用的Token资源，
    确保在有限的上下文窗口内维持有效对话。
    """
    model_context_window: int = 128000   # 模型上下文窗口大小
    system_prompt_tokens: int = 500      # 系统提示占用
    output_reserve: int = 4096           # 输出预留
    safety_margin: float = 0.1           # 安全余量
    
    # 对话历史
    history: list[dict] = field(default_factory=list)
    history_tokens: int = 0
    
    @property
    def available_tokens(self) -> int:
        """当前可用于新输入的Token数"""
        total_budget = int(
            self.model_context_window * (1 - self.safety_margin)
        )
        used = self.system_prompt_tokens + self.history_tokens + self.output_reserve
        return max(0, total_budget - used)
    
    def add_turn(self, user_msg: str, assistant_msg: str):
        """添加一轮对话"""
        user_tokens = estimate_tokens(user_msg)
        assistant_tokens = estimate_tokens(assistant_msg)
        
        self.history.append({"role": "user", "content": user_msg})
        self.history.append({"role": "assistant", "content": assistant_msg})
        self.history_tokens += user_tokens + assistant_tokens
        
        # 如果超出预算，执行淘汰
        self._trim_if_needed()
    
    def _trim_if_needed(self):
        """当历史超出预算时，淘汰早期消息"""
        while self.history_tokens > self._history_budget():
            if len(self.history) <= 2:
                break  # 至少保留1轮
            removed = self.history.pop(0)
            self.history_tokens -= estimate_tokens(removed['content'])
            removed = self.history.pop(0)
            self.history_tokens -= estimate_tokens(removed['content'])
    
    def _history_budget(self) -> int:
        """历史消息可用的Token预算"""
        return int(self.model_context_window * 0.4)  # 最多用40%给历史

7.2 消息摘要压缩

当对话轮次过多时，将早期消息压缩为摘要：

class ConversationWithSummary:
    """带摘要压缩的对话管理"""
    
    def __init__(self, max_history_tokens: int = 6000):
        self.max_history = max_history_tokens
        self.summary: Optional[str] = None
        self.recent_messages: list[dict] = []
        self.recent_tokens: int = 0
    
    def add_message(self, role: str, content: str, client):
        self.recent_messages.append({"role": role, "content": content})
        self.recent_tokens += estimate_tokens(content)
        
        # 超出限制时压缩
        if self.recent_tokens > self.max_history:
            self._compress(client)
    
    def _compress(self, client):
        """将旧消息压缩为摘要，保留最近的对话"""
        # 保留最近4轮
        keep_count = min(8, len(self.recent_messages))
        to_compress = self.recent_messages[:-keep_count]
        self.recent_messages = self.recent_messages[-keep_count:]
        
        if not to_compress:
            return
        
        # 生成摘要
        compress_text = "\n".join([
            f"{m['role']}: {m['content'][:200]}" 
            for m in to_compress
        ])
        
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"用100字总结以下对话的要点：\n{compress_text}"
            }],
            max_tokens=200
        )
        
        new_summary = response.choices[0].message.content
        
        # 合并已有摘要
        if self.summary:
            self.summary = f"{self.summary}\n{new_summary}"
        else:
            self.summary = new_summary
        
        # 重新计算Token
        self.recent_tokens = sum(
            estimate_tokens(m['content']) for m in self.recent_messages
        )
    
    def get_context(self) -> list[dict]:
        """获取完整的对话上下文"""
        messages = []
        if self.summary:
            messages.append({
                "role": "system",
                "content": f"[对话历史摘要]\n{self.summary}"
            })
        messages.extend(self.recent_messages)
        return messages

8. Token优化工具链

8.1 常用工具一览

8.2 LangChain Token管理

from langchain.callbacks import get_openai_callback
from langchain.chat_models import ChatOpenAI
from langchain.chains import ConversationChain
from langchain.memory import ConversationSummaryBufferMemory

llm = ChatOpenAI(model="gpt-4o", max_tokens=1000)

# 使用摘要缓冲记忆 - 自动管理对话历史
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # 超过2000 token时自动摘要
    return_messages=True
)

chain = ConversationChain(llm=llm, memory=memory)

# 追踪Token使用
with get_openai_callback() as cb:
    result = chain.invoke({"input": "你好，我想了解微服务架构"})
    print(f"总Token: {cb.total_tokens}")
    print(f"总费用: ${cb.total_cost:.4f}")

8.3 自定义Token中间件

from functools import wraps
import time

class TokenTracker:
    """Token使用追踪中间件"""
    
    def __init__(self, daily_budget: float = 10.0):
        self.daily_budget = daily_budget
        self.daily_usage: dict[str, float] = {}
        self.call_log: list[dict] = []
    
    def track(self, func):
        """装饰器：追踪函数调用的Token消耗"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            today = time.strftime("%Y-%m-%d")
            
            # 检查日预算
            if self.daily_usage.get(today, 0) >= self.daily_budget:
                raise BudgetExceededError(
                    f"日预算 ${self.daily_budget} 已用完"
                )
            
            result = func(*args, **kwargs)
            
            # 记录使用量
            if hasattr(result, 'usage'):
                cost = self._calculate_cost(result)
                self.daily_usage[today] = self.daily_usage.get(today, 0) + cost
                self.call_log.append({
                    "time": time.time(),
                    "function": func.__name__,
                    "tokens": result.usage.total_tokens,
                    "cost": cost
                })
            
            return result
        return wrapper
    
    def _calculate_cost(self, response) -> float:
        """根据模型和Token数计算费用"""
        # 简化示例，实际需根据模型查表
        usage = response.usage
        input_cost = usage.prompt_tokens * 2.5 / 1_000_000
        output_cost = usage.completion_tokens * 10.0 / 1_000_000
        return input_cost + output_cost

class BudgetExceededError(Exception):
    pass

# 使用
tracker = TokenTracker(daily_budget=5.0)

@tracker.track
def call_llm(messages, model="gpt-4o"):
    client = OpenAI()
    return client.chat.completions.create(model=model, messages=messages)

9. 成本监控与告警

9.1 监控体系架构

┌─────────────────────────────────────────────────┐
│                 成本监控系统                       │
├─────────────────────────────────────────────────┤
│  ┌─────────┐  ┌─────────┐  ┌─────────┐         │
│  │ 数据采集 │→│ 存储聚合 │→│ 告警引擎 │         │
│  └─────────┘  └─────────┘  └─────────┘         │
│       │            │            │               │
│  API回调日志   TimescaleDB   阈值告警           │
│  应用埋点      Prometheus    趋势异常           │
│  代理网关      Grafana       预算封顶           │
└─────────────────────────────────────────────────┘

9.2 实现成本追踪服务

import json
import time
from datetime import datetime, timedelta
from collections import defaultdict

class CostMonitor:
    """Token成本监控与告警服务"""
    
    # 模型定价表（/1M tokens）
    PRICING = {
        "gpt-4o":          {"input": 2.5,  "output": 10.0},
        "gpt-4o-mini":     {"input": 0.15, "output": 0.6},
        "claude-3-5-sonnet": {"input": 3.0, "output": 15.0},
        "deepseek-chat":   {"input": 0.14, "output": 0.28},
    }
    
    def __init__(self):
        self.usage_log: list[dict] = []
        self.alerts: list[dict] = []
        self.budgets = {
            "daily": 50.0,
            "weekly": 200.0,
            "monthly": 500.0,
        }
        self.alert_thresholds = {
            "daily": 0.8,    # 80%时告警
            "weekly": 0.75,
            "monthly": 0.7,
        }
    
    def log_usage(self, model: str, input_tokens: int, 
                  output_tokens: int, metadata: dict = None):
        """记录一次API调用的Token使用"""
        pricing = self.PRICING.get(model, {"input": 5.0, "output": 15.0})
        cost = (input_tokens * pricing["input"] + 
                output_tokens * pricing["output"]) / 1_000_000
        
        entry = {
            "timestamp": datetime.now().isoformat(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost": round(cost, 6),
            "metadata": metadata or {}
        }
        self.usage_log.append(entry)
        
        # 检查告警
        self._check_alerts()
        
        return entry
    
    def get_usage_summary(self, period: str = "daily") -> dict:
        """获取指定时间段的使用摘要"""
        now = datetime.now()
        if period == "daily":
            start = now.replace(hour=0, minute=0, second=0)
        elif period == "weekly":
            start = now - timedelta(days=now.weekday())
            start = start.replace(hour=0, minute=0, second=0)
        elif period == "monthly":
            start = now.replace(day=1, hour=0, minute=0, second=0)
        else:
            start = now - timedelta(days=1)
        
        filtered = [
            e for e in self.usage_log
            if datetime.fromisoformat(e["timestamp"]) >= start
        ]
        
        total_cost = sum(e["cost"] for e in filtered)
        total_input = sum(e["input_tokens"] for e in filtered)
        total_output = sum(e["output_tokens"] for e in filtered)
        
        # 按模型分组
        by_model = defaultdict(lambda: {"calls": 0, "cost": 0})
        for e in filtered:
            by_model[e["model"]]["calls"] += 1
            by_model[e["model"]]["cost"] += e["cost"]
        
        return {
            "period": period,
            "total_calls": len(filtered),
            "total_cost": round(total_cost, 4),
            "total_input_tokens": total_input,
            "total_output_tokens": total_output,
            "budget": self.budgets.get(period, 0),
            "budget_usage_pct": round(total_cost / self.budgets.get(period, 1) * 100, 1),
            "by_model": dict(by_model)
        }
    
    def _check_alerts(self):
        """检查是否需要触发告警"""
        for period in ["daily", "weekly", "monthly"]:
            summary = self.get_usage_summary(period)
            usage_pct = summary["budget_usage_pct"] / 100
            threshold = self.alert_thresholds[period]
            
            if usage_pct >= threshold:
                alert = {
                    "level": "warning" if usage_pct < 1.0 else "critical",
                    "period": period,
                    "usage_pct": f"{usage_pct:.0%}",
                    "cost": summary["total_cost"],
                    "budget": summary["budget"],
                    "message": f"⚠️ {period}预算已使用{usage_pct:.0%}"
                }
                self.alerts.append(alert)
                self._send_alert(alert)
    
    def _send_alert(self, alert: dict):
        """发送告警通知"""
        # 实际实现中可接入邮件、钉钉、Slack等
        print(f"[ALERT] {alert['level'].upper()}: {alert['message']}")

10. 企业级Token管理方案

10.1 架构设计

┌──────────────────────────────────────────────────────────┐
│                    AI Gateway（API网关层）                 │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐   │
│  │ 认证鉴权  │ │ 限流控制  │ │ 路由分发  │ │ 缓存代理  │   │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘   │
├──────────────────────────────────────────────────────────┤
│                    应用层                                  │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐   │
│  │ Prompt    │ │ Token    │ │ 成本     │ │ 审计     │   │
│  │ 管理中心  │ │ 预算引擎  │ │ 分析平台 │ │ 日志     │   │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘   │
├──────────────────────────────────────────────────────────┤
│                    数据层                                  │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐                 │
│  │ 使用量DB  │ │ 用户配额  │ │ 模型定价  │                 │
│  └──────────┘ └──────────┘ └──────────┘                 │
└──────────────────────────────────────────────────────────┘

10.2 多租户配额管理

from datetime import datetime, timedelta
from enum import Enum

class PlanTier(Enum):
    FREE = "free"
    BASIC = "basic"
    PRO = "professional"
    ENTERPRISE = "enterprise"

class TenantQuota:
    """多租户Token配额管理"""
    
    TIER_LIMITS = {
        PlanTier.FREE:       {"daily_tokens": 100_000,   "daily_cost": 1.0,   "max_tokens_per_req": 4096},
        PlanTier.BASIC:      {"daily_tokens": 1_000_000, "daily_cost": 10.0,  "max_tokens_per_req": 8192},
        PlanTier.PRO:        {"daily_tokens": 10_000_000,"daily_cost": 100.0, "max_tokens_per_req": 32768},
        PlanTier.ENTERPRISE: {"daily_tokens": -1,        "daily_cost": -1,    "max_tokens_per_req": 128000},
    }
    
    def __init__(self):
        self.tenants: dict[str, dict] = {}
    
    def register_tenant(self, tenant_id: str, tier: PlanTier):
        self.tenants[tenant_id] = {
            "tier": tier,
            "daily_usage": {"tokens": 0, "cost": 0.0, "date": datetime.now().date()},
            "rate_limiter": {"requests": 0, "window_start": datetime.now()},
        }
    
    def check_quota(self, tenant_id: str, 
                    estimated_tokens: int) -> tuple[bool, str]:
        """检查请求是否在配额内"""
        tenant = self.tenants.get(tenant_id)
        if not tenant:
            return False, "租户不存在"
        
        tier = tenant["tier"]
        limits = self.TIER_LIMITS[tier]
        
        # 企业版无限制
        if limits["daily_tokens"] == -1:
            return True, "OK"
        
        # 重置日使用量（如果是新的一天）
        self._reset_daily_if_needed(tenant)
        
        # 检查单次请求限制
        if estimated_tokens > limits["max_tokens_per_req"]:
            return False, f"单次请求超过限制 ({limits['max_tokens_per_req']} tokens)"
        
        # 检查日额度
        if tenant["daily_usage"]["tokens"] + estimated_tokens > limits["daily_tokens"]:
            return False, "日Token额度即将用完"
        
        return True, "OK"
    
    def record_usage(self, tenant_id: str, tokens: int, cost: float):
        """记录使用量"""
        tenant = self.tenants[tenant_id]
        self._reset_daily_if_needed(tenant)
        tenant["daily_usage"]["tokens"] += tokens
        tenant["daily_usage"]["cost"] += cost
    
    def _reset_daily_if_needed(self, tenant: dict):
        today = datetime.now().date()
        if tenant["daily_usage"]["date"] != today:
            tenant["daily_usage"] = {"tokens": 0, "cost": 0.0, "date": today}

10.3 Prompt版本管理

import hashlib
import json

class PromptRegistry:
    """Prompt模板注册与版本管理"""
    
    def __init__(self):
        self.prompts: dict[str, dict] = {}
    
    def register(self, name: str, template: str, 
                 metadata: dict = None) -> str:
        """注册Prompt模板，返回版本ID"""
        version_id = hashlib.md5(template.encode()).hexdigest()[:8]
        
        if name not in self.prompts:
            self.prompts[name] = {"versions": [], "current": None}
        
        version = {
            "id": version_id,
            "template": template,
            "token_count": estimate_tokens(template),
            "created_at": datetime.now().isoformat(),
            "metadata": metadata or {}
        }
        
        self.prompts[name]["versions"].append(version)
        self.prompts[name]["current"] = version_id
        
        return version_id
    
    def get(self, name: str, version_id: str = None) -> str:
        """获取指定版本的Prompt"""
        prompt = self.prompts[name]
        if version_id:
            for v in prompt["versions"]:
                if v["id"] == version_id:
                    return v["template"]
        # 返回当前版本
        current_id = prompt["current"]
        return self.get(name, current_id)
    
    def compare_tokens(self, name: str, 
                       v1_id: str, v2_id: str) -> dict:
        """对比两个版本的Token差异"""
        v1 = next(v for v in self.prompts[name]["versions"] if v["id"] == v1_id)
        v2 = next(v for v in self.prompts[name]["versions"] if v["id"] == v2_id)
        diff = v2["token_count"] - v1["token_count"]
        return {
            "v1_tokens": v1["token_count"],
            "v2_tokens": v2["token_count"],
            "diff": diff,
            "pct_change": f"{diff/v1['token_count']*100:+.1f}%"
        }

11. 各模型Token限制对比

11.1 主流模型上下文窗口

11.2 如何选择合适的模型

def select_optimal_model(
    input_tokens: int,
    output_tokens: int,
    budget: float,
    needs_reasoning: bool = False,
    needs_long_context: bool = False
) -> str:
    """
    根据需求自动选择最优模型
    
    考虑因素：上下文大小、输出长度、预算、是否需要推理
    """
    candidates = [
        {
            "name": "gpt-4o-mini",
            "context": 128000, "max_output": 16000,
            "input_price": 0.15, "output_price": 0.6,
            "reasoning": False
        },
        {
            "name": "gpt-4o",
            "context": 128000, "max_output": 16000,
            "input_price": 2.5, "output_price": 10.0,
            "reasoning": False
        },
        {
            "name": "deepseek-chat",
            "context": 128000, "max_output": 8000,
            "input_price": 0.14, "output_price": 0.28,
            "reasoning": False
        },
        {
            "name": "deepseek-reasoner",
            "context": 128000, "max_output": 8000,
            "input_price": 0.55, "output_price": 2.19,
            "reasoning": True
        },
    ]
    
    for model in sorted(candidates, key=lambda m: m["output_price"]):
        # 检查上下文窗口
        if input_tokens > model["context"]:
            continue
        
        # 检查输出限制
        if output_tokens > model["max_output"]:
            continue
        
        # 检查推理需求
        if needs_reasoning and not model["reasoning"]:
            continue
        
        # 检查预算
        cost = (input_tokens * model["input_price"] + 
                output_tokens * model["output_price"]) / 1_000_000
        if cost > budget:
            continue
        
        return model["name"]
    
    return "gpt-4o-mini"  # 默认回退

12. 最佳实践总结

12.1 Prompt设计原则

1. 精简原则：每个Token都有成本，删除不必要的修饰词和重复说明
2. 结构化原则：使用Markdown格式组织Prompt，便于模型理解和缓存
3. 模板化原则：将Prompt抽象为模板，避免每次从头编写
4. 版本化原则：对Prompt进行版本管理，追踪优化效果

12.2 上下文管理清单

• System Prompt控制在500 Token以内
• 对话历史实现滑动窗口或摘要压缩
• 长文档使用分块+RAG方案
• 启用Prompt Caching减少重复计算
• 设置Token使用预算和告警阈值
• 定期审查和优化高消耗Prompt

12.3 成本控制策略

12.4 常见陷阱

1. 忽略输出Token成本：输出单价通常是输入的4-5倍，max_tokens设置过大是最大的浪费源
2. System Prompt过长：每轮对话都重复发送，累积成本惊人
3. 不做缓存：相似请求反复计算，白白消耗Token
4. 盲目使用最大模型：简单任务用GPT-4o是浪费，GPT-4o-mini可能就够了
5. 缺乏监控：没有成本追踪，等到账单来了才发现超支

本教程涵盖了Token优化与上下文管理的核心技术栈。掌握这些知识后，你可以构建既高效又经济的LLM应用，在控制成本的同时最大化AI能力的价值。