Compare commits

..

2 Commits

2 changed files with 633 additions and 437 deletions

View File

@ -1,18 +1,15 @@
import milvus_connection
from traceback import format_exc
import argparse
import logging
from aiohttp import web
from llmengine.base_connection import get_connection_class
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.webapp import webserver
import os
import json
logger = logging.getLogger(__name__)
helptext = """Milvus Connection Service API (using pymilvus Collection API):
1. Create Collection Endpoint:
@ -20,48 +17,54 @@ path: /v1/createcollection
method: POST
headers: {"Content-Type": "application/json"}
data: {
"db_type": "textdb"
"db_type": "textdb" // 可选若不提供则使用默认集合 ragdb
}
response:
- Success: HTTP 200, {"status": "success", "collection_name": "ragdb_textdb", "message": "集合 ragdb_textdb 创建成功"}
- Error: HTTP 400, {"status": "error", "collection_name": "ragdb_textdb", "message": "<error message>"}
- Success: 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>"}
2. Delete Collection Endpoint:
path: /v1/deletecollection
method: POST
headers: {"Content-Type": "application/json"}
data: {
"db_type": "textdb"
"db_type": "textdb" // 可选若不提供则删除默认集合 ragdb
}
response:
- Success: HTTP 200, {"status": "success", "collection_name": "ragdb_textdb", "message": "集合 ragdb_textdb 删除成功"}
- Error: HTTP 400, {"status": "error", "collection_name": "ragdb_textdb", "message": "<error message>"}
- Success: HTTP 200, {"status": "success", "collection_name": "ragdb" or "ragdb_textdb", "message": "集合 ragdb 或 ragdb_textdb 删除成功"}
- 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:
path: /v1/insertfile
method: POST
headers: {"Content-Type": "application/json"}
data: {
"file_path": "/path/to/file.txt",
"userid": "user1",
"db_type": "textdb"
"file_path": "/path/to/file.txt", // 必填文件路径
"userid": "user123", // 必填用户 ID
"db_type": "textdb", // 可选若不提供则使用默认集合 ragdb
"knowledge_base_id": "kb123" // 必填知识库 ID
}
response:
- Success: HTTP 200, {"status": "success", "document_id": "<uuid>", "collection_name": "ragdb_textdb", "message": "文件 /path/to/file.txt 成功嵌入并处理三元组"}
- Error: HTTP 400, {"status": "error", "document_id": "<uuid>", "collection_name": "ragdb_textdb", "message": "<error message>"}
- Success: HTTP 200, {"status": "success", "document_id": "<uuid>", "collection_name": "ragdb" or "ragdb_textdb", "message": "文件 <file_path> 成功嵌入并处理三元组", "status_code": 200}
- 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
method: POST
headers: {"Content-Type": "application/json"}
data: {
"db_type": "textdb",
"userid": "user1",
"filename": "test.txt"
"userid": "user123", // 必填用户 ID
"filename": "file.txt", // 必填文件名
"db_type": "textdb", // 可选若不提供则使用默认集合 ragdb
"knowledge_base_id": "kb123" // 必填知识库 ID
}
response:
- Success: HTTP 200, {"status": "success", "collection_name": "ragdb_textdb", "message": "成功删除 X 条记录userid=user1, filename=test.txt"}
- Error: HTTP 400, {"status": "error", "collection_name": "ragdb_textdb", "message": "<error message>"}
- 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}
- 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:
path: /v1/fusedsearchquery
@ -70,7 +73,7 @@ headers: {"Content-Type": "application/json"}
data: {
"query": "苹果公司在北京开设新店",
"userid": "user1",
"db_type": "textdb",
"db_type": "textdb", // 可选若不提供则使用默认集合 ragdb
"file_paths": ["/path/to/file.txt"],
"limit": 5,
"offset": 0,
@ -103,7 +106,7 @@ headers: {"Content-Type": "application/json"}
data: {
"query": "知识图谱的知识融合是什么?",
"userid": "user1",
"db_type": "textdb",
"db_type": "textdb", // 可选若不提供则使用默认集合 ragdb
"file_paths": ["/path/to/file.txt"],
"limit": 5,
"offset": 0,
@ -165,6 +168,21 @@ response:
path: /v1/docs
method: GET
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():
@ -173,6 +191,7 @@ def init():
rf.register('deletecollection', delete_collection)
rf.register('insertfile', insert_file)
rf.register('deletefile', delete_file)
rf.register('deleteknowledgebase', delete_knowledge_base)
rf.register('fusedsearchquery', fused_search_query)
rf.register('searchquery', search_query)
rf.register('listuserfiles', list_user_files)
@ -192,22 +211,17 @@ async def create_collection(request, params_kw, *params, **kw):
debug(f'{params_kw=}')
se = ServerEnv()
engine = se.engine
db_type = params_kw.get('db_type')
if db_type is None:
debug(f'db_type 未提供')
return web.json_response({
"status": "error",
"message": "db_type 参数未提供"
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
db_type = params_kw.get('db_type', '')
collection_name = "ragdb" if not db_type else f"ragdb_{db_type}"
try:
result = await engine.handle_connection("create_collection", {"db_type": db_type})
debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e:
debug(f'创建集合失败: {str(e)}')
error(f'创建集合失败: {str(e)}')
return web.json_response({
"status": "error",
"collection_name": f"ragdb_{db_type}",
"collection_name": collection_name,
"message": str(e)
}, 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=}')
se = ServerEnv()
engine = se.engine
db_type = params_kw.get('db_type')
if db_type is None:
debug(f'db_type 未提供')
return web.json_response({
"status": "error",
"message": "db_type 参数未提供"
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
db_type = params_kw.get('db_type', '')
collection_name = "ragdb" if not db_type else f"ragdb_{db_type}"
try:
result = await engine.handle_connection("delete_collection", {"db_type": db_type})
debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e:
debug(f'删除集合失败: {str(e)}')
error(f'删除集合失败: {str(e)}')
return web.json_response({
"status": "error",
"collection_name": f"ragdb_{db_type}",
"collection_name": collection_name,
"message": str(e)
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
async def insert_file(request, params_kw, *params, **kw):
debug(f'{params_kw=}')
debug(f'Received params: {params_kw=}')
se = ServerEnv()
engine = se.engine
file_path = params_kw.get('file_path')
userid = params_kw.get('userid')
db_type = params_kw.get('db_type')
if not all([file_path, userid, db_type]):
debug(f'file_path, userid 或 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)
file_path = params_kw.get('file_path', '')
userid = params_kw.get('userid', '')
db_type = params_kw.get('db_type', '')
knowledge_base_id = params_kw.get('knowledge_base_id', '')
collection_name = "ragdb" if not db_type else f"ragdb_{db_type}"
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", {
"file_path": file_path,
"userid": userid,
"db_type": db_type
"db_type": db_type,
"knowledge_base_id": knowledge_base_id
})
debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
debug(f'Insert result: {result=}')
# 根据 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:
debug(f'插入文件失败: {str(e)}')
error(f'插入文件失败: {str(e)}')
return web.json_response({
"status": "error",
"collection_name": collection_name,
"document_id": "",
"collection_name": f"ragdb_{db_type}",
"message": str(e)
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
async def delete_file(request, params_kw, *params, **kw):
debug(f'{params_kw=}')
debug(f'Received delete_file params: {params_kw=}')
se = ServerEnv()
engine = se.engine
db_type = params_kw.get('db_type')
userid = params_kw.get('userid')
filename = params_kw.get('filename')
if not all([db_type, userid, filename]):
debug(f'db_type, userid 或 filename 未提供')
return web.json_response({
"status": "error",
"message": "db_type, userid 或 filename 未提供"
}, dumps=lambda obj: json.dumps(obj, ensure_ascii=False), status=400)
userid = params_kw.get('userid', '')
filename = params_kw.get('filename', '')
db_type = params_kw.get('db_type', '')
knowledge_base_id = params_kw.get('knowledge_base_id', '')
collection_name = "ragdb" if not db_type else f"ragdb_{db_type}"
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", {
"db_type": db_type,
"userid": userid,
"filename": filename
"filename": filename,
"db_type": db_type,
"knowledge_base_id": knowledge_base_id
})
debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
debug(f'Delete 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:
debug(f'删除文件失败: {str(e)}')
error(f'删除文件失败: {str(e)}')
return web.json_response({
"status": "error",
"collection_name": f"ragdb_{db_type}",
"message": str(e)
"collection_name": collection_name,
"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)
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
query = params_kw.get('query')
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')
limit = params_kw.get('limit', 5)
offset = params_kw.get('offset', 0)
use_rerank = params_kw.get('use_rerank', True)
if not all([query, userid, db_type, file_paths]):
debug(f'query, userid, db_type 或 file_paths 未提供')
if not all([query, userid, file_paths]):
debug(f'query, userid 或 file_paths 未提供')
return web.json_response({
"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)
try:
result = await engine.handle_connection("fused_search", {
@ -323,7 +383,7 @@ async def fused_search_query(request, params_kw, *params, **kw):
debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e:
debug(f'融合搜索失败: {str(e)}')
error(f'融合搜索失败: {str(e)}')
return web.json_response({
"status": "error",
"message": str(e)
@ -335,16 +395,16 @@ async def search_query(request, params_kw, *params, **kw):
engine = se.engine
query = params_kw.get('query')
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')
limit = params_kw.get('limit', 5)
offset = params_kw.get('offset', 0)
use_rerank = params_kw.get('use_rerank', True)
if not all([query, userid, db_type, file_paths]):
debug(f'query, userid, db_type 或 file_paths 未提供')
if not all([query, userid, file_paths]):
debug(f'query, userid 或 file_paths 未提供')
return web.json_response({
"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)
try:
result = await engine.handle_connection("search_query", {
@ -359,7 +419,7 @@ async def search_query(request, params_kw, *params, **kw):
debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e:
debug(f'纯向量搜索失败: {str(e)}')
error(f'纯向量搜索失败: {str(e)}')
return web.json_response({
"status": "error",
"message": str(e)
@ -383,7 +443,7 @@ async def list_user_files(request, params_kw, *params, **kw):
debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e:
debug(f'查询用户文件列表失败: {str(e)}')
error(f'查询用户文件列表失败: {str(e)}')
return web.json_response({
"status": "error",
"message": str(e)
@ -406,7 +466,7 @@ async def handle_connection(request, params_kw, *params, **kw):
debug(f'{result=}')
return web.json_response(result, dumps=lambda obj: json.dumps(obj, ensure_ascii=False))
except Exception as e:
debug(f'处理连接操作失败: {str(e)}')
error(f'处理连接操作失败: {str(e)}')
return web.json_response({
"status": "error",
"message": str(e)
@ -418,7 +478,7 @@ def main():
parser.add_argument('-p', '--port', default='8888')
parser.add_argument('connection_path')
args = parser.parse_args()
logger.debug(f"Arguments: {args}")
debug(f"Arguments: {args}")
Klass = get_connection_class(args.connection_path)
se = ServerEnv()
se.engine = Klass()

File diff suppressed because it is too large Load Diff