Python调用AI API完全指南（附代码）

环境准备

Python 版本怎么选

跑 AI API 不吃机器配置，Python 3.9 以上都能跑，建议直接上 3.11 或 3.12，asyncio 和类型提示的改进都是实打实的。3.8 及以下别再用了，新版 openai SDK 已经放弃支持。

python --version          # 确认 >= 3.9
python -m venv venv
venv\Scripts\activate     # Windows
source venv/bin/activate  # macOS / Linux

安装 openai SDK

只需要一个包：

pip install --upgrade openai
pip show openai | findstr Version   # Windows
pip show openai | grep Version      # macOS / Linux

版本要大于 1.40。1.0 之前的旧接口签名完全不同，照着老教程写 openai.ChatCompletion.create 会立刻报错，认准 1.x 的新写法。

把 Key 放进环境变量

$env:OPENAI_API_KEY="sk-你的小鱼API密钥"      # Windows PowerShell，当前窗口有效
export OPENAI_API_KEY="sk-你的小鱼API密钥"    # macOS / Linux

Key 在 xyuai.cc 后台生成，最低充 7 元就能开跑。把 Key 硬编码进源码再推到 GitHub，基本活不过半小时就会被爬虫扫走刷爆。

验证连通性

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://xyuapi.top/v1",
)

resp = client.chat.completions.create(
    model="deepseek-v3.2-thinking",
    messages=[{"role": "user", "content": "只回复两个字：通了"}],
)
print(resp.choices[0].message.content)

输出「通了」说明 Key、网络、模型名三项全对。401 是 Key 无效，404 是模型名不对，连接超时是网络或代理问题，这三种报错放在最后一节单独讲。

这里用的 base_url 是小鱼API 的 OpenAI 兼容接入地址 https://xyuapi.top/v1。它和官方 SDK 完全兼容，调用方式一模一样，只是把域名换成了国内节点，不用挂代理、不用配转发，网络抖动也比绕道海外少得多。同一个 Key 覆盖全系列模型，不用为每家单独申请账号。

最小可运行示例

完整代码

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://xyuapi.top/v1",
    timeout=120.0,     # 思考型模型首字延迟高，给足时间
    max_retries=2,     # SDK 内置重试，只对 429 / 5xx 生效
)

def chat(prompt: str, model: str = "deepseek-v3.2-thinking") -> str:
    resp = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "你是一个简洁的中文技术助手。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.7,
    )
    return resp.choices[0].message.content

if __name__ == "__main__":
    print(chat("用一句话解释什么是 HTTP 长连接"))

三个参数必须搞清楚

• base_url 指向 https://xyuapi.top/v1，末尾不要多加斜杠，SDK 会自己补上 /chat/completions
• timeout 默认值对思考模型来说太短，经常还没吐第一个字就断了，写 120 秒起步比较稳
• max_retries 是 SDK 自带的自动重试，默认 2 次，遇到 429 或 5xx 会自己退避重试，你不用再手写一层

换模型只改一个字符串

一个 Key 能调全线模型，切模型就是改 model 参数：

for m in ["gemini-2.5-pro", "grok-4.5", "claude-sonnet-4-5-thinking"]:
    print(m, "->", chat("1+1 等于几", model=m))

小鱼API 采用按次计费，发 1 个字和发 20 万字同一个价，所以拿多个模型横评效果时完全不用心疼调用次数，这是和按 token 计费平台最大的区别。

流式输出

为什么要开流式

关掉流式，用户要盯着空白界面等十几秒，然后一大段文字一次性蹦出来，体感很差。开流式后第一个字通常 1 到 3 秒就出现了。

完整流式代码

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://xyuapi.top/v1",
    timeout=180.0,
)

def stream_chat(prompt: str, model: str = "grok-4.5") -> str:
    pieces = []
    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    for chunk in stream:
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta
        # 思考型模型的推理过程在 reasoning_content 里
        if getattr(delta, "reasoning_content", None):
            continue
        if delta.content:
            pieces.append(delta.content)
            print(delta.content, end="", flush=True)
    print()
    return "".join(pieces)

if __name__ == "__main__":
    stream_chat("写一段 200 字的 Python 装饰器讲解")

流式下的三个坑

• chunk.choices 可能是空数组，最后一个带 usage 的包里就没有 choices，不做判断会 IndexError
• 带 -thinking 的模型先吐 reasoning_content 再吐 content，想把思考过程展示出来就分开渲染，不想展示直接跳过
• 流到一半断网会抛异常，已经打印的内容要自己接住，别让整个回答白跑

做命令行工具或者日志流式输出时，记得给 print 加 flush=True，否则 Python 会把输出攒在缓冲区里，看起来跟没流式一样。写到文件里的场景则建议先落盘再刷新界面，避免用户看到一半刷新页面内容全丢。

多轮对话上下文管理

messages 数组长什么样

接口本身是无状态的，模型根本不记得你上一句说了什么。所谓「记忆」是你每次把历史消息整段发回去实现的：

messages = [
    {"role": "system", "content": "你是技术支持。"},
    {"role": "user", "content": "报错 401"},
    {"role": "assistant", "content": "401 一般是 Key 无效。"},
    {"role": "user", "content": "那 404 呢？"},
]

一个能直接用的会话类

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://xyuapi.top/v1",
    timeout=120.0,
)

class ChatSession:
    def __init__(self, system: str = "你是一个中文助手", model: str = "kimi-k2.5"):
        self.model = model
        self.messages = [{"role": "system", "content": system}]

    def send(self, text: str) -> str:
        self.messages.append({"role": "user", "content": text})
        resp = client.chat.completions.create(
            model=self.model,
            messages=self.messages,
        )
        reply = resp.choices[0].message.content
        self.messages.append({"role": "assistant", "content": reply})
        return reply

    def reset(self) -> None:
        self.messages = self.messages[:1]

s = ChatSession()
print(s.send("我叫小王，在做电商后台"))
print(s.send("我叫什么？在做什么？"))
print("上下文条数：", len(s.messages))

上下文爆了怎么处理

历史消息无限堆下去，迟早撞上模型的上下文上限，报 400 或者回答质量断崖式下跌。三种做法：只保留 system 加最近二十条；把早期对话摘要成一段话塞回去；换成长上下文模型，例如 gemini-3.1-pro-preview 或 claude-opus-4-6-thinking。小鱼API 按次计费，上下文拉长不会额外加钱，所以做上下文管理主要是为了模型的理解质量，不是为了省钱。

线上服务还要记住一件事：ChatSession 这种对象不能挂在进程内存里就完事。多开几个 worker，同一个用户的下一句请求可能落到另一个进程，历史就断了。正规做法是把 messages 存进 Redis 或者数据库，用会话 ID 去取，进程重启也不会丢。

函数调用与结构化输出

工具定义格式

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询指定城市的天气",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名，例如 杭州"}
            },
            "required": ["city"],
        },
    },
}]

完整的工具调用循环

import json, os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://xyuapi.top/v1",
    timeout=180.0,
)

def get_weather(city: str) -> str:
    fake = {"杭州": "26℃ 多云", "北京": "19℃ 晴"}
    return fake.get(city, "暂无该城市数据")

def run_with_tools(question: str, model: str = "gpt-5.5") -> str:
    messages = [{"role": "user", "content": question}]
    for _ in range(5):   # 防止模型无限调用工具
        resp = client.chat.completions.create(
            model=model, messages=messages, tools=tools,
        )
        msg = resp.choices[0].message
        if not msg.tool_calls:
            return msg.content
        messages.append(msg)
        for call in msg.tool_calls:
            args = json.loads(call.function.arguments)
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": get_weather(**args),
            })
    return "工具调用次数超限"

print(run_with_tools("杭州今天穿什么衣服合适"))

两个容易翻车的细节

• 追加 assistant 消息时必须带上 tool_calls 字段，只发文本会丢上下文，模型下一轮会重复发起同一个调用，白白烧钱
• 每条 tool 结果都要带 tool_call_id，多个工具并行调用时错填会串结果

让模型吐出结构化 JSON

import json
from pydantic import BaseModel, ValidationError

schema = {
    "type": "json_schema",
    "json_schema": {
        "name": "user_profile",
        "schema": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "age": {"type": "integer"},
                "skills": {"type": "array", "items": {"type": "string"}},
            },
            "required": ["name", "age", "skills"],
            "additionalProperties": False,
        },
    },
}

resp = client.chat.completions.create(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "从这句话抽人：张涛，31 岁，会 Python 和 Go。"}],
    response_format=schema,
)
data = json.loads(resp.choices[0].message.content)

class Profile(BaseModel):
    name: str
    age: int
    skills: list[str]

try:
    profile = Profile(**data)
    print(profile.name, profile.skills)
except ValidationError as e:
    print("模型输出不符合预期：", e)

上面的代码里顺手做了兜底：部分模型收到 json_schema 会直接返回 400，这时候换成 response_format={"type": "json_object"}，在提示词里把字段说明写死，再让 Pydantic 校验一遍。多这一层校验不亏，模型偶尔会漏字段，或者把整数顺手写成字符串，靠类型校验能第一时间发现。

错误处理与重试

异常类型对应表

异常类	HTTP	含义	处理方式
AuthenticationError	401	Key 无效或余额不足	检查 Key，不要重试
NotFoundError	404	模型名不存在	对照模型列表改名
RateLimitError	429	触发限流	指数退避重试
APITimeoutError	-	请求超时	提高 timeout 后重试
APIConnectionError	-	网络不通或代理异常	排查网络后重试
InternalServerError	5xx	上游异常	退避重试

带抖动的指数退避

import random, time, os
from openai import OpenAI, RateLimitError, APITimeoutError, APIConnectionError

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://xyuapi.top/v1",
    timeout=180.0,
)

RETRYABLE = (RateLimitError, APITimeoutError, APIConnectionError)

def call_with_retry(fn, *args, attempts: int = 5, **kwargs):
    for i in range(attempts):
        try:
            return fn(*args, **kwargs)
        except RETRYABLE as e:
            if i == attempts - 1:
                raise
            wait = min(2 ** i, 30) * (0.5 + random.random())
            print(f"第 {i + 1} 次失败（{type(e).__name__}），{wait:.1f}s 后重试")
            time.sleep(wait)

reply = call_with_retry(lambda: client.chat.completions.create(
    model="deepseek-v4-flash-thinking",
    messages=[{"role": "user", "content": "你好"}],
))
print(reply.choices[0].message.content)

加随机抖动是为了避免一批并发任务同时退避、又同时打回来，形成二次拥堵。另外把失败次数和错误类型打进日志，跑批的时候一眼就能看出是网络在抖还是模型名写错了，这比事后翻上传的原始响应省事得多。

超时到底设多少

单一 timeout 不够用，拆开会更稳：连接超时 10 秒，读取超时按模型分档。普通模型 60 秒够，claude-opus-4-6-thinking、gpt-5.5-pro-thinking 这类深度思考模型给 180 秒。

import httpx
client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://xyuapi.top/v1",
    timeout=httpx.Timeout(10.0, read=180.0),
)

异步并发批量请求

asyncio 加 AsyncOpenAI

串行跑 100 条请求要几分钟，并发跑十几秒就完了：

import asyncio, os
from openai import AsyncOpenAI

aclient = AsyncOpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://xyuapi.top/v1",
    timeout=180.0,
)

async def one(prompt: str, sem: asyncio.Semaphore) -> str:
    async with sem:
        r = await aclient.chat.completions.create(
            model="deepseek-v4-flash-thinking",
            messages=[{"role": "user", "content": prompt}],
        )
        return r.choices[0].message.content

async def batch(prompts, concurrency: int = 8):
    sem = asyncio.Semaphore(concurrency)
    return await asyncio.gather(
        *(one(p, sem) for p in prompts), return_exceptions=True
    )

if __name__ == "__main__":
    prompts = [f"第 {i} 条：用一句话介绍数字 {i}" for i in range(1, 21)]
    results = asyncio.run(batch(prompts))
    ok = [r for r in results if isinstance(r, str)]
    bad = [r for r in results if not isinstance(r, str)]
    print(f"成功 {len(ok)} 条，失败 {len(bad)} 条")
    print(ok[0])

并发数怎么定

Semaphore 是必须的。不限并发直接 gather 一千个请求，本地端口会先耗光，上游也可能返回 429。经验值从 5 到 10 开始，跑稳了再加。小鱼API 国内直连不降智不限速，但客户端自己的连接池也有上限。

异常兜底不能省

asyncio.gather 默认一个任务抛异常就整体炸掉。加上 return_exceptions=True，失败的返回异常对象，其余结果照常拿到，批量任务才不会一条失败全军覆没。

图片输入（多模态）

base64 传本地图片

import base64, os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://xyuapi.top/v1",
    timeout=180.0,
)

def encode_image(path: str) -> str:
    with open(path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

b64 = encode_image("./screenshot.png")
resp = client.chat.completions.create(
    model="gemini-2.5-pro",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "这张图报了什么错？给出修复建议。"},
            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}},
        ],
    }],
)
print(resp.choices[0].message.content)

图片 URL 传远程图

远程图直接给链接，不用下载到本地：

{"type": "image_url", "image_url": {"url": "https://example.com/diagram.jpg"}}

图片相关的三个注意事项

• 大图先压到 1MB 以内，base64 之后体积还会再涨约三分之一，太大会被服务端直接拒掉
• 支持看图的模型有 gemini-2.5-pro、gpt-5.5、claude-sonnet-4-5-thinking 等，grok-4.5 系列解析图表也够用
• 要生成图片就换生图模型，NanoBanana-Pro 单次 0.27 元，gpt-image-2-pro 单次 0.25 元
• 传图之前把无关区域裁掉，模型读图比读文字慢得多，多余的像素只会拖慢响应、还可能分散注意力

成本控制与真实价格表

从响应里读用量

resp = client.chat.completions.create(
    model="deepseek-v3.2-thinking",
    messages=[{"role": "user", "content": "写一首关于秋天的五言绝句"}],
)
u = resp.usage
print(u.prompt_tokens, u.completion_tokens, u.total_tokens)

小鱼API 的计费逻辑

小鱼API 走按次计费：一次对话请求算一次钱，输入 1 个字和输入 20 万字同一个价，输出多长也不影响单价。usage 字段拿来做质量监控和日志分析可以，但它不参与算钱，这点和按 token 计费的平台完全不同。整库代码审查、长合同比对这类原本很烧钱的场景，在这里成本是可预期的。

真实价格表

模型	单价（元/次）	类型
gemini-2.5-pro	0.031	对话
gemini-3-pro-preview	0.05	对话
gemini-3.1-pro-preview	0.09	对话
NanoBanana-Pro	0.27	生图
claude-sonnet-4-5-thinking	0.09	思考
claude-opus-4-5-thinking	0.12	思考
claude-sonnet-4-6-thinking	0.2	思考
claude-opus-4-6-thinking	0.25	思考
claude-sonnet-4-7-thinking	0.2	思考
claude-opus-4-7-thinking	0.25	思考
grok-4.1	0.05	对话
grok-4.5	0.09	对话
grok-4.6	0.09	对话
deepseek-v3.2-thinking	0.049	思考
deepseek-r1-thinking	0.049	思考
deepseek-v4-flash-thinking	0.05	思考
deepseek-v4-pro-thinking	0.09	思考
gpt-5.5	0.2	对话
gpt-5.5-pro-thinking	0.2	思考
gpt-5.3-pro	0.3	对话
gpt-image-2-pro	0.25	生图
kimi-k2.5	0.09	对话
kimi-k2.6	0.09	对话

最低 7 元起充，用多少扣多少。日调用量上千次的话，无限卡套餐比单次扣费更划算。

三种用量的月成本估算

按次计费的好处是不用等账单，可以直接手算：调用次数乘单价就是月成本。下面按 0.05 元这一档估三种典型用量。

使用强度	每天调用	每月调用	月成本估算
个人学习、零星调试	30 次	900 次	约 45 元
小团队跑批量任务	500 次	15000 次	约 750 元
线上服务日常调用	2000 次	60000 次	约 3000 元

个人这一档完全没必要买套餐，最低 7 元起充，按周补额度就够，用不掉也不浪费。跑量上来之后单价再低也架不住次数多，那时候再看无限卡套餐，把成本摊成固定支出更合算。

唯一要动脑子的是估准自己的真实调用次数。插件或脚本装好先跑一周，把日志里的请求条数统计出来再乘单价，别拍脑袋填一个「大概每天几百次」。不确定就按中间那一档准备预算，跑满一个月再按真实数据往下调。

省钱的三条实战经验

• 打磨逻辑阶段用 deepseek-v3.2-thinking（0.049 元/次），定稿再拿 claude-opus-4-6-thinking 复检一遍
• 批量任务先跑 5 条验证提示词，确认输出稳定再放全量，别一上来就几千条
• 加一层本地缓存，同样的输入直接读缓存，压根不产生调用

最容易踩的四个坑

超时没设，程序假死

SDK 默认超时在思考模型上常常不够，请求挂在那里一动不动，日志里什么也没有。统一写 timeout=120.0 起步，思考类给 180 秒，再用 httpx.Timeout 把连接和读取分开设，出问题能立刻定位在连接还是读取阶段。

代理配置冲突

openai.APIConnectionError: Connection error.

机器上挂着梯子时，所有请求都会绕一圈，反而更慢甚至直接失败。小鱼API 是国内直连，最省事的做法是先清掉代理变量再测：

$env:HTTP_PROXY=""; $env:HTTPS_PROXY=""

unset HTTP_PROXY HTTPS_PROXY ALL_PROXY

确实要走代理，就给 httpx 显式传参，别依赖系统环境变量，环境变量很容易被别的软件改掉。

编码问题

Windows 控制台默认 GBK，打印模型返回的 emoji 或特殊符号会直接 UnicodeEncodeError。脚本开头加一行就好：

import sys
sys.stdout.reconfigure(encoding="utf-8")

读本地文件同理，open(path, encoding="utf-8") 里的编码参数别省，否则中文文档读成乱码，喂给模型的结果自然也是错的。

模型名写错

openai.NotFoundError: Error code: 404

模型名区分大小写和连字符，claude-sonnet-4-5-thinking 写成 claude-sonnet-4.5-thinking 就是 404。直接从上面的价格表复制，别手打。