From d18a0a103983d8276cbc1f00468c9285035a7e1b Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Mon, 14 Oct 2024 22:10:56 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=B9=B6=E5=8F=91=E9=97=AE?= =?UTF-8?q?=E9=A2=98=EF=BC=8C=E4=BF=AE=E5=A4=8Dconfig=E4=B8=8D=E6=9B=B4?= =?UTF-8?q?=E6=96=B0=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/main.py | 2 +- src/routers/base_router.py | 4 +- src/routers/chat_router.py | 55 +++++++++----- src/routers/data_router.py | 1 - web/src/components/ChatComponent.vue | 101 ++++++++++++++++++-------- web/src/components/DebugComponent.vue | 9 +-- web/src/layouts/AppLayout.vue | 1 + 7 files changed, 112 insertions(+), 61 deletions(-) diff --git a/src/main.py b/src/main.py index 18a97011..d831e3d0 100644 --- a/src/main.py +++ b/src/main.py @@ -24,5 +24,5 @@ logger = setup_logger("server:main") if __name__ == "__main__": - uvicorn.run(app, host="0.0.0.0", port=5000) + uvicorn.run(app, host="0.0.0.0", port=5000, threads=10, workers=10) diff --git a/src/routers/base_router.py b/src/routers/base_router.py index dda125a5..31ac506c 100644 --- a/src/routers/base_router.py +++ b/src/routers/base_router.py @@ -16,7 +16,7 @@ async def route_index(): return {"message": "You Got It!"} @base.get("/config") -async def get_config(): +def get_config(): return startup.config @base.post("/config") @@ -32,7 +32,7 @@ async def restart(): return {"message": "Restarted!"} @base.get("/log") -async def get_log(): +def get_log(): from src.utils.logging_config import LOG_FILE from collections import deque diff --git a/src/routers/chat_router.py b/src/routers/chat_router.py index d6dcc0d5..f7a9271c 100644 --- a/src/routers/chat_router.py +++ b/src/routers/chat_router.py @@ -1,31 +1,41 @@ -from fastapi import APIRouter, HTTPException, Request -from fastapi.responses import StreamingResponse +import time +import uuid +from fastapi import APIRouter, HTTPException, Request, Body +from fastapi.responses import StreamingResponse, Response +from concurrent.futures import ThreadPoolExecutor import json +import asyncio from src.core import HistoryManager from src.core.startup import startup from src.utils.logging_config import setup_logger chat = APIRouter(prefix="/chat") logger = setup_logger("server-chat") +# 创建线程池 +executor = ThreadPoolExecutor() + +refs_pool = {} @chat.get("/") async def chat_get(): return "Chat Get!" @chat.post("/") -async def chat_post(request: Request): - request_data = await request.json() - query = request_data['query'] - meta = request_data.get('meta') - history_manager = HistoryManager(request_data['history']) +def chat_post( + query: str = Body(...), + meta: dict = Body(None), + history: list = Body(...), + cur_res_id: str = Body(...)): + history_manager = HistoryManager(history) new_query, refs = startup.retriever(query, history_manager.messages, meta) + refs_pool[cur_res_id] = refs messages = history_manager.get_history_with_msg(new_query, max_rounds=meta.get('history_round')) history_manager.add_user(query) logger.debug(f"Web history: {history_manager.messages}") - async def generate_response(): + def generate_response(): content = "" for delta in startup.model.predict(messages, stream=True): if not delta.content: @@ -36,20 +46,29 @@ async def chat_post(request: Request): else: content += delta.content - response_chunk = json.dumps({ - "history": history_manager.update_ai(content), + logger.debug(f"Response: {content}") + + _chunk = json.dumps({ "response": content, - "refs": refs # TODO: 优化 refs，不需要每次都返回 - }, ensure_ascii=False).encode('utf8') + b'\n' - yield response_chunk + "history": history_manager.update_ai(content), + }, ensure_ascii=False).encode('utf-8') + b"\n" + yield _chunk return StreamingResponse(generate_response(), media_type='application/json') @chat.post("/call") -async def call(request: Request): - request_data = await request.json() - query = request_data['query'] - response = startup.model.predict(query) +async def call(query: str = Body(...), meta: dict = Body(None)): + async def predict_async(query): + loop = asyncio.get_event_loop() + return await loop.run_in_executor(executor, startup.model.predict, query) + + response = await predict_async(query) logger.debug({"query": query, "response": response.content}) - return {"response": response.content} \ No newline at end of file + return {"response": response.content} + +@chat.get("/refs") +def get_refs(cur_res_id: str): + global refs_pool + refs = refs_pool.pop(cur_res_id, None) + return {"refs": refs} \ No newline at end of file diff --git a/src/routers/data_router.py b/src/routers/data_router.py index f5073f98..2a19805b 100644 --- a/src/routers/data_router.py +++ b/src/routers/data_router.py @@ -1,7 +1,6 @@ import os from typing import List, Optional from fastapi import APIRouter, File, UploadFile, HTTPException, Depends, Body -from pydantic import BaseModel from src.utils import setup_logger, hashstr from src.core.startup import startup diff --git a/web/src/components/ChatComponent.vue b/web/src/components/ChatComponent.vue index 504834ea..40d84b8c 100644 --- a/web/src/components/ChatComponent.vue +++ b/web/src/components/ChatComponent.vue @@ -16,10 +16,9 @@
- +
+ 图数据库 +
@@ -147,6 +146,7 @@ import { BookFilled, CompassFilled, GoldenFilled, + GoldOutlined, SettingOutlined, SettingFilled, PlusCircleOutlined, @@ -282,7 +282,6 @@ const scrollToBottom = () => { }, 10) } - const generateRandomHash = (length) => { let chars = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'; let hash = ''; @@ -314,14 +313,31 @@ const appendAiMessage = (message, refs=null) => { const updateMessage = (text, id, refs, status) => { const message = conv.value.messages.find((message) => message.id === id); + if (message) { - message.refs = refs; - message.status = status; - message.text = text; - message.model_name = refs.model_name + // 只有在 text 不为空时更新 + if (text !== null && text !== undefined && text !== '') { + message.text = text; + } + + // 只有在 refs 不为空时更新 + if (refs !== null && refs !== undefined) { + message.refs = refs; + + // 如果 refs 里面的 model_name 不为空时更新 + if (refs.model_name !== null && refs.model_name !== undefined && refs.model_name !== '') { + message.model_name = refs.model_name; + } + } + + // 只有在 status 不为空时更新 + if (status !== null && status !== undefined && status !== '') { + message.status = status; + } } else { console.error('Message not found'); } + scrollToBottom(); }; @@ -380,51 +396,74 @@ const fetchChatResponse = (user_input, cur_res_id) => { body: JSON.stringify({ query: user_input, history: conv.value.history, - meta: meta + meta: meta, + cur_res_id: cur_res_id, }), headers: { 'Content-Type': 'application/json' } - }).then((response) => { + }) + .then((response) => { + if (!response.body) throw new Error("ReadableStream not supported."); const reader = response.body.getReader(); - const decoder = new TextDecoder(); + const decoder = new TextDecoder("utf-8"); let buffer = ''; - // 逐步读取响应文本 const readChunk = () => { return reader.read().then(({ done, value }) => { if (done) { - // 处理完成 - updateStatus(cur_res_id, "finished"); + fetchRefs(cur_res_id).then((data) => { + console.log(data) + updateMessage(null, cur_res_id, data, "finished"); + updateStatus(cur_res_id, "finished"); + }) isStreaming.value = false; if (conv.value.messages.length === 2) { renameTitle(); } - return; // 结束读取 + return; } - buffer += decoder.decode(value, { stream: true }); - const messages = buffer.trim().split('\n'); - - messages.forEach((message) => { - try { - const data = JSON.parse(message); - updateMessage(data.response, cur_res_id, data.refs, "loading"); - conv.value.history = data.history; - } catch (e) { - // console.error('JSON 解析错误:', e); - } - }); - buffer = ''; // 清空缓冲区 + const chunk = decoder.decode(value, { stream: true }); + buffer += chunk; + try { + const data = JSON.parse(chunk); + updateMessage(data.response, cur_res_id, data.refs, "loading"); + console.debug(data.response) + conv.value.history = data.history; + } catch (e) { + // console.debug('JSON 解析错误:', e, chunk); + } return readChunk(); // 继续读取 }); }; - return readChunk(); - isStreaming.value = false; + readChunk(); + + }) .catch((error) => { console.error(error); updateStatus(cur_res_id, "error"); isStreaming.value = false; }) + .finally(() => { + isStreaming.value = false; + }); +} + +const fetchRefs = (cur_res_id) => { + return new Promise((resolve, reject) => { + fetch(`/api/chat/refs?cur_res_id=${cur_res_id}`, { + method: "GET", + headers: { + "Content-Type": "application/json" + }, + }).then(response => response.json()) + .then(data => { + resolve(data.refs) + }) + .catch((error) => { + reject(error) + }) + }) } // 更新后的 sendMessage 函数 diff --git a/web/src/components/DebugComponent.vue b/web/src/components/DebugComponent.vue index 2bd5ff73..8adc586f 100644 --- a/web/src/components/DebugComponent.vue +++ b/web/src/components/DebugComponent.vue @@ -25,20 +25,13 @@ const fetchLogs = async () => { state.fetching = true; try { - // 清空之前的错误信息 error.value = ''; - - // 发送请求获取日志数据 - const response = await fetch('/api/log'); // 替换为你的 API 路径 - + const response = await fetch('/api/log'); if (!response.ok) { throw new Error('Failed to fetch logs'); } - // 解析 JSON 数据 const data = await response.json(); - - // 将日志信息赋值给 logs 变量 logs.value = data.log; // 等待 DOM 更新完成后自动滚动到最底部 diff --git a/web/src/layouts/AppLayout.vue b/web/src/layouts/AppLayout.vue index 8e4a53f6..09902290 100644 --- a/web/src/layouts/AppLayout.vue +++ b/web/src/layouts/AppLayout.vue @@ -50,6 +50,7 @@ const getRemoteDatabase = () => { } onMounted(() => { + getRemoteConfig() getRemoteDatabase() configStore.refreshConfig() })