1. 引言ai-gateway 是一个面向 Python 开发者的 AI 网关库用于统一管理多个大语言模型LLM提供商的 API 调用。它通过统一的接口封装 OpenAI、Anthropic、Google Gemini 等主流模型服务帮助开发者屏蔽底层差异实现模型路由、负载均衡、请求重试和成本统计等功能。本文将从功能特性、安装方式、核心语法与参数、9 个实际应用案例以及常见错误与注意事项五个方面系统介绍 ai-gateway 包的使用方法。2. 核心功能ai-gateway 包主要提供以下能力统一接口使用同一套 API 调用不同厂商的模型无需为每个提供商单独编写适配代码。模型路由根据预设规则自动选择模型支持按成本、延迟、可用性等维度进行路由。负载均衡在多个同质模型或 API Key 之间分发请求避免单点过载。自动重试对限流、超时、临时性错误自动重试支持指数退避策略。流式响应原生支持流式输出适合聊天机器人等实时交互场景。成本与用量统计记录每次调用的 Token 消耗和费用便于预算控制。插件机制支持自定义中间件方便接入日志、监控、缓存等扩展能力。3. 安装ai-gateway 可以通过 pip 直接安装推荐在虚拟环境中使用pip install ai-gateway如果需要使用流式响应或特定提供商扩展可以安装附加依赖pip install ai-gateway[streaming] pip install ai-gateway[openai,anthropic]安装完成后可以通过以下命令验证是否成功import ai_gateway print(ai_gateway.__version__)4. 基本语法与参数4.1 初始化网关使用 ai-gateway 的第一步是创建网关实例并配置提供商from ai_gateway import Gateway gateway Gateway( providers{ openai: {api_key: sk-xxx, model: gpt-4o}, anthropic: {api_key: sk-ant-xxx, model: claude-3-5-sonnet}, }, default_provideropenai, retry_times3, timeout30, )主要初始化参数说明参数类型说明providersdict提供商配置字典键为提供商名称值为 API Key、模型等配置。default_providerstr默认使用的提供商名称。retry_timesint失败自动重试次数默认为 0。timeoutint请求超时时间秒默认为 30。routerstr路由策略可选 cost、latency、random 等。4.2 发起对话请求完成初始化后可以通过 chat 方法发起对话response gateway.chat( messages[ {role: system, content: 你是一个乐于助人的助手。}, {role: user, content: 请用一句话介绍 Python。}, ], provideropenai, temperature0.7, max_tokens500, ) print(response.content)chat 方法常用参数参数类型说明messageslist对话消息列表包含 role 和 content 字段。providerstr指定使用的提供商不传则使用默认值。temperaturefloat采样温度控制输出的随机性范围 0 到 2。max_tokensint生成的最大 Token 数。streambool是否启用流式输出默认为 False。4.3 流式输出流式输出适合实时交互场景使用方式如下for chunk in gateway.chat( messages[{role: user, content: 讲一个笑话}], streamTrue, ): print(chunk.delta, end, flushTrue)5. 实际应用案例5.1 案例一多提供商自动切换当某个模型服务不可用时自动切换到备用提供商gateway Gateway( providers{ openai: {api_key: sk-xxx, model: gpt-4o}, anthropic: {api_key: sk-ant-xxx, model: claude-3-5-sonnet}, }, default_provideropenai, retry_times2, ) try: resp gateway.chat(messages[{role: user, content: 你好}]) except Exception: resp gateway.chat( messages[{role: user, content: 你好}], provideranthropic, ) print(resp.content)5.2 案例二基于成本的模型路由根据任务复杂度自动选择低成本或高能力模型gateway Gateway( providers{ openai-gpt4o: {api_key: sk-xxx, model: gpt-4o}, openai-gpt35: {api_key: sk-xxx, model: gpt-3.5-turbo}, }, routercost, ) simple_resp gateway.chat(messages[{role: user, content: 11?}]) complex_resp gateway.chat(messages[{role: user, content: 解释量子纠缠}]) print(simple_resp.content) print(complex_resp.content)5.3 案例三流式聊天机器人构建一个逐字输出的聊天机器人def chat_bot(user_input): gateway Gateway( providers{openai: {api_key: sk-xxx, model: gpt-4o}}, ) full_response for chunk in gateway.chat( messages[{role: user, content: user_input}], streamTrue, ): full_response chunk.delta print(chunk.delta, end, flushTrue) return full_response chat_bot(请介绍杭州)5.4 案例四批量文本分类对多条文本进行情感分类并统计 Token 消耗texts [这个产品很好用, 服务太差了, 中规中矩吧] gateway Gateway( providers{openai: {api_key: sk-xxx, model: gpt-3.5-turbo}}, ) for text in texts: resp gateway.chat( messages[ {role: system, content: 你是一个情感分类器只输出正面或负面。}, {role: user, content: text}, ], max_tokens10, ) print(f{text} - {resp.content}) print(fToken 消耗: {resp.usage.total_tokens})5.5 案例五带重试的稳定调用配置自动重试应对临时性限流错误gateway Gateway( providers{openai: {api_key: sk-xxx, model: gpt-4o}}, retry_times5, timeout60, ) resp gateway.chat( messages[{role: user, content: 生成一份周报}], temperature0.3, ) print(resp.content)5.6 案例六多 Key 负载均衡在多个 API Key 之间分发请求避免单个 Key 被限流gateway Gateway( providers{ openai: { api_keys: [sk-key1, sk-key2, sk-key3], model: gpt-4o, } }, ) for i in range(6): resp gateway.chat(messages[{role: user, content: f第{i1}次请求}]) print(resp.content)5.7 案例七自定义中间件记录日志通过插件机制记录每次请求的耗时和结果from ai_gateway import Gateway, Middleware class LogMiddleware(Middleware): def before_request(self, request): request.start_time time.time() print(f请求开始: {request.messages[-1][content][:20]}) def after_response(self, request, response): cost time.time() - request.start_time print(f请求完成: 耗时 {cost:.2f}s, Token {response.usage.total_tokens}) gateway Gateway( providers{openai: {api_key: sk-xxx, model: gpt-4o}}, middlewares[LogMiddleware()], ) resp gateway.chat(messages[{role: user, content: 测试日志}]) print(resp.content)5.8 案例八结构化输出解析让模型返回 JSON 格式并自动解析为 Python 对象import json from ai_gateway import Gateway gateway Gateway( providers{openai: {api_key: sk-xxx, model: gpt-4o}}, ) resp gateway.chat( messages[ {role: system, content: 只输出 JSON 格式。}, {role: user, content: 提取这句话中的人名和地点张三去了北京。}, ], response_format{type: json_object}, ) data json.loads(resp.content) print(data)5.9 案例九异步并发调用使用异步接口并发处理多个独立请求提升吞吐量import asyncio from ai_gateway import AsyncGateway async def main(): gateway AsyncGateway( providers{openai: {api_key: sk-xxx, model: gpt-3.5-turbo}}, ) tasks [ gateway.chat(messages[{role: user, content: f问题{i}}]) for i in range(5) ] results await asyncio.gather(*tasks) for r in results: print(r.content) asyncio.run(main())6. 常见错误与注意事项6.1 常见错误错误类型可能原因解决方案AuthenticationErrorAPI Key 无效或过期检查 Key 是否正确重新生成。RateLimitError请求频率超过限制增加重试次数或使用多 Key 负载均衡。TimeoutError请求超时增大 timeout 参数或检查网络。ModelNotFoundError模型名称拼写错误核对提供商支持的模型列表。InvalidRequestErrormessages 格式不正确确保每条消息包含 role 和 content 字段。6.2 使用注意事项API Key 安全不要把 Key 硬编码在代码中建议使用环境变量或密钥管理服务。成本控制为 max_tokens 设置合理上限避免意外产生高额费用。错误处理生产环境务必捕获异常并设置降级策略不要只依赖自动重试。流式响应使用流式输出时注意处理连接中断和部分输出的情况。版本兼容不同版本的 ai-gateway 参数可能有差异升级前阅读变更日志。数据隐私发送给模型的内容可能被服务商记录敏感数据需脱敏处理。7. 总结ai-gateway 通过统一抽象层简化了多模型接入的复杂度其路由、重试、流式和统计能力能够显著提升开发效率。建议从简单的单提供商调用开始逐步引入路由和中间件机制并结合实际业务场景设计合理的降级与监控方案。《AI提示工程必知必会》为读者提供了丰富的AI提示工程知识与实战技能主要包括各类提示词的应用如问答式、指令式、状态类、建议式、安全类和感谢类提示词以及如何通过实战演练掌握提示词的使用技巧使用提示词进行文本摘要、改写重述、语法纠错、机器翻译等语言处理任务以及在数据挖掘、程序开发等领域的应用AI在绘画创作上的应用百度文心一言和阿里通义大模型这两大智能平台的特性与功能以及市场调研中提示词的实战应用。通过阅读《AI提示工程必知必会》读者可掌握如何有效利用AI提示工程提升工作效率创新工作流程并在职场中脱颖而出。