如果你已经写过 AI 助手相关的小项目大概率会有一种体感功能第一次跑通的那几分钟特别爽后面维护起来却越来越别扭。换一个模型要翻代码加一个系统提示词要改函数签名想记录一下每次请求的耗时和 token 消耗又得把业务逻辑和 HTTP 调用搅在一起。这些问题不是“能跑”阶段会暴露的而是当你真正想把它做成一个能长期使用的工具时才会集中爆发。今天这篇开发教程主题就是把“能跑的 AI 助手代码”升级成“能维护的 AI 助手架构”。我们会围绕双龙虾接口模块的设计思路把 AI 助手与模型服务之间的接口层重新梳理一遍并以枫云AI 作为接入示例完成一个最小可运行的 CLI 助手。读完你可以掌握一套适合中小型项目的多 Provider 接入方案以后切换模型、新增厂商、加日志和做上下文管理都只需要在接口模块内部处理不需要把主流程拆得七零八落。先说明一点这里的“双龙虾接口模块”并不是某个开源框架的官方组件而是本期教程给接口层模块起的代号。真正值得学习的不是名字本身而是它背后的一层设计思想——把 AI 服务调用从散落的业务代码中收拢为一个独立模块再通过配置驱动不同的 AI 服务商。接下来我会从问题、设计、代码到排错完整走一遍。1. 这篇文章真正要解决的问题很多开发者接入大模型 API 时代码路径几乎一模一样先把 API Key 复制到代码里然后写一个函数把用户输入转发给模型接口拿到返回结果直接打印。这种做法在最早期没有问题因为它能最快验证“这个模型能不能回答我的问题”。但一旦进入真实使用阶段问题就会一个一个冒出来。第一类是密钥管理问题。API Key 写在代码常量里团队协作时很容易被提交到 Git 仓库造成不必要的泄露风险。更麻烦的是当你有多个 AI 服务商每个都有各自的 Key散落在多个文件里几乎没法统一管理。第二类是服务商切换问题。不同厂商的接口风格虽然越来越接近 OpenAI 格式但鉴权方式、请求体字段、错误返回、限流策略都不完全相同。如果业务代码里直接写了requests.post(https://xxx/v1/chat/completions)那么每换一家服务商甚至每换一个模型版本都要去业务代码里改一遍 URL 和参数。第三类是消息结构问题。多轮对话需要维护历史消息如果没有统一的 Message 结构可能会出现“user 和 assistant 消息拼接在不同类型对象里”的状态处理上下文截断时非常痛苦。第四类是测试问题。当 AI 接口调用和业务逻辑耦合在一起时你没办法用本地假数据测试自己的助手逻辑。每次调试都要真实调用远程接口既慢又消耗额度。所以这篇文章要解决的核心问题不是“怎么调用一个 AI 接口”而是“当你的 AI 助手开始长大时怎么让接口层不拖后腿”。2. 双龙虾接口模块是什么为什么接口要模块化2.1 从“函数调用”到“接口模块”在没有模块化之前一个 AI 助手的核心代码通常长这样一个get_response(user_input)函数内部生成一段 JSON里面带有 model、messages、temperature 等字段然后发起 HTTP 请求解析返回结果取出文本。这个函数可以工作但它的职责太多了既要知道怎么发 HTTP又要知道怎么处理该厂商的返回格式还要负责拼装多轮对话历史。接口模块要做的事情是把这些职责拆分出来。双龙虾接口模块可以理解为一个适配层它对外提供统一的chat(messages)方法对内隐藏不同 AI 服务商的接口细节。业务层只需要关心“我发了一条消息助手返回了一段文本”不需要关心请求是发给了谁、用了什么路径、带了多少历史消息。为了做到这一点接口模块至少需要包含几样东西统一的消息数据结构、统一的 Provider 抽象、负责创建 Provider 的工厂逻辑以及具体的厂商实现。这套结构在小型项目中看起来是有点“重”但它带来的收益会随着项目复杂度增加而迅速放大。2.2 没有接口模块时会遇到什么假设你一开始只接入了一家 AI 服务商业务代码里写了这样一段逻辑请求成功就把resp.json()[choices][0][message][content]取出来请求失败就抛异常。这个逻辑看起来没毛病直到你决定接入第二家服务商或者同一场景下需要用不同模型。如果第二家服务商的响应结构不是choices[0].message.content而是output.text你就要在业务代码里加一个 if 判断。再加上不同的认证方式一家需要Authorization: Bearer另一家需要自定义头你的代码会逐渐变成一份“厂商分支大全”。这种代码没有架构上的“错误”但它已经不具备可维护性。接口模块化之后这些分支全部收敛到 Provider 适配器里。新增一家服务商不是去改业务逻辑而是新增一个 Provider 文件并把它注册到工厂里。业务层代码一行都不用改。2.3 接口模块带来的三个核心收益第一个收益是解耦。业务层只依赖抽象接口不依赖具体厂商。这样模型从 A 家切换到 B 家不会影响用户消息处理、上下文管理、日志记录这些核心逻辑。第二个收益是低切换成本。即使枫云AI 后续调整了网关路径或者你决定换一个模型服务改动范围也被限制在具体的 Provider 类内不会扩散到整个项目。第三个收益是可测试性。有了抽象接口你可以写一个 MockProvider在本地运行所有业务逻辑测试。这样既不消耗线上额度又能稳定触发各种边界情况。3. 枫云AI 的接入设计与接口约定在开始写代码前我们先把接入目标定下来。本期教程选择枫云AI 作为示例 AI 服务。这里不会展开它的后台注册、充值、密钥获取流程因为这类信息在不同时期变化较快而且每个同学拿到的服务配置可能不同。更值得关注的是当我们要接入一个具体的 AI 服务商时接口模块应该怎么设计才不会把代码锁死。现在市面上的大模型服务商越来越多地采用与 OpenAI Chat Completions 风格兼容的 HTTP 接口。核心约定通常是这样的请求方法为 POST路径形如/chat/completions请求头中携带认证信息常见格式为Authorization: Bearer api_key请求体里包含model和messages字段messages是一个数组每项包含role和content响应体里包含choices数组其中第一项的message.content就是助手回复文本如果你的目标服务商完全兼容这套协议那适配层的工作量会非常小。如果服务商的格式有差异也不需要恐慌我们只要在对应的 Provider 类里做字段映射把外部返回格式转换为统一结构即可。接口模块需要屏蔽的差异主要有四类差异类型举例处理方式认证方式自定义请求头、Token 参数在 Provider 中封装 header 生成逻辑接口路径/v1/chat/completions或自定义路径在配置中指定 base_url 和路径请求体字段名model、messages命名差异在 Provider 内做转换响应结构choices[0].message.content或output.text在 Provider 内解析并统一返回4. 环境准备与项目结构4.1 环境要求做这个示例项目不需要很重的框架。推荐使用 Python 3.10 及以上版本依赖只需要一个requests库。如果你已经安装了 Python 和 pip再创建虚拟环境即可。具体版本号以你本机环境为准本文重点演示的是接口模块的通用设计思路。4.2 项目目录结构为了让代码清晰我建议按下面的结构组织ai_assistant/ ├── config/ │ └── settings.json ├── core/ │ ├── __init__.py │ ├── message.py │ └── factory.py ├── providers/ │ ├── __init__.py │ ├── base.py │ └── fengyun.py ├── app.py ├── requirements.txt └── README.md简单说明一下各部分职责config/settings.jsonAI 服务商的配置包括接口地址、密钥、模型名。core/message.py统一的消息数据结构。core/factory.pyProvider 工厂根据配置创建具体的服务商实例。providers/base.pyProvider 抽象基类定义统一的调用接口。providers/fengyun.py枫云AI 的 Provider 实现。app.py命令行入口演示完整的多轮对话流程。4.3 初始化项目先创建项目目录和虚拟环境然后安装依赖。mkdir ai_assistant cd ai_assistant python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate pip install requests如果你希望项目更严格可以把依赖写入 requirements.txtrequests2.31.05. 核心代码实现这一部分是整个教程的核心。我们从配置文件开始一步步实现一个完整的多 Provider AI 助手接口模块。5.1 配置文件config/settings.json{ provider: fengyun, fengyun: { base_url: https://api.example.com/v1, api_key: your-fengyun-api-key, model: fengyun-chat, timeout: 60 } }这里的api.example.com是占位地址你需要替换成自己在枫云AI 后台获得的真实网关地址。api_key和model同理都以你在服务商后台实际申请到的配置为准。我们使用 JSON 作为配置格式是因为它的层级结构直观适合表达不同厂商的独立配置块。一个小提醒这个文件不要直接提交到公共 Git 仓库。更安全的做法是把它加入.gitignore或者只在本地保留settings.local.json仓库中放一个不带密钥的settings.example.json。5.2 统一消息结构core/message.py在开始实现 Provider 之前先把消息结构定义好。多轮对话中我们至少需要区分三种角色系统提示词system、用户user、助手assistant。# core/message.py from dataclasses import dataclass from typing import List dataclass class ChatMessage: role: str content: str def to_openai_messages(messages: List[ChatMessage]) - List[dict]: 转换为 OpenAI 风格的消息数组 return [{role: m.role, content: m.content} for m in messages]使用dataclass定义消息结构可以减少样板代码。to_openai_messages是一个纯函数它负责把内部消息对象转换为发送给服务商的格式。如果后续枫云AI 的请求体格式有差异只需要在这里或者 Provider 内部做一次转换业务逻辑不需要感知。5.3 Provider 抽象基类providers/base.py接口模块的核心是抽象。我们定义一个BaseProvider声明所有 Provider 必须实现的方法。# providers/base.py from abc import ABC, abstractmethod from typing import List from core.message import ChatMessage class BaseProvider(ABC): 所有 AI 服务商的统一接口 abstractmethod def chat(self, messages: List[ChatMessage]) - str: 接收完整对话历史返回模型生成的文本。 注意这里的 messages 是完整的历史消息列表 是否截断由调用方负责Provider 只负责转发。 pass abstractmethod def get_model_name(self) - str: 返回当前使用的模型名称便于日志记录 pass这个抽象类看起来很薄但它是整个模块化设计的关键。有了它业务层就可以只依赖BaseProvider而不是具体某个服务商。5.4 枫云AI Provider 实现providers/fengyun.py现在我们来实现具体的枫云AI Provider。假设枫云AI 的服务网关提供与 OpenAI Chat Completions 风格兼容的接口那么代码可以这样写。# providers/fengyun.py import logging from typing import List import requests from core.message import ChatMessage, to_openai_messages from providers.base import BaseProvider logger logging.getLogger(__name__) class FengyunProvider(BaseProvider): def __init__(self, config: dict): self.base_url config[base_url].rstrip(/) self.api_key config[api_key] self.model config[model] self.timeout config.get(timeout, 60) def _build_headers(self) - dict: return { Authorization: fBearer {self.api_key}, Content-Type: application/json, } def chat(self, messages: List[ChatMessage]) - str: url f{self.base_url}/chat/completions payload { model: self.model, messages: to_openai_messages(messages), } logger.info(请求 model%s历史消息数%d, self.model, len(messages)) resp requests.post( url, headersself._build_headers(), jsonpayload, timeoutself.timeout, ) resp.raise_for_status() data resp.json() try: return data[choices][0][message][content] except (KeyError, IndexError) as e: raise RuntimeError(f解析响应失败: {data}) from e def get_model_name(self) - str: return self.model这段代码做了几件事在__init__中从配置字典读取接口地址、密钥和模型名。_build_headers专门负责认证头后续如果要换认证方式只需改这个方法。chat方法负责组装请求体、发起请求、解析响应。使用resp.raise_for_status()让 HTTP 错误在调用处被统一捕获更符合 Python 开发习惯。解析响应时做了异常处理避免因为响应结构变化导致难以排查的 KeyError。如果你的枫云AI 网关不是 OpenAI 兼容格式只需要修改chat方法里的 URL 拼接、请求体结构和响应解析逻辑其他代码完全不用动。5.5 Provider 工厂core/factory.py有了具体 Provider还需要一个工厂来动态创建实例。这样配置里写fengyun就能自动创建FengyunProvider。# core/factory.py from providers.base import BaseProvider from providers.fengyun import FengyunProvider def create_provider(provider_name: str, config: dict) - BaseProvider: if provider_name fengyun: return FengyunProvider(config) raise ValueError(f不支持的 AI Provider: {provider_name})这个工厂目前只支持fengyun。后续新增其他服务商时只需要在函数里增加一个分支或者改用注册表机制。对小型项目来说if 分支已经足够清晰。5.6 命令行入口app.py最后是主程序。这里不做复杂业务只实现一个能跑多轮对话的 CLI。# app.py import json import logging from typing import List from core.factory import create_provider from core.message import ChatMessage logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, ) logger logging.getLogger(app) def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return json.load(f) def main(): config load_config(config/settings.json) provider_name config.get(provider, fengyun) provider_config config[provider_name] provider create_provider(provider_name, provider_config) print(AI 助手已启动输入 exit 或 quit 退出) history: List[ChatMessage] [] while True: user_input input(你: ).strip() if user_input.lower() in (exit, quit): break history.append(ChatMessage(roleuser, contentuser_input)) try: reply provider.chat(history) history.append(ChatMessage(roleassistant, contentreply)) print(fAI: {reply}) except Exception as e: logger.exception(请求接口失败) print(f请求出错: {e}) history.pop() if __name__ __main__: main()注意history.pop()这一段。如果某次请求失败用户消息已经加入了历史列表但助手没有回复这时应该把这条用户消息从历史里去掉避免后续消息带上一次失败请求影响模型对上下文的判断。这是一个很容易被忽略的细节。6. 完整示例与运行验证完成上述代码后就可以运行项目了。先确保你已经在config/settings.json中填入了正确的网关地址、API Key 和模型名。6.1 启动命令在项目根目录执行python app.py如果一切正常你会看到提示AI 助手已启动输入 exit 或 quit 退出 你:然后输入第一句话。比如你: 你好请用一句话介绍你自己模型返回后程序会打印AI: 你好我是基于大模型构建的智能助手可以帮你回答问题、梳理思路和编写代码。这里要注意实际输出内容完全取决于你在枫云AI 配置的模型效果上面只是示意。关键在于你能看到请求成功返回并且下一次提问时会携带上一轮的历史消息。6.2 如何判断运行成功几个简单的判断标准日志中能看到请求 modelfengyun-chat历史消息数1之类的信息说明请求已经发出。打印出的 AI 回复内容与输入问题语义相关说明模型调用链路完整。连续提问两轮后模型能感知到上下文说明历史消息维护逻辑正确。6.3 如果运行失败先看哪里第一次运行最常见的失败点是配置问题。建议按以下顺序排查看config/settings.json中的base_url是否填写正确末尾不要有多余斜杠。看api_key是否被正确读取不要在字符串前后留空格。看model名称是否在服务商的支持列表中。再看控制台输出的异常信息确认是网络错误、认证错误还是响应解析错误。7. 常见问题与排查方法问题现象可能原因排查方式解决方案请求返回 401API Key 错误或已过期检查配置文件和日志中的鉴权头在服务商后台重新生成 Key请求返回 404接口地址或路径不正确查看请求 URL 与服务商文档对比修正base_url或路径拼接逻辑请求返回 400 invalid model模型名称不存在检查日志中 payload 的model字段换成服务商实际支持的模型名请求超时网络连通性异常或模型推理过慢先用 curl 测试网关连通性调整timeout参数或检查网络响应解析报 KeyError服务商响应格式与预期不符打印原始响应 JSON 查看结构调整响应解析逻辑做兼容处理多轮对话语义不连贯历史消息被截断或未正确维护打印history长度和内容增加上下文长度控制只保留最近 N 轮控制台没有日志日志级别配置不正确检查logging.basicConfig配置把级别调成 INFO 或 DEBUG这里最值得提醒的是响应解析问题。不同服务商即使都宣称兼容 OpenAI 格式也可能在流式模式、错误响应、usage 字段上存在细微差别。遇到这类错误时最好的办法不是猜而是在resp.json()之后先打印原始数据确认结构后再写解析逻辑。8. 最佳实践与工程建议代码跑通只是第一步真正让它变成一个可维护项目还需要补充一些工程细节。8.1 密钥管理配置与代码分离推荐的做法是把 API Key 放到环境变量中配置读取时做优先级判断环境变量优先配置文件兜底。这样即使settings.json被误提交也不会直接泄露真实 Key。import os def get_api_key(config: dict) - str: return os.getenv(FENGYUN_API_KEY, config.get(api_key, ))在实际项目中还可以引入 python-dotenv 来加载.env文件。但核心原则是仓库里只保留示例配置真实密钥信息永远放在本地环境或密钥管理服务中。8.2 日志与可观测性接入 AI 服务后日志里至少应该记录三类信息请求的模型名、请求的消息条数、请求耗时。这个日志用于日常排查很有效但不建议把完整对话内容直接写入日志因为用户输入可能包含隐私信息。如果确实需要记录建议先做脱敏处理。8.3 错误处理与重试策略在 Provider 的chat方法里raise_for_status()会触发 HTTP 错误。但并不是所有错误都应该重试。比如 401 认证失败重试多少次都不会成功而 429 限流或 5xx 服务器错误则可以通过短暂等待后重试解决。一个简单的重试逻辑如下import time def chat_with_retry(provider, messages, retries3, delay1.5): for attempt in range(retries): try: return provider.chat(messages) except requests.exceptions.HTTPError as e: if e.response.status_code in (401, 400, 404): raise if attempt retries - 1: raise time.sleep(delay * (attempt 1))这里区分了“不可重试”和“可重试”的错误类型。如果对稳定性要求更高还可以考虑指数退避和抖动但小项目中简单的等退已经足够。8.4 多轮上下文的长度控制大模型对上下文长度有限制所以历史消息不能无限堆积。一个简单的策略是限制最大轮数超过后把最旧的消息丢弃。更精细的做法是按 token 估算但需要额外引入 tokenizer对初学者来说可以先从轮数控制开始。MAX_HISTORY_ROUNDS 10 def append_user_message(history, message): history.append(message) if len(history) MAX_HISTORY_ROUNDS * 2: del history[: len(history) - MAX_HISTORY_ROUNDS * 2]这里MAX_HISTORY_ROUNDS * 2是因为每一轮会新增 user 和 assistant 两条消息。系统提示词可以单独存放在history[0]截断时要留意不要把它删掉。8.5 安全边界在 AI 助手中用户输入最终会传给模型因此输入内容最好先做长度限制和基本校验。另外如果 AI 助手会访问本地文件或执行命令那一部分需要非常谨慎的权限设计。本期教程只完成对话接口不涉及本地操作但你要始终记住模型生成的输出不可完全信任涉及敏感操作时必须有确认环节。8.6 可测试性用 Mock Provider 替换真实调用有了BaseProvider抽象你可以很容易写一个 MockProvider 用于本地测试业务逻辑。# tests/mock_provider.py from typing import List from core.message import ChatMessage from providers.base import BaseProvider class MockProvider(BaseProvider): def chat(self, messages: List[ChatMessage]) - str: return fmock reply for {len(messages)} messages def get_model_name(self) - str: return mock-model这样在写单元测试时就不需要真实调用远程接口也不依赖网络环境和 API 额度。9. 总结与后续学习方向这一期通过双龙虾接口模块的设计把一个 AI 助手中最容易被忽略的接口层重新做了梳理。核心收获可以总结为三点统一消息结构解决了多轮上下文的数据格式问题Provider 抽象解决了多服务商切换的耦合问题配置驱动解决了密钥和模型名散落各处的问题。代码本身并不复杂但每个设计决策都对应真实项目中会遇到的坑。如果你正在做自己的 AI 助手下一步可以继续尝试给 Provider 增加流式输出让回复逐字显示接入多个服务商实现按场景路由给接口模块增加缓存减少重复请求或者把命令行入口替换为 Web API变成一个真正的后端服务。这些方向都可以在这个接口模块的基础上继续叠加。建议先按本文的代码把项目跑通然后把你要接入的真实枫云AI 配置填进去体验一次从配置到运行的完整流程。代码跑通之后再回头审视如果现在要接入第二个服务商你的改动能控制在多少行内如果你已经把接口模块做好这个改动应该非常小。 AI助手接口模块化设计：双龙虾架构实现多Provider接入
如果你已经写过 AI 助手相关的小项目大概率会有一种体感功能第一次跑通的那几分钟特别爽后面维护起来却越来越别扭。换一个模型要翻代码加一个系统提示词要改函数签名想记录一下每次请求的耗时和 token 消耗又得把业务逻辑和 HTTP 调用搅在一起。这些问题不是“能跑”阶段会暴露的而是当你真正想把它做成一个能长期使用的工具时才会集中爆发。今天这篇开发教程主题就是把“能跑的 AI 助手代码”升级成“能维护的 AI 助手架构”。我们会围绕双龙虾接口模块的设计思路把 AI 助手与模型服务之间的接口层重新梳理一遍并以枫云AI 作为接入示例完成一个最小可运行的 CLI 助手。读完你可以掌握一套适合中小型项目的多 Provider 接入方案以后切换模型、新增厂商、加日志和做上下文管理都只需要在接口模块内部处理不需要把主流程拆得七零八落。先说明一点这里的“双龙虾接口模块”并不是某个开源框架的官方组件而是本期教程给接口层模块起的代号。真正值得学习的不是名字本身而是它背后的一层设计思想——把 AI 服务调用从散落的业务代码中收拢为一个独立模块再通过配置驱动不同的 AI 服务商。接下来我会从问题、设计、代码到排错完整走一遍。1. 这篇文章真正要解决的问题很多开发者接入大模型 API 时代码路径几乎一模一样先把 API Key 复制到代码里然后写一个函数把用户输入转发给模型接口拿到返回结果直接打印。这种做法在最早期没有问题因为它能最快验证“这个模型能不能回答我的问题”。但一旦进入真实使用阶段问题就会一个一个冒出来。第一类是密钥管理问题。API Key 写在代码常量里团队协作时很容易被提交到 Git 仓库造成不必要的泄露风险。更麻烦的是当你有多个 AI 服务商每个都有各自的 Key散落在多个文件里几乎没法统一管理。第二类是服务商切换问题。不同厂商的接口风格虽然越来越接近 OpenAI 格式但鉴权方式、请求体字段、错误返回、限流策略都不完全相同。如果业务代码里直接写了requests.post(https://xxx/v1/chat/completions)那么每换一家服务商甚至每换一个模型版本都要去业务代码里改一遍 URL 和参数。第三类是消息结构问题。多轮对话需要维护历史消息如果没有统一的 Message 结构可能会出现“user 和 assistant 消息拼接在不同类型对象里”的状态处理上下文截断时非常痛苦。第四类是测试问题。当 AI 接口调用和业务逻辑耦合在一起时你没办法用本地假数据测试自己的助手逻辑。每次调试都要真实调用远程接口既慢又消耗额度。所以这篇文章要解决的核心问题不是“怎么调用一个 AI 接口”而是“当你的 AI 助手开始长大时怎么让接口层不拖后腿”。2. 双龙虾接口模块是什么为什么接口要模块化2.1 从“函数调用”到“接口模块”在没有模块化之前一个 AI 助手的核心代码通常长这样一个get_response(user_input)函数内部生成一段 JSON里面带有 model、messages、temperature 等字段然后发起 HTTP 请求解析返回结果取出文本。这个函数可以工作但它的职责太多了既要知道怎么发 HTTP又要知道怎么处理该厂商的返回格式还要负责拼装多轮对话历史。接口模块要做的事情是把这些职责拆分出来。双龙虾接口模块可以理解为一个适配层它对外提供统一的chat(messages)方法对内隐藏不同 AI 服务商的接口细节。业务层只需要关心“我发了一条消息助手返回了一段文本”不需要关心请求是发给了谁、用了什么路径、带了多少历史消息。为了做到这一点接口模块至少需要包含几样东西统一的消息数据结构、统一的 Provider 抽象、负责创建 Provider 的工厂逻辑以及具体的厂商实现。这套结构在小型项目中看起来是有点“重”但它带来的收益会随着项目复杂度增加而迅速放大。2.2 没有接口模块时会遇到什么假设你一开始只接入了一家 AI 服务商业务代码里写了这样一段逻辑请求成功就把resp.json()[choices][0][message][content]取出来请求失败就抛异常。这个逻辑看起来没毛病直到你决定接入第二家服务商或者同一场景下需要用不同模型。如果第二家服务商的响应结构不是choices[0].message.content而是output.text你就要在业务代码里加一个 if 判断。再加上不同的认证方式一家需要Authorization: Bearer另一家需要自定义头你的代码会逐渐变成一份“厂商分支大全”。这种代码没有架构上的“错误”但它已经不具备可维护性。接口模块化之后这些分支全部收敛到 Provider 适配器里。新增一家服务商不是去改业务逻辑而是新增一个 Provider 文件并把它注册到工厂里。业务层代码一行都不用改。2.3 接口模块带来的三个核心收益第一个收益是解耦。业务层只依赖抽象接口不依赖具体厂商。这样模型从 A 家切换到 B 家不会影响用户消息处理、上下文管理、日志记录这些核心逻辑。第二个收益是低切换成本。即使枫云AI 后续调整了网关路径或者你决定换一个模型服务改动范围也被限制在具体的 Provider 类内不会扩散到整个项目。第三个收益是可测试性。有了抽象接口你可以写一个 MockProvider在本地运行所有业务逻辑测试。这样既不消耗线上额度又能稳定触发各种边界情况。3. 枫云AI 的接入设计与接口约定在开始写代码前我们先把接入目标定下来。本期教程选择枫云AI 作为示例 AI 服务。这里不会展开它的后台注册、充值、密钥获取流程因为这类信息在不同时期变化较快而且每个同学拿到的服务配置可能不同。更值得关注的是当我们要接入一个具体的 AI 服务商时接口模块应该怎么设计才不会把代码锁死。现在市面上的大模型服务商越来越多地采用与 OpenAI Chat Completions 风格兼容的 HTTP 接口。核心约定通常是这样的请求方法为 POST路径形如/chat/completions请求头中携带认证信息常见格式为Authorization: Bearer api_key请求体里包含model和messages字段messages是一个数组每项包含role和content响应体里包含choices数组其中第一项的message.content就是助手回复文本如果你的目标服务商完全兼容这套协议那适配层的工作量会非常小。如果服务商的格式有差异也不需要恐慌我们只要在对应的 Provider 类里做字段映射把外部返回格式转换为统一结构即可。接口模块需要屏蔽的差异主要有四类差异类型举例处理方式认证方式自定义请求头、Token 参数在 Provider 中封装 header 生成逻辑接口路径/v1/chat/completions或自定义路径在配置中指定 base_url 和路径请求体字段名model、messages命名差异在 Provider 内做转换响应结构choices[0].message.content或output.text在 Provider 内解析并统一返回4. 环境准备与项目结构4.1 环境要求做这个示例项目不需要很重的框架。推荐使用 Python 3.10 及以上版本依赖只需要一个requests库。如果你已经安装了 Python 和 pip再创建虚拟环境即可。具体版本号以你本机环境为准本文重点演示的是接口模块的通用设计思路。4.2 项目目录结构为了让代码清晰我建议按下面的结构组织ai_assistant/ ├── config/ │ └── settings.json ├── core/ │ ├── __init__.py │ ├── message.py │ └── factory.py ├── providers/ │ ├── __init__.py │ ├── base.py │ └── fengyun.py ├── app.py ├── requirements.txt └── README.md简单说明一下各部分职责config/settings.jsonAI 服务商的配置包括接口地址、密钥、模型名。core/message.py统一的消息数据结构。core/factory.pyProvider 工厂根据配置创建具体的服务商实例。providers/base.pyProvider 抽象基类定义统一的调用接口。providers/fengyun.py枫云AI 的 Provider 实现。app.py命令行入口演示完整的多轮对话流程。4.3 初始化项目先创建项目目录和虚拟环境然后安装依赖。mkdir ai_assistant cd ai_assistant python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate pip install requests如果你希望项目更严格可以把依赖写入 requirements.txtrequests2.31.05. 核心代码实现这一部分是整个教程的核心。我们从配置文件开始一步步实现一个完整的多 Provider AI 助手接口模块。5.1 配置文件config/settings.json{ provider: fengyun, fengyun: { base_url: https://api.example.com/v1, api_key: your-fengyun-api-key, model: fengyun-chat, timeout: 60 } }这里的api.example.com是占位地址你需要替换成自己在枫云AI 后台获得的真实网关地址。api_key和model同理都以你在服务商后台实际申请到的配置为准。我们使用 JSON 作为配置格式是因为它的层级结构直观适合表达不同厂商的独立配置块。一个小提醒这个文件不要直接提交到公共 Git 仓库。更安全的做法是把它加入.gitignore或者只在本地保留settings.local.json仓库中放一个不带密钥的settings.example.json。5.2 统一消息结构core/message.py在开始实现 Provider 之前先把消息结构定义好。多轮对话中我们至少需要区分三种角色系统提示词system、用户user、助手assistant。# core/message.py from dataclasses import dataclass from typing import List dataclass class ChatMessage: role: str content: str def to_openai_messages(messages: List[ChatMessage]) - List[dict]: 转换为 OpenAI 风格的消息数组 return [{role: m.role, content: m.content} for m in messages]使用dataclass定义消息结构可以减少样板代码。to_openai_messages是一个纯函数它负责把内部消息对象转换为发送给服务商的格式。如果后续枫云AI 的请求体格式有差异只需要在这里或者 Provider 内部做一次转换业务逻辑不需要感知。5.3 Provider 抽象基类providers/base.py接口模块的核心是抽象。我们定义一个BaseProvider声明所有 Provider 必须实现的方法。# providers/base.py from abc import ABC, abstractmethod from typing import List from core.message import ChatMessage class BaseProvider(ABC): 所有 AI 服务商的统一接口 abstractmethod def chat(self, messages: List[ChatMessage]) - str: 接收完整对话历史返回模型生成的文本。 注意这里的 messages 是完整的历史消息列表 是否截断由调用方负责Provider 只负责转发。 pass abstractmethod def get_model_name(self) - str: 返回当前使用的模型名称便于日志记录 pass这个抽象类看起来很薄但它是整个模块化设计的关键。有了它业务层就可以只依赖BaseProvider而不是具体某个服务商。5.4 枫云AI Provider 实现providers/fengyun.py现在我们来实现具体的枫云AI Provider。假设枫云AI 的服务网关提供与 OpenAI Chat Completions 风格兼容的接口那么代码可以这样写。# providers/fengyun.py import logging from typing import List import requests from core.message import ChatMessage, to_openai_messages from providers.base import BaseProvider logger logging.getLogger(__name__) class FengyunProvider(BaseProvider): def __init__(self, config: dict): self.base_url config[base_url].rstrip(/) self.api_key config[api_key] self.model config[model] self.timeout config.get(timeout, 60) def _build_headers(self) - dict: return { Authorization: fBearer {self.api_key}, Content-Type: application/json, } def chat(self, messages: List[ChatMessage]) - str: url f{self.base_url}/chat/completions payload { model: self.model, messages: to_openai_messages(messages), } logger.info(请求 model%s历史消息数%d, self.model, len(messages)) resp requests.post( url, headersself._build_headers(), jsonpayload, timeoutself.timeout, ) resp.raise_for_status() data resp.json() try: return data[choices][0][message][content] except (KeyError, IndexError) as e: raise RuntimeError(f解析响应失败: {data}) from e def get_model_name(self) - str: return self.model这段代码做了几件事在__init__中从配置字典读取接口地址、密钥和模型名。_build_headers专门负责认证头后续如果要换认证方式只需改这个方法。chat方法负责组装请求体、发起请求、解析响应。使用resp.raise_for_status()让 HTTP 错误在调用处被统一捕获更符合 Python 开发习惯。解析响应时做了异常处理避免因为响应结构变化导致难以排查的 KeyError。如果你的枫云AI 网关不是 OpenAI 兼容格式只需要修改chat方法里的 URL 拼接、请求体结构和响应解析逻辑其他代码完全不用动。5.5 Provider 工厂core/factory.py有了具体 Provider还需要一个工厂来动态创建实例。这样配置里写fengyun就能自动创建FengyunProvider。# core/factory.py from providers.base import BaseProvider from providers.fengyun import FengyunProvider def create_provider(provider_name: str, config: dict) - BaseProvider: if provider_name fengyun: return FengyunProvider(config) raise ValueError(f不支持的 AI Provider: {provider_name})这个工厂目前只支持fengyun。后续新增其他服务商时只需要在函数里增加一个分支或者改用注册表机制。对小型项目来说if 分支已经足够清晰。5.6 命令行入口app.py最后是主程序。这里不做复杂业务只实现一个能跑多轮对话的 CLI。# app.py import json import logging from typing import List from core.factory import create_provider from core.message import ChatMessage logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, ) logger logging.getLogger(app) def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return json.load(f) def main(): config load_config(config/settings.json) provider_name config.get(provider, fengyun) provider_config config[provider_name] provider create_provider(provider_name, provider_config) print(AI 助手已启动输入 exit 或 quit 退出) history: List[ChatMessage] [] while True: user_input input(你: ).strip() if user_input.lower() in (exit, quit): break history.append(ChatMessage(roleuser, contentuser_input)) try: reply provider.chat(history) history.append(ChatMessage(roleassistant, contentreply)) print(fAI: {reply}) except Exception as e: logger.exception(请求接口失败) print(f请求出错: {e}) history.pop() if __name__ __main__: main()注意history.pop()这一段。如果某次请求失败用户消息已经加入了历史列表但助手没有回复这时应该把这条用户消息从历史里去掉避免后续消息带上一次失败请求影响模型对上下文的判断。这是一个很容易被忽略的细节。6. 完整示例与运行验证完成上述代码后就可以运行项目了。先确保你已经在config/settings.json中填入了正确的网关地址、API Key 和模型名。6.1 启动命令在项目根目录执行python app.py如果一切正常你会看到提示AI 助手已启动输入 exit 或 quit 退出 你:然后输入第一句话。比如你: 你好请用一句话介绍你自己模型返回后程序会打印AI: 你好我是基于大模型构建的智能助手可以帮你回答问题、梳理思路和编写代码。这里要注意实际输出内容完全取决于你在枫云AI 配置的模型效果上面只是示意。关键在于你能看到请求成功返回并且下一次提问时会携带上一轮的历史消息。6.2 如何判断运行成功几个简单的判断标准日志中能看到请求 modelfengyun-chat历史消息数1之类的信息说明请求已经发出。打印出的 AI 回复内容与输入问题语义相关说明模型调用链路完整。连续提问两轮后模型能感知到上下文说明历史消息维护逻辑正确。6.3 如果运行失败先看哪里第一次运行最常见的失败点是配置问题。建议按以下顺序排查看config/settings.json中的base_url是否填写正确末尾不要有多余斜杠。看api_key是否被正确读取不要在字符串前后留空格。看model名称是否在服务商的支持列表中。再看控制台输出的异常信息确认是网络错误、认证错误还是响应解析错误。7. 常见问题与排查方法问题现象可能原因排查方式解决方案请求返回 401API Key 错误或已过期检查配置文件和日志中的鉴权头在服务商后台重新生成 Key请求返回 404接口地址或路径不正确查看请求 URL 与服务商文档对比修正base_url或路径拼接逻辑请求返回 400 invalid model模型名称不存在检查日志中 payload 的model字段换成服务商实际支持的模型名请求超时网络连通性异常或模型推理过慢先用 curl 测试网关连通性调整timeout参数或检查网络响应解析报 KeyError服务商响应格式与预期不符打印原始响应 JSON 查看结构调整响应解析逻辑做兼容处理多轮对话语义不连贯历史消息被截断或未正确维护打印history长度和内容增加上下文长度控制只保留最近 N 轮控制台没有日志日志级别配置不正确检查logging.basicConfig配置把级别调成 INFO 或 DEBUG这里最值得提醒的是响应解析问题。不同服务商即使都宣称兼容 OpenAI 格式也可能在流式模式、错误响应、usage 字段上存在细微差别。遇到这类错误时最好的办法不是猜而是在resp.json()之后先打印原始数据确认结构后再写解析逻辑。8. 最佳实践与工程建议代码跑通只是第一步真正让它变成一个可维护项目还需要补充一些工程细节。8.1 密钥管理配置与代码分离推荐的做法是把 API Key 放到环境变量中配置读取时做优先级判断环境变量优先配置文件兜底。这样即使settings.json被误提交也不会直接泄露真实 Key。import os def get_api_key(config: dict) - str: return os.getenv(FENGYUN_API_KEY, config.get(api_key, ))在实际项目中还可以引入 python-dotenv 来加载.env文件。但核心原则是仓库里只保留示例配置真实密钥信息永远放在本地环境或密钥管理服务中。8.2 日志与可观测性接入 AI 服务后日志里至少应该记录三类信息请求的模型名、请求的消息条数、请求耗时。这个日志用于日常排查很有效但不建议把完整对话内容直接写入日志因为用户输入可能包含隐私信息。如果确实需要记录建议先做脱敏处理。8.3 错误处理与重试策略在 Provider 的chat方法里raise_for_status()会触发 HTTP 错误。但并不是所有错误都应该重试。比如 401 认证失败重试多少次都不会成功而 429 限流或 5xx 服务器错误则可以通过短暂等待后重试解决。一个简单的重试逻辑如下import time def chat_with_retry(provider, messages, retries3, delay1.5): for attempt in range(retries): try: return provider.chat(messages) except requests.exceptions.HTTPError as e: if e.response.status_code in (401, 400, 404): raise if attempt retries - 1: raise time.sleep(delay * (attempt 1))这里区分了“不可重试”和“可重试”的错误类型。如果对稳定性要求更高还可以考虑指数退避和抖动但小项目中简单的等退已经足够。8.4 多轮上下文的长度控制大模型对上下文长度有限制所以历史消息不能无限堆积。一个简单的策略是限制最大轮数超过后把最旧的消息丢弃。更精细的做法是按 token 估算但需要额外引入 tokenizer对初学者来说可以先从轮数控制开始。MAX_HISTORY_ROUNDS 10 def append_user_message(history, message): history.append(message) if len(history) MAX_HISTORY_ROUNDS * 2: del history[: len(history) - MAX_HISTORY_ROUNDS * 2]这里MAX_HISTORY_ROUNDS * 2是因为每一轮会新增 user 和 assistant 两条消息。系统提示词可以单独存放在history[0]截断时要留意不要把它删掉。8.5 安全边界在 AI 助手中用户输入最终会传给模型因此输入内容最好先做长度限制和基本校验。另外如果 AI 助手会访问本地文件或执行命令那一部分需要非常谨慎的权限设计。本期教程只完成对话接口不涉及本地操作但你要始终记住模型生成的输出不可完全信任涉及敏感操作时必须有确认环节。8.6 可测试性用 Mock Provider 替换真实调用有了BaseProvider抽象你可以很容易写一个 MockProvider 用于本地测试业务逻辑。# tests/mock_provider.py from typing import List from core.message import ChatMessage from providers.base import BaseProvider class MockProvider(BaseProvider): def chat(self, messages: List[ChatMessage]) - str: return fmock reply for {len(messages)} messages def get_model_name(self) - str: return mock-model这样在写单元测试时就不需要真实调用远程接口也不依赖网络环境和 API 额度。9. 总结与后续学习方向这一期通过双龙虾接口模块的设计把一个 AI 助手中最容易被忽略的接口层重新做了梳理。核心收获可以总结为三点统一消息结构解决了多轮上下文的数据格式问题Provider 抽象解决了多服务商切换的耦合问题配置驱动解决了密钥和模型名散落各处的问题。代码本身并不复杂但每个设计决策都对应真实项目中会遇到的坑。如果你正在做自己的 AI 助手下一步可以继续尝试给 Provider 增加流式输出让回复逐字显示接入多个服务商实现按场景路由给接口模块增加缓存减少重复请求或者把命令行入口替换为 Web API变成一个真正的后端服务。这些方向都可以在这个接口模块的基础上继续叠加。建议先按本文的代码把项目跑通然后把你要接入的真实枫云AI 配置填进去体验一次从配置到运行的完整流程。代码跑通之后再回头审视如果现在要接入第二个服务商你的改动能控制在多少行内如果你已经把接口模块做好这个改动应该非常小。