大模型最舒服的交互是打字机式流式输出，而前面 W4/W5 的"发消息→等回复"模型天生不吃这套：要么憋一整段回、要么靠企微轮询刷新。智能机器人是官方给 AI 场景开的口子——它的流式消息机制就是为大模型设计的。

一、接口速览表

二、两种模式怎么选

• 内网部署 / 想省事 → 长连接：不开公网、不写加解密（W5 那一整套省了），SDK 官方给了 Node.js/Python
• 已有公网服务 / 要高可靠 → Webhook：无状态更抗造，但要走 W5 验签解密
• 别为了"高级"硬上长连接——它是单连接，多副本部署要做主备选举，反而比 Webhook 复杂

三、流式状态机（防御版，本地实测跑通）

大模型边生成边推的关键，是正确管理 stream.id 的生命周期。核心坑：content 传全量（每次带完整文本，不是 delta）、finish 后封死、10 分钟红线。

"""智能机器人流式回复状态机（对齐官方 stream 规则，本地模拟 LLM 输出，dry-run 不出网）
规则：同 stream.id 首发创建/再发刷新(content全量非增量)/finish=true封存；
     首次推送起 10 分钟内必须 finish，否则自动结束；content ≤20480字节。"""
import time

class StreamBox:
    LIMIT = 10 * 60          # 官方：从首次推送10分钟内需finish
    MAX_B = 20480
    def __init__(self, req_id, stream_id):
        self.req_id, self.sid = req_id, stream_id
        self.created_at = None
        self.closed = False
        self.pushes = 0
    def frame(self, content, finish=False, now=None):
        now = now or time.time()
        if self.closed:
            raise RuntimeError('stream %s 已finish，不可再推送' % self.sid)
        if self.created_at is None:
            self.created_at = now                     # 首次：创建流式消息
        if now - self.created_at > self.LIMIT and not finish:
            raise RuntimeError('超10分钟未finish，消息已自动结束')
        b = len(content.encode('utf-8'))
        if b > self.MAX_B:
            raise ValueError('content %d字节 >20480（官方上限）' % b)
        self.pushes += 1
        if finish:
            self.closed = True
        return {'cmd': 'aibot_respond_msg', 'headers': {'req_id': self.req_id},
                'body': {'msgtype': 'stream',
                         'stream': {'id': self.sid, 'finish': finish, 'content': content}}}

if __name__ == '__main__':
    box = StreamBox(req_id='REQ_77', stream_id='s_leave_001')
    # 模拟 LLM 逐步吐字（注意：content 是"全量"不是增量）
    steps = ['正在为您查询', '正在为您查询李明的年假余额', '正在为您查询李明的年假余额：剩余 6.5 天，',
             '正在为您查询李明的年假余额：剩余 6.5 天。建议安排 11 月中旬调休错峰。']
    for i, s in enumerate(steps):
        fr = box.frame(s, finish=(i == len(steps) - 1))
        tag = 'finish=true 封存' if fr['body']['stream']['finish'] else '刷新'
        print('%s: %s | 内容=%s' % (fr['cmd'], tag, s[:26] + '…' if len(s) > 26 else s))
    try:
        box.frame('还想追加')
    except RuntimeError as e:
        print('⑤ finish 后再推送 →', e)
    late = StreamBox('REQ_78', 's_slow')
    late.frame('开始', now=time.time())
    try:
        late.frame('还在想', now=time.time() + 11 * 60)
    except RuntimeError as e:
        print('⑥ 超10分钟刷新 →', e)
    print('总推送次数:', box.pushes)

实测输出：

aibot_respond_msg: 刷新 | 内容=正在为您查询
aibot_respond_msg: 刷新 | 内容=正在为您查询李明的年假余额
aibot_respond_msg: 刷新 | 内容=正在为您查询李明的年假余额：剩余 6.5 天，
aibot_respond_msg: finish=true 封存 | 内容=正在为您查询李明的年假余额：剩余 6.5 天。建议安…
⑤ finish 后再推送 → stream s_leave_001 已finish，不可再推送
⑥ 超10分钟刷新 → 超10分钟未finish，消息已自动结束
总推送次数: 4

前 3 帧刷新、第 4 帧 finish 封存，两条违规（finish 后追加、超时刷新）都被拦下。接入真实大模型时，把 steps 换成 LLM 的 token 流、每积累一定字数或到一个句子边界推一帧即可（不要每个 token 推一帧，既浪费 30 条/分钟配额，客户端也只展示最终全量）。

四、错误码表 / 频控红线

五、可复制提示词：把 LLM 输出接进 stream 的分片策略

这段直接决定流式体验（本文为本文原创）：

你是流式输出节流器。LLM 原始 token 流进来，你决定何时推一帧给企微 stream：
规则：
1. 累积到"完整句子"或"每满 40 个汉字"才成帧（避免逐 token 刷屏、省30次/分钟配额）
2. 每次输出"从头到当前的全量文本"（企微 stream.content 是全量覆盖，非增量）
3. 遇到超时风险（累计接近10分钟）立即成帧并置 finish=true 收尾
4. 输出 JSON：{"frame_text":"全量内容","finish":true|false}
当前 token 增量：{delta}   已累计全量：{accumulated}   已耗时：{elapsed_sec}秒

第 2 条是全量还是增量的关键——企微这里是全量，接增量思维（很多流式 API 是 delta）的模型会写反，务必锁死。

六、学完有什么

到这里企业微信的"发/收/交互/流式"四板斧齐了。最后一篇 W8 补上对外场景——客户联系（微信客户加到你员工，AI 自动生成个性化欢迎语并带 welcome_code 时效发送），把这套能力从"对内办公"延伸到"对客运营"。

七、参考与说明

本文系列涉及的提示词模板与配套教程合集，收录于提示岛（prompts.ssjintaitong.com）。

接口原文在企业微信开发者中心官方文档（管理后台侧栏可进入，或搜"智能机器人长连接"“回复消息”），本文两种模式对比、stream 全量刷新机制、10 分钟与 20480 字节限制、连接数与频控规则均以 2026-11-05 官方页面为准；第三节代码实测跑通（输出见正文）。

智能机器人两种API模式/BotID+Secret/单连接踢旧/ping 30秒/stream.id全量刷新+finish封存/10分钟红线/20480字节/会话30条分钟1000条小时/模式互斥——逐条摘自企业微信开发者中心官方文档 2026-11-05 核对原文；状态机代码本地真跑；示例 BotID/Stream 均为占位假值。