"""L2 训练营知识库工具 — 按需加载 + 语义搜索 本文件包含知识库查询工具的实现。修改主提示词时请勿改动此文件。 """ from __future__ import annotations import re from pathlib import Path from common.knowledge_loader import list_knowledge_files, load_knowledge_file # ───────────────────────────── Markdown 清理 ───────────────────────────── def _strip_markdown(text: str) -> str: """将 Markdown 格式转为纯文本，方便微信等平台显示""" # 移除代码块 text = re.sub(r"```[\s\S]*?```", lambda m: m.group(0).strip("`\n"), text) # 移除行内代码反引号 text = re.sub(r"`([^`]+)`", r"\1", text) # 移除加粗/斜体 text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) text = re.sub(r"\*(.+?)\*", r"\1", text) text = re.sub(r"__(.+?)__", r"\1", text) text = re.sub(r"_(.+?)_", r"\1", text) # 移除标题标记，保留文字 text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE) # 移除链接，保留文字 text = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", text) # 移除图片 text = re.sub(r"!\[([^\]]*)\]\([^)]+\)", "", text) # 移除水平线 text = re.sub(r"^---+$", "", text, flags=re.MULTILINE) # 移除引用标记 text = re.sub(r"^>\s?", "", text, flags=re.MULTILINE) # 移除列表标记（- 和 * 开头） text = re.sub(r"^[\-\*]\s+", "· ", text, flags=re.MULTILINE) # 移除编号列表 text = re.sub(r"^\d+\.\s+", "", text, flags=re.MULTILINE) # 清理多余空行（最多保留一个空行） text = re.sub(r"\n{3,}", "\n\n", text) return text.strip() # ───────────────────────────── 配置 ───────────────────────────── _KB_DIR = Path(__file__).parent / "knowledge_base" # 知识库元数据（与文件名一一对应） _KB_METADATA: dict[str, dict] = { "01_课程层次与问题判断.md": { "title": "课程层次与问题判断", "description": "判断299、1280七天训练营、大本营之间的层次关系，以及学员当前问题卡在哪一层。", "keywords": [ "课程层次", "299", "1280", "七天训练营", "启蒙课", "训练场", "大本营", "层次判断", "知道但做不到", "课程信心", "没体感", "后续训练", "回溯", "回溯现象", "未完成事件", "事实法则", "流动法则", "现实风险层", "问题判断", "卡在哪一层", ], }, "02_即时回复表达原则.md": { "title": "即时回复表达原则", "description": "生成有阅读感、人情味、有分寸、可供辅导老师参考和改写的回复。", "keywords": [ "表达", "回复", "参考回复", "怎么回", "怎么说", "语气", "措辞", "风格", "原则", "阅读感", "人情味", "中肯方向", "长故事", "短问题", "群内回复", "私聊回复", "AI味", "课程词", "慎用术语", "话术表达", "成交表达", ], }, "03_常见学员问题回应素材.md": { "title": "常见学员问题回应素材", "description": "妈妈父母、孩子、伴侣、课程失望、知道但做不到、想放弃等常见场景回应素材。", "keywords": [ "常见问题", "学员问题", "回应素材", "妈妈", "父母", "原生家庭", "孩子", "不上学", "躺平", "伴侣", "老公", "出轨", "离婚", "课程没用", "没体感", "没变化", "失望", "知道但做不到", "反应过度", "小事失控", "想放弃", "不想写输出", "不想分享", "群内长篇分享", "私聊长故事", "攻击课程", "攻击老师", "刷屏", "互相影响", "没救了", "别人有反应我没有", "依赖老师", ], }, "04_功课与练习反馈边界.md": { "title": "功课与练习反馈边界", "description": "功课、练习、释放、万能释放法、练习反馈和练习后不适的处理边界。", "keywords": [ "功课", "练习", "打卡", "反馈", "边界", "作业", "释放", "释放不是宣泄", "万能释放法", "万能释放公式", "看见且释放", "没感觉", "不知道释放什么", "做完哭", "做完抖", "做完堵", "头晕", "胸闷", "呼吸不适", "飘", "空", "回不来", "看到画面", "前世", "小时候画面", "旧记忆", "功课回溯", "最小功课", "带练", "暂停练习", ], }, "05_安全边界与现实风险.md": { "title": "安全边界与现实风险", "description": "安全、医疗、心理危机、法律、家暴、未成年人、严重身体不适等最高优先级边界。", "keywords": [ "安全", "风险", "危机", "高风险", "自伤", "自杀", "想死", "不想活", "伤害自己", "伤害他人", "家暴", "人身安全", "未成年人", "孩子安全", "离家出走", "暴力", "报警", "上报", "班主任", "督导", "现实支持", "严重身体不适", "呼吸困难", "胸闷", "精神混乱", "幻觉", "妄想", "解离", "飘", "空", "麻", "回不来", "用药", "停药", "换药", "诊断", "治疗", "医疗", "法律", "财产", "版权", "保密", "意识许可", "课程资料外传", ], }, "06_1280核心术语词典.md": { "title": "1280核心术语词典", "description": "1280核心术语的人话解释、适用场景、禁用场景和老师提醒。", "keywords": [ "术语", "词典", "概念", "定义", "解释", "是什么意思", "看见自己", "回到自己", "看见且释放", "万能释放法", "万能释放公式", "情绪心钥", "抗拒心钥", "信念心钥", "程序", "程序心钥", "人生剧本", "回溯现象", "未完成事件", "事实法则", "流动法则", "派生情绪", "原生情绪", "抓取", "抗拒", "原生数据", "先天数据", "潜意识数据", "代际", "家族数据", "数息法", "身体框架法", "认知透镜", "知道但做不到", "自动反应", "自我的七个层面", "百慕大能量三角形", "场域支持", "借光而行", "借事修心", "生命数据化", "输出", "支点", "格物致知", "知行合一", "良知", ], }, "07_大本营承接与成交话术.md": { "title": "大本营承接与成交话术", "description": "大本营承接时机、成交判断、课程信心修复、异议处理和老师成交提醒。", "keywords": [ "大本营", "承接", "成交", "异议", "销售", "推课", "后续课程", "继续学", "持续训练", "持续场域", "场域支持", "借光而行", "课程信心", "课程没用", "没体感", "怕没效果", "知道但做不到", "怕结束后回到原样", "没时间", "没钱", "价格", "权益", "优惠", "名额", "截止时间", "报名", "退款", "家人不同意", "觉得被销售", "不想继续", "不适合承接", "轻轻埋种子", "明确承接", "高风险停止成交", ], }, } _ROUTE_RULES: list[dict] = [ { "filename": "05_安全边界与现实风险.md", "boost": 5.0, "keywords": [ "自伤", "自残", "自杀", "想死", "不想活", "活不下去", "伤害自己", "伤害他人", "杀了", "家暴", "打我", "被打", "报警", "人身安全", "孩子安全", "未成年人", "离家出走", "呼吸困难", "胸闷", "喘不过气", "头晕", "严重身体不适", "幻觉", "妄想", "精神混乱", "解离", "回不来", "停药", "换药", "用药", "诊断", "治疗", "医院", "医生", "法律", "财产", "版权", "保密", "课程资料", "外传", "意识许可", ], }, { "filename": "04_功课与练习反馈边界.md", "boost": 3.5, "keywords": [ "功课", "作业", "练习", "带练", "打卡", "释放", "万能释放", "看见且释放", "不知道释放什么", "没感觉", "做完哭", "做完抖", "做完堵", "练完", "头晕", "胸闷", "呼吸不适", "飘", "空", "麻", "回不来", "看到画面", "前世", "小时候画面", "旧记忆", "暂停练习", "最小功课", ], }, { "filename": "07_大本营承接与成交话术.md", "boost": 3.5, "keywords": [ "大本营", "承接", "成交", "推课", "销售", "后续课程", "继续学", "继续练", "持续训练", "报名", "价格", "多少钱", "权益", "优惠", "名额", "截止", "退款", "没时间", "没钱", "怕没效果", "家人不同意", "觉得被销售", "不想继续", "怕结束后回到原样", "适不适合推", ], }, { "filename": "06_1280核心术语词典.md", "boost": 3.0, "keywords": [ "是什么意思", "怎么解释", "术语", "概念", "定义", "看见自己", "回到自己", "情绪心钥", "抗拒心钥", "信念心钥", "程序心钥", "人生剧本", "回溯现象", "未完成事件", "事实法则", "流动法则", "派生情绪", "原生情绪", "原生数据", "先天数据", "潜意识数据", "代际", "家族数据", "身体框架法", "认知透镜", "自我的七个层面", "百慕大", "借光而行", "借事修心", "生命数据化", "格物致知", "知行合一", "良知", ], }, { "filename": "03_常见学员问题回应素材.md", "boost": 2.8, "keywords": [ "妈妈", "父母", "原生家庭", "孩子", "不上学", "躺平", "老公", "伴侣", "出轨", "离婚", "课程没用", "没体感", "没变化", "失望", "知道但做不到", "反应过度", "小事失控", "想放弃", "不想写输出", "不想分享", "群里刷屏", "攻击课程", "攻击老师", "没救了", "别人都有反应", "我没有反应", "依赖老师", ], }, { "filename": "02_即时回复表达原则.md", "boost": 2.4, "keywords": [ "怎么回", "怎么说", "如何回复", "参考回复", "话术", "语气", "表达", "阅读感", "人情味", "长故事", "短问题", "群内", "私聊", "AI味", "不要像AI", "课程词", "慎用术语", ], }, { "filename": "01_课程层次与问题判断.md", "boost": 2.2, "keywords": [ "课程层次", "299", "1280", "七天训练营", "启蒙课", "训练场", "卡在哪", "层次判断", "课程区别", "大本营偏向", "回溯", "未完成事件", "事实法则", "流动法则", "现实风险层", "后续训练需求", ], }, ] # ───────────────────────────── 工具函数 ───────────────────────────── def _tokenize(text: str) -> set[str]: """中文分词（使用jieba）""" try: import jieba return set(jieba.cut(text)) except ImportError: # 降级：按字拆分 return set(text) def _calc_relevance(query: str, content: str) -> float: """计算查询与内容的相关度（0-1）""" query_tokens = _tokenize(query.lower()) content_tokens = _tokenize(content.lower()) if not query_tokens: return 0.0 overlap = query_tokens & content_tokens return len(overlap) / len(query_tokens) def _normalize_text(text: str) -> str: """用于短语命中的轻量归一化。""" return re.sub(r"\s+", "", text.lower()) def _count_phrase_hits(query: str, phrases: list[str]) -> int: """统计查询命中的业务短语数量。""" normalized_query = _normalize_text(query) hits = 0 for phrase in phrases: normalized_phrase = _normalize_text(phrase) if normalized_phrase and normalized_phrase in normalized_query: hits += 1 return hits def _metadata_text(filename: str) -> str: """拼接知识库元数据，供检索加权。""" meta = _KB_METADATA.get(filename, {}) return " ".join( [ meta.get("title", ""), meta.get("description", ""), " ".join(meta.get("keywords", [])), ] ) def _calc_metadata_score(query: str, filename: str) -> float: """计算文件元数据与查询的相关度。""" meta = _KB_METADATA.get(filename, {}) keywords = meta.get("keywords", []) phrase_hits = _count_phrase_hits(query, keywords) relevance = _calc_relevance(query, _metadata_text(filename)) return relevance + min(phrase_hits * 0.35, 2.0) def _calc_route_boost(query: str, filename: str) -> float: """根据 V7.1 固定业务场景给目标知识库加权。""" boost = 0.0 for rule in _ROUTE_RULES: hits = _count_phrase_hits(query, rule["keywords"]) if not hits: continue if rule["filename"] == filename: boost += rule["boost"] + min(hits * 0.3, 1.5) return boost def _extract_relevant_sections(content: str, query: str, max_sections: int = 3) -> list[str]: """从知识库内容中提取与查询最相关的章节""" # 按二级标题分割 sections = re.split(r"\n(?=##\s+)", content) scored = [] for section in sections: if not section.strip(): continue score = _calc_relevance(query, section) scored.append((score, section)) scored.sort(key=lambda x: x[0], reverse=True) return [s for _, s in scored[:max_sections]] def _normalize_top_k(top_k: int) -> int: """限制 top_k，避免模型误传过大值导致输出过长。""" try: value = int(top_k) except (TypeError, ValueError): value = 3 return max(1, min(value, 8)) # ───────────────────────────── 公开工具 ───────────────────────────── def list_kb() -> str: """列出所有可用的知识库文件 当你不确定该查哪个知识库文件时，先调用此工具查看列表。 """ lines = ["【可用知识库文件】\n"] for fname, meta in _KB_METADATA.items(): lines.append(f"◆ {meta['title']}") lines.append(f" 描述: {meta['description']}") lines.append(f" 关键词: {', '.join(meta['keywords'][:5])}...\n") return "\n".join(lines) def load_kb(filename: str) -> str: """加载指定知识库文件的完整内容 Args: filename: 知识库文件名（如 "01_课程层次与问题判断.md"） 当你已确定需要哪个文件的内容时，调用此工具加载完整内容。 """ if filename not in _KB_METADATA: available = ", ".join(_KB_METADATA.keys()) return f"错误: 文件 '{filename}' 不存在。可用文件: {available}" content = load_knowledge_file(_KB_DIR, filename) if not content: return f"错误: 无法读取文件 '{filename}'" return _strip_markdown(content) def search_kb(query: str, top_k: int = 3) -> str: """根据关键词搜索所有知识库文件，返回最相关的片段 Args: query: 搜索关键词（中文） top_k: 返回结果数量（默认3条） 当你不确定具体文件，但知道要查什么内容时，调用此工具。 """ top_k = _normalize_top_k(top_k) all_files = list_knowledge_files(_KB_DIR) results: list[tuple[float, str, str]] = [] for fname in all_files: content = load_knowledge_file(_KB_DIR, fname) if not content: continue # 计算文件级、元数据级、硬路由级相关度 file_score = _calc_relevance(query, content) metadata_score = _calc_metadata_score(query, fname) route_boost = _calc_route_boost(query, fname) # 提取最相关的章节 sections = _extract_relevant_sections(content, query, max_sections=3) for section in sections: section_score = _calc_relevance(query, section) section_phrase_hits = _count_phrase_hits( query, re.findall(r"^#{1,6}\s+(.+)$", section, flags=re.MULTILINE), ) # 综合得分：硬路由 > 元数据 > 章节 > 全文，优先保证关键业务场景命中正确文件。 final_score = ( route_boost + metadata_score * 0.7 + section_score * 1.2 + file_score * 0.3 + min(section_phrase_hits * 0.4, 1.2) ) results.append((final_score, fname, section)) # 排序并取 top_k results.sort(key=lambda x: x[0], reverse=True) top_results = results[:top_k] if not top_results: return f"未找到与 '{query}' 相关的知识库内容。" lines = [f"【知识库搜索结果: '{query}'】\n"] for i, (score, fname, section) in enumerate(top_results, 1): meta = _KB_METADATA.get(fname, {}) title = meta.get("title", fname) lines.append(f"结果{i}（相关度: {score:.2f}）| 来源: {title}") lines.append(f"{_strip_markdown(section[:800])}\n") return "\n".join(lines)