diff --git a/pyproject.toml b/pyproject.toml index 8ed57adca..2902a0e9b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,10 @@ dependencies = [ "markitdown>=0.0.2", "openai>=1.64.0", "ray[default]>=2.43.0", + "mcp>=1.5.0", + "anthropic>=0.46.0", + "fastmcp>=0.4.1", + "starlette>=0.41.3", ] [dependency-groups] diff --git a/ragondin/api.py b/ragondin/api.py index 22eb5cc4d..b4dc0bfcf 100644 --- a/ragondin/api.py +++ b/ragondin/api.py @@ -20,6 +20,7 @@ from routers.openai import router as openai_router from routers.search import router as search_router from utils.dependencies import vectordb +from fastmcp import FastMCP config = load_config() DATA_DIR = Path(config.paths.data_dir) @@ -55,6 +56,7 @@ class ChatMsg(BaseModel): app.mount( "/static", StaticFiles(directory=DATA_DIR.resolve(), check_dir=True), name="static" ) +mcp_server = FastMCP("Ragondin") def static_base_url_dependency(request: Request) -> str: diff --git a/ragondin/controllers/indexer_controller.py b/ragondin/controllers/indexer_controller.py new file mode 100644 index 000000000..e878d036c --- /dev/null +++ b/ragondin/controllers/indexer_controller.py @@ -0,0 +1,162 @@ +from typing import List, Optional +from ..models.indexer import ( + IndexationRequest, DeletionRequest, MetadataUpdateRequest, SearchRequest, + IndexationResult, DeletionResult, MetadataUpdateResult, SearchResult +) +from ..components.indexer.indexer import Indexer +from ..utils.config import Config +import logging +import ray + +logger = logging.getLogger(__name__) + +class IndexerController: + """Contrôleur pour les opérations d'indexation + + Cette classe expose des méthodes de haut niveau pour interagir avec + le système d'indexation, en servant d'intermédiaire entre les + requêtes API et l'Indexer. + """ + + def __init__(self, config: Config) -> None: + """Initialise le contrôleur avec une configuration + + Args: + config: Configuration du système + """ + self.config = config + self.indexer = Indexer.remote(config=config, logger=logger) + self.logger = logger + + async def index_documents(self, request: IndexationRequest) -> IndexationResult: + """Indexe des documents selon les paramètres de la requête + + Args: + request: Paramètres d'indexation incluant chemin, métadonnées et partition + + Returns: + Résultat de l'opération d'indexation + """ + try: + self.logger.info(f"Démarrage de l'indexation des documents: {request.path}") + + # Utiliser directement l'Indexer avec Ray + await ray.get(self.indexer.add_files2vdb.remote( + path=request.path, + metadata=request.metadata, + partition=request.partition + )) + + files_count = 1 if isinstance(request.path, str) else len(request.path) + + return IndexationResult( + success=True, + message=f"Indexation réussie de {files_count} fichier(s)", + file_count=files_count, + chunk_count=0 # On ne peut pas facilement compter les chunks ici + ) + + except Exception as e: + self.logger.error(f"Erreur lors de l'indexation: {str(e)}") + return IndexationResult( + success=False, + message=f"Échec de l'indexation: {str(e)}", + file_count=0, + chunk_count=0 + ) + + async def delete_document(self, request: DeletionRequest) -> DeletionResult: + """Supprime un document selon les paramètres de la requête + + Args: + request: Paramètres de suppression incluant l'ID du fichier et la partition + + Returns: + Résultat de l'opération de suppression + """ + try: + self.logger.info(f"Suppression du document: {request.file_id} dans {request.partition}") + + # Utiliser directement l'Indexer avec Ray + await ray.get(self.indexer.delete_file.remote( + file_id=request.file_id, + partition=request.partition + )) + + return DeletionResult( + success=True, + message=f"Suppression réussie du document {request.file_id}", + deleted_points=1 # On ne peut pas facilement compter les points supprimés + ) + + except Exception as e: + self.logger.error(f"Erreur lors de la suppression: {str(e)}") + return DeletionResult( + success=False, + message=f"Échec de la suppression: {str(e)}", + deleted_points=0 + ) + + async def update_metadata(self, request: MetadataUpdateRequest) -> MetadataUpdateResult: + """Met à jour les métadonnées d'un document + + Args: + request: Paramètres de mise à jour incluant l'ID du fichier, + les nouvelles métadonnées et la partition + + Returns: + Résultat de l'opération de mise à jour + """ + try: + self.logger.info(f"Mise à jour des métadonnées pour: {request.file_id}") + + # Utiliser directement l'Indexer avec Ray + await ray.get(self.indexer.update_file_metadata.remote( + file_id=request.file_id, + metadata=request.metadata, + partition=request.partition + )) + + return MetadataUpdateResult( + success=True, + message=f"Mise à jour réussie des métadonnées pour {request.file_id}", + updated_chunks=1 # On ne peut pas facilement compter les chunks mis à jour + ) + + except Exception as e: + self.logger.error(f"Erreur lors de la mise à jour des métadonnées: {str(e)}") + return MetadataUpdateResult( + success=False, + message=f"Échec de la mise à jour des métadonnées: {str(e)}", + updated_chunks=0 + ) + + async def search_documents(self, request: SearchRequest) -> List[SearchResult]: + """Recherche des documents selon les paramètres de la requête + + Args: + request: Paramètres de recherche incluant requête, top_k, seuil, partition et filtres + + Returns: + Liste des résultats de recherche + """ + try: + self.logger.info(f"Recherche pour la requête: {request.query}") + + # Utiliser directement l'Indexer avec Ray + documents = await ray.get(self.indexer.asearch.remote( + query=request.query, + top_k=request.top_k, + similarity_threshold=request.similarity_threshold, + partition=request.partition, + filter=request.filter + )) + + # Convertir les documents en résultats de recherche + results = [SearchResult.from_document(doc) for doc in documents] + + return results + + except Exception as e: + self.logger.error(f"Erreur lors de la recherche: {str(e)}") + return [] \ No newline at end of file diff --git a/ragondin/mcp/llm_client.py b/ragondin/mcp/llm_client.py new file mode 100644 index 000000000..8fbf5ad5d --- /dev/null +++ b/ragondin/mcp/llm_client.py @@ -0,0 +1,236 @@ +from typing import List, Optional, Dict, Any, Type, TypeVar, Union +from pydantic import BaseModel +from loguru import logger +import openai +from mcp import MCPTool, MCPResource +from fastapi import HTTPException +from openai.types.chat import ( + ChatCompletionMessage, + ChatCompletionMessageParam, + ChatCompletionSystemMessageParam, + ChatCompletionToolParam, + ChatCompletionUserMessageParam, +) + +ModelT = TypeVar("ModelT", bound=BaseModel) + +class ChatMessage(BaseModel): + """Modèle pour les messages du chat""" + role: str + content: str + +class ChatCompletionRequest(BaseModel): + """Modèle pour les requêtes de complétion""" + messages: List[ChatMessage] + model: str = "gpt-3.5-turbo" + temperature: float = 0.7 + max_tokens: Optional[int] = None + top_p: float = 1.0 + frequency_penalty: float = 0.0 + presence_penalty: float = 0.0 + stop: Optional[List[str]] = None + tools: Optional[List[Dict[str, Any]]] = None + tool_choice: Optional[Union[str, Dict[str, str]]] = None + +class ChatCompletionResponse(BaseModel): + """Modèle pour les réponses de complétion""" + id: str + choices: List[Dict[str, Any]] + created: int + model: str + usage: Dict[str, int] + +class OnPremiseLLMClient: + """Client pour interagir avec notre LLM on-premise via MCP""" + + def __init__(self, base_url: str, api_key: str): + """Initialise le client + + Args: + base_url: URL de base de notre API on-premise + api_key: Clé API pour l'authentification + """ + self.base_url = base_url + self.api_key = api_key + self.mcp_tool = MCPTool() + self._setup_resources() + + def _setup_resources(self): + """Configure les ressources MCP""" + # Ressource pour la complétion de chat + chat_resource = MCPResource( + name="chat/completions", + description="Génère des réponses de chat en utilisant notre LLM on-premise", + parameters={ + "messages": List[ChatMessage], + "model": str, + "temperature": float, + "max_tokens": Optional[int], + "top_p": float, + "frequency_penalty": float, + "presence_penalty": float, + "stop": Optional[List[str]], + "tools": Optional[List[Dict[str, Any]]], + "tool_choice": Optional[Union[str, Dict[str, str]]] + } + ) + + # Wrapper la fonction de complétion + wrapped_func = chat_resource(self.chat_completion) + self.mcp_tool.add_resource(chat_resource) + + async def chat_completion( + self, + messages: List[ChatMessage], + model: str = "gpt-3.5-turbo", + temperature: float = 0.7, + max_tokens: Optional[int] = None, + top_p: float = 1.0, + frequency_penalty: float = 0.0, + presence_penalty: float = 0.0, + stop: Optional[List[str]] = None, + tools: Optional[List[Dict[str, Any]]] = None, + tool_choice: Optional[Union[str, Dict[str, str]]] = None + ) -> ChatCompletionResponse: + """Génère une complétion de chat + + Args: + messages: Liste des messages du chat + model: Nom du modèle à utiliser + temperature: Température pour la génération + max_tokens: Nombre maximum de tokens à générer + top_p: Paramètre top_p pour la génération + frequency_penalty: Pénalité de fréquence + presence_penalty: Pénalité de présence + stop: Liste des séquences d'arrêt + tools: Liste des outils disponibles + tool_choice: Choix de l'outil à utiliser + + Returns: + Réponse de complétion + """ + try: + # Préparer la requête + request = ChatCompletionRequest( + messages=messages, + model=model, + temperature=temperature, + max_tokens=max_tokens, + top_p=top_p, + frequency_penalty=frequency_penalty, + presence_penalty=presence_penalty, + stop=stop, + tools=tools, + tool_choice=tool_choice + ) + + # Configurer le client OpenAI avec notre URL on-premise + client = openai.OpenAI( + base_url=self.base_url, + api_key=self.api_key + ) + + # Appeler l'API + response = await client.chat.completions.create( + **request.model_dump(exclude_none=True) + ) + + # Convertir la réponse en notre format + return ChatCompletionResponse( + id=response.id, + choices=[ + { + "message": { + "role": choice.message.role, + "content": choice.message.content, + "tool_calls": [ + { + "id": tool_call.id, + "type": tool_call.type, + "function": { + "name": tool_call.function.name, + "arguments": tool_call.function.arguments + } + } + for tool_call in choice.message.tool_calls + ] if hasattr(choice.message, "tool_calls") else None + }, + "finish_reason": choice.finish_reason + } + for choice in response.choices + ], + created=response.created, + model=response.model, + usage={ + "prompt_tokens": response.usage.prompt_tokens, + "completion_tokens": response.usage.completion_tokens, + "total_tokens": response.usage.total_tokens + } + ) + + except Exception as e: + logger.error(f"Erreur lors de la génération de la complétion: {str(e)}") + raise HTTPException(status_code=500, detail=str(e)) + + async def generate_str( + self, + message: str, + system_prompt: Optional[str] = None, + tools: Optional[List[Dict[str, Any]]] = None, + tool_choice: Optional[Union[str, Dict[str, str]]] = None + ) -> str: + """Génère une réponse textuelle + + Args: + message: Message de l'utilisateur + system_prompt: Prompt système optionnel + tools: Liste des outils disponibles + tool_choice: Choix de l'outil à utiliser + + Returns: + Réponse générée + """ + messages = [] + if system_prompt: + messages.append(ChatMessage(role="system", content=system_prompt)) + messages.append(ChatMessage(role="user", content=message)) + + response = await self.chat_completion( + messages=messages, + tools=tools, + tool_choice=tool_choice + ) + return response.choices[0]["message"]["content"] + + async def generate_structured( + self, + message: str, + response_model: Type[ModelT], + system_prompt: Optional[str] = None, + tools: Optional[List[Dict[str, Any]]] = None, + tool_choice: Optional[Union[str, Dict[str, str]]] = None + ) -> ModelT: + """Génère une réponse structurée + + Args: + message: Message de l'utilisateur + response_model: Modèle Pydantic pour la réponse + system_prompt: Prompt système optionnel + tools: Liste des outils disponibles + tool_choice: Choix de l'outil à utiliser + + Returns: + Réponse structurée + """ + messages = [] + if system_prompt: + messages.append(ChatMessage(role="system", content=system_prompt)) + messages.append(ChatMessage(role="user", content=message)) + + response = await self.chat_completion( + messages=messages, + tools=tools, + tool_choice=tool_choice + ) + content = response.choices[0]["message"]["content"] + return response_model.model_validate_json(content) \ No newline at end of file diff --git a/ragondin/mcp/prompts.py b/ragondin/mcp/prompts.py new file mode 100644 index 000000000..da5b4c7bf --- /dev/null +++ b/ragondin/mcp/prompts.py @@ -0,0 +1,30 @@ +"""Prompts système et configurations pour le client MCP""" + +DEFAULT_SYSTEM_PROMPT = """Tu es un assistant IA expert qui aide les utilisateurs avec leurs questions. +Tu dois toujours répondre en français et de manière professionnelle. +Tu dois être précis et concis dans tes réponses. +Si tu ne connais pas la réponse, dis-le honnêtement.""" + +RAG_SYSTEM_PROMPT = """Tu es un assistant IA expert qui aide les utilisateurs en utilisant le système RAG (Retrieval Augmented Generation). +Tu dois toujours répondre en français et de manière professionnelle. +Tu dois utiliser le contexte fourni pour enrichir tes réponses. +Si le contexte ne contient pas d'informations pertinentes, dis-le honnêtement. +Base tes réponses sur le contexte fourni, mais n'hésite pas à ajouter des informations générales si nécessaire.""" + +DEFAULT_CONFIG = { + "model": "gpt-3.5-turbo", + "temperature": 0.7, + "max_tokens": 1000, + "top_p": 1.0, + "frequency_penalty": 0.0, + "presence_penalty": 0.0 +} + +RAG_CONFIG = { + "model": "gpt-3.5-turbo", + "temperature": 0.5, # Plus bas pour des réponses plus factuelles + "max_tokens": 1000, + "top_p": 0.9, + "frequency_penalty": 0.2, # Pour encourager la diversité + "presence_penalty": 0.2 # Pour encourager la diversité +} \ No newline at end of file diff --git a/ragondin/mcp_api.py b/ragondin/mcp_api.py new file mode 100644 index 000000000..9f72e6abe --- /dev/null +++ b/ragondin/mcp_api.py @@ -0,0 +1,267 @@ +import os +import json +from typing import Dict, Any, List, Optional + +import anthropic +from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks +from pydantic import BaseModel, Field +from starlette.routes import Mount + +from mcp.server.fastmcp import FastMCP + +from .utils.mcp_dependencies import MCPDependencies, get_mcp_dependencies +from .models.indexer import ( + IndexationRequest, DeletionRequest, MetadataUpdateRequest, SearchRequest, + IndexationResult, DeletionResult, MetadataUpdateResult, SearchResult +) + + +# Initialisation de FastAPI +app = FastAPI(title="RAGondin MCP API", description="API RAGondin avec intégration MCP et Anthropic") + +# Initialisation du serveur MCP +mcp_server = FastMCP("RAGondin MCP") + +# Client Anthropic +client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY")) + + +class RAGQueryRequest(BaseModel): + """Modèle de requête pour la génération RAG""" + query: str = Field(..., description="Question de l'utilisateur") + partition: Optional[str] = Field(None, description="Partition spécifique à utiliser") + top_k: int = Field(5, description="Nombre de résultats à récupérer") + similarity_threshold: float = Field(0.75, description="Seuil de similarité") + model: str = Field("claude-3-sonnet-20240229", description="Modèle Anthropic à utiliser") + max_tokens: int = Field(1000, description="Nombre maximum de tokens dans la réponse") + temperature: float = Field(0.7, description="Température pour la génération") + + +class RAGResponse(BaseModel): + """Modèle de réponse pour la génération RAG""" + query: str = Field(..., description="Question originale") + answer: str = Field(..., description="Réponse générative") + sources: List[Dict[str, Any]] = Field(..., description="Sources utilisées") + model: str = Field(..., description="Modèle utilisé") + + +# Montage du serveur MCP sur FastAPI +app.routes.append(Mount("/mcp", app=mcp_server.sse_app())) + + +# Définition des ressources MCP +@mcp_server.resource("indexation://partitions") +async def get_partitions_resource( + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +) -> str: + """Récupère la liste des partitions disponibles""" + partitions = await dependencies.provider.get_partitions() + return json.dumps(partitions, indent=2) + + +@mcp_server.resource("indexation://files/{partition}") +async def get_files_in_partition_resource( + partition: str, + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +) -> str: + """Récupère la liste des fichiers dans une partition""" + files = await dependencies.provider.get_files_in_partition(partition=partition) + return json.dumps(files, indent=2) + + +# Définition des outils MCP +@mcp_server.tool() +async def index_documents( + path: str, + metadata: Optional[Dict[str, Any]] = None, + partition: Optional[str] = None, + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +) -> str: + """Indexe des documents""" + request = IndexationRequest( + path=path, + metadata=metadata or {}, + partition=partition + ) + + result = await dependencies.index_documents(request) + return json.dumps(result.dict(), indent=2) + + +@mcp_server.tool() +async def search_documents( + query: str, + top_k: int = 5, + similarity_threshold: float = 0.80, + partition: Optional[str] = None, + filter: Optional[Dict[str, Any]] = None, + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +) -> str: + """Recherche des documents""" + request = SearchRequest( + query=query, + top_k=top_k, + similarity_threshold=similarity_threshold, + partition=partition, + filter=filter or {} + ) + + results = await dependencies.search_documents(request) + results_dict = [r.dict() for r in results] + return json.dumps(results_dict, indent=2) + + +# Définition des prompts MCP +@mcp_server.prompt() +async def rag_prompt( + query: str, + context_documents: List[Dict[str, Any]], +) -> dict: + """Crée un prompt RAG pour Anthropic""" + # Construction du contexte + context_str = "" + for i, doc in enumerate(context_documents): + context_str += f"\n--- Document {i+1} ---\n" + context_str += f"Contenu: {doc['content']}\n" + context_str += f"Métadonnées: {json.dumps(doc['metadata'])}\n" + + # Construire le prompt pour Claude + return { + "messages": [ + { + "role": "user", + "content": f"""Voici une question de l'utilisateur. Utilise le contexte fourni pour y répondre de façon précise. + +## Question: +{query} + +## Contexte: +{context_str} + +Réponds uniquement en te basant sur les informations du contexte. Si le contexte ne contient pas assez d'informations pour répondre, indique-le clairement.""" + } + ], + "system": "Tu es un assistant IA spécialisé dans la recherche documentaire, appelé RAGondin. Tu fournis des réponses précises basées uniquement sur les documents de contexte qui te sont fournis." + } + + +# Routes FastAPI +@app.post("/rag/query", response_model=RAGResponse) +async def rag_query( + request: RAGQueryRequest, + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +) -> RAGResponse: + """Génère une réponse à une question en utilisant RAG et Anthropic Claude + + Args: + request: Paramètres de la requête + dependencies: Dépendances MCP + + Returns: + Réponse générative avec sources + """ + try: + # Recherche de documents pertinents + search_request = SearchRequest( + query=request.query, + top_k=request.top_k, + similarity_threshold=request.similarity_threshold, + partition=request.partition + ) + + search_results = await dependencies.search_documents(search_request) + + if not search_results: + raise HTTPException( + status_code=404, + detail="Aucun document pertinent trouvé pour cette requête" + ) + + # Préparation du contexte pour le prompt + context_documents = [ + { + "content": r.content, + "metadata": r.metadata, + "score": r.score + } for r in search_results + ] + + # Création du prompt pour Claude + prompt_data = await rag_prompt( + query=request.query, + context_documents=context_documents + ) + + # Appel à Anthropic + message = client.messages.create( + model=request.model, + max_tokens=request.max_tokens, + temperature=request.temperature, + system=prompt_data.get("system", ""), + messages=prompt_data.get("messages", []) + ) + + # Construction de la réponse + return RAGResponse( + query=request.query, + answer=message.content[0].text, + sources=context_documents, + model=request.model + ) + + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Erreur lors du traitement de la requête RAG: {str(e)}" + ) + + +@app.post("/indexation/index", response_model=IndexationResult) +async def api_index_documents( + request: IndexationRequest, + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +) -> IndexationResult: + """Indexe des documents via l'API + + Args: + request: Paramètres d'indexation + dependencies: Dépendances MCP + + Returns: + Résultat de l'indexation + """ + try: + return await dependencies.index_documents(request) + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Erreur lors de l'indexation des documents: {str(e)}" + ) + + +@app.post("/indexation/search", response_model=List[SearchResult]) +async def api_search_documents( + request: SearchRequest, + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +) -> List[SearchResult]: + """Recherche des documents via l'API + + Args: + request: Paramètres de recherche + dependencies: Dépendances MCP + + Returns: + Résultats de la recherche + """ + try: + return await dependencies.search_documents(request) + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"Erreur lors de la recherche: {str(e)}" + ) + + +if __name__ == "__main__": + import uvicorn + uvicorn.run("ragondin.mcp_api:app", host="0.0.0.0", port=8000, reload=True) \ No newline at end of file diff --git a/ragondin/mcp_client.py b/ragondin/mcp_client.py new file mode 100644 index 000000000..034762eb5 --- /dev/null +++ b/ragondin/mcp_client.py @@ -0,0 +1,174 @@ +import os +import json +import asyncio +from typing import Dict, Any, List, Optional + +import anthropic +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client +from mcp.client import MCPClient +from mcp.server.fastmcp import FastMCP + +from .mcp_server import mcp_server + +# Paramètres du serveur MCP +server_params = StdioServerParameters( + command="python", + args=["-m", "ragondin.mcp_server"], + env={"ANTHROPIC_API_KEY": os.environ.get("ANTHROPIC_API_KEY")} +) + +# Client Anthropic +client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY")) + + +async def run_rag_query(query: str, partition: Optional[str] = None) -> str: + """Exécute une requête RAG en utilisant le serveur MCP et Anthropic Claude + + Args: + query: La question de l'utilisateur + partition: Partition spécifique à utiliser (optionnel) + + Returns: + Réponse du modèle + """ + async with stdio_client(server_params) as (read, write): + async with ClientSession(read, write) as session: + # Initialiser la connexion + await session.initialize() + + # Chercher les documents pertinents + filter_args = {} + if partition: + filter_args["partition"] = partition + + results_json = await session.call_tool( + "search_documents", + arguments={ + "query": query, + "top_k": 5, + "similarity_threshold": 0.75, + **({"partition": partition} if partition else {}) + } + ) + + # Parser les résultats + results = json.loads(results_json) + + if not results: + return "Aucun document pertinent trouvé pour cette requête." + + # Obtenir le prompt RAG + prompt_result = await session.get_prompt( + "rag_prompt", + arguments={ + "query": query, + "context_documents": results + } + ) + + # Appeler Anthropic Claude avec le prompt + message = client.messages.create( + model="claude-3-sonnet-20240229", + max_tokens=1000, + system=prompt_result.get("system", ""), + messages=prompt_result.get("messages", []) + ) + + return message.content[0].text + + +async def index_documents_with_mcp( + path: str, + metadata: Optional[Dict[str, Any]] = None, + partition: Optional[str] = None +) -> str: + """Indexe des documents en utilisant le serveur MCP + + Args: + path: Chemin vers les documents à indexer + metadata: Métadonnées à associer (optionnel) + partition: Partition à utiliser (optionnel) + + Returns: + Résultat de l'indexation + """ + async with stdio_client(server_params) as (read, write): + async with ClientSession(read, write) as session: + # Initialiser la connexion + await session.initialize() + + # Indexer les documents + result_json = await session.call_tool( + "index_documents", + arguments={ + "path": path, + **({"metadata": metadata} if metadata else {}), + **({"partition": partition} if partition else {}) + } + ) + + return result_json + + +async def list_partitions() -> List[str]: + """Liste les partitions disponibles + + Returns: + Liste des partitions + """ + async with stdio_client(server_params) as (read, write): + async with ClientSession(read, write) as session: + # Initialiser la connexion + await session.initialize() + + # Lire la ressource des partitions + content, _ = await session.read_resource("indexation://partitions") + return json.loads(content) + + +async def run_openai_query(prompt: str, model: str = "gpt-3.5-turbo", max_tokens: int = 1000) -> str: + """Exécute une requête en utilisant l'API OpenAI via le MCP + Args: + prompt: Le prompt pour l'API OpenAI + model: Le modèle à utiliser + max_tokens: Nombre maximum de tokens dans la réponse + Returns: + Réponse générée par l'API OpenAI + """ + async with MCPClient(mcp_server) as session: + result = await session.call_tool("call_openai_api", prompt=prompt, model=model, max_tokens=max_tokens) + return result + + +async def main(): + """Exemple d'utilisation de l'intégration MCP avec RAGondin et Anthropic Claude""" + # Vérification de la clé API Anthropic + if not os.environ.get("ANTHROPIC_API_KEY"): + print("⚠️ Veuillez définir la variable d'environnement ANTHROPIC_API_KEY") + return + + # Exemple: Indexer un document + print("📑 Indexation d'un document...") + result = await index_documents_with_mcp( + path="./data/sample.txt", + metadata={"source": "exemple", "auteur": "RAGondin"}, + partition="test" + ) + print(f"Résultat de l'indexation: {result}") + + # Exemple: Lister les partitions + print("\n🗂️ Partitions disponibles:") + partitions = await list_partitions() + print(partitions) + + # Exemple: Requête RAG + print("\n❓ Réponse à une question...") + question = "Que contient le document sample.txt?" + answer = await run_rag_query(question, partition="test") + print(f"Question: {question}") + print(f"Réponse: {answer}") + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file diff --git a/ragondin/mcp_server.py b/ragondin/mcp_server.py new file mode 100644 index 000000000..8acf2c894 --- /dev/null +++ b/ragondin/mcp_server.py @@ -0,0 +1,259 @@ +import os +import json +from typing import Dict, List, Any, Optional + +from mcp.server.fastmcp import FastMCP +import openai + +from .models.indexer import ( + IndexationRequest, DeletionRequest, MetadataUpdateRequest, SearchRequest, + IndexationResult, DeletionResult, MetadataUpdateResult, SearchResult +) +from .utils.api_dependencies import MCPDependencies, get_mcp_dependencies +from .utils.config import Config, get_config + + +# Initialisation du serveur MCP +mcp = FastMCP("RAGondin") + +# Création des dépendances +config = get_config() +dependencies = MCPDependencies(config) + +# Configuration de l'API OpenAI +openai.api_key = os.environ.get("OPENAI_API_KEY") + + +@mcp.resource("indexation://partitions") +async def get_partitions_resource() -> str: + """Récupère la liste des partitions disponibles sous forme de ressource""" + partitions = await dependencies.provider.get_partitions() + return json.dumps(partitions, indent=2) + + +@mcp.resource("indexation://files/{partition}") +async def get_files_in_partition_resource(partition: str) -> str: + """Récupère la liste des fichiers dans une partition spécifique""" + files = await dependencies.provider.get_files_in_partition(partition=partition) + return json.dumps(files, indent=2) + + +@mcp.resource("document://{partition}/{file_id}") +async def get_document_chunks_resource(partition: str, file_id: str) -> str: + """Récupère les chunks d'un document spécifique""" + # Simuler une recherche pour trouver tous les chunks d'un document spécifique + search_request = SearchRequest( + query="", # Recherche vide pour récupérer tous les chunks + top_k=100, # Nombre élevé pour récupérer tous les chunks + similarity_threshold=0.0, # Seuil minimal pour tout récupérer + partition=partition, + filter={"file_id": file_id} # Filtre sur l'ID du fichier + ) + + results = await dependencies.search_documents(search_request) + + # Formater les résultats + formatted_chunks = [ + { + "content": r.content, + "metadata": r.metadata + } for r in results + ] + + return json.dumps(formatted_chunks, indent=2) + + +@mcp.tool() +async def index_documents( + path: str, + metadata: Optional[Dict[str, Any]] = None, + partition: Optional[str] = None +) -> str: + """Indexe un ou plusieurs documents + + Args: + path: Chemin ou liste de chemins vers les fichiers à indexer + metadata: Métadonnées à associer aux documents + partition: Partition dans laquelle indexer les documents + + Returns: + Résultat de l'opération d'indexation au format JSON + """ + request = IndexationRequest( + path=path, + metadata=metadata or {}, + partition=partition + ) + + result = await dependencies.index_documents(request) + return json.dumps(result.dict(), indent=2) + + +@mcp.tool() +async def delete_document( + file_id: str, + partition: str +) -> str: + """Supprime un document + + Args: + file_id: ID du fichier à supprimer + partition: Partition contenant le fichier + + Returns: + Résultat de l'opération de suppression au format JSON + """ + request = DeletionRequest( + file_id=file_id, + partition=partition + ) + + result = await dependencies.delete_document(request) + return json.dumps(result.dict(), indent=2) + + +@mcp.tool() +async def update_metadata( + file_id: str, + metadata: Dict[str, Any], + partition: str +) -> str: + """Met à jour les métadonnées d'un document + + Args: + file_id: ID du fichier à mettre à jour + metadata: Nouvelles métadonnées + partition: Partition contenant le fichier + + Returns: + Résultat de l'opération de mise à jour au format JSON + """ + request = MetadataUpdateRequest( + file_id=file_id, + metadata=metadata, + partition=partition + ) + + result = await dependencies.update_metadata(request) + return json.dumps(result.dict(), indent=2) + + +@mcp.tool() +async def search_documents( + query: str, + top_k: int = 5, + similarity_threshold: float = 0.80, + partition: Optional[str] = None, + filter: Optional[Dict[str, Any]] = None +) -> str: + """Recherche des documents + + Args: + query: Requête de recherche + top_k: Nombre de résultats à retourner + similarity_threshold: Seuil de similarité minimal + partition: Partition dans laquelle effectuer la recherche + filter: Filtres supplémentaires + + Returns: + Résultats de recherche au format JSON + """ + request = SearchRequest( + query=query, + top_k=top_k, + similarity_threshold=similarity_threshold, + partition=partition, + filter=filter or {} + ) + + results = await dependencies.search_documents(request) + + # Convertir les résultats en dictionnaires pour la sérialisation JSON + results_dict = [r.dict() for r in results] + return json.dumps(results_dict, indent=2) + + +@mcp.prompt() +async def rag_prompt( + query: str, + context_documents: Optional[List[Dict[str, Any]]] = None +) -> dict: + """Crée un prompt RAG pour Anthropic + + Args: + query: La question de l'utilisateur + context_documents: Documents de contexte (optionnel) + + Returns: + Prompt structuré pour le modèle + """ + # Si les documents de contexte ne sont pas fournis, rechercher dans l'index + if not context_documents: + search_request = SearchRequest( + query=query, + top_k=5, + similarity_threshold=0.80 + ) + + results = await dependencies.search_documents(search_request) + context_documents = [ + { + "content": r.content, + "metadata": r.metadata + } for r in results + ] + + # Construction du contexte + context_str = "" + for i, doc in enumerate(context_documents): + context_str += f"\n--- Document {i+1} ---\n" + context_str += f"Contenu: {doc['content']}\n" + context_str += f"Métadonnées: {json.dumps(doc['metadata'])}\n" + + # Construire le prompt pour Claude + return { + "messages": [ + { + "role": "user", + "content": f"""Voici une question de l'utilisateur. Utilise le contexte fourni pour y répondre de façon précise. + +## Question: +{query} + +## Contexte: +{context_str} + +Réponds uniquement en te basant sur les informations du contexte. Si le contexte ne contient pas assez d'informations pour répondre, indique-le clairement.""" + } + ], + "system": "Tu es un assistant IA spécialisé dans la recherche documentaire, appelé RAGondin. Tu fournis des réponses précises basées uniquement sur les documents de contexte qui te sont fournis." + } + + +@mcp.tool() +async def call_openai_api( + prompt: str, + model: str = "gpt-3.5-turbo", + max_tokens: int = 1000, +) -> str: + """Appelle l'API OpenAI pour générer une réponse + Args: + prompt: Le prompt pour l'API OpenAI + model: Le modèle à utiliser + max_tokens: Nombre maximum de tokens dans la réponse + Returns: + Réponse générée par l'API OpenAI + """ + try: + response = openai.ChatCompletion.create( + model=model, + messages=[{"role": "user", "content": prompt}], + max_tokens=max_tokens, + ) + return response.choices[0].message.content + except Exception as e: + return f"Erreur lors de l'appel à l'API OpenAI: {str(e)}" + + +if __name__ == "__main__": + mcp.run() \ No newline at end of file diff --git a/ragondin/models/indexer.py b/ragondin/models/indexer.py index 907c86a8c..56fc94e4c 100644 --- a/ragondin/models/indexer.py +++ b/ragondin/models/indexer.py @@ -1,8 +1,62 @@ -from typing import Optional +from typing import Dict, List, Optional, Union, Any +from pydantic import BaseModel, Field +from langchain_core.documents.base import Document -from pydantic import BaseModel +class IndexationRequest(BaseModel): + """Modèle de requête pour l'indexation de documents""" + path: Union[str, List[str]] = Field(..., description="Chemin ou liste de chemins vers les fichiers à indexer") + metadata: Dict[str, Any] = Field(default_factory=dict, description="Métadonnées à associer aux documents") + partition: Optional[str] = Field(None, description="Partition dans laquelle indexer les documents") +class DeletionRequest(BaseModel): + """Modèle de requête pour la suppression de documents""" + file_id: str = Field(..., description="ID du fichier à supprimer") + partition: str = Field(..., description="Partition contenant le fichier") + +class MetadataUpdateRequest(BaseModel): + """Modèle de requête pour la mise à jour de métadonnées""" + file_id: str = Field(..., description="ID du fichier à mettre à jour") + metadata: Dict[str, Any] = Field(..., description="Nouvelles métadonnées à associer") + partition: str = Field(..., description="Partition contenant le fichier") class SearchRequest(BaseModel): - query: str - top_k: Optional[int] = 5 # default to 5 if not provided + """Modèle de requête pour la recherche de documents""" + query: str = Field(..., description="Requête de recherche") + top_k: int = Field(default=5, description="Nombre de résultats à retourner") + similarity_threshold: float = Field(default=0.80, description="Seuil de similarité minimal") + partition: Optional[Union[str, List[str]]] = Field(None, description="Partition(s) dans laquelle(s) effectuer la recherche") + filter: Dict[str, Any] = Field(default_factory=dict, description="Filtres supplémentaires") + +class SearchResult(BaseModel): + """Modèle de résultat pour la recherche de documents""" + content: str = Field(..., description="Contenu du document") + metadata: Dict[str, Any] = Field(..., description="Métadonnées associées au document") + score: float = Field(..., description="Score de similarité") + + @classmethod + def from_document(cls, doc: Document, score: Optional[float] = None) -> "SearchResult": + """Crée un résultat de recherche à partir d'un document LangChain""" + return cls( + content=doc.page_content, + metadata=doc.metadata, + score=score or doc.metadata.get("score", 0.0) + ) + +class IndexationResult(BaseModel): + """Modèle de résultat pour l'indexation de documents""" + success: bool = Field(..., description="Indique si l'opération a réussi") + message: str = Field(..., description="Message de résultat") + file_count: int = Field(default=0, description="Nombre de fichiers indexés") + chunk_count: int = Field(default=0, description="Nombre de chunks indexés") + +class DeletionResult(BaseModel): + """Modèle de résultat pour la suppression de documents""" + success: bool = Field(..., description="Indique si l'opération a réussi") + message: str = Field(..., description="Message de résultat") + deleted_points: int = Field(default=0, description="Nombre de points supprimés") + +class MetadataUpdateResult(BaseModel): + """Modèle de résultat pour la mise à jour de métadonnées""" + success: bool = Field(..., description="Indique si l'opération a réussi") + message: str = Field(..., description="Message de résultat") + updated_chunks: int = Field(default=0, description="Nombre de chunks mis à jour") diff --git a/ragondin/models/openai.py b/ragondin/models/openai.py index 23e600f9c..df2602463 100644 --- a/ragondin/models/openai.py +++ b/ragondin/models/openai.py @@ -1,5 +1,4 @@ -from typing import Dict, List, Literal, Optional - +from typing import List, Optional, Dict, Any, Union from pydantic import BaseModel, Field @@ -7,19 +6,26 @@ class OpenAIMessage(BaseModel): """Modèle représentant un message dans l'API OpenAI.""" - role: Literal["user", "assistant", "system"] + role: str content: str + tool_calls: Optional[List[Dict[str, Any]]] = None class OpenAICompletionRequest(BaseModel): """Modèle représentant une requête de complétion pour l'API OpenAI.""" - model: str = Field(..., description="model name") - messages: List[OpenAIMessage] - temperature: Optional[float] = Field(0.7) - top_p: Optional[float] = Field(1.0) - stream: Optional[bool] = Field(False) - max_tokens: Optional[int] = Field(None) + model: str + messages: List[ChatMsg] + temperature: Optional[float] = 0.7 + top_p: Optional[float] = 1.0 + n: Optional[int] = 1 + stream: Optional[bool] = False + stop: Optional[List[str]] = None + max_tokens: Optional[int] = None + presence_penalty: Optional[float] = 0.0 + frequency_penalty: Optional[float] = 0.0 + tools: Optional[List[Tool]] = None + tool_choice: Optional[Union[str, ToolChoice]] = None class OpenAICompletionChoice(BaseModel): @@ -27,7 +33,7 @@ class OpenAICompletionChoice(BaseModel): index: int message: OpenAIMessage - finish_reason: str + finish_reason: Optional[str] = None class OpenAIUsage(BaseModel): @@ -53,7 +59,7 @@ class OpenAICompletionChunkChoice(BaseModel): """Modèle représentant un choix de segment de complétion en streaming dans l'API OpenAI.""" index: int - delta: Dict[str, str] + delta: Dict[str, Any] finish_reason: Optional[str] = None @@ -65,3 +71,30 @@ class OpenAICompletionChunk(BaseModel): created: int model: str choices: List[OpenAICompletionChunkChoice] + + +class ChatMsg(BaseModel): + """Modèle pour un message de chat""" + role: str + content: str + + +class Tool(BaseModel): + """Modèle pour un outil OpenAI""" + type: str = Field(default="function") + function: Dict[str, Any] + + +class ToolChoice(BaseModel): + """Modèle pour le choix d'un outil OpenAI""" + type: str = Field(default="function") + function: Dict[str, str] + + +# Mapping des rôles vers les types de messages +mapping = { + "user": ChatMsg, + "assistant": ChatMsg, + "system": ChatMsg, + "tool": ChatMsg +} diff --git a/ragondin/routers/collection.py b/ragondin/routers/collection.py deleted file mode 100644 index de57f80f5..000000000 --- a/ragondin/routers/collection.py +++ /dev/null @@ -1,15 +0,0 @@ -from components import Indexer -from config.config import load_config -from fastapi import APIRouter, Depends -from utils.dependencies import get_indexer - -# load config -config = load_config() -DATA_DIR = config.paths.data_dir -# Create an APIRouter instance -router = APIRouter() - - -@router.get("/collections/", summary="Get existant collections") -async def get_collections(indexer: Indexer = Depends(get_indexer)) -> list[str]: - return await indexer.vectordb.get_collections() diff --git a/ragondin/routers/collection_api.py b/ragondin/routers/collection_api.py new file mode 100644 index 000000000..ed28ed52d --- /dev/null +++ b/ragondin/routers/collection_api.py @@ -0,0 +1,22 @@ +from typing import List +from fastapi import APIRouter, Depends, HTTPException +from fastapi.responses import JSONResponse + +from ..utils.api_dependencies import get_api_dependencies, APIDependencies + +router = APIRouter( + prefix="/collections", + tags=["collections"], + responses={404: {"description": "Collection non trouvée"}}, +) + +@router.get("/") +async def get_collections( + dependencies: APIDependencies = Depends(get_api_dependencies) +) -> JSONResponse: + """Récupère toutes les collections existantes""" + try: + collections = await dependencies.get_collections() + return JSONResponse(content={"collections": collections}, status_code=200) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) \ No newline at end of file diff --git a/ragondin/routers/indexer_api.py b/ragondin/routers/indexer_api.py new file mode 100644 index 000000000..6a29d5ed3 --- /dev/null +++ b/ragondin/routers/indexer_api.py @@ -0,0 +1,88 @@ +import json +from pathlib import Path +from typing import List, Dict, Any, Optional +from fastapi import APIRouter, Depends, HTTPException, File, Form, UploadFile, Request, Response, status +from fastapi.responses import JSONResponse + +from ..models.indexer import ( + IndexationRequest, DeletionRequest, MetadataUpdateRequest, SearchRequest, + IndexationResult, DeletionResult, MetadataUpdateResult, SearchResult +) +from ..utils.mcp_dependencies import get_mcp_dependencies, MCPDependencies +from ..utils.dependencies import vectordb +from config.config import load_config + +# Charger la configuration +config = load_config() +DATA_DIR = config.paths.data_dir + +router = APIRouter( + prefix="/partition", + tags=["indexation"], + responses={404: {"description": "Opération non trouvée"}}, +) + +@router.post("/{partition}/file/{file_id}") +async def add_file( + request: Request, + partition: str, + file_id: str, + file: UploadFile = File(...), + metadata: Optional[Any] = Form(None), + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +): + """Ajoute un fichier dans une partition spécifique""" + result = await dependencies.add_file(partition, file_id, file, metadata) + return JSONResponse( + status_code=status.HTTP_201_CREATED, + content=result + ) + +@router.delete("/{partition}/file/{file_id}") +async def delete_file( + partition: str, + file_id: str, + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +): + """Supprime un fichier d'une partition spécifique""" + await dependencies.delete_file(partition, file_id) + return Response(status_code=status.HTTP_204_NO_CONTENT) + +@router.put("/{partition}/file/{file_id}") +async def put_file( + request: Request, + partition: str, + file_id: str, + file: UploadFile = File(...), + metadata: Optional[Any] = Form(None), + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +): + """Met à jour un fichier dans une partition spécifique""" + result = await dependencies.update_file(partition, file_id, file, metadata) + return JSONResponse( + status_code=status.HTTP_202_ACCEPTED, + content=result + ) + +@router.patch("/{partition}/file/{file_id}") +async def patch_file( + partition: str, + file_id: str, + metadata: Optional[Any] = Form(None), + dependencies: MCPDependencies = Depends(get_mcp_dependencies) +): + """Met à jour les métadonnées d'un fichier""" + result = await dependencies.update_metadata(partition, file_id, metadata) + return JSONResponse( + status_code=status.HTTP_200_OK, + content=result + ) + +@router.post("/sync-db/") +async def sync_db(dependencies: MCPDependencies = Depends(get_mcp_dependencies)): + """Synchronise la base de données avec les fichiers""" + result = await dependencies.sync_database() + return JSONResponse( + content=result, + status_code=200 + ) \ No newline at end of file diff --git a/ragondin/routers/openai_api.py b/ragondin/routers/openai_api.py new file mode 100644 index 000000000..8254d02b9 --- /dev/null +++ b/ragondin/routers/openai_api.py @@ -0,0 +1,46 @@ +from typing import Optional, List, Dict, Any, Union +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import StreamingResponse +from loguru import logger +from pydantic import BaseModel, Field +from omegaconf import OmegaConf + +from ..models.openai import ( + OpenAICompletionRequest, OpenAICompletion, OpenAICompletionChoice, + OpenAICompletionChunk, OpenAICompletionChunkChoice, OpenAIMessage, + OpenAIUsage, ChatMsg, mapping, Tool, ToolChoice +) +from ..utils.api_dependencies import get_api_dependencies +from config import load_config + +router = APIRouter(prefix="/v1", tags=["openai"]) + +@router.post("/chat/completions") +async def chat_completions( + request: OpenAICompletionRequest, + static_base_url: str, + app_state: Any, + api_deps = Depends(get_api_dependencies) +): + """Endpoint pour la complétion de chat OpenAI + + Args: + request: Requête de complétion + static_base_url: URL de base pour les ressources statiques + app_state: État de l'application + api_deps: Dépendances de l'API + + Returns: + Réponse de complétion (streaming ou non) + """ + try: + # Appeler le service de complétion + return await api_deps.chat_completions( + request=request, + static_base_url=static_base_url, + app_state=app_state + ) + + except Exception as e: + logger.error(f"Erreur lors de la complétion de chat: {str(e)}") + raise HTTPException(status_code=500, detail=str(e)) \ No newline at end of file diff --git a/ragondin/routers/search_api.py b/ragondin/routers/search_api.py new file mode 100644 index 000000000..a1c51e344 --- /dev/null +++ b/ragondin/routers/search_api.py @@ -0,0 +1,92 @@ +from typing import List, Optional +from fastapi import Depends, HTTPException, Request +from fastapi.responses import JSONResponse + +from ..utils.api_dependencies import get_api_dependencies, APIDependencies +from ..utils.router_manager import RouterManager, MCPToolRoute + +# Créer une instance du gestionnaire de routes +router_manager = RouterManager(__file__) + +async def search_multiple_partitions( + request: Request, + partitions: Optional[List[str]] = None, + text: str = None, + top_k: int = 5, + dependencies: APIDependencies = Depends(get_api_dependencies) +) -> JSONResponse: + """Recherche dans plusieurs partitions""" + try: + results = await dependencies.search_multiple_partitions( + request=request, + partitions=partitions, + text=text, + top_k=top_k + ) + return JSONResponse(content=results, status_code=200) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) +@MCPToolRoute(request_type="GET", + path="/partition/{partition}", + router=router, + mcp_server=mcp_server) +async def search_one_partition( + request: Request, + partition: str, + text: str, + top_k: int = 5, + dependencies: APIDependencies = Depends(get_api_dependencies) +) -> JSONResponse: + """Recherche dans une partition spécifique""" + try: + results = await dependencies.search_one_partition( + request=request, + partition=partition, + text=text, + top_k=top_k + ) + return JSONResponse(content=results, status_code=200) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + +async def search_file( + request: Request, + partition: str, + file_id: str, + query: str, + top_k: int = 5, + dependencies: APIDependencies = Depends(get_api_dependencies) +) -> JSONResponse: + """Recherche dans un fichier spécifique""" + try: + results = await dependencies.search_file( + request=request, + partition=partition, + file_id=file_id, + query=query, + top_k=top_k + ) + return JSONResponse(content=results, status_code=200) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + +async def get_extract( + extract_id: str, + dependencies: APIDependencies = Depends(get_api_dependencies) +) -> JSONResponse: + """Récupère un extrait par son ID""" + try: + results = await dependencies.get_extract(extract_id=extract_id) + return JSONResponse(content=results, status_code=200) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + +# Ajouter les routes au gestionnaire +router_manager.append("get", "/", search_multiple_partitions) +router_manager.append("get", "/partition/{partition}", search_one_partition) +router_manager.append("get", "/partition/{partition}/file/{file_id}", search_file) +router_manager.append("get", "/{extract_id}", get_extract) + +# Configurer le router et les ressources MCP +router = router_manager.setup_router() +mcp_tool = router_manager.setup_mcp() \ No newline at end of file diff --git a/ragondin/utils/api_dependencies.py b/ragondin/utils/api_dependencies.py new file mode 100644 index 000000000..b01c6d06c --- /dev/null +++ b/ragondin/utils/api_dependencies.py @@ -0,0 +1,914 @@ +from typing import Optional, Dict, Any, Callable, List, Union +from functools import lru_cache +from fastapi import Depends, HTTPException, status, UploadFile, Form, Request +from pathlib import Path +import json +from loguru import logger +import openai +from fastapi.responses import StreamingResponse +import uuid +import time +from mcp import MCPTool, MCPResource, MCPPrompt +from mcp.types import CallToolRequest, CallToolRequestParams, CallToolResult +from pydantic_core import from_json + +from ..models.indexer import ( + IndexationRequest, DeletionRequest, MetadataUpdateRequest, SearchRequest, + IndexationResult, DeletionResult, MetadataUpdateResult, SearchResult +) +from ..models.openai import ( + OpenAICompletionRequest, OpenAICompletion, OpenAICompletionChoice, + OpenAICompletionChunk, OpenAICompletionChunkChoice, OpenAIMessage, + OpenAIUsage, ChatMsg, mapping +) +from ..controllers.indexer_controller import IndexerController +from ..components.indexer.indexer import Indexer +from config import load_config +from omegaconf import OmegaConf +from .dependencies import vectordb + +def source2url(s: dict, static_base_url: str): + s["url"] = f"{static_base_url}/{s['sub_url_path']}" + s.pop("source") + s.pop("sub_url_path") + return s + +def get_indexer(config: OmegaConf = Depends(load_config)) -> Indexer: + """Dépendance pour obtenir une instance d'Indexer + + Args: + config: Configuration du système + + Returns: + Instance d'Indexer configurée + """ + return Indexer(config=config, logger=logger) + + +def get_indexer_controller(config: OmegaConf = Depends(load_config)) -> IndexerController: + """Dépendance pour obtenir une instance du contrôleur d'indexation + + Args: + config: Configuration du système + + Returns: + Instance du contrôleur d'indexation + """ + return IndexerController(config=config) + + +class APIDependencies: + """Classe qui gère les dépendances de l'API et les opérations d'API""" + + def __init__(self, config: OmegaConf): + """Initialise les dépendances avec une configuration + + Args: + config: Configuration du système + """ + self.config = config + self.controller = IndexerController(config=config) + self.data_dir = Path(config.paths.data_dir) + self.mcp_tool = MCPTool() + self._setup_mcp_resources() + + def _setup_mcp_resources(self): + """Configure les ressources MCP""" + # Ressource pour lister les outils + tools_resource = MCPResource( + name="tools", + description="Liste tous les outils disponibles", + parameters={} + ) + self.mcp_tool.add_resource(tools_resource) + + # Prompt pour la contextualisation des outils + contextualize_prompt = MCPPrompt( + name="contextualize_tools", + description="Prompt pour contextualiser la question utilisateur avec les outils disponibles", + template="""Tu es un assistant IA qui peut utiliser des outils pour répondre aux questions. +Voici les outils disponibles: + +{available_tools} + +Historique du chat: +{chat_history} + +Question de l'utilisateur: {question} + +Analyse la question et décide si l'utilisation d'outils est nécessaire. +Si oui, structure ta réponse pour utiliser les outils appropriés. +Si non, réponds directement à la question. +Réponds toujours en français.""", + parameters={ + "question": str, + "chat_history": List[Dict[str, str]], + "available_tools": List[Dict[str, Any]] + } + ) + self.mcp_tool.add_prompt(contextualize_prompt) + + async def add_file( + self, + partition: str, + file_id: str, + file: UploadFile, + metadata: Optional[Any] = Form(None) + ) -> Dict[str, Any]: + """Ajoute un fichier dans une partition spécifique + + Args: + partition: Nom de la partition + file_id: Identifiant du fichier + file: Fichier à uploader + metadata: Métadonnées optionnelles + + Returns: + Réponse JSON avec le statut de l'opération + """ + # Vérifier si le fichier existe déjà + if vectordb.file_exists(file_id, partition): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"File '{file_id}' already exists in partition {partition}", + ) + + # Charger les métadonnées + try: + metadata = metadata or "{}" + metadata = json.loads(metadata) + except json.JSONDecodeError: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Invalid JSON in metadata" + ) + if not isinstance(metadata, dict): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Metadata must be a dictionary", + ) + + # Ajouter file_id aux métadonnées + metadata["file_id"] = file_id + + # Créer un répertoire temporaire pour stocker les fichiers + save_dir = self.data_dir + save_dir.mkdir(parents=True, exist_ok=True) + + # Sauvegarder le fichier uploadé + file_path = save_dir / Path(file.filename).name + try: + with open(file_path, "wb") as buffer: + buffer.write(await file.read()) + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to save file: {str(e)}", + ) + + # Indexer le fichier + try: + result = await self.controller.index_documents( + IndexationRequest( + path=file_path, + metadata=metadata, + partition=partition + ) + ) + if not result.success: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=result.message + ) + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Indexing error: {str(e)}", + ) + + return { + "message": f"File '{file_id}' successfully indexed in partition '{partition}'" + } + + async def delete_file(self, partition: str, file_id: str) -> None: + """Supprime un fichier d'une partition spécifique + + Args: + partition: Nom de la partition + file_id: Identifiant du fichier à supprimer + """ + try: + result = await self.controller.delete_document( + DeletionRequest( + file_id=file_id, + partition=partition + ) + ) + if not result.success: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=result.message + ) + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Error while deleting file '{file_id}': {str(e)}", + ) + + async def update_file( + self, + partition: str, + file_id: str, + file: UploadFile, + metadata: Optional[Any] = Form(None) + ) -> Dict[str, Any]: + """Met à jour un fichier dans une partition spécifique + + Args: + partition: Nom de la partition + file_id: Identifiant du fichier + file: Nouveau fichier + metadata: Nouvelles métadonnées optionnelles + + Returns: + Réponse JSON avec le statut de l'opération + """ + # Vérifier l'existence du fichier + if not vectordb.file_exists(file_id, partition): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"File '{file_id}' not found in partition '{partition}'.", + ) + + # Supprimer l'ancien fichier + try: + result = await self.controller.delete_document( + DeletionRequest( + file_id=file_id, + partition=partition + ) + ) + if not result.success: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=result.message + ) + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to delete existing file: {str(e)}", + ) + + # Parser les métadonnées + try: + metadata = metadata or "{}" + metadata = json.loads(metadata) + if not isinstance(metadata, dict): + raise ValueError("Metadata is not a dictionary.") + except Exception as e: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid metadata: {str(e)}", + ) + + metadata["file_id"] = file_id + + # Sauvegarder le nouveau fichier + save_dir = self.data_dir + save_dir.mkdir(parents=True, exist_ok=True) + file_path = save_dir / Path(file.filename).name + try: + with open(file_path, "wb") as buffer: + buffer.write(await file.read()) + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to save file: {str(e)}", + ) + + # Indexer le nouveau fichier + try: + result = await self.controller.index_documents( + IndexationRequest( + path=file_path, + metadata=metadata, + partition=partition + ) + ) + if not result.success: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=result.message + ) + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Indexing error: {str(e)}", + ) + + return { + "message": f"File '{file_id}' successfully updated in partition '{partition}'" + } + + async def update_metadata( + self, + partition: str, + file_id: str, + metadata: Optional[Any] = Form(None) + ) -> Dict[str, Any]: + """Met à jour les métadonnées d'un fichier + + Args: + partition: Nom de la partition + file_id: Identifiant du fichier + metadata: Nouvelles métadonnées + + Returns: + Réponse JSON avec le statut de l'opération + """ + # Vérifier l'existence du fichier + if not vectordb.file_exists(file_id, partition): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"File '{file_id}' not found in partition '{partition}'.", + ) + + # Parser les métadonnées + try: + metadata = metadata or "{}" + metadata = json.loads(metadata) + if not isinstance(metadata, dict): + raise ValueError("Metadata must be a JSON object.") + except Exception as e: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid metadata: {str(e)}", + ) + + metadata["file_id"] = file_id + + # Mettre à jour les métadonnées + try: + result = await self.controller.update_metadata( + MetadataUpdateRequest( + file_id=file_id, + metadata=metadata, + partition=partition + ) + ) + if not result.success: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=result.message + ) + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to update metadata: {str(e)}", + ) + + return { + "message": f"Metadata for file '{file_id}' successfully updated." + } + + async def sync_database(self) -> Dict[str, Any]: + """Synchronise la base de données avec les fichiers + + Returns: + Réponse JSON avec le résumé de la synchronisation + """ + try: + if not self.data_dir.exists(): + raise HTTPException(status_code=400, detail="DATA_DIR does not exist") + + sync_summary = {} + + for collection_path in self.data_dir.iterdir(): + if collection_path.is_dir(): # S'assurer que c'est un dossier de collection + collection_name = collection_path.name + up_to_date_files = [] + missing_files = [] + + for file_path in collection_path.iterdir(): + if file_path.is_file() and file_path.suffix != ".md": + if vectordb.file_exists(file_path.name, collection_name): + up_to_date_files.append(file_path.name) + else: + missing_files.append(file_path.name) + await self.controller.index_documents( + IndexationRequest( + path=file_path, + metadata={}, + partition=collection_name + ) + ) + + if not missing_files: + logger.info(f"Collection '{collection_name}' is already up to date.") + else: + logger.info(f"Collection '{collection_name}' updated. Added files: {missing_files}") + + sync_summary[collection_name] = { + "up_to_date": up_to_date_files, + "added": missing_files, + } + + return { + "message": "Database sync completed.", + "details": sync_summary + } + + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + async def search_multiple_partitions( + self, + request: Request, + partitions: Optional[List[str]] = None, + text: str = None, + top_k: int = 5 + ) -> Dict[str, Any]: + """Recherche dans plusieurs partitions + + Args: + request: Requête FastAPI pour la génération des URLs + partitions: Liste des partitions à rechercher + text: Texte à rechercher + top_k: Nombre de résultats à retourner + + Returns: + Résultats de la recherche avec liens HATEOAS + """ + try: + results = await self.controller.search_documents( + SearchRequest( + query=text, + top_k=top_k, + partition=partitions + ) + ) + + # Construire la réponse HATEOAS + documents = [ + { + "link": str( + request.url_for("get_extract", extract_id=doc.metadata["_id"]) + ) + } + for doc in results + ] + + return {"Documents": documents} + + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + async def search_one_partition( + self, + request: Request, + partition: str, + text: str, + top_k: int = 5 + ) -> Dict[str, Any]: + """Recherche dans une partition spécifique + + Args: + request: Requête FastAPI pour la génération des URLs + partition: Partition à rechercher + text: Texte à rechercher + top_k: Nombre de résultats à retourner + + Returns: + Résultats de la recherche avec liens HATEOAS + """ + try: + results = await self.controller.search_documents( + SearchRequest( + query=text, + top_k=top_k, + partition=partition + ) + ) + + # Construire la réponse HATEOAS + documents = [ + { + "link": str( + request.url_for("get_extract", extract_id=doc.metadata["_id"]) + ) + } + for doc in results + ] + + return {"Documents": documents} + + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + async def search_file( + self, + request: Request, + partition: str, + file_id: str, + query: str, + top_k: int = 5 + ) -> Dict[str, Any]: + """Recherche dans un fichier spécifique + + Args: + request: Requête FastAPI pour la génération des URLs + partition: Partition contenant le fichier + file_id: Identifiant du fichier + query: Texte à rechercher + top_k: Nombre de résultats à retourner + + Returns: + Résultats de la recherche avec liens HATEOAS + """ + try: + results = await self.controller.search_documents( + SearchRequest( + query=query, + top_k=top_k, + partition=partition, + filter={"file_id": file_id} + ) + ) + + # Construire la réponse HATEOAS + documents = [ + { + "link": str( + request.url_for("get_extract", extract_id=doc.metadata["_id"]) + ) + } + for doc in results + ] + + return {"Documents": documents} + + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + async def get_extract(self, extract_id: str) -> Dict[str, Any]: + """Récupère un extrait par son ID + + Args: + extract_id: Identifiant de l'extrait + + Returns: + Contenu et métadonnées de l'extrait + """ + try: + doc = vectordb.get_chunk_by_id(extract_id) + return { + "page_content": doc.page_content, + "metadata": doc.metadata + } + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + async def get_collections(self) -> List[str]: + """Récupère toutes les collections existantes + + Returns: + Liste des noms des collections + """ + try: + return await self.controller.vectordb.get_collections() + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + async def get_available_tools(self) -> List[Dict[str, Any]]: + """Récupère la liste des outils disponibles via MCP + + Returns: + Liste des outils avec leurs descriptions et paramètres + """ + try: + tools = await self.mcp_tool.get_tools() + return tools + except Exception as e: + logger.error(f"Erreur lors de la récupération des outils: {str(e)}") + return [] + + async def call_tool(self, tool_call: Dict[str, Any]) -> Any: + """Appelle un outil via MCP + + Args: + tool_call: Appel d'outil à exécuter + + Returns: + Résultat de l'appel d'outil + """ + try: + request = CallToolRequest( + method="tools/call", + params=CallToolRequestParams( + name=tool_call["function"]["name"], + arguments=from_json(tool_call["function"]["arguments"], allow_partial=True) + ) + ) + result = await self.mcp_tool.call_tool(request) + return result + except Exception as e: + logger.error(f"Erreur lors de l'appel de l'outil: {str(e)}") + raise HTTPException(status_code=500, detail=str(e)) + + async def get_contextualized_prompt( + self, + question: str, + chat_history: List[Dict[str, str]], + available_tools: List[Dict[str, Any]] + ) -> str: + """Récupère le prompt contextualisé via MCP + + Args: + question: Question de l'utilisateur + chat_history: Historique du chat + available_tools: Liste des outils disponibles + + Returns: + Prompt contextualisé + """ + try: + # Formater l'historique du chat + formatted_history = "\n".join([ + f"{msg['role']}: {msg['content']}" + for msg in chat_history + ]) + + # Formater les outils disponibles + formatted_tools = json.dumps(available_tools, indent=2) + + # Récupérer le prompt formaté + prompt = await self.mcp_tool.get_prompt( + "contextualize_tools", + { + "question": question, + "chat_history": formatted_history, + "available_tools": formatted_tools + } + ) + return prompt + except Exception as e: + logger.error(f"Erreur lors de la récupération du prompt contextualisé: {str(e)}") + return question # En cas d'erreur, retourner la question brute + + async def chat_completions( + self, + request: OpenAICompletionRequest, + static_base_url: str, + app_state: Any + ) -> StreamingResponse | OpenAICompletion: + """Gère la logique de complétion de chat OpenAI avec RAG + + Args: + request: Requête de complétion OpenAI + static_base_url: URL de base pour les ressources statiques + app_state: État de l'application + + Returns: + Réponse de l'API OpenAI (streaming ou non) + """ + # Récupérer le dernier message utilisateur + user_messages = [msg for msg in request.messages if msg.role == "user"] + if not user_messages: + raise HTTPException( + status_code=400, detail="At least one user message is required" + ) + + new_user_input = user_messages[-1].content + + # Convertir l'historique des messages + chat_history = [] + for msg in request.messages[:-1]: # Exclure le dernier message utilisateur + if msg.role in ["user", "assistant"]: + chat_history.append(ChatMsg(role=msg.role, content=msg.content)) + + msgs = None + if chat_history: + msgs = [ + mapping[chat_msg.role](content=chat_msg.content) + for chat_msg in chat_history + ] + + # Récupérer les outils disponibles + available_tools = await self.get_available_tools() + + # Récupérer le prompt contextualisé + contextualized_prompt = await self.get_contextualized_prompt( + question=new_user_input, + chat_history=[{"role": msg.role, "content": msg.content} for msg in chat_history], + available_tools=available_tools + ) + + # Exécuter le pipeline RAG avec le prompt contextualisé + answer_stream, context, sources = await app_state.ragpipe.run( + partition=["all"], question=contextualized_prompt, chat_history=msgs + ) + + # Gérer les sources + sources = list(map(lambda x: source2url(x, static_base_url), sources)) + src_json = json.dumps(sources) + + # Créer l'ID de réponse + response_id = f"chatcmpl-{str(uuid.uuid4())}" + created_time = int(time.time()) + model_name = app_state.model_name + + if request.stream: + # Réponse streaming compatible OpenAI + async def stream_response(): + full_response = "" + chunk = OpenAICompletionChunk( + id=response_id, + created=created_time, + model=model_name, + choices=[ + OpenAICompletionChunkChoice( + index=0, delta={"role": "assistant"}, finish_reason=None + ) + ], + ) + yield f"data: {chunk.model_dump_json()}\n\n" + + # Envoyer les tokens un par un + async for token in answer_stream: + full_response += token.content + chunk = OpenAICompletionChunk( + id=response_id, + created=created_time, + model=model_name, + choices=[ + OpenAICompletionChunkChoice( + index=0, + delta={"content": token.content}, + finish_reason=None, + ) + ], + ) + yield f"data: {chunk.model_dump_json()}\n\n" + + # Vérifier si la réponse contient des appels d'outils + if hasattr(answer_stream, "tool_calls") and answer_stream.tool_calls: + # Gérer les appels d'outils + tool_results = [] + for tool_call in answer_stream.tool_calls: + result = await self.call_tool(tool_call) + tool_results.append((tool_call.id, result)) + + # Envoyer les appels d'outils + chunk = OpenAICompletionChunk( + id=response_id, + created=created_time, + model=model_name, + choices=[ + OpenAICompletionChunkChoice( + index=0, + delta={ + "tool_calls": [ + { + "id": tool_call.id, + "type": tool_call.type, + "function": { + "name": tool_call.function.name, + "arguments": tool_call.function.arguments + } + } + for tool_call in answer_stream.tool_calls + ] + }, + finish_reason="tool_calls", + ) + ], + ) + yield f"data: {chunk.model_dump_json()}\n\n" + + # Envoyer les résultats des outils + for tool_id, result in tool_results: + chunk = OpenAICompletionChunk( + id=response_id, + created=created_time, + model=model_name, + choices=[ + OpenAICompletionChunkChoice( + index=0, + delta={ + "role": "tool", + "tool_call_id": tool_id, + "content": str(result) + }, + finish_reason=None, + ) + ], + ) + yield f"data: {chunk.model_dump_json()}\n\n" + else: + # Envoyer le chunk final sans outils + chunk = OpenAICompletionChunk( + id=response_id, + created=created_time, + model=model_name, + choices=[ + OpenAICompletionChunkChoice(index=0, delta={}, finish_reason="stop") + ], + ) + yield f"data: {chunk.model_dump_json()}\n\n" + + yield "data: [DONE]\n\n" + + return StreamingResponse( + stream_response(), + media_type="text/event-stream", + headers={"X-Metadata-Sources": src_json}, + ) + else: + # Réponse non streaming + full_response = "" + async for token in answer_stream: + full_response += token.content + + # Vérifier si la réponse contient des appels d'outils + if hasattr(answer_stream, "tool_calls") and answer_stream.tool_calls: + # Récupérer les outils disponibles + available_tools = await self.get_available_tools() + + # Gérer les appels d'outils + tool_results = [] + for tool_call in answer_stream.tool_calls: + result = await self.call_tool(tool_call) + tool_results.append((tool_call.id, result)) + + # Créer la réponse avec les appels d'outils + completion = OpenAICompletion( + id=response_id, + created=created_time, + model=model_name, + choices=[ + OpenAICompletionChoice( + index=0, + message=OpenAIMessage( + role="assistant", + content=full_response, + tool_calls=[ + { + "id": tool_call.id, + "type": tool_call.type, + "function": { + "name": tool_call.function.name, + "arguments": tool_call.function.arguments + } + } + for tool_call in answer_stream.tool_calls + ] + ), + finish_reason="tool_calls", + ) + ], + usage=OpenAIUsage( + prompt_tokens=100, # Valeurs approximatives + completion_tokens=len(full_response.split()) * 4 // 3, # Estimation + total_tokens=100 + len(full_response.split()) * 4 // 3, + ), + ) + else: + # Réponse normale sans outils + completion = OpenAICompletion( + id=response_id, + created=created_time, + model=model_name, + choices=[ + OpenAICompletionChoice( + index=0, + message=OpenAIMessage(role="assistant", content=full_response), + finish_reason="stop", + ) + ], + usage=OpenAIUsage( + prompt_tokens=100, # Valeurs approximatives + completion_tokens=len(full_response.split()) * 4 // 3, # Estimation + total_tokens=100 + len(full_response.split()) * 4 // 3, + ), + ) + + return completion + + +# Fonction utilitaire pour obtenir une instance des dépendances de l'API +def get_api_dependencies(config: OmegaConf = Depends(load_config)) -> APIDependencies: + """Fournit une instance des dépendances de l'API + + Args: + config: Configuration du système + + Returns: + Instance de APIDependencies + """ + return APIDependencies(config=config) \ No newline at end of file diff --git a/ragondin/utils/router_manager.py b/ragondin/utils/router_manager.py new file mode 100644 index 000000000..66d3b013f --- /dev/null +++ b/ragondin/utils/router_manager.py @@ -0,0 +1,128 @@ +from typing import Dict, List, Tuple, Callable, Any +from pathlib import Path +from fastapi import APIRouter +from mcp import MCPTool, MCPResource +from mcp.server.fastmcp import FastMCP + +from fastapi import FastAPI +from fastmcp import MCP +from fastmcp.tools import Tool +from typing import get_type_hints, Callable +from pydantic import create_model +from inspect import signature + +app = FastAPI() +mcp = MCP() + +def MCPToolRoute( + request_type: str = "POST", + path: str = "/", + router: APIRouter = None, + mcp_server :FastMCP=None, + prefix: str = "" +): + + + def decorator(func: Callable): + sig = signature(func) + hints = get_type_hints(func) + name = func.__name__ + fields = { + param: (hints[param], ...) + for param in sig.parameters + } + InputModel = create_model(f"{name.capitalize()}Input", **fields) + + # Créer la fonction FastAPI qui appelle func + + + # Attacher la route dynamiquement + route_func = getattr(router, request_type.lower())(path)(func) + + # Enregistrer comme outil FastMCP + + if request_type == "POST": + mcp_server.tool(name)(func) + elif request_type == "GET": + mcp_server.resource(f"{prefix}/{path}")(func) + + # Retourner la fonction décorée FastAPI + return route_func + + return decorator + + +class RouterManager: + """Gestionnaire de routes API et de ressources MCP""" + + def __init__(self, file_path: str): + """Initialise le gestionnaire avec le chemin du fichier + + Args: + file_path: Chemin du fichier contenant les routes + """ + self.file_path = file_path + self.prefix = Path(file_path).stem # Nom du fichier sans extension + self.router = APIRouter(prefix=f"/{self.prefix}", tags=[self.prefix]) + self.mcp_server = FastMCP(self.prefix) + self.mcp_tool = MCPTool() + self.routes: List[Tuple[str, str, Callable]] = [] # [(request_type, path, func)] + self.api_wrappers: Dict[Callable, Callable] = {} # {func: wrapped_func} + self.mcp_wrappers: Dict[Callable, Callable] = {} # {func: wrapped_func} + + def append(self, request_type: str, path: str, func: Callable) -> None: + """Ajoute une route à configurer + + Args: + request_type: Type de requête (get, post, etc.) + path: Chemin de la route + func: Fonction à wrapper + """ + self.routes.append((request_type, path, func)) + + def setup_router(self) -> APIRouter: + """Configure le router FastAPI avec toutes les routes + + Returns: + Router FastAPI configuré + """ + for request_type, path, func in self.routes: + # Wrapper la fonction avec le décorateur approprié + wrapped_func = getattr(self.router, request_type)(path)(func) + self.api_wrappers[func] = wrapped_func + + return self.router + + def setup_mcp(self) -> MCPTool: + """Configure les ressources MCP pour les routes GET + + Returns: + MCPTool configuré + """ + for request_type, path, func in self.routes: + if request_type == "get": + # Créer une ressource MCP pour chaque route GET + resource = MCPResource( + name=f"{self.prefix}/{path}", + description=f"Resource for {path}", + parameters=func.__annotations__ if hasattr(func, '__annotations__') else {} + ) + + # Wrapper la fonction pour la ressource MCP + wrapped_func = resource(func) + self.mcp_wrappers[func] = wrapped_func + + # Ajouter la ressource au tool + self.mcp_tool.add_resource(resource) + else: + # Si ce n'est pas une route GET, c'est un outil + tool = MCPTool() + + # Wrapper la fonction pour l'outil + wrapped_func = tool(func) + self.mcp_wrappers[func] = wrapped_func + + # Ajouter l'outil au tool + self.mcp_tool.add_tool(tool) + + return self.mcp_tool \ No newline at end of file diff --git a/uv.lock b/uv.lock index 2868ec239..b84f9cb0a 100644 --- a/uv.lock +++ b/uv.lock @@ -160,13 +160,13 @@ name = "anthropic" version = "0.46.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "anyio", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, - { name = "distro", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, - { name = "httpx", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, - { name = "jiter", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, - { name = "pydantic", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, - { name = "sniffio", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, - { name = "typing-extensions", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, + { name = "anyio" }, + { name = "distro" }, + { name = "httpx" }, + { name = "jiter" }, + { name = "pydantic" }, + { name = "sniffio" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/d4/68/3b4c045edf6dc6933895e8f279cc77c7684874c8aba46a4e6241c8b147cf/anthropic-0.46.0.tar.gz", hash = "sha256:eac3d43271d02321a57c3ca68aca84c3d58873e8e72d1433288adee2d46b745b", size = 202191 } wheels = [ @@ -1151,6 +1151,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7b/03/ab118cb743dcf671da01ad0cfd7564465dda115db32976fdc95e21ce8feb/faster_whisper-1.1.0-py3-none-any.whl", hash = "sha256:0f2d025676bbff1e46c4108b6f9a82578d6e33826c174af2990e45b33fab6182", size = 1118168 }, ] +[[package]] +name = "fastmcp" +version = "0.4.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "httpx" }, + { name = "mcp" }, + { name = "pydantic" }, + { name = "pydantic-settings" }, + { name = "python-dotenv" }, + { name = "typer" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/6f/84/17b549133263d7ee77141970769bbc401525526bf1af043ea6842bce1a55/fastmcp-0.4.1.tar.gz", hash = "sha256:713ad3b8e4e04841c9e2f3ca022b053adb89a286ceffad0d69ae7b56f31cbe64", size = 785575 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/79/0b/008a340435fe8f0879e9d608f48af2737ad48440e09bd33b83b3fd03798b/fastmcp-0.4.1-py3-none-any.whl", hash = "sha256:664b42c376fb89ec90a50c9433f5a1f4d24f36696d6c41b024b427ae545f9619", size = 35282 }, +] + [[package]] name = "filelock" version = "3.17.0" @@ -2657,6 +2674,25 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ac/c2/0d5aae823bdcc42cc99327ecdd4d28585e15ccd5218c453b7bcd827f3421/matplotlib-3.10.1-cp313-cp313t-win_amd64.whl", hash = "sha256:bc411ebd5889a78dabbc457b3fa153203e22248bfa6eedc6797be5df0164dbf9", size = 8134832 }, ] +[[package]] +name = "mcp" +version = "1.5.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "httpx" }, + { name = "httpx-sse" }, + { name = "pydantic" }, + { name = "pydantic-settings" }, + { name = "sse-starlette" }, + { name = "starlette" }, + { name = "uvicorn" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/6d/c9/c55764824e893fdebe777ac7223200986a275c3191dba9169f8eb6d7c978/mcp-1.5.0.tar.gz", hash = "sha256:5b2766c05e68e01a2034875e250139839498c61792163a7b221fc170c12f5aa9", size = 159128 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c1/d1/3ff566ecf322077d861f1a68a1ff025cad337417bd66ad22a7c6f7dfcfaf/mcp-1.5.0-py3-none-any.whl", hash = "sha256:51c3f35ce93cb702f7513c12406bbea9665ef75a08db909200b07da9db641527", size = 73734 }, +] + [[package]] name = "mdurl" version = "0.1.2" @@ -4466,12 +4502,14 @@ version = "0.1.0" source = { virtual = "." } dependencies = [ { name = "aiopath" }, + { name = "anthropic" }, { name = "chainlit" }, { name = "debugpy" }, { name = "docling" }, { name = "einops" }, { name = "fastembed-gpu", version = "0.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.13' and platform_machine == 'x86_64' and sys_platform == 'darwin'" }, { name = "fastembed-gpu", version = "0.5.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.13' or platform_machine != 'x86_64' or sys_platform != 'darwin'" }, + { name = "fastmcp" }, { name = "hydra-core" }, { name = "langchain-community" }, { name = "langchain-core" }, @@ -4485,6 +4523,7 @@ dependencies = [ { name = "marker-pdf", version = "0.2.17", source = { registry = "https://pypi.org/simple" }, marker = "platform_machine == 'x86_64' and sys_platform == 'darwin'" }, { name = "marker-pdf", version = "1.6.1", source = { registry = "https://pypi.org/simple" }, marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, { name = "markitdown" }, + { name = "mcp" }, { name = "openai" }, { name = "psutil" }, { name = "pydub" }, @@ -4494,6 +4533,7 @@ dependencies = [ { name = "ragatouille" }, { name = "ray", extra = ["default"] }, { name = "spire-doc" }, + { name = "starlette" }, { name = "whisperx" }, ] @@ -4505,11 +4545,13 @@ dev = [ [package.metadata] requires-dist = [ { name = "aiopath", specifier = ">=0.7.7" }, + { name = "anthropic", specifier = ">=0.46.0" }, { name = "chainlit", specifier = ">=2.2.1" }, { name = "debugpy", specifier = ">=1.8.13" }, { name = "docling", specifier = ">=2.24.0" }, { name = "einops", specifier = ">=0.8.1" }, { name = "fastembed-gpu", specifier = ">=0.4.2" }, + { name = "fastmcp", specifier = ">=0.4.1" }, { name = "hydra-core", specifier = ">=1.3.2" }, { name = "langchain-community", specifier = ">=0.3.18" }, { name = "langchain-core", specifier = ">=0.3.39" }, @@ -4522,6 +4564,7 @@ requires-dist = [ { name = "loguru", specifier = ">=0.7.3" }, { name = "marker-pdf", specifier = ">=0.2.17" }, { name = "markitdown", specifier = ">=0.0.2" }, + { name = "mcp", specifier = ">=1.5.0" }, { name = "openai", specifier = ">=1.64.0" }, { name = "psutil", specifier = ">=7.0.0" }, { name = "pydub", specifier = ">=0.25.1" }, @@ -4531,6 +4574,7 @@ requires-dist = [ { name = "ragatouille", specifier = ">=0.0.9" }, { name = "ray", extras = ["default"], specifier = ">=2.43.0" }, { name = "spire-doc", specifier = ">=13.1.0" }, + { name = "starlette", specifier = ">=0.41.3" }, { name = "whisperx", specifier = ">=3.3.1" }, ] @@ -5303,6 +5347,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/3a/e2/745aeba88a8513017fbac2fd2f9f07b8a36065e51695f818541eb795ec0c/srsly-2.5.1-cp313-cp313-win_amd64.whl", hash = "sha256:e73712be1634b5e1de6f81c273a7d47fe091ad3c79dc779c03d3416a5c117cee", size = 630634 }, ] +[[package]] +name = "sse-starlette" +version = "2.2.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "starlette" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/71/a4/80d2a11af59fe75b48230846989e93979c892d3a20016b42bb44edb9e398/sse_starlette-2.2.1.tar.gz", hash = "sha256:54470d5f19274aeed6b2d473430b08b4b379ea851d953b11d7f1c4a2c118b419", size = 17376 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d9/e0/5b8bd393f27f4a62461c5cf2479c75a2cc2ffa330976f9f00f5f6e4f50eb/sse_starlette-2.2.1-py3-none-any.whl", hash = "sha256:6410a3d3ba0c89e7675d4c273a301d64649c03a5ef1ca101f10b47f895fd0e99", size = 10120 }, +] + [[package]] name = "standard-aifc" version = "3.13.0" @@ -5830,6 +5887,8 @@ dependencies = [ { name = "torch", version = "2.6.0", source = { registry = "https://pypi.org/simple" }, marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, ] wheels = [ + { url = "https://files.pythonhosted.org/packages/52/5b/76ca113a853b19c7b1da761f8a72cb6429b3bd0bf932537d8df4657f47c3/torchvision-0.21.0-1-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:ffa2a16499508fe6798323e455f312c7c55f2a88901c9a7c0fb1efa86cf7e327", size = 2329878 }, + { url = "https://files.pythonhosted.org/packages/4e/fe/5e193353706dab96fe73ae100d5a633ff635ce310e0d92f3bc2958d075b1/torchvision-0.21.0-1-cp313-cp313-manylinux_2_28_aarch64.whl", hash = "sha256:7e9e9afa150e40cd2a8f0701c43cb82a8d724f512896455c0918b987f94b84a4", size = 2280711 }, { url = "https://files.pythonhosted.org/packages/6e/1b/28f527b22d5e8800184d0bc847f801ae92c7573a8c15979d92b7091c0751/torchvision-0.21.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:97a5814a93c793aaf0179cfc7f916024f4b63218929aee977b645633d074a49f", size = 1784140 }, { url = "https://files.pythonhosted.org/packages/36/63/0722e153fd27d64d5b0af45b5c8cb0e80b35a68cf0130303bc9a8bb095c7/torchvision-0.21.0-cp312-cp312-manylinux1_x86_64.whl", hash = "sha256:b578bcad8a4083b40d34f689b19ca9f7c63e511758d806510ea03c29ac568f7b", size = 7238673 }, { url = "https://files.pythonhosted.org/packages/bb/ea/03541ed901cdc30b934f897060d09bbf7a98466a08ad1680320f9ce0cbe0/torchvision-0.21.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:5083a5b1fec2351bf5ea9900a741d54086db75baec4b1d21e39451e00977f1b1", size = 14701186 },