OpenAI SDK 快速接入：Python / Node.js 完整示例

OpenAI SDK 是目前兼容性最广的大模型接入方式。绝大多数中转站和国内模型（DeepSeek、Qwen、GLM 等）都支持 OpenAI 兼容接口，学会这一套 SDK，几乎可以调用所有主流模型。

安装

Python（需要 Python 3.8+）：

pip install openai

Node.js（需要 Node 18+）：

npm install openai

基础对话

最简单的用法：发一条消息，拿到回复。

Python：

from openai import OpenAI

client = OpenAI(
    api_key="sk-你的密钥",
    base_url="https://api.deepseek.com/v1"  # 换成你的中转站或官方地址
)

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一个有帮助的助手。"},
        {"role": "user", "content": "用一句话解释什么是 API"}
    ]
)

print(response.choices[0].message.content)
# 输出示例：API 是应用程序之间通信的标准接口，让不同软件可以互相调用功能。

Node.js：

import OpenAI from "openai";

const client = new OpenAI({
  apiKey: "sk-你的密钥",
  baseURL: "https://api.deepseek.com/v1",
});

const response = await client.chat.completions.create({
  model: "deepseek-chat",
  messages: [
    { role: "system", content: "你是一个有帮助的助手。" },
    { role: "user", content: "用一句话解释什么是 API" },
  ],
});

console.log(response.choices[0].message.content);

流式输出（打字机效果）

流式输出让用户不用等完整回复生成完毕，逐字看到结果，体验更好。

Python：

stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "写一首关于编程的五言绝句"}],
    stream=True
)

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

Node.js：

const stream = await client.chat.completions.create({
  model: "deepseek-chat",
  messages: [{ role: "user", content: "写一首关于编程的五言绝句" }],
  stream: true,
});

for await (const chunk of stream) {
  const content = chunk.choices[0]?.delta?.content;
  if (content) process.stdout.write(content);
}
console.log();

多轮对话

AI 模型本身不记忆历史，需要你把之前的对话内容一起发送。

messages = [
    {"role": "system", "content": "你是一个 Python 编程助手。"}
]

# 第一轮
messages.append({"role": "user", "content": "怎么读取 JSON 文件？"})
resp1 = client.chat.completions.create(model="deepseek-chat", messages=messages)
assistant_msg = resp1.choices[0].message.content
messages.append({"role": "assistant", "content": assistant_msg})
print("助手:", assistant_msg)

# 第二轮（模型能看到上一轮的上下文）
messages.append({"role": "user", "content": "如果文件很大怎么办？"})
resp2 = client.chat.completions.create(model="deepseek-chat", messages=messages)
print("助手:", resp2.choices[0].message.content)

注意：每轮对话都会把完整历史发送给 API，历史越长费用越高。对于长对话，可以定期截断早期消息或做摘要压缩。

图像输入（多模态）

支持 Vision 的模型（如 GPT-4o、Gemini）可以理解图片内容。

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "这张图片里有什么？"},
                {
                    "type": "image_url",
                    "image_url": {"url": "https://example.com/photo.jpg"}
                }
            ]
        }
    ]
)
print(response.choices[0].message.content)

也支持 base64 编码的本地图片：

import base64

with open("screenshot.png", "rb") as f:
    img_base64 = base64.b64encode(f.read()).decode()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "描述这张截图的内容"},
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/png;base64,{img_base64}"}
                }
            ]
        }
    ]
)

Function Calling（工具调用）

让 AI 决定何时调用你定义的函数，实现”AI + 外部工具”的组合。

import json

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        }
    }
]

# 发送带工具定义的请求
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=tools
)

# 检查模型是否要调用工具
message = response.choices[0].message
if message.tool_calls:
    tool_call = message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    print(f"模型想调用: {tool_call.function.name}({args})")
    
    # 你执行实际的函数，把结果返回给模型
    weather_result = "北京，晴，28°C"  # 实际中调用天气 API
    
    # 把工具结果发回模型
    messages = [
        {"role": "user", "content": "北京今天天气怎么样？"},
        message,  # 包含 tool_calls 的助手消息
        {
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": weather_result
        }
    ]
    final = client.chat.completions.create(model="gpt-4o", messages=messages)
    print(final.choices[0].message.content)
    # 输出示例：北京今天天气晴朗，气温 28°C，适合户外活动。

常用参数说明

切换模型

OpenAI SDK 的最大优势：换模型只需改 base_url 和 model，其他代码不变。

# DeepSeek
client = OpenAI(api_key="sk-xxx", base_url="https://api.deepseek.com/v1")
model = "deepseek-chat"

# 通过中转站用 Claude
client = OpenAI(api_key="sk-xxx", base_url="https://中转站地址/v1")
model = "claude-3-5-sonnet-20241022"

# 通过 SiliconFlow 用 Qwen
client = OpenAI(api_key="sk-xxx", base_url="https://api.siliconflow.cn/v1")
model = "Qwen/Qwen2.5-72B-Instruct"

错误处理

生产环境必须处理 API 错误：

from openai import (
    APIConnectionError,
    RateLimitError,
    APIStatusError
)
import time

def chat_with_retry(messages, model="deepseek-chat", max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(
                model=model, messages=messages
            )
        except RateLimitError:
            # 被限速，等待后重试
            wait = 2 ** attempt
            print(f"限速，{wait}s 后重试...")
            time.sleep(wait)
        except APIConnectionError:
            # 网络问题
            print("网络连接失败，重试中...")
            time.sleep(1)
        except APIStatusError as e:
            # 其他 API 错误（如 400、500）
            print(f"API 错误 {e.status_code}: {e.message}")
            raise
    raise Exception("重试次数用尽")

下一步

• 想了解如何在 Cursor 中使用自定义模型？→ 阅读《Cursor 接入自定义 AI 模型》
• 想搭建自动化工作流？→ 阅读《n8n 搭建 AI 自动化工作流》
• 想了解如何保护 API Key？→ 阅读《API Key 安全使用指南》