简介面向 Android 开发者的 Kimi API 流式输出项目参考包聚焦在移动端调用 Moonshot AI 的 Kimi 对话接口时如何处理流式返回结果。内容涵盖完整的 Android 工程结构Kotlin/Java 源码展示接口请求与回调解析逻辑xml/json 用于布局与数据传输配置gradle、properties 标记工程依赖与构建参数同时包含 dex/jar/class 编译产物及可安装 apk便于直接对照源码查看流式输出在真实调用中的组织方式。压缩包共 813 个文件大小约 14.99MB文件数量多、类型较杂适合需要参考 Android 端大模型流式交互实现、或想快速梳理 Kimi API 接入流程的初中级开发者。已有 1230 人学习下载说明该示例对同类需求具备一定参考价值。通过学习可获取一条完整的调用路径从网络请求建立、流式数据增量接收到界面层状态刷新与最终文本展示并结合工程内各类配置与编译文件进一步还原项目构建与运行细节。1. KIMI API 流式输出让聊天机器人先开口再思考KIMI API 流式输出就是在调用 Kimi 大模型接口时把 stream 参数打开让模型边生成边把 token 推送回来而不是等整段回答想完再一次性返回。做对话机器人的开发者基本都遇到过这种场面非流式接口要干等十几秒页面白屏用户以为服务挂了改成流式后一两秒内第一个字就出来后面像打字机一样持续渲染。这套机制也是 kimi 网页版聊天体验的底层来源。下面从 SSE 协议、Python 手动解析、前后端联动讲到高频报错排查适合正在对接 Kimi API、或用 Spring AI 搭聊天后端的开发者照着落地。2. 流式输出的协议与最小调用SSE 推流怎么工作2.1 流式与一次性返回差的不只是首字延迟非流式调用下服务端要等整轮生成完毕把完整回答包在一个 JSON 里返回。模型生成 800 个 token、按每秒 50 token 算就是 16 秒这 16 秒里调用方只能干等用户那边就是白屏。把 stream 置为 true 之后服务端边生成边按帧推送首 token 延迟通常压到 1 秒以内体感完全是两个产品。流式还有一个常被忽略的好处中途止损。用户看到前三句方向不对点一下停止连接断开后面没生成的 token 就不会继续计费。对多轮问答、长文写作这类生成时间长的产品这是实打实的成本节约。另外要注意首 token 延迟会随服务端排队情况波动高峰期请求量大时前面几帧变慢很正常接口侧也有对应的优先级策略被测出慢时先确认是不是队列问题。非流式也不是一无是处。它的响应结构完整、不容易出现半截内容适合批量总结、离线抽取这类没人盯着看的任务流式则几乎只为人机实时交互而生。做选择时先想清楚用户到底看不看生成过程别为了看起来高级给后台任务也开流式徒增拆包断流这些麻烦。2.2 SSE 协议一次 HTTP 长连接上的持续写入Kimi API 兼容 OpenAI 的 Chat Completions 约定流式响应走的是 SSEServer-Sent Events。它本质上就是一次普通 HTTP 响应Content-Type 是 text/event-stream服务端不关闭连接持续写入以 data: 开头的文本帧帧与帧之间用空行分隔。SSE 和 WebSocket 有本质区别SSE 是单向的、不需要协议升级握手、不维护连接状态浏览器原生 EventSource 就能读。对后端开发者而言SSE 意味着不需要引入新的连接层一个生成器加一个响应头就够。一帧的原始形态长这样data: {id:chatcmpl-xxx,choices:[{index:0,delta:{content:你},finish_reason:null}]} data: [DONE]两条铁律data: 冒号后面带一个空格再接内容最后一帧是字面量 [DONE]不是 JSON。这两条手动解析时会反复踩坑后面专门讲。有人试图用轮询模拟流式每 500ms 拉一次完整结果那等于每轮都带着全部上下文重新请求token 成本翻倍还拿不到逐字效果。这套 SSE 解析方式不止 Kimi 在用DeepSeek 这类同样兼容 OpenAI 接口的服务也是同一套约定学会一次到处通用。2.3 用 OpenAI SDK 跑通最小流式调用因为协议兼容最省事的接入方式是直接用 openai 的 Python SDK把 base_url 指到 Kimi 服务地址其余用法和调用 OpenAI 完全一致。from openai import OpenAI client OpenAI( api_keysk-你的密钥, base_urlhttps://api.moonshot.cn/v1, ) stream client.chat.completions.create( modelkimi-k3, messages[ {role: system, content: 你是一个只讲重点的助手。}, {role: user, content: 用三句话解释什么是流式输出}, ], streamTrue, temperature0.3, max_tokens1024, ) for chunk in stream: delta chunk.choices[0].delta if delta and delta.content: print(delta.content, end, flushTrue)streamTrue 之后create() 返回的不再是完整的 ChatCompletion 对象而是一个可迭代生成器。每次迭代拿到一个 chunkchunk.choices[0].delta.content 是这一帧新增的文字逐个拼起来就是完整回答。判断整轮结束不能靠没有内容了要看最后一个 chunk 的 finish_reason 是否变成 stop这个字段后面排障还会用到。提示Kimi 开放平台对模型名的开放范围会随账号和时段调整。写死 model 名之前先去控制台确认当前可用模型 id经常有人认证都过了还是报 400其实是模型名没对上。快速验证用 curl 更直接适合写进部署文档当冒烟测试export KIMI_API_KEYsk-你的密钥 curl https://api.moonshot.cn/v1/chat/completions \ -H Authorization: Bearer $KIMI_API_KEY \ -H Content-Type: application/json \ -d {model:kimi-k3,messages:[{role:user,content:你好}],stream:true}看到终端里一行行 data: 帧推出来就说明密钥、地址、模型名三个前置条件全对。这一步能省掉后面排障时一半的猜谜时间。参数方面temperature 控制采样随机度对话类产品 0.20.4 比较稳max_tokens 是本次回答生成上限流式不改变它的语义到达上限就会截断。两个参数在流式和非流式下行为一致不存在开了流式就不能设的说法。如果想要流式结束时的用量统计可以给 create() 加 stream_options{include_usage: True}SDK 版本太老的话升级后即可识别。3. 不依赖 SDK 手动解析流式帧requests 逐行读与参数对照3.1 最小实现requests 发起流式请求SDK 方便但生产环境常有不得不用原生的理由内网环境装不了包、要在网关层做自定义转发、要把上游流原样透传。这时候用 requests 就能干。关键在于两个地方post 时带 streamTrue读取时用 iter_lines。import json import requests API_KEY sk-你的密钥 URL https://api.moonshot.cn/v1/chat/completions resp requests.post( URL, headers{ Authorization: fBearer {API_KEY}, Content-Type: application/json, }, json{ model: kimi-k3, messages: [{role: user, content: 用一句话介绍你自己}], stream: True, max_tokens: 2048, }, streamTrue, # 连接保持打开响应体按块返回 timeout(5, 120), # 连接超时 5 秒读间隔超时 120 秒 ) for line in resp.iter_lines(decode_unicodeTrue): if not line: continue if not line.startswith(data:): continue data line[5:].strip() # 去掉 data: 前缀 if data [DONE]: break chunk json.loads(data) delta chunk[choices][0][delta] content delta.get(content) if content: print(content, end, flushTrue)iter_lines 会按行把响应体拆出来SSE 帧之间那个空行恰好就是行分隔。decode_unicodeTrue 让 requests 按响应头里的字符集自动解码避免手写 decode 时把 UTF-8 多字节字符拦腰截断。timeout 传元组时第一个值是连接阶段超时第二个值是每次读操作之间的间隔超时不是总时长。流式场景下模型可能长时间没有输出深度思考模型在推理阶段就是不出字这个读超时宁可给到 30 秒以上不然长难问被自己设计的超时掐断界面卡半截还不好排查。requests 够用如果服务里需要连接复用httpx 的 client.stream 写法更顺解析逻辑与上面完全一致换库时不用改帧处理代码。3.2 帧内容拆解delta、finish_reason 与可选的 usage一个标准帧展开是这样的 JSON{ id: chatcmpl-xxx, object: chat.completion.chunk, choices: [ { index: 0, delta: {content: 你好}, finish_reason: null } ] }三条信息最重要。delta.content 是这一帧新增的文字正常回答过程里基本不为空但代码里不能假设它不是 None。Kimi 的深度思考类模型正式回答前有一段推理过程那个阶段推送的是 reasoning_content 字段content 会短暂为空多轮消息结构里角色切换也会产生一个只有 role 的空 delta。手动解析必须容忍没有 content 的帧直接把这类帧跳过别当异常处理。finish_reason 在中间帧都是 null直到最后出现stop 表示正常结束length 表示触发了 max_tokens 上限content_filter 表示内容被策略层拦截。它是判断这轮回答是否被截断的唯一依据。很多线上问题最后定位到是 length 截断但因为当时没记录 finish_reason排查走了大半天弯路。建议每个流式处理循环里都把这个字段存下来连同拼接结果一起落日志。usage 在纯流式下默认是 null需要显式传 stream_options{include_usage: True}而且携带 usage 的那一帧 delta 为空出现在 [DONE] 之前。解析顺序上要处理这个空帧否则容易在前端渲染一个空片段。3.3 参数怎么设流式下的关键取值对照参数流式下的作用常见取值说明stream是否开启逐帧推送true配合 stream_options 使用temperature采样随机度0.20.7越高越发散代码生成建议偏低max_tokens本次生成最大 token 数5124096到达即截断finish_reasonlengthtop_p核采样阈值0.81.0与 temperature 建议二选一调stream_options.include_usage流尾是否返回 token 用量true计费核对用不加就是 null注意max_tokens 不是上下文长度。上下文长度由模型窗口决定max_tokens 只限生成部分。这两个概念经常被混在一起导致第 5 章那个 context length 报错被误读成max_tokens 设小了。手动解析还有一个隐藏麻烦TCP 层可能把多个 SSE 帧粘在同一个包里也可能把一个帧拆成两个包。iter_lines 按 \n 处理能解决大部分情况但如果你前面挂了代理或网关且它对响应做过缓存重写按行解析就会不稳。稳妥做法是维护一个字符串缓冲区按 \n\n 切帧把残留半帧留到下一轮。这个模式在第 4 章前端解析里还会再用一次属于流式解析的通用底座。如果发现 finish_reason length客户端该提示用户回答被截断而不是默默接受半截内容。4. 前后端联动SSE 转发、浏览器逐字渲染与中断停止4.1 为什么中间要加一层后端生产环境不应该让浏览器直连 Kimi哪怕接口支持跨域也不行。密钥暴露、调用不可审计、无法限流这三个理由足够让任何正经产品加一层自己的后端。后端的职责是替前端持有密钥把上游流式帧转成前端能消费的 SSE 帧顺带记日志和用量。如果你在 Java 侧用 Spring AI 搭对话机器人底层也是同一套东西Spring AI 对 OpenAI 兼容接口的流式支持返回的是 Flux 本质就是把上游 SSE 帧里 delta.content 逐个发射出来再以 text/event-stream 推给浏览器。协议没变只是换了解析库。选型时不用纠结语言谁都能做关键是转发层别丢帧、别缓冲。4.2 FastAPI 转发把 OpenAI 流生成器包装成 StreamingResponseimport json from fastapi import FastAPI from fastapi.responses import StreamingResponse from openai import OpenAI app FastAPI() client OpenAI(api_keysk-你的密钥, base_urlhttps://api.moonshot.cn/v1) def generate_sse(messages): stream client.chat.completions.create( modelkimi-k3, messagesmessages, streamTrue, temperature0.3, ) for chunk in stream: choices chunk.choices if not choices: continue delta choices[0].delta if delta and delta.content: frame {text: delta.content} yield fdata: {json.dumps(frame, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n app.post(/api/chat) async def chat(request: Request): body await request.json() messages body.get(messages, [{role: user, content: body.get(question, )}]) return StreamingResponse( generate_sse(messages), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no}, )StreamingResponse 会把生成器里 yield 的内容一帧帧写回客户端。这里我把上游 chunk 重新包了一层只含 text 的 JSON 帧目的有两个一是给前端一个干净的协议前端不需要知道 Kimi 内部 delta 结构二是方便在转发层加统一处理比如日志追加、敏感词标记、调用量计数。headers 里的 X-Accel-Buffering: no 是给 Nginx 看的防止代理层把整个响应缓存下来再一次性吐给前端——这是流式被降级成非流式的最常见元凶。注意 generate_sse 是同步生成器FastAPI 会把它丢到线程池跑如果担心并发可以改用 AsyncOpenAI 加 async 生成器但同步写法在多数内部工具场景下完全够用不必过度设计。4.3 浏览器用 fetch 解析 SSE 帧浏览器原生 EventSource 只能发 GET而聊天后端通常要 POST 消息体所以生产上更常见的是 fetch ReadableStream 手动解析。最大的坑是网络层不保证一次 read() 恰好返回一个完整 SSE 帧可能一次给两帧也可能给半帧。必须维护一个字符串缓冲区按 \n\n 切帧。async function chatStream(messages, onText, onDone) { const resp await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ messages }), }); const reader resp.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const frames buffer.split(\n\n); buffer frames.pop(); // 最后一段可能是不完整帧留到下一轮 for (const frame of frames) { for (const line of frame.split(\n)) { if (!line.startsWith(data:)) continue; const payload line.slice(5).trim(); if (payload [DONE]) { onDone onDone(); return; } try { const json JSON.parse(payload); if (json.text) onText(json.text); } catch { // 半帧残留忽略等下一轮补全 } } } } }关键逻辑是 frames.pop()把切剩下的最后一段放回 buffer这一段很可能是被 TCP 拆开的半帧。JSON.parse 包了 try/catch缓冲区策略下理论上不该出现半 JSON但真实网络什么都有这条防线值得留。decoder.decode 第二个参数传 { stream: true }是为了处理多字节字符被拆在两个 chunk 里的情况TextDecoder 会缓存未完成的字节序列。4.4 停止生成AbortController 与连接断开的连锁反应用户点停止按钮前端用 AbortController 把 fetch 掐断let controller null; function stop() { if (controller) { controller.abort(); controller null; } } // 发起请求时 controller new AbortController(); fetch(/api/chat, { signal: controller.signal });前端断连后浏览器关闭 TCP 连接后端 StreamingResponse 的生成器在下一次迭代时收到 GeneratorExit上游对 Kimi 的连接随之关闭模型停止继续生成。这一整条链路是流式特有的后悔药非流式请求一旦发出就无法中途取消流式可以。两个细节注意一是 stop() 之后界面要给明确的结束状态否则用户看到内容停在半截会以为报错二是后端生成器最好包一层 try/finally在 finally 里记一条用户中断日志把正常结束和人为中断区分开月底对账时才知道哪些是用户主动止损。5. Kimi 流式接口高频报错排查401、400、半路断流与用量核对下面五条是我对接 Kimi 流式输出攒下的血泪经验每一条按现象、原因、解决三个顺序写照着对应排查能省大半天。5.1 401 unauthorized报 incorrect API key先查环境变量而不是密钥现象调用立刻抛 unexpected status 401 unauthorized: incorrect api key provided: sk-svcac****。原因这个报错九成不是密钥本身错了而是代码实际拿到的密钥不对。常见情况环境变量名拼错代码读 KIMI_API_KEYshell 里 export 的却是 MOONSHOT_API_KEY.env 文件里密钥行尾有看不见的回车符密钥复制成两段中间夹了空格或者把网页版登录态当成了 API Key。sk-svcac 这类统一前缀说明密钥格式正常问题出在取值链路。解决先把环境变量和实际值对一遍打印出密钥头部和长度做检查。再到控制台重新生成一把新 Key写死在代码里验证能通再回去排查加载链路。切记不要把真 Key 打全量日志前后各留四位足够定位。5.2 400 context length一接多轮对话就翻车现象api error 400 this models maximum context length is 1048576 tokens. Howeve...原因上下文窗口超限。Kimi 新一代模型的窗口很大但大也架不住多轮对话把完整历史全塞进去再加一两个长文档超限就是一瞬间的事。开流式本身不会导致这个错但它往往在流式联调时才暴露因为你终于开始跑连续对话了。解决发给 API 之前对 messages 做裁剪。常见做法是只保留最近 N 轮加一条系统提示或者用 token 计数函数估算超了就从最老的消息开始丢需要带长文时把检索结果单独截断而不是整个塞进 system。流式模式下这一点尤其重要因为运行中报 400用户那边的打字机是突然停的体验比非流式还糟。粗略估算 token 可以先按中文字符数换算量级精确值以控制台的 token 计数工具为准。5.3 流式中途断流代理层把 SSE 缓冲成了整体返回现象本地 curl 直连 Kimi 好好的走 Nginx 之后页面卡着不出字过很久一次性全出来或者干脆中途断掉。原因Nginx 默认对上游响应做缓冲后端是一帧帧推的代理层攒够了才往客户端吐流式的实时性被磨平如果代理的 read timeout 短于模型思考时间还会直接把连接掐掉。解决对聊天接口做两个配置——proxy_buffering off 关掉缓冲proxy_read_timeout 调大到 300 秒以上。后端响应头带上 X-Accel-Buffering: no 也能让 Nginx 对该响应关闭缓冲。排查时用 curl -N 分别打后端和打 Nginxcurl -N http://127.0.0.1:8000/api/chat \ -H Content-Type: application/json \ -d {question:你好}对比两边的首字到达时间就能定位是哪一层把流拖住了。5.4 流式帧被拆包粘包按行解析导致 JSON 解析崩溃现象日志里频繁抛 JSONDecodeError或者页面内容偶尔少一段字。原因TCP 层不保证消息边界一个 SSE 帧可能被拆成两半两个帧也可能粘在一起。requests 的 iter_lines 和前端 fetch 的按行读都只处理了行这一层没处理帧这一层。SDK 内部封装了解析所以不报错一旦自己写就容易踩。解决统一用缓冲区按 \n\n 切帧的写法第 3.3 和第 4.3 已经给了两个实现。帧分隔符是空行把响应体按 \n\n 分割最后一段残留放回下一轮。手动解析和前端解析都用这一个模式能覆盖绝大多数粘包拆包场景。如果按帧切了还在乱序那基本是自定义代理层在作怪去查代理而不是查 Kimi。5.5 流式模式下 usage 一直为 null调用量与计费对不上账现象跑完一轮流式日志里 chunk 从没出现过 token 用量控制台的调用量统计也对不上。原因OpenAI 兼容协议在流式下默认不返回 usage 字段需要显式打开 stream_optionsKimi 接口同样遵循这个约定。很多人以为流式没有用量就放弃了纯属白亏一个核对手段。解决create() 里加上 stream_options{include_usage: True}这样 [DONE] 之前会有一个 delta 为空的帧携带 usage。把 prompt_tokens 和 completion_tokens 落库第二天对调用量账单就有据可查。解析时记住 usage 帧的 choices 为空先判空再取字段。6. 进阶把首字延迟与完整落库一起管起来把链路跑通之后流式输出只是能用了离能上线还差两件事出问题时能不能快速定位完整回答有没有可靠落库。分享三个我一直在用的处理习惯。第一个习惯是给流式加一层追踪包装器。首字延迟也就是从请求发出到收到第一个有内容帧的耗时是流式接口最重要的健康指标。实现很简单把上游 stream 包装一层import time def traced_stream(stream, request_id): start time.time() first_token True parts [] for chunk in stream: delta chunk.choices[0].delta if chunk.choices else None if delta and delta.content: if first_token: print(f[{request_id}] first_token_ms{(time.time() - start) * 1000:.0f}) first_token False parts.append(delta.content) yield chunk print(f[{request_id}] last_token_ms{(time.time() - start) * 1000:.0f})这个包装器和第 4 章的转发层可以叠加转发层外层包 trace日志留在后端前端感知不到。第二轮优化时可以把这几个指标推到监控看板首字延迟异常抬升时告警。第二个习惯是落库时机放在后端。别等前端拼完文本再 POST 回后端存储那份文本在传输和拼接过程中可能丢帧。正确做法是后端生成器跑完、finish_reason 为 stop 时把拼好的完整内容写库。这样即使浏览器断连后端手里仍然有完整回答日志和数据库始终对齐。第三个习惯是上线前的固定验证清单用一组固定问题跑流式接口记录三条线——首字延迟、断流次数、usage 里 token 数与预估的对应关系。断流率超过 1% 就先查代理层和超时配置别急着怪模型。我现在的习惯是每个接入流式的项目都留一个 trace 开关默认开启日志落到独立文件。线上出问题先看 first_token_ms 有没有异常抬升再看有没有 finish_reasonlength 的截断记录大部分诡异问题都能在几分钟内定位。流式输出本身不玄学把每一帧的来龙去脉记清楚了它就是个老老实实的 HTTP 协议。希望帮到你。本文还有配套的精品资源点击获取