手摸手带你实现前端乞丐版的 ChatGPT

本文首发于个人博客

前言

一开始看到 chatgpt 的流式渲染，有点好奇流式渲染是如何实现的，无意之间发现 vercel 的库：ai，仔细学习了它的代码，写的小而精，把它看明白了之后，想着写一篇文章来输出一下。

所以我就根据 ai 这个库，来一步步带大家来实现前端乞丐版 chatgpt ，把这篇文章看完之后，也可以去看看 ai 这个库，代码是写的真不错（会发现代码都是抄它的，哈哈哈哈）！！！

本文会使用 Next13，不熟悉也没关系，用到的 api 不多，不懂的 api 可以查看 Next13 文档。

注意：需要申请有一个 openai 的 apiKey，不然就无法调用接口哦。

源代码在这里

效果展示：

什么是网络流

先看看一个概念：网络流，平时可能用不到。流是一种用于访问数据的数据结构，比如说：文件、接口返回的数据等等。

使用流有两个好处：

• 可以处理大量数据，流可以将它们分成更小的部分(chunk)，可以一次处理一个(chunk)。
• 可以使用相同的数据结构、流，同时处理不同的数据，这使得代码变得更加复用。

在网络流中，一个 chunk 通常是：

• 文本流：string
• 二进制流：Uint8Arrays

流的分类

网络流主要有三种：

• ReadableStream：用于从数据源中读取数据。执行此操作的代码成为消费者。
• WritableStream：用户将数据写入。执行此操作的代码成为生产者。
• TransformStream 由两个流组成：
  • 它从其可写端（WritableStream）接收输入。
  • 它将输出发送到它的可读端，一个 ReadableStream。

本文中只会使用到 ReadableStream 和 TransformStream。

ReadableStream

ReadableStream 可以从各种来源读取数据块，类型声明如下：

interface ReadableStream<TChunk> {
  getReader(): ReadableStreamDefaultReader<TChunk>;
  pipeThrough<TChunk2>(
    transform: ReadableWritablePair<TChunk2, TChunk>,
    options?: StreamPipeOptions
  ): ReadableStream<TChunk2>;
  readonly locked: boolean;

  // 忽略
  // [Symbol.asyncIterator](): AsyncIterator<TChunk>;
  // cancel(reason?: any): Promise<void>;
  // pipeTo(
  //   destination: WritableStream<TChunk>,
  //   options?: StreamPipeOptions
  // ): Promise<void>;
  // tee(): [ReadableStream<TChunk>, ReadableStream<TChunk>];
}

这三个属性的作用是：

• getReader()：返回一个 Reader，可以从 ReadableStream 读取的对象，返回的 Reader 类似于迭代器。
• locked：每个 ReadableStream 一次只能有一个活动的 Reader，当 Reader 在使用时，ReadableStream 被锁定并且getReader()不能被调用。
• pipeThrough()：将其 ReadableStream 连接到 ReadableWritablePair(一个 TransformStream)。它返回一个新的 ReadableStream（类比一下：把它理解成一个数组的 map 方法）。

下面来看看 getReader() 的返回类型：

interface ReadableStreamGenericReader {
  cancel(reason?: any): Promise<void>;
  // 忽略
  // readonly closed: Promise<undefined>;
}

interface ReadableStreamDefaultReader<TChunk>
  extends ReadableStreamGenericReader {
  releaseLock(): void;
  read(): Promise<ReadableStreamReadResult<TChunk>>;
}

interface ReadableStreamReadResult<TChunk> {
  done: boolean;
  value: TChunk | undefined;
}

• cancel()：在一个活动的 Reader 中，这个方法取消关联的 ReadableStream。
• releaseLock()：停用 Reader 并解锁流。
• read()返回来两个属性的 ReadableStreamReadResult 的 Promise:
  • done：布尔值，false 表示可以读取，true 表示最后一个块。
  • value：块(chunk)。

师傅，别念 api 了，再念人都要傻了，赶紧来一个 demo 吧。

以下是通过 getReader 方式来读取 ReadableStream 的小例子。

const reader = readableStream.getReader(); // (A)
console.log(readableStream.locked); // true (B)
let result = '';

try {
  while (true) {
    const { done, value } = await reader.read(); // (C)
    if (done) {
      break;
    }
    result += value; // (D)
  }
} finally {
  reader.releaseLock(); // (E)
}

console.log('result', result);

• A 行：不能直接读取 readableStream，需要调用getReader()来获取 Reader。
• B 行：在 getReader() 之后，readableStream 被锁定，所以 B 行打印的是 true，如果想再次调用getReader()，必须调用releaseLock()(E 行)。
• C 行：read()返回属性 done 和 value，如果 done 为 true，表示是最后一个块，
• D 行：可以对返回的 value 进行操作，这里是将返回的 value 全部都加在一个字符串里。

通过包装将数据源转化为 ReadableStream

如果想通过 ReadableStream 读取外部源，可以将其包装在适配器对象中并将该对象传递给构造函数 ReadableStream。

new ReadableStream(underlyingSource?, queuingStrategy?)

以下是类型声明：

interface UnderlyingSource<TChunk> {
  start?(controller: ReadableStreamController<TChunk>): void | Promise<void>;
  // 忽略
  // pull?(controller: ReadableStreamController<TChunk>): void | Promise<void>;
  // cancel?(reason?: any): void | Promise<void>;
  // type: 'bytes' | undefined;
  // autoAllocateChunkSize: bigint;
}

• start(controller) 调用构造函数后立即调用 start 方法。

controller 的参数类型如下：

type ReadableStreamController<TChunk> =
  | ReadableStreamDefaultController<TChunk>
  | ReadableByteStreamController<TChunk>; //  先忽略

interface ReadableStreamDefaultController<TChunk> {
  enqueue(chunk?: TChunk): void;
  close(): void;
  // 忽略
  // readonly desiredSize: number | Null;
  // error(err?: any): void;
}

• enqueue(chunk)：添加 chunk 到 ReadableStream 的内部队列。
• close()：关闭 ReadableStream，消费者仍然可以清空队列，在那之后，流结束。

自定义 ReadableStream demo

async function test14() {
  const readableStream = new ReadableStream({
    start(controller) {
      controller.enqueue('First Line\n'); // (A)
      controller.enqueue('Second Line\n'); // (B)
      controller.close(); // (C)
    },
  });

  for await (const chunk of readableStream) {
    console.log(chunk);
  }
}
test14();

// First Line
// Second Line

ReadableStream 是异步可迭代的，可以使用for-await-of来进行迭代。

使用控制器创建一个包含两个块的流（A 和 B 行），关闭流（C 行）很重要，否则for-await-of永远不会结束。

TransformStream

转化流：

• 通过其可写端（WritableStream）接受输入。
• 然后它可能会或可能不会转换输入。
• 结果可以通过 ReadableStream 读取，它是可读的。

使用TransformStream最常见的方式是pipeThrough。

const transformStream = readableStream.pipeThrough(transformStream);

.pipeThrough()将 readableStream 传输到 transformStream 的可写端，并进行转换返回其可读端。

换句话说: 创建了一个新的 ReadableStream，它是 ReadableStream 的转换版本，类似于数组的 map。

一个简单的 demo：

async function test21() {
  const encoder = new TextEncoder();
  const readableByteStream = new ReadableStream({
    start(controller) {
      controller.enqueue(encoder.encode('hello\n'));
      controller.enqueue(encoder.encode('world\n'));
    },
  });
  const readableStream = readableByteStream.pipeThrough(
    new TextDecoderStream('utf-8')
  );
  for await (const stringChunk of readableStream) {
    console.log(stringChunk);
  }
}
test21();

TextEncoder.encode()：将字符串作为输入，并返回 Uint8Array 包含 UTF-8 编码的文本。

使用了内置 TransformStream：TextDecoderStream()，作用就是将接收到的二进制流转换为可读的文本流(Uint8Array -> string)。

自定义 TransformStream

跟上面的 ReadableStream 类似，如果要自定义 TransformStream，也可以传递适配器对象给构造函数 TransformStream。

它具有以下类型:

interface TransformStream<InChunk, OutChunk> {
  start?(
    controller: TransformStreamDefaultController<OutChunk>
  ): void | Promise<void>;
  transform?(
    chunk: InChunk,
    controller: TransformStreamDefaultController<OutChunk>
  ): void | Promise<void>;
  // 忽略
  // flush?(
  //   chunk: InChunk,
  //   controller: TransformStreamDefaultController<OutChunk>
  // ): void | Promise<void>;
}

上面属性的解释：

• start(controller)：在调用构造函数后立即被调用，可以在转换之前做一些准备。
• transform(chunk, controller) 执行实际的转化 。接受一个输入块，并可以使用 controller 将一个或多个转换后的输出块排队。

该 contrller 具有以下类型：

interface TransformStreamDefaultController<OutChunk> {
  enqueue(chunk?: OutChunk): void;
  terminate(): void;
  // 忽略
  // readonly desiredSize: number | null;
  // error(err?: any): void;
}

• enqueue(chunk)：添加 chunk 到 TransformStream 的可读端（输出）。
• terminate()：关闭 TransformStream 的可读端（输出）并在可写端（输入）出错。如果转换器对可写端（输入）的剩余块不感兴趣并想跳过他们，则可以使用它。

小 demo

说了一大堆 api，来一个简单的例子：

async function test20() {
  // 创建一个 ReadableStream 对象
  const readableStream = new ReadableStream({
    start(controller) {
      controller.enqueue('hello');
      controller.enqueue('world');
      controller.close();
    },
  });

  // 创建一个 TransformStream 对象
  const transformer = new TransformStream({
    transform(chunk, controller) {
      // 对输入数据进行转换处理
      const transformedChunk = chunk.toUpperCase();
      // 将转换后的数据通过 controller.enqueue() 方法推送到输出流 ReadableStream
      controller.enqueue(transformedChunk);
    },
  });

  // 通过 TransformStream 进行处理流的转换
  const newReadableStream = readableStream.pipeThrough(transformer);

  // 使用 reader 方式来读取 ReadableStream
  const reader = newReadableStream.getReader();

  while (true) {
    const { done, value } = await reader.read();
    if (done) {
      break;
    }
    console.log(value); // HELLO WORLD
  }
}

test20();

重新回顾 fetch API

Fetch API 是一种用于获取和发送网络资源的现代 Web API。它提供了一种替代 XMLHttpRequest 的方式，可以更简单、更灵活地进行网络请求。

Fetch API 使用 Promise 对象来返回请求结果，可以轻松地将其与 async/await 结合使用。

简单的小例子：

async function main() {
  const res = await fetch('https://example.com');
  const data = await res.json();
  console.log('data', data);
}
main();

这里要用的是 fetch 返回的属性body，它返回的是ReadableStream。

async function main() {
  const res = await fetch('https://example.com');
  const body = res.body; // ReadableStream

  const reader = body.getReader();

  while (true) {
    const { done, value } = await reader.read();
    if (done) {
      break;
    }
    console.log('value', value);
  }
}
main();

用到了上面提到的 ReadableStream，也算是简单的回顾一下。

需要注意的是：fetch.body返回的是二进制流，后面会再提到。

开始干活

讲到这里，终于把前置的知识熟悉一下，我知道你很急，但是你先别急。

我们来进入实战环节。

初始化项目

首先使用pnpm create next-app初始化一个 Next13 项目。

• What is your project named? openai-stream
• Would you like to use TypeScript with this project? Yes
• Would you like to use ESLint with this project? Yes
• Would you like to use Tailwind CSS with this project? Yes
• Would you like to use src/ directory with this project? No
• Use App Router (recommended)? Yes
• Would you like to customize the default import alias? No

初始化之后就会安装 TypeScript、Eslint 和 Tailwind CSS。

当个 UI 仔也挺好

第一步就开始画 UI。

UI 的话主要有两个部分：

1. 消息列表展示。
2. input 输入框。

MessageCard 渲染信息

创建类型文件type.ts定义关于 message 的类型。

// types.ts
export type Message = {
  id: string;
  createdAt?: Date;
  content: string;
  role: 'system' | 'user' | 'assistant';
};

MessageCard 组件用来渲染输入和 openai 返回的信息。

import { Message } from '@/types';
import classNames from 'classnames';

interface MessageCardProps {
  message: Message;
}

type AvatarProps = Pick<Message, 'role'>;

const Avatar = ({ role }: AvatarProps) => {
  const getName = () => (role === 'user' ? 'U' : 'AI');

  return (
    <span
      className={classNames(
        'w-6 h-6 inline-flex items-center justify-center rounded-full min-w-[24px]',
        role === 'user' ? 'bg-orange-300' : 'bg-green-400'
      )}
    >
      {getName()}
    </span>
  );
};

const MessageCard = ({ message }: MessageCardProps) => {
  return (
    <div key={message.id} className="flex items-center">
      <Avatar role={message.role} />
      <div className="ml-2">{message.content}</div>
    </div>
  );
};

export default MessageCard;

基础页面 + input 输入框

下一步画基础的页面和 input 输入框。

直接在app/page.tsx里面书写即可。

// app/page.tsx
'use client';

import MessageCard from './MessageCard';
import { Message } from '@/types';

const Chat = () => {
  const messages: Message[] = [
    { id: '1', content: 'hello', role: 'user' },
    { id: '2', content: 'world', role: 'assistant' },
  ];
  return (
    <div className="flex h-full flex-col w-full max-w-xl pb-36 pt-9 mx-auto stretch">
      <ul className="space-y-4">
        {messages.map((message) => (
          <MessageCard key={message.id} message={message} />
        ))}
      </ul>

      <form
        onSubmit={(e) => {
          e.preventDefault();
        }}
      >
        <input
          className="w-full p-3 focus-visible:outline-gray-300 border border-gray-300 rounded-md shadow-xl focus:shadow-2xl transition-all"
          placeholder="随便说点什么..."
        />
      </form>
    </div>
  );
};

export default Chat;

先 mock 消息列表，看看展示效果咋样。

处理请求

在 app 目录下新建文件app/api/chat/route.ts，用来处理 api 请求。

// app/api/chat/route.ts
// GET 是书写规范!!!
export async function GET() {
  const data = 'hello world';

  return NextResponse.json({ data });
}

可以直接访问 http://localhost:3000/api/chat 可以看到返回的数据。

关于 Next.js 的 Route 可以查看相关文档，不过多赘述。

需要发送 POST 请求将 message 传递给 openai，通过export function POST就可以处理 POST 请求。

// app/api/chat/route.ts
export const runtime = 'edge'; // 如果要流式渲染则需要加上这一行

export async function POST(req: Request) {
  const stream = AIStream();
  console.log('stream:', stream);
  return new StreamingTextResponse(stream);
}

libs/streaming-text-response.ts就是对于 Response 的简单封装，将状态码置为 200。

// libs/streaming-text-response.ts
export class StreamingTextResponse extends Response {
  constructor(res: ReadableStream, init?: ResponseInit) {
    super(res as any, {
      ...init,
      status: 200,
      headers: {
        'Content-Type': 'text/plain; charset=utf-8',
        ...init?.headers,
      },
    });
  }
}

创建文件libs/ai-stream.ts用来处理网络流，使用 ReadableStream 先 mock 两条数据。

// libs/ai-stream.ts
export function AIStream(): ReadableStream {
  const stream = new ReadableStream({
    start(controller) {
      controller.enqueue('hello\n');
      controller.enqueue('world\n');
      controller.close();
    },
  });

  return stream;
}

hook: use-chat

下一步写一个 hook 来进行页面交互，创建文件hooks/use-chat.ts。

装包

需要安装一些依赖包：nanoid 和 swr。

• nanoid 是可以生成唯一 ID 的库。
• swr 是用于数据请求和缓存的库。

pnpm add nanoid swr

类型

将 CreateMessage 和 UseChatOptions 添加到types.ts文件，定义好 use-chat 的类型声明。

// types.ts
// 省略其他代码...
export type CreateMessage = {
  id?: string;
  createdAt?: Date;
  content: string;
  role: 'system' | 'user' | 'assistant';
};

// type.ts
export type UseChatOptions = {
  // 指定聊天功能的API地址，默认为'/api/chat'
  api?: string;

  // 指定聊天的唯一标识符，如果没有指定，则使用useId生成唯一的Hook ID
  id?: string;

  // 消息列表初始化内容，默认为空数组
  initialMessages?: Message[];

  // 输入框的初始内容，默认为空字符串
  initialInput?: string;
};

// hooks/use-chat.ts
export type UseChatHelpers = {
  // 当前消息列表
  messages: Message[];

  // 请求过程中可能出现的错误
  error: undefined | Error;

  // 向聊天中追加一条消息
  append: (
    message: Message | CreateMessage
  ) => Promise<string | null | undefined>;

  // 设置聊天消息列表
  setMessages: (messages: Message[]) => void;

  // 输入框内容
  input: string;

  // 设置输入框内容
  setInput: Dispatch<SetStateAction<string>>;

  // 处理输入框内容变化
  handleInputChange: (
    e: ChangeEvent<HTMLInputElement> | ChangeEvent<HTMLTextAreaElement>
  ) => void;

  // 处理表单提交
  handleSubmit: (e: FormEvent<HTMLFormElement>) => void;

  // 请求是否正在加载中
  isLoading: boolean;
};

export function useChat(options?: UseChatOptions): UseChatHelpers;

use-chat 具体逻辑

下面来写对应的 use-chat 具体逻辑：

export function useChat({
  api = '/api/chat',
  id,
  initialInput = '',
  initialMessages = [],
}: UseChatOptions = {}): UseChatHelpers {
  const hookId = useId();
  // 生成一个chatId
  const chatId = id || hookId;

  const { data, mutate } = useSWR<Message[]>([api, chatId], null, {
    fallbackData: initialMessages,
  });
  const messages = data!;

  // 用 ref 保存最新的消息列表
  const messagesRef = useRef<Message[]>(messages);
  useEffect(() => {
    messagesRef.current = messages;
  }, [messages]);

  // 处理input输入
  const [input, setInput] = useState(initialInput);

  const handleInputChange = useCallback(
    (e: ChangeEvent<HTMLInputElement> | ChangeEvent<HTMLTextAreaElement>) => {
      setInput(e.target.value);
    },
    []
  );

  // 设置消息列表
  const setMessages = useCallback(
    (messages: Message[]) => {
      mutate(messages, false);
      messagesRef.current = messages;
    },
    [mutate]
  );

  return {
    messages,
    setMessages,
    input,
    setInput,
    handleInputChange,
  };
}

使用useSWR来声明一个状态，第二个参数(fetcher)传入 null，表示不需要进行网络请求，可以把它当成本地状态来处理，返回的 mutate 函数可以对这个状态进行更新。

小小的工具函数

新建utils.ts文件用来保存两个工具函数: nanoid、createChunkDecoder。

• nanoid：使用 customAlphabet 函数创建了自定义的 ID 生成器 nanoid，用于生成唯一 ID。
• createChunkDecoder：用于将 Uint8Array 类型的数据块解码成字符串（Uint8Array -> string）。

// utils.ts
import { customAlphabet } from 'nanoid';

export const nanoid = customAlphabet(
  '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz',
  7
);

export function createChunkDecoder() {
  const decoder = new TextDecoder();
  return function (chunk: Uint8Array | undefined): string {
    if (!chunk) return '';
    return decoder.decode(chunk, { stream: true });
  };
}

请求接口

接下来就是发送网络请求到/api/chat，请求成功之后将数据渲染出来。

export function useChat({
  api = '/api/chat',
  id,
  initialInput = '',
  initialMessages = [],
}: UseChatOptions = {}): UseChatHelpers {
  // 省略其它代码...

  const { error, trigger, isMutating } = useSWRMutation<
    string | null,
    any,
    [string, string],
    Message[]
  >(
    [api, chatId],
    async (_, { arg: messagesSnapshot }) => {
      try {
        const abortController = new AbortController();
        abortControllerRef.current = abortController;

        // 保存上一次的消息列表，用于时光回溯
        const previousMessages = messagesRef.current;
        // 先更新UI
        mutate(messagesSnapshot, false);

        // 处理参数
        const body = messagesSnapshot.map(({ role, content }) => ({
          role,
          content,
        }));

        const res = await fetch(api, {
          method: 'POST',
          body: JSON.stringify({ messages: body }),
          signal: abortController.signal,
        }).catch((err) => {
          // 如果报错了，回退到上一次的消息列表
          mutate(previousMessages, false);
          throw err;
        });

        if (!res.ok) {
          // 如果接口请求不成功，回退到上一次的消息列表
          mutate(previousMessages, false);
          throw new Error(
            (await res.text()) || 'Faild to fetch the chat response.'
          );
        }

        // body为空，直接报错
        if (!res.body) {
          throw new Error('The response body is empty.');
        }

        let result = '';
        const createdAt = new Date();
        // 创建唯一的消息ID
        const replyId = nanoid();
        // 使用 reader 的方式读取ReadableStream
        const reader = res.body.getReader();
        // 使用 TextDecoder 进行将 Uint8Array 解码成 string
        const decode = createChunkDecoder();

        while (true) {
          const { done, value } = await reader.read();
          if (done) {
            break;
          }
          // 将二进制value 解析成字符串，然后进行拼接
          result += decode(value);
          // 及时更新UI
          mutate(
            [
              ...messagesSnapshot,
              {
                id: replyId,
                createdAt,
                content: result,
                role: 'assistant',
              },
            ],
            false
          );

          // 如果请求取消了，则需要暂停读取stream
          if (abortControllerRef.current === null) {
            reader.cancel();
            break;
          }
        }

        abortControllerRef.current = null;
        return result;
      } catch (err) {
        if ((err as any).name === 'AbortError') {
          abortControllerRef.current = null;
          return null;
        }
        throw err;
      }
    },
    {
      revalidate: false, // 不需要重新验证缓存
    }
  );

  const append = useCallback(
    async (message: Message | CreateMessage) => {
      if (!message.id) {
        message.id = nanoid();
      }
      // 将消息添加到消息列表，手动触发接口请求
      return trigger(messagesRef.current.concat(message as Message));
    },
    [trigger]
  );

  const handleSubmit = useCallback(
    (e: FormEvent<HTMLFormElement>) => {
      e.preventDefault();
      // 如果输入框没有内容，直接返回
      if (!input) {
        return;
      }
      // 将内容添加到消息列表中
      append({
        content: input,
        role: 'user',
        createdAt: new Date(),
      });
      setInput('');
    },
    [append, input]
  );

  return {
    error,
    append,
    handleSubmit,
    isLoading: isMutating,
  };
}

事件处理

接下来就是将上面的 hook 逻辑和视图绑定在一起。

// app/page.tsx
const Chat = () => {
  const { messages, input, handleSubmit, handleInputChange } = useChat();
  return (
    <div className="flex h-full flex-col w-full max-w-xl pb-36 pt-9 mx-auto stretch">
      <ul className="space-y-4">
        {messages.map((message) => (
          <MessageCard key={message.id} message={message} />
        ))}
      </ul>

      <form onSubmit={handleSubmit}>
        <input
          value={input}
          onChange={handleInputChange}
          className="w-full p-3 focus-visible:outline-gray-300 rounded shadow-xl focus:shadow-2xl transition-all"
          placeholder="随便说点什么..."
        />
      </form>
    </div>
  );
};
export default Chat;

连通性验证

不出意外的报 bug 了。

问题在于fetch.body方法返回的是一个二进制流（Uint8Array），fetch 提供了text()、json()、blob()等方式将二进制流转化为其他数据格式。

刚刚在libs/ai-stream.ts里面推到队列里面的是字符串，所以就会报错。

export function AIStream(): ReadableStream {
  const stream = new ReadableStream({
    start(controller) {
      controller.enqueue('hello\n'); // 字符串
      controller.enqueue('world\n'); // 字符串
      controller.close();
    },
  });
  return stream;
}

有两种解决方式。

1. enqueue 推到队列的类型改成二进制形式：

const textEncoder = new TextEncoder();
const stream = new ReadableStream({
  start(controller) {
    controller.enqueue(textEncoder.encode('hello\n'));
    controller.enqueue(textEncoder.encode('world\n'));
    controller.close();
  },
});

创建了 TextEncoder 对象，用于将字符串编码为 Uint8Array 对象。

2. 采用 TransformStream 可以对输入的数据进行转换处理：

export function createCallbacksTransformer() {
  const encoder = new TextEncoder();

  return new TransformStream<string, Uint8Array>({
    async transform(message, controller): Promise<void> {
      controller.enqueue(encoder.encode(message));
    },
  });
}

export function AIStream(): ReadableStream {
  const stream = new ReadableStream({
    start(controller) {
      controller.enqueue('hello\n');
      controller.enqueue('world\n');
      controller.close();
    },
  });

  return stream.pipeThrough(createCallbacksTransformer());
}

通过使用 pipeThrough 方法，将 AIStream 的输出流连接到 createCallbacksTransformer 的输入流，实现了数据的转换和传递。

实现将 string 转换为 Uint8Array 对象的流处理过程。

本文后面会使用第二种方式。

我们来看看效果：

🎉🎉🎉 完成了一大步 🎉🎉🎉。

连接 openai

在连接 openai 之前，需要先安装 openai-edge。

可以点击查看openai-edge具体的 api。

pnpm add openai-edge

接下来就是在app/api/chat/route.ts中连接 openai。

// app/api/chat/route.ts
const config = new Configuration({
  // 申请好的 OPENAI_API_KEY
  apiKey: process.env.OPENAI_API_KEY,
});
const openai = new OpenAIApi(config);

export const runtime = 'edge';

export async function POST(req: Request) {
  // 获得请求参数
  const { messages } = await req.json();

  const response = await openai.createChatCompletion({
    model: 'gpt-3.5-turbo',
    stream: true,
    messages: messages.map((message: any) => ({
      content: message.content,
      role: message.role,
    })),
  });

  // 将 response 传递给 AIStream 进行处理
  const stream = AIStream(response);
  return new StreamingTextResponse(stream);
}

这里process.env.OPENAI_API_KEY读取的是环境变量，我们可以新建一个.env.local的文件。

OPENAI_API_KEY=api key

新建完成之后需要重启一下服务，加载一下环境变量。

使用.env.local 命名是为了防止将该文件提交到 git，如果上传到 github 会暴露 apiKey。

在服务端请求 openai api，并将 openai 返回的 response 传递给了 AIStream，AIStream 也需要更新一下。

// libs/ai-stream.ts
export function AIStream(res: Response): ReadableStream {
  if (!res.ok) {
    throw new Error(
      `Failed to convert the response to stream. Received status code: ${res.status}.`
    );
  }

  const stream =
    res.body ||
    new ReadableStream({
      start(controller) {
        controller.close();
      },
    });

  return stream;
  // .pipeThrough(createCallbacksTransformer);
}

res.body返回的是ReadableStream<Uint8Array>，之前写的 pipeThrough(createCallbacksTransformer)是将 string -> Uint8Array，目前先用不到，先注释掉。

再来看看效果：

sse 数据

从 gif 图上看到返回"data: json 字符串"这样子的格式，这是 sse 数据格式。

sse 数据：每一次发送的信息，由若干个 message 组成，每个 message 之间用\n\n 分隔。

每个 message 内部由若干行组成，每一行都是如下格式。

[field]: value\n

上面的 field 可以取四个值。

data
event
id
retry

sse 数据格式可以看看阮一峰老师的这篇文章，写的还是挺详细的。

如何解析 sse 数据

sse 数据如何进行解析呢？ 可以使用eventsource-parser这个库，这个库就是用来解析 sse 数据的。

pnpm add eventsource-parser

fetch.body 返回的是 Uint8Array ，还是使用 TransformStream，在 TransformStream 中进行 sse 数据的解析。

1. 创建 TransformStream 将 Uint8Array 格式先转成 string 格式，解析 sse 数据之后再进一步把 content 读取出来。
2. 然后延用上面的 createCallbacksTransformer 将 string 格式再转成 Uint8Array 格式。

先新建函数createEventStreamTransformer：

// libs/ai-stream.ts
import {
  createParser,
  type EventSourceParser,
  type ParseEvent,
} from 'eventsource-parser';

export function createEventStreamTransformer(customParser: AIStreamParser) {
  const decoder = new TextDecoder();
  let parser: EventSourceParser;

  return new TransformStream<Uint8Array, string>({
    async start(controller) {
      function onParse(event: ParseEvent) {
        if (event.type === 'event') {
          const data = event.data; // data 就是解析之后的每个数据
          if (data === '[DONE]') {
            // 如果是[DONE] 表示解析完毕。
            controller.terminate();
            return;
          }
          // 将 json 数据字符串传入到自定义解析器进行过滤
          const message = customParser(data);
          // 如果 message 有值的话，就推到队列中。
          if (message) controller.enqueue(message);
        }
      }

      // 创建解析器
      parser = createParser(onParse);
    },
    transform(chunk) {
      parser.feed(decoder.decode(chunk));
    },
  });
}

在 start 方法中，创建了 parser，用来解析数据流中的 sse 数据。

在 transform 方法中，此时 chunk 是 Uint8Array 类型，先使用 TextDecoder 将 Uint8Array 解析成字符串，然后通过 feed 进行 sse 数据的解析，解析后的数据传递给 onParse 函数。

这里为了逻辑解耦，使用了 customParser 自定义解析器，传入 customParser 来实现解析具体的逻辑（下面会介绍）。

流可以使用多次 pipeThrough，进行多次转化。

// libs/ai-stream.ts
export function AIStream(res: Response): ReadableStream {
  // 省略其他代码...

  return stream
    .pipeThrough(createEventStreamTransformer())
    .pipeThrough(createCallbacksTransformer());
}

• 第一次 pipeThrough：将 Uint8Array -> string。
  • 具体的这一步流程是：Uint8Array -> eventsource-parser 解析 sse 数据 -> 走到 onParse 函数 -> customParser(sse 数据 json 字符串) -> string
• 第二次 pipeThrough：将第一次 pipeThrough 的结果 string -> Uint8Array。

AIStream 最终返回的依然是 Uint8Array 格式的流。

自定义解析器

createEventStreamTransformer方法传入了自定义解析器，customParser 该怎么写呢？

libs/ai-stream.ts 文件就只负责流的处理，将 customParser 作为参数传入即可，具体如何解析取决于上层应用。

// libs/ai-stream.ts
export function AIStream(
  res: Response,
  customParser: AIStreamParser // 新增
): ReadableStream {
  // 省略其他代码...

  return stream
    .pipeThrough(createEventStreamTransformer(customParser)) // 传入
    .pipeThrough(createCallbacksTransformer());
}

新建文件libs/openai-stream.ts实现 openai 流的 customParser。

// libs/openai-stream.ts
import { AIStream } from './ai-stream';

export function trimStartOfStreamHelper() {
  let start = true;
  return (text: string) => {
    if (start) text = text.trimStart();
    if (text) start = false;
    return text;
  };
}

function parseOpenAIStream(): (data: string) => string | void {
  const trimStartOfStream = trimStartOfStreamHelper();

  return (data) => {
    const json = JSON.parse(data); // 将 json 字符串解析成对象

    const text = trimStartOfStream(
      json.choices[0]?.delta?.content ?? json.choices[0]?.text ?? ''
    ); // 读取对应的字段

    return text;
  };
}

export function OpenAIStream(res: Response): ReadableStream {
  return AIStream(res, parseOpenAIStream());
}

一个消息头部可能有多个空格，使用trimStartOfStreamHelper辅助函数把 chunk 最前面的空格给去掉。

因为 data 传入的时候已经是 sse 的数据部分，是 json 字符串，可以使用JSON.parse的方式来解析成对象，最后在读取相对应的字段即可。

再更新一下app/api/chat/route.ts文件，将 AIStream 替换成 OpenAIStream。

// app/api/chat/route.ts
export async function POST(req: Request) {
  // 省略其他代码...
  const stream = OpenAIStream(response); // AIStream() -> OpenAIStream()
  return new StreamingTextResponse(stream);
}

再来看看效果

🎉🎉🎉 牛哇，实现啦！！！ 🎉🎉🎉

扩展

实现 regenerate

如果对当前生成的结果不满意，重新生成新的结果。

// hooks/use-chat.ts
export type UseChatHelpers = {
  reload: () => Promise<string | null | undefined>;
};

export function useChat({
  api = '/api/chat',
  id,
  initialInput = '',
  initialMessages = [],
}: UseChatOptions = {}): UseChatHelpers {
  // 省略其他代码...

  const reload = useCallback(async () => {
    if (messagesRef.current.length === 0) return null;

    const lastMessage = messagesRef.current[messagesRef.current.length - 1];
    // 如果最后一条消息是 chatgpt 生成的
    if (lastMessage.role === 'assistant') {
      // 去掉消息列表的最后一条消息，然后触发接口请求
      return trigger(messagesRef.current.slice(0, -1));
    }
    return trigger(messagesRef.current);
  }, [trigger]);

  return {
    // 省略其他的代码...
    reload,
  };
}

新增 react-feather 添加几个好看的图标，顺便也把 input 输入框美化一下。

// app/page.tsx
// 省略其他代码...
import { Pause, Send, RotateCw, MoreHorizontal } from 'react-feather';
import classNames from 'classnames';

const Chat = () => {
  // 省略其他代码...
  const { reload, messages } = useChat();

  const disabledClassName = isLoading
    ? 'cursor-not-allowed pointer-events-none opacity-70'
    : '';

  return (
    // 省略其他代码...
    <div className="fixed w-full left-0 bottom-0 py-4 bg-gray-100 border-t border-t-gray-300">
      {messages.length > 0 ? (
        <button
          className="mb-2 mx-auto border border-gray-300 bg-gray-100 text-gray-600 p-2 px-8 rounded-md hover:bg-gray-200 transition-all flex items-center"
          onClick={reload}
        >
          <RotateCw className="mr-2" size={16} />
          重新生成
        </button>
      ) : null}
      <form
        onSubmit={handleSubmit}
        className="max-w-xl w-full mx-auto relative"
      >
        <div className="relative">
          <input
            disabled={isLoading}
            value={input}
            onChange={handleInputChange}
            className={classNames(
              'w-full p-3 focus-visible:outline-gray-300 border border-gray-300 bg-gray-100 rounded-md shadow-xl focus:shadow-2xl transition-all',
              disabledClassName
            )}
            placeholder="随便说点什么..."
          />
          {isLoading ? (
            <button
              className={classNames(
                'absolute right-3 bg-gray-200 p-1 top-1/2 -translate-y-1/2 rounded max-w-xs transition-all',
                disabledClassName
              )}
            >
              <MoreHorizontal size={16} />
            </button>
          ) : (
            <button className="absolute right-3 bg-gray-200 hover:text-white p-1 top-1/2 -translate-y-1/2 rounded max-w-xs hover:bg-green-400 transition-all">
              <Send size={16} />
            </button>
          )}
        </div>
      </form>
    </div>
  );
};

看看具体的效果：

实现 stop

在 hooks/use-chat.ts 中使用了 AbortController 传递给了 fetch 函数，写一个 stop 方法来实现暂停。

// hooks/use-chat.ts
export type UseChatHelpers = {
  stop: () => void;
};

export function useChat({
  api = '/api/chat',
  id,
  initialInput = '',
  initialMessages = [],
}: UseChatOptions = {}): UseChatHelpers {
  // 省略其他代码...

  const stop = useCallback(() => {
    if (abortControllerRef.current) {
      // 取消请求
      abortControllerRef.current.abort();
      abortControllerRef.current = null;
    }
  }, []);

  return {
    // 省略其他的代码...
    stop,
  };
}

如果在生成中会出现暂停的图标，再更新一下 UI。

// app/page.tsx
import { Pause, Send } from 'react-feather';

const Chat = () => {
  // 省略其它代码...
  const { stop, isLoading, messages } = useChat();

  const getBtnContent = () => {
    if (isLoading) {
      return (
        <>
          <Pause className="mr-2" size={16} />
          暂停生成
        </>
      );
    }
    return (
      <>
        <RotateCw className="mr-2" size={16} />
        重新生成
      </>
    );
  };

  return (
    // 省略其他代码...
    <div className="fixed w-full left-0 bottom-0 py-4 bg-gray-100 border-t border-t-gray-300">
      {messages.length > 0 ? (
        <button
          className="mb-2 mx-auto border border-gray-300 bg-gray-100 text-gray-600 p-2 px-8 rounded-md hover:bg-gray-200 transition-all flex items-center"
          onClick={isLoading ? stop : reload}
        >
          {getBtnContent()}
        </button>
      ) : null}
      <form
        onSubmit={handleSubmit}
        className="max-w-xl w-full mx-auto relative"
      >
        {/* 省略其他代码... */}
      </form>
    </div>
  );
};

来看看效果

有一点小缺陷：点击 stop 只是前端不再读取网络流，所以只是前端的渲染暂停，但是此时网络流还是没有断，更好的方式应该是发送消息给后端，后端主动中断流，然后再停止前端的渲染。

渲染 markdown

openai 是支持写 markdown 格式的，可以引入 markdown -> html 的包进行渲染。

需要安装 marked 和 @tailwindcss/typography。

• marked：将 markdown 解析成 html。
• @tailwindcss/typography：漂亮的文案排版。

pnpm add marked
pnpm add @tailwindcss/typography @types/marked -D

新建libs/marked.ts文件，用来写 markdown -> HTML 的方法

// libs/marked.ts
import { marked } from 'marked';

export const markdownToHTML = (markdown: string) => {
  if (!markdown || typeof markdown !== 'string') {
    return '';
  }

  return marked.parse(markdown);
};

@tailwindcss/typography的配置也比较简单，在tailwind.config.js的 plugin 配置一下。

// tailwind.config.js
/** @type {import('tailwindcss').Config} */
module.exports = {
  // 省略其他代码...
  plugins: [require('@tailwindcss/typography')],
};

在app/MessageCard.tsx文件中，给需要进行渲染 markdown html 的内容添加上类名：prose 。

// app/MessageCard.tsx
import { markdownToHTML } from '@/libs/marked';

const MessageCard = ({ message }: MessageCardProps) => {
  const content = markdownToHTML(message.content);

  return (
    <div className="flex items-start">
      <Avatar role={message.role} />
      <div
        className="pl-2 leading-6 prose transition-all max-w-xl" // 在这里!
        dangerouslySetInnerHTML={{ __html: content }}
      ></div>
    </div>
  );
};

来看看效果：

保持滚动条在底部

如果渲染的消息过长，或者消息过多时，得手动进行滚动，写一个 hook 要滚动条一直维持在底部。

新建 hook：hooks/use-scroll-bottom。

需要先安装 lodash.throttle

pnpm add lodash.throttle

// hooks/use-scroll-bottom.ts
import throttle from 'lodash.throttle';
import { RefObject, useEffect } from 'react';

interface UseScrollBottomOptions {
  scrollRef: RefObject<HTMLElement>;
}

const useScrollBottom = ({ scrollRef }: UseScrollBottomOptions) => {
  useEffect(() => {
    const scrollingElement = scrollRef.current;

    const callback: MutationCallback = function (mutationsList) {
      for (let mutation of mutationsList) {
        if (mutation.type === 'childList') {
          window.scrollTo(0, document.body.scrollHeight);
        }
      }
    };
    const throttleCallback = throttle(callback, 1000 / 16);

    const observer = new MutationObserver(throttleCallback);
    if (scrollingElement) {
      observer.observe(scrollingElement!, {
        subtree: true,
        childList: true,
      });
    }

    return () => {
      observer.disconnect();
    };
  }, []);
};

export default useScrollBottom;

在app/page.tsx中导入进行使用即可。

// app/page.tsx
const Chat = () => {
  const scrollRef = useRef<HTMLUListElement | null>(null);
  useScrollBottom({ scrollRef });

  return (
    <ul className="space-y-4" ref={scrollRef}>
      // 绑定ref
      {messages.map((message) => (
        <MessageCard key={message.id} message={message} />
      ))}
    </ul>
  );
};

后续

写到这里，前端乞丐版的 chatgpt 就写完啦，不知不觉这篇文章就写了这么长，看似简单的东西却有这么多东西可以写。

本来只想写流式渲染的，发现写着写着可以写成一个乞丐版的 chatgpt，然后就一路写写写写下去了。

回首上一次更新技术文章，已经不知道啥时候，最近一直在摆烂中（其实是偷懒，下班了不想动），直到端午节给自己找了点事情做才有的这篇文章，还是要继续加油不断地学习呀。

参考资料

1. Vercel AI 强烈推荐大家看看代码，写的真不错
2. Mdn TextEncoder
3. Mdn ReadableStream
4. Mdn TransformStream
5. openai-edge Github