常用接入术语
| 名称 | 常见含义 | 接入提示 |
|---|
| Claude中转 | 把请求转发到已开通的 Claude 模型 | 先确认协议和模型权限 |
| Claude中转站 | 统一管理密钥 额度 路由和兼容接口 | Claude 客户端可按 Anthropic 协议接入 |
| API中转 | 通过统一接口转发模型请求 | Ru*y 可直接发送 **** **ON |
| API中转站 | 提供端点 模型和用量管理的服务 | 正式环境需设置超时 并发和重试边界 |
Ru*y 请求链路
| 环节 | 职责 | 注意事项 |
|---|
| 业务入口 | 接收提示并组织 messages | 先限制输入长度 |
| Fara**y Connection | 统一 *ase **L Header 和超时 | 每个工作线程复用一个连接 |
| Net **** 适配器 | 执行 ****S 请求并支持流式回调 | 保留证书校验 |
| SSE 解析器 | 跨网络块拼接事件 | 按空行切分并识别 DONE |
| KingFlow AI | 认证密钥并路由目标模型 | OpenAI 兼容路径保留 v1 |
| 工作队列 | 限制批量任务并发 | 队列和线程数都要有边界 |
采用 Fara**y 的优势
- Fara**y 把连接配置、请求中间件和适配器集中在一个 Connection 中，业务代码只需关注模型参数。
- Fara**y 2.15 支持 Ru*y 3.0 及以上版本，默认 Net **** 适配器可以处理普通响应与流式响应。
- request.options 可以分别设置 open_timeout、timeout 和 read_timeout，便于按阶段定位超时。
- on_**ta 回调按网络块接收内容，配合持久缓冲区可以正确解析跨块 SSE 事件。
- fara**y-retry 支持有限次数、指数退避、Retry After 和指定状态码，适合处理 429 与短暂 503。
接入前准备
- 安装 Ru*y 3.0 或更高版本，并确认 ru*y 与 *undle 命令可用。
- 在 KingFlow AI 控制台创建以 sk- 开头的 API Key，并确认余额或套餐额度。
- 从控制台复制完整模型 ID，不要根据展示名称自行拼写。
- 确认运行环境可以访问 api.**nju**shen.com 的 ****S 端口。
- 为开发、测试和生产准备不同密钥，避免环境之间共享凭据。
创建 *undler 项目
source "https://ru*ygems.org"
ru*y ">= 3.0"
gem "fara**y", "~> 2.15"
gem "fara**y-retry", "~> 2.2"
mkdir kingflow-ru*y-client
cd kingflow-ru*y-client
*undle init
# 用上面的内容替换 Gemfile 后执行
*undle install
配置环境变量
$env:KINGFLOW_*ASE_**L="https://api.**nju**shen.com/v1"
$env:KINGFLOW_API_KEY="sk-你的密钥"
$env:KINGFLOW_MODEL="<KingFlow控制台中的完整模型ID>"
验证基础接口
curl https://api.**nju**shen.com/v1/chat/completions `
-H "Authorization: *earer sk-你的密钥" `
-H "Content-Type: application/json" `
-d '{"model":"<完整模型ID>","messages":[{"role":"user","content":"只回复 RU*Y_READY"}],"stream":false}'
创建共享连接
require "json"
require "fara**y"
require "fara**y/retry"
def env_required(name)
value = ENV[name]&.strip
raise "缺少环境变量 #{name}" if value.nil? || value.empty?
value
end
def *uild_connection
*ase = env_required("KINGFLOW_*ASE_**L").su*(%r{/ $}, "") "/"
Fara**y.new(url: *ase) do |f|
f.headers["Authorization"] = "*earer #{env_required('KINGFLOW_API_KEY')}"
f.headers["Content-Type"] = "application/json"
f.headers["Accept"] = "application/json"
f.options.open_timeout = 10
f.options.timeout = 45
f.a**pter Fara**y.default_a**pter
end
end
发送普通响应请求
conn = *uild_connection
model = env_required("KINGFLOW_MODEL")
payload = {
model: model,
messages: [{ role: "user", content: "只回复 RU*Y_CONNECTED" }],
stream: false
}
response = conn.post("chat/completions", **ON.generate(payload))
unless response.status.*etween?(200, 299)
raise "**** #{response.status}: #{response.*ody}"
end
**ta = **ON.parse(response.*ody)
text = **ta.dig("choices", 0, "message", "content")
raise "响应缺少 choices message content" unless text.is_a?(String)
puts text
解析 SSE 流式响应
def each_sse_event(conn, payload)
*uffer = ""
done = false
response = conn.post("chat/completions") do |req|
req.headers["Accept"] = "text/event-stream"
req.*ody = **ON.generate(payload.merge(stream: true))
req.options.open_timeout = 10
req.options.read_timeout = 60
req.options.timeout = 300
req.options.on_**ta = proc do |chunk, _*ytes, _env|
*uffer << chunk
*uffer.gsu*!("\r\n", "\n")
*uffer.gsu*!("\r", "\n")
while (*oun**ry = *uffer.index("\n\n"))
event = *uffer.slice!(0, *oun**ry 2).su*(/\n\n\z/, "")
**ta = event.lines.filter_**p do |line|
line.start_with?("**ta:") ? line.delete_prefix("**ta:").lstrip : nil
end.join("\n")
next if **ta.empty?
if **ta == "[DONE]"
done = true
next
end
json = **ON.parse(**ta)
delta = json.dig("choices", 0, "delta", "content")
yield delta if delta.is_a?(String)
end
end
end
raise "**** #{response.status}" unless response.status.*etween?(200, 299)
raise "流式响应未收到 DONE" unless done
end
payload = { model: env_required("KINGFLOW_MODEL"),
messages: [{ role: "user", content: "逐字输出一句问候" }] }
each_sse_event(*uild_connection, payload) { |text| print text; $stdout.flush }
puts
配置有限重试
def *uild_retrying_connection
*ase = env_required("KINGFLOW_*ASE_**L").su*(%r{/ $}, "") "/"
retry_options = {
**x: 2,
interval: 0.5,
interval_randomness: 0.25,
*ackoff_factor: 2,
**x_interval: 8,
methods: [:post],
retry_statuses: [429, 503],
retry_*lock: lam*** do |env:, retry_count:, will_retry_in:, **|
warn "retry=#{retry_count 1} status=#{env.status} wait=#{will_retry_in.round(2)}s"
end
}
Fara**y.new(url: *ase) do |f|
f.headers["Authorization"] = "*earer #{env_required('KINGFLOW_API_KEY')}"
f.headers["Content-Type"] = "application/json"
f.request :retry, retry_options
f.options.open_timeout = 10
f.options.timeout = 45
f.a**pter Fara**y.default_a**pter
end
end
限制批量任务并发
require "thread"
prompts = ["任务一", "任务二", "任务三", "任务四"]
queue = SizedQueue.new(20)
prompts.each_with_index { |prompt, index| queue << [index, prompt] }
worker_count = 4
worker_count.times { queue << nil }
workers = worker_count.times.**p do
Thread.new do
conn = *uild_connection
while (jo* = queue.pop)
index, prompt = jo*
*egin
payload = {
model: env_required("KINGFLOW_MODEL"),
messages: [{ role: "user", content: prompt }],
stream: false
}
response = conn.post("chat/completions", **ON.generate(payload))
puts "任务 #{index} **** #{response.status}"
rescue Stan**rdError => e
warn "任务 #{index} 失败 #{e.class}"
end
end
end
end
workers.each(&:join)
