大语言模型请求响应周期全解析：从API调用到错误处理

在大语言模型（LLM）应用开发过程中，很多开发者虽然能够调用API完成基本功能，但对请求响应周期的完整流程缺乏系统理解。当遇到"token exchange failed"、"request timed out"、"response exceeded token maximum"等错误时，往往只能盲目尝试解决方案。本文将完整拆解LLM请求响应周期的每个环节，从输入处理到输出生成，帮助开发者建立完整的认知框架。

1. LLM请求响应周期概述

1.1 什么是LLM请求响应周期

LLM请求响应周期是指从用户向大语言模型提交问题开始，到模型返回完整答案结束的完整过程。这个周期包含了多个关键阶段：输入预处理、令牌化、模型推理、文本生成和后期处理。理解这个周期对于优化应用性能、处理错误和设计可靠系统至关重要。

在实际开发中，一个完整的请求响应周期通常涉及以下参与方：

• 客户端应用（前端界面、移动应用、后端服务）
• API网关或代理层
• 模型推理服务
• 底层计算资源（GPU/CPU集群）

1.2 为什么需要深入理解请求响应周期

深入理解请求响应周期可以帮助开发者：

• 准确诊断和解决各类API错误
• 优化请求参数提升响应速度
• 合理设置超时时间和重试机制
• 有效管理token使用成本
• 设计更稳定的LLM集成架构

根据网络热词中出现的常见错误，如"token exchange failed"、"request timed out"等，都是由于对请求响应周期理解不足导致的典型问题。

2. 请求准备阶段：输入预处理

2.1 提示词构建与格式化

在发送请求之前，首先需要构建合适的提示词。提示词的质量直接影响模型输出的准确性和相关性。一个良好的提示词应该包含清晰的指令、必要的上下文和期望的输出格式。

# 示例：构建对话式提示词 def build_chat_prompt(messages): """ 构建OpenAI Chat API兼容的提示词格式 """ prompt = [] for message in messages: if message["role"] == "system": prompt.append({"role": "system", "content": message["content"]}) elif message["role"] == "user": prompt.append({"role": "user", "content": message["content"]}) elif message["role"] == "assistant": prompt.append({"role": "assistant", "content": message["content"]}) return prompt # 使用示例 messages = [ {"role": "system", "content": "你是一个有帮助的AI助手。"}, {"role": "user", "content": "请解释机器学习的基本概念。"} ] formatted_prompt = build_chat_prompt(messages)

2.2 参数配置与优化

LLM请求通常包含多个重要参数，这些参数直接影响请求的处理和响应生成：

# 完整的请求参数配置 request_params = { "model": "gpt-3.5-turbo", # 指定模型版本 "messages": formatted_prompt, # 提示词内容 "max_tokens": 1000, # 最大输出token数 "temperature": 0.7, # 创造性程度（0-2） "top_p": 0.9, # 核采样参数 "frequency_penalty": 0.5, # 频率惩罚 "presence_penalty": 0.3, # 存在惩罚 "stream": False, # 是否流式输出 "timeout": 30 # 请求超时时间（秒） }

每个参数的作用需要仔细理解：

• max_tokens：限制输出长度，防止生成过长内容
• temperature：控制输出的随机性，值越低输出越确定
• top_p：与temperature配合使用，控制词汇选择的多样性

2.3 输入验证与清理

在发送请求前，必须对输入进行验证和清理，避免无效请求或安全风险：

def validate_llm_request(params): """验证LLM请求参数的合法性""" errors = [] # 检查必填参数 if not params.get("model"): errors.append("模型名称不能为空") if not params.get("messages"): errors.append("提示词内容不能为空") # 检查参数范围 if params.get("max_tokens", 0) > 4000: errors.append("max_tokens不能超过4000") if not 0 <= params.get("temperature", 1) <= 2: errors.append("temperature必须在0-2之间") # 检查内容长度 total_content = "".join([msg["content"] for msg in params["messages"]]) if len(total_content) > 10000: errors.append("输入内容过长") return len(errors) == 0, errors # 使用验证函数 is_valid, error_messages = validate_llm_request(request_params) if not is_valid: print(f"参数验证失败: {error_messages}") return

3. 请求发送与网络传输

3.1 API请求构造

构造HTTP请求是LLM调用的核心环节。不同的模型提供商可能有不同的API设计，但基本模式相似：

import requests import json import time class LLMClient: def __init__(self, api_key, base_url="https://api.openai.com/v1"): self.api_key = api_key self.base_url = base_url self.session = requests.Session() self.session.headers.update({ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }) def send_request(self, endpoint, data, timeout=30): """发送LLM API请求""" url = f"{self.base_url}/{endpoint}" try: start_time = time.time() response = self.session.post( url, json=data, timeout=timeout, # 重要：设置合理的重试策略 hooks={'response': self._log_response_time} ) end_time = time.time() # 记录请求耗时 print(f"请求耗时: {end_time - start_time:.2f}秒") if response.status_code == 200: return response.json() else: self._handle_error(response) except requests.exceptions.Timeout: print("请求超时，请检查网络连接或增加超时时间") raise except requests.exceptions.ConnectionError: print("网络连接错误，请检查API端点是否正确") raise def _log_response_time(self, response, *args, **kwargs): """记录响应时间""" response.elapsed_time = time.time() - response.request.start_time def _handle_error(self, response): """处理API错误响应""" error_data = response.json() error_code = response.status_code error_handlers = { 400: "请求参数错误，请检查输入格式", 401: "API密钥无效或过期", 429: "请求频率超限，请稍后重试", 500: "服务器内部错误", 503: "服务暂时不可用" } default_message = error_handlers.get(error_code, "未知错误") print(f"API错误 {error_code}: {default_message}") print(f"详细错误: {error_data}") raise Exception(f"API调用失败: {default_message}")

3.2 网络超时与重试机制

网络不稳定是LLM应用中的常见问题，需要实现健壮的重试机制：

def send_request_with_retry(client, params, max_retries=3, base_delay=1): """带重试机制的请求发送""" for attempt in range(max_retries + 1): try: return client.send_request("chat/completions", params) except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: if attempt == max_retries: print("达到最大重试次数，请求失败") raise # 指数退避策略 delay = base_delay * (2 ** attempt) print(f"第{attempt + 1}次请求失败，{delay}秒后重试: {e}") time.sleep(delay) except Exception as e: # 非网络错误立即抛出 print(f"非网络错误，不再重试: {e}") raise return None

3.3 请求头与认证处理

正确的认证信息是请求成功的前提，需要妥善管理API密钥等敏感信息：

import os from datetime import datetime class SecureLLMClient(LLMClient): def __init__(self, api_key=None): # 优先从环境变量获取API密钥 if api_key is None: api_key = os.getenv("LLM_API_KEY") if not api_key: raise ValueError("未提供API密钥且环境变量LLM_API_KEY未设置") super().__init__(api_key) # 添加额外的安全头 self.session.headers.update({ "User-Agent": "LLM-Client/1.0", "X-Request-ID": self._generate_request_id() }) def _generate_request_id(self): """生成唯一请求ID用于追踪""" timestamp = datetime.now().strftime("%Y%m%d%H%M%S") random_suffix = os.urandom(4).hex() return f"req_{timestamp}_{random_suffix}" def update_headers(self, additional_headers): """动态更新请求头""" self.session.headers.update(additional_headers)

4. 服务器端处理流程

4.1 请求接收与验证

当请求到达服务器后，API网关会进行初步验证：

# 模拟服务器端请求验证逻辑 def validate_incoming_request(request_data): """服务器端请求验证""" validation_checks = [ # 认证验证 lambda: validate_auth(request_data.get('auth_token')), # 速率限制检查 lambda: check_rate_limit(request_data.get('user_id')), # 参数格式验证 lambda: validate_parameters(request_data.get('parameters')), # 内容安全检查 lambda: content_safety_check(request_data.get('prompt')) ] for check in validation_checks: result, error = check() if not result: return False, error return True, "验证通过" def validate_auth(auth_token): """验证认证令牌""" if not auth_token: return False, "缺少认证令牌" # 模拟令牌验证逻辑 if not auth_token.startswith("sk-"): return False, "无效的认证令牌格式" return True, None def check_rate_limit(user_id): """检查用户请求频率""" # 模拟速率限制逻辑 current_minute = datetime.now().minute request_key = f"rate_limit:{user_id}:{current_minute}" # 这里应该是Redis或其他存储的查询 current_requests = 0 # 从存储获取实际值 if current_requests >= 60: # 每分钟60次限制 return False, "请求频率超限" return True, None

4.2 令牌化与输入编码

输入文本需要转换为模型可以理解的数字序列，这个过程称为令牌化：

# 模拟令牌化过程 class Tokenizer: def __init__(self, vocab_file): self.vocab = self.load_vocabulary(vocab_file) self.vocab_size = len(self.vocab) def load_vocabulary(self, vocab_file): """加载词汇表""" # 实际实现中会从文件加载词汇表 return {"<|endoftext|>": 0, "hello": 1, "world": 2, "the": 3, "a": 4} def tokenize(self, text): """将文本转换为令牌序列""" tokens = [] words = text.lower().split() for word in words: if word in self.vocab: tokens.append(self.vocab[word]) else: # 处理未知词汇 tokens.append(self.vocab["<|endoftext|>"]) return tokens def encode(self, text, max_length=2048): """编码文本并处理长度限制""" tokens = self.tokenize(text) # 截断或填充到固定长度 if len(tokens) > max_length: tokens = tokens[:max_length] elif len(tokens) < max_length: tokens.extend([0] * (max_length - len(tokens))) return tokens # 使用示例 tokenizer = Tokenizer("vocab.txt") input_text = "Hello world, this is a test" token_ids = tokenizer.encode(input_text) print(f"输入文本: {input_text}") print(f"令牌序列: {token_ids}")

4.3 模型推理与计算

令牌序列进入模型进行前向传播计算，生成下一个令牌的概率分布：

import numpy as np class LLMInference: def __init__(self, model_weights): self.weights = model_weights self.hidden_size = 512 # 示例隐藏层大小 def forward_pass(self, input_ids): """模拟模型前向传播""" batch_size, seq_length = input_ids.shape # 嵌入层 embeddings = self.embedding_lookup(input_ids) # 多层Transformer处理 hidden_states = embeddings for layer in range(12): # 12层Transformer hidden_states = self.transformer_layer(hidden_states, layer) # 输出层 logits = self.output_projection(hidden_states) return logits def embedding_lookup(self, input_ids): """词嵌入查找""" # 简化实现 return np.random.random((input_ids.shape[0], input_ids.shape[1], self.hidden_size)) def transformer_layer(self, hidden_states, layer_idx): """单层Transformer计算""" # 自注意力机制 attention_output = self.self_attention(hidden_states) # 前馈网络 output = self.feed_forward(attention_output) return output def generate_next_token(self, logits, temperature=1.0, top_p=0.9): """根据logits生成下一个令牌""" # 应用温度参数 logits = logits / temperature # 应用top-p采样 probs = self.softmax(logits) sorted_probs = np.sort(probs)[::-1] cumulative_probs = np.cumsum(sorted_probs) # 找到top-p阈值 idx = np.where(cumulative_probs >= top_p)[0][0] top_p_threshold = sorted_probs[idx] # 过滤低概率令牌 filtered_probs = np.where(probs >= top_p_threshold, probs, 0) filtered_probs = filtered_probs / np.sum(filtered_probs) # 采样下一个令牌 next_token = np.random.choice(len(filtered_probs), p=filtered_probs) return next_token

5. 响应生成与流式输出

5.1 自回归生成过程

LLM通过自回归方式逐个生成令牌，直到满足停止条件：

def autoregressive_generation(model, prompt_ids, max_length=100, stop_tokens=[0]): """自回归文本生成""" generated_ids = prompt_ids.copy() current_ids = prompt_ids for step in range(max_length - len(prompt_ids)): # 获取下一个令牌的logits logits = model.forward_pass(current_ids) # 只关注最后一个位置的logits next_token_logits = logits[:, -1, :] # 生成下一个令牌 next_token = model.generate_next_token(next_token_logits) # 检查停止条件 if next_token in stop_tokens: break # 将新令牌添加到序列中 generated_ids.append(next_token) current_ids = np.array([generated_ids]) return generated_ids # 生成过程示例 prompt_tokens = [1, 2, 3] # 初始提示词令牌 generated_tokens = autoregressive_generation( model=llm_model, prompt_ids=prompt_tokens, max_length=50, stop_tokens=[0] # 结束令牌 )

5.2 流式输出处理

对于长文本生成，流式输出可以提升用户体验：

import json def stream_generation(client, prompt, callback=None): """流式生成文本""" stream_params = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": prompt}], "stream": True, # 启用流式输出 "max_tokens": 1000 } response = client.send_request("chat/completions", stream_params) collected_content = "" for chunk in response: if chunk.get('choices') and chunk['choices'][0].get('delta'): content = chunk['choices'][0]['delta'].get('content', '') if content: collected_content += content # 调用回调函数处理每个新内容块 if callback: callback(content) # 检查生成是否完成 if chunk['choices'][0].get('finish_reason'): break return collected_content # 使用流式生成的示例 def print_chunk(content): """简单的流式输出回调函数""" print(content, end='', flush=True) # 调用流式生成 full_response = stream_generation( client=llm_client, prompt="请详细解释人工智能的发展历史", callback=print_chunk )

5.3 停止条件与长度控制

合理的停止条件确保生成的文本既完整又不过长：

class GenerationController: def __init__(self, max_tokens=1000, stop_sequences=None): self.max_tokens = max_tokens self.stop_sequences = stop_sequences or ["\n\n", "。", "!"] self.generated_tokens = 0 def should_stop(self, current_text, new_token_text): """检查是否应该停止生成""" self.generated_tokens += 1 # 检查令牌数量限制 if self.generated_tokens >= self.max_tokens: return True, "max_tokens_reached" current_text += new_token_text # 检查停止序列 for stop_seq in self.stop_sequences: if current_text.endswith(stop_seq): return True, "stop_sequence" # 检查自然结束点 if self._is_natural_ending(current_text): return True, "natural_end" return False, None def _is_natural_ending(self, text): """判断文本是否自然结束""" # 简单的自然结束判断逻辑 endings = [".", "!", "?", "。", "！", "？"] return any(text.strip().endswith(ending) for ending in endings) # 使用生成控制器 controller = GenerationController(max_tokens=500, stop_sequences=["\n\n", "总结："]) def controlled_generation(prompt): current_text = prompt while True: # 生成下一个令牌 next_token = generate_next_token(current_text) current_text += next_token # 检查停止条件 should_stop, reason = controller.should_stop(current_text, next_token) if should_stop: print(f"生成停止，原因: {reason}") break return current_text

6. 响应后处理与返回

6.1 响应格式标准化

将模型原始输出转换为标准化的API响应格式：

def format_api_response(raw_output, request_id, usage_stats): """格式化API响应""" response = { "id": f"chatcmpl-{request_id}", "object": "chat.completion", "created": int(time.time()), "model": "gpt-3.5-turbo", "choices": [ { "index": 0, "message": { "role": "assistant", "content": raw_output }, "finish_reason": "stop" } ], "usage": { "prompt_tokens": usage_stats["prompt_tokens"], "completion_tokens": usage_stats["completion_tokens"], "total_tokens": usage_stats["total_tokens"] } } return response # 使用示例 raw_output = "人工智能是模拟人类智能的计算机系统..." usage_stats = { "prompt_tokens": 150, "completion_tokens": 200, "total_tokens": 350 } formatted_response = format_api_response( raw_output=raw_output, request_id="abc123", usage_stats=usage_stats ) print(json.dumps(formatted_response, indent=2, ensure_ascii=False))

6.2 内容安全与质量检查

在返回响应前进行内容安全检查：

class ContentSafetyFilter: def __init__(self, sensitive_words_file): self.sensitive_words = self.load_sensitive_words(sensitive_words_file) def load_sensitive_words(self, file_path): """加载敏感词库""" # 实际实现中从文件加载 return ["暴力", "违法", "不良内容"] def check_safety(self, text): """检查内容安全性""" violations = [] for word in self.sensitive_words: if word in text: violations.append({ "type": "sensitive_content", "word": word, "position": text.find(word) }) # 检查文本质量 if self._is_low_quality(text): violations.append({"type": "low_quality"}) return len(violations) == 0, violations def _is_low_quality(self, text): """判断文本质量""" # 简单的质量检查逻辑 if len(text) < 10: return True if text.count(" ") / len(text) > 0.5: # 空格比例过高 return True return False def filter_content(self, text, replacement="[已过滤]"): """过滤敏感内容""" safe_text = text for word in self.sensitive_words: safe_text = safe_text.replace(word, replacement) return safe_text # 使用安全过滤器 safety_filter = ContentSafetyFilter("sensitive_words.txt") is_safe, violations = safety_filter.check_safety(generated_text) if not is_safe: print(f"内容安全问题: {violations}") safe_text = safety_filter.filter_content(generated_text) else: safe_text = generated_text

6.3 性能监控与日志记录

记录详细的请求响应信息用于监控和分析：

import logging from datetime import datetime class RequestLogger: def __init__(self, log_file="llm_requests.log"): self.logger = logging.getLogger("LLMRequestLogger") self.logger.setLevel(logging.INFO) # 创建文件处理器 file_handler = logging.FileHandler(log_file) formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) file_handler.setFormatter(formatter) self.logger.addHandler(file_handler) def log_request(self, request_id, endpoint, params, start_time): """记录请求信息""" log_data = { "request_id": request_id, "endpoint": endpoint, "params_keys": list(params.keys()), "start_time": start_time.isoformat(), "prompt_length": len(str(params.get('messages', ''))) } self.logger.info(f"REQUEST: {json.dumps(log_data)}") def log_response(self, request_id, response, end_time, duration): """记录响应信息""" log_data = { "request_id": request_id, "end_time": end_time.isoformat(), "duration_seconds": duration, "response_tokens": response.get('usage', {}).get('completion_tokens', 0), "status": "success" } self.logger.info(f"RESPONSE: {json.dumps(log_data)}") def log_error(self, request_id, error_type, error_message): """记录错误信息""" log_data = { "request_id": request_id, "error_type": error_type, "error_message": error_message, "timestamp": datetime.now().isoformat() } self.logger.error(f"ERROR: {json.dumps(log_data)}") # 使用日志记录器 request_logger = RequestLogger() def monitored_request(client, endpoint, params): request_id = client._generate_request_id() start_time = datetime.now() request_logger.log_request(request_id, endpoint, params, start_time) try: response = client.send_request(endpoint, params) end_time = datetime.now() duration = (end_time - start_time).total_seconds() request_logger.log_response(request_id, response, end_time, duration) return response except Exception as e: request_logger.log_error(request_id, type(e).__name__, str(e)) raise

7. 常见错误分析与解决方案

7.1 认证与权限错误

认证错误是LLM集成中最常见的问题之一：

# 常见认证错误及解决方案 AUTH_ERRORS = { "invalid_api_key": { "症状": "API密钥无效或格式错误", "原因": "密钥过期、拼写错误或格式不正确", "解决方案": [ "检查API密钥是否正确复制", "验证密钥是否在有效期内", "确认密钥格式符合要求" ] }, "insufficient_quota": { "症状": "额度不足或超过使用限制", "原因": "账户余额不足或达到速率限制", "解决方案": [ "检查账户余额和使用情况", "调整请求频率或升级套餐", "联系客服申请提高限制" ] }, "account_not_active": { "症状": "账户未激活或受限", "原因": "新账户未完成验证或账户被暂停", "解决方案": [ "完成邮箱验证和手机验证", "检查账户状态和通知信息", "联系支持团队解决账户问题" ] } } def handle_auth_error(error_code, details): """处理认证错误""" if error_code in AUTH_ERRORS: error_info = AUTH_ERRORS[error_code] print(f"认证错误: {error_info['症状']}") print(f"可能原因: {error_info['原因']}") print("解决方案:") for i, solution in enumerate(error_info['解决方案'], 1): print(f" {i}. {solution}") else: print(f"未知认证错误: {error_code}") print(f"错误详情: {details}")

7.2 网络与超时错误

网络相关问题通常需要客户端和服务器端协同解决：

NETWORK_ERRORS = { "request_timeout": { "症状": "请求超时，未在指定时间内获得响应", "原因": [ "网络连接不稳定", "服务器负载过高", "请求内容过于复杂", "客户端超时设置过短" ], "解决方案": [ "增加请求超时时间（如从30秒增加到60秒）", "检查网络连接稳定性", "简化请求内容或减少max_tokens", "实现指数退避重试机制" ] }, "connection_reset": { "症状": "连接被重置或中断", "原因": [ "防火墙或代理设置问题", "服务器端主动断开连接", "网络中间设备问题" ], "解决方案": [ "检查防火墙和代理配置", "验证API端点URL是否正确", "尝试使用不同的网络环境" ] } } def handle_network_error(error_type, context): """处理网络错误""" print(f"网络错误类型: {error_type}") if error_type in NETWORK_ERRORS: error_info = NETWORK_ERRORS[error_type] print("可能的原因和解决方案:") for cause in error_info["原因"]: print(f" - 原因: {cause}") for solution in error_info["解决方案"]: print(f" - 解决方案: {solution}") # 提供具体的调试建议 print("\n调试建议:") print("1. 使用ping或traceroute检查网络连通性") print("2. 检查DNS解析是否正常") print("3. 验证SSL证书有效性") print("4. 查看服务器状态页面确认服务可用性")

7.3 内容与参数错误

参数配置错误会导致请求被拒绝或生成质量不佳：

PARAMETER_ERRORS = { "max_tokens_exceeded": { "症状": "max_tokens参数超过模型限制", "原因": "设置的令牌数超过模型最大支持", "解决方案": "减少max_tokens值，通常保持在2048以内" }, "invalid_temperature": { "症状": "temperature参数超出有效范围", "原因": "温度值不在0-2范围内", "解决方案": "将temperature调整为0到2之间的值" }, "content_filter_violation": { "症状": "输入或输出内容触发安全过滤", "原因": "提示词或生成内容包含敏感信息", "解决方案": "修改提示词避免敏感话题，或调整内容过滤设置" } } def validate_parameters_before_send(params): """发送前参数验证""" errors = [] # 检查max_tokens max_tokens = params.get('max_tokens', 0) if max_tokens > 4000: errors.append("max_tokens不能超过4000") # 检查temperature temperature = params.get('temperature', 1.0) if not 0 <= temperature <= 2: errors.append("temperature必须在0到2之间") # 检查提示词长度 messages = params.get('messages', []) total_length = sum(len(msg.get('content', '')) for msg in messages) if total_length > 10000: errors.append("提示词总长度超过限制") return errors # 使用参数验证 param_errors = validate_parameters_before_send(request_params) if param_errors: print("参数错误发现:") for error in param_errors: print(f" - {error}") # 修正参数或抛出异常

8. 性能优化与最佳实践

8.1 请求批处理优化

对于多个相似请求，使用批处理可以显著提升效率：

def batch_requests(requests_list, batch_size=5): """批处理多个LLM请求""" batches = [requests_list[i:i + batch_size] for i in range(0, len(requests_list), batch_size)] all_responses = [] for batch in batches: # 准备批处理请求 batch_requests = [] for req in batch: batch_requests.append({ "model": req["model"], "messages": req["messages"], "max_tokens": req.get("max_tokens", 1000) }) # 发送批处理请求（如果API支持） try: batch_response = send_batch_request(batch_requests) all_responses.extend(batch_response) except Exception as e: print(f"批处理请求失败: {e}") # 回退到单个请求 for req in batch: try: single_response = send_single_request(req) all_responses.append(single_response) except Exception as single_error: print(f"单个请求也失败: {single_error}") all_responses.append(None) return all_responses def send_batch_request(batch_requests): """发送批处理请求（模拟实现）""" # 实际实现中会使用支持批处理的API responses = [] for req in batch_requests: # 模拟批处理延迟优势 response = send_single_request(req) responses.append(response) return responses

8.2 缓存策略实现

对重复或相似的请求使用缓存减少API调用：

import hashlib from functools import lru_cache class LLMCache: def __init__(self, max_size=1000): self.cache = {} self.max_size = max_size def _generate_cache_key(self, params): """生成缓存键""" # 基于请求参数生成唯一键 key_data = { "model": params.get("model"), "messages": params.get("messages", []), "temperature": params.get("temperature", 1.0), "max_tokens": params.get("max_tokens", 1000) } key_string = json.dumps(key_data, sort_keys=True) return hashlib.md5(key_string.encode()).hexdigest() def get(self, params): """从缓存获取响应""" key = self._generate_cache_key(params) if key in self.cache: cached_item = self.cache[key] # 检查缓存是否过期（例如1小时） if time.time() - cached_item["timestamp"] < 3600: return cached_item["response"] return None def set(self, params, response): """设置缓存""" if len(self.cache) >= self.max_size: # 简单的LRU策略：移除最旧的项 oldest_key = min(self.cache.keys(), key=lambda k: self.cache[k]["timestamp"]) del self.cache[oldest_key] key = self._generate_cache_key(params) self.cache[key] = { "response": response, "timestamp": time.time() } # 使用缓存的客户端 class CachedLLMClient(LLMClient): def __init__(self, api_key, cache_size=1000): super().__init__(api_key) self.cache = LLMCache(cache_size) def send_cached_request(self, params): """带缓存的请求发送""" # 检查缓存 cached_response = self.cache.get(params) if cached_response: print("缓存命中") return cached_response # 发送实际请求 response = self.send_request("chat/completions", params) # 缓存响应 self.cache.set(params, response) return response

8.3 监控与告警配置

建立完整的监控体系确保服务可靠性：

class LLMMonitor: def __init__(self, metrics_collector): self.metrics = metrics_collector self.alerts_config = self.load_alerts_config() def load_alerts_config(self): """加载告警配置""" return { "high_error_rate": { "threshold": 0.1, # 10%错误率 "window_minutes": 5, "message": "错误率超过阈值" }, "high_latency": { "threshold": 10.0, # 10秒平均延迟 "window_minutes": 5, "message": "平均响应时间过长" }, "quota_usage": { "threshold": 0.8, #