simonkoson/tps-dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Obsidian 知识库批量导入 + 语义搜索体验升级

Browse Source 
- 新增 import_obsidian_kb.py 批量导入脚本（164篇入库，知识库186条）
- parse_md_file 补强：来源fallback、相关装备/应用领域实体提取、
  Obsidian双链去括号、原始类别/源文件路径存metadata、TYPE_MAP扩充
- search_similar 改进：智能摘要（中文2-gram拆词+加权段落匹配）、
  min_similarity=0.3过滤、top_k 5→10、返回完整content_md
- 前端搜索卡片升级：展开全文、关键词加粗渲染、相关度分档样式
- CLAUDE.md 状态更新

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
simonkoson
2026-07-08 15:26:17 +08:00
parent c21bc7d093
commit db9d10a795
6 changed files with 426 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/knowledge.py
+8 -2
View File
@@ -109,7 +109,8 @@ def get_grouped_knowledge_items(
class SearchRequest(BaseModel):
query: str
top_k: int = 5
top_k: int = 10
min_similarity: float = 0.3
@router.post("/search")
@@ -121,10 +122,15 @@ def search_knowledge(
"""
语义检索：输入一段文字，返回最相关的知识库条目及相似度。
查询向量用 type="query"（区分于存入时的 type="db"）。
top_k 默认 10，min_similarity 默认 0.3 过滤低相关条目。
三角色均可读。
"""
svc = KnowledgeService()
results = svc.search_similar(query_text=body.query, top_k=body.top_k)
results = svc.search_similar(
query_text=body.query,
top_k=body.top_k,
min_similarity=body.min_similarity,
)
return {
"results": results,
"query": body.query,
backend/app/services/knowledge_service.py
+117 -12
View File
@@ -23,8 +23,15 @@ class KnowledgeService:
SOURCE_TYPE_MAP = {
"杂志文章": "military_report",
"军报": "military_report",
"军报文章": "military_report",
"节目文稿": "manuscript",
"报题单": "baoti",
"装备": "manual",
"技术": "manual",
"动态": "manual",
"术语": "manual",
"厂商": "manual",
"索引": "manual",
}
# 来源大类固定显示顺序（制片人 Obsidian 习惯）
@@ -89,6 +96,12 @@ class KnowledgeService:
else:
source_detail = None
# fallback：如果 source_detail 仍为空，取"来源"字段
if not source_detail:
raw_source = str(fm.get("来源", "") or "").strip()
if raw_source:
source_detail = raw_source
# —— 播出日期：容错 "待补充" 等非日期文本——
raw_date = str(fm.get("播出日期", "") or "").strip()
publish_date = None
@@ -107,17 +120,22 @@ class KnowledgeService:
# —— 权重（不展示，存 JSONB 备 Phase 4）——
weight = str(fm.get("权重", "") or "").strip() or None
# —— 相关实体（涉及装备/涉及技术/涉及厂商/主题）——
# —— 相关实体（涉及装备/涉及技术/涉及厂商/主题/相关装备/应用领域）——
import re as _re
related_entities = []
for key in ("涉及装备", "涉及技术", "涉及厂商", "主题"):
for key in ("涉及装备", "涉及技术", "涉及厂商", "主题", "相关装备", "应用领域"):
val = fm.get(key)
if val:
if isinstance(val, list):
related_entities.extend(val)
for item in val:
# 去掉 Obsidian 双链格式 [[xxx]]
cleaned = _re.sub(r"\[\[|\]\]", "", str(item)).strip()
if cleaned:
related_entities.append(cleaned)
elif isinstance(val, str):
# 可能是 "山东舰, 福建舰" 这样的逗号分隔字符串
for item in val.replace("，", ",").split(","):
item = item.strip()
item = _re.sub(r"\[\[|\]\]", "", item).strip()
if item:
related_entities.append(item)
@@ -130,6 +148,26 @@ class KnowledgeService:
# related_concepts 字段预留给双链解析（Phase 4），本 Task 原样存入
metadata["double_bracket_links"] = self._extract_double_brackets(parsed.content)
# 保留原始 Obsidian 类别（映射到 manual 的类型，如装备/技术/动态/术语/厂商等）
if raw_type and source_type == "manual":
metadata["obsidian_category"] = raw_type
# 源文件路径保留（预埋 PDF 原文链接）
raw_source_files = fm.get("源文件")
if raw_source_files:
cleaned_files = []
if isinstance(raw_source_files, list):
for sf in raw_source_files:
cleaned = _re.sub(r'[\[\]"]', "", str(sf)).strip()
if cleaned:
cleaned_files.append(cleaned)
elif isinstance(raw_source_files, str):
cleaned = _re.sub(r'[\[\]"]', "", raw_source_files).strip()
if cleaned:
cleaned_files.append(cleaned)
if cleaned_files:
metadata["source_files"] = cleaned_files
# —— 正文（去掉 frontmatter 的纯内容）——
content_md = parsed.content
@@ -229,13 +267,11 @@ class KnowledgeService:
sources.add(tags["source_detail"])
return sorted(list(sources))
def search_similar(self, query_text: str, top_k: int = 5) -> list[dict]:
def search_similar(self, query_text: str, top_k: int = 10, min_similarity: float = 0.3) -> list[dict]:
"""
语义检索：查询句转为向量，用 SQL 余弦距离（<=>）在数据库层检索
返回 top_k 条相似笔记，含相似度分数 + 原文片段（SQL 端截断前 200 字）。
注意：当前取前 200 字是已知妥协（整篇向量检索无法定位中段命中点），
Phase 4a 做切块检索（chunk）时可优化为取最相关片段。
返回 top_k 条相似笔记，含相似度分数 + 智能摘要片段。
过滤掉 similarity < min_similarity 的条目。
"""
query_vector = self.embedder.embed_single(query_text, embed_type="query")
vec_str = "[" + ",".join(str(v) for v in query_vector) + "]"
@@ -248,7 +284,7 @@ class KnowledgeService:
ki.source_type,
ki.author,
ki.tags,
SUBSTRING(ki.content_md, 1, 200) AS snippet,
ki.content_md,
1 - (ke.embedding <=> '{vec_str}'::vector) AS similarity
FROM knowledge_embeddings ke
JOIN knowledge_items ki ON ke.knowledge_id = ki.id
@@ -260,19 +296,88 @@ class KnowledgeService:
rows = session.execute(stmt).all()
results = []
for r in rows:
similarity = round(r.similarity, 4)
# 过滤低于阈值的条目
if similarity < min_similarity:
continue
tags = r.tags or {}
source_detail = tags.get("source_detail") if isinstance(tags, dict) else None
snippet = self._extract_smart_snippet(r.content_md, query_text)
results.append({
"id": r.id,
"title": r.title,
"source_type": r.source_type,
"author": r.author,
"source_detail": source_detail,
"snippet": r.snippet,
"similarity": round(r.similarity, 4),
"snippet": snippet,
"content_md": r.content_md,
"similarity": similarity,
})
return results
def _extract_smart_snippet(self, content_md: str, query_text: str, max_len: int = 300) -> str:
"""
智能摘要提取：根据搜索词定位最相关段落，截取摘要并加粗关键词。
中文连续字符串会被拆成 2-gram 子串以提高段落匹配命中率。
"""
import re as _re
if not content_md:
return ""
# 1. 切分关键词：先按空格/标点拆，再把长中文词拆成 2-gram
raw_parts = _re.split(r'[\s，。！？、；：""''（）\(\)\[\]\-]+', query_text)
raw_parts = [p.strip() for p in raw_parts if len(p.strip()) > 1]
keywords = []
for part in raw_parts:
keywords.append(part)
if len(part) > 2 and _re.fullmatch(r'[一-鿿]+', part):
for i in range(len(part) - 1):
bigram = part[i:i+2]
if bigram not in keywords:
keywords.append(bigram)
if not keywords:
return content_md[:max_len]
# 2. 按段落分割（跳过纯 markdown 标记行如 # ## --- 等）
paragraphs = content_md.split("\n\n")
if len(paragraphs) < 3:
paragraphs = content_md.split("\n")
paragraphs = [p.strip() for p in paragraphs
if p.strip() and not _re.fullmatch(r'[#\-=\s>]+', p.strip())]
# 3. 计算每段关键词命中数（完整词权重 3，bigram 权重 1）
best_para = ""
best_score = 0
for para in paragraphs:
score = 0
for kw in keywords:
if kw in para:
score += 3 if kw in raw_parts else 1
if score > best_score:
best_score = score
best_para = para
# 4. fallback：无关键词命中时用正文前 max_len 字
if best_score == 0:
snippet = content_md[:max_len]
else:
snippet = best_para[:max_len]
# 5. 加粗命中关键词（优先标记完整词，避免 bigram 重复标记）
marked = set()
for kw in sorted(keywords, key=len, reverse=True):
if kw in snippet and kw not in marked:
snippet = snippet.replace(kw, f"**{kw}**", 1)
marked.add(kw)
for sub_kw in keywords:
if sub_kw != kw and sub_kw in kw:
marked.add(sub_kw)
return snippet
def get_item_count(self) -> int:
with Session(engine) as session:
return len(session.exec(select(KnowledgeItem)).all())