kplam
51dc8dc4f6
- FastAPI + Qdrant + Redis + Ollama 技术栈 - L1→L2→L3→chunk 四层分层检索（dense + sparse RRF 融合） - 文档三级总结与 2.5 级回退 - query 解析路由与分类 - /admin 管理页面
69 lines
2.9 KiB
Python
69 lines
2.9 KiB
Python
"""知识分类（taxonomy）相关的数据模型与加载逻辑"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import structlog
|
|
from pydantic import BaseModel, Field
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
# 未分类常量：分类置信度不足或无法归类时使用
|
|
UNCATEGORIZED = "uncategorized"
|
|
|
|
|
|
class TaxonomyCategory(BaseModel):
|
|
"""知识分类类目定义"""
|
|
|
|
name: str = Field(description="类目名称，全局唯一")
|
|
description: str = Field(default="", description="类目描述，用于辅助分类判断")
|
|
|
|
|
|
class CategoryResult(BaseModel):
|
|
"""文档/查询的分类结果"""
|
|
|
|
main_category: str = Field(description="主类目名称")
|
|
tags: list[str] = Field(default_factory=list, description="附加标签列表")
|
|
confidence: float = Field(ge=0, le=1, description="分类置信度，范围 [0, 1]")
|
|
|
|
|
|
def _default_taxonomy() -> list[TaxonomyCategory]:
|
|
"""内置默认类目集（通用企业知识库场景）"""
|
|
return [
|
|
TaxonomyCategory(name="技术文档", description="架构设计、API 文档、开发规范、运维手册等技术资料"),
|
|
TaxonomyCategory(name="产品手册", description="产品功能介绍、使用说明、版本发布说明"),
|
|
TaxonomyCategory(name="运营规范", description="运营流程、活动方案、内容规范、客服话术"),
|
|
TaxonomyCategory(name="财务行政", description="财务制度、报销流程、行政通知、办公管理"),
|
|
TaxonomyCategory(name="市场资料", description="市场分析、竞品调研、营销素材、品牌规范"),
|
|
TaxonomyCategory(name="人事制度", description="招聘、考勤、绩效、培训、员工手册等 HR 制度"),
|
|
TaxonomyCategory(name="法律法规", description="合同模板、合规要求、法律条文、知识产权"),
|
|
TaxonomyCategory(name=UNCATEGORIZED, description="无法归入其他类目的文档"),
|
|
]
|
|
|
|
|
|
def load_taxonomy(path: str = "") -> list[TaxonomyCategory]:
|
|
"""加载 taxonomy 类目集
|
|
|
|
path 为空时使用内置默认类目集；非空时从 JSON 文件加载，
|
|
文件格式为 [{"name": ..., "description": ...}]。
|
|
校验类目 name 唯一；若缺少 uncategorized 类目则自动追加。
|
|
"""
|
|
if not path:
|
|
return _default_taxonomy()
|
|
|
|
raw = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
categories = [TaxonomyCategory.model_validate(item) for item in raw]
|
|
|
|
# 校验 name 唯一
|
|
names = [c.name for c in categories]
|
|
if len(names) != len(set(names)):
|
|
duplicates = sorted({n for n in names if names.count(n) > 1})
|
|
raise ValueError(f"taxonomy 类目 name 重复: {duplicates}")
|
|
|
|
# 必含 uncategorized，缺失则自动追加
|
|
if UNCATEGORIZED not in names:
|
|
logger.warning("taxonomy 缺少 uncategorized 类目，已自动追加", path=path)
|
|
categories.append(TaxonomyCategory(name=UNCATEGORIZED, description="无法归入其他类目的文档"))
|
|
|
|
return categories
|