Android接入Kimi API流式输出：SSE解析与实时打字机效果实战

简介：在移动端集成大模型能力时，请求等待常导致体验卡顿。流式输出技术通过Server-Sent Events（SSE）逐token推送数据，显著降低首字延迟（TTFT），让用户看到即时反馈。OkHttp作为Android主流网络库，可基于Okio逐行读取SSE流，实现增量解析与UI实时刷新。本文结合Kimi API实战，讲解流式请求的鉴权、参数调优、粘包处理、生命周期取消等工程问题，并介绍打字机效果的实现技巧，帮助开发者构建更流畅的AI对话界面。

1. KIMI API流式输出：我为什么放弃“等它一次生成完”

第一次在Android端接入Kimi API，我心里想的是：接口通了不就行了？结果让产品试用了一版非流式对接，用户问一句话，转圈能转十几秒，长回复基本等于卡死。换成KIMI API流式输出之后，首字一两秒就上屏，剩下的内容逐字补全，观感立刻从“加载中”变成“正在打字”。这篇文章是那次改造的完整复盘：流式原理、OkHttp怎么吃下SSE流、参数怎么调、四个实战坑，以及最后把流式输出做成打字机效果的技巧。适合正在给Android App接Kimi API的开发者，照着落地能省不少排查时间。

2. 原理与接口基础：Kimi的SSE响应长什么样

2.1 流式输出为什么快：TTFT与生成节奏

先弄清楚为什么流式输出能“快”。大模型生成是按token来的，一个token大概是几个字符到半个词，模型每次预测一个token，把它送回上下文再做下一步预测。非流式接法会等整段序列生成完，一次性把几千个token粘成一个JSON返回给你；流式则是在每个token或每个小批次生成后立刻推到客户端。影响感知的指标不是“总延迟”，而是“首token时间”（TTFT，Time To First Token）。普通请求可能总耗时8秒，流式可能在0.6秒看到第一个字，剩下7秒是慢慢补完的。对用户来说，转圈从“卡死”变成了“正在输出中”，这个差别直接决定App留不留得住人。

Kimi API走的也是这个逻辑。你在请求体里把stream字段设成true，返回的响应体就从“一个大JSON”变成“一长串SSE事件”。SSE全称Server-Sent Events，是HTTP协议上的一种简单流式推送：服务端保持连接打开，按data:前缀逐段推送文本，客户端一直接收，直到收到结束标志。它和WebSocket的区别是单向的——只有服务端往客户端推，不需要客户端回传消息。做AI对话恰恰只需要这个方向，所以用SSE足够。

SSE在传输层有几个特征值得记住。第一，响应头的Content-Type是text/event-stream，不是常见的application/json；第二，连接是长连接，整个生成过程中连接保持打开，所有数据分块到达；第三，事件之间用空行分隔，每行可以是data:、id:、event:前缀，但对Kimi这种场景，你只需要关心data:。这些特征意味着你用OkHttp、Okio甚至原生HttpURLConnection都能读，不需要WebSocket那样复杂的握手和帧协议。

2.2 Kimi接口的请求要素：鉴权、消息结构、stream开关

Kimi的API在形式上兼容大多数对话模型的惯例：POST一个/v1/chat/completions风格的地址，请求头带Authorization: Bearer <你的Key>，请求体带模型名、消息数组和stream标志。下面是我在模拟项目X中用到的请求体骨架：

{ "model": "kimi-7b-chat-v1", "messages": [ {"role": "system", "content": "你是助理"}, {"role": "user", "content": "写一段400字的文案"} ], "stream": true, "temperature": 0.8, "max_tokens": 2048 }

• model：具体取值要以你在官方控制台创建API Key时页面显示的模型名称为准，不同时期开放的模型标识不完全一样。
• messages：必须按system→user→assistant顺序传，角色信息参与生成。
• stream：false时响应体是一个完整JSON；true时是SSE流。
• temperature：控制随机性，0到1之间，聊天建议0.7～0.9，代码或翻译任务往0.2降。
• max_tokens：单次回复的最大token数，Kimi有上下文上限，不是设了就能一定达到，还要看输入占用。

鉴权头没有太多花活，Authorization: Bearer sk-xxx是标准写法。真正容易踩的是两类错误：一类是Key过期或无效，接口直接返回401，响应体会带Authentication Error字样；另一类是并发超限或余额不足，返回429。流式场景下429可能出现在连接建立时，也可能在生成了几百字后断开，所以UI对错误的处理不能只看首包。如果是非流式，你会拿到类似choices[0].message.content里的一整段文本；改成流式后，同样的内容被拆成多个choices[0].delta.content片段。delta就是“增量”的意思：第一段可能只有一两个字，后面每小段补充几个字，直到服务端发来一个[DONE]事件收尾。理解这个差异后，Android端的接入思路就清楚了：不用等整个响应，只需要把每个delta片段推进UI。

还有一个查看用量的小习惯：非流式响应里会带着usage字段，包含prompt_tokens和completion_tokens；流式响应则把usage放在最后一个chunk里，并且那个chunk往往还带一个finish_reason。如果你需要统计成本，记得从结束事件附近取，不要每次流都单独计费。Kimi的SSE事件是按行分割的，典型的一小段长这样：

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"你好"},"finish_reason":null}]}

连续多行data:拼起来，就是完整的回复。最后会有一行是data: [DONE]。抓包确认了这事之后，我决定用OkHttp来做网络层：它既有同步阻塞接口，也有异步回调，底层还能逐行读响应体，不用引入重量级WebSocket框架。

提示：data:后面的内容是一段JSON，不是纯文本。解析时必须先去掉data:前缀和首尾空白，再喂给JSON解析器，否则会报Expected BEGIN_OBJECT之类的错。

3. Android端接入：OkHttp逐行解析与流式事件分发

3.1 用OkHttp发起流式请求：依赖与超时配置

Android端的网络层用OkHttp是最顺手的。先在build.gradle里加依赖：

implementation 'com.squareup.okhttp3:okhttp:4.12.0' implementation 'com.squareup.okio:okio:3.6.0'

OkHttp内部依赖Okio，Okio负责处理字节流和行分割。版本上4.12.0属于4.x的稳定版，如果项目里已经有更高版本，不建议强行降级，保持依赖一致即可。请求的关键点是readTimeout不能按普通请求那样设成10秒或30秒——流式响应可能持续几十秒，期间token之间的间隔在1到2秒内，设短了会误杀连接。我的习惯是连接超时设15秒，读取超时设0表示不限制，再靠应用层做空闲超时控制。

private val client = OkHttpClient.Builder() .connectTimeout(15, TimeUnit.SECONDS) .writeTimeout(30, TimeUnit.SECONDS) .readTimeout(0, TimeUnit.MILLISECONDS) // 流式场景关闭读取超时 .retryOnConnectionFailure(true) .build()

readTimeout(0, MILLISECONDS)是流式接口的关键配置，防止长回复中途停顿被判超时。代价是如果服务端一直不吐数据，这个连接会一直挂着，所以后面还要在应用层设“首包等待超时”和“整体看门狗”，这节放第4章讲。另外提醒一点：Retrofit默认的OkHttpCall会把整个响应体读进内存再交给Converter，不适合直接做流式解析，除非你自定义Converter拿到ResponseBody再逐行读。所以从零接入我建议直接用OkHttp，不要多套一层Retrofit增加理解成本。

3.2 逐行解析SSE事件：Okio的readUtf8Line

下面是我在模拟项目X中写的核心解析代码。通过response.body?.source()拿到Okio的BufferedSource，然后用readUtf8Line()一行行读，只处理以data:开头的行：

client.newCall(request).enqueue(object : Callback { override fun onFailure(call: Call, e: IOException) { viewModel.onError(e.message ?: "网络错误") } override fun onResponse(call: Call, response: Response) { if (!response.isSuccessful) { viewModel.onError("HTTP ${response.code}") return } val source = response.body?.source() ?: return try { while (!source.exhausted()) { val line = source.readUtf8Line() ?: break if (!line.startsWith("data:")) continue val payload = line.removePrefix("data:").trim() if (payload == "[DONE]") { viewModel.onDone() break } val delta = extractDelta(payload) if (delta.isNotBlank()) { viewModel.onDelta(delta) } } } catch (e: IOException) { viewModel.onError(e.message ?: "读取中断") } finally { response.close() } } })

代码逻辑分四步：先判断HTTP状态码，响应成功才继续；再逐行读取，Okio的readUtf8Line()会把换行前的字节转成字符串；接着只挑data:行，去掉前缀后先判断结束标志；最后把每段的JSON解析出delta.content，抛给ViewModel去更新UI。这里有一个细节：不要用response.body?.string()，它会等整个流读完才返回，那就退化成了非流式。

extractDelta是一个纯函数，负责从单个SSE数据块里抽出增量文本。我用的是org.json解析，不引第三方库：

private fun extractDelta(payload: String): String { return try { val obj = JSONObject(payload) val choices = obj.optJSONArray("choices") ?: return "" val choice = choices.optJSONObject(0) ?: return "" val delta = choice.optJSONObject("delta") ?: return "" delta.optString("content", "") } catch (e: JSONException) { "" } }

注意optJSONObject和optString这类带opt前缀的方法，解析失败不会抛异常，而是返回空值，适合在脏数据场景兜底。但如果SSE的一行被网络代理或负载均衡切成两半，按行解析还是会翻车——这种情况我会把连续data:行拼成一个完整JSON再做解析，后文避坑章会说。

3.3 用EventSource库减少手动解析

OkHttp官方生态里有个okhttp-sse扩展模块，把SSE的粘包、多行data拼接、事件分发都封装好了。如果你不想自己维护逐行解析逻辑，可以引入它：

implementation 'com.squareup.okhttp3:okhttp-sse:4.12.0'

使用方式很直观，注册一个EventSourceListener：

val request = Request.Builder() .url(KIMI_API_URL) .addHeader("Authorization", "Bearer ${Config.API_KEY}") .addHeader("Content-Type", "application/json") .post(body.toRequestBody("application/json; charset=utf-8".toMediaType())) .build() val factory = EventSources.createFactory(client) val eventSource = factory.newEventSource(request, object : EventSourceListener() { override fun onEvent(eventSource: EventSource, id: String?, type: String?, data: String?) { if (data == "[DONE]") { viewModel.onDone() } else { val delta = extractDelta(data.orEmpty()) if (delta.isNotBlank()) viewModel.onDelta(delta) } } override fun onClosed(eventSource: EventSource) { viewModel.onDone() } override fun onFailure(eventSource: EventSource, t: Throwable?, response: Response?) { viewModel.onError(t?.message ?: "连接失败") } })

这段代码和手动解析相比，最大的收益是省掉了对data:行的理解，onEvent直接拿到已经拼好的data内容。手写解析的好处是零依赖、可控性强；用EventSource的好处是少踩粘包坑。我的建议是：项目里已有OkHttp且不想加依赖，手写够用；如果是从零接，直接上EventSource更稳。另外onClosed和[DONE]可能会出现两次，所以viewModel.onDone()要设计成幂等的，重复调用只更新一次状态，避免连续弹两次结束标记。

4. 参数调优与上下文管理：从8k到128k的取舍

4.1 模型上下文档位与temperature的配合

Kimi API的上下文长度选项通常在8k到128k之间，具体可见档位以控制台实际列出的为准。选长上下文会带来两个代价：一是单次请求的输入token数变大，按token计费时成本上升；二是首token时间变长，因为服务端要处理更多输入。做普通对话，8k或16k足够；做长文档问答、合同分析、代码库解读，才需要上32k以上。下面是我整理的选择参考：

我一般把上下文选型放在产品定位之后决定：模拟项目X的AI助手模块只是日常问答，8k够用，但用户会连续追问，导致messages数组越来越大，所以必须做裁剪。temperature我有意分成三档：闲聊0.9，说明类内容0.6，需要稳定输出的解析任务0.2。流式输出下如果发现回复前后风格飘忽，可以考虑把temperature降到0.7以下。top_p和temperature是两种控制随机性的手段，Kimi接口两者都支持，日常用temperature一个就够了，两个同时调容易互相干扰。

4.2 多轮对话的messages裁剪策略

messages数组把整段对话历史都发给Kimi，越聊越长。不裁剪有两个后果：一是token超限，接口直接报错；二是即使不超限，历史越长，首token越慢，成本越高。裁剪策略我推荐“系统提示词固定 + 最近N轮 + 字符预算”三个组合：

fun buildMessages( systemPrompt: String, history: List<Pair<String, String>>, newQuestion: String, maxInputChars: Int = 4000 ): List<Map<String, String>> { val messages = mutableListOf<Map<String, String>>() messages += mapOf("role" to "system", "content" to systemPrompt) var totalChars = systemPrompt.length for ((q, a) in history.takeLast(10)) { val round = "用户: $q\n助手: $a" if (totalChars + round.length > maxInputChars) break messages += mapOf("role" to "user", "content" to q) messages += mapOf("role" to "assistant", "content" to a) totalChars += round.length } messages += mapOf("role" to "user", "content" to newQuestion) return messages }

takeLast(10)限制了最多带10轮历史，maxInputChars做字符预算，预算超了就丢掉更早的轮次。这是种粗糙的按字符估算token的方式——中文一个字约等于1到2个token，英文一个词约1到2个token。如果你需要精确控制，可以离线跑几次接口，对比输入字符数和返回的usage字段，找出自己场景下的经验比例。要提醒的是：如果产品要求严格命中上下文窗口上限，最好用官方tokenizer在本地做计数，不要只用字符猜。

token计数在流式场景下有点玄学，因为输入侧和输出侧的计费逻辑不完全一样。我在日志里额外打了一条：每次请求前记录buildMessages生成的字符数，响应后记录usage，长期积累就能摸清自己的业务到底什么量级。如果有一天发现messages越剪越多，说明产品的对话轮次设计有问题，该考虑做“摘要压缩历史”，而不是继续堆原始消息。

4.3 流式请求的看门狗与取消策略

服务端长时间没有推送任何数据，原因很多：网络被代理切断、服务端排队、客户端被系统挂起。我采用双层看门狗：第一层是“首包等待”，连接建立后5秒内没有收到第一行data:，就主动cancel；第二层是“包间空闲”，相邻两个数据块间隔超过3秒就视为异常。实现上不叠加OkHttp的readTimeout，而是起一个Handler循环：

private val watchdog = Handler(Looper.getMainLooper()) private val resetRunnable = Runnable { cancelCurrentStream("读取闲置超时") } private fun onPacketReceived() { watchdog.removeCallbacks(resetRunnable) watchdog.postDelayed(resetRunnable, 3000L) }

每收到一个包就重置定时器，包间超过3秒就触发取消。这种方式比改OkHttp全局timeout更精准，因为不同接口的推送速度不一样。取消时调用call.cancel()，然后通过ViewModel通知UI显示“响应中断，可重试”。要特别注意：收到[DONE]后一定要watchdog.removeCallbacks(resetRunnable)，否则正常结束还会被看门狗误杀一次。

重试策略也要克制。我默认最多重试一次，且只在首包之前失败才自动重试；如果是中途断开，不自动重试，而是在UI给一个“重新生成”按钮。因为对话模型靠上下文生成，中途断了再接上，历史messages里已经包含了之前收到的内容，直接重发会导致输出重复。处理办法是把断流前收到的文本暂存，重试时在UI层回填，再让后续增量接着往上追加。

整个流式过程建议用一个状态机管理：Idle、Streaming、Error、Done。Idle状态允许发起请求；进入Streaming后禁用发送按钮，只允许“停止生成”；Error状态展示错误文案并提供重试；Done状态恢复发送并清理光标。状态机看着简单，但能把取消、重试、看门狗这些散落的逻辑收到一处，排查问题只看状态流转日志就够了。

5. Android流式输出的常见问题与排查：四个实测坑

5.1 提前规避的四个高频坑

这四个坑每个都来自真实调试记录，算是给后来者的血泪经验。

第一个坑：UI 15秒不更新，Log里出现SocketTimeoutException。现象是首包发出来后，中间停了几秒，然后整个连接被掐断。原因是OkHttp默认的readTimeout被设成10秒，而长回复中某两个token间隔恰好超过10秒，判断为超时。解决方法是把readTimeout设为0，应用层自己用看门狗控制空闲时间，上面第四章的3秒兜底方案就是为这个设计的。如果你不想改成0，至少也要把它设成60秒以上，给长回复留够余量。

第二个坑：JSON解析到一半抛出MalformedJsonException。现象是日志打印出一段不完整JSON，内容戛然而止。原因是Wi-Fi网关或透明代理把流式响应拆成了更小的TCP分片，data:的一整行被切成两段，按行解析时不完整。解决方法是不要直接按行解析JSON，而是先判断payload是否以{开头且能完整解析；如果解析失败，就把它缓存起来，和下一行拼起来再解析一次。EventSource库内部已经处理了这个粘包逻辑，这也是我推荐用它替代手写逐行解析的原因。

第三个坑：Activity销毁后回调还在跑，导致崩溃。现象是用户退出页面后，日志里蹦出IllegalStateException或空指针。原因是OkHttp的回调线程没有感知生命周期，而回调里更新了已销毁的UI引用。解决思路是三层防守：在Activity的onDestroy或ViewModel的onCleared里调用call.cancel()；所有UI更新通过ViewModel的LiveData/StateFlow下发；生命周期结束后在回调入口检查isActive，不活跃就直接返回。ViewModel的好处是即使Activity重建，回调还能把状态保留到新实例上。

第四个坑：用editText.setText(fullText)刷新，导致打字机效果变闪烁。现象是每次收到delta都重建整个文本，光标跳动，表情符号分裂。原因是setText全量替换导致重绘负载过高。解决方法是只在收到增量时执行append或SpannableStringBuilder.append，并用50毫秒节流合并高频段。这个方法在第6章展开讲。

5.2 从现象反推的排查套路

如果问题不好定位，我习惯从三个方向做二分。先看接收侧：在onEvent入口打印一行标记，确认是客户端没收到数据，还是收到了但UI没更新。这个二分能区分网络问题还是渲染问题。再看传输侧：用抓包工具看TCP流，确认服务端确实推送了连续数据，中间有没有长时间空隙；如果有空隙，回到看门狗策略。最后看解析侧：把收到的原始data字符串单独落盘，手动解析，如果不报错，那问题就在行分割逻辑上。

排查顺序上，我推荐按“日志最可能暴露真相”的优先级来排。下面是顺手整理的一张对照表，方便新手对号入座：

提示：Android 9及以上默认禁止明文HTTP，Kimi的官方接口是HTTPS，正常不会受影响。但如果你们团队在测试环境用HTTP代理抓包，记得在network security config里给代理域名加豁免，否则会一直报“Cleartext HTTP traffic not permitted”。

6. 把流式输出做成打字机效果：增量刷新、自动滚动与节流控制

6.1 增量刷新与节流窗口

流式数据到了ViewModel，最终要落到TextView上。单纯append会触发多次布局和重绘，token密集时每秒可能刷十几次，屏幕会闪。我给模拟项目X做的方案是用Handler做50毫秒节流：把一段时间内的delta拼进缓冲，每50毫秒统一提交一次到UI。

fun onDeltaArrived(delta: String) { pending.append(delta) if (!isFlushScheduled) { isFlushScheduled = true handler.postDelayed(::flush, 50L) } } private fun flush() { binding.content.append(pending.toString()) binding.content.invalidate() binding.scrollView.post { binding.scrollView.fullScroll(View.FOCUS_DOWN) } pending.clear() isFlushScheduled = false }

flush()里做了两件事：把缓冲文本追加到TextView的TextBuffer；然后把ScrollView滚到底部。这里有一个细节：自动滚动一定要post到主线程下一帧再执行，否则TextView的测量高度还没更新，滚动距离算不准，表现为“滚不到底”。另外，如果用户正在往上翻看历史内容，强制滚底部会打断阅读。我会在滚动前检查scrollView是否已经处于底部，只有接近底部时才自动跟随。

打字机效果还有一层是光标仿真。我用的办法是在文本末尾追加一个零宽度的闪动方块字符，配合ValueAnimator控制alpha闪烁。等流式结束后移除光标，同时把max_tokens等生成中的标记清掉。这个细节很便宜但对观感提升很大。

6.2 验证流式是否达标的方法

接完后我一般做一个3项检查。第一项是首包时间：从发请求到第一个可见字符上屏，应该在1到2秒内，如果超过5秒，检查是否误用了非流式接口，或请求体里stream没传true。第二项是断线恢复：飞行模式开关模拟网络中断，确认UI进入可重试状态且不会崩溃。第三项是字数一致性：用非流式接口跑同一段prompt，统计完整输出的字符数，再对流式收到的所有delta做拼接，对比两个结果是否一致。不一致说明有丢包或拼接遗漏，优先查粘包处理。

这三项过完，基本可以放心交付。从那以后我每次接新的SSE接口，都会强制走一遍“格式确认→增量上屏→生命周期取消→飞行模式断网”四步检查，少走弯路。希望帮到你。

本文还有配套的精品资源，点击获取