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 类里做字段映射，把外部返回格式转换为统一结构即可。

接口模块需要屏蔽的差异主要有四类：

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.json：AI 服务商的配置，包括接口地址、密钥、模型名。
• core/message.py：统一的消息数据结构。
• core/factory.py：Provider 工厂，根据配置创建具体的服务商实例。
• providers/base.py：Provider 抽象基类，定义统一的调用接口。
• 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.txt：

requests>=2.31.0

5. 核心代码实现

这一部分是整个教程的核心。我们从配置文件开始，一步步实现一个完整的多 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": f"Bearer {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, headers=self._build_headers(), json=payload, timeout=self.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( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) logger = logging.getLogger("app") def load_config(path: str) -> dict: with open(path, "r", encoding="utf-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(role="user", content=user_input)) try: reply = provider.chat(history) history.append(ChatMessage(role="assistant", content=reply)) print(f"AI: {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 如何判断运行成功

几个简单的判断标准：

• 日志中能看到请求 model=fengyun-chat，历史消息数=1之类的信息，说明请求已经发出。
• 打印出的 AI 回复内容与输入问题语义相关，说明模型调用链路完整。
• 连续提问两轮后，模型能感知到上下文，说明历史消息维护逻辑正确。

6.3 如果运行失败，先看哪里

第一次运行最常见的失败点是配置问题。建议按以下顺序排查：

1. 看config/settings.json中的base_url是否填写正确，末尾不要有多余斜杠。
2. 看api_key是否被正确读取，不要在字符串前后留空格。
3. 看model名称是否在服务商的支持列表中。
4. 再看控制台输出的异常信息，确认是网络错误、认证错误还是响应解析错误。

7. 常见问题与排查方法

这里最值得提醒的是响应解析问题。不同服务商即使都宣称兼容 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, retries=3, delay=1.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 f"mock reply for {len(messages)} messages" def get_model_name(self) -> str: return "mock-model"

这样在写单元测试时，就不需要真实调用远程接口，也不依赖网络环境和 API 额度。

9. 总结与后续学习方向

这一期通过双龙虾接口模块的设计，把一个 AI 助手中最容易被忽略的接口层重新做了梳理。核心收获可以总结为三点：统一消息结构解决了多轮上下文的数据格式问题；Provider 抽象解决了多服务商切换的耦合问题；配置驱动解决了密钥和模型名散落各处的问题。代码本身并不复杂，但每个设计决策都对应真实项目中会遇到的坑。

如果你正在做自己的 AI 助手，下一步可以继续尝试：给 Provider 增加流式输出，让回复逐字显示；接入多个服务商，实现按场景路由；给接口模块增加缓存，减少重复请求；或者把命令行入口替换为 Web API，变成一个真正的后端服务。这些方向都可以在这个接口模块的基础上继续叠加。

建议先按本文的代码把项目跑通，然后把你要接入的真实枫云AI 配置填进去，体验一次从配置到运行的完整流程。代码跑通之后，再回头审视：如果现在要接入第二个服务商，你的改动能控制在多少行内？如果你已经把接口模块做好，这个改动应该非常小。