Compare commits

..

2 Commits

2 changed files with 633 additions and 437 deletions

View File

@ -1,18 +1,15 @@
import milvus_connection import milvus_connection
from traceback import format_exc from traceback import format_exc
import argparse import argparse
import logging
from aiohttp import web from aiohttp import web
from llmengine.base_connection import get_connection_class from llmengine.base_connection import get_connection_class
from appPublic.registerfunction import RegisterFunction from appPublic.registerfunction import RegisterFunction
from appPublic.log import debug, exception from appPublic.log import debug, error, info, exception
from ahserver.serverenv import ServerEnv from ahserver.serverenv import ServerEnv
from ahserver.webapp import webserver from ahserver.webapp import webserver
import os import os
import json import json
logger = logging.getLogger(__name__)
helptext = """Milvus Connection Service API (using pymilvus Collection API): helptext = """Milvus Connection Service API (using pymilvus Collection API):
1. Create Collection Endpoint: 1. Create Collection Endpoint:
@ -20,48 +17,54 @@ path: /v1/createcollection
method: POST method: POST
headers: {"Content-Type": "application/json"} headers: {"Content-Type": "application/json"}
data: { data: {
"db_type": "textdb" "db_type": "textdb" // 可选若不提供则使用默认集合 ragdb
} }
response: response:
- Success: HTTP 200, {"status": "success", "collection_name": "ragdb_textdb", "message": "集合 ragdb_textdb 创建成功"} - Success: HTTP 200, {"status": "success", "collection_name": "ragdb" or "ragdb_textdb", "message": "集合 ragdb 或 ragdb_textdb 创建成功"}
- Error: HTTP 400, {"status": "error", "collection_name": "ragdb_textdb", "message": "<error message>"} - Error: HTTP 400, {"status": "error", "collection_name": "ragdb" or "ragdb_textdb", "message": "<error message>"}
2. Delete Collection Endpoint: 2. Delete Collection Endpoint:
path: /v1/deletecollection path: /v1/deletecollection
method: POST method: POST
headers: {"Content-Type": "application/json"} headers: {"Content-Type": "application/json"}
data: { data: {
"db_type": "textdb" "db_type": "textdb" // 可选若不提供则删除默认集合 ragdb
} }
response: response:
- Success: HTTP 200, {"status": "success", "collection_name": "ragdb_textdb", "message": "集合 ragdb_textdb 删除成功"} - Success: HTTP 200, {"status": "success", "collection_name": "ragdb" or "ragdb_textdb", "message": "集合 ragdb 或 ragdb_textdb 删除成功"}
- Error: HTTP 400, {"status": "error", "collection_name": "ragdb_textdb", "message": "<error message>"} - Success (collection does not exist): HTTP 200, {"status": "success", "collection_name": "ragdb" or "ragdb_textdb", "message": "集合 ragdb 或 ragdb_textdb 不存在,无需删除"}
- Error: HTTP 400, {"status": "error", "collection_name": "ragdb" or "ragdb_textdb", "message": "<error message>"}
3. Insert File Endpoint: 3. Insert File Endpoint:
path: /v1/insertfile path: /v1/insertfile
method: POST method: POST
headers: {"Content-Type": "application/json"} headers: {"Content-Type": "application/json"}
data: { data: {
"file_path": "/path/to/file.txt", "file_path": "/path/to/file.txt", // 必填文件路径
"userid": "user1", "userid": "user123", // 必填用户 ID
"db_type": "textdb" "db_type": "textdb", // 可选若不提供则使用默认集合 ragdb
"knowledge_base_id": "kb123" // 必填知识库 ID
} }
response: response:
- Success: HTTP 200, {"status": "success", "document_id": "<uuid>", "collection_name": "ragdb_textdb", "message": "文件 /path/to/file.txt 成功嵌入并处理三元组"} - Success: HTTP 200, {"status": "success", "document_id": "<uuid>", "collection_name": "ragdb" or "ragdb_textdb", "message": "文件 <file_path> 成功嵌入并处理三元组", "status_code": 200}
- Error: HTTP 400, {"status": "error", "document_id": "<uuid>", "collection_name": "ragdb_textdb", "message": "<error message>"} - Success (triples failed): HTTP 200, {"status": "success", "document_id": "<uuid>", "collection_name": "ragdb" or "ragdb_textdb", "message": "文件 <file_path> 成功嵌入,但三元组处理失败: <error>", "status_code": 200}
- Error: HTTP 400, {"status": "error", "document_id": "", "collection_name": "ragdb" or "ragdb_textdb", "message": "<error message>", "status_code": 400}
4. Delete File Endpoint: 4. Delete Document Endpoint:
path: /v1/deletefile path: /v1/deletefile
method: POST method: POST
headers: {"Content-Type": "application/json"} headers: {"Content-Type": "application/json"}
data: { data: {
"db_type": "textdb", "userid": "user123", // 必填用户 ID
"userid": "user1", "filename": "file.txt", // 必填文件名
"filename": "test.txt" "db_type": "textdb", // 可选若不提供则使用默认集合 ragdb
"knowledge_base_id": "kb123" // 必填知识库 ID
} }
response: response:
- Success: HTTP 200, {"status": "success", "collection_name": "ragdb_textdb", "message": "成功删除 X 条记录userid=user1, filename=test.txt"} - Success: HTTP 200, {"status": "success", "document_id": "<uuid1,uuid2>", "collection_name": "ragdb" or "ragdb_textdb", "message": "成功删除 <count> 条 Milvus 记录,<nodes> 个 Neo4j 节点,<rels> 个 Neo4j 关系userid=<userid>, filename=<filename>", "status_code": 200}
- Error: HTTP 400, {"status": "error", "collection_name": "ragdb_textdb", "message": "<error message>"} - Success (no records): HTTP 200, {"status": "success", "document_id": "", "collection_name": "ragdb" or "ragdb_textdb", "message": "没有找到 userid=<userid>, filename=<filename>, knowledge_base_id=<knowledge_base_id> 的记录,无需删除", "status_code": 200}
- Success (collection missing): HTTP 200, {"status": "success", "document_id": "", "collection_name": "ragdb" or "ragdb_textdb", "message": "集合 <collection_name> 不存在,无需删除", "status_code": 200}
- Error: HTTP 400, {"status": "error", "document_id": "", "collection_name": "ragdb" or "ragdb_textdb", "message": "<error message>", "status_code": 400}
5. Fused Search Query Endpoint: 5. Fused Search Query Endpoint:
path: /v1/fusedsearchquery path: /v1/fusedsearchquery
@ -70,7 +73,7 @@ headers: {"Content-Type": "application/json"}
data: { data: {
"query": "苹果公司在北京开设新店", "query": "苹果公司在北京开设新店",
"userid": "user1", "userid": "user1",
"db_type": "textdb", "db_type": "textdb", // 可选若不提供则使用默认集合 ragdb
"file_paths": ["/path/to/file.txt"], "file_paths": ["/path/to/file.txt"],
"limit": 5, "limit": 5,
"offset": 0, "offset": 0,
@ -103,7 +106,7 @@ headers: {"Content-Type": "application/json"}
data: { data: {
"query": "知识图谱的知识融合是什么?", "query": "知识图谱的知识融合是什么?",
"userid": "user1", "userid": "user1",
"db_type": "textdb", "db_type": "textdb", // 可选若不提供则使用默认集合 ragdb
"file_paths": ["/path/to/file.txt"], "file_paths": ["/path/to/file.txt"],
"limit": 5, "limit": 5,
"offset": 0, "offset": 0,
@ -165,6 +168,21 @@ response:
path: /v1/docs path: /v1/docs
method: GET method: GET
response: This help text response: This help text
10.Delete Knowledge Base Endpoint:
path: /v1/deleteknowledgebase
method: POST
headers: {"Content-Type": "application/json"}
data: {
"userid": "user123", // 必填用户 ID
"knowledge_base_id": "kb123",// 必填知识库 ID
"db_type": "textdb" // 可选若不提供则使用默认集合 ragdb
}
response:
- Success: HTTP 200, {"status": "success", "document_id": "<uuid1,uuid2>", "collection_name": "ragdb" or "ragdb_textdb", "message": "成功删除 <count> 条 Milvus 记录,<nodes> 个 Neo4j 节点,<rels> 个 Neo4j 关系userid=<userid>, knowledge_base_id=<knowledge_base_id>", "status_code": 200}
- Success (no records): HTTP 200, {"status": "success", "document_id": "", "collection_name": "ragdb" or "ragdb_textdb", "message": "没有找到 userid=<userid>, knowledge_base_id=<knowledge_base_id> 的记录,无需删除", "status_code": 200}
- Success (collection missing): HTTP 200, {"status": "success", "document_id": "", "collection_name": "ragdb" or "ragdb_textdb", "message": "集合 <collection_name> 不存在,无需删除", "status_code": 200}
- Error: HTTP 400, {"status": "error", "document_id": "", "collection_name": "ragdb" or "ragdb_textdb", "message": "<error message>", "status_code": 400}
""" """
def init(): def init():
@ -173,6 +191,7 @@ def init():
rf.register('deletecollection', delete_collection) rf.register('deletecollection', delete_collection)
rf.register('insertfile', insert_file) rf.register('insertfile', insert_file)
rf.register('deletefile', delete_file) rf.register('deletefile', delete_file)
rf.register('deleteknowledgebase', delete_knowledge_base)
rf.register('fusedsearchquery', fused_search_query) rf.register('fusedsearchquery', fused_search_query)
rf.register('searchquery', search_query) rf.register('searchquery', search_query)
rf.register('listuserfiles', list_user_files) rf.register('listuserfiles', list_user_files)
@ -192,22 +211,17 @@ async def create_collection(request, params_kw, *params, **kw):
debug(f'{params_kw=}') debug(f'{params_kw=}')
se = ServerEnv() se = ServerEnv()
engine = se.engine engine = se.engine
db_type = params_kw.get('db_type') db_type = params_kw.get('db_type', '')
if db_type is None: collection_name = "ragdb" if not db_type else f"ragdb_{db_type}"
debug(f'db_type 未提供')
return web.json_response({
"status": "error",
"message": "db_type 参数未提供"
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
try: try:
result = await engine.handle_connection("create_collection", {"db_type": db_type}) result = await engine.handle_connection("create_collection", {"db_type": db_type})
debug(f'{result=}') debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False)) return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e: except Exception as e:
debug(f'创建集合失败: {str(e)}') error(f'创建集合失败: {str(e)}')
return web.json_response({ return web.json_response({
"status": "error", "status": "error",
"collection_name": f"ragdb_{db_type}", "collection_name": collection_name,
"message": str(e) "message": str(e)
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400) }, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
@ -215,82 +229,128 @@ async def delete_collection(request, params_kw, *params, **kw):
debug(f'{params_kw=}') debug(f'{params_kw=}')
se = ServerEnv() se = ServerEnv()
engine = se.engine engine = se.engine
db_type = params_kw.get('db_type') db_type = params_kw.get('db_type', '')
if db_type is None: collection_name = "ragdb" if not db_type else f"ragdb_{db_type}"
debug(f'db_type 未提供')
return web.json_response({
"status": "error",
"message": "db_type 参数未提供"
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
try: try:
result = await engine.handle_connection("delete_collection", {"db_type": db_type}) result = await engine.handle_connection("delete_collection", {"db_type": db_type})
debug(f'{result=}') debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False)) return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e: except Exception as e:
debug(f'删除集合失败: {str(e)}') error(f'删除集合失败: {str(e)}')
return web.json_response({ return web.json_response({
"status": "error", "status": "error",
"collection_name": f"ragdb_{db_type}", "collection_name": collection_name,
"message": str(e) "message": str(e)
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400) }, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
async def insert_file(request, params_kw, *params, **kw): async def insert_file(request, params_kw, *params, **kw):
debug(f'{params_kw=}') debug(f'Received params: {params_kw=}')
se = ServerEnv() se = ServerEnv()
engine = se.engine engine = se.engine
file_path = params_kw.get('file_path') file_path = params_kw.get('file_path', '')
userid = params_kw.get('userid') userid = params_kw.get('userid', '')
db_type = params_kw.get('db_type') db_type = params_kw.get('db_type', '')
if not all([file_path, userid, db_type]): knowledge_base_id = params_kw.get('knowledge_base_id', '')
debug(f'file_path, userid 或 db_type 未提供') collection_name = "ragdb" if not db_type else f"ragdb_{db_type}"
return web.json_response({
"status": "error",
"message": "file_path, userid 或 db_type 未提供"
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
try: try:
# 仅检查必填字段是否存在
required_fields = ['file_path', 'userid', 'knowledge_base_id']
missing_fields = [field for field in required_fields if field not in params_kw or not params_kw[field]]
if missing_fields:
raise ValueError(f"缺少必填字段: {', '.join(missing_fields)}")
debug(
f'Calling insert_document with: file_path={file_path}, userid={userid}, db_type={db_type}, knowledge_base_id={knowledge_base_id}')
result = await engine.handle_connection("insert_document", { result = await engine.handle_connection("insert_document", {
"file_path": file_path, "file_path": file_path,
"userid": userid, "userid": userid,
"db_type": db_type "db_type": db_type,
"knowledge_base_id": knowledge_base_id
}) })
debug(f'{result=}') debug(f'Insert result: {result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False)) # 根据 result 的 status 设置 HTTP 状态码
status = 200 if result.get("status") == "success" else 400
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=status)
except Exception as e: except Exception as e:
debug(f'插入文件失败: {str(e)}') error(f'插入文件失败: {str(e)}')
return web.json_response({ return web.json_response({
"status": "error", "status": "error",
"collection_name": collection_name,
"document_id": "", "document_id": "",
"collection_name": f"ragdb_{db_type}",
"message": str(e) "message": str(e)
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400) }, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
async def delete_file(request, params_kw, *params, **kw): async def delete_file(request, params_kw, *params, **kw):
debug(f'{params_kw=}') debug(f'Received delete_file params: {params_kw=}')
se = ServerEnv() se = ServerEnv()
engine = se.engine engine = se.engine
db_type = params_kw.get('db_type') userid = params_kw.get('userid', '')
userid = params_kw.get('userid') filename = params_kw.get('filename', '')
filename = params_kw.get('filename') db_type = params_kw.get('db_type', '')
if not all([db_type, userid, filename]): knowledge_base_id = params_kw.get('knowledge_base_id', '')
debug(f'db_type, userid 或 filename 未提供') collection_name = "ragdb" if not db_type else f"ragdb_{db_type}"
return web.json_response({
"status": "error",
"message": "db_type, userid 或 filename 未提供"
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
try: try:
required_fields = ['userid', 'filename', 'knowledge_base_id']
missing_fields = [field for field in required_fields if field not in params_kw or not params_kw[field]]
if missing_fields:
raise ValueError(f"缺少必填字段: {', '.join(missing_fields)}")
debug(
f'Calling delete_document with: userid={userid}, filename={filename}, db_type={db_type}, knowledge_base_id={knowledge_base_id}')
result = await engine.handle_connection("delete_document", { result = await engine.handle_connection("delete_document", {
"db_type": db_type,
"userid": userid, "userid": userid,
"filename": filename "filename": filename,
"db_type": db_type,
"knowledge_base_id": knowledge_base_id
}) })
debug(f'{result=}') debug(f'Delete result: {result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False)) status = 200 if result.get("status") == "success" else 400
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=status)
except Exception as e: except Exception as e:
debug(f'删除文件失败: {str(e)}') error(f'删除文件失败: {str(e)}')
return web.json_response({ return web.json_response({
"status": "error", "status": "error",
"collection_name": f"ragdb_{db_type}", "collection_name": collection_name,
"message": str(e) "document_id": "",
"message": str(e),
"status_code": 400
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
async def delete_knowledge_base(request, params_kw, *params, **kw):
debug(f'Received delete_knowledge_base params: {params_kw=}')
se = ServerEnv()
engine = se.engine
userid = params_kw.get('userid', '')
knowledge_base_id = params_kw.get('knowledge_base_id', '')
db_type = params_kw.get('db_type', '')
collection_name = "ragdb" if not db_type else f"ragdb_{db_type}"
try:
required_fields = ['userid', 'knowledge_base_id']
missing_fields = [field for field in required_fields if field not in params_kw or not params_kw[field]]
if missing_fields:
raise ValueError(f"缺少必填字段: {', '.join(missing_fields)}")
debug(
f'Calling delete_knowledge_base with: userid={userid}, knowledge_base_id={knowledge_base_id}, db_type={db_type}')
result = await engine.handle_connection("delete_knowledge_base", {
"userid": userid,
"knowledge_base_id": knowledge_base_id,
"db_type": db_type
})
debug(f'Delete knowledge base result: {result=}')
status = 200 if result.get("status") == "success" else 400
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=status)
except Exception as e:
error(f'删除知识库失败: {str(e)}')
return web.json_response({
"status": "error",
"collection_name": collection_name,
"document_id": "",
"message": str(e),
"status_code": 400
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400) }, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
async def fused_search_query(request, params_kw, *params, **kw): async def fused_search_query(request, params_kw, *params, **kw):
@ -299,16 +359,16 @@ async def fused_search_query(request, params_kw, *params, **kw):
engine = se.engine engine = se.engine
query = params_kw.get('query') query = params_kw.get('query')
userid = params_kw.get('userid') userid = params_kw.get('userid')
db_type = params_kw.get('db_type') db_type = params_kw.get('db_type', '')
file_paths = params_kw.get('file_paths') file_paths = params_kw.get('file_paths')
limit = params_kw.get('limit', 5) limit = params_kw.get('limit', 5)
offset = params_kw.get('offset', 0) offset = params_kw.get('offset', 0)
use_rerank = params_kw.get('use_rerank', True) use_rerank = params_kw.get('use_rerank', True)
if not all([query, userid, db_type, file_paths]): if not all([query, userid, file_paths]):
debug(f'query, userid, db_type 或 file_paths 未提供') debug(f'query, userid 或 file_paths 未提供')
return web.json_response({ return web.json_response({
"status": "error", "status": "error",
"message": "query, userid, db_type 或 file_paths 未提供" "message": "query, userid 或 file_paths 未提供"
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400) }, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
try: try:
result = await engine.handle_connection("fused_search", { result = await engine.handle_connection("fused_search", {
@ -323,7 +383,7 @@ async def fused_search_query(request, params_kw, *params, **kw):
debug(f'{result=}') debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False)) return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e: except Exception as e:
debug(f'融合搜索失败: {str(e)}') error(f'融合搜索失败: {str(e)}')
return web.json_response({ return web.json_response({
"status": "error", "status": "error",
"message": str(e) "message": str(e)
@ -335,16 +395,16 @@ async def search_query(request, params_kw, *params, **kw):
engine = se.engine engine = se.engine
query = params_kw.get('query') query = params_kw.get('query')
userid = params_kw.get('userid') userid = params_kw.get('userid')
db_type = params_kw.get('db_type') db_type = params_kw.get('db_type', '')
file_paths = params_kw.get('file_paths') file_paths = params_kw.get('file_paths')
limit = params_kw.get('limit', 5) limit = params_kw.get('limit', 5)
offset = params_kw.get('offset', 0) offset = params_kw.get('offset', 0)
use_rerank = params_kw.get('use_rerank', True) use_rerank = params_kw.get('use_rerank', True)
if not all([query, userid, db_type, file_paths]): if not all([query, userid, file_paths]):
debug(f'query, userid, db_type 或 file_paths 未提供') debug(f'query, userid 或 file_paths 未提供')
return web.json_response({ return web.json_response({
"status": "error", "status": "error",
"message": "query, userid, db_type 或 file_paths 未提供" "message": "query, userid 或 file_paths 未提供"
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400) }, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
try: try:
result = await engine.handle_connection("search_query", { result = await engine.handle_connection("search_query", {
@ -359,7 +419,7 @@ async def search_query(request, params_kw, *params, **kw):
debug(f'{result=}') debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False)) return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e: except Exception as e:
debug(f'纯向量搜索失败: {str(e)}') error(f'纯向量搜索失败: {str(e)}')
return web.json_response({ return web.json_response({
"status": "error", "status": "error",
"message": str(e) "message": str(e)
@ -383,7 +443,7 @@ async def list_user_files(request, params_kw, *params, **kw):
debug(f'{result=}') debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False)) return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e: except Exception as e:
debug(f'查询用户文件列表失败: {str(e)}') error(f'查询用户文件列表失败: {str(e)}')
return web.json_response({ return web.json_response({
"status": "error", "status": "error",
"message": str(e) "message": str(e)
@ -406,7 +466,7 @@ async def handle_connection(request, params_kw, *params, **kw):
debug(f'{result=}') debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False)) return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e: except Exception as e:
debug(f'处理连接操作失败: {str(e)}') error(f'处理连接操作失败: {str(e)}')
return web.json_response({ return web.json_response({
"status": "error", "status": "error",
"message": str(e) "message": str(e)
@ -418,7 +478,7 @@ def main():
parser.add_argument('-p', '--port', default='8888') parser.add_argument('-p', '--port', default='8888')
parser.add_argument('connection_path') parser.add_argument('connection_path')
args = parser.parse_args() args = parser.parse_args()
logger.debug(f"Arguments: {args}") debug(f"Arguments: {args}")
Klass = get_connection_class(args.connection_path) Klass = get_connection_class(args.connection_path)
se = ServerEnv() se = ServerEnv()
se.engine = Klass() se.engine = Klass()

File diff suppressed because it is too large Load Diff