234 lines
9.2 KiB
Python
234 lines
9.2 KiB
Python
"""L2 训练营知识库工具 — 按需加载 + 语义搜索
|
||
|
||
本文件包含知识库查询工具的实现。修改主提示词时请勿改动此文件。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from pathlib import Path
|
||
from typing import List
|
||
|
||
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": [
|
||
"表达", "回复", "语气", "措辞", "风格", "原则", "禁忌",
|
||
"怎么说", "话术", "沟通", "表达技巧", "回复模板",
|
||
],
|
||
},
|
||
"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": [
|
||
"大本营", "承接", "成交", "异议", "销售", "推课",
|
||
"后续课程", "持续训练", "课程信心", "报名", "价格", "权益",
|
||
"没时间", "没钱", "怕没效果", "家人不同意", "不想继续",
|
||
],
|
||
},
|
||
}
|
||
|
||
# ───────────────────────────── 工具函数 ─────────────────────────────
|
||
|
||
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 _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 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条）
|
||
|
||
当你不确定具体文件，但知道要查什么内容时，调用此工具。
|
||
"""
|
||
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)
|
||
|
||
# 提取最相关的章节
|
||
sections = _extract_relevant_sections(content, query, max_sections=2)
|
||
|
||
for section in sections:
|
||
section_score = _calc_relevance(query, section)
|
||
# 综合得分：文件级 + 章节级
|
||
final_score = (file_score + section_score) / 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)
|