"""知识分类(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