编程

用 OpenAI 兼容 API 搭最小聊天 Agent

从鉴权、消息结构到流式输出与错误重试的工程清单。

!封面:用 OpenAI 兼容 API 搭最小聊天 Agent

## 用 OpenAI 兼容 API 搭最小聊天 Agent

协议面

OpenAI 兼容 API(OpenAI-compatible API)是以 Chat Completions 接口为核心的标准化协议,几乎所有公开或自托管模型服务商(包括各类中转站)都遵循此格式。你可以用同一套代码接入任何支持该协议的平台,无需改动模型名称或消息结构。这正是 GrokCode /official-api/api-transit 的核心优势——官方订阅价 vs 中转站综合倍率与稳定性一站比对。

要使用兼容 API,首先需要获取 API Token。推荐路径是先在 /official-prices 查看官方订阅的 ChatGPT Plus 试用订阅(热门商品示例),再通过 /api-transit 的 Sub Cailai One 等中转站获取 token。典型配置:

  • base_url:官方为 https://api.openai.com/v1,中转站通常为 https://your-transit.com/v1https://api.你的中转平台.com/v1
  • api_key:从 /channels 卡网有货/质保价页面或中转样本(状态=active,系统=最低充值=$1)获取

Python 示例(使用官方 openai 库兼容模式):


from openai import OpenAI

import os



client = OpenAI(

    base_url="https://api.你的中转平台.com/v1",  # 替换为中转或官方地址

    api_key=os.getenv("OPENAI_API_KEY") or "sk-你的token"

)



response = client.chat.completions.create(

    model="gpt-4o-mini",  # 支持 openai×26, xai×13 等模型族

    messages=[{"role": "user", "content": "你好"}]

)

print(response.choices[0].message.content)

此步骤确保了协议层面的“最小可运行”。接下来进入消息结构层面。

消息与工具

OpenAI 兼容 API 的消息结构严格遵循 OpenAI 标准:

  • role:必须为 systemuserassistant
  • content:字符串或数组(支持多模态)
  • 可选 nametool_callstool_call_id

工具(function calling)扩展了 Agent 能力。兼容 API 支持 tools 参数和 tool_calls 响应。最小聊天 Agent 可通过工具实现简单函数调用,例如天气查询或计算器。

示例工具函数(Python):


def get_weather(location: str) -> str:

    # 模拟工具实现

    return f"北京天气:25°C,晴"



tools = [

    {

        "type": "function",

        "function": {

            "name": "get_weather",

            "description": "获取指定位置的天气",

            "parameters": {

                "type": "object",

                "properties": {

                    "location": {"type": "string", "description": "城市名"}

                },

                "required": ["location"]

            }

        }

    }

]



messages = [

    {"role": "system", "content": "你是一个助手,可以调用工具"},

    {"role": "user", "content": "北京的天气如何?"}

]



response = client.chat.completions.create(

    model="gpt-4o-mini",

    messages=messages,

    tools=tools,

    tool_choice="auto"

)



# 处理工具调用

if response.choices[0].message.tool_calls:

    for tool_call in response.choices[0].message.tool_calls:

        if tool_call.function.name == "get_weather":

            args = json.loads(tool_call.function.arguments)

            result = get_weather(args["location"])

            messages.append({

                "role": "tool",

                "tool_call_id": tool_call.id,

                "content": result

            })

            # 继续调用以获取最终回复

这一层构建了具备“工具链”的最小 Agent。更多 Agent 构建知识可参考 /guides/build-opc-gateway(OPC 中转网关)或 /guides/openai-compatible-opc

流式与超时

用户期望实时对话,因此必须支持流式输出(stream=True)。兼容 API 的流式响应是 SSE 格式,每 chunk 包含 delta 内容,可实时拼接显示。

超时与重试是生产环境必备:

  • 设置 timeout 参数(单位秒)
  • 使用指数退避重试(避免 429 限流)

Python 完整流式 + 重试示例(含超时与重试):


import json

import time

from openai import OpenAI, APIError, RateLimitError



client = OpenAI(base_url="https://api.你的中转平台.com/v1", api_key="你的token")



messages = [{"role": "user", "content": "解释量子计算"}]



def generate_stream(messages, model="gpt-4o-mini"):

    for attempt in range(3):

        try:

            stream = client.chat.completions.create(

                model=model,

                messages=messages,

                stream=True,

                timeout=30  # 总超时 30 秒

            )

            for chunk in stream:

                if chunk.choices[0].delta.content:

                    print(chunk.choices[0].delta.content, end="", flush=True)

            return  # 成功结束

        except RateLimitError:

            print("限流,指数退避中...")

            time.sleep(2 ** attempt)

        except APIError as e:

            print(f"API 错误:{e},重试 {attempt+1}/3")

            time.sleep(2 ** attempt)

            if attempt == 2:

                raise



generate_stream(messages)

流式输出可直接对接前端 WebSocket 或终端打印,实现“响应即刻”。相关知识见 /guides/rate-limit-abuse(限流防御指南)。

观测与限流

生产 Agent 需要观测请求、模型调用耗时与限流状态。推荐集成简单日志 + Prometheus 指标,或直接打印。

限流处理已在流式示例中体现:使用指数退避 + 最大重试次数。常见错误码(兼容 API 通用的)包括 429(限流)、500(服务器错误)、400(参数错误)。

最小观测代码:


import time



start = time.time()

response = client.chat.completions.create(...)

end = time.time()

print(f"请求耗时:{end-start:.2f}s")



# 统计调用次数与限流事件

将所有 Agent 流程串联即可:从消息结构到工具调用,再到流式输出与重试,就构建了一个可观测的最小聊天 Agent。结合 /channels 卡网中转与 /official-prices 官方价,成本可控制在最低。

上线 checklist

1. 环境准备:Python 3.9+,pip install openai python-dotenv

2. Token 配置:优先 /api-transit Sub Cailai One(active 状态,最低充值 $1)或 /official-api 官方 token

3. 安全:仅存储 key,不硬编码;使用环境变量

4. 测试:本地模拟 + 中转站验证稳定性

5. 部署:Docker / VPS / 服务器,无需额外 SLA

6. 监控:集成日志 + 限流报警

7. 合规:遵守服务条款,仅作个人/合法用途

延伸阅读

  • /guides/openai-compatible-opc
  • /guides/build-opc-gateway
  • /guides/rate-limit-abuse

风险与边界

本文仅讨论合法、防御性运维知识,例如消息验证、超时设置与重试逻辑的工程实践。非法律意见:实际应用中请参考官方文档与法律顾问,遵守各平台的使用政策与数据保护法规。GrokCode 站点不提供 SLA 担保,不协助规避任何地区政策或违法用途,仅作为比价与技术参考。

知识体系位置

本文是「AI 低价订阅与中转 API 比价站」知识地图中的核心编程模块:上接 /official-prices 官方订阅与 /api-transit 中转聚合,下接硬件算力(本地部署)与更深编程主题(多代理系统)。通过 /channels 卡网比价 + /guides 系列,可快速构建从入门到进阶的 OpenAI 兼容 Agent 能力。

(全文约 2450 汉字,含 1 个 Markdown 表格:常见错误码与处理策略如下)

错误类型 常见现象 防御性处理建议
APIError 500/400 立即重试,记录日志
RateLimitError 429 指数退避 + 计数限流
Timeout 超过指定秒数 缩短超时或增加重试次数
Connection 连接失败 检查 base_url 与网络

此清单可直接复制到生产代码中。