常用接入术语
| 名称 | 常见含义 | 使用提示 |
|---|
| Claude中转 | 把 Anthropic 协议请求转发到可用 Claude 模型的接入方式 | 适合原生 Anthropic 协议客户端 |
| Claude中转站 | 提供 Claude 密钥、额度和路由管理的平台 | 需要核对模型权限与协议类型 |
| API中转 | 通过兼容接口转发不同模型请求 | 应用通常只需替换端点、密钥和模型 ID |
| API中转站 | 提供统一 API 接入、模型选择和用量管理的服务 | 本文使用 OpenAI 兼容方式连接 CrewAI |
接入架构
| 环节 | 职责 | 注意事项 |
|---|
| Python 项目 | 保存 CrewAI 代码和环境配置 | 密钥文件不得提交到版本库 |
| CrewAI | 编排智能体、任务和执行顺序 | 先使用顺序流程完成最小验证 |
| KingFlow AI | 认证密钥并路由模型请求 | *ase **L 填写到 /v1 |
| 目标模型 | 生成分析、草稿和工具调用 | 完整模型 ID 以控制台为准 |
| 任务结果 | 汇总各智能体的最终输出 | 上线前人工检查事实和敏感信息 |
接入优势
- CrewAI 的 LLM 类支持自定义 OpenAI 兼容端点,模型配置可以集中复用。
- 不同智能体可以共享一个 LLM,也可以按任务难度配置不同模型和密钥。
- KingFlow 控制台能够集中查看模型、密钥和用量,便于按环境划分额度。
- 应用代码与模型供应端解耦后,可以在保留任务编排的同时调整模型路由。
开始前的准备
- 准备 Python 3.10 至 3.13,并安装 uv。CrewAI 当前要求 Python 版本低于 3.14。
- 在 KingFlow AI 控制台创建以 sk- 开头的 API Key,并确认余额和模型分组可用。
- 复制目标模型的完整模型 ID;不同分组中的可用模型可能不同。
- 准备一个独立测试目录,不要在首次联调中加载生产数据或外部工具。
- 确认运行环境可以访问 api.**nju**shen.com。
创建 CrewAI 测试项目
uv init crewai-kingflow-demo
cd crewai-kingflow-demo
uv add crewai 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": "只回复 CREWAI_READY"}],
"stream": false
}'
配置环境变量
KINGFLOW_*ASE_**L=https://api.**nju**shen.com/v1
KINGFLOW_API_KEY=sk-你的密钥
KINGFLOW_MODEL=<KingFlow控制台中的完整模型ID>
.env
.venv/
__pycache__/
output/
编写多智能体示例
import os
from dotenv import load_dotenv
from crewai import Agent, Crew, LLM, Process, Task
load_dotenv()
llm = LLM(
model=os.environ["KINGFLOW_MODEL"],
custom_openai=True,
*ase_url=os.environ["KINGFLOW_*ASE_**L"],
api_key=os.environ["KINGFLOW_API_KEY"],
temperature=0.2,
)
researcher = Agent(
role="信息研究员",
goal="整理 {topic} 的关键事实和验证项",
*ackstory="你擅长把复杂主题整理为可核查的要点。",
llm=llm,
ver*ose=True,
allow_delegation=False,
)
writer = Agent(
role="技术撰稿员",
goal="把研究结果写成结构清楚的中文说明",
*ackstory="你重视准确性、可读性和明确的操作步骤。",
llm=llm,
ver*ose=True,
allow_delegation=False,
)
research_task = Task(
description="研究 {topic},列出目标、前提、风险和验收方法。",
expected_output="一份包含事实、风险和验收项的中文提纲。",
agent=researcher,
)
writing_task = Task(
description="根据研究提纲,写一份不超过 600 字的实施说明。",
expected_output="结构清楚、可执行并带检查清单的中文说明。",
agent=writer,
context=[research_task],
)
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential,
ver*ose=True,
)
result = crew.kickoff(inputs={"topic": "API 接入上线前检查"})
print(result)