Python OpenAI SDK 接入指南
使用官方 openai Python 包直连青柠网关,零代码结构改动,支持同步/异步流式输出与函数调用 (Tool Calling)。
pip install openai
1. 同步流式对话 (Streaming)
from openai import OpenAI
client = OpenAI(
api_key="sk-live-your-callai-key",
base_url="https://api.callaiapi.com/v1"
)
response = client.chat.completions.create(
model="claude-3-7-sonnet",
messages=[
{"role": "system", "content": "你是一位资深全栈架构师。"},
{"role": "user", "content": "请用 Python 演示并解释快速排序算法。"}
],
stream=True
)
for chunk in response:
content = chunk.choices[0].delta.content or ""
print(content, end="", flush=True)2. 异步高并发与长思考超时保护 (AsyncOpenAI)
import asyncio
from openai import AsyncOpenAI
# 针对 DeepSeek R1 等深度推理模型,显式设置 timeout 为 120 秒
client = AsyncOpenAI(
api_key="sk-live-your-callai-key",
base_url="https://api.callaiapi.com/v1",
timeout=120.0
)
async def main():
stream = await client.chat.completions.create(
model="deepseek-r1",
messages=[{"role": "user", "content": "请推导三维空间中点到平面的最短距离公式"}],
stream=True
)
async for chunk in stream:
delta = chunk.choices[0].delta.content or ""
print(delta, end="", flush=True)
asyncio.run(main())🛡️客户端流式断连保护 (Streaming Abort)
青柠AI 网关原生支持 Web Streams 断连捕获。如果用户在前端关闭网页或在 Python 中中断流(break),网关立即切断与上游的连接并精确停止扣费,杜绝空跑浪费。
一键生成完整可执行的 Python 脚本?
支持切换模型、自定义温度、携带上下文并导出源码。