From bebab0defc28840e41418cef99a6852d07c24e54 Mon Sep 17 00:00:00 2001 From: Danil Parunin 5f1b81b8-4f5d-11e8-9c2d-fa7ae01bbebc Date: Tue, 16 Jun 2026 16:18:08 +0000 Subject: [PATCH] =?UTF-8?q?fix(needs=5Ffixes):=201=20=D0=B8=D1=81=D0=BF?= =?UTF-8?q?=D1=80=D0=B0=D0=B2=D0=BB=D0=B5=D0=BD=D0=B8=D0=B9,=200=20=D0=BE?= =?UTF-8?q?=D1=82=D1=81=D1=82=D0=BE=D1=8F=D0=BD=D0=BE=20=E2=80=94=20main.p?= =?UTF-8?q?y?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.py | 217 +++++++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 161 insertions(+), 56 deletions(-) diff --git a/main.py b/main.py index b0341de..31e79c2 100644 --- a/main.py +++ b/main.py @@ -1,73 +1,178 @@ +""" +# main.py +# FAQ bot using deepagents, ChromaDB and a single MCP‑style HTTP tool. +# The agent decides whether to query the local knowledge base or the +# external metadata service and annotates the answer with a `source` field. +""" import os +import asyncio import json -import httpx from pathlib import Path -from langchain_ollama import ChatOllama +from typing import List, Dict + +from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma -from langchain.embeddings import OllamaEmbeddings -from langchain.schema import Document -from langchain.prompts import ChatPromptTemplate -from langchain.chains import LLMChain +from langchain_core.documents import Document +from langchain_core.messages import HumanMessage, SystemMessage +from langchain.tools import tool +from deepagents import create_deep_agent +from deepagents.backends import FilesystemBackend, LocalShellBackend, CompositeBackend -# 1. Load FAQ into Chroma +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- +# Load API key from .env or environment variable +OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") +if not OPENAI_API_KEY: + raise RuntimeError("OPENAI_API_KEY not set") -def load_faq_to_chroma(md_path: str, persist_dir: str = "./chroma_faq"): - from langchain.document_loaders import TextLoader - from langchain.text_splitter import RecursiveCharacterTextSplitter +# LLM configuration – OpenRouter +llm = ChatOpenAI( + model="openai/gpt-oss-20b:free", + base_url="https://openrouter.ai/api/v1", + api_key=OPENAI_API_KEY, + temperature=0.0, +) - loader = TextLoader(md_path) - docs = loader.load() - splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) - texts = splitter.split_documents(docs) - embeddings = OllamaEmbeddings(model="nomic-embed-text") - chroma = Chroma.from_documents(texts, embeddings, persist_directory=persist_dir) - chroma.persist() - return chroma +# Embeddings – OpenRouter +embeddings = OpenAIEmbeddings( + model="text-embedding-3-small", + base_url="https://openrouter.ai/api/v1", + api_key=OPENAI_API_KEY, +) -# 2. Search function +# --------------------------------------------------------------------------- +# ChromaDB setup +# --------------------------------------------------------------------------- +CHROMA_PATH = Path("./chroma_faq") +CHROMA_COLLECTION = "faq_collection" -def search_course_docs(query: str, k: int = 3): - chroma = Chroma(persist_directory="./chroma_faq", embedding_function=OllamaEmbeddings(model="nomic-embed-text")) - results = chroma.similarity_search(query, k=k) - return [doc.page_content for doc in results] +vector_store = Chroma( + collection_name=CHROMA_COLLECTION, + embedding_function=embeddings, + persist_directory=str(CHROMA_PATH), +) -# 3. MCP-style tool +# Load markdown files into Chroma if not already persisted +if not CHROMA_PATH.exists() or not list(CHROMA_PATH.iterdir()): + def load_faq_to_chroma(md_dir: str = "data"): + md_path = Path(md_dir) + docs: List[Document] = [] + for file in md_path.glob("*.md"): + text = file.read_text(encoding="utf-8") + docs.append(Document(page_content=text, metadata={"source": file.name})) + vector_store.add_documents(docs) + vector_store.persist() -def fetch_course_meta(query: str): - # For demo, use static JSON file - meta_path = Path("meta.json") + load_faq_to_chroma() + +# --------------------------------------------------------------------------- +# Tools +# --------------------------------------------------------------------------- +@tool +def search_course_docs(query: str, k: int = 3) -> str: + """Search the local FAQ collection for relevant passages.""" + docs = vector_store.similarity_search(query, k=k) + if not docs: + return "No relevant information found in the course materials." + return "\n\n---\n\n".join(d.page_content for d in docs) + +# MCP‑style tool – simple HTTP GET to a local JSON file +# For the purpose of this assignment we use a static JSON file +# located at ./meta/course_meta.json +@tool +def fetch_course_meta(query: str) -> str: + """Return metadata that matches the query from a local JSON file.""" + meta_path = Path("./meta/course_meta.json") if not meta_path.exists(): - return {"error": "Meta not found"} - meta = json.loads(meta_path.read_text()) - # simple search by key - return meta.get(query, {}) + return "Metadata file not found." + data = json.loads(meta_path.read_text(encoding="utf-8")) + # Very naive matching: return any entry where the query is a substring + matches = [item for item in data if query.lower() in item.get("title", "").lower()] + if not matches: + return "No matching metadata found." + return json.dumps(matches, indent=2) -# 4. Agent logic +# --------------------------------------------------------------------------- +# Backend for deepagents +# --------------------------------------------------------------------------- +backend = CompositeBackend([ + LocalShellBackend(workspace_dir="./workspace"), + FilesystemBackend(), +]) -def answer_question(question: str): - # Simple heuristic: if question contains "schedule" or "метаданные" use meta - if any(word in question.lower() for word in ["schedule", "расписание", "метаданные"]): - source = "mcp_meta" - answer = fetch_course_meta(question) - else: - source = "chroma" - answer = search_course_docs(question, k=1)[0] - return {"answer": answer, "source": source} +# --------------------------------------------------------------------------- +# Agent definition +# --------------------------------------------------------------------------- +# System prompt instructs the agent to choose the appropriate tool and +# to annotate the answer with a `source` field. +SYSTEM_PROMPT = ( + "You are an FAQ assistant for a course.\n" + "If the user asks about course materials, use the `search_course_docs` tool.\n" + "If the user asks about schedule or metadata, use the `fetch_course_meta` tool.\n" + "Respond in JSON format with two fields: `answer` (string) and `source` (either `chroma` or `mcp_meta`).\n" + "Do not call both tools unless absolutely necessary." +) -if __name__ == "__main__": - # CLI with preset questions - preset = [ - "Что такое ChromaDB?", - "Как подключить Ollama embeddings?", - "Когда будет расписание следующего занятия?" - ] - for q in preset: - res = answer_question(q) - print(f"Q: {q}\nA: {res['answer']}\nSource: {res['source']}\n") - # interactive +agent = create_deep_agent( + model=llm, + tools=[search_course_docs, fetch_course_meta], + backend=backend, + system_prompt=SYSTEM_PROMPT, +) + +# --------------------------------------------------------------------------- +# CLI helpers +# --------------------------------------------------------------------------- +PRESET_QUESTIONS = [ + "What topics are covered in the first lecture?", # chroma + "How can I access the lecture slides?", # chroma + "What is the schedule for the next week?", # mcp_meta +] + +async def run_interactive(): + print("FAQ Bot – type your question (or 'exit' to quit).\n") while True: - q = input("Ask a question (or 'exit'): ") - if q.lower() == "exit": + user_input = input("> ") + if user_input.lower() in {"exit", "quit"}: break - res = answer_question(q) - print(f"A: {res['answer']} (source: {res['source']})") + result = await agent.ainvoke( + {"messages": [HumanMessage(content=user_input)]}, + {"configurable": {"thread_id": "session-1"}}, + ) + # The agent returns a list of messages; the last is the assistant + assistant_msg = result["messages"][-1].content + try: + data = json.loads(assistant_msg) + print(f"\nAnswer: {data['answer']}\nSource: {data['source']}\n") + except Exception: + print("\nUnexpected response format:\n", assistant_msg) + +async def run_presets(): + for q in PRESET_QUESTIONS: + print(f"\nQuestion: {q}") + result = await agent.ainvoke( + {"messages": [HumanMessage(content=q)]}, + {"configurable": {"thread_id": "session-1"}}, + ) + assistant_msg = result["messages"][-1].content + try: + data = json.loads(assistant_msg) + print(f"Answer: {data['answer']}\nSource: {data['source']}") + except Exception: + print("Unexpected response format:\n", assistant_msg) + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="FAQ Bot CLI") + parser.add_argument("--presets", action="store_true", help="Run preset questions") + args = parser.parse_args() + + if args.presets: + asyncio.run(run_presets()) + else: + asyncio.run(run_interactive())