LLM API请求全流程解析：从令牌化到流式传输的工程实践

1. 先搞清楚一次 LLM 请求到底包含哪些环节当你调用一个大语言模型LLM时无论是通过 OpenAI、Claude、DeepSeek 的 API还是本地部署的开源模型背后都是一套完整的请求-响应循环。这个循环远不止“发个问题等个答案”那么简单。从你的代码发出请求到最终拿到可用的结果中间至少经过六个关键环节请求构造把你的自然语言问题转换成模型能理解的格式上下文管理处理历史对话、系统提示词和当前问题的拼接令牌化将文本拆分成模型认识的数字序列推理生成模型基于输入逐词预测输出流式传输实时返回生成结果而不是等全部完成后处理对原始输出进行格式化、截断或安全过滤我见过很多开发者一上来就纠结“为什么响应慢”或“为什么输出不完整”其实问题往往出在前三个环节。比如上下文超长导致截断、令牌化后实际输入远超预期、或者请求格式不符合 API 要求。2. 请求构造别让格式问题拖慢整个流程2.1 基础请求结构大多数 LLM API 都遵循类似的 RESTful 设计。以 OpenAI 风格的接口为例一个完整的请求需要包含这些核心字段import requests import json payload { model: gpt-3.5-turbo, # 指定模型版本 messages: [ {role: system, content: 你是一个有帮助的助手}, {role: user, content: 请解释量子计算的基本原理} ], max_tokens: 500, # 控制输出长度 temperature: 0.7, # 控制随机性 stream: True # 是否启用流式输出 } headers { Content-Type: application/json, Authorization: Bearer your-api-key } response requests.post( https://api.openai.com/v1/chat/completions, headersheaders, datajson.dumps(payload) )这里最容易出问题的是messages字段的格式。我见过有人直接把字符串当消息体发送或者混淆了role的取值。正确的角色应该是system、user、assistant三者之一分别对应系统提示、用户输入和模型之前的回复。2.2 参数选择的实际影响max_tokens不是设得越大越好。这个参数直接影响响应时间和 API 成本。如果你的场景只需要简短回答设为 100-200 就足够如果需要长文生成也要考虑模型的实际能力上限。temperature参数控制输出的创造性0.0 表示完全确定性输出每次相同输入得到相同结果1.0 表示最大随机性。对于代码生成或事实问答我通常用 0.1-0.3对于创意写作可以提到 0.7-0.9。实测建议第一次调用时先把stream设为False确认基础流程能走通后再开启流式传输。流式能提升用户体验但会增加连接管理的复杂度。3. 上下文管理决定模型理解深度的关键3.1 令牌计数与长度限制每个 LLM 都有上下文窗口限制比如 GPT-4 通常是 128K 令牌Claude 3 能达到 200K。但“能支持”不等于“能用好”。上下文越长推理速度越慢成本也越高。你需要时刻关注实际使用的令牌数。OpenAI 提供了tiktoken库来精确计算import tiktoken def count_tokens(text, modelgpt-4): encoding tiktoken.encoding_for_model(model) return len(encoding.encode(text)) messages [ {role: system, content: 你是一个专业的技术文档写手}, {role: user, content: 请为Redis集群部署写一份操作指南} ] total_tokens sum(count_tokens(msg[content]) for msg in messages) print(f当前对话使用令牌数: {total_tokens})当令牌数接近模型上限时API 会返回类似maximum context length is 4096 tokens的错误。这时你需要精简输入内容或启用自动截断策略。3.2 对话历史的管理策略多轮对话中历史消息的保留方式直接影响模型的表现。常见的策略有全量保留保留所有历史记录适合需要长期记忆的场景滑动窗口只保留最近 N 轮对话控制上下文长度关键摘要对早期对话生成摘要用摘要替代原始内容我个人的经验是对于技术问答类应用滑动窗口保留最近5-10轮通常足够对于需要长期上下文的创作任务可以结合摘要和全量保留。4. 令牌化文本到数字的转换过程4.1 为什么令牌化影响实际效果令牌化不是简单的按词切割。模型使用的令牌化器Tokenizer会把文本拆分成子词单元比如 unfortunately 可能被拆成 [un, fort, un, ate, ly]。不同模型的令牌化方式不同这导致相同文本在不同模型中的令牌数可能差异很大某些专业术语可能被拆分成无意义的片段中英文混合文本需要特别处理如果你发现模型对某些专业词汇理解有偏差很可能是令牌化出了问题。这时可以在提示词中明确给出术语的定义或使用同义词替换。4.2 令牌化实战检查在发送请求前先用对应模型的令牌化器检查一下# 检查GPT系列的令牌化 import tiktoken text 深度学习模型在自然语言处理中的应用 encoding tiktoken.get_encoding(cl100k_base) # GPT-4使用的编码 tokens encoding.encode(text) print(f文本: {text}) print(f令牌数: {len(tokens)}) print(f令牌列表: {tokens}) print(f反向解码: {encoding.decode(tokens)})这个检查能帮你发现潜在的令牌化问题比如特殊符号被错误处理、空格计数异常等。5. 推理生成模型如何产生文本5.1 自回归生成过程LLM 的文本生成是典型的自回归过程根据已有文本预测下一个词不断重复直到满足停止条件。在 API 层面这个过程对应着这些参数max_tokens生成的最大令牌数达到即停止stop_sequences遇到特定字符串时停止生成top_p核采样限制候选词的概率累积和控制多样性frequency_penalty降低重复词汇的出现概率关键理解max_tokens限制的是本次生成的新令牌数不是总上下文长度。如果你设置了max_tokens100但输入已经用了 3900 个令牌在 4096 限制的模型上请求会因超限而失败。5.2 流式传输的实际优势启用流式传输后你不需要等待整个响应完成就能开始处理结果import requests import json def stream_chat_completion(api_key, messages): url https://api.openai.com/v1/chat/completions headers { Content-Type: application/json, Authorization: fBearer {api_key} } data { model: gpt-3.5-turbo, messages: messages, stream: True, max_tokens: 500 } response requests.post(url, headersheaders, jsondata, streamTrue) for line in response.iter_lines(): if line: line line.decode(utf-8) if line.startswith(data: ): json_str line[6:] if json_str ! [DONE]: chunk json.loads(json_str) if choices in chunk and chunk[choices]: delta chunk[choices][0].get(delta, {}) if content in delta: yield delta[content] # 使用示例 messages [{role: user, content: 请介绍Python的装饰器}] for chunk in stream_chat_completion(your-api-key, messages): print(chunk, end, flushTrue)流式传输特别适合需要实时显示生成结果的场景比如聊天应用或代码补全工具。6. 错误处理与重试机制6.1 常见错误类型及应对LLM API 调用中常见的错误包括429 Too Many Requests速率限制需要实现指数退避重试500 Internal Server Error服务端问题短暂等待后重试400 Bad Request请求格式错误需要检查参数合法性401 UnauthorizedAPI 密钥问题检查密钥有效性一个健壮的重试机制应该这样设计import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_session_with_retries(): session requests.Session() retry_strategy Retry( total3, # 最大重试次数 status_forcelist[429, 500, 502, 503, 504], # 需要重试的状态码 method_whitelist[POST], # 只对POST请求重试 backoff_factor1 # 重试间隔1, 2, 4秒 ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) return session # 使用带重试的session session create_session_with_retries() response session.post(api_url, headersheaders, jsonpayload)6.2 超时设置与连接管理网络不稳定的环境需要设置合理的超时try: response requests.post( api_url, headersheaders, jsonpayload, timeout(3.05, 30) # 连接超时3.05秒读取超时30秒 ) except requests.exceptions.Timeout: print(请求超时可能是网络问题或服务器响应慢) except requests.exceptions.ConnectionError: print(连接错误检查网络连接和API端点)对于生产环境我建议把超时时间设得比平均响应时间稍长但要设置上限防止无限等待。7. 成本控制与性能优化7.1 令牌使用监控LLM API 的成本直接与输入输出令牌数相关。你需要监控每次调用的实际消耗def calculate_cost(response, model_pricing): 计算单次请求的成本 input_tokens response[usage][prompt_tokens] output_tokens response[usage][completion_tokens] total_tokens response[usage][total_tokens] input_cost (input_tokens / 1000) * model_pricing[input] output_cost (output_tokens / 1000) * model_pricing[output] return { input_tokens: input_tokens, output_tokens: output_tokens, total_tokens: total_tokens, input_cost: input_cost, output_cost: output_cost, total_cost: input_cost output_cost } # GPT-4 Turbo定价示例每千令牌 gpt4_pricing {input: 0.01, output: 0.03} cost_info calculate_cost(api_response, gpt4_pricing)定期分析令牌使用模式能帮你发现优化机会比如过长的系统提示词、不必要的上下文保留等。7.2 批量请求处理如果需要处理大量相似请求考虑使用批量接口如果API支持或合理的并发控制import asyncio import aiohttp async def make_async_request(session, url, headers, payload): async with session.post(url, headersheaders, jsonpayload) as response: return await response.json() async def batch_requests(api_requests): async with aiohttp.ClientSession() as session: tasks [] for request in api_requests: task make_async_request(session, request[url], request[headers], request[payload]) tasks.append(task) results await asyncio.gather(*tasks, return_exceptionsTrue) return results # 使用示例 requests_list [ { url: https://api.openai.com/v1/chat/completions, headers: headers, payload: payload1 }, { url: https://api.openai.com/v1/chat/completions, headers: headers, payload: payload2 } ] results asyncio.run(batch_requests(requests_list))重要提醒并发请求要遵守API的速率限制否则会收到429错误。先了解服务的具体限制再设计合适的并发策略。8. 生产环境最佳实践8.1 日志与监控在生产环境中你需要记录完整的请求-响应循环信息请求时间戳和唯一ID使用的模型和参数输入输出令牌数响应时间和状态码错误信息如果有这能帮你分析性能瓶颈、成本趋势和错误模式。8.2 缓存策略对于重复性查询可以考虑实现缓存层import redis import hashlib import json class LLMCache: def __init__(self, redis_client, ttl3600): # 默认缓存1小时 self.redis redis_client self.ttl ttl def _get_cache_key(self, model, messages, parameters): 生成基于请求内容的缓存键 content f{model}{json.dumps(messages, sort_keysTrue)}{json.dumps(parameters, sort_keysTrue)} return hashlib.md5(content.encode()).hexdigest() def get(self, model, messages, parameters): key self._get_cache_key(model, messages, parameters) cached self.redis.get(key) return json.loads(cached) if cached else None def set(self, model, messages, parameters, response): key self._get_cache_key(model, messages, parameters) self.redis.setex(key, self.ttl, json.dumps(response)) # 使用示例 cache LLMCache(redis_client) cached_response cache.get(model, messages, parameters) if not cached_response: response make_llm_request(model, messages, parameters) cache.set(model, messages, parameters, response)缓存能显著降低成本和延迟但要注意不适合实时性要求极高的场景。8.3 降级方案当主要API不可用时应该有备选方案备用模型GPT-4不可用时降级到GPT-3.5本地模型云端服务中断时使用本地部署的轻量模型规则引擎对于简单查询使用基于规则的回复我建议把这些经验落实到你的LLM应用开发中先确保单次请求稳定可靠再考虑批量处理和性能优化。很多时候问题不是出在模型能力上而是请求构造、错误处理或资源管理不到位。