rag/uapi_service.py

@@ -1,7 +1,6 @@
 from appPublic.log import debug, error
 from typing import Dict, Any, List
 import uuid
-import json
 from ahserver.serverenv import ServerEnv
 from uapi.appapi import UAPI
 from appPublic.dictObject import DictObject
@@ -14,10 +13,9 @@ class APIService:
         """调用嵌入服务获取文本向量"""
         try:
             uapi = UAPI(request, DictObject(**globals()))
-            debug(f'{uapi=}, {type(uapi.call)}')
+            debug(f'{uapi=}')
             params_kw = {"input": texts}
-            b = await uapi.call(upappid, apiname, user, params_kw)
-            d = json.loads(b.decode('utf-8'))
+            d = await uapi.request(upappid, apiname, user, params_kw)
             if d.get("object") != "list" or not d.get("data"):
                 error(f"嵌入服务响应格式错误: {d}")
                 raise RuntimeError("嵌入服务响应格式错误")
@@ -36,8 +34,7 @@ class APIService:
                 raise ValueError("查询文本不能为空")
             uapi = UAPI(request, DictObject(**globals()))
             params_kw = {"query": query}
-            b = await uapi.call(upappid, apiname, user, params_kw)
-            d = json.loads(b.decode('utf-8'))
+            d = await uapi.request(upappid, apiname, user, params_kw)
             if d.get("object") != "list" or not d.get("data"):
                 error(f"实体识别服务响应格式错误: {d}")
                 raise RuntimeError("实体识别服务响应格式错误")
@@ -57,8 +54,7 @@ class APIService:
         try:
             uapi = UAPI(request, DictObject(**globals()))
             params_kw = {"text": text}
-            b = await uapi.call(upappid, apiname, user, params_kw)
-            d = json.loads(b.decode('utf-8'))
+            d = await uapi.request(upappid, apiname, user, params_kw)
             if d.get("object") != "list":
                 error(f"request #{request_id} invalid response format: {d}")
                 raise RuntimeError("三元组抽取服务响应格式错误")
@@ -91,8 +87,7 @@ class APIService:
                 "documents": documents,
                 "top_n": top_n
             }
-            b = await uapi.call(upappid, apiname, user, params_kw)
-            d = json.loads(b.decode('utf-8'))
+            d = await uapi.request(upappid, apiname, user, params_kw)
             if d.get("object") != "rerank.result" or not d.get("data"):
                 error(f"重排序服务响应格式错误: {d}")
                 raise RuntimeError("重排序服务响应格式错误")
@@ -115,8 +110,7 @@ class APIService:
         try:
             uapi = UAPI(request, DictObject(**globals()))
             params_kw = {}
-            b = await uapi.call(upappid, apiname, user, params_kw)
-            d = json.loads(b.decode('utf-8'))
+            d = await uapi.request(upappid, apiname, user, params_kw)
             if d.get("status") != 200:
                 error(f"Neo4j 文档请求失败，状态码: {d.get('status')}")
                 raise RuntimeError(f"Neo4j 文档请求失败: {d.get('status')}")
@@ -131,9 +125,7 @@ class APIService:
         """初始化 Neo4j 服务"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {}
-        b = await uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def neo4j_insert_triples(self, request, triples: list, document_id: str, knowledge_base_id: str, userid: str, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """插入三元组到 Neo4j"""
@@ -144,25 +136,19 @@ class APIService:
             "knowledge_base_id": knowledge_base_id,
             "userid": userid
         }
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d 
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def neo4j_delete_document(self, request, document_id: str, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """删除指定文档"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {"document_id": document_id}
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def neo4j_delete_knowledgebase(self, request, userid: str, knowledge_base_id: str, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """删除用户知识库"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {"userid": userid, "knowledge_base_id": knowledge_base_id}
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def neo4j_match_triplets(self, request, query: str, query_entities: list, userid: str, knowledge_base_id: str, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """根据实体匹配相关三元组"""
@@ -173,26 +159,20 @@ class APIService:
             "userid": userid,
             "knowledge_base_id": knowledge_base_id
         }
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     # Milvus 服务
     async def milvus_create_collection(self, request, upappid: str, apiname: str, user: str, db_type: str = "") -> Dict[str, Any]:
         """创建 Milvus 集合"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {"db_type": db_type}
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def milvus_delete_collection(self, request, upappid: str, apiname: str, user: str, db_type: str = "") -> Dict[str, Any]:
         """删除 Milvus 集合"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {"db_type": db_type}
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def milvus_insert_document(self, request, chunks: List[Dict], upappid: str, apiname: str, user: str, db_type: str = "") -> Dict[str, Any]:
         """添加 Milvus 记录"""
@@ -201,9 +181,11 @@ class APIService:
             "chunks": chunks,
             "dbtype": db_type
         }
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        payload = json.dumps(params_kw)  # 转换为 JSON 字符串
+        payload_bytes = payload.encode()  # 编码为字节
+        payload_size = len(payload_bytes)  # 获取字节数
+        debug(f"request payload size for insertdocument: {payload_size} bytes")
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def milvus_delete_document(self, request, userid: str, file_path: str, knowledge_base_id: str, document_id: str, upappid: str, apiname: str, user: str, db_type: str = "") -> Dict[str, Any]:
         """删除 Milvus 记录"""
@@ -215,17 +197,13 @@ class APIService:
             "document_id": document_id,
             "dbtype": db_type
         }
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def milvus_delete_knowledgebase(self, request, userid: str, knowledge_base_id: str, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """删除 Milvus 知识库"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {"userid": userid, "knowledge_base_id": knowledge_base_id}
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def milvus_search_query(self, request, query_vector: List[float], userid: str, knowledge_base_ids: list, limit: int, offset: int, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """根据用户知识库检索 Milvus"""
@@ -237,38 +215,32 @@ class APIService:
             "limit": limit,
             "offset": offset
         }
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def milvus_list_user_files(self, request, userid: str, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """列出 Milvus 用户知识库列表"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {"userid": userid}
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def milvus_list_all_knowledgebases(self, request, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """列出 Milvus 数据库中所有数据"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {}
-        b = uapi.call(upappid, apiname, user, params_kw)
-        d = json.loads(b.decode('utf-8'))
-        return d
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     # RAG 服务
     async def rag_create_collection(self, request, upappid: str, apiname: str, user: str, db_type: str = "") -> Dict[str, Any]:
         """创建 RAG 集合"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {"db_type": db_type}
-        return await uapi.call(upappid, apiname, user, params_kw)
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def rag_delete_collection(self, request, upappid: str, apiname: str, user: str, db_type: str = "") -> Dict[str, Any]:
         """删除 RAG 集合"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {"db_type": db_type}
-        return await uapi.call(upappid, apiname, user, params_kw)
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def rag_insert_file(self, request, file_path: str, userid: str, knowledge_base_id: str, document_id: str, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """添加 RAG 记录"""
@@ -279,7 +251,7 @@ class APIService:
             "knowledge_base_id": knowledge_base_id,
             "document_id": document_id
         }
-        return await uapi.call(upappid, apiname, user, params_kw)
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def rag_delete_file(self, request, userid: str, file_path: str, knowledge_base_id: str, document_id: str, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """删除 RAG 记录"""
@@ -290,13 +262,13 @@ class APIService:
             "knowledge_base_id": knowledge_base_id,
             "document_id": document_id
         }
-        return await uapi.call(upappid, apiname, user, params_kw)
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def rag_delete_knowledgebase(self, request, userid: str, knowledge_base_id: str, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """删除 RAG 知识库"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {"userid": userid, "knowledge_base_id": knowledge_base_id}
-        return await uapi.call(upappid, apiname, user, params_kw)
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def rag_search_query(self, request, query: str, userid: str, knowledge_base_ids: list, limit: int, offset: int, use_rerank: bool, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """根据用户知识库检索 RAG"""
@@ -309,7 +281,7 @@ class APIService:
             "offset": offset,
             "use_rerank": use_rerank
         }
-        return await uapi.call(upappid, apiname, user, params_kw)
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def rag_fused_search_query(self, request, query: str, userid: str, knowledge_base_ids: list, limit: int, offset: int, use_rerank: bool, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """根据用户知识库+知识图谱检索 RAG"""
@@ -322,22 +294,22 @@ class APIService:
             "offset": offset,
             "use_rerank": use_rerank
         }
-        return await uapi.call(upappid, apiname, user, params_kw)
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def rag_list_user_files(self, request, userid: str, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """列出 RAG 用户知识库列表"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {"userid": userid}
-        return await uapi.call(upappid, apiname, user, params_kw)
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def rag_list_all_knowledgebases(self, request, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """列出 RAG 数据库中所有数据"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {}
-        return await uapi.call(upappid, apiname, user, params_kw)
+        return await uapi.request(upappid, apiname, user, params_kw)
 
     async def rag_docs(self, request, upappid: str, apiname: str, user: str) -> Dict[str, Any]:
         """获取 RAG 帮助文档"""
         uapi = UAPI(request, DictObject(**globals()))
         params_kw = {}
-        return await uapi.call(upappid, apiname, user, params_kw)
+        return await uapi.request(upappid, apiname, user, params_kw)