javascript - 如何通过 API 将流数据发送到 Nuxt3 应用程序？

Question

您好，我一直在制作一个 API 以将分块数据实时返回到 Nuxt3 前端。我实际上正在尝试实现此 git 代码在我的 Nuxt3 应用程序中为 Next.js 实现的 ChatGPT API 响应流。

https://github.com/shivanshubisht/chatgpt

以下是我试图实现的数据流。

(a)Nuxt3前端<->(b)在Nuxt3上运行的包装API<->(c)在硝基上运行的ChatGPT API处理程序<->(d)ChatGPT API

环境:“nuxt”:“^3.2.3”，Windows 11，node.js 16

我在 (b) 包装器和 (c) 处理程序之间进行数据传输时遇到问题，而 (c) 处理程序本身可以完美地从 ChatGPT API 实时提取流数据。似乎存在以下问题:(c) 返回从 ChatGPT 接收到的相同分块数据，或/和 (b) 从流中的 (c) 接收数据。不过，我想两者都有问题。

以下是我认为有问题的代码:

(b) 包装 API:/server/api/response.post.ts

import { OpenAIStream, OpenAIStreamPayload } from '@/server/api/openaistream'

type RequestData = {
  currentModel: string
  message: string
}

export default defineEventHandler(async (event) => {
  const { message } = await readBody(event);

  const payload: OpenAIStreamPayload = {
    model: 'gpt-3.5-turbo',
    messages: [{ role: 'user', content: message }],
    temperature: 0.7,
    top_p: 1,
    frequency_penalty: 0,
    presence_penalty: 0,
    max_tokens: 500,
    stream: true,
    n: 1,
  }

  const stream = await OpenAIStream(payload);
  return new Response(stream);
})

(c) ChatGPT 处理程序:/server/api/openaistream.ts

import {
  createParser,
  ParsedEvent,
  ReconnectInterval,
} from 'eventsource-parser';

export type ChatGPTAgent = 'user' | 'system';
export interface ChatGPTMessage {
  //abbreviated
}
export interface OpenAIStreamPayload {
  //abbreviated
}

export async function OpenAIStream(payload: OpenAIStreamPayload) {
  const encoder = new TextEncoder();
  const decoder = new TextDecoder();

  let counter = 0;

  const res = await fetch('https://api.openai.com/v1/chat/completions', {
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${process.env.OPENAI_API_KEY ?? ''}`,
    },
    method: 'POST',
    body: JSON.stringify(payload),
  });

  const stream = new ReadableStream({
    async start(controller) {
      // callback
      function onParse(event: ParsedEvent | ReconnectInterval) {
        if (event.type === 'event') {
          const data = event.data;
          // https://beta.openai.com/docs/api-reference/completions/create#completions/create-stream
          if (data === '[DONE]') {
            console.log("END!");
            controller.close();
            return;
          }
          try {
            const json = JSON.parse(data);
            const text = json.choices[0].delta?.content || '';
            // check if it receives chunked data from chatgpt API
            console.log(text);
            if (counter < 2 && (text.match(/\n/) || []).length) {
              // this is a prefix character (i.e., "\n\n"), do nothing
              return;
            }
            const queue = encoder.encode(text);
            controller.enqueue(queue);
            counter++;
          } catch (e) {
            // maybe parse error
            controller.error(e);
          }
        }
      }

      for await (const chunk of res.body as any) {
        //check if it can decode the chunk data to send as stream data
        console.log(decoder.decode(chunk))
        parser.feed(decoder.decode(chunk));
      }
    },
  });

  return stream;
}

我猜想 Nuxt3 的 h3/硝基在处理来自 Next.js/Vercel 的流数据时可能会有所不同，但由于资源很少，我无法弄清楚它应该如何工作。有人可以分享任何想法或建议吗？谢谢!

API 处理程序正确地将分块流数据实时发送到在 nitro 上运行的另一个 API。

Answer 1

自行解决。 Next.js 的另一个包以某种方式工作: https://github.com/gptlabs/openai-streams

sendStream、fetch 和 getReader 的组合完美运行。下面是示例代码，供Nuxt用户将来引用。

/server/api/text.get.js

import { OpenAI } from "openai-streams/node";

export default defineEventHandler(async (event) => {
  const decoder = new TextDecoder('utf-8');
  console.log("received request!");
  const stream = await OpenAI(
    "chat",
    {
      model: "gpt-3.5-turbo",
      messages: [
        { "role": "user", "content": "test input" }
      ],
      stream: true
    },
  );
  return sendStream(event, stream);
});

---

/test.vue

<script>
const keyword = ref('');
var generateText = ref('');

const handleClick = async () => {
  const completion = await fetch('/api/test');

  const reader = completion.body.getReader();
  const decoder = new TextDecoder('utf-8');
  const read = async () => {
      const { done, value } = await reader.read();
      if (done) {
        console.log("release locked");
        return reader.releaseLock();
      }
      
      const chunk = decoder.decode(value, { stream: true });
      const temp = chunk.replace(/\}/g, '},'); //JSON.parse(JSON.stringify(chunk));
      const jsonData = temp
        .split(',')
        .map((data) => {
          const trimData = data.trim();
          console.log(trimData);
          if (trimData === '') return undefined;
          if (trimData === '{"content":"\n\n"}') return undefined;
          if (trimData === '[DONE]') return undefined;
          return trimData;
        })
        .filter((data) => data);
      
      var textOutput = ''

      for (let i = 0; i < jsonData.length; i++) {
        try {
          if (JSON.parse(jsonData[i]).content === "\n\n") {
            textOutput = '';
          } else {
            textOutput = JSON.parse(jsonData[i]);
          }
        } catch(e) {
          console.log(e);
        }

        if (textOutput.content !== undefined) {
          generateText.value = generateText.value + textOutput.content;
        }
      }
      return read();
    };
    await read();
}
</script>