⚠️ 待更新·2026-08-29核验 · 更新时间待核验 · 本文信息可能已过期，请以官方文档为准
更新时间：2026-08-29 · 核验状态：待更新 · 官方溯源待补
简介
Groq、Together AI、DeepSeek、NVIDIA NIM、OpenRouter 都遵循 OpenAI Chat Completions 协议。这意味着一套 openai SDK 代码,只换 base_url 与 api_key,就能在所有这些平台之间无缝切换。本篇给出统一封装,并演示流式与异步两种进阶用法。
架构图
flowchart TD
A[OpenAI SDK codebase] --> B{Switch base_url + api_key}
B -->|groq.com| C[Groq LPU]
B -->|together.xyz| D[Together AI]
B -->|deepseek.com| E[DeepSeek]
B -->|integrate.api.nvidia.com| F[NVIDIA NIM]
C --> G[Same SDK shape]
D --> G
E --> G
F --> G
安装
pip install openai python-dotenv
统一封装
import os
from dataclasses import dataclass
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
@dataclass
class Provider:
name: str
base_url: str
api_key_env: str
model: str
PROVIDERS = {
"groq": Provider("Groq", "https://api.groq.com/openai/v1", "GROQ_API_KEY", "llama-3.3-70b-versatile"),
"together": Provider("Together", "https://api.together.xyz/v1", "TOGETHER_API_KEY", "meta-llama/Llama-3.3-70B-Instruct-Turbo"),
"deepseek": Provider("DeepSeek", "https://api.deepseek.com/v1", "DEEPSEEK_API_KEY", "deepseek-chat"),
"nim": Provider("NIM", "https://integrate.api.nvidia.com/v1", "NVIDIA_API_KEY", "meta/llama-3.1-8b-instruct"),
"openrouter": Provider("OpenRouter", "https://openrouter.ai/api/v1", "OPENROUTER_API_KEY","meta-llama/llama-3.3-70b-instruct:free"),
}
def client_for(name: str) -> OpenAI:
p = PROVIDERS[name]
return OpenAI(api_key=os.environ[p.api_key_env], base_url=p.base_url)
def chat(name: str, prompt: str, system: str = "你是简洁的中文助手") -> str:
p = PROVIDERS[name]
resp = client_for(name).chat.completions.create(
model=p.model,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": prompt},
],
temperature=0.3,
max_tokens=256,
)
return resp.choices[0].message.content
if __name__ == "__main__":
for name in PROVIDERS:
try:
print(f"[{name}] {chat(name, '用一句话解释什么是反向索引')}")
except Exception as e:
print(f"[{name}] ERROR: {e}")
流式封装
def chat_stream(name: str, prompt: str):
p = PROVIDERS[name]
stream = client_for(name).chat.completions.create(
model=p.model,
messages=[{"role": "user", "content": prompt}],
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield delta
for piece in chat_stream("groq", "写一首秋天的诗"):
print(piece, end="", flush=True)
print()
异步并发(Async)
跑批处理时,AsyncOpenAI 能并发请求多家 provider:
import asyncio
from openai import AsyncOpenAI
async def achat(name, prompt):
p = PROVIDERS[name]
client = AsyncOpenAI(api_key=os.environ[p.api_key_env], base_url=p.base_url)
r = await client.chat.completions.create(
model=p.model,
messages=[{"role":"user","content":prompt}],
max_tokens=64,
)
return name, r.choices[0].message.content
async def main():
results = await asyncio.gather(*[achat(n, "hi") for n in PROVIDERS])
for name, text in results:
print(f"[{name}] {text}")
asyncio.run(main())
选型策略
- 最低延迟:Groq(LPU)
- 最便宜:OpenRouter
:free 模型
- 中文最强:DeepSeek
- 本地化部署:NIM 容器
- 多 provider 容灾:OpenRouter 的 fallback 字段
容灾与降级
生产环境推荐组合使用:主链路用付费稳定 provider(如 OpenAI/Anthropic),备份链路配置多家免费 provider,主链路异常时自动切换。可借助 OpenRouter 的 fallback 字段实现零代码容灾,也可在外层包 try/except 手动切换。并发场景下用 asyncio.Semaphore 控制并发度,批处理时把请求存到队列,worker 按各家限速消费,避免触发 429。跨时区场景下,境外 provider 在中国早晚高峰延迟波动大,可加连接池复用与本地代理。
流式输出封装
所有 provider 都支持 OpenAI 风格的流式响应,封装一个统一的生成器即可复用。下面这段代码把流式 token 转成 Python 生成器,业务侧无需关心是哪家 provider:
def chat_stream(name: str, prompt: str):
p = PROVIDERS[name]
stream = client_for(name).chat.completions.create(
model=p.model,
messages=[{"role": "user", "content": prompt}],
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield delta
for piece in chat_stream("groq", "写一首秋天的诗"):
print(piece, end="", flush=True)
print()
这样前端 SSE 推送、命令行打字机效果都能复用同一份生成器。
常见问题
401:对应 provider 的环境变量没设或拼错,检查 api_key_env。model not found:不同平台的模型 ID 不同,以各自 /v1/models 返回为准。- 想加自动 fallback:在外层包 try/except,失败时切换到下一个 provider。
- token 计费口径不一:各家对 system prompt 是否计入缓存不同,成本对比需统一基准。
掌握这套模式后,切换模型只需改一个字符串。
最佳实践
- 抽象一个
make_client(provider) 工厂：统一返回 OpenAI(base_url, api_key)，业务代码不关心 provider。
- 环境变量按 provider 命名：
GROQ_API_KEY、TOGETHER_API_KEY、DEEPSEEK_API_KEY 等分开存。
base_url 末尾要带 /v1：少了 /v1 会 404，所有兼容层都遵循 OpenAI 路径约定。- 错误码语义统一：不同 provider 的 429 含义可能不同（RPM vs TPM），网关层要归一化。
⚠️ Pending Update · 2026-08-29 Verification · Content may be outdated, please refer to official docs
Updated: 2026-08-29 · Status: Pending Verification
Introduction
Groq, Together AI, DeepSeek, NVIDIA NIM, and OpenRouter all implement the OpenAI Chat Completions protocol. That means a single openai SDK codebase — with only base_url and api_key swapped — can move seamlessly across all of them. This article provides a unified wrapper and demos streaming and async patterns.
架构图
flowchart TD
A[OpenAI SDK codebase] --> B{Switch base_url + api_key}
B -->|groq.com| C[Groq LPU]
B -->|together.xyz| D[Together AI]
B -->|deepseek.com| E[DeepSeek]
B -->|integrate.api.nvidia.com| F[NVIDIA NIM]
C --> G[Same SDK shape]
D --> G
E --> G
F --> G
Install
pip install openai python-dotenv
Unified Wrapper
import os
from dataclasses import dataclass
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
@dataclass
class Provider:
name: str
base_url: str
api_key_env: str
model: str
PROVIDERS = {
"groq": Provider("Groq", "https://api.groq.com/openai/v1", "GROQ_API_KEY", "llama-3.3-70b-versatile"),
"together": Provider("Together", "https://api.together.xyz/v1", "TOGETHER_API_KEY", "meta-llama/Llama-3.3-70B-Instruct-Turbo"),
"deepseek": Provider("DeepSeek", "https://api.deepseek.com/v1", "DEEPSEEK_API_KEY", "deepseek-chat"),
"nim": Provider("NIM", "https://integrate.api.nvidia.com/v1", "NVIDIA_API_KEY", "meta/llama-3.1-8b-instruct"),
"openrouter": Provider("OpenRouter", "https://openrouter.ai/api/v1", "OPENROUTER_API_KEY","meta-llama/llama-3.3-70b-instruct:free"),
}
def client_for(name: str) -> OpenAI:
p = PROVIDERS[name]
return OpenAI(api_key=os.environ[p.api_key_env], base_url=p.base_url)
def chat(name: str, prompt: str, system: str = "You are a concise assistant.") -> str:
p = PROVIDERS[name]
resp = client_for(name).chat.completions.create(
model=p.model,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": prompt},
],
temperature=0.3,
max_tokens=256,
)
return resp.choices[0].message.content
if __name__ == "__main__":
for name in PROVIDERS:
try:
print(f"[{name}] {chat(name, 'Explain an inverted index in one sentence.')}")
except Exception as e:
print(f"[{name}] ERROR: {e}")
Streaming Wrapper
def chat_stream(name: str, prompt: str):
p = PROVIDERS[name]
stream = client_for(name).chat.completions.create(
model=p.model,
messages=[{"role": "user", "content": prompt}],
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield delta
for piece in chat_stream("groq", "Write a poem about autumn."):
print(piece, end="", flush=True)
print()
Async Concurrency
For batch jobs, AsyncOpenAI lets you fan out across providers:
import asyncio
from openai import AsyncOpenAI
async def achat(name, prompt):
p = PROVIDERS[name]
client = AsyncOpenAI(api_key=os.environ[p.api_key_env], base_url=p.base_url)
r = await client.chat.completions.create(
model=p.model,
messages=[{"role":"user","content":prompt}],
max_tokens=64,
)
return name, r.choices[0].message.content
async def main():
results = await asyncio.gather(*[achat(n, "hi") for n in PROVIDERS])
for name, text in results:
print(f"[{name}] {text}")
asyncio.run(main())
Selection Strategy
- Lowest latency: Groq (LPU)
- Cheapest: OpenRouter
:free models
- Best Chinese: DeepSeek
- On-prem: NIM container
- Multi-provider failover: OpenRouter's
fallback field
Failover and Degradation
In production, combine providers: use a stable paid provider (OpenAI/Anthropic) on the main path, with several free providers as backups that activate on failure. OpenRouter's fallback field gives zero-code failover; alternatively wrap calls in try/except for manual switching. Use asyncio.Semaphore to cap concurrency in async mode; for batch processing, push requests into a queue and have workers consume at each provider's rate limit. For cross-region workloads, overseas providers see higher latency variance at peak hours — reuse connections and route via a local proxy.
Streaming Wrapper
All providers support OpenAI-style streaming responses, so one generator wraps them all. The snippet below turns streamed tokens into a Python generator, so the business layer stays provider-agnostic:
def chat_stream(name: str, prompt: str):
p = PROVIDERS[name]
stream = client_for(name).chat.completions.create(
model=p.model,
messages=[{"role": "user", "content": prompt}],
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield delta
for piece in chat_stream("groq", "Write a poem about autumn."):
print(piece, end="", flush=True)
print()
This single generator can power SSE push, CLI typewriter effects, and more.
Troubleshooting
401: The env var for the provider is unset or misspelled. Check api_key_env.model not found: Model IDs differ across platforms. Trust each platform's /v1/models response.- Auto fallback: Wrap calls in try/except and switch to the next provider on failure.
- Token accounting differs: Providers vary on whether system prompts count toward cache. Normalize the baseline before comparing costs.
Master this pattern and switching models becomes a one-line change.
Best Practices
- Abstract a
make_client(provider) factory: return OpenAI(base_url, api_key) uniformly; business code stays provider-agnostic.
- Name env vars per provider:
GROQ_API_KEY, TOGETHER_API_KEY, DEEPSEEK_API_KEY, etc. — separate.
base_url must end with /v1: missing the /v1 returns 404; all compatibility layers follow OpenAI path conventions.- Normalize error codes: 429 means different things across providers (RPM vs TPM); the gateway must normalize.