最近做的企微二开，业务方盯着 ChatGPT 那种"一个字一个字蹦"的回复体验，要求我们的企微机器人也搞流式——客户问一句话，机器人别憋半天才回一大坨，要边生成边发。但企微消息接口本身没有真正的流式（没有 edit message 能力），落地得绕一下。把踩的坑和最终方案记下来。

底层用的是 Eyun 平台开放的企微 API，统一 POST+JSON，鉴权用 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 "ok"

Webhook 路径 /wx-api/webhook/，首次创建返回 secret 用于验签。回调要 3 秒内回 200，所以流式生成放在后台任务跑，回调立即返回。否则平台会重试，导致同一问题生成多份答案。

第二步：调大模型流式接口（SSE）

大模型要开 stream=true，模型一边生成一边吐 token，首字延迟从几秒缩到几百毫秒。我们用 OpenAI 兼容接口：

def stream_llm(messages, on_chunk):
    resp = requests.post(LLM_URL, json={
        "model": "gpt-4",
        "messages": messages,
        "stream": True
    }, stream=True)
    buffer = ""
    for line in resp.iter_lines():
        if line.startswith(b"data: "):
            chunk = json.loads(line[6:])["choices"][0].get("delta", {}).get("content", "")
            if chunk:
                buffer += chunk
                on_chunk(chunk, buffer)
    return buffer

on_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_len=400):
    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 能力，靠分段发送模拟。把切分策略选对、上限守住、兜底做严，机器人回复体验从"半天憋一坨"变成"边想边答"，业务方和客户的反馈都会明显不同。