"""知识库：专家系统风格的规则与知识表示（零依赖，纯标准库）。 设计原则（对齐《可行性调研与落地实现路线报告》第八章"专家系统内核"）： - 领域知识显式化：写在规则文件里（config/knowledge/.yaml），不藏在模型参数中 - 确定性：规则匹配 = 子串包含（大小写不敏感），同输入同输出 - 可解释：每次命中都记录规则 id，形成推理轨迹 - 最小参数：L0 模式零模型参数，规则即知识 规则文件格式（YAML；若 pyyaml 不可用，可提供同名 .json）： domain: code rules: - id: code-sort priority: 90 # 越大越先触发 patterns: ["排序", "sort"] # 任一子串命中即触发 template: code-implement # 可选：Planner 任务模板 id output: | # 可选：输出模板（{query} 等占位符） （规则输出）... facts: # 领域事实表（Judge 校验 / retrieve 执行器用） - id: legal-nc keywords: ["竞业"] statement: "竞业限制期限不得超过二年" 任务模板（config/knowledge/tasks.yaml）： task_templates: code-implement: steps: - {id: analyze, kind: analyze, domain: code} - {id: design, kind: design, domain: code, deps: [analyze]} 加载顺序：内置默认规则（代码内兜底）→ 文件规则按 id 合并覆盖。 """ from __future__ import annotations import json from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional DEFAULT_RULES_DIR = Path(__file__).resolve().parent.parent / "config" / "knowledge" @dataclass class Rule: """一条领域规则。""" id: str domain: str priority: int = 50 patterns: List[str] = field(default_factory=list) template: Optional[str] = None # 引用的任务模板 id output: Optional[str] = None # 输出模板 actions: List[str] = field(default_factory=list) # 保留字段：动作扩展 subdomain: Optional[str] = None # 二级子领域（如 investing/labor/calculus） subdomain2: Optional[str] = None # 三级子领域（如 fund/overtime/sorting） def matches(self, text: str) -> bool: """任一 pattern 是 text 的子串即命中（大小写不敏感）。""" if not self.patterns: return False q = text.lower() return any(p.lower() in q for p in self.patterns) # --------------------------------------------------------------- # 三级子领域映射（rule_id -> subdomain2） # 集中维护：新增规则时在此加一行即可完成三级细化标注 # --------------------------------------------------------------- SUBDOMAIN2_MAP: Dict[str, str] = { # ---- code ---- "code-sort": "sorting", "code-debug": "error-analysis", "code-algorithm": "algorithm-general", "code-refactor": "code-quality", "code-database": "sql", "code-explain": "code-reading", "code-test": "unit-test", "code-web": "web-dev", "code-implement-general": "implementation", "code-git-knowledge": "git", "code-docker-knowledge": "container", "code-python-knowledge": "python-env", # ---- math ---- "math-equation": "equation", "math-calculus": "calculus", "math-algebra": "algebra", "math-geometry": "geometry", "math-proof": "proof", "math-probability": "probability", "math-number-theory": "number-theory", "math-trigonometry": "trigonometry", "math-optimization": "optimization", "math-general": "math-general", # ---- legal ---- "legal-contract": "contract", "legal-labor": "labor", "legal-ip": "intellectual-property", "legal-housing": "housing", "legal-marriage": "family-law", "legal-tax": "tax", "legal-consumer": "consumer-rights", "legal-litigation": "litigation", "legal-compliance": "compliance", "legal-general": "legal-general", # ---- medical ---- "medical-hypertension": "hypertension", "medical-drug": "medication", "medical-common": "common-illness", "medical-chronic": "chronic-disease", "medical-digestive": "digestive", "medical-nutrition": "nutrition", "medical-mental": "mental-health", "medical-firstaid": "first-aid", "medical-pediatrics": "pediatrics", "medical-general": "medical-general", # ---- finance ---- "finance-investing": "investing", "finance-saving": "saving", "finance-loan": "loan", "finance-insurance": "insurance", "finance-credit-card": "credit", "finance-personal-budget": "budgeting", "finance-general": "finance-general", # ---- life ---- "life-food": "cooking", "life-travel": "travel", "life-home": "home", "life-pet": "pet", "life-fitness": "fitness", "life-weather": "weather", "life-general": "life-general", # ---- education ---- "edu-study-method": "study-method", "edu-exam": "exam", "edu-language": "language", "edu-course": "course", "edu-career": "career", "edu-general": "education-general", # ---- general ---- "general-explain": "explain", "general-writing": "writing", "general-compare": "compare", "general-translate": "translate", "general-knowledge": "explain", } # --------------------------------------------------------------- # 二级子领域映射（rule_id -> subdomain） # 三级 subdomain2 的父级类别；与 SUBDOMAIN2_MAP 按 rule_id 对齐维护。 # --------------------------------------------------------------- SUBDOMAIN_MAP: Dict[str, str] = { # ---- code ---- "code-sort": "algorithm", "code-debug": "debugging", "code-algorithm": "algorithm", "code-refactor": "quality", "code-database": "data", "code-explain": "reading", "code-test": "quality", "code-web": "web", "code-implement-general": "implementation", "code-git-knowledge": "tooling", "code-docker-knowledge": "tooling", "code-python-knowledge": "tooling", # ---- math ---- "math-equation": "algebra", "math-calculus": "analysis", "math-algebra": "algebra", "math-geometry": "geometry", "math-proof": "proof", "math-probability": "probability", "math-number-theory": "number-theory", "math-trigonometry": "trigonometry", "math-optimization": "optimization", "math-general": "general", # ---- legal ---- "legal-contract": "contract", "legal-labor": "labor", "legal-ip": "ip", "legal-housing": "civil", "legal-marriage": "civil", "legal-tax": "tax", "legal-consumer": "consumer", "legal-litigation": "procedure", "legal-compliance": "compliance", "legal-general": "general", # ---- medical ---- "medical-hypertension": "chronic", "medical-drug": "medication", "medical-common": "common", "medical-chronic": "chronic", "medical-digestive": "common", "medical-nutrition": "nutrition", "medical-mental": "mental", "medical-firstaid": "emergency", "medical-pediatrics": "pediatrics", "medical-general": "general", # ---- finance ---- "finance-investing": "investing", "finance-saving": "personal-finance", "finance-loan": "credit", "finance-insurance": "insurance", "finance-credit-card": "credit", "finance-personal-budget": "personal-finance", "finance-general": "general", # ---- life ---- "life-food": "daily", "life-travel": "daily", "life-home": "daily", "life-pet": "daily", "life-fitness": "health", "life-weather": "daily", "life-general": "general", # ---- education ---- "edu-study-method": "learning", "edu-exam": "learning", "edu-language": "language", "edu-course": "learning", "edu-career": "development", "edu-general": "general", # ---- general ---- "general-explain": "explanation", "general-writing": "writing", "general-compare": "analysis", "general-translate": "language", "general-knowledge": "explanation", } # --------------------------------------------------------------- # 内置默认规则（兜底：即使规则文件缺失/损坏，系统仍可运行） # --------------------------------------------------------------- BUILTIN_RULES: List[Dict[str, Any]] = [ # ---- code ---- {"id": "code-sort", "domain": "code", "priority": 90, "patterns": ["排序", "快速排序", "排序算法", "sort", "quicksort"], "template": "code-implement"}, {"id": "code-debug", "domain": "code", "priority": 85, "patterns": ["报错", "错误", "调试", "bug", "debug", "typeerror", "异常", "报 TypeError"], "template": "code-debug"}, {"id": "code-implement-general", "domain": "code", "priority": 50, "patterns": ["实现", "编写", "写一个", "函数", "代码", "编程", "用 python", "用 java", "用 javascript", "sql", "接口", "算法"], "template": "code-implement"}, # ---- math ---- {"id": "math-equation", "domain": "math", "priority": 90, "patterns": ["方程", "求解", "求根", "solve", "equation", "解方程"], "template": "math-solve"}, {"id": "math-calculus", "domain": "math", "priority": 85, "patterns": ["积分", "导数", "微积分", "求导", "integral", "derivative", "∫"], "template": "math-solve"}, {"id": "math-general", "domain": "math", "priority": 50, "patterns": ["数学", "证明", "定理", "概率", "统计", "计算", "等于", "math", "不等式"], "template": "math-solve"}, # ---- legal ---- {"id": "legal-contract", "domain": "legal", "priority": 90, "patterns": ["合同", "条款", "违约", "离职", "竞业", "劳动", "contract", "clause", "赔偿"], "template": "legal-advice"}, {"id": "legal-ip", "domain": "legal", "priority": 85, "patterns": ["专利", "版权", "商标", "知识产权", "patent", "copyright", "trademark"], "template": "legal-advice"}, {"id": "legal-general", "domain": "legal", "priority": 50, "patterns": ["法律", "合规", "诉讼", "仲裁", "法条", "law", "legal", "法规"], "template": "legal-advice"}, # ---- medical ---- {"id": "medical-hypertension", "domain": "medical", "priority": 90, "patterns": ["高血压", "hypertension", "血压"], "template": "medical-advice"}, {"id": "medical-drug", "domain": "medical", "priority": 85, "patterns": ["药物", "吃药", "剂量", "副作用", "退烧药", "降压药", "dosage", "prescription"], "template": "medical-advice"}, {"id": "medical-general", "domain": "medical", "priority": 50, "patterns": ["医疗", "症状", "诊断", "治疗", "感冒", "发烧", "糖尿病", "医生", "患者", "体检", "疫苗", "medical", "symptom", "disease"], "template": "medical-advice"}, # ---- finance ---- {"id": "finance-investing", "domain": "finance", "priority": 90, "patterns": ["基金", "定投", "收益率", "股票", "投资", "炒股", "证券", "invest", "stock"]}, {"id": "finance-saving", "domain": "finance", "priority": 85, "patterns": ["存款", "储蓄", "利息", "零钱通", "余额宝", "saving"]}, {"id": "finance-loan", "domain": "finance", "priority": 80, "patterns": ["贷款", "房贷", "借款", "按揭", "loan"]}, {"id": "finance-insurance", "domain": "finance", "priority": 75, "patterns": ["保险", "理赔", "保单", "投保", "insurance"]}, {"id": "finance-credit-card", "domain": "finance", "priority": 70, "patterns": ["信用卡", "花呗", "白条", "credit card"]}, {"id": "finance-personal-budget", "domain": "finance", "priority": 60, "patterns": ["预算", "记账", "开销", "省钱", "budget"]}, {"id": "finance-general", "domain": "finance", "priority": 50, "patterns": ["金融", "财务", "外汇", "汇率", "finance"]}, # ---- life ---- {"id": "life-food", "domain": "life", "priority": 90, "patterns": ["做饭", "做菜", "菜谱", "食谱", "烹饪", "cooking"]}, {"id": "life-travel", "domain": "life", "priority": 85, "patterns": ["旅游", "旅行", "攻略", "景点", "签证", "travel"]}, {"id": "life-home", "domain": "life", "priority": 80, "patterns": ["装修", "租房", "家电", "清洁", "搬家", "home"]}, {"id": "life-pet", "domain": "life", "priority": 75, "patterns": ["宠物", "养猫", "养狗", "撸猫", "pet"]}, {"id": "life-fitness", "domain": "life", "priority": 70, "patterns": ["健身", "减肥", "跑步", "锻炼", "fitness"]}, {"id": "life-weather", "domain": "life", "priority": 65, "patterns": ["天气", "下雨", "台风", "降温", "weather"]}, {"id": "life-general", "domain": "life", "priority": 50, "patterns": ["生活", "日常", "家居", "life"]}, # ---- education ---- {"id": "edu-study-method", "domain": "education", "priority": 90, "patterns": ["学习方法", "记忆", "做笔记", "笔记法", "专注力"]}, {"id": "edu-exam", "domain": "education", "priority": 85, "patterns": ["考试", "考研", "复习", "真题", "四六级", "exam"]}, {"id": "edu-language", "domain": "education", "priority": 80, "patterns": ["英语", "单词", "口语", "语法", "english"]}, {"id": "edu-course", "domain": "education", "priority": 75, "patterns": ["课程", "网课", "慕课", "选修", "course"]}, {"id": "edu-career", "domain": "education", "priority": 70, "patterns": ["职业规划", "求职", "面试", "简历", "校招", "career"]}, {"id": "edu-general", "domain": "education", "priority": 50, "patterns": ["教育", "大学", "专业选择", "education"]}, # ---- general ---- {"id": "general-explain", "domain": "general", "priority": 30, "patterns": ["总结", "介绍", "解释", "为什么", "优缺点", "是什么", "翻译", "邮件", "summarize", "explain", "what is", "写一封"], "template": "general-explain"}, ] # 内置默认任务模板（兜底） BUILTIN_TASKS: Dict[str, Dict[str, Any]] = { "code-implement": {"steps": [ {"id": "analyze", "kind": "analyze", "domain": "code", "desc": "需求与约束分析"}, {"id": "design", "kind": "design", "domain": "code", "deps": ["analyze"], "desc": "算法与数据结构设计"}, {"id": "implement", "kind": "implement", "domain": "code", "deps": ["design"], "desc": "实现代码"}, {"id": "verify", "kind": "verify", "domain": "code", "deps": ["implement"], "desc": "自测校验"}, ]}, "code-debug": {"steps": [ {"id": "analyze", "kind": "analyze", "domain": "code", "desc": "错误现象与复现分析"}, {"id": "diagnose", "kind": "diagnose", "domain": "code", "deps": ["analyze"], "desc": "定位错误根因"}, {"id": "fix", "kind": "fix", "domain": "code", "deps": ["diagnose"], "desc": "给出修复方案"}, {"id": "verify", "kind": "verify", "domain": "code", "deps": ["fix"], "desc": "修复后验证"}, ]}, "math-solve": {"steps": [ {"id": "conditions", "kind": "analyze", "domain": "math", "desc": "明确已知条件与目标"}, {"id": "solve", "kind": "solve", "domain": "math", "deps": ["conditions"], "desc": "选择方法并求解"}, {"id": "verify", "kind": "verify", "domain": "math", "deps": ["solve"], "desc": "检查边界与验证"}, ]}, "legal-advice": {"steps": [ {"id": "facts", "kind": "analyze", "domain": "legal", "desc": "梳理事实与法律问题"}, {"id": "retrieve", "kind": "retrieve", "domain": "legal", "deps": ["facts"], "desc": "检索适用法规"}, {"id": "conclude", "kind": "conclude", "domain": "legal", "deps": ["retrieve"], "desc": "给出法律意见"}, {"id": "disclaimer", "kind": "disclaimer", "domain": "legal", "deps": ["conclude"], "desc": "免责提示"}, ]}, "medical-advice": {"steps": [ {"id": "symptoms", "kind": "analyze", "domain": "medical", "desc": "梳理症状与背景"}, {"id": "advise", "kind": "advise", "domain": "medical", "deps": ["symptoms"], "desc": "给出一般建议"}, {"id": "warning", "kind": "disclaimer", "domain": "medical", "deps": ["advise"], "desc": "就医警示"}, ]}, "general-explain": {"steps": [ {"id": "outline", "kind": "analyze", "domain": "general", "desc": "梳理主题要点"}, {"id": "explain", "kind": "explain", "domain": "general", "deps": ["outline"], "desc": "展开解释"}, {"id": "conclude", "kind": "conclude", "domain": "general", "deps": ["explain"], "desc": "总结"}, ]}, } # 内置默认事实表（兜底） BUILTIN_FACTS: Dict[str, List[Dict[str, Any]]] = { "legal": [ {"id": "legal-noncompete", "keywords": ["竞业", "离职", "同业"], "statement": "竞业限制期限不得超过二年，且用人单位应在限制期内按月给予经济补偿"}, {"id": "legal-renew-compensation", "keywords": ["不续签", "经济补偿", "劳动合同"], "statement": "劳动合同期满用人单位不续签的，通常应支付经济补偿（每满一年一个月工资）"}, ], "medical": [ {"id": "medical-hypertension-diet", "keywords": ["高血压", "饮食"], "statement": "高血压患者应低盐低脂饮食、控制体重、规律运动、戒烟限酒，并在医生指导下用药"}, {"id": "medical-fever-drug", "keywords": ["发烧", "退烧"], "statement": "体温超过 38.5℃ 可在药师指导下使用退烧药；持续发热或出现严重症状应及时就医"}, ], } def _try_load_yaml(path: Path) -> Optional[Dict[str, Any]]: try: import yaml # type: ignore except ImportError: return None try: with open(path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) return data if isinstance(data, dict) else None except Exception: return None def _try_load_json(path: Path) -> Optional[Dict[str, Any]]: json_path = path.with_suffix(".json") if not json_path.exists(): return None try: with open(json_path, "r", encoding="utf-8") as f: data = json.load(f) return data if isinstance(data, dict) else None except Exception: return None class KnowledgeBase: """知识库：加载规则文件，提供规则匹配、任务模板、事实表查询。""" def __init__(self, rules_dir: Optional[str | Path] = None): self.rules_dir = Path(rules_dir) if rules_dir else DEFAULT_RULES_DIR self._rules: Dict[str, Rule] = {} self._tasks: Dict[str, Dict[str, Any]] = {} self._facts: Dict[str, List[Dict[str, Any]]] = {} self.load() # ---- 加载 ---- def load(self) -> None: """内置默认 + 规则文件合并（文件规则按 id 覆盖内置）。""" self._rules = {} self._tasks = dict(BUILTIN_TASKS) for item in BUILTIN_RULES: self._register_rule(item) self._facts = {d: [dict(f) for f in facts] for d, facts in BUILTIN_FACTS.items()} if self.rules_dir.is_dir(): for f in sorted(self.rules_dir.glob("*.yaml")): data = _try_load_yaml(f) if data is not None: self._load_file_data(f, data) for f in sorted(self.rules_dir.glob("*.json")): if f.name not in {p.name for p in self.rules_dir.glob("*.yaml")}: data = _try_load_json(f) if data is not None: self._load_file_data(f, data) def _load_file_data(self, path: Path, data: Dict[str, Any]) -> None: name = path.stem if name == "tasks": for tid, tpl in (data.get("task_templates") or {}).items(): if isinstance(tpl, dict) and isinstance(tpl.get("steps"), list): self._tasks[tid] = tpl return domain = data.get("domain", name) for item in data.get("rules") or []: if isinstance(item, dict) and item.get("id"): self._register_rule({**item, "domain": domain}) for fact in data.get("facts") or []: if isinstance(fact, dict) and fact.get("id"): self._facts.setdefault(domain, []).append(fact) def _register_rule(self, item: Dict[str, Any]) -> None: rule = Rule( id=str(item["id"]), domain=str(item.get("domain", "general")), priority=int(item.get("priority", 50)), patterns=[str(p) for p in item.get("patterns", [])], template=item.get("template"), output=item.get("output"), actions=[str(a) for a in item.get("actions", [])], subdomain=item.get("subdomain") or SUBDOMAIN_MAP.get(str(item["id"])), subdomain2=item.get("subdomain2") or SUBDOMAIN2_MAP.get(str(item["id"])), ) self._rules[rule.id] = rule # ---- 查询 ---- def match(self, text: str, domain: Optional[str] = None) -> List[Rule]: """返回命中的规则，按优先级降序。domain 为空则全领域匹配。""" hits = [] for rule in self._rules.values(): if domain is not None and rule.domain != domain: continue if rule.matches(text): hits.append(rule) hits.sort(key=lambda r: r.priority, reverse=True) return hits def rule(self, rule_id: str) -> Optional[Rule]: return self._rules.get(rule_id) def rules_count(self) -> int: return len(self._rules) def task_template(self, tid: str) -> Optional[Dict[str, Any]]: return self._tasks.get(tid) def task_ids(self) -> List[str]: return sorted(self._tasks.keys()) def facts(self, domain: str) -> List[Dict[str, Any]]: return self._facts.get(domain, []) def domains(self) -> List[str]: return sorted({r.domain for r in self._rules.values()})