ai_coin 🪙 —— 可插拔的 AI 接入层

像游戏机的投币口：拧下来，装进任何工程就能用。 统一封装 API 接入 / 多供应商 / 模型管理 / 连接测试 / 工具调用 / 结构化输出 / 流式 / 多模态。

为什么有它

每次做聊天软件都要重写一遍"配置 API 提供商 + 发请求 + 解析响应"的代码， 而且 AI 经常不听话（返回的 JSON 残缺、裹在 markdown 里、被截断、tool call 参数解析崩掉）。 ai_coin 把这一切封装成一个库，调用方只需要几个方法。

安装

pip install ai-coin        # PyPI 发布后
# 或直接拷 ai_coin/ 目录进工程

快速上手

from ai_coin import AICoin

ai = AICoin("config.db")   # 配置存 SQLite，随文件走，可迁移

# ① 加供应商：用内置预设一键填充（DeepSeek/通义/Kimi/智谱/Claude/Gemini/ollama… 共 16 家）
ai.add_provider("DeepSeek", preset="deepseek", api_key="sk-xxx")
#   也可手填：
#   ai.add_provider("自定义", "https://api.xxx.com/v1", "sk-xxx", api_type="openai_compatible")

# ② 管理模型：拉取或手工添加（预设已自动建好默认模型）
pid = ai.list_providers()[0].id
ai.pull_models(pid)              # 从供应商 GET /models 拉取自动入库
ai.add_model(pid, "my-custom-model")  # 手工加

# ③ 连接测试（可选哪个测哪个，结果写回库）
ok, msg = ai.test_model(1)

# ④ 一句话对话
reply = ai.chat(1, "你好")

支持的供应商

• 内置 16 个预设（add_provider(preset=...) 一键配置）： DeepSeek / OpenAI / Kimi(Moonshot) / 智谱GLM / 通义千问 / 豆包(火山) / MiniMax / 阶跃星辰 / Groq / xAI(Grok) / Mistral / Together / Ollama(本地) / Anthropic / Gemini / Azure
• 自定义任意 OpenAI 兼容供应商（api_type="openai_compatible"）
• 四种协议适配器：openai_compatible / azure / anthropic / gemini

ai.list_presets()   # 查看所有预设

核心接口

高级用法

预设一键配置 + 自动建默认模型

ai.add_provider("DeepSeek", preset="deepseek", api_key="sk-xxx")
# → base_url / api_type 自动填好，deepseek-chat、deepseek-reasoner 自动入库

工具调用（function calling）—— 自动循环

def get_weather(city: str) -> str:
    return f"{city} 今天晴，25°C"

reply = ai.chat(model_id, "北京天气怎么样？", tools=[{
    "name": "get_weather",
    "description": "查询城市天气",
    "parameters": {"type": "object",
                   "properties": {"city": {"type": "string"}},
                   "required": ["city"]},
    "function": get_weather,       # 传 Python 函数，框架自动执行并回传
}])

框架自动处理多轮 tool_call → 执行 → 回传，直到 AI 输出最终回答。 工具参数解析失败不会崩，会作为错误回传给模型。

结构化输出（半成品克星）

story = ai.chat(model_id, "编个3行小故事", json_schema={
    "type": "object",
    "properties": {"title": {"type": "string"}, "lines": {"type": "array"}},
    "required": ["title", "lines"],
})
# 返回校验通过的 dict —— 不是可能残缺的文本

内部流程：response_format 强制 JSON → 从响应里捞 JSON（剥 markdown / 修复截断 / 括号补齐）→ 按 schema 校验，不合格自动带错误回喂 AI 重试，最多 4 轮。

流式输出

for chunk in ai.stream_chat(model_id, "讲个故事"):
    print(chunk, end="", flush=True)

推理模型：原始输出不丢失

推理模型（DeepSeek-R1、o1 系列等）的思考过程在 reasoning_content 里， 普通 chat() 会丢掉它。用 chat_full / stream_chat_full 拿原始输出：

# 非流式：完整结果（含思考过程 + token 用量）
full = ai.chat_full(model_id, "1+1=?")
full.content              # 最终答案
full.reasoning_content    # 思考过程
full.usage                # token 用量

# 流式：原始增量块
for chunk in ai.stream_chat_full(model_id, "推理题"):
    print(chunk.reasoning, end="")   # 思考过程增量
    print(chunk.content, end="")     # 正文增量

旧接口 chat() / stream_chat() 仍只返回正文，完全向后兼容。

多模态（图片 / 音频 / 视频）

from ai_coin import image_part, audio_part, video_part, media_part

reply = ai.chat(model_id, {
    "role": "user",
    "content": [
        {"type": "text", "text": "这张图里有什么？"},
        image_part("photo.png"),                  # 图片（本地或 URL）
        # audio_part("record.wav"),               # 音频（本地文件）
        # video_part("clip.mp4"),                 # 视频（本地或 URL）
        # media_part("photo.png")                 # 或按扩展名自动识别类型
    ],
})

*_part() 返回 OpenAI 兼容的媒体片段；Anthropic 适配器会自动转成 它的 content block 格式（image / document），调用方无需关心协议差异。

自定义请求头 / 模型列表端点

ai.add_provider("定制", "https://api.xxx.com/v1", "sk", api_type="openai_compatible",
                models_endpoint="https://api.xxx.com/v1/models",   # 自定义拉模型URL
                extra_headers={"X-Project": "my-app"})             # 额外请求头

错误分类，方便做 UI 提示

from ai_coin import RateLimitError, AuthError, ContextOverflowError, TimeoutError

429 → 限流、401/403 → 鉴权失败、上下文超长 / 请求超时 / 网络错误，全部区分开。

架构

ai_coin/
├── __init__.py     # from ai_coin import AICoin
├── models.py       # Provider / Model / ChatMessage 数据模型
├── store.py        # SQLite 持久化（零依赖，单文件，线程安全，老库自动迁移）
├── providers.py    # 供应商/模型管理 + 拉取列表 + 连接测试
├── presets.py      # ★ 16 家供应商预设
├── adapters.py     # OpenAI兼容 / Azure / Anthropic / Gemini 适配器（含流式）
├── json_fixer.py   # JSON 提取 / 截断修复
├── core.py         # AICoin 主类：对话/工具循环/结构化输出/重试/流式
└── errors.py       # 错误分类

• 数据库化：SQLite 单文件，配置随库走，天然可迁移；老库自动补列升级
• 高度封装：chat() 一个方法搞定适配、重试、工具、结构化输出
• 可扩展：新接入一家供应商 = 写一个 Adapter 子类 + 注册一行

开发

pip install -e ".[dev]"
pytest -v              # 29 个测试：mock 服务器端到端覆盖