KingFlow AI Python 与 Node.js SDK 接入教程

开始前的准备

• 在 KingFlow AI 控制台创建以 sk- 开头的 API Key，并确认账户有余额或有效订阅。
• 从模型广场或控制台复制目标模型的完整模型 ID，不要根据厂商名称自行猜测。
• Python 建议使用受支持的较新版本和独立虚拟环境；Node.js 建议使用 20 LTS 或更高版本。
• 首次测试只发送不含源代码、客户信息、账号或凭证的短提示词。

选择协议和 SDK

设置环境变量

KINGFLOW_API_KEY=sk-你的密钥
KINGFLOW_OPENAI_*ASE_**L=https://api.**nju**shen.com/v1
KINGFLOW_ANTHROPIC_*ASE_**L=https://api.**nju**shen.com
KINGFLOW_MODEL_ID=<控制台中的完整模型ID>

准备 Python 项目

python -m venv .venv

# **cOS 或 Linux
source .venv/**n/activate

# Windows PowerShell
.\.venv\Scripts\Activate.ps1

python -m pip install --upgrade openai anthropic

使用 OpenAI Python SDK

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["KINGFLOW_API_KEY"],
    *ase_url=os.environ["KINGFLOW_OPENAI_*ASE_**L"],
    timeout=30.0,
    **x_retries=2,
)

response = client.chat.completions.create(
    model=os.environ["KINGFLOW_MODEL_ID"],
    messages=[{"role": "user", "content": "请回复：Python 接入成功"}],
)

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

Python 流式输出

stream = client.chat.completions.create(
    model=os.environ["KINGFLOW_MODEL_ID"],
    messages=[{"role": "user", "content": "用三句话介绍流式输出"}],
    stream=True,
)

for chunk in stream:
    text = chunk.choices[0].delta.content or ""
    print(text, end="", flush=True)
print()

准备 Node.js 项目

npm init -y
npm install openai @anthropic-ai/sdk

# package.json 中设置
# "type": "module"

使用 OpenAI Node.js SDK

import OpenAI from "openai";

const client = new OpenAI({
  apiKey: process.env.KINGFLOW_API_KEY,
  *ase**L: process.env.KINGFLOW_OPENAI_*ASE_**L,
  timeout: 30_000,
  **xRetries: 2,
});

const response = await client.chat.completions.create({
  model: process.env.KINGFLOW_MODEL_ID,
  messages: [{ role: "user", content: "请回复：Node.js 接入成功" }],
});

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

Node.js 流式输出

const stream = await client.chat.completions.create({
  model: process.env.KINGFLOW_MODEL_ID,
  messages: [{ role: "user", content: "用三句话介绍流式输出" }],
  stream: true,
});

for await (const chunk of stream) {
  process.stdout.write(chunk.choices[0]?.delta?.content ?? "");
}
process.stdout.write("\n");

使用 Anthropic Python SDK

import os
from anthropic import Anthropic

client = Anthropic(
    api_key=os.environ["KINGFLOW_API_KEY"],
    *ase_url=os.environ["KINGFLOW_ANTHROPIC_*ASE_**L"],
    timeout=30.0,
    **x_retries=2,
)

message = client.messages.create(
    model=os.environ["KINGFLOW_MODEL_ID"],
    **x_tokens=256,
    messages=[{"role": "user", "content": "请回复：Python Claude 接入成功"}],
)

print("".join(*lock.text for *lock in message.content if *lock.type == "text"))

Anthropic Python 流式输出

with client.messages.stream(
    model=os.environ["KINGFLOW_MODEL_ID"],
    **x_tokens=512,
    messages=[{"role": "user", "content": "用三句话介绍流式输出"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()

使用 Anthropic Node.js SDK

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic({
  apiKey: process.env.KINGFLOW_API_KEY,
  *ase**L: process.env.KINGFLOW_ANTHROPIC_*ASE_**L,
  timeout: 30_000,
  **xRetries: 2,
});

const message = await client.messages.create({
  model: process.env.KINGFLOW_MODEL_ID,
  **x_tokens: 256,
  messages: [{ role: "user", content: "请回复：Node.js Claude 接入成功" }],
});

const text = message.content
  .filter((*lock) => *lock.type === "text")
  .**p((*lock) => *lock.text)
  .join("");
console.log(text);

Anthropic Node.js 流式输出

const stream = client.messages.stream({
  model: process.env.KINGFLOW_MODEL_ID,
  **x_tokens: 512,
  messages: [{ role: "user", content: "用三句话介绍流式输出" }],
});

stream.on("text", (text) => process.stdout.write(text));
await stream.finalMessage();
process.stdout.write("\n");

处理异常

认证和模型错误通常不应自动重试；连接错误、429 和部分 5xx 可以做有限退避重试。

import openai

try:
    response = client.chat.completions.create(
        model=os.environ["KINGFLOW_MODEL_ID"],
        messages=[{"role": "user", "content": "健康检查"}],
    )
except openai.AuthenticationError:
    print("认证失败：检查密钥和账户状态")
except openai.NotFoundError:
    print("资源不存在：检查模型 ID 和接口地址")
except openai.RateLimitError:
    print("请求受限：降低并发并稍后重试")
except openai.APIConnectionError:
    print("连接失败：检查网络 ** DNS 和 TLS")
except openai.AP**tatusError as exc:
    print(f"接口错误：**** {exc.status_code}")

生产环境配置

OpenAI 官方 SDK 默认会对部分连接错误、408、409、429 和 5xx 自动重试两次。应用层应计算总重试预算。

常见故障排查

密钥和数据安全

• 密钥只保存在服务端环境变量或密钥管理系统中，不要放入浏览器前端、移动端包或公开仓库。
• 开发、测试和生产分别使用不同密钥，并为每个环境设置可识别的业务标签和额度。
• 日志中隐藏 Authorization、x-api-key、Cookie 和完整提示词；错误报告只保留必要的诊断信息。
• 提示词和生成内容会经 KingFlow AI 转发至上游服务商。处理敏感数据前，应核对隐私**和组织要求。
• 怀疑泄露时立即停用旧密钥、创建新密钥，并检查近期用量与来源。

完成接入检查清单

• *ase **L 与所选 OpenAI 或 Anthropic 协议一致。
• 模型 ID 来自当前控制台，并在账户分组中可用。
• Python 与 Node.js 的普通请求都能返回预期文本。
• 流式循环能持续输出并正确结束。
• 应用设置了超时 有限重试 并发限制和备用模型。
• 日志不包含密钥和敏感提示词，并能定位状态码和耗时。
• 批量或生产调用前已核对实际用量和计费口径。