Token 分词(下篇):工程化实践经验分享
前两篇讲原理。这篇全是工程经验分享包括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,000RAG 文档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(); ListMessage history trimHistory(conversation.getMessages(), budget.historyBudget); ListChunk 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 计算但估算和精确是两码事用错场景会出问题。精确计算用对应模型的 tokenizerfrom transformers import AutoTokenizer tokenizer AutoTokenizer.from_pretrained(Qwen/Qwen2.5-7B-Instruct) token_count len(tokenizer.encode(text, add_special_tokensFalse))注意不同模型必须用对应的 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:i1000] for i in range(0, len(doc), 1000)]问题在于不同文本类型的 token 密度差异极大文本类型1000 字符大约 token 数英文~250中文~500–700JSON可能超过 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_tokensFalse) # 当前段落加进去会超限先保存当前 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 sizeFAQ 知识库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每次调用 APIresponse 里都有finish_reason很多团队从来不检查它finish_reason含义stop正常结束模型生成了 EOS tokenlength达到max_output_tokens上限输出可能不完整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_limitoutput 那部分必须提前留出来。坑 3只算 content不算 Chat Templaterole 标记、消息分隔符、特殊 token 都吃 token用apply_chat_template之后再算。坑 4RAG 召回全塞进去召回 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 是大模型工程的输入边界、成本边界和能力边界。把这个搞透你在这条路上会少踩很多坑。