tian/kangda-robot-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改异步保存机制

Browse Source
This commit is contained in:
Tian jianyong 2025-12-24 20:23:45 +08:00
parent bfe2492c4a
commit ac756fb4ee
1 changed files with 23 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py
View File

@ -8,7 +8,7 @@ import json
import time
from typing import Any, Dict, Generator, Optional, Tuple
from fastapi import APIRouter, Depends, Request
from fastapi import APIRouter, BackgroundTasks, Depends, Request
from fastapi.responses import StreamingResponse
from module_admin.service.login_service import LoginService
@ -96,6 +96,7 @@ def remove_style_hint(question: str) -> Tuple[str, bool]:
@ragflowController.post('/converse_with_chat_assistant')
async def converse_with_chat_assistant(
request: Request,
background_tasks: BackgroundTasks,
converse_params: ConverseWithChatAssistantModel,
):
"""
@ -202,20 +203,21 @@ async def converse_with_chat_assistant(
store_hash = get_question_hash(cache_question)
logger.info(f'[RAG_CACHE] 准备存储 | chat_id={converse_params.chat_id} | question={cache_question} | hash={store_hash}')
async def make_cache_store(chat_id: str, question: str):
async def store_answer(answer: str):
if redis and answer and len(answer.strip()) >= 10:
try:
await _async_store_qa(chat_id, question, answer, redis)
logger.info(f'[RAG_CACHE] 语义缓存存储成功 | chat_id={chat_id} | answer_length={len(answer)}')
except Exception as e:
logger.warning(f'语义缓存存储失败: {e}')
return store_answer
answer_queue = asyncio.Queue()
cache_store = await make_cache_store(converse_params.chat_id, cache_question)
async def cache_store_background():
try:
answer = await answer_queue.get()
if redis and answer and len(answer.strip()) >= 10:
await _async_store_qa(converse_params.chat_id, cache_question, answer, redis)
logger.info(f'[RAG_CACHE] 语义缓存存储成功 | chat_id={converse_params.chat_id} | answer_length={len(answer)}')
except Exception as e:
logger.warning(f'[RAG_CACHE] 语义缓存存储失败: {e}')
background_tasks.add_task(cache_store_background)
return StreamingResponse(
stream_ragflow_response(result, converse_params.chat_id, start_time, cache_store_func=cache_store),
stream_ragflow_response(result, converse_params.chat_id, start_time, answer_queue=answer_queue),
media_type='text/event-stream',
headers={
'Cache-Control': 'no-cache',
@ -230,7 +232,7 @@ async def converse_with_chat_assistant(
return ResponseUtil.error(msg=f'对话服务异常: {str(e)}')
def stream_ragflow_response(result: Generator, chat_id: str, start_time: float, cache_store_func=None) -> Generator[str, None, None]:
def stream_ragflow_response(result: Generator, chat_id: str, start_time: float, answer_queue=None) -> Generator[str, None, None]:
"""
流式处理RAGFlow响应 - 同步版本，修复首token延迟
"""
@ -323,13 +325,17 @@ def stream_ragflow_response(result: Generator, chat_id: str, start_time: float,
total_time = time.perf_counter() - start_time
logger.info(f'[RAG_SERVER {time.time():.3f}] ⏱️ Total Stream Duration ({chat_id}): {total_time:.3f}s')
if cache_store_func and last_answer and len(last_answer.strip()) >= 10:
if answer_queue and last_answer and len(last_answer.strip()) >= 10:
try:
import asyncio
asyncio.run(cache_store_func(last_answer))
logger.info(f'[RAG_CACHE] 缓存存储完成 | chat_id={chat_id} | answer_length={len(last_answer)}')
loop = asyncio.get_event_loop()
if loop.is_running():
asyncio.run_coroutine_threadsafe(answer_queue.put(last_answer), loop)
else:
loop.run_until_complete(answer_queue.put(last_answer))
logger.info(f'[RAG_CACHE] 已将答案放入队列 | chat_id={chat_id} | answer_length={len(last_answer)}')
except Exception as cache_err:
logger.warning(f'[RAG_CACHE] 缓存存储失败: {cache_err}')
logger.warning(f'[RAG_CACHE] 队列写入失败: {cache_err}')
# 聊天助手列表