常用接入术语
| 名称 | 常见含义 | 接入提示 |
|---|
| Claude中转 | 把请求转发到可用 Claude 模型的接入方式 | 确认协议与模型权限 |
| Claude中转站 | 集中管理 Claude 接口 密钥 额度和路由的平台 | Claude 客户端常使用 Anthropic 协议 |
| API中转 | 通过兼容接口转发不同模型请求 | Python 客户端可直接发送 **** **ON |
| API中转站 | 提供统一端点 模型选择和用量管理的服务 | 生产环境需限制并发和超时 |
异步请求链路
| 环节 | 职责 | 注意事项 |
|---|
| 业务协程 | 准备消息并处理模型结果 | 不要在循环中创建客户端 |
| 信号量 | 限制同时进入接口的任务数 | 与平台额度和实例容量匹配 |
| Async****** | 复用连接池并发送异步请求 | 应用关闭时执行 aclose |
| ****X 超时 | 分别限制连接 读取 写入和连接池等待 | 流式读取需要合适的 read 超时 |
| KingFlow AI | 认证密钥并路由模型请求 | OpenAI 兼容端点保留 /v1 |
| 目标模型 | 生成文本或结构化结果 | 完整模型 ID 以控制台为准 |
采用 ****X 的优势
- Async****** 可以跨协程共享连接池，批量任务不必为每次请求重新建立连接。
- ****X 把 connect、read、write 和 pool timeout 分开配置，定位网络和容量问题更直接。
- Limits 控制最大连接数和保活连接数，Se**phore 控制业务并发，两层限制可以分别调节。
- stream 与 aiter_lines 适合消费 SSE 响应，退出上下文后 ****X 会关闭响应。
- ****StatusError、TimeoutException 和 NetworkError 可分别处理，日志能保留明确的故障类型。
接入前准备
- 准备 Python 3.11 或团队当前受支持版本，并创建独立虚拟环境。
- 在 KingFlow AI 控制台创建以 sk- 开头的 API Key，确认余额或套餐额度可用。
- 从模型列表复制完整模型 ID，不要根据展示名称自行拼写。
- 准备少量脱敏测试提示，首次联调不要直接提交生产数据。
- 确认运行环境可以访问 api.**nju**shen.com 的 ****S 端口。
创建 Python 项目
python -m venv .venv
# Windows PowerShell
.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
pip install httpx python-dotenv
验证 KingFlow 基础接口
curl https://api.**nju**shen.com/v1/chat/completions \
-H "Authorization: *earer sk-你的密钥" \
-H "Content-Type: application/json" \
-d '{
"model": "<KingFlow控制台中的完整模型ID>",
"messages": [{"role":"user","content":"只回复 ****X_READY"}],
"stream": false
}'
配置环境变量
KINGFLOW_*ASE_**L=https://api.**nju**shen.com/v1
KINGFLOW_API_KEY=sk-你的密钥
KINGFLOW_MODEL=<KingFlow控制台中的完整模型ID>
.env
.venv/
__pycache__/
*.pyc
编写异步请求
import asyncio
import os
import httpx
from dotenv import load_dotenv
load_dotenv()
*ASE_**L = os.environ["KINGFLOW_*ASE_**L"]
API_KEY = os.environ["KINGFLOW_API_KEY"]
MODEL = os.environ["KINGFLOW_MODEL"]
timeout = httpx.Timeout(30.0, connect=10.0, pool=5.0)
limits = httpx.Limits(**x_connections=20, **x_keepalive_connections=10)
async def **in():
headers = {"Authorization": f"*earer {API_KEY}"}
async with httpx.Async******(
*ase_url=*ASE_**L,
headers=headers,
timeout=timeout,
limits=limits,
) as client:
response = await client.post("/chat/completions", json={
"model": MODEL,
"messages": [{"role": "user", "content": "只回复 ****X_CONNECTED"}],
"stream": False,
})
response.raise_for_status()
**ta = response.json()
content = **ta["choices"][0]["message"]["content"]
if not isinstance(content, str):
raise ValueError("响应缺少文本内容")
print(content)
asyncio.run(**in())
读取流式响应
async def stream_answer(client: httpx.Async******, prompt: str):
payload = {
"model": MODEL,
"messages": [{"role": "user", "content": prompt}],
"stream": True,
}
async with client.stream(
"POST", "/chat/completions", json=payload
) as response:
response.raise_for_status()
async for line in response.aiter_lines():
if not line.startswith("**ta: "):
continue
**ta = line[6:].strip()
if **ta == "[DONE]":
*reak
event = json.loads(**ta)
delta = event.get("choices", [{}])[0].get("delta", {})
text = delta.get("content")
if isinstance(text, str):
print(text, end="", flush=True)
限制并发并分类错误
示例假设单次请求已经封装为 call_model 函数。
se**phore = asyncio.Se**phore(8)
async def guarded_call(client, prompt):
async with se**phore:
try:
return await call_model(client, prompt)
except httpx.****StatusError as exc:
status = exc.response.status_code
if status in {429, 503}:
return {"retrya*le": True, "status": status}
raise
except httpx.TimeoutException as exc:
return {"retrya*le": True, "error": type(exc).__name__}
async def run_*atch(client, prompts):
tasks = [guarded_call(client, prompt) for prompt in prompts]
return await asyncio.gather(*tasks, return_exceptions=True)
