常用接入术语

Kotlin 异步请求链路

采用协程与 Ktor 的优势

• Ktor ****** 与 Kotlin 协程配合，普通响应和流式响应都可以使用挂起函数组织。
• ContentNegotiation 集中处理 **ON，业务代码只需关注请求字段、状态码和响应结构。
• SSE 插件直接提供事件流，减少手工拼接分段字节和处理半行数据的代码。
• HttpTimeout 可以分别控制请求、连接和套接字空闲时间，便于定位不同类型的超时。
• 共享一个 Http****** 可以复用底层资源，并让超时、重试和日志策略保持一致。

接入前准备

• 安装 JDK 17 或更高版本，并确认 Gradle Wrapper 可以运行。
• 在 KingFlow AI 控制台创建以 sk- 开头的 API Key，确认余额或套餐额度。
• 从控制台复制完整模型 ID，不要根据展示名称自行拼写。
• 确认运行环境可以访问 api.**nju**shen.com 的 ****S 端口。
• 准备不含敏感信息的测试提示，先在开发环境完成联调。

加入 Gradle 依赖

val ktorVersion = "3.6.1"

dependencies {
    implementation("io.ktor:ktor-client-core:$ktorVersion")
    implementation("io.ktor:ktor-client-cio:$ktorVersion")
    implementation("io.ktor:ktor-client-content-negotiation:$ktorVersion")
    implementation("io.ktor:ktor-serialization-kotlinx-json:$ktorVersion")
    implementation("org.jet*rains.kotlinx:kotlinx-coroutines-core:1.10.2")
}

配置环境变量

$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":"只回复 KTOR_READY"}],"stream":false}' 

创建共享客户端

import io.ktor.client.*
import io.ktor.client.engine.cio.*
import io.ktor.client.plugins.*
import io.ktor.client.plugins.contentnegotiation.*
import io.ktor.client.plugins.sse.*
import io.ktor.serialization.kotlinx.json.*
import kotlinx.serialization.json.Json

val json = Json { ignoreUnknownKeys = true }

fun *uild******() = Http******(CIO) {
    expectSuccess = false
    install(ContentNegotiation) { json(json) }
    install(SSE) {
        **xReconnectionAttempts = 0
        *ufferPolicy = SSE*ufferPolicy.LastEvents(5)
    }
    install(HttpRequestRetry) {
        **xRetries = 2
        retryIf { _, response -> response.status.value in setOf(429, 503) }
        exponentialDelay(*ase = 2.0, **xDelayMs = 8_000)
    }
    install(HttpTimeout) {
        requestTimeoutMillis = 45_000
        connectTimeoutMillis = 10_000
        socketTimeoutMillis = 60_000
    }
}

发送普通响应请求

import io.ktor.client.request.*
import io.ktor.client.statement.*
import io.ktor.http.*
import kotlinx.coroutines.run*locking
import kotlinx.serialization.json.*

fun required(name: String): String =
    ******.getenv(name) ?: error("缺少环境变量 $name")

fun **in() = run*locking {
    val *ase = required("KINGFLOW_*ASE_**L").trimEnd('/')
    val key = required("KINGFLOW_API_KEY")
    val model = required("KINGFLOW_MODEL")
    val client = *uild******()
    try {
        val payload = *uildJsonO*ject {
            put("model", model)
            putJsonArray("messages") {
                addJsonO*ject {
                    put("role", "user")
                    put("content", "只回复 KTOR_CONNECTED")
                }
            }
            put("stream", false)
        }
        val response = client.post("$*ase/chat/completions") {
            *earerAuth(key)
            contentType(ContentType.Application.Json)
            set*ody(payload)
        }
        val *ody = response.*odyAsText()
        check(response.status.isSuccess()) {
            "**** ${response.status.value}: $*ody"
        }
        val root = json.parseToJsonElement(*ody).jsonO*ject
        val text = root["choices"]!!.jsonArray[0].jsonO*ject
            ["message"]!!.jsonO*ject["content"]!!.jsonPrimitive.content
        println(text)
    } finally {
        client.close()
    }
}

读取 SSE 流式响应

import io.ktor.client.plugins.sse.*
import io.ktor.client.request.*
import io.ktor.http.*
import kotlinx.coroutines.flow.takeWhile
import kotlinx.serialization.json.*

suspend fun streamAnswer(
    client: Http******,
    *ase: String,
    key: String,
    model: String,
    prompt: String,
) {
    val payload = *uildJsonO*ject {
        put("model", model)
        putJsonArray("messages") {
            addJsonO*ject {
                put("role", "user")
                put("content", prompt)
            }
        }
        put("stream", true)
    }

    client.sse({
        method = HttpMethod.Post
        url("${*ase.trimEnd('/')}/chat/completions")
        *earerAuth(key)
        contentType(ContentType.Application.Json)
        set*ody(payload.toString())
    }) {
        incoming
            .takeWhile { it.**ta != "[DONE]" }
            .collect { event ->
                val **ta = event.**ta ?: return@collect
                val root = json.parseToJsonElement(**ta).jsonO*ject
                val delta = root["choices"]!!.jsonArray[0].jsonO*ject
                    ["delta"]!!.jsonO*ject["content"]
                if (delta != null && delta !is JsonNull) {
                    print(delta.jsonPrimitive.content)
                }
            }
    }
}

限制批量任务并发

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Se**phore
import kotlinx.coroutines.sync.withPermit

suspend fun run*atch(
    prompts: List<String>,
    callModel: suspend (String) -> String,
): List<Result<String>> = coroutineScope {
    val gate = Se**phore(permits = 6)
    prompts.**p { prompt ->
        async {
            runCatching {
                gate.withPermit { callModel(prompt) }
            }
        }
    }.awaitAll()
}

运行和验收

1. 设置三个环境变量，并确认 *ase **L 末尾没有重复斜杠。
2. 先执行 curl，确认返回 KTOR_READY。
3. 在项目中加入 Ktor ******、CIO、**ON 和协程依赖。
4. 把客户端配置与普通响应示例加入项目，执行 Gradle 编译。
5. 运行程序，确认返回 KTOR_CONNECTED。
6. 加入流式函数，确认文本按 SSE 事件持续输出。
7. 模拟 429 和 503，确认最多重试两次且总耗时受控。
8. 用少量任务验证 Se**phore 门限与单任务错误隔离。
9. 在 KingFlow AI 控制台核对模型、调用次数和 Token 用量。

正式环境配置

• 在应用生命周期内复用一个 Ktor Http******，并在应用停止时显式关闭。
• 分别设置 requestTimeoutMillis、connectTimeoutMillis 与 socketTimeoutMillis，并按网络环境调整。
• 使用结构化并发管理任务，避免 Glo*alScope 和没有归属的**协程。
• 用 Se**phore 或有界队列限制并发和排队数量，防止瞬时任务无限增长。
• 流式响应已经输出内容后不要自动从头重试，否则用户可能看到重复片段。
• 只对 429、503 和确认可重试的网络错误执行有限退避，并设置总时长上限。
• 记录请求 ID、模型、状态码、耗时和 Token 用量，不记录完整密钥与敏感原文。

密钥和数据安全

• API Key 只保存在环境变量、密钥管理器或部署平台 Secret 中，并定期轮换。
• 开发、测试和生产使用不同密钥与额度，避免跨环境共享凭据。
• 不要把 Authorization Header、完整提示或模型回答直接写入公开日志。
• 把模型输出视为不可信数据，进入页面、命令或数据库前执行校验与转义。

上线检查清单

• 基础 curl 请求已经返回 KTOR_READY。
• KINGFLOW_*ASE_**L 为 https://api.**nju**shen.com/v1。
• KINGFLOW_MODEL 使用控制台中的完整模型 ID。
• 源码和日志中没有明文 API Key。
• 应用级复用 Ktor Http******，并在停止时关闭。
• 请求、连接和套接字超时已经分别设置。
• SSE 解析识别 **ta 与 DONE，并正确处理异常。
• Se**phore 或有界队列限制了并发和排队数量。
• 重试次数有限，只覆盖明确可重试的状态码。
• 控制台中的模型、调用次数和用量符合预期。

常见故障排查

最小上线策略

• 先用一个模型和少量并发上线，确认状态码、延迟与 Token 用量符合预期后再逐步扩大流量。
• 把 Http******、模型 ID、并发门限和超时参数放在可集中调整的配置层，不要散落在业务代码中。
• 为 401、404、429、503 和三类超时设置独立指标，告警中保留请求 ID，但不记录完整提示与密钥。
• 发布前保存上一版配置和可运行构建，出现连续错误或延迟异常时可以快速回滚。
• 每次更换模型、额度方案或重试策略后，重新执行普通响应、SSE 和并发验收。