English | 简体中文

流式请求工具库使用指南

useAgent 提供的流式请求工具库专为大模型应用场景设计，支持 Server-Sent Events (SSE) 流式 POST 通信，具备消息缓存、顺序保证、全链路错误处理等核心能力，帮助开发者快速实现稳定可靠的流式交互体验。

安装

使用任意包管理器安装：

npm install @lesliechueng/stream-fetch-manage --save-dev

yarn add @lesliechueng/stream-fetch-manage --save-dev

pnpm add @lesliechueng/stream-fetch-manage --save-dev

基础用法

import { StreamFetchClient } from "@lesliechueng/stream-fetch-manage";

interface IContent {
  content: string;
  sequenceNumber: number;
  done?: boolean;
}

const streamFetch = new StreamFetchClient<IContent>(
  {
    baseUrl: "/api/chat",
    headers: {
      "Content-Type": "application/json",
    },
    overErrorTimer: 60 * 1000, // 流式中间超时时间，单位为毫秒
  },
  {
    onMessage: (data: IContent) => {
      console.log("收到消息:", data);
    },
    onClose: (lastData: IContent) => {
      console.log("连接关闭", lastData);
    },
    onServerError: (data: IContent, error: Error) => {
      console.error("服务器错误", error);
    },
    onStreamConnectionError: (data: IContent, error: Error) => {
      console.error("流连接错误:", error);
    },
    onConnectionError: (data: IContent, error: Error) => {
      console.error("连接错误:", error);
    },
    onParseError: (data: IContent, error: Error) => {
      console.error("解析错误:", error);
    },
  }
);

// 开始发起请求，下面是具体的参数
streamFetch.sendStreamRequest({
  // 流式中间请求参数
});

// 暂停请求
streamFetch.disconnect();

高级配置

消息顺序保证与缓存，当需要处理可能乱序到达的流式消息时，可配置消息处理器实现顺序保证。

import { StreamFetchClient } from "@lesliechueng/stream-fetch-manage";

interface IContent {
  content: string;
  sequenceNumber: number;
  done?: boolean;
}

const streamFetch = new StreamFetchClient<IContent>(
  {
    baseUrl: "/api/chat",
    headers: {
      "Content-Type": "application/json",
    },
    overErrorTimer: 60 * 1000, // 流式中间超时时间，单位为毫秒
  },
  {
    onMessage: (data: IContent) => {
      console.log("收到消息:", data);
    },
    onClose: (lastData: IContent) => {
      console.log("连接关闭", lastData);
    },
    onServerError: (data: IContent, error: Error) => {
      console.error("服务器错误", error);
    },
    onStreamConnectionError: (data: IContent, error: Error) => {
      console.error("流连接错误:", error);
    },
    onConnectionError: (data: IContent, error: Error) => {
      console.error("连接错误:", error);
    },
    onParseError: (data: IContent, error: Error) => {
      console.error("解析错误:", error);
    },
  },
  {
    maxCacheSize: 6, // 最大缓存大小，单位为条
    cacheTimeout: 5000, // 缓存超时时间，单位为毫秒
    expectedSeq: 0, // 期望的初始化消息索引值
    handleValidateMessageFormat: (data: IContent) => {
      // 校验消息序号的数据类型
      if (typeof data.sequenceNumber !== "number") {
        throw new Error("Message must have a numeric seq field");
      }
    },
    // 使得消息处理器获取消息序号的索引值
    getIndexValue: (data: IContent) => data.sequenceNumber,
  }
);

核心 API

StreamFetchClient 构造函数

new StreamFetchClient(config, eventHandlers, processorConfig?)

| 参数 | 类型 | 说明 | | --------------- | -------------------------- | ------------------------------------ | | config | IStreamFetchClientConfig | 基础配置 | | eventHandlers | ICurrentEventHandlers | 事件处理函数集合 | | processorConfig | IProcessorConfig | 消息处理器配置（可选，用于消息排序） |

IStreamFetchClientConfig

| 属性 | 类型 | 默认值 | 说明 | | -------------- | ------------------------ | ---------------------------------------- | ------------------- | | baseUrl | string | '' | 流式请求基础 URL | | headers | Record<string, string> | { 'Content-Type': 'application/json' } | 请求头 | | overErrorTimer | number | 60000 | 无消息超时时间 (ms) |

ICurrentEventHandlers

| 方法 | 说明 | | ----------------------- | -------------------- | | onMessage | 收到消息时触发 | | onStreamConnectionError | 连接超时时触发 | | onConnectionError | 连接建立失败时触发 | | onServerError | 服务器返回错误时触发 | | onParseError | 消息解析失败时触发 | | onClose | 连接关闭时触发 |

实例方法

sendStreamRequest

发送流式请求

async sendStreamRequest(
  payload: Record<string, any>,
  eventHandlers?: ICurrentEventHandlers<T> | null,
  config?: IStreamFetchClientConfig
)

| 参数 | 类型 | 说明 | | ------------- | -------------------------- | ---------------------- | | payload | Record<string, any> | 请求体数据 | | eventHandlers | ICurrentEventHandlers | 临时事件处理器（可选） | | config | IStreamFetchClientConfig | 临时配置（可选） |

disconnect

断开当前流式连接并清理资源

disconnect();

消息处理机制

当配置了  processorConfig  时，消息处理流程如下：

1. 收到消息后先进行格式验证（handleValidateMessageFormat）
2. 提取消息序号（getIndexValue）
3. 如果序号与预期一致，直接处理并更新预期序号
4. 如果序号大于预期，缓存消息
5. 处理完当前消息后，自动检查缓存中是否有下一条预期消息

这种机制可以确保即使消息乱序到达，最终也能按正确顺序处理。

注意事项

1. 务必在组件卸载或不需要流式连接时调用  disconnect()  方法，避免内存泄漏
2. 根据业务需求合理配置  maxCacheSize  和  cacheTimeout，平衡内存占用和消息可靠性
3. handleValidateMessageFormat  中应严格验证消息格式，避免处理非法数据
4. 对于需要长期运行的流式连接，建议实现重连机制