最近做的企微二开业务方盯着 ChatGPT 那种一个字一个字蹦的回复体验要求我们的企微机器人也搞流式——客户问一句话机器人别憋半天才回一大坨要边生成边发。但企微消息接口本身没有真正的流式没有 edit message 能力落地得绕一下。把踩的坑和最终方案记下来。底层用的是Eyun 平台开放的企微 API统一 POSTJSON鉴权用 App Token 加 appid请求头带Authorization: Bearer eyk_xxxx路径统一{BASE_URL}/wx-api/api/模块/动作响应封套{code, data, detail, message, time}code 为 0 成功。所谓流式回复在企微里本质是分段发送——大模型流式生成、缓冲到自然段切分点、调多次 sendText 把段落分批发出去模拟流式体验。第一步消息进来——Webhook 收提问客户在企微发问题回调进来app.route(/wx-api/webhook/, methods[POST]) def webhook(): if request.headers.get(X-Eyun-Event) ! message: return ok data request.json[data] stream_reply(data[appid], data[fromUin], data[content]) return okWebhook 路径/wx-api/webhook/首次创建返回 secret 用于验签。回调要 3 秒内回 200所以流式生成放在后台任务跑回调立即返回。否则平台会重试导致同一问题生成多份答案。第二步调大模型流式接口SSE大模型要开 streamtrue模型一边生成一边吐 token首字延迟从几秒缩到几百毫秒。我们用 OpenAI 兼容接口def stream_llm(messages, on_chunk): resp requests.post(LLM_URL, json{ model: gpt-4, messages: messages, stream: True }, streamTrue) buffer for line in resp.iter_lines(): if line.startswith(bdata: ): chunk json.loads(line[6:])[choices][0].get(delta, {}).get(content, ) if chunk: buffer chunk on_chunk(chunk, buffer) return bufferon_chunk是回调每次模型吐 token 都被调用。但企微没有 edit message 接口每吐一个字就发一条消息是灾难——客户会被刷屏刷到卸载企微。所以 on_chunk 里不能直接发要缓冲。第三步缓冲到自然段切分点再发策略是按自然段切分。模型吐出来的 token 累积到出现换行、句号、问号这种自然停顿点时把累积的段落发出去。这样既保留了边生成边发的体验又控制了消息条数。def stream_reply(appid, to_uin, content): messages build_messages_with_history(appid, to_uin, content) pending sent_count 0 def on_chunk(chunk, buffer): nonlocal pending, sent_count pending chunk # 自然段切分点换行、句号、问号 if should_flush(pending) and sent_count 3: send_text(appid, to_uin, pending.strip()) pending sent_count 1 final stream_llm(messages, on_chunk) if pending.strip(): send_text(appid, to_uin, pending.strip())should_flush判断当前缓冲是否到了自然段末尾。sent_count 3是硬上限——一条回答最多分 4 段发再多就是刷屏。超过 3 段就把剩余内容合并到最后一段一次发出去宁可丢流式感也不能刷屏。第四步首字占位——别让客户干等模型生成首字可能要 1-2 秒这几秒客户看不到任何反应会以为机器人卡了。接收到回调后立刻发一条占位消息def stream_reply(appid, to_uin, content): send_text(appid, to_uin, 正在思考...) # 然后开始流式生成 ...占位消息用 message/sendText 发to字段填提问者 uin。占位不要发请稍等这种话——客户看了心烦发个正在思考...这种拟人化提示就行。第五步超长答案分段策略模型答案超过 500 字时即使按自然段切分也会超长。企微单条消息有长度上限超过会被截断。处理方式是按字符数硬切分def split_long_text(text, max_len400): paragraphs text.split(\n) chunks, current [], for p in paragraphs: if len(current) len(p) max_len: if current: chunks.append(current) current p else: current (current \n p) if current else p if current: chunks.append(current) return chunks def send_long_answer(appid, to_uin, text): for chunk in split_long_text(text): send_text(appid, to_uin, chunk) time.sleep(0.3) # 避免发太快触发频控按段落切分而不是按字符硬切避免把一句话从中间断开。time.sleep(0.3)是必要的——企微对同一会话的发送频率有上限连发会触发频控报错。频控报错码在网关层是字符串rate_limit要识别后做退避重试。第六步流式中的失败兜底流式生成中途模型断流网络抖动、模型限流怎么办已经发了前几段后半段没了客户看到半截答案。处理方式是检测到流中断时调一次非流式补全def stream_reply(appid, to_uin, content): ... try: final stream_llm(messages, on_chunk) except StreamInterrupted: # 流断了用非流式把剩余补全 final non_stream_llm(messages [{role: assistant, content: pending}]) send_text(appid, to_uin, final[len(pending):])把已经发出去的内容作为 assistant 消息拼回去让模型从断点续写。客户感知不到断流只看到答案继续往下走。几个权衡点流式回复的体验提升是实在的——客户首字等待从 5 秒缩到 1 秒但代价是消息条数增加、消息历史被流式段落切碎。要权衡几个点短回答 100 字不开流式直接一次发完避免占位消息和正式消息两条刷屏长回答按自然段切分最多 4 段占位消息只在模型生成超过 1 秒时发模型快就省掉占位失败要兜底别让客户看到半截答案流式这套东西本质是多次 sendText 调用模拟出来企微接口本身不支持真流式——把切分点选好、上限守住、失败兜底做严客户用起来感觉是真的流式就够了。写在最后流式回复这套本质是 Webhook 收消息、大模型流式生成SSE、缓冲到自然段切分点调 message/sendText 多次发送、超长按段落硬切、失败用非流式补全。AI 流式吐 token 的体验好但企微端没有 edit message 能力靠分段发送模拟。把切分策略选对、上限守住、兜底做严机器人回复体验从半天憋一坨变成边想边答业务方和客户的反馈都会明显不同。