HL🌱数字花园

从零开始：实现 ChatGpt 打字机聊天效果

2024年1月18日14分钟阅读

本文介绍使用 ChatGPT API 实现类似官网打字机回复效果的思路和方法，并附核心模块代码示例（vue3 + TS）

  • 调用流式接口、解析数据
  • 渲染 markdown
  • 代码块高亮
  • 打字机效果 + 移动光标
  • 其他 (一些思路）

准备工作

  • 注册 OpenAI 账号
  • 生成调用接口鉴权的 APIkey
  • 一个不被墙的网络环境

这一部分方法很多, 也存在时效性的问题, 大家可以自行查找合适的方法

接口调用

API

URL : api.openai.com/v1/chat/com…

请求方法: POST

接口官方文档： platform.openai.com/docs/api-re…

简单介绍一下我们用到的几个关键参数

参数说明值
Content-Type内容类型application/json
Authorization鉴权, 传获取的 KEYBearer 你的 KEY

Body

参数说明值
model使用的模型gpt-3.5-turbo
messages对话内容:, role 可传'user'和'system', 通过这个字段可以实现上下文对话，或者预设对话的前文[{'role': 'user', content: '问题内容'}]
strame是否流传输，如果是 false 将会一次性返回结果true

发送请求

async fetch(messages: GptMsgs) {
    return await fetch('https://api.openai.com/v1/chat/completions', {
      method: 'POST',
      body: JSON.stringify({
        model: 'gpt-3.5-turbo',
        messages,
        stream: true
      }),
      headers: {
        'Content-Type': 'application/json',
        Authorization: `Bearer ${this.key}`
      }
    })
  }

将 stream 设置为 true, 接口会将结果作为 EventStream 的一系列事件（event）返回

这里用的是浏览器原生支持的 FetchAPI

axios 在 config 中设置 responseType: ‘stream’也能实现同样的效果

解析流式数据

  1. 从返回的 body 中通过 getReader 获取 reader
if (!res.body) return
// 从response中获取reader
const reader = res.body.getReader()

  1. 通过 while 循环 + reader.read() 获取每次传输的数据

    通过控制台打印出来是这样的:

    const { done, value } = await reader.read()
console.log(value)

  2. 解析这些 Uint8Array 数据

    1. 用浏览器原生支持的 TextDecoder 将 buffer 解析成字符串，解析出來的格式如下图

    2. 使用正则表达式去匹配其中的 JSON 格式

    const parsePack = (str: string) => {
  // 定义正则表达式匹配模式
  const pattern = /data:\s*({.*?})\s*\n/g
  // 定义一个数组来存储所有匹 配到的 JSON 对象
  const result = []
  // 使用正则表达式匹配完整的 JSON 对象并解析它们
  let match
  while ((match = pattern.exec(str)) !== null) {
    const jsonStr = match[1]
    try {
      const json = JSON.parse(jsonStr)
      result.push(json)
    } catch (e) {
      console.log(e)
    }
  }
  // 输出所有解析出的 JSON 对象
  return result
}
    1. 从每个解析出来的 JSON 数据结构中取出我们要的内容
    // 解析出的结构
{
    "id": "chatcmpl-7KM1Fg6dyjS1u8ZynusrTT0N5y0Sj",
    "object": "chat.completion.chunk",
    "created": 1685085557,
    "model": "gpt-3.5-turbo-0301",
    "choices": [
        {
            "delta": {
                "content": "助"
            },
            "index": 0,
            "finish_reason": null
        }
    ]
}



    // 获取新增的字符
if (!json.choices || json.choices.length === 0) {
  return
}
const text = json.choices[0].delta.conten

在实际的封装中增加了一些钩子函数来拓展其他的功能，完整的代码如下

  • onStart: 调用函数后请求发出前
  • onCreated: 发出请求收到第一个回包后执行
  • onPatch: 有新的内容更新时执行
  • onDone: 传输结束时执行
export interface GptMsg {
  role: string
  content: string
}
export type GptMsgs = Array<GptMsg>
export class StreamGpt {
  onStart: (prompt: string) => void
  onCreated: () => void
  onDone: () => void
  onPatch: (text: string) => void
  constructor(private key: string, options: {
    onStart: (prompt: string) => void
    onCreated: () => void
    onDone: () => void
    onPatch: (text: string) => void
  }) {
    const { onStart, onCreated, onDone, onPatch } = options
    this.onStart = onStart
    this.onCreated = onCreated
    this.onPatch = onPatch
    this.onDone = onDone
  }
  async stream(prompt: string, history: GptMsgs = []) {
    let finish = false
    let count = 0
    // 触发onStart
    this.onStart(prompt)
    // 发起请求
    const res = await this.fetch([...history, { 'role': 'user', content: prompt }])
    if (!res.body) return
    // 从response中获取reader
    const reader = res.body.getReader()
    const decoder: TextDecoder = new TextDecoder()
    // 循环读取内容
    while (!finish) {
      const { done, value } = await reader.read()
      // console.log(value)
      if (done) {
        finish = true
        this.onDone()
        break
      }
      count++
      const jsonArray = parsePack(decoder.decode(value))
      if (count === 1) {
        this.onCreated()
      }
      jsonArray.forEach((json: any) => {
        if (!json.choices || json.choices.length === 0) {
          return
        }
        const text = json.choices[0].delta.content
        this.onPatch(text)
      })
    }
  }
  async fetch(messages: GptMsgs) {
    return await fetch('https://api.openai.com/v1/chat/completions', {
      method: 'POST',
      body: JSON.stringify({
        model: 'gpt-3.5-turbo',
        messages,
        stream: true
      }),
      headers: {
        'Content-Type': 'application/json',
        Authorization: `Bearer ${this.key}`
      }
    })
  }
}

打字机效果

打字机队列

现在我们有了一个不断补充内容的字符串, 但在传输过程中每次停顿后会跳出一串内容然后又停顿一会, 阅读体验有些不流畅, 就像玩游戏时帧数低卡顿的感觉, 我们用一个队列让它逐字地展示出来, 并且根据传输速度控制输出的速度

  • 队列提供入队和消费功能
  • 定时器在时间间隔中输出字符
  • 消费的间隔采用动态计算的方式，尽可能保证队列中的剩余字符能在两秒钟内消费完
  • 结束时将剩余的字符一次性消费
// 打字机队列
export class Typewriter {
  private queue: string[] = []
  private consuming = false
  private timmer: any
  constructor(private onConsume: (str: string) => void) {
  }
  // 输出速度动态控制
  dynamicSpeed() {
    const speed = 2000 / this.queue.length
    if (speed > 200) {
      return 200
    } else {
      return speed
    }
  }
  // 添加字符串到队列
  add(str: string) {
    if (!str) return
    this.queue.push(...str.split(''))
  }
  // 消费
  consume() {
    if (this.queue.length > 0) {
      const str = this.queue.shift()
      str && this.onConsume(str)
    }
  }
  // 消费下一个
  next() {
    this.consume()
    // 根据队列中字符的数量来设置消耗每一帧的速度，用定时器消耗
    this.timmer = setTimeout(() => {
      this.consume()
      if (this.consuming) {
        this.next()
      }
    }, this.dynamicSpeed())
  }
  // 开始消费队列
  start() {
    this.consuming = true
    this.next()
  }
  // 结束消费队列
  done() {
    this.consuming = false
    clearTimeout(this.timmer)
    // 把queue中剩下的字符一次性消费
    this.onConsume(this.queue.join(''))
    this.queue = []
  }
}

在 VUE3 中使用

封装 hook

  • 定义了 streamingText 来存储传输中的字符串
  • streaming 标识当前是否在传输中
  • msgList 保存历史的对话记录
  • history 参数控制是否在每次请求时带上历史记录
  • 注意: api 的接口 token 的上限为 4096，实际的话要判断是否超出长度而抛弃一些早前的记录
// 整合封装 vue3 composition api
import { ref } from 'vue'
import { StreamGpt, Typewriter, GptMsgs } from '../scripts'
export const useGpt = (key: string, history: boolean = false) => {
  const streamingText = ref('')
  const streaming = ref(false)
  const msgList = ref<GptMsgs>([])
  const typewriter = new Typewriter((str: string) => {
    streamingText.value += str || ''
    console.log('str', str)
  })
  const gpt = new StreamGpt(key, {
    onStart: (prompt: string) => {
      streaming.value = true
      msgList.value.push({
        role: 'user',
        content: prompt
      })
    },
    onPatch: (text: string) => {
      console.log('onPatch', text)
      typewriter.add(text)
    },
    onCreated: () => {
      typewriter.start()
    },
    onDone: () => {
      typewriter.done()
      streaming.value = false
      msgList.value.push({
        role: 'system',
        content: streamingText.value
      })
      streamingText.value = ''
    }
  })
  // 如果是history模式，则在strame时将msgList传入
  const stream = (prompt: string) => {
    gpt.stream(prompt, history ? msgList.value : undefined)
  }
  return {
    streamingText,
    streaming,
    msgList,
    stream
  }
}

使用

const { msgList, streaming, streamingText, stream } = useGpt('你的key', true)
// 发送内容
const handleSubmit = (content: string) => {
  if (content === '') return
  stream(content)
}

渲染 Markdown

现在我们有了解析好的内容, 可以看到是 Markdown 的格式，我们要将它转换成 HTML，试了几个库后选择了 Markdown-it

  • 安装 markdown-it
npm install markdown-it
  • 引入并初始化
import MarkdownIt from 'markdown-it'
const md: MarkdownIt = MarkdownIt()
  • 解析 md 内容
let html = md.render(props.content)

代码块高亮

如果要让 markdown 中的代码呈现出代码块的样式并且对特定语言的语法进行高亮，我们要借助 hightlight.js 这个库

  • 安装 highlight.js
npm install highlight.js
  • 引入并初始化
import hljs from 'highlight.js'
import 'highlight.js/styles/atom-one-dark-reasonable.css'

从 node_modules 中引入相应的样式文件, 在对应的文件夹下可以找到更多代码的样式主题，路径引入即可

  • 在 markdown-it 提供的钩子函数中将代码部分用 hightlight.js 改写

    const md: MarkdownIt = MarkdownIt({
  highlight: function (str: string, lang: string) {
    if (lang && hljs.getLanguage(lang)) {
      try {
        return `<div><div><span>${lang}</span></div><div><code>${hljs.highlight(str, { language: lang, ignoreIllegals: true }).value
          }</code></div></div>`
      } catch (__) {
        console.log(__, 'error')
      }
    }
    return `<div><div><span>${lang}</span></div><div><code>${md.utils.escapeHtml(
      str
    )}</code></div></div>`
  }
})

同时在钩子函数给代码块插入一个 class 为 hl-code-header 的头部，来实现代码语言的显示

光标闪烁效果

在打字机输出文字时，给尾部加一个闪烁的光标

  1. 在解析出的 HTML 结构中，找到最后一个元素并在这个元素的尾部插入一个光标元素
  2. 给它加上闪烁动画
  3. 对一些特殊的标签特殊处理，例如代码块的 PRE 标签
  4. 对于深层嵌套的元素递归查找到最后一个元素
      // 获取展示内容的容器
      const parent = popRef.value
      if (!parent) return
      // 获取最后一个子元素节点
      let lastChild = parent.lastElementChild || parent
      // 如果是pre标签，就在pre标签中找到class为hljs的元素
      if (lastChild.tagName === 'PRE') {
        lastChild = lastChild.getElementsByClassName('hljs')[0] || lastChild
      }
      // 兼容是ul标签的情况，找到OL标签内部的最后一个元素
      if (lastChild.tagName === 'OL') {
        lastChild = findLastElement(lastChild as HTMLElement)
      }
      // 向最后一个子元素中插入span标签实现光标
      lastChild?.insertAdjacentHTML('beforeend', '<span class="input-cursor"></span>') 
 
 
// 递归找到DOM下最后一个元素节点
function findLastElement(element: HTMLElement): HTMLElement {
  // 如果该DOM没有子元素，则返回自身
  if (!element.children.length) {
    return element
  }
  const lastChild = element.children[element.children.length - 1]
  // 如果最后一个子元素是元素节点，则递归查找
  if (lastChild.nodeType === Node.ELEMENT_NODE) {
    return findLastElement(lastChild as HTMLElement)
  }
  return element
}

实时滚到到底部

通过 vue 的 watch 监听 streamingText 来判断是否新增了内容，让页面随着内容的输出滚动到底部，让阅读的体验更上一层楼

给大家推荐一个 VUE3 的 hooks 库 vueuse，功能十分齐全，这里滚动的逻辑就用里面的 scroll 模块简单快速地实现

import { useScroll } from '@vueuse/core'
// ...
// 滚动的元素
const listEl = ref()
const { y } = useScroll(listEl)
const scrollToBottom = () => {
  nextTick(() => {
    y.value = listEl.value?.scrollHeight || 0
  })
}
// 监听streamingText变化，滚动到底部
watch(streamingText, (val) => {
  if (val) {
    scrollToBottom()
  }
})

其他

角色扮演

可以在 message 中加入一些预设对话内容，让 GPT 进行角色扮演, 例如：

const preset = [{ 'role': 'user', content: '请扮演一个不会说中文的人和我对话' }, { 'role': 'system', content: 'sorry, I do not speak Chinese' }]
// 发起请求时在messages中加入加入预设的内容
const res = await this.fetch([...preset, ..._history, { 'role': 'user', content: prompt }])

我们让 GPT 扮演一个不会中文的人，结果如下:

(好家伙，不会说中文但是能看懂中文是吧 -.-!)

超长对话

由于 API 的接口有请求的字数上限 (4096 个 Token)

想要实现超长上下文记录的对话可以用到一个技巧, 当对话记录达到一定量时让 GPT 对前面的对话记录生成总结摘要

在后续对话的上文中只提供总结摘要，那么 GPT 就会大致清楚本次对话的主题

等再达到更大一个量级时, 再对摘要生成摘要

虽然越往后会有一部分 “记忆缺失”，但对于对话的大致方向还是可以记录的

关系图谱

反向链接

  • 无法找到反向链接