L’IA agente avec LangChain — Partie 6 : RAG agenté alimenté par MCP

Ce guide reconstitue le parcours allant des matières premières à un système fonctionnel pour : l’IA agente avec LangChain — Partie 6 : RAG agenté alimenté par MCP et LangGraph. L’accent est mis sur des étapes opérationnelles, des vérifications explicites, ainsi que du code que vous pouvez intégrer directement dans un dépôt sans devoir deviner l’intention. Pour une vue d’ensemble, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans avoir à deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès, et refusez toute complétion partielle silencieuse.

Concepts techniques

Lorsque vous travaillez sur les concepts techniques, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût des jetons ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le système passe de l’environnement de démonstration à des environnements partagés. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Changer fréquemment les prompts ne résout que rarement un système de récupération insuffisant.

1. Configurations du système

Lorsque vous travaillez sur la section 1. Configurations du système, notez d’abord les exigences : entrées requises, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Conservez les configurations en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les administrateurs peuvent auditer sans devoir lire l’ensemble du système. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.

OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

model_config = SettingsConfigDict(env_file=".env", extra="ignore")

2. Structure du projet

Lorsque vous travaillez sur la section 2. Structure du projet, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations apportées ultérieurement. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.

project/
├── rag_process/       # Pinecone RAG operations
│   ├── __init__.py
│   ├── extractor.py
|   └── service.py
│
├── graphs             # implement ReAct and Reflection StateGraph
│   ├── nodes.py       # graph nodes. Defined as Python functions
│   ├── graph.py       # graph workflow for MCP in SES mode
│   ├── graph_stdio.py # graph workflow for MCP in STDIO mode
│   ├── consts.py      # define node names as constants
│   └── state.py       # typed dictionary to store graph states
│

├── mcp_client_sse.py     # mcp client in sse mode
├── mcp_client_stdio.py   # mcp client in stdio mode
├── server_sse.py         # mcp server in sse mode
│── server_stdio.py       # mcp server in stdio mode
|── schemas.py          # pydantic data models
|── upload.py           # python functions to upload files asynchronously
├── main.py             # fastAPI app
├── config.py           # pydantic BaseConfig
│── utils.py            # utility extracting question/docs from message list
├── .env                # environment variable definition
├── client.py           # streamlit client
├── .gitignore
└── requirements.txt     # package requirements

3. MCP (Model Context Protocol)

Lorsque vous travaillez sur le 3. MCP (Model Context Protocol), notez d’abord le contrat : les entrées requises, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Préférez des unités petites et testables à des scripts volumineux. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus complexe et embrouillé. Mémorisez les instructions du système stables ainsi que les schémas des outils. Envoyer à nouveau un préambule identique est une cause fréquente de gaspillage.

3.1. Concept

Lorsque vous travaillez sur la section 3.1. Concept, notez d’abord les conditions du contrat : entrées requises, signal de succès, et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux éléments générés, définez des vérifications de succès, et refusez les terminaisons partielles silencieuses. Évaluez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.

3.2 Architecture MCP

Lorsque vous travaillez sur l’architecture MCP 3.2, notez d’abord le contrat : les entrées requises, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite des factures inattendues lorsque le système passe de l’environnement de démonstration à des environnements partagés. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Changer fréquemment les prompts ne résout que rarement un système de récupération insuffisant.

3.3. Modes de transport MCP

Lorsque vous travaillez sur la section 3.3. Modes de transport MCP, notez d’abord les exigences : entrées requises, signal de succès, et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les administrateurs peuvent auditer sans devoir lire l’ensemble du système. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.

3.3.1. Mode STDIO

Lorsque vous travaillez sur la section 3.3.1. Mode STDIO, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Documentez à la fois le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations apportées ultérieurement. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant. Lorsque vous travaillez sur la section 3.3.1. Mode STDIO, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des critères de succès et refusez les complétions partielles silencieuses.

3.3.2. Mode SSE

3.3.2. Le mode SSE fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le parcours passe de l’environnement de démonstration aux environnements partagés. Séparez la politique de segmentation de la politique de récupération : modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité changent.

4. Mise en œuvre de MCP en mode STDIO

1. L’implémentation de MCP en mode STDIO fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple réussi, un cas d’échec ainsi que la note de réversion avant d’élargir le périmètre. Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Séparez la politique de segmentation de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent.

4.1. Serveur MCP

4.1. Le MCP Server fonctionne le mieux lorsqu’il est considéré comme une entité mesurable. Capturez un exemple idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Documentez en même temps le parcours normal et celui de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure. Séparez la politique de segmentation de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent. 4.1. Le MCP Server fonctionne le mieux lorsqu’il est considéré comme une entité mesurable. Capturez un exemple idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des critères de succès et refusez les terminations partielles silencieuses.

# initialize an STDIO FastMCP instance
mcp = FastMCP("Search Server")

# create mcp tools
@mcp.tool()
async def search_agentic_docs(query: str) -> str:
    """
    Good for questions about generative agents, prompt engineering, and adversarial attacks.
    Pass a natural language search query to retrieve context from the agentic ai database.
    """
    # Simply invoke your existing vector service retriever
    docs = await vector_service.retriever.ainvoke(query)

    # Flatten the document content into a single string for the LLM
    return "\n\n".join([d.page_content for d in docs])

# start MCP server in stdio mode
if __name__ == "__main__":
   mcp.run(transport="stdio")

4.2. MCP Client

Pour le client MCP 4.2, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des jetons ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce du coût évite les factures inattendues lorsque le parcours passe de l’environnement de démonstration à des environnements partagés. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.

# define location of server file
BASE_DIR = Path(__file__).parent
SERVER_SCRIPT = BASE_DIR / "server.py"

# Local server running directly via Python over stdio
SERVER_CONFIG = {

    "local_search_server": {
        "transport": "stdio",
        "command": "python",
        "args": [str(SERVER_SCRIPT)],
    },
}


# create a MCP client and return
def get_mcp_client() :
    """Works for stdio, sse, or a mix of both!"""
    client = MultiServerMCPClient(SERVER_CONFIG)
    return client

4.3. Utilisation des outils MCP dans l’agent

Pour 4.3. Utilisation des outils MCP dans l’agent : définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.

from mcp_client_stdio import get_mcp_client


async def build_graph():
    """
    build StateGraph
    :return:
    """
    # initialize a MCP client instance
    client = get_mcp_client()

    # retrieve MCP tools
    mcp_tools = await client.get_tools()

    # create an llM
    llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
    tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY)] + mcp_tools

    # bind tools to ToolNode and llm
    tool_node = ToolNode(tools)
    workflow = StateGraph(GraphState)
    llm_with_tools = llm.bind_tools(tools)

4.4. Test du flux de travail STDIO

Pour 4.4. Tester le flux de travail STDIO, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez conjointement le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’une mise en forme ultérieure. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation. Pour 4.4. Tester le flux de travail STDIO, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définissez des vérifications de succès et refusez les terminations partielles silencieuses.

question_1 = "what is prompt engineer ?"

async def main():
    print("hello ReAct agent by Langraph")
    message = HumanMessage(content=question_1)
    app = await build_graph()
    res = await app.ainvoke({"messages": [message]})


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

5. Mise en œuvre de MCP en mode SSE

Lorsque vous travaillez sur la section 5. Mise en œuvre de MCP en mode SSE, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le système passe d’un environnement de démonstration à des environnements partagés. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Changer fréquemment les prompts ne résout que rarement un système de récupération insuffisant.

5.1. Serveur MCP

Lorsque vous travaillez sur la section 5.1. MCP Server, notez d’abord le contrat : les entrées requises, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les administrateurs peuvent auditer sans devoir lire l’ensemble du système. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.

# for SSE
mcp = FastMCP(
    name="My MCP Server",
    host="0.0.0.0",
    port=8001
)

# implement mcp tools here

if __name__ == "__main__":
    mcp.run(transport="sse")

5.2. MCP Client

Lorsque vous travaillez sur la section 5.2. MCP Client, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Documentez à la fois le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations apportées ultérieurement. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent de prompts ne résout que rarement un système de récupération insuffisant. Lorsque vous travaillez sur la section 5.2. MCP Client, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définites des vérifications de succès et refusez les complétions partielles silencieuses.

# specify config of the SSE MCP server that the client will connect to
SERVER_CONFIG = {
    # config for mcp by sse transportation
    "fastapi_mcp_server": {
        "transport": "sse",
        "url": "http://localhost:8001/sse",
    }
}


# create a client
def get_mcp_client() :
    client = MultiServerMCPClient(SERVER_CONFIG)
    return client

5.3. Utilisation des outils MCP dans StateGraph

5.3. L’utilisation des outils MCP dans StateGraph fonctionne le mieux lorsqu’ils sont considérés comme une surface mesurable. Capturez un enregistrement exemplaire, un cas d’échec et la note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts permet d’éviter des factures inattendues lorsque le chemin passe de l’environnement de démonstration à des environnements partagés. Séparez la politique de segmentation de la politique de récupération : modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité changent.

async def build_graph(mcp_tools: list):
    """
    build StateGraph
    :return:
    """

    # create an llm
    llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
    tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY)] + mcp_tools

    # bind tools to ToolNode and llm
    tool_node = ToolNode(tools)
    llm_with_tools = llm.bind_tools(tools)

5.4. Pourquoi l’implémentation SSE diffère de STDIO

5.4. Pourquoi la mise en œuvre de SSE diffère de STDIO : elle fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple réussi, un cas d’échec ainsi que la note de réversion avant d’élargir le périmètre. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les administrateurs peuvent auditer sans devoir lire l’ensemble du système.

6. Intégration du client SSE MCP avec FastAPI

from mcp_client_sse import get_mcp_client

# CREATES a manager for FastAPI startup/shutdown
@asynccontextmanager
async def lifespan(fastapi_app: FastAPI):

    # create an MCP client, and retrieve MCP tools
    mcp_client = get_mcp_client()
    mcp_tools = await mcp_client.get_tools()

    # Compile graph dynamically with tools and store it on app state
    fastapi_app.state.agent = await build_graph(mcp_tools)

    # PAUSES here while FastAPI runs and handles requests
    yield


app = FastAPI(lifespan=lifespan)

7. Réconstruction du StateGraph dans la partie 5

7.1. Point d’entrée FastAPI

@app.post("/generate_text", response_model=RAGResponse)
async def query_by_RAG_controller(request: Request, body: RAGRequest=Body(...)) -> RAGResponse:
    try:
        # wrap input query in a HumanMessage
        message = HumanMessage(content=body.question)

        # loads the StateGraph agent from the request
        # this agent is created and maintained by lifespan context
        agent = request.app.state.agent

        # Async invokes the agent and returns the response
        generation = await agent.ainvoke({"messages": [message]})
        response = {"answer": generation.get("generation", "")}

        return RAGResponse(**response)

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

7.2. Interface utilisateur Streamlit

8. Outils gérés par MCP vs outils locaux LangChain

9. Résumé final

Liste de contrôle opérationnelle