前两篇讲原理。这篇全是工程经验分享,包括Token Budget Manager 怎么设计、RAG 分块的正确姿势、日志和 JSON 怎么处理、多租户限流怎么做、token 安全风控要注意什么。
把 Token 当成资源管理
很多团队在 AI 项目出了问题之后才意识到:token 应该像 CPU、内存、数据库连接一样被管理,而不是随便传进去、调完再看 usage 是多少。
一个成熟的企业级 AI 系统,token 的生命周期应该是这样的:
在请求真正发出去之前,你应该知道这次请求大概要花多少 token,超了该怎么裁,裁完之后再发。不是发出去之后再看报错。
Token Budget Manager 设计
核心思路是给每个 context 部分分配固定预算,总和不超过模型的上下文窗口。
以 32k context 的模型为例:
部分 | token 预算 |
|---|---|
系统提示词 | 2,000 |
用户当前问题 | 1,000 |
历史对话 | 5,000 |
RAG 文档 | 18,000 |
工具定义 + 工具结果 | 3,000 |
预留输出 | 2,000 |
安全冗余 | 1,000 |
| 合计 | 32,000 |
这个分配不是固定的,应该根据业务场景调整。RAG 场景文档多就多给 RAG,客服场景历史对话重要就多给历史。
裁剪的优先级:
必须保留:系统提示词核心规则、用户当前问题 尽量保留:最近几轮对话、高相关 RAG 文档 可以裁剪:早期历史对话、低相关文档、冗余 JSON 字段、重复日志下面是一个完整的请求构建流程(Java 伪代码):
class AiRequestBuilder { AiRequest build(UserMessage userMessage, Conversation conversation) { // 1. 定义预算 TokenBudget budget = TokenBudget.builder() .contextLimit(32_000) .reservedOutput(2_000) .systemBudget(2_000) .historyBudget(5_000) .ragBudget(18_000) .toolBudget(3_000) .safetyMargin(1_000) .build(); // 2. 构建各部分 String systemPrompt = buildSystemPrompt(); List<Message> history = trimHistory(conversation.getMessages(), budget.historyBudget); List<Chunk> ragChunks = selectChunksByBudget(ragRetriever.retrieve(userMessage), budget.ragBudget); String toolContext = compressToolResult(buildToolContext(userMessage), budget.toolBudget); Prompt prompt = Prompt.of(systemPrompt, history, userMessage, ragChunks, toolContext); // 3. 计算实际 token 数,超限就裁剪 int maxInput = budget.contextLimit - budget.reservedOutput - budget.safetyMargin; while (tokenCounter.count(prompt) > maxInput) { if (canRemoveLowScoreChunk(ragChunks)) { removeLowestScoreChunk(ragChunks); } elseif (canCompressHistory(history)) { history = historyCompressor.compress(history); } elseif (canCompressToolContext(toolContext)) { toolContext = toolResultCompressor.compressMore(toolContext); } else { hardTruncateLeastImportantPart(prompt); break; } prompt = Prompt.of(systemPrompt, history, userMessage, ragChunks, toolContext); } return AiRequest.builder() .prompt(prompt) .maxOutputTokens(budget.reservedOutput) .estimatedInputTokens(tokenCounter.count(prompt)) .build(); } }调用完之后,usage 必须落库:
class UsageRecorder { void record(AiResponse response, AiRequest request) { Usage u = response.getUsage(); usageLogRepo.save(UsageLog.builder() .requestId(request.getRequestId()) .userId(request.getUserId()) .tenantId(request.getTenantId()) .model(request.getModel()) .scene(request.getScene()) .inputTokens(u.getInputTokens()) .outputTokens(u.getOutputTokens()) .totalTokens(u.getTotalTokens()) .latencyMs(response.getLatencyMs()) .promptVersion(request.getPromptVersion()) .createdAt(LocalDateTime.now()) .build()); } }落库书库是为了后续成本分析、模型选型、Prompt 优化的数据来源。没有这些数据,你永远不知道钱花在哪里。
Token 计算:估算 vs 精确
实际项目里经常要做 token 计算,但估算和精确是两码事,用错场景会出问题。
精确计算,用对应模型的 tokenizer:
from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct") token_count = len(tokenizer.encode(text, add_special_tokens=False))注意:不同模型必须用对应的 tokenizer。用 GPT 的 tiktoken 去估算 Qwen 的消耗,误差可以到 20–40%。
粗估,只适合低精度场景:
# 经验粗估,不精确 def rough_estimate(text: str) -> int: 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)粗估适合的场景:前端 UI 显示、提前限流(防止明显超长输入)、快速排查是否需要精确计算。不适合精确计费、接近上下文上限时的裁剪判断。JSON、代码、日志、Base64 的实际 token 数和粗估可能差很多。
另一个原则:生产环境里,以 API 返回的usage.input_tokens作为最终账单依据,不要相信自己本地算的数字。
RAG 分块
RAG 系统里 token 影响两个关键点:文档分块和上下文拼接。两个都做错的团队不少。
最常见的错误是按字符数机械切分:
# 错误做法 chunks = [doc[i:i+1000] for i in range(0, len(doc), 1000)]问题在于不同文本类型的 token 密度差异极大:
文本类型 | 1000 字符大约 token 数 |
|---|---|
英文 | ~250 |
中文 | ~500–700 |
JSON | 可能超过 1000 |
日志 | 可能更高 |
同样 1000 字符,中文 chunk 的 token 数可能是英文的两三倍,导致后面拼 prompt 时预算超出。
正确原则:先按语义结构切,再用 token 控制长度。
def chunk_document(text: str, tokenizer, chunk_size: int = 800, overlap: int = 100) -> list[dict]: """ 先按段落/章节切分,超长的再按 token 拆分,保留 overlap 防语义截断。 """ # Step 1: 按段落切 paragraphs = split_by_paragraph(text) chunks = [] current_ids = [] for para in paragraphs: para_ids = tokenizer.encode(para, add_special_tokens=False) # 当前段落加进去会超限,先保存当前 chunk if current_ids and len(current_ids) + len(para_ids) > chunk_size: chunks.append({ "text": tokenizer.decode(current_ids), "token_count": len(current_ids) }) # overlap:把末尾 overlap 个 token 带到下一个 chunk current_ids = current_ids[-overlap:] # 单个段落本身就超长,递归拆分 if len(para_ids) > chunk_size: for i in range(0, len(para_ids), chunk_size - overlap): sub_ids = para_ids[i:i + chunk_size] chunks.append({ "text": tokenizer.decode(sub_ids), "token_count": len(sub_ids) }) current_ids = para_ids[-(overlap):] else: current_ids.extend(para_ids) if current_ids: chunks.append({ "text": tokenizer.decode(current_ids), "token_count": len(current_ids) }) return chunksChunk 的元数据设计也很重要,检索之后 rerank 和溯源都要用:
class Chunk { String docId; String chunkId; String text; int tokenCount; // 每个 chunk 存 token 数,拼 prompt 时直接用 String titlePath; // 标题路径,如 "第三章 > 3.2 退款规则" Integer pageNumber; String sourceUrl; LocalDateTime updatedAt; float rerankScore; // 检索后赋值 }常见的 chunk size 参考值:
场景 | 建议 chunk size |
|---|---|
FAQ 知识库 | 300–800 tokens |
技术文档 | 500–1200 tokens |
合同/政策文件 | 800–1500 tokens |
代码文件 | 按函数/类切,再控制 token 上限 |
日志分析 | 先聚合摘要,再控制 token |
几个典型场景的 token 管理
客服机器人
用户问:"我的订单为什么还没发货?"
系统需要整合的上下文来源很多:系统提示词、用户信息、订单详情、物流信息、售后政策、历史对话。
错误做法是把所有订单塞进去、完整 JSON 塞进去、最近 100 轮对话全带上。这样成本高,响应慢,而且多数内容和当前问题无关,会分散模型注意力。
正确做法是只给完成当前任务所需的最小充分上下文:
{ "orderId": "A1001", "status": "PAID_NOT_SHIPPED", "paidAt": "2026-05-01 10:30:00", "expectedShipBefore": "2026-05-03 23:59:59", "policy": "付款后 48 小时内发货,预售商品除外" }不需要的字段(内部日志、交易流水 ID、仓库内部编码)通通不传。
日志分析
日志是最大的 token 黑洞。5000 行日志直接塞给模型,超限是必然的,而且有效信息也会淹没在噪声里。
正确做法是后端先做:
def preprocess_logs(raw_logs: list[str]) -> str: """先用程序过滤,再给模型,而不是整坨丢过去""" # 1. 按 traceId 聚合 # 2. 只保留 ERROR / WARN # 3. 去重堆栈(同一异常只保留首次) # 4. 统计出现次数 # 5. 截断超长字段 # 6. 去掉心跳日志、定时任务日志 summary = extract_error_summary(raw_logs) return f""" 异常类型:{summary.exception_type} 出现次数:{summary.count} 首次时间:{summary.first_seen} 典型堆栈(前 20 行): {summary.stack_trace[:20_lines]} 影响接口:{summary.affected_endpoints} """给模型的是摘要,不是原始日志。
代码助手
代码文件的 token 密度高,而且大部分内容和用户当前的问题无关。
用户报错:NullPointerException in OrderService.getOrderStatus() 需要传给模型的: 完整的错误堆栈 OrderService.getOrderStatus() 方法体 相关的 Order 实体类(只传字段定义,不传方法) 数据库表结构(只传相关字段) 不需要传的: 整个 Controller 文件 无关的 Service 方法 配置文件(除非报错和配置有关) 测试代码多租户限流和成本治理
企业系统里,不同用户、租户、业务线对 token 的消耗差异可能很大。需要从多个维度管理:
维度 | 用途 |
|---|---|
user_id | 用户级别的每日/每月限额 |
tenant_id | 租户级成本控制和账单分摊 |
scene | 分析哪个业务场景最贵 |
model | 对比不同模型的成本效益 |
prompt_version | 判断 Prompt 优化是否降低了成本 |
配额设计示例:
class TokenQuota { // 用户级 long userDailyLimit = 100_000; // 免费用户每天 10 万 tokens long userMonthlyLimit = 2_000_000; // 租户级 long tenantDailyLimit = 10_000_000; long tenantMonthlyLimit = 200_000_000; // 单请求保护 int maxInputTokensPerRequest = 20_000; int maxOutputTokensPerRequest = 4_000; }限流检查应该在请求进入 AI Gateway 时就做,不要等到模型调用时:
if (tokenEstimate > userPlan.getMaxInputTokens()) { throw new BizException("输入内容过长,请缩短后重试"); } if (usageCache.getUserDailyUsage(userId) + tokenEstimate > quota.userDailyLimit) { throw new BizException("今日使用额度已用完"); }Token 安全风控
这个常被忽视,但确实有实际风险。
攻击者可能构造超长输入来消耗你的 token 配额:大量重复文本、超长 Base64、巨大 JSON、Prompt Injection 攻击。后果是成本攻击、上下文污染,甚至让模型产生异常输出。
后端要做的基本防护:
public void validateInput(String userInput) { // 字符数硬限制(粗筛,快) if (userInput.length() > 50_000) { thrownew BizException("输入内容过长"); } // token 数精确检查(接近上限时) int estimatedTokens = roughEstimate(userInput); if (estimatedTokens > 15_000) { int exactTokens = tokenCounter.count(userInput); if (exactTokens > userPlan.getMaxInputTokens()) { thrownew BizException("输入内容超过 token 限制"); } } // Prompt Injection 基础检测 if (containsInjectionPattern(userInput)) { log.warn("Possible prompt injection: userId={}", userId); // 隔离处理或拒绝 } // Base64 / HTML 异常长度检测 if (looksLikeBase64(userInput) && userInput.length() > 10_000) { thrownew BizException("不支持此类型内容"); } }重视finish_reason
每次调用 API,response 里都有finish_reason,很多团队从来不检查它:
finish_reason | 含义 |
|---|---|
stop | 正常结束,模型生成了 EOS token |
length | 达到 |
content_filter | 内容安全策略触发 |
tool_calls | 模型选择调用工具 |
如果finish_reason == length,说明模型的回答被截断了,你应该:
检查
max_output_tokens是否设置太小检查是否 input token 太多,压缩了 output 空间
对于需要完整输出的场景(比如代码生成),考虑做续写(把截断内容加回历史继续请求)
这个字段应该在 usage 日志里记录,用于监控和排查。
常见踩坑总结
坑 1:把 context 窗口当字符数32k context = 32k tokens,不是 32k 汉字。中文实际可容纳的汉字数大约是 context 限制的一半不到。
坑 2:没有预留输出 token输入 token 占满了上下文,模型没地方生成,要么报错要么输出被截断。input + output <= context_limit,output 那部分必须提前留出来。
坑 3:只算 content,不算 Chat Templaterole 标记、消息分隔符、特殊 token 都吃 token,用apply_chat_template之后再算。
坑 4:RAG 召回全塞进去召回 Top 30 全堆进 prompt,不仅贵,还会触发 lost-in-the-middle——模型对 prompt 中间位置的内容注意力偏弱,关键信息放在两端效果更好。召回后用 reranker 筛到 Top 5,控制总 token 在合理范围。
坑 5:按字符机械切 RAG 文档已经说过,先按语义结构切,再用 token 控制长度。
坑 6:工具定义过多按场景动态选工具,不要一股脑全传。
坑 7:不同模型共用一个 tokenizer 估算GPT 的 tiktoken 和 Qwen 的 tokenizer 差异可能 20–40%,别用错了。
坑 8:完整日志/JSON 直接塞给模型先过滤、聚合、字段裁剪,把摘要给模型,而不是原始数据。
坑 9:以为加特殊 token 只需要改词表新增 token 后需要扩展 Embedding 矩阵并训练,不是改个 JSON 文件那么简单。
回顾一下
读完这三篇文章,如果你真正掌握了,应该能做到:
解释 token、字符、字节、单词、子词的区别
解释 token ID 和 Embedding 矩阵的关系
解释为什么 Tokenizer 和模型参数绑定,不能随便换
能推导 BPE 的训练和推理过程
解释 Chat Template 如何影响 token 数
独立设计一个 Token Budget Manager
为 RAG 系统设计基于语义结构和 token 预算的分块策略
根据 usage 日志分析成本异常
排查上下文超限问题
排查输出被截断问题(
finish_reason == length)设计多租户 token 限流和配额体系
判断是否需要从零训练 Tokenizer(大多数业务不需要)
总结
理解 token 不是为了知道"文本怎么被切开",而是为了设计出真正可靠的大模型应用:不超上下文、不浪费成本、不乱塞文档、不让历史对话无限堆积、不暴露过多工具、能监控 usage、能做限流和预算、能排查复杂的生成问题。
Token 是大模型工程的输入边界、成本边界和能力边界。把这个搞透,你在这条路上会少踩很多坑。