From 43af078724c421fafe2bf0c565996b5b78b7dde0 Mon Sep 17 00:00:00 2001
From: Tian jianyong <11429339@qq.com>
Date: Wed, 17 Dec 2025 17:58:29 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BD=BF=E7=94=A8=20aiohttp=20=E5=92=8C?=
 =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E6=B1=A0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../module_admin/service/ragflow_service.py   | 59 +++++++++++++++----
 1 file changed, 48 insertions(+), 11 deletions(-)

diff --git a/ruoyi-fastapi-backend/module_admin/service/ragflow_service.py b/ruoyi-fastapi-backend/module_admin/service/ragflow_service.py
index 32255ad..3694759 100644
--- a/ruoyi-fastapi-backend/module_admin/service/ragflow_service.py
+++ b/ruoyi-fastapi-backend/module_admin/service/ragflow_service.py
@@ -10,8 +10,13 @@ from module_admin.entity.vo.ragflow_vo import (
     ConverseWithChatAssistantModel
 )
 from utils.ragflow_util import RAGFlowClient
+from utils.ragflow_asy_util import AsyncRAGFlowClient
 from config.env import RAGFlowConfig
 from utils.ragflow_client_manager import get_ragflow_client
+import asyncio
+import logging
+
+logger = logging.getLogger(__name__)
 
 
 class RAGFlowService:
@@ -19,6 +24,28 @@ class RAGFlowService:
         RAGFlow服务 - 简化版本，使用同步操作
     """
     
+    # 全局异步客户端实例 - 连接池复用
+    _async_client: AsyncRAGFlowClient = None
+    
+    @classmethod
+    def _get_async_client(cls) -> AsyncRAGFlowClient:
+        """获取或创建异步客户端实例 - 连接池复用"""
+        if cls._async_client is None:
+            logger.info("🔧 创建aiohttp连接池客户端...")
+            cls._async_client = AsyncRAGFlowClient(
+                base_url=RAGFlowConfig.RAGFLOW_BASE_URL,
+                api_key=RAGFlowConfig.RAGFLOW_API_KEY,
+                timeout=30,
+                connector_kwargs={
+                    "limit": 100,              # 连接池大小
+                    "limit_per_host": 30,      # 每主机连接数
+                    "keepalive_timeout": 30,   # Keep-alive超时
+                    "ssl": False               # 禁用SSL验证
+                }
+            )
+            logger.info("✅ aiohttp连接池创建完成")
+        return cls._async_client
+    
     # 获取数据集列表
     @classmethod
     def get_ragflow_dataset_list_services(cls, query_db, rage_flow_query: RagflowListQueryModel):
@@ -125,18 +152,28 @@ class RAGFlowService:
         result = client.create_session_with_chat(**(create_params.model_dump()))
         return result
     
-    # 与助手聊天 - 核心方法
+    # 与助手聊天 - 核心方法 (已优化为aiohttp版本)
     @classmethod
-    def converse_with_chat_assistant_services(cls, converse_params: ConverseWithChatAssistantModel):
-        """与聊天助手对话 - 同步版本，返回Generator"""
-        client = RAGFlowClient(RAGFlowConfig.RAGFLOW_BASE_URL, RAGFlowConfig.RAGFLOW_API_KEY)
-        # 直接返回Generator，支持流式响应
-        return client.converse_with_chat_assistant(
-            chat_id=converse_params.chat_id,
-            question=converse_params.question,
-            stream=converse_params.stream,
-            session_id=converse_params.session_id
-        )
+    async def converse_with_chat_assistant_services(cls, converse_params: ConverseWithChatAssistantModel):
+        """与聊天助手对话 - aiohttp异步版本，支持连接池复用"""
+        client = cls._get_async_client()
+        
+        # 使用异步客户端调用 - 利用连接池
+        endpoint = f"/api/v1/chats/{converse_params.chat_id}/conversations"
+        data = {
+            "question": converse_params.question,
+            "stream": converse_params.stream,
+            "session_id": converse_params.session_id
+        }
+        
+        logger.info(f"🚀 使用aiohttp连接池发起对话请求: chat_id={converse_params.chat_id}")
+        
+        if converse_params.stream:
+            # 流式响应
+            return client._stream_request('POST', endpoint, json=data)
+        else:
+            # 同步响应
+            return await client._request('POST', endpoint, json=data)
 
     # 与助手聊天 (OpenAI Compatible) - 同步版本
     @classmethod