基于GPT搭建私有知识库聊天机器人（六）仿chatGPT打字机效果

文章链接：

基于GPT搭建私有知识库聊天机器人（一）实现原理

基于GPT搭建私有知识库聊天机器人（二）环境安装

基于GPT搭建私有知识库聊天机器人（三）向量数据训练

基于GPT搭建私有知识库聊天机器人（四）问答实现

基于GPT搭建私有知识库聊天机器人（五）函数调用

在前几篇文章中，我们已经了解了如何使用 GPT 模型来搭建一个简单的聊天机器人，并在后端使用私有知识库来提供答案。

现在，我们将继续改进我们的聊天界面，实现类似chatGPT打字机的效果聊天，避免长时间等待接口数据返回，以提升用户体验。

1、效果展示

PS：一本正经的胡说八道

2、Server-Sent Events (SSE) 技术简介

在本篇文章中，我们将使用 SSE 技术来实现打字机效果输出。SSE 是一种 HTML5 技术，允许服务器向客户端推送数据，而不需要客户端主动请求。通过 SSE，我们可以在服务器端有新消息时，实时将消息推送到前端，从而实现动态的聊天效果。

3、前端代码

首先，我们需要编写前端的JavaScript 代码，以便使用 SSE 技术与服务器进行实时通信。

<!DOCTYPE html>
<html>
<head><title>ChatGPT-like Interface</title><link rel="stylesheet" href="static/styles.css">
</head>
<body><div class="chat-container"><div class="chat-history" id="chatHistory"><!-- Chat messages will be dynamically added here --></div><div class="user-input"><input type="text" id="userInput" placeholder="请输入您的问题..."><button id="sendButton">发送</button></div></div><script>// Your existing chat interface code here...// Server communication codevar eventSource; // Declare the eventSource variable outside the click handlerdocument.getElementById("sendButton").addEventListener("click", function () {var userMessage = document.getElementById("userInput").value.trim();if (userMessage === '') {alert('Please enter a message!');return;}appendMessage('user', userMessage); // Add the user's message to the chat history// Close the previous SSE connection (if exists)if (eventSource) {eventSource.close();}// Establish SSE connection with the user's message as a parametereventSource = new EventSource(`/print_stream?question=${encodeURIComponent(userMessage)}`);eventSource.onmessage = function (event) {var botMessage = event.data;appendMessage('bot', botMessage);};eventSource.onerror = function (error) {console.error("Error occurred with SSE connection:", error);// Handle the error if necessaryisFirstToken = true;eventSource.close();};document.getElementById("userInput").value = '';});var chatHistoryDiv = document.getElementById("chatHistory"); // 获取 chatHistory 的元素var isFirstToken = true; // 用于跟踪是否是第一次返回 tokenfunction appendMessage(sender, message) {if (isFirstToken) {// 如果是第一次返回 token，创建新的 <div> 元素，并将 isFirstToken 设置为 falsevar messageDiv = document.createElement('div');messageDiv.className = `chat-message ${sender === 'user' ? 'user-message' : 'bot-message'}`;chatHistoryDiv.appendChild(messageDiv);if(sender === 'bot') {isFirstToken = false;}} else {// 如果不是第一次返回 token，直接获取最后一个 <div> 元素，将新的消息内容追加到现有的元素中var messageDiv = chatHistoryDiv.lastElementChild;}messageDiv.innerText += message; // 将新的消息内容追加到 <div> 中chatHistoryDiv.scrollTop = chatHistoryDiv.scrollHeight; // 将滚动条滚动到最底部}</script>
</body>
</html>

为了实现对话效果，我们需要调整 CSS 样式表中的部分样式。以下是 CSS 样式表：

body {font-family: Arial, sans-serif;background-color: #f0f0f0;margin: 0;padding: 0;display: flex;justify-content: center;align-items: center;min-height: 100vh;
}.chat-container {width: 800px;border: 1px solid #ccc;border-radius: 5px;box-shadow: 0 2px 6px rgba(0, 0, 0, 0.1);background-color: #fff;overflow: hidden;
}.chat-history {max-height: 800px;overflow-y: auto;padding: 10px;
}.chat-message {margin-bottom: 10px;padding: 8px 12px;border-radius: 20px;max-width: 70%; /* 设置最大宽度，使得消息在一行中不会过长 */align-self: flex-end; /* 靠右显示 */word-wrap: break-word; /* 处理长文本的自动换行 */overflow-wrap: break-word; /* 处理长文本的自动换行 */
}.user-message {color: #007bff;background-color: #e6e6e6; /* 用户消息气泡背景色 */text-align: right; /* 靠右显示文本内容 */align-self: flex-end; /* 靠右显示气泡 */margin-left: auto; /* 添加额外的间距，让气泡靠右 */
}.bot-message {color: #555;background-color: #d9edf7; /* 机器人消息气泡背景色 */text-align: left; /* 靠左显示文本内容 */align-self: flex-start; /* 靠左显示气泡 */margin-right: auto; /* 添加额外的间距，让气泡靠左 */
}.user-input {display: flex;align-items: center;padding: 10px;
}#userInput {flex-grow: 1;padding: 8px;border: 1px solid #ccc;border-radius: 5px;margin-right: 10px;
}#sendButton {padding: 8px 15px;border: none;border-radius: 5px;background-color: #007bff;color: #fff;cursor: pointer;
}#sendButton:hover {background-color: #0056b3;
}

4、后端代码

本文依旧使用的langchain框架实现访问openAI，以及利用回调函数接收token数据。

首先，是API入口：

from flask import Flask, request, Response, stream_with_context@app.route("/print_stream")
def print_stream():question = request.args.get('question')ans = search_schedule(question)return Response(stream_with_context(ans), content_type='text/event-stream')

其次，是访问openAI代码（不太了解的可以看下前几篇文章）：

def search_schedule(query: str) -> str:stream_to_web = StreamToWeb()llm = ChatOpenAI(temperature=0,model="gpt-3.5-turbo-0613",callback_manager=CallbackManager([stream_to_web]),streaming=True)bus_tools = [BusTool()]open_ai_agent = initialize_agent(bus_tools,llm,agent=AgentType.OPENAI_FUNCTIONS,verbose=True)chain_thread = threading.Thread(target=process_query,kwargs={"question": query,"open_ai_agent": open_ai_agent})chain_thread.start()resp = stream_to_web.generate_tokens()return resp

注意：上面调用openai部分代码必须使用异步执行，才能做到一边接收返回token，一边返回前端，否则无法实现打字机效果。

最后，打字机效果核心代码：

class StreamToWeb(StreamingStdOutCallbackHandler):def __init__(self):self.tokens = []# 记得结束后这里置trueself.finish = Falsedef on_llm_new_token(self, token: str, **kwargs):self.tokens.append(token)def on_llm_end(self, response: any, **kwargs: any) -> None:self.finish = 1def on_llm_error(self, error: Exception, **kwargs: any) -> None:print(str(error))self.tokens.append(str(error))def generate_tokens(self):while not self.finish or self.tokens:if self.tokens:data = self.tokens.pop(0)yield f"data: {data}\n\n"else:pass