diff --git a/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py b/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py index 9b68d83..f6a2c44 100644 --- a/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py +++ b/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py @@ -123,32 +123,46 @@ def stream_ragflow_response(result: Generator, chat_id: str, start_time: float) """ 流式处理RAGFlow响应 - 同步版本，修复首token延迟 """ + import time + + server_stream_start = time.time() + logger.info(f"[RAG_SERVER {server_stream_start:.3f}] 🚀 开始流式响应处理，chat_id: {chat_id}") + last_answer = "" first_token_received = False start_stream_time = time.perf_counter() # 立即发送连接建立消息，解决首token延迟 - yield "event: ping\ndata: {\"status\": \"connected\"}\n\n" + connection_time = time.time() + ping_message = "event: ping\ndata: {\"status\": \"connected\"}\n\n" + yield ping_message + logger.info(f"[RAG_SERVER {connection_time:.3f}] 📡 连接建立ping发送，耗时: {connection_time - server_stream_start:.3f}s") + chunk_count = 0 try: for chunk in result: + chunk_time = time.time() + chunk_count += 1 + # 检查第一个token的延迟 if not first_token_received: first_token_received = True latency = time.perf_counter() - start_stream_time - logger.info(f"[RAG_PROFILE] Time to First Token ({chat_id}): {latency:.3f}s") + logger.info(f"[RAG_SERVER {chunk_time:.3f}] 🎯 首token到达，耗时: {latency:.3f}s") # 处理chunk数据 payload = chunk.get('data') if isinstance(chunk, dict) else chunk if not payload: + logger.info(f"[RAG_SERVER {chunk_time:.3f}] ⚠️ 空chunk跳过") continue body = payload if isinstance(payload, dict) else {'data': payload} # 处理错误 if isinstance(chunk, dict) and chunk.get('code') and chunk.get('code') != 0: - logger.error(f"RAGFlow Stream Error: {chunk}") - yield format_sse({'message': chunk.get('message', '流式处理异常')}, event='error') + logger.error(f"[RAG_SERVER {chunk_time:.3f}] ❌ RAGFlow Stream Error: {chunk}") + error_message = format_sse({'message': chunk.get('message', '流式处理异常')}, event='error') + yield error_message continue # 处理answer字段的增量更新 @@ -161,23 +175,32 @@ def stream_ragflow_response(result: Generator, chat_id: str, start_time: float) if delta: body['answer'] = delta last_answer = current_answer + logger.info(f"[RAG_SERVER {chunk_time:.3f}] 📝 Chunk #{chunk_count} 处理完成，delta长度: {len(delta)}") yield format_sse(body) else: # 上下文重置的备用处理 last_answer = current_answer + logger.info(f"[RAG_SERVER {chunk_time:.3f}] 🔄 Chunk #{chunk_count} 上下文重置") yield format_sse(body) else: + logger.info(f"[RAG_SERVER {chunk_time:.3f}] 📦 Chunk #{chunk_count} 其他数据: {body}") yield format_sse(body) # 流结束 - yield format_sse({'status': 'completed'}, event='end') + stream_end_time = time.time() + end_message = format_sse({'status': 'completed'}, event='end') + yield end_message + logger.info(f"[RAG_SERVER {stream_end_time:.3f}] 🏁 流式响应完成，chat_id: {chat_id}") + logger.info(f"[RAG_SERVER {stream_end_time:.3f}] 📊 总共处理chunk数量: {chunk_count}") except Exception as exc: - logger.exception(f'[RAG_PROFILE] ragflow流式对话异常: {exc}') - yield format_sse({'message': str(exc)}, event='error') + error_time = time.time() + logger.exception(f'[RAG_SERVER {error_time:.3f}] ragflow流式对话异常: {exc}') + error_message = format_sse({'message': str(exc)}, event='error') + yield error_message finally: total_time = time.perf_counter() - start_time - logger.info(f'[RAG_PROFILE] Total Stream Duration ({chat_id}): {total_time:.3f}s') + logger.info(f'[RAG_SERVER {time.time():.3f}] ⏱️ Total Stream Duration ({chat_id}): {total_time:.3f}s') # 聊天助手列表 diff --git a/ruoyi-fastapi-backend/test/test_sse.py b/ruoyi-fastapi-backend/test/test_sse.py index 4d7b142..1e4ea6f 100644 --- a/ruoyi-fastapi-backend/test/test_sse.py +++ b/ruoyi-fastapi-backend/test/test_sse.py @@ -11,51 +11,78 @@ import requests # 同步 HTTP 客户端 # 下面的默认参数可以通过环境变量覆盖，避免把机密写死在代码里。 API_URL = os.environ.get( - "RAGFLOW_URL", - "http://10.0.0.202:9099/system/ragflow/converse_with_chat_assistant", + "LOCAL_API_URL", + "http://localhost:9099/system/ragflow/converse_with_chat_assistant", ) AUTH_TOKEN = os.environ.get( - "RAGFLOW_TOKEN", + "LOCAL_API_TOKEN", "Bearer ", ) DEFAULT_CHAT_ID = os.environ.get( - "RAGFLOW_CHAT_ID", "db4bb966895b11f08cda0242ac130006" + "LOCAL_CHAT_ID", "db4bb966895b11f08cda0242ac130006" ) DEFAULT_SESSION_ID = os.environ.get( - "RAGFLOW_SESSION_ID", "38d765e48a3811f0be310242ac130006" + "LOCAL_SESSION_ID", "38d765e48a3811f0be310242ac130006" ) DEFAULT_QUESTION = os.environ.get( - "RAGFLOW_QUESTION", "你好，请用简洁的语言介绍你自己" + "LOCAL_QUESTION", "你好，请用简洁的语言介绍你自己" ) LOGIN_URL = os.environ.get( - "RAGFLOW_LOGIN_URL", - "http://10.0.0.202:9099/login", + "LOCAL_LOGIN_URL", + "http://localhost:9099/login", ) +def print_log(message: str): + """打印带时间戳的日志""" + current_time = time.time() + print(f"[CLIENT {current_time:.3f}] {message}") + + async def login_get_token() -> str: + """ + 获取本地服务token + """ + print_log("开始登录获取token...") + login_start = time.time() + payload = { "username": "admin", "password": "admin123" } headers = { - "Authorization": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoiMSIsInVzZXJfbmFtZSI6ImFkbWluIiwiZGVwdF9uYW1lIjoiXHU3ODE0XHU1M2QxXHU5MGU4XHU5NWU4Iiwic2Vzc2lvbl9pZCI6IjYwNjdlMzkxLTRiNGMtNGUwYy1hZjM5LTVmNzczZjIzMmE3OSIsImxvZ2luX2luZm8iOnsiaXBhZGRyIjpudWxsLCJsb2dpbkxvY2F0aW9uIjoiXHU2NzJhXHU3N2U1IiwiYnJvd3NlciI6Ik90aGVyIiwib3MiOiJPdGhlciIsImxvZ2luVGltZSI6IjIwMjUtMTItMDIgMTA6NTk6NDAifSwiZXhwIjoxNzY3MjM2MzgwfQ.NM9j0emNz8trxU1DX87PVZZaWWHlFFwF7o0Pxop6B5c" + "Content-Type": "application/json" } async with aiohttp.ClientSession() as session: - # 使用data参数发送form-data而不是json async with session.post( - LOGIN_URL, data=payload, headers=headers, timeout=60 + LOGIN_URL, json=payload, headers=headers, timeout=60 ) as resp: - response_data = await resp.json() # 将响应转换为JSON - token = response_data.get("token") # 获取token字段 - return token + if resp.status == 200: + response_data = await resp.json() + token = response_data.get("token") or response_data.get("access_token") + login_end = time.time() + print_log(f"登录成功，耗时: {login_end - login_start:.3f}s") + if token: + return token + else: + print_log("登录响应无token，返回空字符串") + return "" + else: + print_log(f"登录失败，状态码: {resp.status}") + return "" async def stream_chat_async(question: str, chat_id: str, session_id: str) -> str: """ 异步版（aiohttp），适合已有事件循环的应用，不阻塞主线程。 """ + client_start = time.time() + print_log(f"开始SSE流式请求 - 问题: {question}") + print_log(f"chat_id: {chat_id}, session_id: {session_id}") + token = await login_get_token() + token_received = time.time() + print_log(f"Token获取完成，耗时: {token_received - client_start:.3f}s") payload = { "chatId": chat_id, @@ -68,15 +95,24 @@ async def stream_chat_async(question: str, chat_id: str, session_id: str) -> str "Accept": "text/event-stream", } + print_log("发送HTTP请求到服务器...") + request_sent = time.time() + print_log(f"请求耗时: {request_sent - token_received:.3f}s") + answer_parts = [] event_type = "message" + first_chunk_received = False + chunk_count = 0 async with aiohttp.ClientSession() as session: async with session.post( API_URL, json=payload, headers=headers, timeout=60 ) as resp: - print( - f"# status={resp.status} content-type={resp.headers.get('content-type')}") + response_received = time.time() + print_log(f"HTTP响应接收完成，状态码: {resp.status}") + print_log(f"连接建立耗时: {response_received - request_sent:.3f}s") + print_log(f"Content-Type: {resp.headers.get('content-type')}") + resp.raise_for_status() while True: @@ -90,35 +126,44 @@ async def stream_chat_async(question: str, chat_id: str, session_id: str) -> str event_type = line.split(":", 1)[1].strip() or "message" continue if not line.startswith("data:"): - print(f"[skip] {line}") continue data_str = line[len("data:"):].strip() if not data_str: continue + + current_time = time.time() + + if not first_chunk_received: + first_chunk_received = True + first_chunk_time = current_time + print_log(f"🎯 首块数据到达，耗时: {first_chunk_time - response_received:.3f}s") + try: payload_obj = json.loads(data_str) except json.JSONDecodeError: - print(f"[{event_type}] {data_str}") + print_log(f"[{event_type}] {data_str}") event_type = "message" continue if event_type == "end" or payload_obj.get("status") == "completed": - print("\n[stream] completed") + print_log(f"🏁 流式响应完成，总耗时: {current_time - client_start:.3f}s") + print_log(f"📊 接收到的chunk数量: {chunk_count}") break if payload_obj.get("data") is True: event_type = "message" continue if "answer" in payload_obj: piece = payload_obj.get("answer", "") + chunk_count += 1 answer_parts.append(piece) # 流式输出：每个片段都单独打印，并且添加延时让效果更明显 print(piece, end="", flush=True) time.sleep(0.1) # 每个片段间隔100ms，让流式效果更明显 else: - print(f"[{event_type}] {payload_obj}") + print_log(f"[{event_type}] {payload_obj}") event_type = "message" - print("\n\n[stream completed] - Answer received in real-time via streaming!") + print("\n\n[CLIENT] 流式响应完成 - 通过SSE实时接收！") full_answer = "".join(answer_parts) return full_answer diff --git a/ruoyi-fastapi-backend/utils/ragflow_util.py b/ruoyi-fastapi-backend/utils/ragflow_util.py index 61fb26f..c6772e4 100644 --- a/ruoyi-fastapi-backend/utils/ragflow_util.py +++ b/ruoyi-fastapi-backend/utils/ragflow_util.py @@ -54,6 +54,12 @@ class RAGFlowClient: def _stream_request(self, method: str, endpoint: str, **kwargs) -> Generator[Dict[str, Any], None, None]: """发送流式HTTP请求，修复SSE缓冲问题""" + import time + + # 添加详细的时间戳日志 + server_start_time = time.time() + print(f"[RAG_CLIENT {server_start_time:.3f}] 🔄 开始流式请求: {method} {endpoint}") + url = f"{self.base_url}{endpoint}" headers = kwargs.pop('headers', self.headers.copy()) @@ -64,18 +70,39 @@ class RAGFlowClient: 'X-Accel-Buffering': 'no' # Nginx禁用缓冲 }) + print(f"[RAG_CLIENT {time.time():.3f}] 📤 发送HTTP请求到: {url}") response = requests.request(method, url, headers=headers, stream=True, **kwargs) + response_time = time.time() + print(f"[RAG_CLIENT {response_time:.3f}] 📥 HTTP响应接收完成，状态码: {response.status_code}") + print(f"[RAG_CLIENT {response_time:.3f}] ⏱️ 连接建立耗时: {response_time - server_start_time:.3f}s") + # 立即处理响应，yield每个chunk + chunk_count = 0 + first_chunk_sent = False + for line in response.iter_lines(): if line: + line_time = time.time() line = line.decode('utf-8') if line.startswith('data:'): try: data = json.loads(line[5:].strip()) + + if not first_chunk_sent: + first_chunk_sent = True + print(f"[RAG_CLIENT {line_time:.3f}] 🎯 首块数据开始发送，耗时: {line_time - response_time:.3f}s") + + chunk_count += 1 + print(f"[RAG_CLIENT {line_time:.3f}] 📦 Chunk #{chunk_count} 发送中... (耗时: {line_time - server_start_time:.3f}s)") yield data except json.JSONDecodeError: + print(f"[RAG_CLIENT {line_time:.3f}] ❌ JSON解析错误: {line}") continue + + end_time = time.time() + print(f"[RAG_CLIENT {end_time:.3f}] 🏁 流式请求完成，总耗时: {end_time - server_start_time:.3f}s") + print(f"[RAG_CLIENT {end_time:.3f}] 📊 总共发送chunk数量: {chunk_count}") # ==================== # OpenAI兼容API