1
// @/utils/server.ts
2

3
/**
4
 * 获取大模型响应流
5
 * @param completion 大模型响应流 await service.chat.completions.create 返回的流式数据 create({ stream: true })
6
 * @param model 模型名称
7
 * @param type 类型
8
 * @returns 流式数据
9
 */
10
export const getStreamData = (completion: Stream<ChatCompletionChunk>) => {
11
  let count = 0;
12
  let thinkingCount = 0;
13
  const stream = new ReadableStream({
14
    async start(controller) {
15
      try {
16
        console.log('Starting stream processing...');
17
        for await (const chunk of completion) {
18
          const { choices } = chunk;
19
          const delta = choices[0]?.delta as Delta ?? {};
20

21
          if (!delta) continue;
22

23
          const { reasoning_content = null, content = null } = delta;
24
          // 添加 thinking 标签
25
          if (count === 0 && reasoning_content) {
26
            controller.enqueue(new TextEncoder().encode('<thinking>'));
27
            // 等待 100ms 后避免批处理
28
            await new Promise(resolve => setTimeout(resolve, 100));
29
          }
30
          if (reasoning_content) {
31
            count++;
32
            thinkingCount++;
33
            controller.enqueue(new TextEncoder().encode(reasoning_content));
34
          }
35
          if (count - thinkingCount === 0 && thinkingCount !== 0 && content) {
36
            controller.enqueue(new TextEncoder().encode('</thinking>'));
37
            // 等待 100ms 后避免批处理
38
            await new Promise(resolve => setTimeout(resolve, 100));
39
          }
40
          if (content) {
41
            count++;
42
            controller.enqueue(new TextEncoder().encode(content));
43
          }
44
        }
45
      } catch (error) {
46
        console.error('Stream processing error:', error);
47
        controller.error(error);
48
      } finally {
49
        if (count === 0) {
50
          controller.enqueue(new TextEncoder().encode('[服务器繁忙，请稍后再试。]'));
51
        } else {
52
          controller.enqueue(new TextEncoder().encode('[DONE]'));
53
        }
54
        controller.close();
55
      }
56
    },
57
  });
58
  return stream;
59
};

1
import OpenAI from 'openai';
2
import { getStreamData } from '@/utils/server';
3

4
const openai = new OpenAI({
5
  baseURL: 'https://api.deepseek.com',
6
  apiKey: process.env.DEEPSEEK_KEY,
7
});
8

9
const completion = await openai.chat.completions.create({
10
  model: "deepseek-chat",
11
  messages: [{ role: "user", content: "Hello, world!" }],
12
  stream: true,
13
});
14
const stream = await getStreamData(completion);
15

16
// 返回流式数据
17
return new Response(stream, {
18
  headers: {
19
    'Content-Type': 'text/event-stream',
20
  },
21
});

1
// @/utils/client.ts
2

3
/**
4
 * 解析流式数据
5
 * @param stream 流式数据 ReadableStream
6
 * @param options 选项
7
 * @param options.output 输出类型 默认 text 如果 output 为 json 则onResult 的 result 返回解析后的 json 对象
8
 * @param options.onStart 流式数据开始
9
 * @param options.onEnd 流式数据结束 形参 thinking 为思考内容，result 为结果
10
 * @param options.onchange 实时更新最新内容 形参 thinking 为思考内容，result 为结果
11
 * @param options.onThinkingStart 如果是思考模型，则 onThinkingStart 会触发
12
 * @param options.onThinkingEnd 如果是思考模型，则 onThinkingEnd 会触发
13
 */
14
export const parseReadableStream = async (stream: ReadableStream<Uint8Array<ArrayBufferLike>>, options: {
15
  output?: 'text' | 'json';
16
  onStart?: () => void;
17
  onEnd?: (thinking: string, result: string | Record<string, unknown>) => void;
18
  onchange?: (thinking: string, result: string) => void;
19
  onThinkingStart?: () => void;
20
  onThinkingEnd?: () => void;
21
}) => {
22
  const { output = 'text', onStart = () => {}, onEnd = () => {}, onThinkingStart = () => {}, onThinkingEnd = () => {}, onchange = () => {} } = options;
23
  const reader = stream?.getReader();
24
  const decoder = new TextDecoder();
25
  let isReasoning = false;
26
  let thinking = '';
27
  let content = '';
28
  onStart();
29
  while (true) {
30
    const { done, value } = await reader!.read();
31
    if (done) break;
32
    const text = decoder.decode(value);
33
    if (text.includes('<thinking>')) {
34
      isReasoning = true;
35
      onThinkingStart();
36
      continue;
37
    }
38
    if (text.includes('</thinking>')) {
39
      isReasoning = false;
40
      onThinkingEnd();
41
      continue;
42
    }
43
    if (text.includes('[DONE]')) {
44
      let result = content;
45
      if (output === 'json') {
46
        // 取出 ```json 和 ``` 之间的内容
47
        const jsonContent = result.match(/```json\s*([\s\S]*?)\s*```/)?.[1] || '';
48
        try {
49
          result = JSON.parse(jsonContent);
50
        } catch (error) {
51
          console.error(error);
52
        }
53
      }
54
      onEnd(thinking, result);
55
      break;
56
    }
57
    if (isReasoning) {
58
      thinking += text;
59
    } else {
60
      content += text;
61
    }
62
    onchange(thinking, content);
63
  }
64
};

1
import { parseReadableStream } from '@/utils/client';
2

3
const stream = await fetch('http://localhost:3000/sse', {
4
  method: 'GET',
5
}).then(res => res.body);
6

7
try {
8
  parseReadableStream(stream, {
9
  output: 'json',
10
  onStart: () => {
11
    console.log('开始');
12
  },
13
  onEnd: (thinking, result) => {
14
    console.log('结束', thinking, result);
15
  },
16
  onchange: (thinking, result) => {
17
    // 如果 output 为 json 则 result 为 已经解析后对象
18
    console.log('实时更新', thinking, result);
19
  },
20
  onThinkingStart: () => {
21
    console.log('思考开始');
22
  },
23
    onThinkingEnd: () => {
24
      console.log('思考结束');
25
    },
26
  });
27
} catch (error) {
28
  console.error(error);
29
}