c410d69970
轻量级 Obsidian Markdown RAG 系统，包含： - Markdown 结构感知分块（标题层级 + 代码块/表格整体保留） - FastEmbed + BAAI/bge-small-zh-v1.5 本地向量化 - SQLite + sqlite-vec 向量库（无需外部服务） - BM25 + 向量混合检索，RRF 融合 - 盘点模式（有哪些/列表/目录类问题） - DeepSeek API 生成回答 - mengya-rag CLI 工具（ask/search/context/read/sync/index/status） - docs/RAG优化方案.md 待实施优化计划 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
117 lines
3.2 KiB
Python
117 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from llama_index.core.schema import TextNode
|
|
|
|
from .config import Settings
|
|
from .embeddings import embed_documents
|
|
from .markdown_chunker import chunk_markdown_file
|
|
from .vector_store import db_path, load_nodes_from_db, write_vector_db
|
|
|
|
|
|
def sync_notes(settings: Settings) -> None:
|
|
settings.notes_dir.mkdir(parents=True, exist_ok=True)
|
|
subprocess.run(
|
|
[
|
|
"rsync",
|
|
"-av",
|
|
"--delete",
|
|
"--exclude=.git/",
|
|
"--exclude=.trash/",
|
|
"--exclude=.obsidian/workspace*",
|
|
settings.notes_remote,
|
|
f"{settings.notes_dir}/",
|
|
],
|
|
check=True,
|
|
)
|
|
|
|
|
|
def markdown_files(notes_dir: Path) -> list[Path]:
|
|
if not notes_dir.exists():
|
|
raise FileNotFoundError(f"笔记目录不存在: {notes_dir}")
|
|
|
|
return sorted(notes_dir.rglob("*.md"))
|
|
|
|
|
|
def chinese_tokenizer(text: str) -> list[str]:
|
|
stopwords = {
|
|
"请",
|
|
"根据",
|
|
"笔记",
|
|
"简要",
|
|
"回答",
|
|
"哪些",
|
|
"什么",
|
|
"内容",
|
|
"一下",
|
|
"我的",
|
|
"有哪些",
|
|
"是什么",
|
|
}
|
|
tokens: list[str] = []
|
|
|
|
for part in re.findall(r"[A-Za-z0-9_#+.-]+|[\u4e00-\u9fff]+", text.lower()):
|
|
if re.fullmatch(r"[A-Za-z0-9_#+.-]+", part):
|
|
if part in {"md", "markdown"}:
|
|
continue
|
|
if part not in stopwords:
|
|
tokens.append(part)
|
|
continue
|
|
|
|
if len(part) == 1:
|
|
if part not in stopwords:
|
|
tokens.append(part)
|
|
continue
|
|
|
|
max_n = min(4, len(part))
|
|
for n in range(2, max_n + 1):
|
|
for index in range(0, len(part) - n + 1):
|
|
token = part[index : index + n]
|
|
if token not in stopwords:
|
|
tokens.append(token)
|
|
|
|
return tokens
|
|
|
|
|
|
def build_index(settings: Settings) -> int:
|
|
nodes: list[TextNode] = []
|
|
for path in markdown_files(settings.notes_dir):
|
|
nodes.extend(chunk_markdown_file(path, settings.notes_dir))
|
|
|
|
if settings.index_dir.exists():
|
|
shutil.rmtree(settings.index_dir)
|
|
settings.index_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
texts = [node_to_embedding_text(node) for node in nodes]
|
|
vectors = embed_documents(settings, texts)
|
|
write_vector_db(db_path(settings.index_dir), nodes, vectors)
|
|
return len(nodes)
|
|
|
|
|
|
def load_index_nodes(settings: Settings) -> list[TextNode]:
|
|
database = db_path(settings.index_dir)
|
|
if not database.exists():
|
|
raise FileNotFoundError(
|
|
f"索引不存在: {database}，请先运行 mengya-build-index"
|
|
)
|
|
return load_nodes_from_db(database)
|
|
|
|
|
|
def node_to_embedding_text(node: TextNode) -> str:
|
|
title = str(node.metadata.get("title", ""))
|
|
rel_path = str(node.metadata.get("rel_path", ""))
|
|
heading_path = str(node.metadata.get("heading_path", ""))
|
|
tags = node.metadata.get("tags", [])
|
|
return (
|
|
f"文件名: {title}\n"
|
|
f"路径: {rel_path}\n"
|
|
f"标题路径: {heading_path}\n"
|
|
f"标签: {tags}\n"
|
|
f"正文:\n{node.get_content()}"
|
|
)
|