HiAPI
Zh
博客

教程

用 Python 构建 OpenAI API 代理：先亲手写一个，再让它对接所有模型

一个可用的 FastAPI 代理，带鉴权、限流和流式传输；以及 HiAPI 的 OpenAI 兼容 API 如何让同一个代理同时服务 DeepSeek、Claude、GPT 等模型。

2026年11月3日
用 Python 构建 OpenAI API 代理：先亲手写一个，再让它对接所有模型

TL;DR

  • OpenAI API 代理是你应用和模型 API 之间的一个小型服务器。它保管真实密钥，并补上 API 本身不提供的能力：按客户划分的密钥、限流、日志和密钥轮换。下面是一个约 50 行的可用 FastAPI 版本。
  • HiAPI 的文本 API 兼容 OpenAI：POST /v1/chat/completions 和 POST /v1/responses 位于 https://api.hiapi.ai/v1。把官方 openai SDK 或你的代理指向这个 base URL，就能用一个密钥调用 DeepSeek、Claude、GPT、Kimi、GLM 和 Grok 文本模型。
  • 图像和视频不一样：它们需要几秒到几分钟，所以运行在 HiAPI 的异步 POST /v1/tasks 上。不要把这类请求塞进同步代理。

第 1 步：一个最小的透传代理

代理会校验客户端发来的内部密钥，替换成真实的上游密钥，然后把响应流式传回。

pip install fastapi uvicorn httpx

# proxy.py
import os
import httpx
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
from starlette.background import BackgroundTask

UPSTREAM = os.environ.get("UPSTREAM_BASE", "https://api.openai.com/v1")
UPSTREAM_KEY = os.environ["UPSTREAM_KEY"]
INTERNAL_KEYS = set(os.environ["INTERNAL_KEYS"].split(","))  # comma-separated

app = FastAPI()
client = httpx.AsyncClient(timeout=120.0)

def auth(request: Request) -> str:
    header = request.headers.get("authorization", "")
    key = header.removeprefix("Bearer ").strip()
    if key not in INTERNAL_KEYS:
        raise HTTPException(401, "invalid internal key")
    return key

@app.post("/v1/{path:path}")
async def forward(path: str, request: Request):
    tenant = auth(request)
    body = await request.body()
    upstream_req = client.build_request(
        "POST", f"{UPSTREAM}/{path}", content=body,
        headers={"Authorization": f"Bearer {UPSTREAM_KEY}",
                 "Content-Type": request.headers.get("content-type", "application/json")},
    )
    upstream = await client.send(upstream_req, stream=True)
    # Stream back so long answers and SSE don't time out; close the upstream stream when done
    return StreamingResponse(upstream.aiter_raw(), status_code=upstream.status_code,
                             media_type=upstream.headers.get("content-type"),
                             background=BackgroundTask(upstream.aclose))

运行它：

UPSTREAM_KEY=your-upstream-key INTERNAL_KEYS=tenant-a,tenant-b \
uvicorn proxy:app --host 0.0.0.0 --port 8000

客户端照常使用 OpenAI SDK，把代理地址设为 base URL：

from openai import OpenAI

client = OpenAI(api_key="tenant-a", base_url="http://localhost:8000/v1")
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
)
print(resp.choices[0].message.content)

对话、流式传输和嵌入都走同一个处理函数，因为代理从不解析请求体内部。

第 2 步：补上上游 API 不提供的能力

按租户限流。没有它，一个繁忙的客户就能耗尽你整个配额。

import time
from collections import defaultdict

_window: dict[str, list[float]] = defaultdict(list)
LIMIT = 60  # requests per 60 s

def check_rate_limit(tenant: str):
    now = time.time()
    _window[tenant] = [t for t in _window[tenant] if now - t < 60]
    if len(_window[tenant]) >= LIMIT:
        raise HTTPException(429, "rate limit exceeded")
    _window[tenant].append(now)

在 auth() 之后立刻调用它。如果有多台服务器进程，就把计数器搬到 Redis（INCR 加 EXPIRE）。

日志。记录租户、路径、状态码、延迟和 token 数量。绝不记录 Authorization 请求头。

密钥轮换。从密钥管理服务加载 UPSTREAM_KEY，并定时刷新。上游密钥在底层更换时，客户端仍继续使用各自的租户密钥。

成本规则。解析请求体，拒绝超出策略的请求，例如在免费套餐上使用昂贵模型，或 max_tokens 超过你的上限。

第 3 步：把多家模型厂商放在同一个代理后面

代理只对接一家厂商很容易。要加上 Anthropic、DeepSeek 或 Moonshot，就意味着不同的鉴权请求头、请求结构和错误格式，薄代理就会变成翻译层。

HiAPI 的文本 API 对其所有文本模型都采用 OpenAI 的请求结构，所以上面的代理无需改动即可工作。设置两个环境变量：

UPSTREAM_BASE=https://api.hiapi.ai/v1 UPSTREAM_KEY=your-hiapi-key INTERNAL_KEYS=tenant-a,tenant-b \
uvicorn proxy:app --host 0.0.0.0 --port 8000

或者跳过代理，直接使用 SDK：

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["HIAPI_KEY"], base_url="https://api.hiapi.ai/v1")
resp = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[{"role": "user", "content": "Explain sparse attention in two sentences."}],
    max_tokens=400,
)
print(resp.choices[0].message.content)

我们用官方 openai Python 包（3.35.1）运行了这段代码，一次常规调用，一次带 stream=True；两次调用都无报错完成，SDK 也照常解析了响应。这次测试有一个坑：当 max_tokens=20 时，回复为空，因为 deepseek-v4-flash 把全部 20 个 token 都花在了推理上（usage.completion_tokens_details.reasoning_tokens 为 22）。给推理模型留几百个 token 的余量。

HiAPI 上的一些文本模型：

模型 ID 路由 输入 / 输出，每 100 万 token 美元
deepseek-v4-flash Chat Completions $0.14 / $0.28
deepseek-v4.1-flash Chat Completions 和 Responses $0.44 / $1.32
gpt-5.6-luna 仅 Responses $0.05 / $0.30
grok-4.6 Chat Completions $0.84 / $2.52
deepseek-v4-pro Chat Completions 和 Responses $1.50 / $4.50
claude-opus-5-5 Chat Completions $2.02 / $10.08
kimi-k3 Chat Completions $2.32 / $11.60
claude-sonnet-4-6 Chat Completions $3.00 / $15.00

HiAPI 实时价格，查询于 2026 年 10 月 20 日。GET https://api.hiapi.ai/v1/models 会列出你的密钥可用的所有模型。

像 gpt-5.6-luna 这样的 GPT 模型只在 Responses API 上提供（client.responses.create），所以转发 /v1/{path} 的代理可以同时覆盖两者。

未知模型 ID 会以 OpenAI 的错误风格返回 HTTP 400，并附带一个 request_id 供联系支持：

{"error": {"code": "model_not_found", "type": "hiapi_error",
           "message": "The requested model does not exist or is not available. Check the exact model ID with GET /v1/models and try again. ...",
           "request_id": "..."}}

第 4 步：图像和视频不该放进同步代理

同步代理会一直保持连接打开，直到答案返回。这对文本没问题。图像要几十秒，视频要几分钟，这比大多数负载均衡器和浏览器愿意等待的时间都长。

HiAPI 在异步端点上运行图像和视频模型：提交任务，拿到 task id，然后轮询或接收回调。

curl -s -X POST https://api.hiapi.ai/v1/tasks \
  -H "Authorization: Bearer $HIAPI_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "z-image",
    "input": {"prompt": "a glass data center at golden hour, isometric", "aspect_ratio": "16:9"},
    "callback": {"url": "https://your-app.example.com/hiapi/callback", "when": "final"}
  }'
# => {"code":200,"message":"success","data":{"taskId":"tk-hiapi-..."}}

你的代理仍然可以转发提交调用，它会立即返回。结果随后通过你的回调 URL 或 GET /v1/tasks/{taskId} 返回。输出链接约 7 天后过期，所以请保存文件。

生产环境检查清单

  • 从日志中移除密钥。无论传入还是传出，都绝不记录 Authorization。
  • 媒体任务用回调。轮询成千上万个任务会浪费请求。把回调处理写成幂等的，并以 taskId 为键，因为投递可能重试。
  • 轮询前先存储 task id。如果进程崩溃，恢复轮询，而不是重新提交并付两次费。
  • 为推理 token 留出空间。把 max_tokens 设得远高于你预期的答案长度。
  • 检查密钥能用什么。GET /v1/models 会返回该密钥对应的列表。

常见问题

我可以在 HiAPI 上使用 OpenAI Python SDK 吗？

可以，用于文本模型。设置 base_url="https://api.hiapi.ai/v1" 并使用你的 HiAPI 密钥。图像和视频使用 /v1/tasks，OpenAI SDK 不覆盖该端点；请用 requests 或 httpx 调用。

我还需要自己的代理吗？

只有出于和任何厂商相同的理由才需要：按租户划分的密钥、限流和审计日志。代理内部不再需要按厂商做转换。

流式传输能用吗？

能。stream=True 会返回以 data: [DONE] 结尾的服务器发送事件，上面的代理会不加缓冲地流式转发它们。

如何在不中断服务的情况下轮换 HiAPI 密钥？

在控制台创建新密钥，部署到你的代理或密钥存储，然后吊销旧密钥。切换期间两个密钥都可用。

更多细节：Chat Completions 文档、Responses 文档以及作为 OpenAI API 代理替代方案的 HiAPI。

标签pythonOpenAIProxyTutorial

相关文章

查看全部文章 →

立即开始构建持久 AI 产物

获取 API 密钥，通过一个基于任务（task）的 API 调用图像、视频和音频模型。

阅读 API 文档