Initial implementation of FAQ bot with ChromaDB and MCP-style tool: update src/utils.py

This commit is contained in:
2026-06-10 13:58:07 +00:00
parent cdb09538fe
commit 9ee82aed51
+19
View File
@@ -1,11 +1,14 @@
import os import os
from pathlib import Path from pathlib import Path
import json
import httpx
from langchain_ollama import OllamaEmbeddings from langchain_ollama import OllamaEmbeddings
from langchain_chroma import Chroma from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_text_splitters import RecursiveCharacterTextSplitter
CHROMA_PATH = Path("./chroma_faq") CHROMA_PATH = Path("./chroma_faq")
# Load FAQ markdown files into ChromaDB
def load_faq_to_chroma(md_dir: str = "data"): def load_faq_to_chroma(md_dir: str = "data"):
Path(md_dir).mkdir(parents=True, exist_ok=True) Path(md_dir).mkdir(parents=True, exist_ok=True)
md_files = list(Path(md_dir).glob("*.md")) md_files = list(Path(md_dir).glob("*.md"))
@@ -22,7 +25,23 @@ def load_faq_to_chroma(md_dir: str = "data"):
chroma.persist() chroma.persist()
return chroma return chroma
# Search the ChromaDB for relevant documents
def search_course_docs(query: str, k: int = 3): def search_course_docs(query: str, k: int = 3):
chroma = Chroma(persist_directory=str(CHROMA_PATH), embedding_function=OllamaEmbeddings(model="nomic-embed-text")) chroma = Chroma(persist_directory=str(CHROMA_PATH), embedding_function=OllamaEmbeddings(model="nomic-embed-text"))
retriever = chroma.as_retriever(search_kwargs={"k": k}) retriever = chroma.as_retriever(search_kwargs={"k": k})
return retriever.get_relevant_documents(query) return retriever.get_relevant_documents(query)
# MCPstyle tool: fetch metadata from a mock HTTP endpoint
# In production this would be a real MCP server. Here we use a local JSON file served by http.server.
def fetch_course_meta(query: str):
url = f"http://localhost:8000/course_meta.json"
try:
response = httpx.get(url, timeout=5.0)
response.raise_for_status()
data = response.json()
except Exception:
# Fallback to local static file if server not running
data = json.loads(Path("data/course_meta.json").read_text(encoding="utf-8"))
# Simple filtering: return items where query is in title or description
results = [item for item in data if query.lower() in item.get("title", "").lower() or query.lower() in item.get("description", "").lower()]
return results