File size: 2,952 Bytes
c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """
Knowledge Base Indexing Pipeline — Markdown loading, preprocessing, and chunking for Qdrant.
"""
import os
import glob
import logging
from typing import List
from langchain_core.documents import Document
from backend.src.services.qdrant_service import QdrantService
from backend.src.config import settings
logger = logging.getLogger("math_assistant.indexer")
class KnowledgeIndexer:
"""Parses local markdown files and indexes them into the Qdrant vector store."""
def __init__(self, kb_dir: str = "knowledge-base"):
self.kb_dir = kb_dir
self.qdrant = QdrantService()
def parse_frontmatter(self, text: str) -> tuple[dict, str]:
"""Extract metadata from markdown frontmatter."""
metadata = {}
content = text
if text.startswith("---\n"):
parts = text.split("---\n", 2)
if len(parts) == 3:
frontmatter = parts[1]
content = parts[2]
for line in frontmatter.strip().split('\n'):
if ':' in line:
key, val = line.split(':', 1)
metadata[key.strip()] = val.strip()
return metadata, content.strip()
def load_markdown_files(self) -> List[Document]:
"""Load all markdown files from the knowledge base directory."""
documents = []
search_pattern = os.path.join(self.kb_dir, "**", "*.md")
filepaths = glob.glob(search_pattern, recursive=True)
logger.info(f"Found {len(filepaths)} markdown files in {self.kb_dir}")
for filepath in filepaths:
try:
with open(filepath, 'r', encoding='utf-8') as f:
text = f.read()
metadata, content = self.parse_frontmatter(text)
metadata["source_file"] = filepath
doc = Document(page_content=content, metadata=metadata)
documents.append(doc)
except Exception as e:
logger.error(f"Failed to process {filepath}: {e}")
return documents
def index_all(self):
"""Load all files, split, and index into Qdrant."""
logger.info("Starting Knowledge Base indexing...")
docs = self.load_markdown_files()
if not docs:
logger.warning("No documents found to index.")
return
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
chunks = splitter.split_documents(docs)
logger.info(f"Created {len(chunks)} chunks. Pushing to Qdrant...")
self.qdrant.add_documents(chunks)
logger.info(f"Indexing complete. Total vectors: {self.qdrant.get_document_count()}")
if __name__ == "__main__":
indexer = KnowledgeIndexer()
indexer.index_all()
|