L’IA agente avec LangChain — Partie 7 : Création de systèmes multi-agents

Ce guide reconstitue le parcours allant des matières premières à un système fonctionnel pour : l’IA agente avec LangChain — Partie 7 : Création de systèmes multi-agents avec 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 derrière lui. Pour l’étape d’aperçu, 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 avoir à deviner l’état caché. Documentez conjointement le parcours normal et les scénarios 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.

Pourquoi les multi-agents sont importants

Lors de l’étape « Pourquoi les multi-agents sont importants », 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. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité et non un processus embrouillé. Créez des points de contrôle après les étapes coûteuses. Le système de reprise ne doit pas facturer à nouveau la même appel d’LLM lorsque l’opérateur réessaie un nœud ultérieur.

Concepts techniques utilisés dans cet article

Lorsque vous travaillez sur les Concepts techniques utilisés à ce stade, 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 garantir l’honnêteté des modifications ultérieures du code. Considérez ce stade comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des vérifications de succès, et refusez toute exécution partielle silencieuse. Faites un point après les étapes coûteuses. Le système de reprise ne doit pas facturer à nouveau la même appel du LLM lorsque l’opérateur réessaie un nœud ultérieur.

Aperçu du projet : Un agent assistant financier

Lors de l’étape « Aperçu du projet – Aspects financiers », notez d’abord les conditions du contrat : entrées requises, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté 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 processus passe de la démonstration aux environnements partagés. Instaurez des points de contrôle après les étapes coûteuses. Le système de reprise ne doit pas facturer à nouveau la même appel d’LLM lorsque l’opérateur réessaie un nœud ultérieur. Lors de l’étape « Aperçu du projet – Aspects financiers », notez d’abord les conditions du contrat : entrées requises, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Documentez conjointement le parcours optimal et les procédures de récupération. Les tentatives de réessai, 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.

1. Configuration du système

La phase de configuration du système fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit pointer vers une seule responsabilité plutôt que vers un processus embrouillé. Maintenez l’état des graphes simple et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

OPENAI_API_KEY="<Your OpenAI API Key>"
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"

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

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

2. Structure du projet

La phase 2 relative à la structure du projet fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Recueillez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Considérez cette phase 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 toute mise en œuvre partielle silencieuse. Maintenez l’état du graphe plat et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

project/
├── tools/
│   ├── __init__.py
│   ├── financial_search.py # fetch stock prices for companies
│   ├── python_REPL.py      # async wrapper of PythonREPL()
|   └── wikipedia_search.py # fetch financial summary of companies
│
├── graphs/            # implement subagents and multi-agents
│   ├── agent.py       # implement research and analyst in a single agent
│   ├── sub_agents.py  # implement research and analyst sub agents
│   ├── multi_agent_supervisor.py # implement supervisor multi-agents
│   └── multi_agent_swarm.py # implement swamp multi-agents
│
├── config.py           # pydantic BaseConfig
│── utils.py            # utility for printing and executing agents
├── .env                # environment variable definition
├── .gitignore
└── requirements.txt     # package requirements

3. Outils de l’agent assistant financier

Les 3 Outils de la scène fonctionnent le mieux lorsqu’ils sont considérés comme une surface mesurable. Capturez un enregistrement idéal, 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 processus passe de la démonstration aux environnements partagés. Exposez des outils dotés de schémas restreints et de labels explicites indiquant les effets secondaires. Les hôtes doivent savoir quels appels modifient l’état avant d’approuver automatiquement. Les 3 Outils de la scène fonctionnent le mieux lorsqu’ils sont considérés comme une surface mesurable. Capturez un enregistrement idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Documentez ensemble le parcours réussi et le parcours de récupération. Les tentatives répétées, les contrôles humains et le traitement des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure.

3.1. Outil de recherche de données existantes

Pour l’étape 3 1 Stock Data, 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é. Préférez des unités petites et testables à des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un pipeline embrouillé. Authentifiez au niveau du gateway et réautorisez au niveau du plan de données. Un token porteur seul ne constitue pas une frontière entre les tenants.

# define LangChain tool to retrieve stock data
@tool
async def stock_data_tool(
        company_ticker: Annotated[str, "The ticker symbol of the company to retrieve their stock performance data."],
        num_days: Annotated[int, "The number of business days of stock data required to respond to the user query."]
) -> str:

        # fetch stock data asynchronously from yfinance
        # check full code for _fetch_stock_sync function
        df = await asyncio.to_thread(_fetch_stock_sync, company_ticker, f"{num_days}d")

        return (f"Successfully executed the stock performance data retrieval tool to retrieve "
                f"the last *{num_days} days* of data for company **{company_ticker}**:\n\n{df.to_markdown()}")

3.2. Outil d’exécution de code Python

Pour l’étape du code Python 3.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é. 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 exécution partielle silencieuse. Séparez la construction du client du cycle de messages afin que les fournisseurs puissent être remplacés sans avoir à réécrire la machine d’état de la conversation.

@tool
async def python_repl_tool(
        code: Annotated[str, "The python code to execute to generate your chart."],
) -> str:

    # wrap LLM generated code and execute it by python.REPL()
    # repl = PythonREPL().
    clean_code = textwrap.dedent(code).strip()
    result = await asyncio.to_thread(repl.run, clean_code)

    return (
        f"Successfully executed the Python REPL tool.\n\n"
        f"Python code executed:\n```python\n{code}\n```\n\n"
        f"Code output:\n```\n{result}```"
    )

3.3. Outil de recherche Wikipedia

Pour l’étape de recherche sur Wikipedia 3 3, 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é. 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 processus passe d’un environnement de démonstration à des environnements partagés. Authentifiez-vous au niveau du gateway et réautorisez-vous au niveau du plan de données. Un simple jeton porteur ne constitue pas une frontière entre les tenants. Pour l’étape de recherche sur Wikipedia 3 3, 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 les scénarios de récupération. Les tentatives de réessai, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’améliorations ultérieures.

@tool()
async def wikipedia_tool(
        query: Annotated[str, "The Wikipedia search to execute to find key summary information."],
) -> str:
    """Use this to search Wikipedia for factual information asynchronously."""
    headers = {"User-Agent": "MyLangChainAgent/1.0 (contact@example.com)"}

    async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:

        # Step 1: Search Wikipedia for matching pages
        search_url = "https://en.wikipedia.org/w/api.php"
        search_params = {
            "action": "query",
            "list": "search",
            "srsearch": query,
            "format": "json"
        }
        response = await client.get(search_url, params=search_params)
        search_data = response.json()
        results = search_data.get("query", {}).get("search", [])
        title = results[0]["title"]

        # Step 2: Fetch the page extract/summary
        summary_url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{title}"
        summary_res = await client.get(summary_url)
        summary_data = summary_res.json()
        summary = summary_data.get("extract", "No summary available.")

        return f"Successfully executed:\nWikipedia summary: {summary}"

4. Swarm Multi-Agents

Lors de la réalisation de l’étape 4 Swarm Multi-Agents, 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 maintenir l’honnêteté des modifications ultérieures du code. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité et non un processus embrouillé. Créez des points de contrôle après les étapes coûteuses. Le système de reprise ne doit pas facturer à nouveau la même appel LLM lorsque l’opérateur réessaie un nœud ultérieur.

4.1. Architecture de Swarm Multi-Agents

Lorsque vous travaillez sur l’Architecture de phase 4.1, 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 garantir l’honnêteté des modifications ultérieures du code. Considérez cette phase comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des vérifications de succès, et refusez les terminations partielles silencieuses. Faites des points de contrôle après les étapes coûteuses. Le système de reprise ne doit pas facturer à nouveau la même appel du LLM lorsque l’opérateur réessaie un nœud ultérieur.

4.2. Mise en œuvre multi-agents Swarm

Lors du travail sur l’étape 4 2 Swarm Multi-agents, 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’intégrité des modifications ultérieures du code. 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 évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Créez un point de contrôle après les étapes coûteuses. La reprise du processus ne doit pas facturer à nouveau la même appel de LLM lorsque l’opérateur réessaie un nœud ultérieur. Lors du travail sur l’étape 4 2 Swarm Multi-agents, 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’intégrité des modifications ultérieures du code. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives de réessai, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’améliorations ultérieures.

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")

# Create a handoff tool for researcher -> analyst
transfer_to_analyst = create_handoff_tool(
    agent_name="analyst",
    description="Transfer user to the analyst assistant, who can create visualizations of provided data.",
)

# create research agent as a subagent
research_agent = create_agent(
    llm,
    tools=[wikipedia_tool, stock_data_tool, transfer_to_analyst],
    system_prompt="You provide summaries from Wikipedia, and can query load raw, numerical stock performance data.",
    name="researcher"
)

# define the checkpointer
checkpointer = InMemorySaver()

# create swarm agents
swarm_agent = create_swarm(
    agents=[research_agent, analyst_agent],
    default_active_agent="researcher"
).compile(checkpointer=checkpointer)

5. Superviseur Multi-Agents

La phase 5 Superviseur Multi-Agents fonctionne le mieux lorsqu’elle est considérée 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. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Maintenez l’état du graphe plat et typé. Les blocs imbriqués cachent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

5.1. Architecture de Superviseur Multi-Agents

L’architecture 5 1 des étapes fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemplaire 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. Nommez les artefacts, définez des vérifications de succès, et refusez toute mise en œuvre partielle silencieuse. Maintenez l’état du graphe plat et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

5.2. Mise en œuvre de superviseurs multi-agents

La phase 5 2 Supervisor Multi-agents fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, 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 processus passe de l’environnement de démonstration à des environnements partagés. Maintenez l’état du graphe plat et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption. La phase 5 2 Supervisor Multi-agents fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Documentez ensemble le parcours réussi et le parcours de récupération. Les tentatives de réessai, les contrôles humains et le traitement des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure.

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")

# create a checkpointer
checkpointer = InMemorySaver()

# create research subagent
research_agent = create_agent(
    llm,
    tools=[wikipedia_tool, stock_data_tool],
    system_prompt="You provide summaries from Wikipedia, and can query load raw, numerical stock performance data.",
    name="researcher"
)

# Create the supervisor multi-agent graph and compile it
supervisor = create_supervisor(
    model=llm,
    agents=[research_agent, analyst_agent],
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent. Assign research and data collection tasks to this agent\n"
        "- an analyst agent. Assign the creation of visualizations via code to this agent\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    add_handoff_back_messages=True,
    output_mode="last_message"
).compile(checkpointer=checkpointer)

6. Résumé final

Pour l’étape de résumé final, 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é. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Faites approuver par un humain les actions qui entraînent des dépenses ou modifient des données de production. La connexion en temps de compilation ne garantit pas la complétude du processus métier.

Liste de contrôle opérationnelle

L’étape de liste de contrôle opérationnelle fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un enregistrement type, un cas d’échec et 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 graphe.

Gardez l’état du graphe plat et typé. Les blocs imbriqués cachent le fait que tel nœud a écrit telle champ et empêchent la reprise après interruption.

Ajoutez un test de base qui exerce le chemin critique dans les processus d’intégration continue en utilisant des fixtures, et non des API payantes en temps réel, chaque fois que le budget le permet.

Dokumentez à la fois le parcours normal et celui de récupération. Les tentatives de répétition, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’une amélioration ultérieure.

Gardez l’état du graphe plat et typé. Les blocs imbriqués cachent le fait que tel nœud a écrit telle champ et empêchent la reprise après interruption.