Python 调用示例

本页面提供 Python 调用 51API 的完整示例，从简单到复杂，满足不同需求。

环境准备

安装依赖

pip

# 推荐：使用 OpenAI 官方 SDK
pip install openai

# 或者使用 requests 库
pip install requests

conda

conda install -c conda-forge openai

poetry

poetry add openai

配置 API Key

推荐使用环境变量存储 API Key：

Linux / macOS

export NEWAPI_API_KEY="sk-你的密钥"

Windows CMD

set NEWAPI_API_KEY=sk-你的密钥

Windows PowerShell

$env:NEWAPI_API_KEY="sk-你的密钥"

---

方式一：使用 OpenAI SDK（推荐）

这是最简单的方式，代码简洁，功能完整。

基础对话

import os
from openai import OpenAI

# 初始化客户端
client = OpenAI(
    api_key=os.getenv("NEWAPI_API_KEY"),  # 从环境变量获取
    base_url="https://www.51api.org/v1"    # 必须包含 /v1
)

# 发送请求
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个有帮助的助手。"},
        {"role": "user", "content": "你好！"}
    ]
)

# 打印回复
print(response.choices[0].message.content)

流式输出

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.getenv("NEWAPI_API_KEY"),
    base_url="https://www.51api.org/v1"
)

# 开启流式输出
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "讲一个简短的故事"}],
    stream=True
)

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

多轮对话

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.getenv("NEWAPI_API_KEY"),
    base_url="https://www.51api.org/v1"
)

# 保存对话历史
messages = [
    {"role": "system", "content": "你是一个友好的助手。"}
]

def chat(user_input):
    # 添加用户消息
    messages.append({"role": "user", "content": user_input})
    
    # 发送请求
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    
    # 获取回复
    reply = response.choices[0].message.content
    
    # 保存到历史
    messages.append({"role": "assistant", "content": reply})
    
    return reply

# 多轮对话示例
print("AI:", chat("我叫小明"))
print("AI:", chat("我刚才说我叫什么？"))

---

方式二：使用 requests 库

适合需要更多自定义控制的场景。

基础请求

import requests
import os

API_KEY = os.getenv("NEWAPI_API_KEY")
API_URL = "https://www.51api.org/v1/chat/completions"

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {API_KEY}"
}

data = {
    "model": "gpt-4o",
    "messages": [
        {"role": "user", "content": "你好！"}
    ]
}

response = requests.post(API_URL, headers=headers, json=data)

if response.status_code == 200:
    result = response.json()
    print(result["choices"][0]["message"]["content"])
else:
    print(f"请求失败: {response.status_code}")
    print(response.text)

流式请求

import requests
import json
import os

API_KEY = os.getenv("NEWAPI_API_KEY")
API_URL = "https://www.51api.org/v1/chat/completions"

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {API_KEY}"
}

data = {
    "model": "gpt-4o",
    "messages": [{"role": "user", "content": "讲个笑话"}],
    "stream": True
}

response = requests.post(API_URL, headers=headers, json=data, stream=True)

print("AI: ", end="")
for line in response.iter_lines(decode_unicode=True):
    if line and line.startswith("data:"):
        data_str = line[5:].strip()
        if data_str == "[DONE]":
            break
        try:
            chunk = json.loads(data_str)
            content = chunk["choices"][0]["delta"].get("content", "")
            print(content, end="", flush=True)
        except json.JSONDecodeError:
            pass
print()

---

高级示例：带重试和保存

适用于生产环境，包含超时控制、自动重试、响应保存等功能。

💡 适用场景

• 调用 DeepSearch、Thinking 等长耗时模型
• 需要保存 AI 回复到文件
• 需要处理网络不稳定的情况

import os
import json
import time
import random
import requests

# ==================== 配置区域 ====================
API_TOKEN = os.getenv("NEWAPI_API_KEY")  # 或直接填写 "sk-xxx"
API_BASE_URL = "https://www.51api.org/v1/chat/completions"
MODEL_NAME = "gpt-4o"  # 可替换为其他模型
PROMPT = "你好，请介绍一下你自己"
SAVE_DIR = "./response_txt"

# 超时设置（秒）
# DeepSearch / Thinking 模型可能需要更长时间，建议设置 600 秒以上
TIMEOUT = (30, 600)  # (连接超时, 读取超时)

# 重试配置
MAX_RETRIES = 2
RETRY_SLEEP_BASE = 2  # 退避时间：2, 4, 8...


# ==================== 工具函数 ====================
def extract_text_from_chunk(data_str: str) -> str:
    """从 SSE 数据块中提取文本内容"""
    try:
        obj = json.loads(data_str)
        choice = obj.get("choices", [{}])[0]
        
        # 流式响应
        delta = choice.get("delta", {})
        if delta.get("content"):
            return delta["content"]
        
        # 非流式响应
        message = choice.get("message", {})
        if message.get("content"):
            return message["content"]
    except (json.JSONDecodeError, KeyError, IndexError):
        pass
    return ""


# ==================== 主程序 ====================
def main():
    if not API_TOKEN:
        raise RuntimeError("请设置 NEWAPI_API_KEY 环境变量")

    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {API_TOKEN}",
    }

    payload = {
        "model": MODEL_NAME,
        "messages": [{"role": "user", "content": PROMPT}],
        "stream": True,
    }

    # 创建保存目录
    os.makedirs(SAVE_DIR, exist_ok=True)
    unique_id = f"{int(time.time())}{random.randint(100, 999)}"
    response_dir = os.path.join(SAVE_DIR, unique_id)
    os.makedirs(response_dir, exist_ok=True)

    raw_file = os.path.join(response_dir, "raw_stream.txt")
    text_file = os.path.join(response_dir, "text.txt")

    start_time = time.time()
    print("=" * 50)
    print(f"模型: {MODEL_NAME}")
    print(f"提问: {PROMPT}")
    print("=" * 50)

    last_error = None

    for attempt in range(MAX_RETRIES + 1):
        try:
            with requests.post(
                API_BASE_URL,
                headers=headers,
                json=payload,
                stream=True,
                timeout=TIMEOUT,
            ) as resp:
                print(f"状态码: {resp.status_code}")

                if resp.status_code != 200:
                    print(f"错误: {resp.text[:500]}")
                    break

                resp.encoding = "utf-8"
                full_text = []

                print("\nAI: ", end="")
                
                with open(raw_file, "w", encoding="utf-8") as f_raw, \
                     open(text_file, "w", encoding="utf-8") as f_text:
                    
                    for line in resp.iter_lines(decode_unicode=True):
                        if not line:
                            continue

                        # 保存原始数据
                        f_raw.write(line + "\n")
                        f_raw.flush()

                        # 解析 SSE 格式
                        if line.startswith("data:"):
                            data = line[5:].strip()
                        else:
                            data = line.strip()

                        if data == "[DONE]":
                            break

                        # 提取并显示文本
                        text = extract_text_from_chunk(data)
                        if text:
                            full_text.append(text)
                            f_text.write(text)
                            f_text.flush()
                            print(text, end="", flush=True)

                print("\n")
                print("=" * 50)
                print(f"✅ 完成！响应已保存到: {response_dir}")
                last_error = None
                break

        except requests.exceptions.ReadTimeout:
            last_error = "读取超时"
        except requests.exceptions.ConnectTimeout:
            last_error = "连接超时"
        except requests.exceptions.ConnectionError as e:
            last_error = f"连接错误: {e}"
        except Exception as e:
            last_error = f"未知错误: {e}"

        if attempt < MAX_RETRIES:
            sleep_time = RETRY_SLEEP_BASE * (2 ** attempt)
            print(f"\n⚠️ {last_error}，{sleep_time}秒后重试...")
            time.sleep(sleep_time)
        else:
            print(f"\n❌ 失败: {last_error}")

    elapsed = time.time() - start_time
    print(f"耗时: {elapsed:.2f} 秒")


if __name__ == "__main__":
    main()

功能说明

功能	说明
⏱️ 超时控制	支持分别设置连接超时和读取超时
🔄 自动重试	网络抖动时自动重试，使用指数退避策略
💾 响应保存	同时保存原始 SSE 数据和纯文本
📺 实时输出	边接收边打印，实时显示 AI 回复

---

图片分析示例

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.getenv("NEWAPI_API_KEY"),
    base_url="https://www.51api.org/v1"
)

# 使用图片 URL
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "这张图片里有什么？"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://example.com/image.jpg"
                    }
                }
            ]
        }
    ],
    max_tokens=300
)

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

---

常见问题

报错 "Model not found"

检查你的令牌分组是否正确。详见 常见问题。

连接超时

1. 检查网络连接
2. 增加超时时间
3. 使用代理（如需要）

其他问题

查看 完整常见问题