mirror of
https://github.com/RYDE-WORK/Langchain-Chatchat.git
synced 2026-01-19 21:37:20 +08:00
a5b203170b
- 新功能
- pypi 包新增 chatchat-kb 命令脚本，对应 init_database.py 功能
- 开发者
- _model_config.py 中默认包含 xinference 配置项
- 所有涉及向量库的操作，前置检查当前 Embed 模型是否可用
- /knowledge_base/create_knowledge_base 接口增加 kb_info 参数
- /knowledge_base/list_files 接口返回所有数据库字段，而非文件名称列表
- 修正 xinference 模型管理脚本
67 lines
2.7 KiB
Python
67 lines
2.7 KiB
Python
import urllib
|
||
from chatchat.server.utils import BaseResponse, ListResponse
|
||
from chatchat.server.knowledge_base.utils import validate_kb_name
|
||
from chatchat.server.knowledge_base.kb_service.base import KBServiceFactory
|
||
from chatchat.server.db.repository.knowledge_base_repository import list_kbs_from_db
|
||
from chatchat.configs import DEFAULT_EMBEDDING_MODEL, logger, log_verbose
|
||
from fastapi import Body
|
||
|
||
|
||
def list_kbs():
|
||
# Get List of Knowledge Base
|
||
return ListResponse(data=list_kbs_from_db())
|
||
|
||
|
||
def create_kb(knowledge_base_name: str = Body(..., examples=["samples"]),
|
||
vector_store_type: str = Body("faiss"),
|
||
kb_info: str = Body("", description="知识库内容简介，用于Agent选择知识库。"),
|
||
embed_model: str = Body(DEFAULT_EMBEDDING_MODEL),
|
||
) -> BaseResponse:
|
||
# Create selected knowledge base
|
||
if not validate_kb_name(knowledge_base_name):
|
||
return BaseResponse(code=403, msg="Don't attack me")
|
||
if knowledge_base_name is None or knowledge_base_name.strip() == "":
|
||
return BaseResponse(code=404, msg="知识库名称不能为空，请重新填写知识库名称")
|
||
|
||
kb = KBServiceFactory.get_service_by_name(knowledge_base_name)
|
||
if kb is not None:
|
||
return BaseResponse(code=404, msg=f"已存在同名知识库 {knowledge_base_name}")
|
||
|
||
kb = KBServiceFactory.get_service(knowledge_base_name, vector_store_type, embed_model, kb_info=kb_info)
|
||
try:
|
||
kb.create_kb()
|
||
except Exception as e:
|
||
msg = f"创建知识库出错： {e}"
|
||
logger.error(f'{e.__class__.__name__}: {msg}',
|
||
exc_info=e if log_verbose else None)
|
||
return BaseResponse(code=500, msg=msg)
|
||
|
||
return BaseResponse(code=200, msg=f"已新增知识库 {knowledge_base_name}")
|
||
|
||
|
||
def delete_kb(
|
||
knowledge_base_name: str = Body(..., examples=["samples"])
|
||
) -> BaseResponse:
|
||
# Delete selected knowledge base
|
||
if not validate_kb_name(knowledge_base_name):
|
||
return BaseResponse(code=403, msg="Don't attack me")
|
||
knowledge_base_name = urllib.parse.unquote(knowledge_base_name)
|
||
|
||
kb = KBServiceFactory.get_service_by_name(knowledge_base_name)
|
||
|
||
if kb is None:
|
||
return BaseResponse(code=404, msg=f"未找到知识库 {knowledge_base_name}")
|
||
|
||
try:
|
||
status = kb.clear_vs()
|
||
status = kb.drop_kb()
|
||
if status:
|
||
return BaseResponse(code=200, msg=f"成功删除知识库 {knowledge_base_name}")
|
||
except Exception as e:
|
||
msg = f"删除知识库时出现意外： {e}"
|
||
logger.error(f'{e.__class__.__name__}: {msg}',
|
||
exc_info=e if log_verbose else None)
|
||
return BaseResponse(code=500, msg=msg)
|
||
|
||
return BaseResponse(code=500, msg=f"删除知识库失败 {knowledge_base_name}")
|