常用接入术语
| 名称 | 常见含义 | 接入提示 |
|---|
| Claude中转 | 把请求转发到已开通的 Claude 模型 | 先确认协议与模型权限 |
| Claude中转站 | 统一管理密钥 额度 路由和兼容接口 | Claude 客户端可按 Anthropic 协议接入 |
| API中转 | 通过统一接口转发模型请求 | Ktor 可直接发送 **** **ON |
| API中转站 | 提供端点 模型和用量管理的服务 | 正式环境需设置超时 并发和重试边界 |
Kotlin 异步请求链路
| 环节 | 职责 | 注意事项 |
|---|
| 协程作用域 | 调度请求并管理任务生命周期 | 避免无归属的**协程 |
| Ktor ****** | 发送请求并复用底层连接 | 应用级创建并统一关闭 |
| ContentNegotiation | 序列化请求并处理 **ON | 忽略未知字段以适配响应扩展 |
| KingFlow AI | 认证密钥并路由目标模型 | OpenAI 兼容路径保留 /v1 |
| SSE 插件 | 按事件持续消费流式响应 | 识别 **ta 与 DONE |
| Se**phore | 限制同时运行的请求数量 | 门限按实例与额度设置 |
采用协程与 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()
}
