Aller au contenu
> 💻 🧠 Code 1001 > LLM : articles, supports pratiques et ressources > Guide pratique pour construire des agents d’IA avec LangGraph et MCP

Guide pratique pour construire des agents d’IA avec LangGraph et MCP

  • par

Cet article est un guide pratique destiné aux développeurs pour la création d’agents d’IA autonomes en Python. Nous ne répéterons pas la théorie sur ce que sont LangChain et LangGraph. Au lieu de cela, nous nous concentrerons sur le code, l’architecture et la résolution de problèmes concrets.

Objectif : Construire deux projets à partir de zéro :

  1. Un Agent de Classification : Un agent multi-étapes avec un état géré, mais sans outils externes.
  2. Un Agent Assistant : Un agent complet avec accès au système de fichiers et à la recherche web via le protocole MCP, basé sur une logique cyclique.

Nous aborderons les meilleures pratiques : la gestion de la configuration, la sélection des modèles et la gestion des erreurs pour créer des systèmes robustes.

In Questo Articolo

Bref aperçu des concepts : L’Agent et le pont MCP

Avant de plonger dans le code, définissons deux concepts :

  • Agent d’IA : Un programme construit autour d’une boucle « raisonnement-action ». Il reçoit une tâche, utilise un LLM pour décider de la prochaine action (par ex., appeler un outil), exécute l’action et répète le cycle jusqu’à ce que la tâche soit terminée.
  • MCP (Model Context Protocol) : Un standard qui sert de pont entre la logique de l’agent et les outils externes. Il permet à l’agent de travailler avec des fichiers, des API ou des recherches de manière unifiée, sans se soucier des détails de leur implémentation.

Partie 1 : Mise en place d’un environnement robuste

Étape 1 : Environnement virtuel et dépendances

Créez et activez un environnement virtuel. Ensuite, créez un fichier requirements.txt :« `

Frameworks principaux

langchain
langgraph

Adaptateurs de modèles

langchain-openai
langchain-google-genai
langchain-mistralai
langchain-community # Pour Ollama

Outils et protocoles

langchain-mcp-adapters
mcp
ollama

Utilitaires

python-dotenv
tenacity # Pour une gestion robuste des erreurs

Installez les dépendances :

bash
pip install -r requirements.txt« `

Étape 2 : Configuration des clés d’API

Créez un fichier .env pour stocker vos clés :

OPENAI_API_KEY="sk-..."
GOOGLE_API_KEY="AIzaSy..."
MISTRAL_API_KEY="..."
BRAVE_API_KEY="..." # Pour l'outil de recherche web via MCP

Étape 3 : Le patron de conception « Fabrique de Modèles »

Pour basculer de manière flexible entre les modèles cloud et locaux sans modifier le code de l’agent, nous utiliserons le patron de conception « fabrique » (factory pattern).

# llm_factory.py
import os
from enum import Enum
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_mistralai import ChatMistralAI
from langchain_community.chat_models import ChatOllama

load_dotenv()

class ModelProvider(Enum):
    OPENAI = "openai"
    GEMINI = "gemini"
    MISTRAL_API = "mistral_api"
    OLLAMA = "ollama"

def get_llm(provider: ModelProvider, model_name: str = None):
    """Une fabrique pour créer des instances de LLM."""
    if provider == ModelProvider.OPENAI:
        return ChatOpenAI(model=model_name or "gpt-4o-mini", temperature=0)
    elif provider == ModelProvider.GEMINI:
        return ChatGoogleGenerativeAI(model=model_name or "gemini-1.5-flash", temperature=0)
    elif provider == ModelProvider.MISTRAL_API:
        return ChatMistralAI(model=model_name or "mistral-large-latest", temperature=0)
    elif provider == ModelProvider.OLLAMA:
        # Assurez-vous qu'Ollama est en cours d'exécution avec le modèle requis
        # docker exec -it ollama ollama pull mistral
        return ChatOllama(model=model_name or "mistral", temperature=0)
    raise ValueError(f"Fournisseur de modèle inconnu : {provider}")

# Exemple d'utilisation
if __name__ == "__main__":
    # local_llm = get_llm(ModelProvider.OLLAMA)
    openai_llm = get_llm(ModelProvider.OPENAI)
    response = openai_llm.invoke("Expliquez le concept de RAG en trois phrases.")
    print(response.content)

Partie 2 : Projet 1 — Agent de Classification d’offres d’emploi

Cet agent montre comment utiliser LangGraph pour créer un graphe linéaire avec un état géré. Il prendra une description de poste et la classifiera séquentiellement selon trois paramètres.

Étape 1 : Définition de l’État

L’état est la « mémoire » de notre graphe, transmise d’un nœud à l’autre.

# vacancy_classifier.py
from typing import TypedDict, Dict

class ClassificationState(TypedDict):
    """État pour l'agent de classification."""
    description: str          # Texte source
    job_type: str             # Type de poste (projet/permanent)
    category: str             # Profession
    search_type: str          # Objectif (cherche un emploi/un prestataire)
    classification_log: list  # Journal de débogage

Étape 2 : Implémentation des Nœuds du Graphe

Chaque nœud est une fonction qui prend l’état, effectue sa part du travail et renvoie l’état mis à jour.

import asyncio
import json
from langchain_core.prompts import ChatPromptTemplate
from llm_factory import get_llm, ModelProvider

class VacancyClassifierAgent:
    def __init__(self):
        self.llm = get_llm(ModelProvider.OPENAI, model_name="gpt-4o-mini")

    async def _classify_job_type(self, state: ClassificationState) -> ClassificationState:
        """Nœud 1 : Détermine le type de poste."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Déterminez le type de poste. La réponse doit être 'projet' ou 'permanent'."),
            ("human", "Description du poste :\n\n{description}")
        ])
        chain = prompt | self.llm
        result = await chain.ainvoke({"description": state["description"]})

        state["job_type"] = result.content.strip()
        state["classification_log"].append("Type de poste déterminé.")
        return state

    async def _classify_category(self, state: ClassificationState) -> ClassificationState:
        """Nœud 2 : Détermine la catégorie de la profession."""
        # Les catégories могут быть загружены из файла или базы данных
        categories = ["Développeur Python", "Designer", "Marketeur", "Animateur 3D"]
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"Choisissez la catégorie la plus appropriée dans la liste : {', '.join(categories)}."),
            ("human", "Description du poste :\n\n{description}")
        ])
        chain = prompt | self.llm
        result = await chain.ainvoke({"description": state["description"]})

        state["category"] = result.content.strip()
        state["classification_log"].append("Catégorie déterminée.")
        return state

    async def _classify_search_type(self, state: ClassificationState) -> ClassificationState:
        """Nœud 3 : Détermine l'objectif de la recherche."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Déterminez l'objectif de l'auteur. La réponse doit être 'recherche d'emploi' ou 'recherche de prestataire'."),
            ("human", "Description du poste :\n\n{description}")
        ])
        chain = prompt | self.llm
        result = await chain.ainvoke({"description": state["description"]})

        state["search_type"] = result.content.strip()
        state["classification_log"].append("Objectif de recherche déterminé.")
        return state

Étape 3 : Assemblage et Exécution du Graphe

Nous assemblons les nœuds en un seul flux de travail.

# ... suite de la classe VacancyClassifierAgent ...
from langgraph.graph import StateGraph, END

    def build_graph(self):
        """Assemble le graphe d'états."""
        workflow = StateGraph(ClassificationState)

        workflow.add_node("job_type_classifier", self._classify_job_type)
        workflow.add_node("category_classifier", self._classify_category)
        workflow.add_node("search_type_classifier", self._classify_search_type)

        workflow.set_entry_point("job_type_classifier")
        workflow.add_edge("job_type_classifier", "category_classifier")
        workflow.add_edge("category_classifier", "search_type_classifier")
        workflow.add_edge("search_type_classifier", END)

        return workflow.compile()

async def main():
    agent = VacancyClassifierAgent()
    graph = agent.build_graph()

    description = "Nous recherchons un développeur Python expérimenté pour rejoindre notre équipe à temps plein pour travailler sur un projet fintech."

    initial_state = ClassificationState(
        description=description,
        job_type="", category="", search_type="",
        classification_log=[]
    )

    final_state = await graph.ainvoke(initial_state)

    print("--- Résultat de la Classification ---")
    print(json.dumps(final_state, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    asyncio.run(main())

Partie 3 : Projet 2 — Agent Assistant avec Outils (MCP)

Cet agent illustre une logique cyclique, où il peut appeler des outils de manière répétée pour résoudre une tâche.

Étape 1 : Gestion de la Configuration

Pour les agents interagissant avec le monde extérieur, une configuration robuste est essentielle.

# mcp_agent_config.py
from dataclasses import dataclass, field
import os
from llm_factory import ModelProvider

@dataclass
class AgentConfig:
    workdir: str = "./agent_workdir"
    model_provider: ModelProvider = ModelProvider.OLLAMA

    def __post_init__(self):
        """Validation post-initialisation."""
        os.makedirs(self.workdir, exist_ok=True)

Étape 2 : Définition de l’État pour le Dialogue

L’état stockera désormais l’historique des messages.

# mcp_agent.py
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

Étape 3 : Implémentation du Graphe Cyclique

Le graphe se composera de deux nœuds principaux et d’une arête conditionnelle qui crée la boucle « raisonnement-action ».

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
from langchain_mcp_adapters.langchain import V1ToolExecutor
from langchain_mcp_adapters.clients import MultiServerMCPClient
from llm_factory import get_llm
from mcp_agent_config import AgentConfig

class MCPAgent:
    def __init__(self, config: AgentConfig):
        self.config = config
        self.llm = get_llm(config.model_provider)
        self.tools = []
        self.tool_executor = None

    async def setup_tools(self):
        """Initialise les outils via MCP."""
        mcp_config = {
            "filesystem": {
                "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", self.config.workdir],
                "transport": "stdio"
            },
            # Ajoutez brave-search si vous avez une BRAVE_API_KEY
        }
        mcp_client = MultiServerMCPClient(mcp_config)
        self.tools = await mcp_client.get_tools()
        self.tool_executor = ToolExecutor([V1ToolExecutor(tool) for tool in self.tools])

        # Liez les outils au modèle
        self.llm = self.llm.bind_tools(self.tools)

    def _should_continue(self, state: AgentState):
        """Arête conditionnelle : décide s'il faut appeler un outil."""
        last_message = state['messages'][-1]
        if not last_message.tool_calls:
            return "end"
        return "continue"

    def _call_model(self, state: AgentState):
        """Nœud 1 : Appelle le LLM pour prendre une décision."""
        response = self.llm.invoke(state['messages'])
        return {"messages": [response]}

    def _call_tool(self, state: AgentState):
        """Nœud 2 : Exécute l'appel à l'outil."""
        last_message = state['messages'][-1]
        tool_call = last_message.tool_calls[0]

        action = {"tool": tool_call["name"], "tool_input": tool_call["args"], "log": ""}
        response = self.tool_executor.invoke(action)

        return {"messages": [response]}

    def build_graph(self):
        workflow = StateGraph(AgentState)
        workflow.add_node("agent", self._call_model)
        workflow.add_node("action", self._call_tool)

        workflow.set_entry_point("agent")
        workflow.add_conditional_edges(
            "agent",
            self._should_continue,
            {"continue": "action", "end": END}
        )
        workflow.add_edge("action", "agent")

        return workflow.compile()

Étape 4 : Exécution et Interaction

# ... suite de mcp_agent.py ...
import asyncio
from langchain_core.messages import HumanMessage
from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
async def run_agent_task(graph, task):
    """Exécute une tâche avec gestion des erreurs."""
    return await graph.ainvoke({"messages": [HumanMessage(content=task)]})

async def main():
    config = AgentConfig(model_provider=ModelProvider.OPENAI) # ou OLLAMA
    agent = MCPAgent(config)
    await agent.setup_tools()
    graph = agent.build_graph()

    task = "Créez un fichier nommé 'salut.txt' dans le répertoire de travail et écrivez 'Bonjour, le monde !' dedans."
    result = await run_agent_task(graph, task)

    print("\n--- Réponse Finale de l'Agent ---")
    print(result['messages'][-1].content)

if __name__ == "__main__":
    asyncio.run(main())

Ici, nous avons ajouté le décorateur tenacity pour la robustesse : si l’appel à l’agent échoue à cause d’une erreur réseau temporaire, il sera automatiquement réessayé.

Conclusion

Nous avons construit deux types d’agents en utilisant des pratiques modernes :

  • Un graphe linéaire est excellent pour les tâches avec une séquence claire d’étapes, comme les processus ETL ou l’analyse multi-étapes.
  • Un graphe cyclique est la base pour créer des assistants interactifs et des agents autonomes capables de résoudre des problèmes complexes avec des outils.

Les patrons de conception architecturaux présentés — la fabrique de modèles, la gestion de la configuration, la séparation de la logique en nœuds et l’utilisation de graphes d’états — sont les fondements pour construire des systèmes d’IA évolutifs et robustes.

Étiquettes:

Laisser un commentaire

Votre adresse e-mail ne sera pas publiée. Les champs obligatoires sont indiqués avec *