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", headers=headers, data=json.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, model="gpt-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_tokens=100,但输入已经用了 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": f"Bearer {api_key}" } data = { "model": "gpt-3.5-turbo", "messages": messages, "stream": True, "max_tokens": 500 } response = requests.post(url, headers=headers, json=data, stream=True) 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='', flush=True)流式传输特别适合需要实时显示生成结果的场景,比如聊天应用或代码补全工具。
6. 错误处理与重试机制
6.1 常见错误类型及应对
LLM API 调用中常见的错误包括:
- 429 Too Many Requests:速率限制,需要实现指数退避重试
- 500 Internal Server Error:服务端问题,短暂等待后重试
- 400 Bad Request:请求格式错误,需要检查参数合法性
- 401 Unauthorized:API 密钥问题,检查密钥有效性
一个健壮的重试机制应该这样设计:
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( total=3, # 最大重试次数 status_forcelist=[429, 500, 502, 503, 504], # 需要重试的状态码 method_whitelist=["POST"], # 只对POST请求重试 backoff_factor=1 # 重试间隔:1, 2, 4秒 ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) return session # 使用带重试的session session = create_session_with_retries() response = session.post(api_url, headers=headers, json=payload)6.2 超时设置与连接管理
网络不稳定的环境需要设置合理的超时:
try: response = requests.post( api_url, headers=headers, json=payload, 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, headers=headers, json=payload) 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_exceptions=True) 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, ttl=3600): # 默认缓存1小时 self.redis = redis_client self.ttl = ttl def _get_cache_key(self, model, messages, parameters): """生成基于请求内容的缓存键""" content = f"{model}{json.dumps(messages, sort_keys=True)}{json.dumps(parameters, sort_keys=True)}" 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应用开发中:先确保单次请求稳定可靠,再考虑批量处理和性能优化。很多时候问题不是出在模型能力上,而是请求构造、错误处理或资源管理不到位。