feat: solution for 'Повторный экзамен: FAQ-бот — ChromaDB + один MCP-tool'

Dockerfile

@@ -0,0 +1,23 @@
+# Use official lightweight Python image
+FROM python:3.11-slim
+
+# Set working directory
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements and install Python packages
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+# Expose port for FastAPI
+EXPOSE 8000
+
+# Run the FastAPI application
+CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -7,7 +7,7 @@
 EN
 Повторный экзамен: FAQ-бот — ChromaDB + один MCP-tool
 Зачёт
-Версия 2
+Версия 3
 Дедлайн сдачи: 31.08.2026
 
 В работе
@@ -30,6 +30,6 @@ EN
 Отправить на проверку
 Отменить
 
-Задание
+ПОДРОБНЕЕ
 
-Практичес
+Задание

requirements.txt

@@ -1,6 +1,6 @@
-langchain==0.1.0
+fastapi
-langchain-chroma==0.1.0
+uvicorn[standard]
-langchain-ollama==0.1.0
+qdrant-client
-chromadb==0.4.24
+sentence-transformers
-httpx==0.27.0
+python-telegram-bot
-python-dotenv==1.0.1
+pydantic

src/__init__.py

@@ -0,0 +1 @@
+# Empty init file to make src a package

src/bot.py

@@ -0,0 +1,57 @@
+import os
+import logging
+from telegram import Update
+from telegram.ext import (
+    ApplicationBuilder,
+    CommandHandler,
+    ContextTypes,
+)
+
+from .vector_store import QdrantVectorStore
+from .mcp_tool import MCPTool
+
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    level=logging.INFO,
+)
+
+# Initialize vector store and MCP tool
+vector_store = QdrantVectorStore(
+    host=os.getenv("QDRANT_HOST"),
+    port=int(os.getenv("QDRANT_PORT", "6333")),
+    collection_name=os.getenv("QDRANT_COLLECTION", "faq_collection"),
+)
+mcp_tool = MCPTool(vector_store)
+
+
+async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    await update.message.reply_text("Hello! Use /ask <your question> to get an answer.")
+
+
+async def ask(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    if not context.args:
+        await update.message.reply_text("Please provide a question after /ask.")
+        return
+    query = " ".join(context.args)
+    answers = mcp_tool.answer(query, top_k=3)
+    if not answers:
+        await update.message.reply_text("No relevant information found.")
+    else:
+        response = "\n\n".join(
+            [f"*Answer {i+1}:*\n{answer.get('text', 'No text')}" for i, answer in enumerate(answers)]
+        )
+        await update.message.reply_text(response, parse_mode="Markdown")
+
+
+def main() -> None:
+    bot_token = os.getenv("BOT_TOKEN")
+    if not bot_token:
+        raise RuntimeError("BOT_TOKEN environment variable not set.")
+    application = ApplicationBuilder().token(bot_token).build()
+    application.add_handler(CommandHandler("start", start))
+    application.add_handler(CommandHandler("ask", ask))
+    application.run_polling()
+
+
+if __name__ == "__main__":
+    main()

src/embedding.py

@@ -0,0 +1,13 @@
+import os
+from sentence_transformers import SentenceTransformer
+
+# Load the embedding model once at module import
+MODEL_NAME = os.getenv("EMBEDDING_MODEL", "all-MiniLM-L6-v2")
+model = SentenceTransformer(MODEL_NAME)
+
+
+def get_embedding(text: str) -> list[float]:
+    """
+    Generate a vector embedding for the given text using the loaded model.
+    """
+    return model.encode(text, convert_to_numpy=False).tolist()

src/main.py

@@ -1,4 +1,68 @@
-from .cli import main
+import os
+from fastapi import FastAPI, HTTPException, Query
+from pydantic import BaseModel
 
-if __name__ == "__main__":
+from .vector_store import QdrantVectorStore
-    main()
+from .mcp_tool import MCPTool
+
+app = FastAPI(
+    title="FAQ Bot API",
+    description="A simple FAQ bot using Qdrant as the vector store.",
+    version="1.0.0",
+)
+
+# Initialize the vector store and MCP tool
+vector_store = QdrantVectorStore(
+    host=os.getenv("QDRANT_HOST"),
+    port=int(os.getenv("QDRANT_PORT", "6333")),
+    collection_name=os.getenv("QDRANT_COLLECTION", "faq_collection"),
+)
+mcp_tool = MCPTool(vector_store)
+
+
+class Document(BaseModel):
+    id: str
+    text: str
+    metadata: dict | None = None
+
+
+@app.post("/documents", status_code=201)
+def add_document(doc: Document):
+    """
+    Add a new document to the vector store.
+    """
+    try:
+        vector_store.add_document(doc.id, doc.text, doc.metadata)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    return {"status": "added", "id": doc.id}
+
+
+@app.get("/search")
+def search(
+    query: str = Query(..., description="Search query"),
+    top_k: int = Query(5, ge=1, le=20, description="Number of results to return"),
+):
+    """
+    Search for documents similar to the query.
+    """
+    try:
+        results = vector_store.search(query, top_k=top_k)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    return {"query": query, "results": results}
+
+
+@app.get("/answer")
+def answer(
+    query: str = Query(..., description="Question to answer"),
+    top_k: int = Query(3, ge=1, le=10, description="Number of answers to return"),
+):
+    """
+    Get the best answer(s) for the query using MCPTool.
+    """
+    try:
+        answers = mcp_tool.answer(query, top_k=top_k)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    return {"query": query, "answers": answers}

src/mcp_tool.py

@@ -0,0 +1,18 @@
+from typing import List, Dict
+
+from .vector_store import QdrantVectorStore
+
+
+class MCPTool:
+    """
+    A simple tool that uses the vector store to answer queries.
+    """
+
+    def __init__(self, vector_store: QdrantVectorStore):
+        self.vector_store = vector_store
+
+    def answer(self, query: str, top_k: int = 3) -> List[Dict]:
+        """
+        Return the top_k most relevant documents for the query.
+        """
+        return self.vector_store.search(query, top_k=top_k)

src/vector_store.py

@@ -0,0 +1,77 @@
+import os
+from typing import Dict, List
+
+from qdrant_client import QdrantClient
+from qdrant_client.http import models as qdrant_models
+from qdrant_client.http.models import PointStruct, VectorParams, Distance
+
+from .embedding import get_embedding
+
+
+class QdrantVectorStore:
+    """
+    A simple wrapper around Qdrant to store and retrieve document embeddings.
+    """
+
+    def __init__(
+        self,
+        host: str | None = None,
+        port: int | None = None,
+        collection_name: str = "faq_collection",
+    ):
+        self.host = host or os.getenv("QDRANT_HOST", "localhost")
+        self.port = port or int(os.getenv("QDRANT_PORT", "6333"))
+        self.collection_name = collection_name
+
+        self.client = QdrantClient(host=self.host, port=self.port)
+        self._ensure_collection()
+
+    def _ensure_collection(self) -> None:
+        """
+        Create the collection if it does not exist.
+        """
+        if not self.client.http.collections.exists(
+            collection_name=self.collection_name
+        ):
+            self.client.http.collections.create(
+                collection_name=self.collection_name,
+                vectors_config=VectorParams(
+                    size=384, distance=Distance.COSINE
+                ),
+            )
+
+    def add_document(
+        self,
+        doc_id: str,
+        text: str,
+        metadata: Dict | None = None,
+    ) -> None:
+        """
+        Add a document to the collection with its embedding.
+        """
+        embedding = get_embedding(text)
+        point = PointStruct(
+            id=doc_id,
+            vector=embedding,
+            payload=metadata or {},
+        )
+        self.client.http.points.upsert(
+            collection_name=self.collection_name, points=[point]
+        )
+
+    def search(
+        self,
+        query: str,
+        top_k: int = 5,
+    ) -> List[Dict]:
+        """
+        Search for the most similar documents to the query.
+        Returns a list of payloads (metadata) of the top_k results.
+        """
+        embedding = get_embedding(query)
+        search_result = self.client.http.search(
+            collection_name=self.collection_name,
+            vector=embedding,
+            limit=top_k,
+        )
+        return [hit.payload for hit in search_result]