feat: solution for 'Повторный экзамен: FAQ-бот — ChromaDB + один MCP-tool'
This commit is contained in:
@@ -0,0 +1,85 @@
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any
|
||||
|
||||
import httpx
|
||||
from langchain_community.document_loaders import TextLoader
|
||||
from langchain_community.embeddings import OllamaEmbeddings
|
||||
from langchain_community.vectorstores import Chroma
|
||||
from langchain_core.documents import Document
|
||||
from langchain_core.tools import tool
|
||||
|
||||
# Path to the data directory
|
||||
DATA_DIR = Path(__file__).parent.parent / "data"
|
||||
CHROMA_DIR = Path(__file__).parent.parent / "chroma_faq"
|
||||
|
||||
def load_faq_to_chroma() -> Chroma:
|
||||
"""
|
||||
Load all .md files from the data directory, chunk them, embed with Ollama,
|
||||
and persist into a Chroma vector store.
|
||||
"""
|
||||
# Check if the Chroma collection already exists
|
||||
if CHROMA_DIR.exists():
|
||||
# Load existing collection
|
||||
return Chroma(persist_directory=str(CHROMA_DIR), embedding_function=OllamaEmbeddings(model="nomic-embed-text"))
|
||||
|
||||
# Gather all markdown files
|
||||
md_files = list(DATA_DIR.glob("*.md"))
|
||||
documents: List[Document] = []
|
||||
|
||||
for md_file in md_files:
|
||||
loader = TextLoader(str(md_file), encoding="utf-8")
|
||||
docs = loader.load()
|
||||
documents.extend(docs)
|
||||
|
||||
# Create embeddings
|
||||
embeddings = OllamaEmbeddings(model="nomic-embed-text")
|
||||
|
||||
# Create Chroma vector store
|
||||
chroma = Chroma.from_documents(
|
||||
documents=documents,
|
||||
embedding=embeddings,
|
||||
persist_directory=str(CHROMA_DIR),
|
||||
)
|
||||
return chroma
|
||||
|
||||
@tool
|
||||
def search_course_docs(query: str, k: int = 3) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Search the local FAQ Chroma vector store for relevant documents.
|
||||
|
||||
Returns a list of dictionaries containing the content and metadata.
|
||||
"""
|
||||
chroma = load_faq_to_chroma()
|
||||
results = chroma.similarity_search(query, k=k)
|
||||
output = []
|
||||
for doc in results:
|
||||
output.append(
|
||||
{
|
||||
"content": doc.page_content,
|
||||
"metadata": doc.metadata,
|
||||
}
|
||||
)
|
||||
return output
|
||||
|
||||
@tool
|
||||
def fetch_course_meta(query: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Simulate an MCP-style HTTP tool that returns course metadata
|
||||
matching the query. The metadata is read from a local JSON file.
|
||||
"""
|
||||
meta_path = DATA_DIR / "course_meta.json"
|
||||
with open(meta_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
|
||||
# Simple keyword matching in schedule and instructor fields
|
||||
results = {}
|
||||
if "schedule" in query.lower():
|
||||
results["schedule"] = data.get("schedule", [])
|
||||
if "instructor" in query.lower() or "professor" in query.lower():
|
||||
results["instructor"] = data.get("instructor", {})
|
||||
if not results:
|
||||
# Default to returning the whole metadata if no keyword matched
|
||||
results = data
|
||||
return results
|
||||
Reference in New Issue
Block a user