我們是袋鼠云數(shù)棧 UED 團隊，致力于打造優(yōu)秀的一站式數(shù)據(jù)中臺產品。我們始終保持工匠精神，探索前端道路，為社區(qū)積累并傳播經驗價值。

本文作者：修能

生產力工具 + AI 是不可逆轉的趨勢，慢慢的大模型能力通過 AI Agent 落地的工程化能力也開始趨于成熟。作為大數(shù)據(jù)產品的數(shù)棧也必然是需要借助 AI 能力提升產品競爭力。
去年 12 月，我們在產品中上線了 AI+ 的功能，借助已經開源的大模型的能力，幫助我們探索和落地更多地應用場景。在初版 AI+ 的功能中，我們實現(xiàn)了基礎功能的通話。

SSE

在 ChatGPT 中，我們在等待大模型生成回答的時間通常不需要很久。這是因為 ChatGPT 通過 server-sent events（SSE）來實現(xiàn)將生成的部分回答通過事件流傳遞到前端。而這就讓前端不必等回答全部生成后再獲取，也就使得不需要請求等待很久。

SSE 是一種基于 HTTP 協(xié)議的單向通信機制，用于服務端向客戶端推送數(shù)據(jù)。

入門使用

// 創(chuàng)建 SSE 的實例
const evtSource = new EventSource("http://api.example.com/ssedemo.php", {
  withCredentials: true,
});

// 添加監(jiān)聽事件
evtSource.onmessage = (event) => {
  const newElement = document.createElement("li");
  const eventList = document.getElementById("list");

  newElement.textContent = `message: ${event.data}`;
  eventList.appendChild(newElement);
};

// 錯誤處理
evtSource.onerror = (err) => {
  console.error("EventSource failed:", err);
};

// 關閉事件流
evtSource.close();

需要注意的是，SSE 請求的服務端響應信息頭的 MIME 類型必須是text/event-stream，否則會無法監(jiān)聽到事件。
另外，由于是基于 HTTP 協(xié)議的，所以在 HTTP/1.1 或更低的時候，會受瀏覽器最大連接數(shù)的限制。

Fields

收到的消息格式一定是具有以下字段的某種組合，其他字段名都將忽略，每行一個：

• event
• data
• id
• retry

: this is a test stream // 第一條消息，這會被解析會注釋

data: some text // 第二條消息

data: another message // 第三條消息
data: with two lines

event: userconnect // 第四條消息
data: {"username": "bobby", "time": "02:33:48"}

如上所示，默認瀏覽器的 EventSource API 雖然可用，但是限制比較多。

1. 只支持 url 和 withCredentials 參數(shù)。不支持往 body 里傳參數(shù)。而通常來說 URL 是有最大長度限制的。
2. 無法自定義請求頭。
3. 只能發(fā)起 GET 請求。

其實，我們也可以通過 Fetch 來實現(xiàn) SSE 的通信，只不過需要額外自行處理數(shù)據(jù)流的傳遞。

實現(xiàn)

首先，我們借助 Fetch 的能力來實現(xiàn)請求。

const response = await fetch(url, options);

通過接受用戶提供的 url 和 options 發(fā)起一個 fetch 的請求。
然后，我們需要排除掉非 SSE 的請求類型，我們可以直接拿響應的 header 中拿 content-type進行判斷。

const contentType = response.headers.get('content-type');
if (!contentType?.startsWith('text/event-stream')) {
    throw new Error('SSE 請求必須設置 content-type 為 text/event-stream');
}

接著，我們業(yè)務場景中通常直接通過 response.json()獲取 JSON 格式的數(shù)據(jù)了，但這里我們由于是事件流，所以我們通過 response.body 拿到的是一個 ReadableStream。我們需要借助相關的 API 進行流的讀取。

const reader = response.body.getReader();
let result: ReadableStreamDefaultReadResult<Uint8Array>;
while (!(result = await reader.read()).done) {
  	// 假定每一次 read 的 value 都是完整的消息
    onmessage(onChunk(result.value));
}

其中 onChunk 函數(shù)就是處理事件流中的每一份數(shù)據(jù)的。

// 偽代碼
function onChunk(arr: Uint8Array){
  const links = seekLinks();
  // 待完善
}

在實現(xiàn) seekLinks 方法之前，我們需要先知道到什么時候算每一行的結束。

從 Fields 可以知道，每一行是以\n作為區(qū)分的。

function seekLinks(arr: Uint8Array){
  const lines = [];
  const buffer = arr;
  const bufLength = buffer.length;
  let position = 0;
  let lineStart = 0;
  while(position < bufLength){
    // '\n'.charCodeAt() === 10;
    if(buffer[position] === 10){
      lines.push(buffer.slice(lineStart, position));
      lineStart = position;
    };
    position += 1;
  }
  return lines;
}

在獲取到所有行后，針對每一行做處理。

// 偽代碼
function onChunk(arr: Uint8Array){
  const links = seekLinks();
  const decoder = new TextDecoder();
  let message = {
    data: '',
    event: '',
    id: '',
    retry: undefined,
  }:
  links.forEach((line) => {
    // ':'.charCodeAt() === 58;
    const colon = line.findIndex(l => l === 58);
    const fieldArr = line.slice(0, colon);
    const valueArr = line.slice(colon);
    if(colon === -1){
      // 當冒號作為開頭的時候，解析成注釋
      return;
    }
    const field = decoder.decode(fieldArr);
    const value = decoder.decode(valueArr);
    switch (field) {
      case 'data':
          message.data = message.data
              ? message.data + '\n' + value
              : value;
          break;
      case 'event':
          message.event = value;
          break;
      case 'id':
          message.id = value;
          break;
      case 'retry':
          const retry = parseInt(value, 10);
          message.retry = retry
          break;
  	}
  });
  return message;
}

大致完成了最簡單的基礎功能的解析，而以上偽代碼參考 fetch-event-source 的源碼。

借助 fetch-event-source 的能力，在數(shù)棧產品中調用的方式和 HTTP 請求基本保持一致。

function sse(url: string, params: any, options: FetchEventSourceInit) {
  const headers = {
    'Content-Type': 'application/json',
    accept: 'text/event-stream',
  };
  fetchEventSource(url, {
    method: 'POST',
    body: JSON.stringify(params),
    headers,
    ...options,
  });
}

打字機效果

接著，我們實現(xiàn)具備科技感的打字機效果：

輸出

這里我們不能直接將響應的消息直接打印到屏幕上，因為響應的消息通常是好多字，這樣子會導致打字機效果顯得非?？D，用戶體驗不佳。
在數(shù)棧產品中，我們通過將響應的消息收集到暫存區(qū)中，然后通過每秒從暫存區(qū)中取出若干個字符打印到屏幕上，優(yōu)化打字機卡頓的效果。

function AIGC(){
   const typing = useTyping({
      // 暫存區(qū)啟動后，每個 delay 的時間都會執(zhí)行該方法將消息打印到屏幕上
      onTyping(val) {
        // ...
      },
  });
	const handleChat = (message: string) => {
      // 標志暫存區(qū)需要開始存響應的消息了
      typing.start();
      requestChat(params, {
        onmessage(event: { data: string }) {
           	const { data } = event;
            // 把響應的消息存入暫存區(qū)中
            typing.push(data);
        },
        onclose() {
            // 關閉或失敗的話，釋放暫存區(qū)的數(shù)據(jù)
            typing.close();
        },
        onerror() {
            typing.close();
        },
    });
  };
}

其中，相關暫存區(qū)的代碼整理成 useTyping 實現(xiàn)。

export default function useTyping({
    onTyping,
    onEnd,
}: {
    onTyping: (val: string) => void;
    onEnd: () => void;
}) {
    const interval = useRef<number>();
    const queue = useRef<string>('');
    const isStart = useRef<boolean>(false);

    function startTyping() {
        if (interval.current) return;
        let index = 0;
        interval.current = window.setInterval(() => {
            if (index < queue.current.length) {
                const str = queue.current;
                onTyping(str.slice(0, index + 1));
                index++;
            } else if (!isStart.current) {
                // 如果發(fā)送了全部的消息且信號關閉，則清空隊列
                window.clearInterval(interval.current);
                interval.current = 0;
                onEnd();
            }
            // 如果發(fā)送了全部的消息，但是信號沒有關閉，則什么都不做繼續(xù)輪訓等待新的消息
        }, 50);
    }

    useEffect(() => {
        return () => {
            window.clearInterval(interval.current);
            interval.current = 0;
        };
    }, []);

    function start() {
        isStart.current = true;
        window.clearInterval(interval.current);
        interval.current = 0;
        queue.current = '';
    }

    function push(str: string) {
        if (!isStart.current) return;
        queue.current += str.replace(/\\n/g, '\n');
        startTyping();
    }

    // 關閉的時候不需要清空隊列，因為可能還有一些消息沒有發(fā)送完畢，統(tǒng)一等消息發(fā)送完畢后關閉
    function close() {
        isStart.current = false;
    }

    return { start, push, close };
}

光標

在實現(xiàn)了打字機效果后，我們還需要添加一個閃爍的光標。
原理比較簡單，就是在消息區(qū)域的最后一個元素的末尾添加元素即可。

.markdown {
  >*:last-child::after {
    content: " ";
    width: 2px;
    height: 13px;
    transform: translate(1px, 2px);
    font-family: Menlo, Monaco, "Courier New", monospace;
    font-weight: normal;
    font-size: 0;
    font-feature-settings: "liga" 0, "calt" 0;
    line-height: 13px;
    letter-spacing: 0;
    display: inline-block;
    visibility: hidden;
    animation: blinker 1s step-end infinite;
    background: #000;
  }

  @keyframes blinker {
    0% {
      visibility: inherit;
    }
    50% {
      visibility: hidden;
    }
    100% {
      visibility: inherit;
    }
  }
}

當然，這里有一些問題，在 markdown 解析出 Code Block 的時候會導致光標錯位，這個問題 ChatGPT 同樣也有。

那么到這里，我們就實現(xiàn)了一個具備基礎功能的 AI+ 的需求。

最后

歡迎關注【袋鼠云數(shù)棧UED團隊】~
袋鼠云數(shù)棧 UED 團隊持續(xù)為廣大開發(fā)者分享技術成果，相繼參與開源了歡迎 star文章來源地址http://www.zghlxwxcb.cn/news/detail-824272.html

• 大數(shù)據(jù)分布式任務調度系統(tǒng)——Taier
• 輕量級的 Web IDE UI 框架——Molecule
• 針對大數(shù)據(jù)領域的 SQL Parser 項目——dt-sql-parser
• 袋鼠云數(shù)棧前端團隊代碼評審工程實踐文檔——code-review-practices
• 一個速度更快、配置更靈活、使用更簡單的模塊打包器——ko
• 一個針對 antd 的組件測試工具庫——ant-design-testing

到了這里，關于袋鼠云數(shù)棧產品中 AI+ 實現(xiàn)原理剖析的文章就介紹完了。如果您還想了解更多內容，請在右上角搜索TOY模板網以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網！