在实际企业级 AI 应用开发中直接调用大模型原生 API 往往会遇到上下文长度限制、配额不足、响应中断、密钥管理复杂等一系列工程挑战。近期月之暗面推出的 Kimi Hosted Agent 平台正是为了帮助企业开发者更稳定、更高效地集成和调用大模型能力其 B 端收入有七成来自 API 调用这反映出市场对可靠 API 服务的强烈需求。本文将围绕如何构建一个稳定、可维护的大模型 API 集成方案展开重点解决开发者在调用 Kimi、DeepSeek、智谱等模型 API 时常见的错误如400 Bad Request上下文超长、402 Insufficient Balance余额不足、响应中途关闭等问题。我们会从环境准备、密钥管理、请求构造、错误处理、生产级最佳实践等多个维度提供一个可复现的实战指南。1. 理解大模型 API 的核心挑战与 Kimi Hosted Agent 的定位在直接调用大模型 API 时开发者通常会遇到几个核心挑战这些挑战也是 Kimi Hosted Agent 这类托管平台着力解决的问题。1.1 常见的 API 错误类型及其根源大模型 API 的调用错误并非偶然其背后有明确的资源限制和规则约束。以下是一些高频错误码及其含义错误码/现象触发条件根本原因400 Bad Request提示上下文超长请求的 tokens 总数超过模型上限模型有固定的上下文窗口如 128K、200K。输入输出的 tokens 数不能超过此限制。402 Insufficient Balance调用 API 时API 密钥关联的账户余额或套餐额度已用完。Connection closed mid-response流式响应过程中连接中断网络不稳定、客户端超时设置过短、或服务端生成响应时间过长。Response exceeded output token maximum模型生成的内容太长即使总上下文未超限单次生成的输出 tokens 数也可能有独立限制。Kimi Hosted Agent 平台通过托管模型实例、自动管理上下文窗口、提供更稳定的网络链路和计费方式旨在降低开发者直接处理这些问题的复杂度。1.2 Hosted Agent 与原生 API 调用的关键差异对于企业开发者而言选择原生 API 还是 Hosted Agent 平台是一个重要的技术选型决策。两者的核心差异如下维度原生 API 调用Hosted Agent 平台上下文管理开发者需自行计算和分块确保单次请求不超限。平台通常提供更优的上下文管理策略甚至支持超长文档的自动处理。稳定性与性能依赖公共网络可能受地域和运营商影响。通常提供专线或优化链路承诺更高的 SLA服务等级协议。计费与配额按 tokens 计费需自行监控余额防止因余额不足导致业务中断。可能提供更灵活的套餐包、月结模式并有用量预警机制。功能扩展仅限于模型提供的标准接口。可能集成文件上传、代码执行、长会话管理等增值功能。理解这些差异有助于我们设计一个更具弹性的集成架构即使暂时不使用托管平台也能借鉴其思路来优化自己的代码。2. 环境准备与依赖配置构建一个健壮的 API 调用客户端首先需要规范开发环境和管理依赖。2.1 项目初始化与依赖管理以一个典型的 Python 项目为例使用pip和requirements.txt来管理依赖。核心库包括用于发起 HTTP 请求的requests和处理环境变量的python-dotenv。创建项目目录并初始化虚拟环境# 创建项目目录 mkdir robust_llm_client cd robust_llm_client # 创建并激活虚拟环境推荐使用 Python 3.8 python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 创建依赖文件 touch requirements.txt在requirements.txt中声明依赖requests2.28.0 python-dotenv1.0.0 tiktoken0.5.0 # 用于精确计算 tokens避免超限安装依赖pip install -r requirements.txt2.2 安全地管理 API 密钥绝对不要将 API Key 硬编码在代码中。使用环境变量或配置文件是基本的安全规范。创建.env文件来存储密钥# .env KIMI_API_KEYyour_kimi_api_key_here DEEPSEEK_API_KEYyour_deepseek_api_key_here ZHIPU_API_KEYyour_zhipu_api_key_here在代码中通过os.getenv读取# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class APIConfig: KIMI_API_KEY os.getenv(KIMI_API_KEY) DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY) ZHIPU_API_KEY os.getenv(ZHIPU_API_KEY) # 各模型 API 的基地址 KIMI_BASE_URL https://api.moonshot.cn/v1 DEEPSEEK_BASE_URL https://api.deepseek.com/v1 ZHIPU_BASE_URL https://open.bigmodel.cn/api/paas/v4同时将.env加入.gitignore以避免意外提交# .gitignore .env __pycache__/ *.pyc3. 构建健壮的 API 客户端类一个良好的客户端类应具备请求构造、令牌计算、错误重试、响应解析等核心功能。3.1 基础客户端结构与令牌计算首先实现一个基础客户端它能够计算提示词的 tokens 数量这是避免400错误的关键。# llm_client.py import requests import tiktoken import time import json from typing import Optional, Dict, Any from config import APIConfig class RobustLLMClient: def __init__(self, provider: str kimi): self.provider provider self.api_key self._get_api_key(provider) self.base_url self._get_base_url(provider) self.encoding tiktoken.get_encoding(cl100k_base) # 多数新模型使用此编码 def _get_api_key(self, provider: str) - str: 安全地获取 API Key key_map { kimi: APIConfig.KIMI_API_KEY, deepseek: APIConfig.DEEPSEEK_API_KEY, zhipu: APIConfig.ZHIPU_API_KEY } key key_map.get(provider) if not key: raise ValueError(fUnsupported provider or missing API key for: {provider}) return key def _get_base_url(self, provider: str) - str: 获取 API 基地址 url_map { kimi: APIConfig.KIMI_BASE_URL, deepseek: APIConfig.DEEPSEEK_BASE_URL, zhipu: APIConfig.ZHIPU_BASE_URL } return url_map.get(provider, ) def count_tokens(self, text: str) - int: 计算一段文本的 tokens 数量 return len(self.encoding.encode(text)) def estimate_conversation_tokens(self, messages: list) - int: 估算一个对话消息列表的总 tokens 数。 注意这只是估算实际 API 计算可能包含额外开销。 total_tokens 0 for message in messages: # 每条消息通常包含 role, content 等字段 content message.get(content, ) total_tokens self.count_tokens(content) # 为 role 和结构开销增加一些 tokens total_tokens 5 return total_tokens3.2 实现带重试机制的请求方法网络波动和服务端瞬时故障是导致Connection closed mid-response的常见原因实现重试机制至关重要。# 在 RobustLLMClient 类中继续添加 class RobustLLMClient: # ... 之前的代码 ... def _make_request_with_retry(self, endpoint: str, payload: Dict, max_retries: int 3, initial_backoff: float 1.0) - Optional[Dict]: 带指数退避重试的请求方法 url f{self.base_url}/{endpoint} headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } for attempt in range(max_retries 1): # 包括首次尝试 try: response requests.post(url, headersheaders, jsonpayload, timeout60) if response.status_code 200: return response.json() elif response.status_code 400: # 业务逻辑错误如 tokens 超限重试无意义 error_info response.json().get(error, {}) raise ValueError(fAPI Request Error (400): {error_info.get(message, Unknown)}) elif response.status_code 402: # 余额不足需要人工处理 raise ValueError(Insufficient balance. Please recharge your account.) elif response.status_code in [429, 500, 502, 503]: # 限流或服务端错误可以重试 if attempt max_retries: backoff_time initial_backoff * (2 ** attempt) # 指数退避 print(fRequest failed with status {response.status_code}. Retrying in {backoff_time}s...) time.sleep(backoff_time) continue else: raise Exception(fAPI request failed after {max_retries} retries. Status: {response.status_code}) else: # 其他错误 response.raise_for_status() except requests.exceptions.Timeout: if attempt max_retries: print(fRequest timeout. Retrying...) time.sleep(initial_backoff * (2 ** attempt)) continue else: raise Exception(Request timed out after multiple retries.) except requests.exceptions.ConnectionError as e: if attempt max_retries: print(fConnection error: {e}. Retrying...) time.sleep(initial_backoff * (2 ** attempt)) continue else: raise Exception(Connection failed after multiple retries.) return None def chat_completion(self, messages: list, model: str kimi-v1, max_tokens: int 2000, temperature: float 0.7) - Dict: 发送聊天补全请求并包含 tokens 检查 # 1. 预检查 tokens 数量 estimated_tokens self.estimate_conversation_tokens(messages) max_tokens print(fEstimated tokens: {estimated_tokens}) # 不同模型的上下文窗口限制示例值需根据实际模型调整 context_limits { kimi-v1: 128000, deepseek-chat: 128000, zhipu-glm-4: 128000 } model_limit context_limits.get(model, 4000) # 默认一个安全值 if estimated_tokens model_limit: raise ValueError(fEstimated tokens ({estimated_tokens}) exceed models context limit ({model_limit}). Please shorten your prompt or reduce max_tokens.) # 2. 构造请求体 payload { model: model, messages: messages, max_tokens: max_tokens, temperature: temperature, stream: False # 非流式响应更简单先确保基础功能稳定 } # 3. 发送请求 endpoint chat/completions # 多数提供商使用此端点 if self.provider zhipu: endpoint chat/completions # 智谱等可能略有不同需参考其文档 return self._make_request_with_retry(endpoint, payload)4. 实战处理长上下文与流式响应对于需要处理长文档或希望实现打字机效果的场景需要更高级的技巧。4.1 智能处理长文本输入当输入文本超过模型限制时简单的截断会丢失信息。更优的策略是进行智能分块和摘要。# 在 RobustLLMClient 类中添加长文本处理方法 class RobustLLMClient: # ... 之前的代码 ... def split_text_into_chunks(self, text: str, chunk_size: int 1000, overlap: int 50) - list: 将长文本按 tokens 数分块块与块之间有一定重叠避免语义断裂。 tokens self.encoding.encode(text) chunks [] start 0 while start len(tokens): end start chunk_size chunk_tokens tokens[start:end] chunk_text self.encoding.decode(chunk_tokens) chunks.append(chunk_text) start end - overlap # 重叠一部分保持上下文连贯 return chunks def summarize_long_document(self, long_text: str, model: str) - str: 通过递归摘要的方式处理超长文档 max_chunk_tokens 30000 # 设定一个安全的分块大小 if self.count_tokens(long_text) max_chunk_tokens: # 如果文本不长直接处理 messages [ {role: user, content: f请为以下文本生成一个简洁的摘要\n\n{long_text}} ] response self.chat_completion(messages, modelmodel, max_tokens500) return response[choices][0][message][content] else: # 文本过长先分块再递归摘要 chunks self.split_text_into_chunks(long_text, chunk_sizemax_chunk_tokens) chunk_summaries [] for i, chunk in enumerate(chunks): print(fSummarizing chunk {i1}/{len(chunks)}...) summary self.summarize_long_document(chunk, model) # 递归调用 chunk_summaries.append(summary) # 将所有分块的摘要合并再生成最终摘要 combined_summaries \n.join(chunk_summaries) final_messages [ {role: user, content: f以下是同一文档多个部分的摘要请将它们整合成一个连贯的总体摘要\n\n{combined_summaries}} ] response self.chat_completion(final_messages, modelmodel, max_tokens800) return response[choices][0][message][content]4.2 实现稳定的流式响应流式响应可以提升用户体验但需要更细致的超时和网络错误处理。# 在 RobustLLMClient 类中添加流式响应方法 class RobustLLMClient: # ... 之前的代码 ... def stream_chat_completion(self, messages: list, model: str kimi-v1, max_tokens: int 2000, temperature: float 0.7): 流式响应版本适用于需要实时显示生成内容的场景 import json url f{self.base_url}/chat/completions headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { model: model, messages: messages, max_tokens: max_tokens, temperature: temperature, stream: True # 开启流式 } try: # 设置更长的超时时间因为流式响应可能持续较久 response requests.post(url, headersheaders, jsonpayload, timeout120, streamTrue) response.raise_for_status() full_content for line in response.iter_lines(): if line: line line.decode(utf-8) if line.startswith(data: ): data_str line[6:] # 去掉 data: 前缀 if data_str [DONE]: break try: data json.loads(data_str) delta data[choices][0][delta] if content in delta: content_piece delta[content] full_content content_piece yield content_piece # 逐块 yield 给调用者 except json.JSONDecodeError: print(fFailed to parse JSON: {data_str}) continue print(f\n[Stream completed. Full response length: {len(full_content)}]) except requests.exceptions.Timeout: yield [ERROR] Stream request timed out. except requests.exceptions.ConnectionError: yield [ERROR] Connection lost during streaming. except Exception as e: yield f[ERROR] An error occurred: {str(e)}5. 生产环境的最佳实践与错误排查将代码用于生产环境时需要额外的保障措施。5.1 配置监控与告警API 调用的稳定性和成本需要被监控。以下是一个简单的监控装饰器示例# monitoring.py import time import functools from datetime import datetime def monitor_llm_call(func): 监控 API 调用的装饰器记录耗时、tokens 用量和状态 functools.wraps(func) def wrapper(*args, **kwargs): start_time time.time() start_dt datetime.now() try: result func(*args, **kwargs) end_time time.time() duration end_time - start_time # 记录成功日志生产环境应接入 ELK、Prometheus 等 log_entry { timestamp: start_dt.isoformat(), function: func.__name__, status: success, duration_seconds: round(duration, 2), input_tokens: kwargs.get(estimated_tokens, N/A), # 需要实际获取 output_tokens: len(result[choices][0][message][content]) if result else N/A # 简化估算 } print(f[MONITOR] {log_entry}) return result except Exception as e: end_time time.time() duration end_time - start_time log_entry { timestamp: start_dt.isoformat(), function: func.__name__, status: error, duration_seconds: round(duration, 2), error: str(e) } print(f[MONITOR] {log_entry}) raise e return wrapper # 使用装饰器 monitor_llm_call def safe_chat_completion(client, messages, model, max_tokens): return client.chat_completion(messages, model, max_tokens)5.2 常见问题排查清单当 API 调用出现问题时可以按以下清单快速定位。问题现象优先检查点解决方案所有请求返回400 Bad Request1. API Key 是否正确且未过期2. 请求的 URL 端点是否正确3. 请求体 JSON 格式是否正确1. 核对 .env 文件中的密钥2. 查阅官方文档确认端点3. 使用 jsonlint 验证格式间歇性429 Too Many Requests1. 是否触发了速率限制2. 同一密钥是否在多处使用1. 降低请求频率加入随机延迟2. 为不同服务使用不同密钥流式响应中途断开1. 客户端或服务端超时设置2. 网络稳定性3. 生成内容过长1. 增加 timeout 参数2. 检查网络连接3. 限制 max_tokens402 Insufficient Balance1. 账户余额是否充足2. 套餐额度是否用完1. 登录平台控制台查看余额2. 升级套餐或充值5.3 密钥轮换与容灾策略对于关键业务应考虑多密钥和多个模型供应商的容灾方案。# 简单的多供应商容灾客户端 class MultiProviderLLMClient: def __init__(self, providers: list None): self.providers providers or [kimi, deepseek, zhipu] self.clients [RobustLLMClient(provider) for provider in self.providers] def chat_with_fallback(self, messages, model_mapNone, **kwargs): 按顺序尝试多个供应商直到有一个成功 model_map model_map or { kimi: kimi-v1, deepseek: deepseek-chat, zhipu: zhipu-glm-4 } for i, client in enumerate(self.clients): provider self.providers[i] model model_map.get(provider) try: print(fTrying provider: {provider}) result client.chat_completion(messages, modelmodel, **kwargs) print(fSuccess with provider: {provider}) return result, provider except Exception as e: print(fProvider {provider} failed: {e}) continue raise Exception(All providers failed.) # 使用示例 fallback_client MultiProviderLLMClient() result, successful_provider fallback_client.chat_with_fallback( messages[{role: user, content: 你好请介绍你自己。}], max_tokens500 )通过上述实践我们构建了一个具备错误处理、重试机制、长文本支持和多供应商容灾能力的稳健的 LLM API 客户端。这种设计思路与 Kimi Hosted Agent 等平台的目标一致即在享受大模型能力的同时最大限度地降低集成复杂度和运维风险。在实际项目中还需根据具体的业务需求、流量规模和合规要求进一步设计限流、降级、审计等高级功能。
企业级大模型API集成实战：解决上下文超长、余额不足等核心挑战
发布时间：2026/7/23 3:06:42
在实际企业级 AI 应用开发中直接调用大模型原生 API 往往会遇到上下文长度限制、配额不足、响应中断、密钥管理复杂等一系列工程挑战。近期月之暗面推出的 Kimi Hosted Agent 平台正是为了帮助企业开发者更稳定、更高效地集成和调用大模型能力其 B 端收入有七成来自 API 调用这反映出市场对可靠 API 服务的强烈需求。本文将围绕如何构建一个稳定、可维护的大模型 API 集成方案展开重点解决开发者在调用 Kimi、DeepSeek、智谱等模型 API 时常见的错误如400 Bad Request上下文超长、402 Insufficient Balance余额不足、响应中途关闭等问题。我们会从环境准备、密钥管理、请求构造、错误处理、生产级最佳实践等多个维度提供一个可复现的实战指南。1. 理解大模型 API 的核心挑战与 Kimi Hosted Agent 的定位在直接调用大模型 API 时开发者通常会遇到几个核心挑战这些挑战也是 Kimi Hosted Agent 这类托管平台着力解决的问题。1.1 常见的 API 错误类型及其根源大模型 API 的调用错误并非偶然其背后有明确的资源限制和规则约束。以下是一些高频错误码及其含义错误码/现象触发条件根本原因400 Bad Request提示上下文超长请求的 tokens 总数超过模型上限模型有固定的上下文窗口如 128K、200K。输入输出的 tokens 数不能超过此限制。402 Insufficient Balance调用 API 时API 密钥关联的账户余额或套餐额度已用完。Connection closed mid-response流式响应过程中连接中断网络不稳定、客户端超时设置过短、或服务端生成响应时间过长。Response exceeded output token maximum模型生成的内容太长即使总上下文未超限单次生成的输出 tokens 数也可能有独立限制。Kimi Hosted Agent 平台通过托管模型实例、自动管理上下文窗口、提供更稳定的网络链路和计费方式旨在降低开发者直接处理这些问题的复杂度。1.2 Hosted Agent 与原生 API 调用的关键差异对于企业开发者而言选择原生 API 还是 Hosted Agent 平台是一个重要的技术选型决策。两者的核心差异如下维度原生 API 调用Hosted Agent 平台上下文管理开发者需自行计算和分块确保单次请求不超限。平台通常提供更优的上下文管理策略甚至支持超长文档的自动处理。稳定性与性能依赖公共网络可能受地域和运营商影响。通常提供专线或优化链路承诺更高的 SLA服务等级协议。计费与配额按 tokens 计费需自行监控余额防止因余额不足导致业务中断。可能提供更灵活的套餐包、月结模式并有用量预警机制。功能扩展仅限于模型提供的标准接口。可能集成文件上传、代码执行、长会话管理等增值功能。理解这些差异有助于我们设计一个更具弹性的集成架构即使暂时不使用托管平台也能借鉴其思路来优化自己的代码。2. 环境准备与依赖配置构建一个健壮的 API 调用客户端首先需要规范开发环境和管理依赖。2.1 项目初始化与依赖管理以一个典型的 Python 项目为例使用pip和requirements.txt来管理依赖。核心库包括用于发起 HTTP 请求的requests和处理环境变量的python-dotenv。创建项目目录并初始化虚拟环境# 创建项目目录 mkdir robust_llm_client cd robust_llm_client # 创建并激活虚拟环境推荐使用 Python 3.8 python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 创建依赖文件 touch requirements.txt在requirements.txt中声明依赖requests2.28.0 python-dotenv1.0.0 tiktoken0.5.0 # 用于精确计算 tokens避免超限安装依赖pip install -r requirements.txt2.2 安全地管理 API 密钥绝对不要将 API Key 硬编码在代码中。使用环境变量或配置文件是基本的安全规范。创建.env文件来存储密钥# .env KIMI_API_KEYyour_kimi_api_key_here DEEPSEEK_API_KEYyour_deepseek_api_key_here ZHIPU_API_KEYyour_zhipu_api_key_here在代码中通过os.getenv读取# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class APIConfig: KIMI_API_KEY os.getenv(KIMI_API_KEY) DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY) ZHIPU_API_KEY os.getenv(ZHIPU_API_KEY) # 各模型 API 的基地址 KIMI_BASE_URL https://api.moonshot.cn/v1 DEEPSEEK_BASE_URL https://api.deepseek.com/v1 ZHIPU_BASE_URL https://open.bigmodel.cn/api/paas/v4同时将.env加入.gitignore以避免意外提交# .gitignore .env __pycache__/ *.pyc3. 构建健壮的 API 客户端类一个良好的客户端类应具备请求构造、令牌计算、错误重试、响应解析等核心功能。3.1 基础客户端结构与令牌计算首先实现一个基础客户端它能够计算提示词的 tokens 数量这是避免400错误的关键。# llm_client.py import requests import tiktoken import time import json from typing import Optional, Dict, Any from config import APIConfig class RobustLLMClient: def __init__(self, provider: str kimi): self.provider provider self.api_key self._get_api_key(provider) self.base_url self._get_base_url(provider) self.encoding tiktoken.get_encoding(cl100k_base) # 多数新模型使用此编码 def _get_api_key(self, provider: str) - str: 安全地获取 API Key key_map { kimi: APIConfig.KIMI_API_KEY, deepseek: APIConfig.DEEPSEEK_API_KEY, zhipu: APIConfig.ZHIPU_API_KEY } key key_map.get(provider) if not key: raise ValueError(fUnsupported provider or missing API key for: {provider}) return key def _get_base_url(self, provider: str) - str: 获取 API 基地址 url_map { kimi: APIConfig.KIMI_BASE_URL, deepseek: APIConfig.DEEPSEEK_BASE_URL, zhipu: APIConfig.ZHIPU_BASE_URL } return url_map.get(provider, ) def count_tokens(self, text: str) - int: 计算一段文本的 tokens 数量 return len(self.encoding.encode(text)) def estimate_conversation_tokens(self, messages: list) - int: 估算一个对话消息列表的总 tokens 数。 注意这只是估算实际 API 计算可能包含额外开销。 total_tokens 0 for message in messages: # 每条消息通常包含 role, content 等字段 content message.get(content, ) total_tokens self.count_tokens(content) # 为 role 和结构开销增加一些 tokens total_tokens 5 return total_tokens3.2 实现带重试机制的请求方法网络波动和服务端瞬时故障是导致Connection closed mid-response的常见原因实现重试机制至关重要。# 在 RobustLLMClient 类中继续添加 class RobustLLMClient: # ... 之前的代码 ... def _make_request_with_retry(self, endpoint: str, payload: Dict, max_retries: int 3, initial_backoff: float 1.0) - Optional[Dict]: 带指数退避重试的请求方法 url f{self.base_url}/{endpoint} headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } for attempt in range(max_retries 1): # 包括首次尝试 try: response requests.post(url, headersheaders, jsonpayload, timeout60) if response.status_code 200: return response.json() elif response.status_code 400: # 业务逻辑错误如 tokens 超限重试无意义 error_info response.json().get(error, {}) raise ValueError(fAPI Request Error (400): {error_info.get(message, Unknown)}) elif response.status_code 402: # 余额不足需要人工处理 raise ValueError(Insufficient balance. Please recharge your account.) elif response.status_code in [429, 500, 502, 503]: # 限流或服务端错误可以重试 if attempt max_retries: backoff_time initial_backoff * (2 ** attempt) # 指数退避 print(fRequest failed with status {response.status_code}. Retrying in {backoff_time}s...) time.sleep(backoff_time) continue else: raise Exception(fAPI request failed after {max_retries} retries. Status: {response.status_code}) else: # 其他错误 response.raise_for_status() except requests.exceptions.Timeout: if attempt max_retries: print(fRequest timeout. Retrying...) time.sleep(initial_backoff * (2 ** attempt)) continue else: raise Exception(Request timed out after multiple retries.) except requests.exceptions.ConnectionError as e: if attempt max_retries: print(fConnection error: {e}. Retrying...) time.sleep(initial_backoff * (2 ** attempt)) continue else: raise Exception(Connection failed after multiple retries.) return None def chat_completion(self, messages: list, model: str kimi-v1, max_tokens: int 2000, temperature: float 0.7) - Dict: 发送聊天补全请求并包含 tokens 检查 # 1. 预检查 tokens 数量 estimated_tokens self.estimate_conversation_tokens(messages) max_tokens print(fEstimated tokens: {estimated_tokens}) # 不同模型的上下文窗口限制示例值需根据实际模型调整 context_limits { kimi-v1: 128000, deepseek-chat: 128000, zhipu-glm-4: 128000 } model_limit context_limits.get(model, 4000) # 默认一个安全值 if estimated_tokens model_limit: raise ValueError(fEstimated tokens ({estimated_tokens}) exceed models context limit ({model_limit}). Please shorten your prompt or reduce max_tokens.) # 2. 构造请求体 payload { model: model, messages: messages, max_tokens: max_tokens, temperature: temperature, stream: False # 非流式响应更简单先确保基础功能稳定 } # 3. 发送请求 endpoint chat/completions # 多数提供商使用此端点 if self.provider zhipu: endpoint chat/completions # 智谱等可能略有不同需参考其文档 return self._make_request_with_retry(endpoint, payload)4. 实战处理长上下文与流式响应对于需要处理长文档或希望实现打字机效果的场景需要更高级的技巧。4.1 智能处理长文本输入当输入文本超过模型限制时简单的截断会丢失信息。更优的策略是进行智能分块和摘要。# 在 RobustLLMClient 类中添加长文本处理方法 class RobustLLMClient: # ... 之前的代码 ... def split_text_into_chunks(self, text: str, chunk_size: int 1000, overlap: int 50) - list: 将长文本按 tokens 数分块块与块之间有一定重叠避免语义断裂。 tokens self.encoding.encode(text) chunks [] start 0 while start len(tokens): end start chunk_size chunk_tokens tokens[start:end] chunk_text self.encoding.decode(chunk_tokens) chunks.append(chunk_text) start end - overlap # 重叠一部分保持上下文连贯 return chunks def summarize_long_document(self, long_text: str, model: str) - str: 通过递归摘要的方式处理超长文档 max_chunk_tokens 30000 # 设定一个安全的分块大小 if self.count_tokens(long_text) max_chunk_tokens: # 如果文本不长直接处理 messages [ {role: user, content: f请为以下文本生成一个简洁的摘要\n\n{long_text}} ] response self.chat_completion(messages, modelmodel, max_tokens500) return response[choices][0][message][content] else: # 文本过长先分块再递归摘要 chunks self.split_text_into_chunks(long_text, chunk_sizemax_chunk_tokens) chunk_summaries [] for i, chunk in enumerate(chunks): print(fSummarizing chunk {i1}/{len(chunks)}...) summary self.summarize_long_document(chunk, model) # 递归调用 chunk_summaries.append(summary) # 将所有分块的摘要合并再生成最终摘要 combined_summaries \n.join(chunk_summaries) final_messages [ {role: user, content: f以下是同一文档多个部分的摘要请将它们整合成一个连贯的总体摘要\n\n{combined_summaries}} ] response self.chat_completion(final_messages, modelmodel, max_tokens800) return response[choices][0][message][content]4.2 实现稳定的流式响应流式响应可以提升用户体验但需要更细致的超时和网络错误处理。# 在 RobustLLMClient 类中添加流式响应方法 class RobustLLMClient: # ... 之前的代码 ... def stream_chat_completion(self, messages: list, model: str kimi-v1, max_tokens: int 2000, temperature: float 0.7): 流式响应版本适用于需要实时显示生成内容的场景 import json url f{self.base_url}/chat/completions headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { model: model, messages: messages, max_tokens: max_tokens, temperature: temperature, stream: True # 开启流式 } try: # 设置更长的超时时间因为流式响应可能持续较久 response requests.post(url, headersheaders, jsonpayload, timeout120, streamTrue) response.raise_for_status() full_content for line in response.iter_lines(): if line: line line.decode(utf-8) if line.startswith(data: ): data_str line[6:] # 去掉 data: 前缀 if data_str [DONE]: break try: data json.loads(data_str) delta data[choices][0][delta] if content in delta: content_piece delta[content] full_content content_piece yield content_piece # 逐块 yield 给调用者 except json.JSONDecodeError: print(fFailed to parse JSON: {data_str}) continue print(f\n[Stream completed. Full response length: {len(full_content)}]) except requests.exceptions.Timeout: yield [ERROR] Stream request timed out. except requests.exceptions.ConnectionError: yield [ERROR] Connection lost during streaming. except Exception as e: yield f[ERROR] An error occurred: {str(e)}5. 生产环境的最佳实践与错误排查将代码用于生产环境时需要额外的保障措施。5.1 配置监控与告警API 调用的稳定性和成本需要被监控。以下是一个简单的监控装饰器示例# monitoring.py import time import functools from datetime import datetime def monitor_llm_call(func): 监控 API 调用的装饰器记录耗时、tokens 用量和状态 functools.wraps(func) def wrapper(*args, **kwargs): start_time time.time() start_dt datetime.now() try: result func(*args, **kwargs) end_time time.time() duration end_time - start_time # 记录成功日志生产环境应接入 ELK、Prometheus 等 log_entry { timestamp: start_dt.isoformat(), function: func.__name__, status: success, duration_seconds: round(duration, 2), input_tokens: kwargs.get(estimated_tokens, N/A), # 需要实际获取 output_tokens: len(result[choices][0][message][content]) if result else N/A # 简化估算 } print(f[MONITOR] {log_entry}) return result except Exception as e: end_time time.time() duration end_time - start_time log_entry { timestamp: start_dt.isoformat(), function: func.__name__, status: error, duration_seconds: round(duration, 2), error: str(e) } print(f[MONITOR] {log_entry}) raise e return wrapper # 使用装饰器 monitor_llm_call def safe_chat_completion(client, messages, model, max_tokens): return client.chat_completion(messages, model, max_tokens)5.2 常见问题排查清单当 API 调用出现问题时可以按以下清单快速定位。问题现象优先检查点解决方案所有请求返回400 Bad Request1. API Key 是否正确且未过期2. 请求的 URL 端点是否正确3. 请求体 JSON 格式是否正确1. 核对 .env 文件中的密钥2. 查阅官方文档确认端点3. 使用 jsonlint 验证格式间歇性429 Too Many Requests1. 是否触发了速率限制2. 同一密钥是否在多处使用1. 降低请求频率加入随机延迟2. 为不同服务使用不同密钥流式响应中途断开1. 客户端或服务端超时设置2. 网络稳定性3. 生成内容过长1. 增加 timeout 参数2. 检查网络连接3. 限制 max_tokens402 Insufficient Balance1. 账户余额是否充足2. 套餐额度是否用完1. 登录平台控制台查看余额2. 升级套餐或充值5.3 密钥轮换与容灾策略对于关键业务应考虑多密钥和多个模型供应商的容灾方案。# 简单的多供应商容灾客户端 class MultiProviderLLMClient: def __init__(self, providers: list None): self.providers providers or [kimi, deepseek, zhipu] self.clients [RobustLLMClient(provider) for provider in self.providers] def chat_with_fallback(self, messages, model_mapNone, **kwargs): 按顺序尝试多个供应商直到有一个成功 model_map model_map or { kimi: kimi-v1, deepseek: deepseek-chat, zhipu: zhipu-glm-4 } for i, client in enumerate(self.clients): provider self.providers[i] model model_map.get(provider) try: print(fTrying provider: {provider}) result client.chat_completion(messages, modelmodel, **kwargs) print(fSuccess with provider: {provider}) return result, provider except Exception as e: print(fProvider {provider} failed: {e}) continue raise Exception(All providers failed.) # 使用示例 fallback_client MultiProviderLLMClient() result, successful_provider fallback_client.chat_with_fallback( messages[{role: user, content: 你好请介绍你自己。}], max_tokens500 )通过上述实践我们构建了一个具备错误处理、重试机制、长文本支持和多供应商容灾能力的稳健的 LLM API 客户端。这种设计思路与 Kimi Hosted Agent 等平台的目标一致即在享受大模型能力的同时最大限度地降低集成复杂度和运维风险。在实际项目中还需根据具体的业务需求、流量规模和合规要求进一步设计限流、降级、审计等高级功能。
![[Android] 可视化音乐制作 -短视频超火的音乐视频制作工具](http://pic.xiahunao.cn/yaotu/[Android] 可视化音乐制作 -短视频超火的音乐视频制作工具)
