Fetch处理大模型流式数据请求与解析

为什么有的大模型可以一次返回多个 data？

Server-Sent Events (SSE)：允许服务器连续发送多个 data: 行，每个代表一个独立的数据块。

流式响应：大模型服务通常以流式响应方式返回数据，提高响应速度。

批量处理：服务器可能将多个数据块打包发送，减少网络开销。

一、问题分析

从上图中只能简易看出这种数据特点前两个特点：

分段返回
每段有多条类似json数据
每段的最后一条json数据有可能不完整

二、解决方案

/*** 异步函数 fetchEventGpt，用于向服务器发送获取 GPT 数据的请求** @param data 要发送的数据对象* @returns 返回响应体内容*/
const fetchEventGpt = async (data: any) => {const response = await fetch('/getGPT', {method: 'Post',headers: { 'Content-Type': 'application/json' },body: JSON.stringify(data),})// 等待直到请求完成console.log(response)return response.body
}/*** 异步解析流数据并返回 Response 对象。** 该函数通过 fetchEventGpt 发送请求，获取 ReadableStream 流数据，然后通过 ReadableStream API 进行解析。* 解析过程中，会将流数据分片并尝试解析 JSON 对象，处理可识别内容，并将结果通过 controller.enqueue 返回。* 如果解析结束，会清空缓存区并关闭连接。** @return {Response} Response 对象，包含解析后的流数据*/
const parseStream = async () => {const response = await fetchEventGpt({messages: [{role: 'user',content: '核能',},{role: 'assistant',content:'<br/>根据您提供的文章内容，以下是对“核能”的相关信息的整理：<br/><br/>1. 知识产权归属 ：本文件及其附件的知识产权属于核电运行研究（上海）有限公司和中核（）供应链管理有限公司，未经书面许可不得擅自使用。<br/><br/>2. 民用核设施事故责任 ：根据《民法典》第一千二百三十七条，民用核设施或运入运出核设施的核材料发生事故造成他人损害的，营运单位应承担侵权责任。但是，如果损害是由战争、武装冲突、暴乱等情形或者受害人故意造成的，则营运单位不承担责任。<br/><br/>3. 招标代理机构招标人 ：本次招标的代理机构为中核（上海）供应链管理有限公司，招标人为核电运行研究（上海）有限公司。<br/><br/>4. 招标投标系统检测 ：根据第九条，实验室应按照《招标投标系统检测技术规范》等进行检测，检测内容包括数据项、业务规则、功能、接口、性能、安全性、可靠性、易用性、运行环境等，并对相关文档进行审核。<br/><br/>5. 招标项目目标 ：项目的目标是突破微波干燥处理核电厂放射性废物的关键技术，研制微波干燥成套装备，解决核电厂放射性废物处理难题。<br/><br/>6. 投标文件提交方式 ：电子投标文件应通过中核集团电子采购平台提交，纸质版投标文件提交地点为中核（上海）供应链管理有限公司。<br/><br/>以上内容涉及核能相关的法律法规、招标流程、技术规范等方面，希望能对您。 ',},{role: 'user',content: '核能',},],classify: '',}).catch(() => {return new Error('请求失败')})if (!(response instanceof ReadableStream)) {return new Error('请求失败')}const reader = response.getReader() as anyconst decoder = new TextDecoder('utf-8')const encoder = new TextEncoder()let jsonBuffer = ''const readableStream = new ReadableStream({async start(controller) {function push() {reader.read().then(({ done, value }: any) => {if (done) {controller.close()return}// 1、流返回的块数据const chunk = decoder.decode(value, { stream: true })// 2、更新到缓存区jsonBuffer += chunk// 3、尝试分片解析jsonlet boundaryIndex = 0// 当前片内容let result = ''while ((boundaryIndex = jsonBuffer.indexOf('\n')) >= 0) {// 3.1 数据块切片const jsonString = jsonBuffer.slice(0, boundaryIndex)// 3.2 更新缓存区jsonBuffer = jsonBuffer.slice(boundaryIndex + 2)try {const jsonStr = jsonString.replace('data:', '')const jsonObject = JSON.parse(jsonStr) // 解析 JSON// 处理可识别内容 - 伪代码，根据实际对象处理const content = jsonObject?.data?.contentcontroller.enqueue(encoder.encode(content))// 解析结束 - 我们业务是根据此字段标识，根据实际情况调整if (jsonObject?.data?.isEnd === true) {// 清空缓存区jsonBuffer = ''break}} catch (error) {console.log('json解析出错', error)}}if (jsonBuffer) {try {const jsonObject = JSON.parse(jsonBuffer)} catch (error) {console.log(error)}}push()}).catch((err: any) => {controller.error(err)})}push()},})return new Response(readableStream)
}