常用接入术语

Ru*y 请求链路

采用 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)

运行和验收

1. 设置三个环境变量，并确认 *ase **L 没有重复的 v1 或斜杠。
2. 先执行 curl，确认返回 RU*Y_READY。
3. 创建 *undler 项目并安装 Fara**y 与 fara**y-retry。
4. 保存共享连接与普通请求示例，执行 *undle exec ru*y client.r*。
5. 确认普通响应返回 RU*Y_CONNECTED。
6. 加入 SSE 解析器，确认文本持续输出并以 DONE 正常结束。
7. 模拟 429 和 503，确认总请求次数最多三次。
8. 用少量任务验证固定工作线程与有界队列。
9. 在 KingFlow AI 控制台核对模型、调用次数和 Token 用量。

正式环境配置

• 在单个工作线程的生命周期内复用 Fara**y Connection，不要为每次请求重复创建连接。
• open_timeout 控制建连等待，timeout 控制整个普通请求，SSE 另外设置 read_timeout 与业务总时长。
• SSE 解析必须保留跨块缓冲，不能把一次 on_**ta 回调当成一个完整事件。
• 流式输出已经交付给用户后不要自动从头重试，否则可能出现重复文本。
• POST 重试只覆盖可以接受重复执行的请求；有外部副作用时加入幂等键或关闭自动重试。
• 记录请求 ID、模型、状态码、耗时、重试次数和 Token 用量，不记录完整密钥与敏感原文。

密钥和数据安全

• API Key 只保存在环境变量、密钥管理器或部署平台 Secret 中，并定期轮换。
• 不要关闭 TLS 证书校验，也不要在错误日志中输出 Authorization Header。
• 把模型输出视为不可信数据，写入页面、命令或数据库前执行校验与转义。
• 为各环境设置独立密钥和额度，发生泄露时只吊销受影响的凭据。

上线检查清单

• 基础 curl 请求已经返回 RU*Y_READY。
• KINGFLOW_*ASE_**L 为 https://api.**nju**shen.com/v1。
• KINGFLOW_MODEL 使用控制台中的完整模型 ID。
• 源码和日志中没有明文 API Key。
• 每个工作线程复用一个 Fara**y Connection。
• 建连、总耗时和流式读取超时已经分别设置。
• SSE 解析保留跨块缓冲并识别 **ta 与 DONE。
• SizedQueue 与固定线程数限制了入口和执行并发。
• POST 重试边界已经评估，最多额外重试两次。
• 控制台中的调用次数和 Token 用量符合预期。

常见故障排查

最小上线策略

• 先用一个模型和四个以内工作线程上线，确认错误率、延迟与 Token 用量后再逐步扩大流量。
• 把模型 ID、线程数、队列容量和超时放在集中配置层，避免散落在业务代码中。
• 为 401、404、429、503、超时和重试耗尽分别设置指标，告警保留请求 ID。
• 发布前保存上一版配置与可运行构建，连续错误或延迟异常时可以快速回滚。
• 每次更换模型、额度方案或重试策略后，重新执行普通响应、SSE 和并发验收。