返回博客
PythonNbilityOpenAI-compatibleChat CompletionsStreamingAPI 错误处理

Python 调用 Nbility：聊天、流式输出、错误处理

面向 Python 开发者的 Nbility OpenAI-compatible API 入门实战：环境变量、openai-python 客户端、Chat Completions、流式输出、超时重试、错误分类、日志与成本控制。

Python 调用 Nbility：聊天、流式输出、错误处理

Python 调用 Nbility 封面

如果你准备把 AI 能力接进自己的脚本、后台任务、Bot 或 Web 服务，最稳的起点不是上来就写复杂 Agent，而是先把三个基础动作跑通：一次普通聊天调用、一次流式输出、一次可诊断的错误处理。

这篇文章用 Python 演示 Nbility 的 OpenAI-compatible Chat Completions 接入。你可以继续使用熟悉的 openai Python SDK，只需要把 base_url 指向 Nbility，把 API Key 放在环境变量里。后面无论是接 Dify、Hermes Agent、Telegram Bot，还是自己写自动化脚本，这套基础代码都能复用。

调用链路先看懂

Python 调用 Nbility 的最小链路

最小链路只有四步：

  1. .env 保存 NBILITY_API_KEY、NBILITY_BASE_URL、NBILITY_MODEL；
  2. Python 用 OpenAI(api_key=..., base_url=...) 创建客户端；
  3. 调用 client.chat.completions.create() 发送 messages；
  4. 读取 choices[0].message.content，同时记录 usage、延迟和错误信息。

Nbility 的 Chat Completions 文档说明 endpoint 是 POST /v1/chat/completions，请求头使用 Bearer token，请求体包含 model、messages，并支持 stream: true 返回 SSE 流。OpenAI Chat Completions API 也使用相同的核心结构，所以 Python SDK 可以自然复用。

准备环境

建议用虚拟环境，避免污染系统 Python：

python -m venv .venv
source .venv/bin/activate
pip install openai python-dotenv

创建 .env：

NBILITY_API_KEY=[REDACTED]
NBILITY_BASE_URL=https://api.nbility.dev/v1
NBILITY_MODEL=gpt-4o

注意：真实 Key 只放在本地环境变量、服务器 Secret 或 CI Secret 里，不要写进 Git 仓库、前端代码、截图和日志。

最小聊天调用

新建 chat_once.py：

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.environ["NBILITY_API_KEY"],
    base_url=os.getenv("NBILITY_BASE_URL", "https://api.nbility.dev/v1"),
)

response = client.chat.completions.create(
    model=os.getenv("NBILITY_MODEL", "gpt-4o"),
    messages=[
        {"role": "system", "content": "你是一个简洁、可靠的技术助手。"},
        {"role": "user", "content": "用三句话解释什么是 OpenAI-compatible API。"},
    ],
    temperature=0.3,
    max_tokens=300,
)

print(response.choices[0].message.content)
if response.usage:
    print("usage:", response.usage)

运行：

python chat_once.py

如果这一步跑通，你已经完成了最重要的验证：Key、Base URL、模型名、网络、SDK 都是可用的。

封装成可复用函数

项目里不要到处散落 API 调用，建议封装一个小函数：

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["NBILITY_API_KEY"],
    base_url=os.getenv("NBILITY_BASE_URL", "https://api.nbility.dev/v1"),
    timeout=60,
    max_retries=2,
)


def ask_ai(prompt: str, system: str = "你是一个可靠的技术助手。") -> str:
    response = client.chat.completions.create(
        model=os.getenv("NBILITY_MODEL", "gpt-4o"),
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": prompt},
        ],
        temperature=0.2,
        max_tokens=800,
    )
    return response.choices[0].message.content or ""

这里有两个关键点：

  • timeout=60：避免请求长时间卡住；
  • max_retries=2：对临时网络问题做有限重试，但不要无限重试。

如果你用 Nbility 做统一模型入口，可以在环境变量里切换模型，而不必改业务代码。比如开发环境用便宜模型，生产环境对关键任务切到更强模型。

流式输出：让长回答先显示出来

聊天产品、命令行助手、社群机器人都很适合流式输出。用户不需要等完整回答生成完，看到第一段内容就知道系统在工作。

新建 chat_stream.py：

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.environ["NBILITY_API_KEY"],
    base_url=os.getenv("NBILITY_BASE_URL", "https://api.nbility.dev/v1"),
)

stream = client.chat.completions.create(
    model=os.getenv("NBILITY_MODEL", "gpt-4o"),
    messages=[{"role": "user", "content": "写一段 200 字以内的 Python API 接入建议。"}],
    stream=True,
    temperature=0.4,
)

for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

print()

Nbility 文档里 stream: true 会返回 Server-Sent Events；OpenAI Python SDK 已经把 SSE 解析成可迭代的 chunk，所以你只需要读取 delta.content。

在 Web 服务里，后端可以继续把这些片段转成 SSE 或 WebSocket 发给浏览器；在 Telegram / QQ / 飞书 Bot 里，建议做“分段编辑消息”或“先回复处理中，再补全结果”，避免每个 token 都发一条消息。

错误处理：不要只写 except Exception

生产环境最容易踩坑的地方不是正常调用，而是失败时不知道哪里错。建议按 HTTP 状态和 SDK 异常分类处理。

Python API 错误处理矩阵

可以先写一个实用版本：

import os
from openai import OpenAI, APIConnectionError, APIStatusError, RateLimitError

client = OpenAI(
    api_key=os.environ["NBILITY_API_KEY"],
    base_url=os.getenv("NBILITY_BASE_URL", "https://api.nbility.dev/v1"),
    timeout=60,
    max_retries=2,
)


def safe_chat(prompt: str) -> str:
    try:
        response = client.chat.completions.create(
            model=os.getenv("NBILITY_MODEL", "gpt-4o"),
            messages=[{"role": "user", "content": prompt}],
            temperature=0.2,
            max_tokens=500,
        )
        return response.choices[0].message.content or ""
    except RateLimitError as exc:
        raise RuntimeError("请求太频繁或额度压力过高，请稍后重试。") from exc
    except APIConnectionError as exc:
        raise RuntimeError("连接 API 失败，请检查网络、Base URL 或代理设置。") from exc
    except APIStatusError as exc:
        status = exc.status_code
        body = getattr(exc, "response", None)
        detail = body.text[:1000] if body is not None else str(exc)
        if status in {401, 403}:
            raise RuntimeError("认证失败：检查 API Key、Bearer 格式、账号权限或余额。") from exc
        if status in {400, 404}:
            raise RuntimeError(f"请求参数可能有误：检查模型名、messages 和 max_tokens。详情：{detail}") from exc
        if status >= 500:
            raise RuntimeError("上游服务临时异常，可以稍后重试或切换模型。") from exc
        raise RuntimeError(f"API 请求失败：HTTP {status}，{detail}") from exc

这个版本故意不把异常静默吞掉。业务层可以把用户可读信息展示给用户，同时把技术细节写入日志。

日志和成本：至少记录这几个字段

AI API 接入不是“能返回文本就结束”。如果你后面要做多人使用、客户服务、Agent 自动化，建议从第一天记录这些字段：

request_id: 本地生成的请求 ID
user_id: 调用用户或业务账号
model: 实际模型名
stream: 是否流式
latency_ms: 请求耗时
prompt_tokens / completion_tokens / total_tokens: token 用量
status: success / failed
type: error type 或 HTTP status

如果 response 里有 usage，就把它写进日志或数据库。这样你才能回答三个很实际的问题：谁在用、用在哪个功能、为什么成本突然变高。

Nbility 适合放在这里做统一入口：文本模型、图片模型、多模型切换和成本观察都可以集中管理。对开发者来说，代码里保持 OpenAI-compatible 写法，后台用 Nbility 管模型和 token，会比每个供应商写一套适配更省心。

常见问题

1. 401 Unauthorized

优先检查：

API Key 是否正确
是否带了 Authorization: Bearer
.env 是否被正确加载
服务器环境变量是否覆盖了本地配置
Key 是否还有余额或权限

2. 404 model not found

通常是模型名写错，或当前账号没有该模型权限。先用 Nbility 控制台可用模型名，替换 .env 里的 NBILITY_MODEL，再用最小聊天脚本验证。

3. 流式输出没有任何内容

检查三件事：

stream=True 是否真的传入
循环里是否读取 chunk.choices[0].delta.content
代理、网关或 Web 框架是否缓冲了响应

很多 Web 框架默认会缓冲响应，导致前端看起来“不流式”。需要确认后端真的以 SSE 或 chunked response 方式向浏览器输出。

4. 什么时候用同步，什么时候用流式？

同步适合后台任务、短回答、结构化处理；流式适合聊天 UI、长回答、移动端 Bot。流式主要改善等待体验，不一定降低 token 成本。

上线检查清单

[ ] API Key 不出现在代码、前端、日志和截图里
[ ] Base URL 统一为 https://api.nbility.dev/v1
[ ] 模型名放在环境变量，方便切换
[ ] 所有请求设置 timeout
[ ] 429 / 5xx 有有限重试或队列退避
[ ] 400 / 401 / 403 / 404 不盲目重试，优先修配置
[ ] usage、latency、model、user_id 写入日志
[ ] 流式输出在目标客户端真实可见，不被网关缓冲

参考链接

相关文章

Node.js 调用 Nbility：从 CLI 小工具到网页应用
Node.jsJavaScriptNbility

Node.js 调用 Nbility：从 CLI 小工具到网页应用

面向 Node.js / JavaScript 开发者的 Nbility OpenAI-compatible API 接入教程：环境变量、openai npm SDK、CLI 小工具、Express 网页接口、流式输出、错误处理、日志与上线检查。

给小红书/推文生成配图工作流
小红书XTwitter

给小红书/推文生成配图工作流

从标题、卖点和平台尺寸出发，用 AI 生成小红书首图、轮播卡片和 X/Twitter 推文配图，并用脚本控制安全区、比例、导出和审核。

用 Nbility 跑通你的 Agent 工作流

获取 API Key，统一接入 OpenAI 兼容模型和开发工具。

管理 API Key