流式消息发送指南

什么是流式消息？

流式消息（Streaming Messages） 是一种渐进式的消息发送方式，支持在消息发送过程中分批次更新内容。适用于以下场景：

✅ AI 智能助理：实时显示 AI 生成的回复内容
✅ 长文本生成：逐步展示较长的文本内容
✅ 代码生成：逐行显示生成的代码
✅ 实时翻译：动态更新翻译结果

流式消息的优势

特性	传统消息	流式消息
响应速度	需等待完整内容	立即响应，逐步显示
用户体验	等待时间长	实时可见，降低焦虑
资源利用	一次性加载	按需传输，节省带宽
交互性	低	高，可中途停止

快速开始

5 分钟上手示例

下面是一个完整的流式消息发送流程示例：

// 步骤 1: 创建占位符消息
const placeholder = await fetch('https://open.qingtui.cn/v1/message/text/stream/send/placeholder', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer ACCESS_TOKEN'
  },
  body: JSON.stringify({
    chatType: '5', // 轻应用单聊
    launchMsgId: 'question_msg_id_123',
    launchAccountId: 10086,
    toObjectId: 'user_openid_456'
  })
});

const { msgId, appMsgId } = await placeholder.json();

// 步骤 2: 建立流式连接并持续发送内容
const streamResponse = await fetch('https://open.qingtui.cn/v1/message/text/stream/send/single', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/x-ndjson', // 使用 NDJSON 格式
    'Authorization': 'Bearer ACCESS_TOKEN',
    'Transfer-Encoding': 'chunked' // 启用分块传输编码
  },
  body: generateStreamBody(msgId, appMsgId)
});

// 生成器函数：持续产生 JSON 行
function* generateStreamBody(msgId, appMsgId) {
  const chunks = ['您好，', '我是智能助理，', '很高兴为您服务。'];

  for (const chunk of chunks) {
    // 每一行都是一个独立的 JSON 对象
    yield JSON.stringify({
      toUser: 'user_openid_456',
      status: 1, // 1 = 生成中
      placeholderMsgId: msgId,
      placeholderAppMsgId: appMsgId,
      launchMsgId: 'question_msg_id_123',
      launchAccountId: '10086',
      message: chunk // 每次新增的内容（不是累积内容）
    }) + '\n';

    // 模拟打字延迟
    await new Promise(resolve => setTimeout(resolve, 500));
  }

  // 最后一行：结束标志
  yield JSON.stringify({
    toUser: 'user_openid_456',
    status: 2, // 2 = 结束
    placeholderMsgId: msgId,
    placeholderAppMsgId: appMsgId,
    launchMsgId: 'question_msg_id_123',
    launchAccountId: '10086',
    message: '您好，我是智能助理，很高兴为您服务。' // 最终完整内容
  }) + '\n';
}

效果预览：

[正在输入...]
您好，
您好，我是智能助理，
您好，我是智能助理，很高兴为您服务。
✓ 消息已发送

完整开发流程

阶段一：创建占位符消息

在正式发送流式内容之前，需要先调用占位符接口创建一个"空消息"，获取后续更新所需的 ID。

接口说明

请求地址： POST https://open.qingtui.cn/v1/message/text/stream/send/placeholder

请求参数：

参数名	类型	必须	说明
chatType	string	是	聊天对话类型： `2`: 群聊 `5`: 轻应用单聊
launchMsgId	string	是	发起提问的消息 ID
launchAccountId	number	是	发起提问的账号 ID
toObjectId	string	是	发送对象 ID（群 ID 或账号 ID）

请求示例：

{
  "chatType": "5",
  "launchMsgId": "msg_123456",
  "launchAccountId": 10086,
  "toObjectId": "user_openid_789"
}

响应示例：

{
  "errcode": 0,
  "errmsg": "success",
  "data": {
    "msgId": "placeholder_msg_abc123",
    "appMsgId": "app_msg_def456"
  }
}

响应参数说明：

参数名	说明	用途
msgId	流式占位消息 ID	用于后续更新消息内容
appMsgId	流式轻应用存储的消息 ID	用于消息管理和追踪

注意事项：

⚠️ 占位符消息创建后需要在 30 秒内开始发送流式内容，否则会自动过期
⚠️ 每个占位符只能被一个流式会话使用
✅ 建议设置超时处理，如果长时间未收到回复应提示用户

阶段二：流式输出内容

获取到占位符 ID 后，建立一个持久的 HTTP 连接，通过分块传输编码（Chunked Transfer Encoding）持续发送 JSON 行数据。

接口说明

发送给个人： POST https://open.qingtui.cn/v1/message/text/stream/send/single

发送给群组： POST https://open.qingtui.cn/v1/message/text/stream/send/channel

关键特性

⚠️ 重要： 流式消息使用 HTTP Chunked Transfer Encoding（分块传输编码）

✅ 单次连接：只建立一次 HTTP 连接
✅ 持续传输：通过同一个连接持续发送数据
✅ JSON 行格式：每行都是一个独立的 JSON 对象（NDJSON 格式）
✅ 自动分块：设置 Transfer-Encoding: chunked 后由 HTTP 协议自动分块

发送给个人

请求头示例：

POST /v1/message/text/stream/send/single HTTP/1.1
Host: open.qingtui.cn
Authorization: Bearer ACCESS_TOKEN
Content-Type: application/x-ndjson
Transfer-Encoding: chunked

请求体格式（JSON Lines）：

{"toUser":"user_openid_789","status":1,"placeholderMsgId":"placeholder_msg_abc123","placeholderAppMsgId":"app_msg_def456","launchMsgId":"msg_123456","launchAccountId":"10086","message":"这是第一段内容..."}\n
{"toUser":"user_openid_789","status":1,"placeholderMsgId":"placeholder_msg_abc123","placeholderAppMsgId":"app_msg_def456","launchMsgId":"msg_123456","launchAccountId":"10086","message":"这是第二段内容..."}\n
{"toUser":"user_openid_789","status":2,"placeholderMsgId":"placeholder_msg_abc123","placeholderAppMsgId":"app_msg_def456","launchMsgId":"msg_123456","launchAccountId":"10086","message":"这是最终内容"}\n

注意事项：

⚠️ 每一行必须是一个完整的 JSON 对象，以换行符 \n 结尾
⚠️ 使用 application/x-ndjson 作为 Content-Type
⚠️ 必须设置 Transfer-Encoding: chunked 启用分块传输
⚠️ 所有 JSON 行通过同一个 HTTP 连接发送

发送给群组

请求头示例：

POST /v1/message/text/stream/send/channel HTTP/1.1
Host: open.qingtui.cn
Authorization: Bearer ACCESS_TOKEN
Content-Type: application/x-ndjson
Transfer-Encoding: chunked

请求体格式（JSON Lines）：

{"channelId":"group_chat_999","status":1,"placeholderMsgId":"placeholder_msg_abc123","placeholderAppMsgId":"app_msg_def456","launchMsgId":"msg_123456","launchAccountId":"10086","message":"这是群组的流式回复内容..."}\n
{"channelId":"group_chat_999","status":2,"placeholderMsgId":"placeholder_msg_abc123","placeholderAppMsgId":"app_msg_def456","launchMsgId":"msg_123456","launchAccountId":"10086","message":"这是群组的最终回复"}\n

内容传输策略：

流式消息使用 JSON Lines (NDJSON) 格式传输：

格式要求：

// 每一行都是一个独立的 JSON 对象
// status=1 时：message 为每次新增的内容片段
// status=2 时：message 为最终的完整内容
JSON.stringify({ ..., status: 1, message: '新增片段 1' }) + '\n'
JSON.stringify({ ..., status: 1, message: '新增片段 2' }) + '\n'
JSON.stringify({ ..., status: 2, message: '完整内容' }) + '\n'

注意事项：

⚠️ status=1 时，message 字段是每次新增的内容片段（不是累积内容）
⚠️ status=2 时，message 字段是最终的完整内容（用于消息落库）
⚠️ 客户端负责将接收到的所有片段拼接成完整内容展示

实现方式：

// 使用生成器函数持续产生 JSON 行
async function* generateStreamBody(msgId, appMsgId) {
  const chunks = ["第一段内容", "第二段内容", "第三段内容"];

  for (const chunk of chunks) {
    // 产生一行 JSON 数据，message 为每次新增的内容
    yield JSON.stringify({
      toUser: "user_openid_123",
      status: 1,
      placeholderMsgId: msgId,
      placeholderAppMsgId: appMsgId,
      launchMsgId: "msg_456",
      launchAccountId: "10086",
      message: chunk, // 注意：是新增内容，不是累积内容
    }) + "\n";

    // 控制发送频率
    await sleep(500);
  }

  // 最后一行：结束标志，包含完整内容
  yield JSON.stringify({
    toUser: "user_openid_123",
    status: 2,
    placeholderMsgId: msgId,
    placeholderAppMsgId: appMsgId,
    launchMsgId: "msg_456",
    launchAccountId: "10086",
    message: "第一段内容第二段内容第三段内容", // 最终完整内容
  }) + "\n";
}

// 发送流式请求
const response = await fetch(
  "https://open.qingtui.cn/v1/message/text/stream/send/single",
  {
    method: "POST",
    headers: {
      "Content-Type": "application/x-ndjson",
      Authorization: "Bearer ACCESS_TOKEN",
      "Transfer-Encoding": "chunked",
    },
    body: generateStreamBody(msgId, appMsgId),
  },
);

建议： 使用生成器函数来管理流式数据的产生和发送节奏。

阶段三：结束流式消息

当所有内容发送完成后，发送最后一行 JSON 数据，将 status 设置为 2 标记消息完成。

最后一行示例：

{"toUser":"user_openid_789","status":2,"placeholderMsgId":"placeholder_msg_abc123","placeholderAppMsgId":"app_msg_def456","launchMsgId":"msg_123456","launchAccountId":"10086","message":"这是最终的完整回复内容。"}\n

注意事项：

⚠️ 必须发送 status=2 的最后一行，否则消息会一直处于"生成中"状态
⚠️ 发送完最后一行后关闭连接
⚠️ 连接关闭后无法再更新该消息内容
✅ 建议在结束前做内容校验，确保没有敏感信息

时序图与流程图

完整交互时序图

状态流转图

发送流程图

实战案例

案例 1：AI 智能助理

class AIAssistant {
  constructor(accessToken) {
    this.accessToken = accessToken;
    this.baseUrl = 'https://open.qingtui.cn';
  }

  async sendStreamingReply(params) {
    const {
      toUser,
      question,
      launchMsgId,
      launchAccountId,
      chatType = '5' // 默认单聊
    } = params;

    try {
      // 1. 创建占位符
      const placeholder = await this.createPlaceholder({
        chatType,
        launchMsgId,
        launchAccountId,
        toObjectId: toUser
      });

      const { msgId, appMsgId } = placeholder;

      // 2. 调用 AI 接口获取流式响应
      const aiStream = await this.callAIModel(question);

      // 3. 流式发送内容（使用生成器函数）
      const streamBody = this.generateStreamBody({
        toUser,
        status: 1,
        placeholderMsgId: msgId,
        placeholderAppMsgId: appMsgId,
        launchMsgId,
        launchAccountId
      }, aiStream);

      // 建立流式连接并持续发送
      await this.sendStreamWithChunkedTransfer(streamBody);

      return { success: true };

    } catch (error) {
      console.error('流式消息发送失败:', error);
      throw error;
    }
  }

  *generateStreamBody(baseParams, aiStream) {
    for await (const chunk of aiStream) {
      // 每次发送新增的内容片段
      yield JSON.stringify({
        ...baseParams,
        message: chunk // 新增内容，不是累积内容
      }) + '\n';

      // 控制发送频率
      this.sleep(100);
    }
  }

  async sendStreamWithChunkedTransfer(streamBody) {
    const response = await fetch(`${this.baseUrl}/v1/message/text/stream/send/single`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/x-ndjson',
        'Authorization': `Bearer ${this.accessToken}`,
        'Transfer-Encoding': 'chunked'
      },
      body: streamBody
    });

    if (!response.ok) {
      throw new Error(`HTTP error! status: ${response.status}`);
    }
  }

  async createPlaceholder(params) {
    const response = await fetch(`${this.baseUrl}/v1/message/text/stream/send/placeholder`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${this.accessToken}`
      },
      body: JSON.stringify(params)
    });

    const result = await response.json();
    if (result.errcode !== 0) {
      throw new Error(result.errmsg);
    }

    return result.data;
  }

  sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

// 使用示例
const assistant = new AIAssistant('your_access_token');

await assistant.sendStreamingReply({
  toUser: 'user_openid_123',
  question: '如何学习编程？',
  launchMsgId: 'msg_question_456',
  launchAccountId: '10086'
});

案例 2：群组机器人

class GroupBot {
  constructor(accessToken) {
    this.accessToken = accessToken;
  }

  async replyToGroup(groupChatId, question, answer) {
    // 1. 创建占位符
    const placeholder = await fetch(
      "https://open.qingtui.cn/v1/message/text/stream/send/placeholder",
      {
        method: "POST",
        headers: {
          Authorization: `Bearer ${this.accessToken}`,
        },
        body: JSON.stringify({
          chatType: "2", // 群聊
          launchMsgId: question.msgId,
          launchAccountId: question.accountId,
          toObjectId: groupChatId,
        }),
      },
    );

    const { msgId, appMsgId } = (await placeholder.json()).data;

    // 2. 分段发送答案（使用生成器函数）
    const paragraphs = answer.split("\n\n");
    const streamBody = this.generateGroupStreamBody(
      {
        channelId: groupChatId,
        status: 1,
        placeholderMsgId: msgId,
        placeholderAppMsgId: appMsgId,
        launchMsgId: question.msgId,
        launchAccountId: question.accountId,
      },
      paragraphs,
    );

    // 建立流式连接并持续发送
    await this.sendStreamWithChunkedTransfer(streamBody);
  }

  *generateGroupStreamBody(baseParams, paragraphs) {
    for (const paragraph of paragraphs) {
      // 每次发送新增的段落
      yield JSON.stringify({
        ...baseParams,
        message: paragraph, // 新增内容，不是累积内容
      }) + "\n";

      // 模拟思考时间
      yield* this.sleep(500);
    }

    // 发送结束标记
    yield JSON.stringify({
      ...baseParams,
      status: 2, // 2 表示流式消息结束
      message: "", // 结束时不需要消息内容
    }) + "\n";
  }

  async sendStreamWithChunkedTransfer(streamBody) {
    await fetch("https://open.qingtui.cn/v1/message/text/stream/send/channel", {
      method: "POST",
      headers: {
        "Content-Type": "application/x-ndjson",
        Authorization: `Bearer ${this.accessToken}`,
        "Transfer-Encoding": "chunked",
      },
      body: streamBody,
    });
  }

  sleep(ms) {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }
}

错误处理

常见错误码

错误码	说明	解决方案
40001	参数错误	检查必填参数和参数格式
40002	占位符已过期	重新创建占位符
40003	占位符不存在	检查 msgId 是否正确
40004	重复使用占位符	每个占位符只能使用一次
50001	服务器内部错误	稍后重试
50002	流式发送超时	检查网络连接，重试

错误处理最佳实践

async function sendWithRetry(params, maxRetries = 3) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await sendStreamMessage(params);
    } catch (error) {
      if (i === maxRetries - 1) {
        // 最后一次重试失败，发送错误提示
        await sendErrorMessage(params);
        throw error;
      }

      // 等待后重试
      await sleep(1000 * Math.pow(2, i));
    }
  }
}

function handleError(error, context) {
  console.error("流式消息错误:", {
    error: error.message,
    code: error.code,
    context,
    timestamp: new Date().toISOString(),
  });

  // 根据错误类型处理
  switch (error.code) {
    case 40002:
      return "占位符已过期，请重新提问";
    case 40003:
      return "消息 ID 无效，请联系管理员";
    case 50001:
      return "服务器繁忙，请稍后重试";
    default:
      return "发生未知错误，请稍后重试";
  }
}

性能优化建议

1. 发送频率控制

// 推荐的发送间隔
const DELAY_CONFIG = {
  minDelay: 100, // 最小间隔 100ms
  maxDelay: 1000, // 最大间隔 1s
  chunkSize: 50, // 每段字符数
};

function calculateDelay(chunkSize) {
  // 根据内容长度动态调整延迟
  const delay = Math.max(
    DELAY_CONFIG.minDelay,
    Math.min(DELAY_CONFIG.maxDelay, chunkSize * 10),
  );
  return delay;
}

2. 内容分块策略

function splitContent(content) {
  const chunks = [];
  const chunkSize = 50; // 每块 50 个字符

  for (let i = 0; i < content.length; i += chunkSize) {
    chunks.push(content.slice(i, i + chunkSize));
  }

  return chunks;
}

// 按句子分割（更自然）
function splitBySentence(content) {
  return content.match(/[^.!?]+[.!?]+/g) || [content];
}

3. 并发控制

class StreamController {
  constructor(maxConcurrent = 5) {
    this.queue = [];
    this.active = 0;
    this.maxConcurrent = maxConcurrent;
  }

  async add(task) {
    if (this.active >= this.maxConcurrent) {
      await new Promise((resolve) => this.queue.push(resolve));
    }

    this.active++;
    try {
      return await task();
    } finally {
      this.active--;
      if (this.queue.length > 0) {
        const next = this.queue.shift();
        next();
      }
    }
  }
}

注意事项

⚠️ 重要提醒

占位符有效期
- 创建后 30 秒内必须开始发送
- 超过 5 分钟未完成会自动清理
内容限制
- 单次发送内容不超过 2000 字符
- 总内容长度不超过 10000 字符
- 禁止发送违规内容
频率限制
- 单个用户每秒最多 10 次更新
- 单个群组每秒最多 5 次更新
- 应用级别有总 QPS 限制
状态管理
- 必须正确设置 status 字段
- 结束后无法再更新
- 异常状态要通知用户

✅ 最佳实践

用户体验
- 显示"正在输入..."提示
- 提供停止生成按钮
- 支持重新生成
错误恢复
- 实现自动重试机制
- 保存生成进度
- 提供友好的错误提示
监控告警
- 记录发送成功率
- 监控平均响应时间
- 设置失败率告警

常见问题 FAQ

Q: 流式消息和普通消息有什么区别？
A: 流式消息支持分批次更新同一条消息的内容，用户可以实时看到生成过程；普通消息是一次性发送完整内容。

Q: 为什么要先创建占位符？
A: 占位符消息会预分配一个消息 ID，后续的流式更新都基于这个 ID 进行，确保所有更新都应用到同一条消息上。

Q: status 字段有哪些值？
A:

1: 流式生成中（持续更新）
2: 流式结束（完成回复）
5: 异常（出错时使用）

Q: 支持发送图片、视频等多媒体吗？
A: 当前版本仅支持文本文案消息，多媒体消息的流式发送将在后续版本支持。

Q: 如果网络中断怎么办？
A: 建议实现断点续传机制，记录最后发送的位置，恢复后从该位置继续发送。

Q: 可以在多个设备间同步吗？
A: 可以，流式消息会同步到所有登录同一账号的设备。

Q: 如何调试流式消息？
A:

使用浏览器开发者工具查看网络请求
检查每个请求的响应状态
确认 msgId 是否正确传递
验证 status 状态转换是否正确

附录

完整代码示例

查看完整的示例代码：GitHub 示例仓库

技术支持

如有问题，请联系：

开发者社区：https://open.qingtui.cn/forum
技术邮箱：developer@qingtui.com

什么是流式消息？​

流式消息的优势​

快速开始​

5 分钟上手示例​

完整开发流程​

阶段一：创建占位符消息​

接口说明​

阶段二：流式输出内容​

接口说明​

关键特性​

发送给个人​

发送给群组​

阶段三：结束流式消息​

时序图与流程图​

完整交互时序图​

状态流转图​

发送流程图​

实战案例​

案例 1：AI 智能助理​

案例 2：群组机器人​

错误处理​

常见错误码​

错误处理最佳实践​

性能优化建议​

1. 发送频率控制​

2. 内容分块策略​

3. 并发控制​

注意事项​

⚠️ 重要提醒​

✅ 最佳实践​

常见问题 FAQ​

附录​

完整代码示例​

相关文档​

技术支持​

什么是流式消息？

流式消息的优势

快速开始

5 分钟上手示例

完整开发流程

阶段一：创建占位符消息

接口说明

阶段二：流式输出内容

接口说明

关键特性

发送给个人

发送给群组

阶段三：结束流式消息

时序图与流程图

完整交互时序图

状态流转图

发送流程图

实战案例

案例 1：AI 智能助理

案例 2：群组机器人

错误处理

常见错误码

错误处理最佳实践

性能优化建议

1. 发送频率控制

2. 内容分块策略

3. 并发控制

注意事项

⚠️ 重要提醒

✅ 最佳实践

常见问题 FAQ

附录

完整代码示例

相关文档

技术支持