OpenAI 特别用法
本页面展示通过 OpenAI SDK 调用 Claude Reasoning 等高级功能的特殊用法,包括 Extended Thinking、Adaptive Thinking 以及流式推理等场景。
基础配置
在开始之前,请确保您已经获取了 API Key。如果还没有,请参考创建API Key。
基础信息
- API Base URL:
https://api.agentsflare.com/v1 - 认证方式: Bearer Token
- 内容类型:
application/json
Claude Reasoning 兼容性测试
以下脚本使用 OpenAI SDK 测试中转站是否支持 Claude Reasoning 功能,涵盖模型列表查询、基础请求、Extended Thinking、Adaptive Thinking 以及流式输出等测试。
python
#!/usr/bin/env python3
# ============================================================
# test_claude_reasoning.py
# 使用 OpenAI SDK 测试第三方中转站是否支持 Claude Reasoning
# 依赖: pip install openai
# ============================================================
import os
import json
import sys
from openai import OpenAI
import httpx
# ==================== 配置区域 ====================
BASE_URL = os.environ.get("BASE_URL", "https://api.agentsflare.com/v1")
API_KEY = os.environ.get("API_KEY", "sk-********")
MODEL = os.environ.get("MODEL", "claude-sonnet-4-6")
MAX_TOKENS = int(os.environ.get("MAX_TOKENS", "16384"))
BUDGET_TOKENS= int(os.environ.get("BUDGET_TOKENS", "8000"))
# ==================================================
# ANSI 颜色
GREEN = "\033[92m"
RED = "\033[91m"
YELLOW = "\033[93m"
CYAN = "\033[96m"
RESET = "\033[0m"
def ok(msg): print(f"{GREEN}✅ {msg}{RESET}")
def fail(msg): print(f"{RED}❌ {msg}{RESET}")
def warn(msg): print(f"{YELLOW}⚠️ {msg}{RESET}")
def info(msg): print(f"{CYAN}{msg}{RESET}")
# 初始化 OpenAI 客户端(指向中转站)
client = OpenAI(
api_key=API_KEY,
base_url=BASE_URL,
)
results: dict[str, bool] = {}
# ============================================================
# 工具函数:打印响应摘要
# ============================================================
def print_response_summary(response):
"""打印响应的关键信息"""
print(f" Model : {response.model}")
print(f" Usage : prompt={response.usage.prompt_tokens}, "
f"completion={response.usage.completion_tokens}, "
f"total={response.usage.total_tokens}")
for i, choice in enumerate(response.choices):
msg = choice.message
# 检测是否有 thinking 内容 (Claude 会在 content 里放 thinking block)
if hasattr(msg, "content") and isinstance(msg.content, list):
for block in msg.content:
if hasattr(block, "type"):
if block.type == "thinking":
think_text = getattr(block, "thinking", "")
print(f" [Thinking] : {think_text[:120]}{'...' if len(think_text) > 120 else ''}")
elif block.type == "text":
print(f" [Answer] : {block.text[:200]}{'...' if len(block.text) > 200 else ''}")
else:
content = msg.content or ""
print(f" [Answer] : {content[:200]}{'...' if len(content) > 200 else ''}")
# ============================================================
# 测试 0:查询支持的模型列表
# ============================================================
def _parse_models(raw):
"""防御式解析各种可能的 /models 返回格式"""
# 标准 OpenAI 格式: {"object":"list","data":[{"id":"..."}]}
if isinstance(raw, dict):
data = raw.get("data", [])
if isinstance(data, list):
for item in data:
if isinstance(item, dict):
if "id" in item:
yield item["id"]
elif "model" in item: # 某些中转站用 model 字段
yield item["model"]
elif isinstance(item, str):
yield item
# 有些中转站直接返回 JSON 数组
elif isinstance(raw, list):
for item in raw:
if isinstance(item, dict):
if "id" in item:
yield item["id"]
elif "model" in item:
yield item["model"]
elif isinstance(item, str):
yield item
def test_list_models():
info("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
info("[测试 0/4] 查询中转站支持的 Claude 模型列表")
info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
claude_models = []
# 1) 先尝试 OpenAI SDK 的标准方式
try:
models = client.models.list()
if hasattr(models, "data") and isinstance(models.data, list):
claude_models = [m.id for m in models.data
if hasattr(m, "id") and "claude" in m.id.lower()]
else:
# 某些 SDK 版本迭代 models 直接得到对象
claude_models = [m.id for m in models
if hasattr(m, "id") and "claude" in m.id.lower()]
except Exception as sdk_err:
warn(f"SDK 解析失败 ({sdk_err}),尝试直接请求...")
# 2) 回退:直接发 HTTP GET 到 /v1/models
try:
url = BASE_URL.rstrip("/") + "/models"
resp = httpx.get(url, headers={"Authorization": f"Bearer {API_KEY}"}, timeout=30)
resp.raise_for_status()
raw = resp.json()
all_models = list(_parse_models(raw))
claude_models = [m for m in all_models if "claude" in m.lower()]
except Exception as http_err:
fail(f"查询失败: SDK 错误={sdk_err}; HTTP fallback 错误={http_err}")
results["模型列表查询"] = False
return
if claude_models:
ok(f"共找到 {len(claude_models)} 个 Claude 模型:")
for name in sorted(claude_models):
print(f" • {name}")
else:
warn("未找到任何 Claude 模型,请检查中转站是否支持 Claude")
results["模型列表查询"] = True
# ============================================================
# 测试 1:基础请求(不启用 Thinking)
# ============================================================
def test_basic():
info("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
info("[测试 1/4] 基础请求(不启用 Thinking)")
info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
try:
response = client.chat.completions.create(
model=MODEL,
max_tokens=1024,
messages=[
{"role": "user", "content": "你好,请用一句话介绍你自己。"}
],
)
ok(f"基础请求成功")
print_response_summary(response)
results["基础请求"] = True
except Exception as e:
fail(f"基础请求失败: {e}")
results["基础请求"] = False
# ============================================================
# 测试 2:Extended Thinking(type: enabled + budget_tokens)
# 适用于 Claude 3.7 Sonnet / Claude Sonnet 4.5 / Claude Opus 4.5
# ============================================================
def test_extended_thinking():
info("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
info("[测试 2/4] Extended Thinking(type: enabled + budget_tokens)")
info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
try:
response = client.chat.completions.create(
model=MODEL,
max_tokens=MAX_TOKENS,
messages=[
{
"role": "user",
"content": (
"请逐步推理:一个农夫有17只羊,除了9只以外都死了,还剩几只活羊?"
),
}
],
# OpenAI SDK 通过 extra_body 传递非标准参数
extra_body={
"thinking": {
"type": "enabled",
"budget_tokens": BUDGET_TOKENS,
}
},
)
ok("Extended Thinking 请求成功")
print_response_summary(response)
results["Extended Thinking"] = True
except Exception as e:
fail(f"Extended Thinking 失败: {e}")
results["Extended Thinking"] = False
# ============================================================
# 测试 3:Adaptive Thinking(type: adaptive,Claude Sonnet/Opus 4.6+)
# ============================================================
def test_adaptive_thinking():
info("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
info("[测试 3/4] Adaptive Thinking(type: adaptive,Claude 4.6+)")
info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
try:
response = client.chat.completions.create(
model=MODEL,
max_tokens=MAX_TOKENS,
messages=[
{
"role": "user",
"content": "请深入思考:如果 3^x + 3^x + 3^x = 3^12,求 x 的值。",
}
],
extra_body={
"thinking": {
"type": "adaptive",
}
},
)
ok("Adaptive Thinking 请求成功")
print_response_summary(response)
results["Adaptive Thinking"] = True
except Exception as e:
warn(f"Adaptive Thinking 失败(可能模型不支持): {e}")
results["Adaptive Thinking"] = False
# ============================================================
# 测试 4:Streaming + Extended Thinking(流式 + 推理)
# ============================================================
def test_streaming_thinking():
info("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
info("[测试 4/4] Streaming + Extended Thinking(流式推理)")
info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
try:
stream = client.chat.completions.create(
model=MODEL,
max_tokens=MAX_TOKENS,
stream=True,
messages=[
{
"role": "user",
"content": "请简单解释什么是量子纠缠?",
}
],
extra_body={
"thinking": {
"type": "enabled",
"budget_tokens": BUDGET_TOKENS,
}
},
)
ok("流式请求已建立,开始接收数据...")
print(" [Stream Output]:", end=" ", flush=True)
char_count = 0
for chunk in stream:
delta = chunk.choices[0].delta if chunk.choices else None
if delta and delta.content:
print(delta.content, end="", flush=True)
char_count += len(delta.content)
if char_count > 300: # 只预览前 300 字符
print(" ...(截断预览)", flush=True)
break
else:
print() # 换行
ok(f"Streaming + Thinking 完成")
results["Streaming + Thinking"] = True
except Exception as e:
fail(f"Streaming + Thinking 失败: {e}")
results["Streaming + Thinking"] = False
# ============================================================
# 汇总报告
# ============================================================
def print_summary():
info("\n╔══════════════════════════════════════════╗")
info("║ 📊 测试结果汇总 ║")
info("╚══════════════════════════════════════════╝")
passed = sum(1 for v in results.values() if v)
total = len(results)
for name, status in results.items():
mark = f"{GREEN}✅ 通过{RESET}" if status else f"{RED}❌ 失败{RESET}"
print(f" {name:<24} {mark}")
print()
if passed == total:
ok(f"全部通过 {passed}/{total} — 中转站完整支持 Claude Reasoning 🎉")
elif passed >= 2:
warn(f"部分通过 {passed}/{total} — 中转站支持基础 Claude 功能,Reasoning 支持不完整")
else:
fail(f"大部分失败 {passed}/{total} — 请检查 BASE_URL / API_KEY / MODEL 配置")
print(f"""
{YELLOW}📌 配置信息:{RESET}
BASE_URL = {BASE_URL}
MODEL = {MODEL}
MAX_TOKENS = {MAX_TOKENS}
BUDGET_TOKENS = {BUDGET_TOKENS}
{YELLOW}📌 说明:{RESET}
• Extended Thinking (type:enabled) → Claude 3.7 / 4.5 系列
• Adaptive Thinking (type:adaptive) → Claude 4.6+ 系列
• OpenAI SDK 通过 extra_body 传递 thinking 参数
""")
# ============================================================
# 主入口
# ============================================================
if __name__ == "__main__":
info("╔══════════════════════════════════════════╗")
info("║ Claude Reasoning 中转站兼容性测试 (Python) ║")
info("╚══════════════════════════════════════════╝")
info(f"🔗 API 地址 : {BASE_URL}")
info(f"🤖 模型名称 : {MODEL}")
info(f"🧠 思考预算 : {BUDGET_TOKENS} tokens")
test_list_models()
test_basic()
test_extended_thinking()
test_adaptive_thinking()
test_streaming_thinking()
print_summary()关键参数说明
| 参数 | 类型 | 说明 |
|---|---|---|
extra_body | dict | OpenAI SDK 中用于传递非标准参数的字段 |
thinking.type | string | enabled(Claude 3.7 / 4.5 系列)或 adaptive(Claude 4.6+ 系列) |
thinking.budget_tokens | int | 为推理过程分配的最大 token 数量 |
支持的模型
以下 Claude 模型可通过本方式调用 Reasoning 功能:
claude-sonnet-4-6Newclaude-opus-4-6claude-sonnet-4-5claude-opus-4-5claude-3-7-sonnet
💡 提示
- 通过
extra_body传递thinking参数是 OpenAI SDK 调用 Claude Reasoning 的推荐方式 adaptive模式仅适用于 Claude 4.6 及以上版本- 流式输出时,thinking 内容块可能与普通内容块交错出现