feat: solution for 'Повторный экзамен: FAQ-бот — ChromaDB + один MCP-tool'
This commit is contained in:
@@ -7,11 +7,19 @@
|
||||
EN
|
||||
Повторный экзамен: FAQ-бот — ChromaDB + один MCP-tool
|
||||
Зачёт
|
||||
Версия 4
|
||||
Версия 5
|
||||
Дедлайн сдачи: 31.08.2026
|
||||
|
||||
В работе
|
||||
|
||||
Требуется доработка
|
||||
|
||||
К сожалению, проект не соответствует заявленному стеку задания. В описании требуется использовать ChromaDB и Ollama‑embed‑text, однако в коде реализовано хранение embeddings через Qdrant. Пересмотрите решение: замените Qdrant на ChromaDB, убедитесь, что все импорты и зависимости соответствуют
|
||||
Решение не соответствует заявленному стеку задания. Пожалуйста, пересмотрите работу и убедитесь в использовании ChromaDB с Ollama‑embed‑text, а также корректной интеграции MCP‑тулов.
|
||||
|
||||
Редактирование ответа
|
||||
|
||||
Заполните ответ и отправьте работу на проверку преподавателю.
|
||||
|
||||
Тип ответа
|
||||
Текст
|
||||
Ссылка
|
||||
+3
-6
@@ -1,6 +1,3 @@
|
||||
fastapi
|
||||
uvicorn[standard]
|
||||
qdrant-client
|
||||
sentence-transformers
|
||||
python-telegram-bot
|
||||
pydantic
|
||||
chromadb==0.4.24
|
||||
ollama==0.1.0
|
||||
mcp-tools==0.1.0
|
||||
+1
-1
@@ -1 +1 @@
|
||||
# Empty init file to make src a package
|
||||
# This file makes src a Python package.
|
||||
+100
@@ -0,0 +1,100 @@
|
||||
"""
|
||||
ChromaDB wrapper for storing and querying FAQ documents.
|
||||
"""
|
||||
|
||||
import chromadb
|
||||
from chromadb.config import Settings
|
||||
from typing import List, Dict, Any
|
||||
|
||||
class ChromaDB:
|
||||
"""
|
||||
Wrapper around ChromaDB to handle FAQ documents.
|
||||
"""
|
||||
|
||||
def __init__(self, persist_path: str = "chromadb"):
|
||||
"""
|
||||
Initialize the ChromaDB client and collection.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
persist_path : str, optional
|
||||
Directory to persist the database. Defaults to "chromadb".
|
||||
"""
|
||||
self.client = chromadb.Client(Settings(persist_directory=persist_path))
|
||||
self.collection_name = "faq"
|
||||
self.collection = self.client.get_or_create_collection(name=self.collection_name)
|
||||
|
||||
def add_document(self, text: str, embedding: List[float], doc_id: str = None):
|
||||
"""
|
||||
Add a single document to the collection.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
text : str
|
||||
The document text.
|
||||
embedding : List[float]
|
||||
The embedding vector for the document.
|
||||
doc_id : str, optional
|
||||
Optional document ID. If None, an auto-generated ID is used.
|
||||
"""
|
||||
if doc_id is None:
|
||||
# Generate a simple incremental ID
|
||||
existing_ids = self.collection.get()["ids"]
|
||||
doc_id = str(len(existing_ids))
|
||||
self.collection.add(
|
||||
documents=[text],
|
||||
embeddings=[embedding],
|
||||
ids=[doc_id]
|
||||
)
|
||||
|
||||
def query(self, embedding: List[float], k: int = 5) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Query the collection for the top-k most similar documents.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
embedding : List[float]
|
||||
The query embedding.
|
||||
k : int, optional
|
||||
Number of results to return. Defaults to 5.
|
||||
|
||||
Returns
|
||||
-------
|
||||
List[Dict[str, Any]]
|
||||
List of dictionaries containing 'id', 'document', and 'distance'.
|
||||
"""
|
||||
results = self.collection.query(
|
||||
query_embeddings=[embedding],
|
||||
n_results=k,
|
||||
include=["documents", "distances", "ids"]
|
||||
)
|
||||
docs = []
|
||||
for doc, dist, doc_id in zip(results["documents"][0], results["distances"][0], results["ids"][0]):
|
||||
docs.append({"id": doc_id, "document": doc, "distance": dist})
|
||||
return docs
|
||||
|
||||
def is_empty(self) -> bool:
|
||||
"""
|
||||
Check if the collection has any documents.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True if empty, False otherwise.
|
||||
"""
|
||||
return len(self.collection.get()["ids"]) == 0
|
||||
|
||||
def load_sample_data(self, sample_data: List[Dict[str, str]]):
|
||||
"""
|
||||
Load a list of sample documents into the collection.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
sample_data : List[Dict[str, str]]
|
||||
List of dictionaries with keys 'text' and optional 'id'.
|
||||
"""
|
||||
for item in sample_data:
|
||||
text = item["text"]
|
||||
doc_id = item.get("id")
|
||||
embedding = embed_text(text)
|
||||
self.add_document(text, embedding, doc_id=doc_id)
|
||||
+26
-11
@@ -1,13 +1,28 @@
|
||||
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.
|
||||
Embedding utilities using Ollama's embed-text model.
|
||||
"""
|
||||
return model.encode(text, convert_to_numpy=False).tolist()
|
||||
|
||||
import ollama
|
||||
from typing import List
|
||||
|
||||
def embed_text(text: str, model: str = "embed-text") -> List[float]:
|
||||
"""
|
||||
Generate an embedding for the given text using Ollama's embed-text model.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
text : str
|
||||
The input text to embed.
|
||||
model : str, optional
|
||||
The Ollama model name. Defaults to "embed-text".
|
||||
|
||||
Returns
|
||||
-------
|
||||
List[float]
|
||||
The embedding vector.
|
||||
"""
|
||||
try:
|
||||
result = ollama.embeddings(model=model, prompt=text)
|
||||
return result["embedding"]
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Failed to embed text: {e}") from e
|
||||
+56
-60
@@ -1,68 +1,64 @@
|
||||
"""
|
||||
Main entry point for the FAQ bot.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
from fastapi import FastAPI, HTTPException, Query
|
||||
from pydantic import BaseModel
|
||||
import json
|
||||
from src.embedding import embed_text
|
||||
from src.database import ChromaDB
|
||||
from src.mcp_tools import generate_answer
|
||||
|
||||
from .vector_store import QdrantVectorStore
|
||||
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):
|
||||
def load_sample_faq() -> list:
|
||||
"""
|
||||
Add a new document to the vector store.
|
||||
Load a small sample FAQ dataset.
|
||||
"""
|
||||
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}
|
||||
return [
|
||||
{"id": "0", "text": "What is the return policy? Our return policy allows returns within 30 days of purchase."},
|
||||
{"id": "1", "text": "How do I track my order? You can track your order using the tracking link sent to your email."},
|
||||
{"id": "2", "text": "What payment methods are accepted? We accept Visa, MasterCard, and PayPal."},
|
||||
{"id": "3", "text": "Do you ship internationally? Yes, we ship to most countries worldwide."},
|
||||
{"id": "4", "text": "How can I contact customer support? You can contact us via email at support@example.com."},
|
||||
]
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="FAQ Bot using ChromaDB and MCP-tools")
|
||||
parser.add_argument("--question", type=str, help="Your question to ask the bot")
|
||||
args = parser.parse_args()
|
||||
|
||||
@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}
|
||||
if not args.question:
|
||||
print("Please provide a question using --question")
|
||||
return
|
||||
|
||||
# Initialize database
|
||||
db = ChromaDB()
|
||||
|
||||
@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}
|
||||
# If the collection is empty, load sample data
|
||||
if db.is_empty():
|
||||
print("Database empty. Loading sample FAQ data...")
|
||||
sample_data = load_sample_faq()
|
||||
for item in sample_data:
|
||||
text = item["text"]
|
||||
doc_id = item.get("id")
|
||||
embedding = embed_text(text)
|
||||
db.add_document(text, embedding, doc_id=doc_id)
|
||||
print("Sample data loaded.")
|
||||
|
||||
# Generate embedding for the question
|
||||
question_embedding = embed_text(args.question)
|
||||
|
||||
# Query the database for relevant documents
|
||||
results = db.query(question_embedding, k=5)
|
||||
|
||||
# Extract context documents
|
||||
context = [res["document"] for res in results]
|
||||
|
||||
# Generate answer using MCP-tools
|
||||
answer = generate_answer(context, args.question)
|
||||
|
||||
# Output the answer
|
||||
print("\nAnswer:")
|
||||
print(answer)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,33 @@
|
||||
"""
|
||||
Integration with MCP-tools for generating answers.
|
||||
"""
|
||||
|
||||
import mcp_tools
|
||||
from typing import List
|
||||
|
||||
def generate_answer(context: List[str], question: str) -> str:
|
||||
"""
|
||||
Generate an answer using MCP-tools given context and a question.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
context : List[str]
|
||||
List of context strings retrieved from the database.
|
||||
question : str
|
||||
The user's question.
|
||||
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
The generated answer.
|
||||
"""
|
||||
# Combine context into a single string
|
||||
context_text = "\n".join(context)
|
||||
# Construct a prompt for MCP-tools
|
||||
prompt = f"Question: {question}\nContext:\n{context_text}\nAnswer:"
|
||||
# Use MCP-tools to generate the answer
|
||||
try:
|
||||
response = mcp_tools.generate(prompt=prompt)
|
||||
return response
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Failed to generate answer with MCP-tools: {e}") from e
|
||||
Reference in New Issue
Block a user