大语言模型代币消耗控制：从监控到优化的工程实践指南

在实际 AI 应用开发中，尤其是在调用大语言模型 API 时，开发者最常遇到的困扰之一就是代币消耗控制。很多项目在原型验证阶段运行良好，一旦进入高频测试或生产数据灌入，账单金额会迅速超出预期。更令人沮丧的是，有时为了优化某个环节的代币使用，反复调整提示词、测试不同参数，结果在“研究如何节省”的过程中，反而因为无效实验和监控缺失，消耗掉了所有预算。这背后暴露的不仅是成本意识问题，更是工程方法上的缺失。

本文将从一个真实项目场景出发，拆解代币消耗的主要环节，给出可落地的监控、优化和管控方案。无论是使用 OpenAI GPT 系列、国产大模型还是开源模型接口，只要涉及按 token 计费，本文提供的思路都能直接套用。我们将先理解 token 计算的基本规则，再构建一个本地化的用量监控体系，然后针对提示词、缓存、采样策略等关键点做具体优化，最后给出生产环境下的配置清单和排错指南。

1. 先弄清代币是怎么被计算和消耗的

代币（Token）是大语言模型中的基本计价单位，它并不完全等同于单词或汉字。例如，英文单词“apple”可能被算作 1 个 token，而“unfortunately”可能被拆成“un”、“for”、“tun”、“ate”、“ly”等多个 token。中文方面，一个汉字通常对应 1~2 个 token，但也要看具体分词算法。

1.1 不同模型的 token 计算方式差异

虽然概念相似，但 OpenAI、Claude、文心一言、通义千问等模型的 token 计算规则并不完全相同。以下是一个常见模型的对比：

如果项目中对成本敏感，必须在开发前期就用实际文本测试 token 数量。OpenAI 提供了官方的 token 计算工具：

import tiktoken def count_tokens(text, model_name="gpt-3.5-turbo"): encoding = tiktoken.encoding_for_model(model_name) return len(encoding.encode(text)) # 测试一段中文文本 text = "本文介绍如何有效控制大语言模型的代币消耗" token_count = count_tokens(text) print(f"Token 数量: {token_count}")

对于国内模型，通常需要查看其官方文档或 SDK 中是否提供类似的计数函数。

1.2 API 调用中哪些环节消耗代币

一次完整的 API 调用，代币消耗来自三个部分：

1. 输入提示词（Prompt）：你发给模型的所有内容，包括系统指令、用户问题、上下文示例等。
2. 生成内容（Completion）：模型返回的答案。
3. 隐藏成本：有些模型会在每次对话中保留一定的上下文缓存，这也可能计入 token 消耗。

更重要的是，如果你使用了函数调用（Function Calling）或 JSON 模式等高级功能，这些结构化描述本身也会消耗 token，而且数量不容忽视。

2. 建立代币用量监控体系，避免“无声”超支

最大的风险不是代币用完，而是在不知不觉中用完。很多开发者在调试阶段反复调用 API，却没有实时监控机制，等到收到账单或额度告警时才后悔莫及。

2.1 在代码层面植入用量统计

无论使用哪个模型的 SDK，都应该封装一个带有统计功能的客户端类：

class TokenAwareClient: def __init__(self, api_key, model_name): self.client = OpenAI(api_key=api_key) # 或其他模型客户端 self.model_name = model_name self.total_prompt_tokens = 0 self.total_completion_tokens = 0 self.total_calls = 0 def chat_completion(self, messages, **kwargs): response = self.client.chat.completions.create( model=self.model_name, messages=messages, **kwargs ) # 统计本次调用的 token 使用量 prompt_tokens = response.usage.prompt_tokens completion_tokens = response.usage.completion_tokens total_tokens = response.usage.total_tokens # 累计统计 self.total_prompt_tokens += prompt_tokens self.total_completion_tokens += completion_tokens self.total_calls += 1 # 打印实时消耗（生产环境应改为日志） print(f"本次消耗: {total_tokens} tokens (Prompt: {prompt_tokens}, Completion: {completion_tokens})") print(f"累计消耗: {self.total_prompt_tokens + self.total_completion_tokens} tokens") return response # 使用示例 client = TokenAwareClient("your-api-key", "gpt-3.5-turbo") response = client.chat_completion([ {"role": "user", "content": "请用一句话介绍 Python"} ])

2.2 设置用量阈值和告警机制

对于重要项目，应该设置硬性限制，防止单次运行消耗过多代币：

class BudgetAwareClient(TokenAwareClient): def __init__(self, api_key, model_name, daily_budget=100000): super().__init__(api_key, model_name) self.daily_budget = daily_budget self.daily_usage = 0 # 从持久化存储加载今日已用量（简化示例用内存） self.load_daily_usage() def chat_completion(self, messages, **kwargs): # 预测本次请求可能消耗的 token 数量（保守估计） estimated_tokens = self.estimate_token_usage(messages, kwargs) if self.daily_usage + estimated_tokens > self.daily_budget: raise Exception(f"今日预算不足: 已用 {self.daily_usage}, 预算 {self.daily_budget}") response = super().chat_completion(messages, **kwargs) # 更新每日用量并持久化 actual_tokens = response.usage.total_tokens self.daily_usage += actual_tokens self.save_daily_usage() return response def estimate_token_usage(self, messages, kwargs): # 简单的估算逻辑：统计所有文本长度乘以系数 total_text = " ".join([msg["content"] for msg in messages if msg.get("content")]) # 保守估计：按平均 1.5 个 token per 字符（中英文混合场景） estimated = int(len(total_text) * 1.5) # 如果指定了 max_tokens，加上这个值 if kwargs.get("max_tokens"): estimated += kwargs["max_tokens"] return estimated def load_daily_usage(self): # 实际项目中应该从数据库或文件读取 # 这里简化为从文件读取 try: with open("daily_usage.txt", "r") as f: self.daily_usage = int(f.read()) except FileNotFoundError: self.daily_usage = 0 def save_daily_usage(self): with open("daily_usage.txt", "w") as f: f.write(str(self.daily_usage))

2.3 区分环境：测试 vs 生产的不同监控策略

在不同环境下，监控的粒度应该有所不同：

在生产环境中，还应该建立用量趋势分析，及时发现异常模式。比如平时每天消耗 1 万 token 的应用，突然某天消耗 10 万 token，即使没超预算也需要排查原因。

3. 优化提示词工程，从源头控制输入 token

提示词是代币消耗的大头，尤其是需要带入大量上下文的场景。优化提示词不仅能节省成本，还能提高模型响应质量。

3.1 精简系统指令和上下文

很多开发者喜欢写冗长的系统指令，但实际上模型对指令的理解存在边际效应：

# 不推荐：过于冗长的系统指令 system_message = """ 你是一个专业的AI助手，擅长回答技术问题。请遵循以下规则： 1. 回答要准确专业 2. 如果不确定要说明 3. 格式要清晰易读 4. 要用中文回答 5. 不要编造不存在的信息 ...（还有10条规则） """ # 推荐：精简核心指令 system_message = "你是一个技术专家，用中文准确回答问题，不确定时明确说明。"

对于需要带入文档内容的场景，不要简单粗暴地全文嵌入：

# 不推荐：直接嵌入长文档 context = "这是一篇关于机器学习的长文档，共有10000字..." # 直接消耗大量token # 推荐：先提取关键信息再嵌入 def extract_relevant_sections(document, query): # 使用简单的关键词匹配或嵌入向量相似度查找相关段落 relevant_parts = find_most_relevant_parts(document, query) return "\n".join(relevant_parts[:3]) # 只返回最相关的3个段落 context = extract_relevant_sections(long_document, user_question) messages = [ {"role": "system", "content": "根据以下上下文回答问题"}, {"role": "user", "content": f"上下文：{context}\n问题：{user_question}"} ]

3.2 使用分层提示策略

对于复杂任务，采用分层策略可以减少单次调用的 token 消耗：

def process_complex_query(user_query): # 第一层：分析查询意图和所需信息 analysis_prompt = f""" 分析以下用户查询，确定： 1. 主要意图是什么（咨询、生成、总结、比较等） 2. 需要哪些关键信息 3. 答案的大致结构 查询：{user_query} """ analysis = client.chat_completion([{"role": "user", "content": analysis_prompt}]) # 第二层：根据分析结果收集必要信息 if "需要对比" in analysis.choices[0].message.content: # 只获取对比所需的关键数据，而不是全部信息 comparison_data = extract_comparison_data(user_query) final_prompt = f"基于以下数据进行比较分析：{comparison_data}" else: # 其他情况的处理逻辑 final_prompt = f"直接回答：{user_query}" # 第三层：生成最终答案 result = client.chat_completion([{"role": "user", "content": final_prompt}]) return result.choices[0].message.content

这种分层方法虽然可能增加调用次数，但每次调用的 token 消耗更可控，总体成本可能更低，而且质量更高。

3.3 利用模型的消息历史记忆能力

现代对话模型能记住上下文，不需要每次重复发送历史消息：

# 不推荐：每次发送完整历史 def bad_chat_approach(new_message, full_history): # full_history 会越来越长，token 消耗指数增长 messages = full_history + [{"role": "user", "content": new_message}] return client.chat_completion(messages) # 推荐：利用模型的上下文记忆 def good_chat_approach(new_message, previous_messages=None): if previous_messages is None: previous_messages = [] # 只保留最近几轮对话，避免历史过长 if len(previous_messages) > 6: # 保留最近3轮对话（一问一答为一轮） previous_messages = previous_messages[-6:] messages = previous_messages + [{"role": "user", "content": new_message}] response = client.chat_completion(messages) # 更新历史，包含本次问答 updated_history = messages + [{"role": "assistant", "content": response.choices[0].message.content}] return response, updated_history

4. 优化生成参数，控制输出 token 数量

模型生成的内容是代币消耗的另一大来源，通过合理设置生成参数，可以在保证质量的前提下有效控制输出长度。

4.1 设置合理的 max_tokens 限制

max_tokens参数直接控制生成内容的最大长度，应该根据实际需求设置：

# 根据任务类型设置不同的 max_tokens def get_appropriate_max_tokens(task_type): limits = { "简短回答": 100, "普通解释": 300, "详细分析": 800, "长文生成": 2000 } return limits.get(task_type, 300) # 默认300 # 在调用时动态设置 task_type = classify_user_query(user_query) max_tokens = get_appropriate_max_tokens(task_type) response = client.chat_completion( messages=messages, max_tokens=max_tokens )

但要注意，设置过小的max_tokens可能导致答案被截断。比较好的做法是设置一个合理上限，同时让模型知道需要简洁回答：

messages = [ {"role": "system", "content": "请用简洁的语言回答，重点突出，避免冗长。"}, {"role": "user", "content": user_query} ] response = client.chat_completion(messages, max_tokens=500)

4.2 使用停止序列（Stop Sequences）

对于格式化的输出，可以使用停止序列来避免模型生成多余内容：

# 生成 JSON 格式数据时，提前停止 response = client.chat_completion( messages=messages, stop=["\n}", "}\n"] # 看到这些序列时停止生成 )

4.3 调整温度（Temperature）和核采样（Top-p）

虽然这些参数不直接影响 token 数量，但通过影响生成质量间接影响成本：

对于成本敏感的应用，建议使用较低的温度值，并结合重试次数限制：

def generate_with_retry(messages, max_retries=2, temperature=0.3): for attempt in range(max_retries): try: response = client.chat_completion( messages=messages, temperature=temperature, max_tokens=500 ) if is_quality_acceptable(response): return response except Exception as e: if attempt == max_retries - 1: raise e return None # 所有重试都失败

5. 实施缓存策略，避免重复计算

很多用户问题具有重复性，为相同的问题重复调用 API 是典型的代币浪费。

5.1 基于问题内容的简单缓存

最基本的缓存是基于问题文本的精确匹配：

import hashlib import json class CachedLLMClient: def __init__(self, api_key, model_name, cache_file="llm_cache.json"): self.client = TokenAwareClient(api_key, model_name) self.cache_file = cache_file self.cache = self.load_cache() def get_cache_key(self, messages, parameters): # 基于消息内容和参数生成唯一键 content = json.dumps({"messages": messages, "params": parameters}, sort_keys=True) return hashlib.md5(content.encode()).hexdigest() def chat_completion(self, messages, **kwargs): cache_key = self.get_cache_key(messages, kwargs) if cache_key in self.cache: print("缓存命中！") return self.cache[cache_key] response = self.client.chat_completion(messages, **kwargs) self.cache[cache_key] = response self.save_cache() return response def load_cache(self): try: with open(self.cache_file, "r") as f: return json.load(f) except FileNotFoundError: return {} def save_cache(self): with open(self.cache_file, "w") as f: json.dump(self.cache, f)

5.2 基于语义相似度的智能缓存

精确匹配缓存只能处理完全相同的查询，更高级的做法是基于语义相似度：

from sentence_transformers import SentenceTransformer import numpy as np class SemanticCachedLLMClient(CachedLLMClient): def __init__(self, api_key, model_name, similarity_threshold=0.9): super().__init__(api_key, model_name) self.similarity_threshold = similarity_threshold self.encoder = SentenceTransformer('all-MiniLM-L6-v2') self.semantic_cache = self.load_semantic_cache() def get_semantic_key(self, messages): # 提取最后一个用户消息进行语义编码 last_user_message = None for msg in reversed(messages): if msg["role"] == "user": last_user_message = msg["content"] break if not last_user_message: return None embedding = self.encoder.encode([last_user_message])[0] return embedding def find_similar_cache(self, current_embedding): for cached_embedding, response in self.semantic_cache.items(): # 将字符串表示的嵌入向量转换回 numpy 数组 cached_embedding_array = np.fromstring(cached_embedding, sep=' ') similarity = np.dot(current_embedding, cached_embedding_array) / ( np.linalg.norm(current_embedding) * np.linalg.norm(cached_embedding_array) ) if similarity > self.similarity_threshold: return response return None def chat_completion(self, messages, **kwargs): current_embedding = self.get_semantic_key(messages) if current_embedding is not None: similar_response = self.find_similar_cache(current_embedding) if similar_response: print("语义缓存命中！") return similar_response response = super().chat_completion(messages, **kwargs) if current_embedding is not None: # 将 numpy 数组转换为可存储的字符串格式 embedding_str = ' '.join(map(str, current_embedding)) self.semantic_cache[embedding_str] = response self.save_semantic_cache() return response def load_semantic_cache(self): try: with open("semantic_cache.json", "r") as f: return json.load(f) except FileNotFoundError: return {} def save_semantic_cache(self): with open("semantic_cache.json", "w") as f: json.dump(self.semantic_cache, f)

6. 生产环境代币管理清单和排错指南

将前述策略整合成可执行的生产清单，并提供常见问题的排查路径。

6.1 代币优化检查清单

在应用上线前，逐项检查以下内容：

提示词优化检查项

• [ ] 系统指令是否简洁明确（不超过 100 字）
• [ ] 是否避免重复发送相同上下文
• [ ] 长文档是否先提取关键段落再嵌入
• [ ] 是否使用分层提示处理复杂任务

生成参数检查项

• [ ] 是否为不同任务类型设置合适的 max_tokens
• [ ] 温度参数是否根据任务确定性需求设置（创意任务 0.7-0.9，确定性任务 0.1-0.3）
• [ ] 是否设置停止序列来避免多余生成
• [ ] 是否实现重试机制并有次数限制

缓存策略检查项

• [ ] 是否实现基于内容的缓存
• [ ] 高频问题是否考虑语义缓存
• [ ] 缓存失效策略是否合理（定时清除或基于大小）

监控告警检查项

• [ ] 是否实时统计 token 使用量
• [ ] 是否设置每日预算阈值
• [ ] 是否有异常用量告警机制
• [ ] 是否区分测试和生产环境的监控策略

6.2 常见问题排查指南

当发现代币消耗异常时，按以下顺序排查：

问题现象：单次调用 token 消耗远高于预期

可能原因：

1. 提示词中包含大量不必要的上下文
2. 系统指令过于冗长
3. 嵌入的文档或数据过大

排查步骤：

# 1. 检查提示词长度 def analyze_prompt_length(messages): total_length = 0 for msg in messages: if msg.get("content"): total_length += len(msg["content"]) print(f"提示词总长度: {total_length} 字符") print(f"估计token数量: {total_length * 1.5}") # 粗略估算 # 2. 检查是否有重复内容 def find_duplicate_content(messages): contents = [msg["content"] for msg in messages if msg.get("content")] for i, content in enumerate(contents): if contents.count(content) > 1: print(f"发现重复内容: {content[:100]}...")

解决方案：

• 精简系统指令到核心要点
• 使用文档摘要而非全文
• 移除对话历史中的重复回合

问题现象：每日用量突然激增

可能原因：

1. 有循环调用或递归调用失控
2. 用户量突然增加但未相应调整预算
3. 某个功能被滥用或出现异常使用模式

排查步骤：

# 检查调用频率模式 def analyze_usage_pattern(usage_logs): import pandas as pd df = pd.DataFrame(usage_logs) hourly_usage = df.groupby(df['timestamp'].dt.hour)['tokens'].sum() # 寻找异常时间点 avg_usage = hourly_usage.mean() outliers = hourly_usage[hourly_usage > avg_usage * 3] # 3倍于平均值为异常 return outliers

解决方案：

• 实现速率限制（Rate Limiting）
• 添加用户级别的用量配额
• 对异常使用模式添加人工审核环节

问题现象：缓存命中率低

可能原因：

1. 用户问题多样性太高
2. 相似度阈值设置不合理
3. 缓存键生成策略有问题

排查步骤：

def analyze_cache_performance(cache_logs): total_requests = len(cache_logs) cache_hits = sum(1 for log in cache_logs if log['hit']) hit_rate = cache_hits / total_requests print(f"缓存命中率: {hit_rate:.2%}") # 分析未命中的请求类型 miss_requests = [log for log in cache_logs if not log['hit']] if miss_requests: print("常见未命中请求示例:") for i, req in enumerate(miss_requests[:3]): print(f"{i+1}. {req['query'][:100]}...")

解决方案：

• 调整语义相似度阈值
• 考虑基于查询意图分类的缓存策略
• 增加缓存容量和多样性

6.3 生产环境配置建议

对于正式上线的应用，建议采用以下配置策略：

环境变量管理

import os class ProductionLLMConfig: def __init__(self): self.api_key = os.getenv('LLM_API_KEY') self.daily_budget = int(os.getenv('LLM_DAILY_BUDGET', '100000')) self.max_tokens_default = int(os.getenv('MAX_TOKENS_DEFAULT', '500')) self.enable_cache = os.getenv('ENABLE_CACHE', 'true').lower() == 'true' self.cache_ttl = int(os.getenv('CACHE_TTL_HOURS', '24')) # 缓存24小时 def get_client(self): client = BudgetAwareClient(self.api_key, "gpt-3.5-turbo", self.daily_budget) if self.enable_cache: client = CachedLLMClient(self.api_key, "gpt-3.5-turbo") return client

多级降级策略当预算接近耗尽时，不应该直接报错，而应该优雅降级：

def get_response_with_fallback(user_query, primary_client, fallback_strategies): try: return primary_client.chat_completion([{"role": "user", "content": user_query}]) except BudgetExceededError: for strategy in fallback_strategies: try: return strategy(user_query) except Exception: continue return {"error": "服务暂时不可用，请稍后重试"} # 降级策略示例 def cached_response_strategy(query): # 只从缓存中查找，即使相似度阈值降低 return semantic_cache.find_similar(query, threshold=0.7) def rule_based_fallback(query): # 基于规则的简单回复 if "价格" in query: return "请咨询客服获取最新价格信息" elif "时间" in query: return "我们的工作时间是工作日9:00-18:00" else: return "暂时无法回答此问题，请稍后重试"

代币管理本质上是一个工程优化问题，需要在整个开发周期中持续关注。从项目开始就建立监控意识，在编码阶段实施优化策略，在测试阶段验证效果，在生产环境设置防护机制，这样才能真正避免"在研究如何节省的过程中花光所有代币"的尴尬局面。最关键的是培养成本意识，把 token 消耗当作与服务器资源、数据库查询同等重要的技术指标来对待。