大模型API工程实践：从OpenAI/Anthropic集成到生产级客户端构建

在实际项目开发中调用大模型 API 已经成为集成 AI 能力的标准路径。无论是构建智能客服、内容生成工具还是进行数据分析增强开发者都需要面对模型选型、API 集成、成本控制和错误处理等一系列工程问题。最近主流模型提供商在定价和功能上的调整直接影响着我们的技术选型和项目预算。理解不同 API 的特性、掌握稳健的集成方法、并建立有效的成本与异常监控机制是当前开发者的必备技能。本文将从工程实践角度出发为你梳理在项目中集成 OpenAI、Anthropic 等主流大模型 API 的完整流程。我们将不局限于简单的“Hello World”调用而是深入探讨如何构建一个健壮、可维护且具备成本意识的客户端涵盖环境准备、SDK 选择、核心调用模式、关键参数解析、全面的错误处理策略以及针对生产环境的优化建议。无论你是初次接触大模型 API还是希望优化现有集成方案都能从中获得可直接落地的代码示例和设计思路。1. 理解大模型 API 的核心概念与工程挑战在编写第一行代码之前我们需要厘清几个关键概念这有助于理解后续的配置和设计决策。1.1 模型即服务与 API 端点大模型提供商如 OpenAI, Anthropic将其训练好的模型封装为可通过 HTTP 请求调用的服务。对于开发者而言我们无需关心模型的训练细节和庞大的基础设施只需通过一个特定的 URL端点发送符合规范的请求即可获得模型的推理结果。这种模式极大地降低了 AI 能力的应用门槛。每个模型都有唯一的标识符例如 OpenAI 的gpt-4o、gpt-4o-mini或 Anthropic 的claude-3-5-sonnet-20241022。模型标识符通常与版本、能力及定价直接关联。1.2 上下文长度与 Token这是大模型 API 中最重要的两个技术概念。上下文长度是指模型单次请求能够处理的最大文本量通常以 Token 为单位。例如一个模型的上下文长度是 128K意味着你的请求系统提示 用户消息和模型的回复加起来其 Token 总数不能超过这个限制。如果超出通常会收到类似400 Bad Request: This model‘s maximum context length is ...的错误。Token是模型处理文本的基本单位。在英文中一个 Token 大约相当于 0.75 个单词或 4 个字符。中文等象形文字通常一个字对应 1-2 个 Token。API 的计费通常基于输入和输出消耗的 Token 总数。因此在设计和优化提示词时控制 Token 消耗是控制成本的关键。1.3 API Key、计费与配额调用 API 需要身份凭证即 API Key。它就像一把钥匙用于认证和计费。每个 API Key 都关联着一个账户和其下的计费计划。计费模式主流采用按使用量付费即按输入/输出的 Token 数计费。不同模型的单价每百万 Token 的价格不同性能更强的模型通常更贵。提供商也可能提供套餐或承诺使用折扣。速率限制为了防止滥用和保障服务稳定性API 提供商会对单个 Key 或账户设置速率限制例如每分钟/每天的最大请求数或 Token 数。超过限制会收到429 Too Many Requests错误。余额与欠费调用前需确保账户有足够余额或已绑定支付方式。余额耗尽会导致调用失败并可能返回402 Insufficient Balance等错误。1.4 流式响应与非流式响应非流式响应客户端发送完整请求后等待服务器处理完整个生成过程一次性返回全部结果。这种方式简单但用户需要等待较长时间才能看到任何内容。流式响应服务器在生成 Token 的过程中就逐步将部分结果返回给客户端。这对于需要实时显示生成内容的应用如聊天界面至关重要能极大提升用户体验。在技术实现上这通常通过 Server-Sent Events (SSE) 或类似技术实现。2. 环境准备与项目初始化我们将使用 Python 作为示例语言因为它拥有最完善的 AI 开发生态。确保你已安装 Python 3.8 或更高版本。2.1 创建虚拟环境与安装依赖首先为项目创建一个独立的虚拟环境避免包依赖冲突。# 创建项目目录并进入 mkdir ai-api-integration cd ai-api-integration # 创建虚拟环境以 venv 为例 python -m venv venv # 激活虚拟环境 # 在 Windows 上: venv\Scripts\activate # 在 macOS/Linux 上: source venv/bin/activate激活虚拟环境后安装必要的 SDK 和工具库。我们将安装 OpenAI 和 Anthropic 的官方 SDK以及用于处理环境变量的python-dotenv。pip install openai anthropic python-dotenv2.2 管理敏感的 API Key绝对不要将 API Key 硬编码在源代码中或提交到版本控制系统如 Git。正确的方法是使用环境变量。在项目根目录创建.env文件。将你的 API Key 填入该文件。# .env 文件内容示例 OPENAI_API_KEYsk-your-openai-api-key-here ANTHROPIC_API_KEYsk-ant-your-anthropic-api-key-here确保.env文件已被添加到.gitignore中防止意外提交。# .gitignore 文件内容示例 venv/ .env *.pyc __pycache__/2.3 项目基础结构创建一个简单的项目结构将配置、工具函数和主逻辑分离。ai-api-integration/ ├── .env # 环境变量保密不上传 ├── .gitignore # Git忽略文件 ├── config.py # 配置加载 ├── clients/ # API 客户端封装 │ ├── __init__.py │ ├── openai_client.py │ └── anthropic_client.py ├── utils/ # 工具函数 │ ├── __init__.py │ └── token_counter.py # 简单的 Token 估算 └── main.py # 主程序入口3. 构建健壮的 API 客户端直接在每个业务函数中初始化 SDK 客户端会导致代码重复且难以管理配置。我们将封装一个统一的客户端类集中处理初始化、错误重试和日志记录。3.1 配置加载模块首先创建config.py安全地加载环境变量。# config.py import os from dotenv import load_dotenv # 加载 .env 文件中的环境变量 load_dotenv() class Config: 应用配置类 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) ANTHROPIC_API_KEY os.getenv(ANTHROPIC_API_KEY) # 基础 API 端点通常使用官方默认仅在需要自定义代理时修改 OPENAI_API_BASE os.getenv(OPENAI_API_BASE, https://api.openai.com/v1) ANTHROPIC_API_BASE os.getenv(ANTHROPIC_API_BASE, https://api.anthropic.com/v1) # 默认模型 DEFAULT_OPENAI_MODEL os.getenv(DEFAULT_OPENAI_MODEL, gpt-4o-mini) DEFAULT_ANTHROPIC_MODEL os.getenv(DEFAULT_ANTHROPIC_MODEL, claude-3-5-sonnet-20241022) # 请求超时设置秒 REQUEST_TIMEOUT int(os.getenv(REQUEST_TIMEOUT, 30)) # 最大重试次数 MAX_RETRIES int(os.getenv(MAX_RETRIES, 3)) classmethod def validate(cls): 验证必要配置是否存在 missing [] if not cls.OPENAI_API_KEY: missing.append(OPENAI_API_KEY) if not cls.ANTHROPIC_API_KEY: missing.append(ANTHROPIC_API_KEY) # 根据实际需求可以只验证其中一个 if missing: raise ValueError(f缺少必要的环境变量: {, .join(missing)}。请在 .env 文件中配置。) # 应用启动时验证配置 Config.validate()3.2 封装 OpenAI 客户端创建clients/openai_client.py封装一个具备重试和基础错误处理能力的客户端。# clients/openai_client.py import logging import time from typing import Optional, Dict, Any from openai import OpenAI, APIError, APITimeoutError, APIConnectionError from config import Config logger logging.getLogger(__name__) class OpenAIClient: 封装 OpenAI SDK 客户端增加重试和统一错误处理 def __init__(self): self.client OpenAI( api_keyConfig.OPENAI_API_KEY, base_urlConfig.OPENAI_API_BASE, timeoutConfig.REQUEST_TIMEOUT, max_retries0 # 我们自定义重试逻辑 ) self.max_retries Config.MAX_RETRIES def create_chat_completion(self, messages: list, model: Optional[str] None, temperature: float 0.7, max_tokens: Optional[int] None, stream: bool False, **kwargs) - Any: 创建聊天补全请求支持自动重试。 Args: messages: 消息列表格式如 [{role: user, content: Hello}] model: 模型名称默认为配置中的默认模型 temperature: 温度参数控制随机性 (0-2) max_tokens: 生成的最大 token 数 stream: 是否使用流式响应 **kwargs: 其他传递给 OpenAI API 的参数 Returns: OpenAI 的响应对象或流式响应迭代器 Raises: Exception: 当重试耗尽后仍失败时抛出 model model or Config.DEFAULT_OPENAI_MODEL last_exception None for attempt in range(self.max_retries 1): # 尝试次数 重试次数 1 try: logger.debug(f调用 OpenAI API (尝试 {attempt 1}/{self.max_retries 1}): model{model}) response self.client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature, max_tokensmax_tokens, streamstream, **kwargs ) return response except (APITimeoutError, APIConnectionError) as e: # 网络超时或连接错误通常值得重试 last_exception e logger.warning(fOpenAI API 网络错误 (尝试 {attempt 1}): {e}) if attempt self.max_retries: wait_time 2 ** attempt # 指数退避 logger.info(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) else: logger.error(fOpenAI API 调用失败已达最大重试次数 {self.max_retries}) raise except APIError as e: # API 错误需要根据状态码判断是否重试 last_exception e error_code getattr(e, status_code, None) error_type getattr(e, type, None) logger.error(fOpenAI API 错误 (尝试 {attempt 1}): code{error_code}, type{error_type}, msg{e.message}) # 400 错误通常是参数错误重试无意义 # 429 是速率限制可以短暂等待后重试 # 401/403 是认证错误重试无用 if error_code 429 and attempt self.max_retries: wait_time 5 * (attempt 1) # 对于限流线性增加等待时间 logger.info(f触发速率限制等待 {wait_time} 秒后重试...) time.sleep(wait_time) continue elif error_code in [400, 401, 403]: # 参数错误或认证错误直接抛出 raise else: # 其他服务器错误 (5xx) 可以重试 if attempt self.max_retries: wait_time 2 ** attempt logger.info(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) else: logger.error(fOpenAI API 调用失败已达最大重试次数) raise except Exception as e: # 其他未知异常 last_exception e logger.exception(f调用 OpenAI API 时发生未知异常 (尝试 {attempt 1})) if attempt self.max_retries: time.sleep(2 ** attempt) else: raise # 理论上不会走到这里因为循环内会 raise raise last_exception or Exception(OpenAI API 调用失败)3.3 封装 Anthropic 客户端类似地创建clients/anthropic_client.py。注意 Anthropic SDK 的调用方式与 OpenAI 略有不同。# clients/anthropic_client.py import logging import time from typing import Optional, Dict, Any import anthropic from anthropic import APIError, APITimeoutError, APIConnectionError from config import Config logger logging.getLogger(__name__) class AnthropicClient: 封装 Anthropic SDK 客户端 def __init__(self): self.client anthropic.Anthropic( api_keyConfig.ANTHROPIC_API_KEY, base_urlConfig.ANTHROPIC_API_BASE, timeoutConfig.REQUEST_TIMEOUT, max_retries0 # 自定义重试 ) self.max_retries Config.MAX_RETRIES def create_message(self, messages: list, model: Optional[str] None, max_tokens: int 1024, temperature: float 0.7, stream: bool False, **kwargs) - Any: 创建消息Anthropic 的聊天补全接口。 Args: messages: 消息列表注意 Anthropic 要求以 user/assistant 角色开始 model: 模型名称 max_tokens: 生成的最大 token 数必需参数 temperature: 温度参数 stream: 是否流式 **kwargs: 其他参数如 thinking_budget (用于 Claude 3.5 Sonnet 思考模式) Returns: Anthropic 的响应对象或流式迭代器 model model or Config.DEFAULT_ANTHROPIC_MODEL last_exception None # Anthropic 对消息格式有特定要求这里做简单适配 # 注意实际生产环境可能需要更复杂的消息历史管理 system_message None anthropic_messages [] for msg in messages: if msg.get(role) system: system_message msg.get(content) else: # 转换 role 到 anthropic 支持的 ‘user‘ 或 ‘assistant‘ role user if msg[role] in [user, developer] else assistant anthropic_messages.append({ role: role, content: msg[content] }) for attempt in range(self.max_retries 1): try: logger.debug(f调用 Anthropic API (尝试 {attempt 1}/{self.max_retries 1}): model{model}) response self.client.messages.create( modelmodel, messagesanthropic_messages, systemsystem_message, max_tokensmax_tokens, temperaturetemperature, streamstream, **kwargs ) return response except (APITimeoutError, APIConnectionError) as e: last_exception e logger.warning(fAnthropic API 网络错误 (尝试 {attempt 1}): {e}) if attempt self.max_retries: wait_time 2 ** attempt logger.info(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) else: logger.error(fAnthropic API 调用失败已达最大重试次数) raise except APIError as e: last_exception e error_code e.status_code error_type e.__class__.__name__ logger.error(fAnthropic API 错误 (尝试 {attempt 1}): code{error_code}, type{error_type}, msg{e.message}) # 处理特定错误例如 thinking_budget 参数错误 if thinking_budget in str(e).lower() and must be a positive integer in str(e).lower(): logger.error(参数 thinking_budget 必须是一个正整数。) raise ValueError(thinking_budget 参数无效) from e if error_code 429 and attempt self.max_retries: wait_time 5 * (attempt 1) logger.info(f触发速率限制等待 {wait_time} 秒后重试...) time.sleep(wait_time) continue elif error_code in [400, 401, 403]: raise else: if attempt self.max_retries: wait_time 2 ** attempt logger.info(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) else: logger.error(fAnthropic API 调用失败已达最大重试次数) raise except Exception as e: last_exception e logger.exception(f调用 Anthropic API 时发生未知异常 (尝试 {attempt 1})) if attempt self.max_retries: time.sleep(2 ** attempt) else: raise raise last_exception or Exception(Anthropic API 调用失败)4. 核心调用模式与参数详解封装好客户端后我们来探讨几种核心的调用模式并解释关键参数。4.1 基础同步调用这是最简单的调用方式适用于不需要实时反馈的后台任务。# main.py 示例片段 import logging from clients.openai_client import OpenAIClient from clients.anthropic_client import AnthropicClient logging.basicConfig(levellogging.INFO) def basic_chat_with_openai(): 使用 OpenAI 进行基础聊天 client OpenAIClient() messages [ {role: system, content: 你是一个乐于助人的助手。}, {role: user, content: 用一句话解释什么是人工智能。} ] try: response client.create_chat_completion( messagesmessages, modelgpt-4o-mini, # 可以覆盖默认模型 temperature0.5, # 降低随机性使输出更确定 max_tokens150 # 限制回复长度 ) # 提取回复内容 reply response.choices[0].message.content print(fOpenAI 回复: {reply}) # 查看使用量 usage response.usage print(fToken 使用: 输入 {usage.prompt_tokens}, 输出 {usage.completion_tokens}, 总计 {usage.total_tokens}) except Exception as e: logging.error(f调用 OpenAI 失败: {e}) def basic_chat_with_anthropic(): 使用 Anthropic 进行基础聊天 client AnthropicClient() messages [ {role: user, content: 用一句话解释什么是机器学习。} ] try: response client.create_message( messagesmessages, modelclaude-3-haiku-20240307, # 使用更快的 Haiku 模型 max_tokens100, # Anthropic 必须指定 max_tokens temperature0.7 ) reply response.content[0].text print(fAnthropic 回复: {reply}) # Anthropic 的 usage 在 response.usage usage response.usage print(fToken 使用: 输入 {usage.input_tokens}, 输出 {usage.output_tokens}) except Exception as e: logging.error(f调用 Anthropic 失败: {e}) if __name__ __main__: basic_chat_with_openai() print(- * 50) basic_chat_with_anthropic()4.2 流式调用流式调用对于构建交互式应用至关重要。下面展示如何处理流式响应。def stream_chat_with_openai(): 使用 OpenAI 流式聊天 client OpenAIClient() messages [{role: user, content: 写一首关于编程的短诗。}] try: stream_response client.create_chat_completion( messagesmessages, streamTrue # 关键参数开启流式 ) print(OpenAI 流式回复: , end, flushTrue) full_reply for chunk in stream_response: # 检查是否有内容增量 if chunk.choices[0].delta.content is not None: content chunk.choices[0].delta.content print(content, end, flushTrue) full_reply content print() # 换行 return full_reply except Exception as e: logging.error(f流式调用 OpenAI 失败: {e}) return None4.3 关键参数解析与调优不同的参数会显著影响模型的输出和行为。理解它们对于生产应用至关重要。参数适用平台含义与影响常用值/范围生产环境建议modelOpenAI, Anthropic指定使用的模型。决定能力、速度、成本和上下文长度。gpt-4o,gpt-4o-mini,claude-3-5-sonnet根据任务复杂度选择简单任务用轻量模型如-mini,-haiku以节省成本复杂任务用高级模型。temperatureOpenAI, Anthropic控制输出的随机性。值越高输出越多样、有创意值越低输出越确定、保守。0.0 ~ 2.0 (OpenAI), 0.0 ~ 1.0 (Anthropic)创意生成如写作可用 0.7~0.9事实问答、代码生成建议用 0.1~0.3。max_tokensOpenAI, Anthropic限制模型生成的最大 Token 数。注意Anthropic 此为必需参数。1 ~ 模型上限务必设置防止意外生成长文本导致高费用或超时。根据历史对话估算。streamOpenAI, Anthropic是否启用流式响应。True/False前端交互场景必须为True后端批量处理可为False。top_p(OpenAI)OpenAI核采样概率。与temperature二选一通常不一起调。0.0 ~ 1.0更精确地控制输出分布。常用 0.9~0.95。thinking_budget(Anthropic)Anthropic (Claude 3.5)为模型的“思考”过程分配 Token 预算。仅特定模型支持。正整数用于需要深度推理的任务。需结合max_tokens考虑总输出 Token 思考 Token 回复 Token。frequency_penalty,presence_penaltyOpenAI频率惩罚和存在惩罚用于减少重复用词。-2.0 ~ 2.0在需要避免重复的场景如文章生成中微调通常从 0.1 开始。5. 生产环境中的错误处理与监控在开发环境能跑通只是第一步生产环境必须考虑各种异常情况。5.1 常见错误码与处理策略下表列出了集成大模型 API 时最常见的错误及其处理建议。错误现象/状态码可能原因检查与处理策略400 Bad Request请求参数错误。如max_tokens超限、thinking_budget非正整数、消息格式错误、上下文长度超限。1. 检查错误信息中的具体描述。2. 验证所有参数值是否符合 API 文档要求。3. 使用工具估算输入 Token 是否超过模型限制。401 UnauthorizedAPI Key 无效、过期或未提供。1. 检查.env文件中的 KEY 是否正确。2. 检查 KEY 是否在代码中被正确加载。3. 在提供商控制台验证 KEY 状态。403 Forbidden权限不足。可能模型未授权、端点不可访问、或触发了安全策略。1. 确认账户是否有权限访问该模型。2. 检查 API 端点 URL 是否正确。3. 查看提供商控制台是否有风控通知。429 Too Many Requests触发速率限制或配额限制。1. 实现带退避机制的重试逻辑如我们封装的客户端。2. 监控调用频率考虑升级套餐或申请提高限额。3. 对于批量任务加入延迟。500,502,503,504服务器内部错误、网关错误或服务暂时不可用。1. 这些错误通常可以重试。2. 实现指数退避重试。3. 检查提供商状态页面。APIConnectionError/ 网络超时网络连接问题、客户端超时设置过短、或服务器响应慢。1. 适当增加timeout参数。2. 检查本地网络和代理设置。3. 实现重试机制。响应内容不完整或中断网络波动、流式响应中断、服务器端问题。1. 对于流式响应实现断线重连和续接逻辑。2. 记录已接收的部分并提示用户或进行补偿。402 Insufficient Balance账户余额不足。1. 调用前检查余额如果 API 支持。2. 设置消费告警。3. 确保支付方式有效。5.2 实现监控与告警对于生产系统需要监控 API 调用的健康度、延迟和成本。日志记录在封装的客户端中我们已经记录了关键操作的日志。应将这些日志接入 ELK、Sentry 或类似系统。指标收集在每次 API 调用后收集以下指标耗时Latency状态码Status Code输入/输出 Token 数模型名称可以推送至 Prometheus、StatsD 或直接由云服务商监控。成本告警定期如每小时查询账户余额或使用量当达到预算的某个百分比如 80%时触发告警邮件、钉钉、Slack。降级策略当主要模型服务不可用或成本超支时应有备用方案。例如可以准备一个开关在故障时切换到另一个提供商的模型或者 fallback 到规则引擎。# 一个简单的成本监控函数示例 import requests from datetime import datetime, timedelta def check_openai_usage(api_key: str, days: int 1): 查询 OpenAI 近期使用量示例实际需参考最新 API headers { Authorization: fBearer {api_key}, } end_date datetime.utcnow().date() start_date end_date - timedelta(daysdays) # 注意OpenAI 用量查询 API 可能有变化此处为示例 url fhttps://api.openai.com/v1/usage?start_date{start_date}end_date{end_date} try: response requests.get(url, headersheaders) response.raise_for_status() usage_data response.json() total_cost usage_data.get(total_usage, 0) / 100 # 假设单位为美分 print(f最近{days}天总费用: ${total_cost:.2f}) return total_cost except Exception as e: logging.error(f查询使用量失败: {e}) return None6. 最佳实践与扩展方向6.1 生产环境检查清单在将大模型 API 集成部署到生产环境前请对照此清单进行检查[ ]配置安全API Key 已通过环境变量或密钥管理服务管理未硬编码。[ ]错误处理已实现针对网络错误、速率限制、服务器错误的重试机制带退避。[ ]超时设置设置了合理的请求超时如 30-60 秒避免线程阻塞。[ ]限流保护在客户端或网关层实施了调用频率限制防止意外超限。[ ]输入验证与清理对用户输入进行基本的清理和长度检查防止恶意输入或过长的提示词消耗大量 Token。[ ]Token 估算与限制实现了简单的 Token 估算例如使用tiktoken库 for OpenAI并在请求前拒绝明显超长的输入。[ ]日志与监控关键操作请求、响应、错误都有日志并接入了监控告警系统。[ ]成本监控设置了基于使用量或费用的告警阈值。[ ]降级方案定义了当主要 API 不可用时的降级策略如切换模型、返回缓存、展示友好错误。[ ]版本管理在配置中指定了具体的模型版本号如gpt-4o-2024-08-06避免自动升级导致的不兼容。6.2 性能与成本优化建议缓存策略对于内容生成类且结果可复用的请求如商品描述生成、标准问答可以将(模型, 提示词, 参数)作为键将结果缓存一段时间如 Redis直接返回缓存结果。批处理请求如果业务允许将多个独立的生成任务合并为一个批处理请求如果 API 支持可以减少网络开销。模型选型并非所有任务都需要最强大的模型。将任务分类对简单的分类、摘要、格式化任务使用轻量级模型如gpt-4o-mini,claude-3-haiku可以大幅降低成本。优化提示词精心设计的提示词Prompt Engineering可以用更少的 Token 获得更准确的结果。避免在系统提示中放入冗长且不变的背景信息考虑将其外部化。异步调用对于不要求实时响应的后台任务使用异步客户端如openai.AsyncOpenAI可以提升吞吐量。6.3 扩展方向构建统一的 AI 网关当项目中使用多个 AI 提供商时维护多个客户端会变得复杂。一个自然的演进方向是构建一个统一的 AI 网关或抽象层。这个网关可以提供统一接口对外暴露一致的completion、embedding接口内部路由到不同的提供商。故障转移当首选提供商失败时自动切换到备用提供商。负载均衡与成本优化根据模型能力、当前延迟和成本智能分配请求。统一监控与审计集中收集所有 AI 调用的日志、指标和成本数据。# 一个极简网关的示意 class AIGateway: def __init__(self): self.clients { openai: OpenAIClient(), anthropic: AnthropicClient(), # 可以接入更多如 DeepSeek, 智谱AI等 } self.default_provider openai def chat_completion(self, messages, providerNone, **kwargs): provider provider or self.default_provider client self.clients.get(provider) if not client: raise ValueError(f不支持的提供商: {provider}) # 可以在这里加入路由逻辑、熔断器、监控等 try: if provider openai: return client.create_chat_completion(messages, **kwargs) elif provider anthropic: # 可能需要适配参数 return client.create_message(messages, **kwargs) except Exception as e: logging.error(fProvider {provider} 调用失败: {e}) # 实现故障转移逻辑 # return self._fallback(messages, **kwargs) raise通过这样的封装业务代码将与具体的 AI 提供商解耦为未来的架构演进打下坚实基础。