简介面向 Android 端接入 Kimi 大模型流式对话能力的实战工程与配套素材适合已具备 Kotlin 与 Gradle 基础、希望实现逐字返回SSE 流式输出效果的移动端开发者参考。压缩包共 813 个文件、约 14.99MBdex、class、jar、kotlin_module 等构成可运行的编译产物与依赖库kt、java、gradle、properties 保留源码与构建配置xml、flat、webp 对应界面布局与打包后的资源json、bin、txt 多为接口返回样例与说明另含 apk 安装包与 bat 脚本目录层次贴近真实 Android 工程。已有 1233 人学习下载。读者可从中了解请求参数组织、分块数据的解析与拼接、UI 增量刷新及异常中断处理等实现细节也能把它作为调试与二次开发的起点对照排查流式响应中常见的粘包、编码错乱与回调时序问题。1. KIMI API 流式输出在 Android 上卡住的往往是最后 200 毫秒有个细节很反直觉把 KIMI API 从非流式改成streamtrueAndroid 客户端首字延迟能从十几秒掉到一秒以内但真正让人返工的从来不是首字——是首字之后那几秒。文字一卡一顿、后半段莫名丢字、切到后台还在闷头烧 token、日志里明明有data:行 UI 却一片空白。这些都不是模型的问题是协议帧、读超时和 UI 刷新节奏三者没对齐。这里拆的是一套可直接复用的 KIMI API 流式输出接入实现重点是 Android 侧从 HTTP 分块读取到StateFlow驱动增量渲染的完整链路以及那些只有真跑过一遍才会遇到的边界。适合已经能发出普通请求、想把对话体验做顺的 Android 开发者只写过一次性请求、没碰过 SSE 的也能跟着走因为每一步都给了可对照的代码和参数。2. 请求参数与 SSE 帧格式streamtrue 的返回究竟长什么样写客户端之前必须先把服务端的行为钉死否则后面每一处异常你都会怀疑是自己代码写错了。KIMI API 的对话接口走的是 OpenAI 兼容协议打开流式的开关只有一个字段stream: true但打开之后返回体的传输形态完全变了Content-Type从application/json变成text/event-stream响应不再是一个完整的 JSON 对象而是一串以空行分隔的文本帧连接会一直挂着直到服务端主动发终止标记。2.1 最小可用参数集哪些字段会影响流的形态先说清楚请求体里每个字段在流式场景下的作用。stream是开关model决定用哪个模型写死在代码里迟早要改抽成常量或配置项messages是完整上下文注意流式不改变多轮拼装规则历史消息还是要自己带全temperature和top_p影响生成随机性对流式只影响内容不影响帧结构max_tokens会直接影响流的长度设得太小会出现finish_reason为length的截断很多人误以为是网络断了。还有一个字段值得单独提如果你需要精确的 token 结算常见做法是在请求体里带上用量统计开关不同接入点字段名以文档为准让最后一条帧把 usage 一起吐出来否则流式过程中你只能自己估算。这一步不做后面做成本核算会很被动。# 先在终端把服务端行为钉死再动客户端代码-N 是关键关掉 curl 自身的输出缓冲 curl -N \ -X POST $KIMI_BASE_URL/v1/chat/completions \ -H Authorization: Bearer $KIMI_API_KEY \ -H Content-Type: application/json \ -H Accept: text/event-stream \ -d { model: $KIMI_MODEL, stream: true, temperature: 0.3, max_tokens: 1024, messages: [ {role: system, content: 你是一个简洁的技术助手}, {role: user, content: 用三句话说明 SSE 的分帧规则} ] }这段命令里最容易被忽略的是-N。不加它curl 自己会缓冲输出你会看到整段内容憋几秒后「唰」地全出来然后误判成服务端没有真正流式转头去怀疑参数写错了——这类翻车我见过不止一次。Accept头不是强制的但加上以后中间层比如某些网关更不容易把响应改成一次性返回。$KIMI_BASE_URL按官方文档给出的接入点填不要硬编码在客户端里方便灰度换环境。2.2 帧结构拆解data 行、空行分隔与终止符流式返回的每一帧都以data:开头后跟一段 JSON帧与帧之间用一个空行分隔最后一条是固定字符串data: [DONE]。一个典型的帧序列长这样data: {id:chunk-1,object:chat.completion.chunk,choices:[{index:0,delta:{role:assistant,content:},finish_reason:null}]} data: {id:chunk-1,object:chat.completion.chunk,choices:[{index:0,delta:{content:SSE},finish_reason:null}]} data: {id:chunk-1,object:chat.completion.chunk,choices:[{index:0,delta:{content: 以空行},finish_reason:null}]} data: {id:chunk-1,object:chat.completion.chunk,choices:[{index:0,delta:{},finish_reason:stop}]} data: [DONE]注意第一帧delta里只有rolecontent是空字符串。这意味着你的 UI 在第一帧收到的「内容」是零长度如果代码里判断「收到内容就新建气泡」会建出一个空气泡。正确的判断条件应该是「角色帧只用来初始化消息对象内容帧才触发渲染」。帧内字段含义客户端是否必须处理choices[0].delta.role首帧标识说话方通常只出现一次是用于创建消息实体choices[0].delta.content本帧新增的文本增量不是全量是累积拼接choices[0].finish_reasonstop正常结束、length被max_tokens截断是决定结束态文案choices为空数组携带用量统计的收尾帧无正文是必须判空data: [DONE]流终止符不是 JSON是收到即关闭连接最后一行「choices为空数组」是最典型的空指针来源。很多人写response.choices[0].delta.content在没有正文的收尾帧上直接崩掉崩溃日志还指向一个看起来完全正常的网络回调。2.3 为什么不能整包解析分块传输与粘包边界HTTP 的流式响应走的是分块传输TCP 层给你的数据块边界和 SSE 的帧边界没有任何对应关系。服务端可能把三帧塞进一个 TCP 包里给你也可能把一帧拆成两半、从一个中文字符的 UTF-8 三字节中间切开。所以任何「收到一段 bytes 就 JSON 解析」的写法必然会在某个时刻炸掉而且因为拆包位置随机表现为时好时坏特别像玄学。正确的解析模型只有一个把响应体当成字节流按行边界\n切分攒够一整行再处理。行内是data:开头才解析 JSON是空行就跳过是[DONE]就收尾。这个模型能天然处理粘包和拆包因为行边界本身就是协议规定的帧边界。\r\n的情况由读行函数处理掉不要自己split(\n)。2.4 用日志把服务端行为固化成基线在写第一行业务代码之前我一般会把上面那条 curl 的输出重定向到文件跑三次一次正常短回答、一次触发max_tokens截断、一次故意传错 key。三次日志分别对应正常结束、finish_reasonlength、HTTP 401 三种形态把这三份日志留在项目里客户端解析器写完之后直接拿它们做单元测试输入比连真实接口调试快得多也省 token。这一步还有个附带好处你能亲眼看到错误响应在流式模式下往往不是 SSE 格式而是一个普通 JSON 错误体状态码也不是 200。所以客户端在进入读行循环之前必须先判response.isSuccessful否则解析器会在一个错误 JSON 上反复报解析异常把真正的 401 盖掉。3. Android 接入骨架OkHttp 逐行读取 Kotlin Flow 落地协议层清楚了Android 侧要解决的就三件事用支持流式的 HTTP 客户端发请求、把响应体按行读出来、把读到的帧变成可取消的数据流。Retrofit 在这件事上不是不能做但它的转换器默认是「等响应体读完再反序列化」正好和流式相反所以这一步我倾向直接用 OkHttp 的ResponseBody.source()可控性最好。3.1 依赖与请求构建流式场景下的 OkHttp 配置差异依赖只有 OkHttp 和 kotlinx-coroutines 两样序列化用kotlinx-serialization-json或者干脆用buildJsonObject手拼——请求体字段就那么几个手拼反而比反射序列化更容易在日志里一眼看懂排查时省事。关键差异在客户端配置。默认的OkHttpClient读超时是 10 秒指的是「两次读到数据之间的最大间隔」。流式生成时模型在思考、在长句中间、在工具调用前后都可能出现超过 10 秒的空档一旦超时连接就被掐断表现是文字打到一半突然停住。这不是服务端断的是你自己的客户端断的。// 流式专用 client超时策略和普通接口完全不同不要复用同一个实例 private val streamClient OkHttpClient.Builder() .connectTimeout(15, TimeUnit.SECONDS) // 建连仍然要设防止卡在握手 .writeTimeout(15, TimeUnit.SECONDS) // 请求体很小15s 足够 .readTimeout(0, TimeUnit.SECONDS) // 0 不设读超时空闲检测交给上层 .retryOnConnectionFailure(false) // 流式请求禁止在传输层自动重试 .build() // 请求体字段少手拼 常量抽取日志里可读性最好 private fun buildBody(prompt: String, model: String): String buildJsonObject { put(model, model) put(stream, true) put(temperature, 0.3) put(max_tokens, 1024) putJsonArray(messages) { addJsonObject { put(role, user) put(content, prompt) } } }.toString()三个参数值得单独解释。readTimeout(0)在 OkHttp 里表示不超时很多团队不敢这么设怕连接变僵尸。实际做法是保留一个上层的心跳检测记录最后一次收到帧的时间超过 60 秒没有任何帧就主动cancel()这样既不会误杀正常的长思考也不会让连接无限挂着。retryOnConnectionFailure(false)更要紧——默认开启时OkHttp 在某些连接异常下会重发请求而流式请求重发意味着同一段内容被生成两次用户看到的可能是重复的开头账单上也是双份。3.2 用 callbackFlow 把 SSE 读成 Flow读循环必须跑在 IO 线程call.execute()是阻塞调用不能放主线程。把它包成Flow的好处是取消能自动传导上层collect的作用域一结束awaitClose里call.cancel()立刻断开连接。fun streamChat(prompt: String): FlowStreamEvent callbackFlow { val request Request.Builder() .url($baseUrl/v1/chat/completions) .addHeader(Authorization, Bearer $apiKey) .addHeader(Accept, text/event-stream) .post(buildBody(prompt, model).toRequestBody(application/json.toMediaType())) .build() val call streamClient.newCall(request) val worker Thread { try { call.execute().use { response - // 错误响应通常不是 SSE 格式必须先分流否则解析器会被错误 JSON 带偏 if (!response.isSuccessful) { trySend(StreamEvent.Failed(response.code, response.body?.string().orEmpty())) returnuse } val source response.body?.source() ?: returnuse // 逐行读这是整个链路的命门绝不能用 body.string() while (!source.exhausted()) { val line source.readUtf8Line() ?: break parseLine(line)?.let { trySend(it) } } } } catch (e: IOException) { trySend(StreamEvent.Failed(-1, e.message.orEmpty())) } finally { close() // 正常结束或被取消都要关闭 channel否则 collect 永远挂着 } }.apply { start() } awaitClose { call.cancel() // 页面销毁立刻断连避免后台继续生成、继续计费 worker.interrupt() // 中断阻塞的读循环 } }response.body?.source()返回的是BufferedSource它和响应体共享同一个底层流读多少拿多少不会一次性把整个响应拉进内存。对比一下response.body?.string()——那一行代码会把连接读到底才返回等于主动把流式退化成非流式而且因为服务端连接一直开着它会一直阻塞到生成结束UI 上表现为「转圈很久然后一次性出全文」。这是流式接入里最常见的一个错误。trySend而不是send是因为send在缓冲区满时会挂起而这里跑在普通线程里没有协程上下文。缓冲区大小默认 64对逐字返回的帧来说偏小长回答时可能出现丢弃稳妥做法是给callbackFlow显式指定channelFlow的缓冲区或者在上层用buffer(256)。这一条不设偶发丢字会非常难查。3.3 解析层的三个纯函数把 data 行变成可测对象把解析逻辑抽成不依赖 Android 和网络的纯函数是整个实现里性价比最高的一步。有了它前面留的三份 curl 日志可以直接当测试用例不用连网就能覆盖[DONE]、空行、截断帧、错误帧四种情况。// 输入一行原始文本输出事件返回 null 表示这行不需要处理 internal fun parseLine(raw: String): StreamEvent? { if (raw.isEmpty()) return null // 空行是帧分隔符不是数据 if (!raw.startsWith(data:)) return null // 注释行和其他 SSE 字段一律跳过 val payload raw.removePrefix(data:).trim() // 标准格式是 data: 冒号后可能带空格 if (payload [DONE]) return StreamEvent.Done return try { val obj json.parseToJsonElement(payload).jsonObject val choices obj[choices]?.jsonArray // 收尾帧 choices 可能为空数组直接取 [0] 必崩 if (choices null || choices.isEmpty()) { val usage obj[usage] return if (usage ! null) StreamEvent.Usage(usage.toString()) else null } val choice choices[0].jsonObject val delta choice[delta]?.jsonObject val content delta?.get(content)?.jsonPrimitive?.contentOrNull val finish choice[finish_reason]?.jsonPrimitive?.contentOrNull when { !content.isNullOrEmpty() - StreamEvent.Delta(content) finish ! null - StreamEvent.Finish(finish) else - StreamEvent.RoleOnly // 首帧只有 role用来建实体 } } catch (e: Exception) { StreamEvent.Malformed(payload) // 不抛降级成可观测事件 } }removePrefix(data:).trim()这一行里的trim()不能省有的中间层会在冒号后补一个空格有的不补trim()两种情况都吃得下。Malformed分支不抛异常而是转成事件是因为流式场景下偶发的半帧不应该中断整轮对话把它投递到监控里比让 App 崩掉划算得多。3.4 取消链路与超时兜底awaitClose里的call.cancel()是资源不泄漏的关键。Android 上最常见的泄漏形态是用户在生成中途退出页面ViewModel销毁了但call还在跑服务端继续生成、账单继续走回到页面再发一次请求两个流同时往一个 UI 写。把streamChat放在viewModelScope里collect作用域销毁时callbackFlow被取消awaitClose触发cancel()断开 socket整条链路才真正停下来。超时兜底单独做一层用SystemClock.elapsedRealtime()记录最后一个有效帧的时间戳每收到一个Delta就刷新配合一个每 5 秒执行一次的检查超过 60 秒没新帧就主动取消并提示「响应超时可重试」。这层逻辑不能放在 OkHttp 的readTimeout里做因为那个值是面向单次读的设太短会误杀长思考设成 0 又没有保护——两件事分开边界才清楚。4. 增量渲染与状态机把 delta 拼成不闪烁的对话流读到了帧、也解析成了事件接下来是 UI。这一步的坑不在网络层而在刷新频率和状态设计。一个中文长回答可能有几百个Delta帧如果每来一帧就notifyDataSetChanged()或者给TextViewsetText 一次主线程会被打得满地找牙列表在滑动时尤其明显。4.1 为什么不能每帧刷新从帧频率到屏幕刷新率模型返回的单帧通常只有一到几个字符一秒内几十帧是常态而屏幕刷新是 60Hz 或 120Hz。也就是说大部分帧在视觉上根本来不及呈现白白消耗了主线程的布局测量时间。人眼对文字出现的感知阈值比这更宽50 毫秒合并一次视觉上已经是连续打字的观感性能开销却降了一个数量级。所以正确的模型是「累积 节流」所有Delta先写进一个可变缓冲区UI 只按固定节奏读缓冲区的快照。这里有个经典错误是用conflate()或sample()直接作用在FlowString上——conflate会丢弃中间值而对增量文本来说丢一帧就是丢字最终得到的是一句残缺的回答。合并不是丢弃是攒起来一次性刷。// 累积缓冲 时间窗口合并注意这是「累加」不是「替换」 private val pending StringBuilder() private var flushJob: Job? null fun onDelta(text: String) { pending.append(text) // 增量追加绝不覆盖 if (flushJob?.isActive true) return // 窗口内已有待执行的刷新直接返回 flushJob viewModelScope.launch { delay(50) // 合并 50ms 内的所有增量 val snapshot pending.toString() _ui.update { state - state.copy( messages state.messages.toMutableList().also { it[it.lastIndex] it.last().copy(content snapshot) }, phase Phase.Streaming ) } } }if (flushJob?.isActive true) return是节流的核心窗口期内新来的增量只进缓冲区不新建协程。delay(50)之后取快照一次update写进StateFlowCompose或RecyclerView各自按自己的机制做差异渲染只有最后一条消息内容变了其他条目不受影响。4.2 状态机设计四个阶段对应的 UI 与交互一条流式消息至少有四个状态等待首帧、流式生成中、正常结束、异常结束。这四个状态决定了按钮能不能点、要不要显示停止按钮、出错时给不给重试入口。把它们压成一个布尔值isLoading是后面所有交互 bug 的源头。阶段触发条件UI 表现可执行操作Waiting请求已发出未收到首个Delta显示思考占位取消Streaming收到首个非空Delta气泡逐字增长停止生成Finished收到Finish(stop)或[DONE]收尾显示用量重新生成、复制FailedHTTP 非 2xx 或异常中断保留已生成部分 错误提示重试特别注意Failed里的「保留已生成部分」。很多人出错时直接把整条消息删掉用户眼看着写了一半的回答消失体验极差。正确做法是保留已收到的文本在气泡底部追加一行状态文案重试时以「新消息」的方式追加而不是覆写原来那条——覆写会让用户失去对照也没法对比两次回答的差异。4.3 结束态的三种收尾方式Finish事件里带的finish_reason要分开处理。stop是正常结束直接收尾即可。length表示被max_tokens截断这时候文本是完整的句子但意思没说完UI 上要给一个明确的「回答过长已截断」提示否则用户会以为 App 卡了。第三种是没有Finish也没有[DONE]、直接读到了流末尾——这通常是中间层断连应该走异常分支并标记为可重试。用量统计如果开了会以一条choices为空的帧送达位置在[DONE]之前。这条帧里的数字用于成本核算和上下文长度预估注意别把它当成异常帧丢掉也别在它上面触发 UI 刷新——它没有内容变化。4.4 停止按钮的真实语义「停止生成」不是把 UI 停住就完事。点下去要做的动作是取消collect的协程 → 触发awaitClose→call.cancel()→ socket 断开 → 服务端停止生成。如果只停了 UI 不停连接用户点完停止账单还在涨这是最冤的一种浪费。同时要处理一个竞态用户点停止的瞬间可能正好有一帧Delta已经进了缓冲区还没刷。这时候要么把缓冲区里剩下的内容刷完再切结束态内容更完整要么直接丢弃响应更快。我倾向刷完因为那部分内容服务端已经生成了、已经计费了丢掉纯亏。5. 避坑与排查五条流式对话翻车记录下面五条都是实际跑起来才会遇到的问题按「现象 → 原因 → 解决」记能省掉不少靠猜的时间。排查顺序建议从连接层往数据层走先看有没有连上、有没有按行读再看帧有没有解析对最后才怀疑 UI。5.1 连接层一次性出全文与固定时长断流现象一日志里能看到多条data:行但 UI 一直空白几秒后整段文字突然全部出现。原因读响应体时用了response.body?.string()或者经过了一个会把响应体整体缓冲的拦截器。string()的语义是读完整个流再返回流式连接又不会主动关闭于是它一直阻塞到生成结束。解决改成response.body?.source()拿BufferedSource用readUtf8Line()逐行读。顺手检查一下有没有装日志拦截器——很多日志拦截器会把 body 打出来副作用同样是读完整个流调试期看着正常上线去掉日志才「突然好了」这类问题最费时间。现象二每次回答到第 10 秒左右就断长短完全固定换个问题也一样。原因readTimeout用了默认值 10 秒。模型在长句中偶尔会有超过 10 秒的思考空档读超时一到连接被客户端自己掐断服务端那边还在正常生成。解决流式专用 client 设readTimeout(0)另做一层基于「最后一帧时间戳」的空闲检测阈值放到 60 秒。两套超时各管各的别混在一起。5.2 数据层乱码、重复与空数组崩溃现象三中文偶尔出现半个字或者一行里混进一个乱码符号。原因按字节读或者按固定长度读从多字节字符中间切开。UTF-8 下一个汉字占三个字节切成[1,2]就解不出来。解决永远按行读用readUtf8Line()而不是readByteArray(n)。行边界天然落在字符边界上不会切开字符。现象四回答里出现重复的片段或者最后一句原样重复了一遍。原因有两个常见来源。一是把delta.content当成全量内容处理写成text content而不是text content表现是只有最后几个字留下看起来像内容被覆盖二是在异常重试时把已经渲染的内容清空重来但缓冲区没清新旧内容拼在一起。解决Delta语义永远是增量只追加不替换重试走「新消息」路径不复用旧缓冲区awaitClose和flushJob一起取消避免上一轮的回调写到新一轮的 UI 上。现象五偶发崩溃堆栈指向choices[0]或者用量统计永远拿不到。原因收尾帧的choices是空数组直接下标越界。上层还常常用一个catch把这类异常吞掉导致用量帧被静默丢弃。解决解析器里先判choices.isNullOrEmpty()为空时检查有没有usage字段并单独投递Usage事件没有就返回null。这条分支最好写成单元测试用例因为它只在每次回答的末尾触发一次手工测试很容易漏。5.3 生命周期层后台继续生成与重复订阅现象退出页面后返回发现消息内容比离开时多了一大截或者同一个问题被回答了两遍。原因collect用的作用域不是viewModelScope而是某个更长命的作用域或者awaitClose里漏了call.cancel()Flow取消了但 socket 还活着。解决统一用viewModelScope.launch { streamChat(prompt).collect { ... } }页面销毁自动取消awaitClose里call.cancel()和worker.interrupt()都要有如果用了repeatOnLifecycle注意它在STOPPED时会取消订阅回到前台重新订阅会重发请求流式场景下应该在 ViewModel 里持有Job手动管理而不是靠生命周期自动重订。提示排查这类问题时先在awaitClose里打一行日志。日志没出现说明取消根本没传导下去问题在作用域日志出现了但服务端还在生成问题在call.cancel()有没有真正执行。6. 进阶验收首字延迟、断流恢复与用量结算的量化验证功能跑通只是及格线真正决定这套流式接入能不能上线的是三个可量化的指标首字延迟、断流后的恢复能力、以及一轮对话的实际用量。这三件事都别靠感觉用代码打点。首字延迟的测量点要卡在「请求发出」和「第一个非空Delta到达」之间而不是「第一个帧到达」——首帧通常只有role内容是空的用它算出来的数字会偏乐观几十到几百毫秒。// TTFT 打点只认第一个非空 content首帧 role 不算 val t0 SystemClock.elapsedRealtime() var ttft -1L fun onEvent(event: StreamEvent) { when (event) { is StreamEvent.Delta - { if (ttft 0L) { ttft SystemClock.elapsedRealtime() - t0 Log.d(stream, TTFT${ttft}ms) // 上报到监控按模型/长度分桶 } onDelta(event.text) } is StreamEvent.Finish - Log.d(stream, total${SystemClock.elapsedRealtime() - t0}ms) else - Unit } }把 TTFT 按「模型 输入长度区间 是否首轮」三个维度分桶上报跑上一周你就能看出哪个组合明显拖后腿而不是笼统地觉得「有时候慢」。断流恢复的边界要提前想清楚。流式请求一旦收到过内容就不该在传输层自动重试因为重试意味着重新生成用户会看到一段重复的开头。可行的做法是只保留已收到的文本把这条消息标记为「未完成」由用户手动触发继续如果业务上必须自动续常见做法是把已生成的部分作为assistant消息追加进上下文再发一条「继续」的用户消息让模型接着写。这条路能走通但不保证衔接自然接入方要在产品文案上说清楚别让用户以为是同一段连续输出。验收项判定方式容易出现的假通过首字延迟首个非空Delta时间戳减去请求发出时间用首帧算数字虚低长回答完整性与max_tokens对照检查finish_reason只看有没有文字漏判length截断中途退出退出页面后抓包确认连接已断只看 UI 停了socket 还在用量准确性Usage事件与服务端账单对账用量帧被空数组分支吞掉错误可恢复401、429、超时三种错误各跑一遍只在网络正常时测过用量结算这块有个习惯值得养成在StreamEvent.Usage到达时把数字原样落库不要只在内存里存着。对账的时候按天比一旦发现客户端累计和服务端账单偏差超过 5%基本就是某类帧被解析器吞了——要么是空choices分支写错要么是错误帧走了Malformed分支被丢弃。注意max_tokens与上下文窗口是两个概念。前者限制本轮输出长度后者是输入加输出的总上限。流式场景下超限往往表现为生成到一半直接断流、既没有Finish也没有[DONE]看着像网络问题实际是参数问题。这套东西我前后改了三版第一版死在readTimeout第二版死在缓冲区节流丢字第三版才把用量对账跑平。从那以后我每次接新的流式接口都会先跑一遍「curl 留三份日志 → 纯函数解析器 → 单元测试喂日志 → 再接 UI」这个顺序一步都不跳。跳过哪一步后面就会在那一层多花两倍时间。希望帮到你。本文还有配套的精品资源点击获取
