Découper les messages humains à multiples intentions en plans de sous-tâches LangGraph

Les superviseurs multi-agents reçoivent souvent un seul message humain contenant plusieurs intentions. En l’absence d’une étape de décomposition explicite, chaque équipe hérite du texte intégral et s’attache au sujet incorrect — par exemple, une vérification de l’état d’une API qui contamine une requête de recherche d’actualités portant sur les marchés plutôt que sur la technologie.

Le schéma de confusion

request_team qui interroge l’API, tandis que la recherche d’URL est gérée par une search_team. En pratique, lorsque le message complet est transmis tel quel, la formulation relative à la vérification de l’état peut fausser le sujet de recherche (par exemple en considérant « API » comme une actualité technologique plutôt que des nouvelles financières). Les traces générées par LangSmith montrent que le même contenu non structuré atteint chaque équipe.

Planification dynamique des sous-tâches

Introduisez des objets structurés pour la Sous-tâche, le Encadrement et un Plan de sous-tâches qui contient une liste de sous-tâches. Étendez l’état partagé de l’agent — le dictionnaire transmis depuis le superviseur principal jusqu’aux superviseurs d’équipe et aux travailleurs — afin que ces champs soient transmis à chaque étape.

Indications pour le superviseur principal et le nœud

Poussez le superviseur principal à ne générer que la structure du plan :

def build_top_graph(model_str, config):
    model = init_chat_model(model_str, temperature=0, max_retries=3, timeout=30)
    PLAN_PROMPT = (
        "Break the user's request into independent subtasks. Each subtask goes to "
        "exactly one team:\n"
        "- 'request_team': checks API health and provides site urls as data to API.\n"
        "- 'search_team': finds relevant sites for a topic and returns full Google "
        "News search URLs.\n\n"
        "A single team may own multiple subtasks — list them separately, each with "
        "self-contained instruction text covering only that piece of the request."
    )

    ROUTE_PROMPT = (
        "Below are the remaining pending subtasks with their ids, teams, and "
        "instructions. Pick the id of the one to dispatch next. If the list is "
        "empty, respond with next_subtask_id='FINISH'."
    )
    FINISH_TOKEN = "FINISH"

Mettez à jour le nœud du superviseur principal de manière que la première branche crée le plan avant toute délégation :

def supervisor_node(
        state: SupervisorState, config: RunnableConfig
    ) -> Command[Literal["request_team", "search_team", "__end__"]]:
        # Phase 1: plan subtasks once, on first entry
        if not state.get("subtasks_planned"):
            # Supervisor planning prompt + human input:
            messages = [SystemMessage(content=PLAN_PROMPT)] + state["messages"]
            # Model to decompose human input into subtasks
            plan = model.with_structured_output(SubtaskPlan).invoke(
                messages, config=config
            )
            # Capture output from model including instruction for each team
            subtasks: list[Subtask] = [
                {
                    "id": str(uuid.uuid4())[:8],
                    "team": s["team"],
                    "instruction": s["instruction"],
                    "status": "pending",
                }
                for s in plan.subtasks
            ]
            state = {**state, "subtasks": subtasks, "subtasks_planned": True}

        pending = [s for s in state["subtasks"] if s["status"] == "pending"]

        if not pending:
            return Command(goto="__end__", update={"next": FINISH_TOKEN})

        def _is_blocked(s: Subtask) -> bool:
          ...
        dispatchable = [s for s in pending if not _is_blocked(s)]
        if not dispatchable:
            return Command(...)
        if len(dispatchable) == 1:
            chosen = dispatchable[0]
        else:
            listing = "\n".join(
                f"- id={s['id']} team={s['team']} instruction={s['instruction']!r}"
                for s in dispatchable
            )
            decision = model.with_structured_output(RoutingDecision).invoke(
                [SystemMessage(content=ROUTE_PROMPT + "\n\n" + listing)],
                config=config,
            )
            if decision.next_subtask_id == FINISH_TOKEN:
                return Command(goto="__end__", update={"next": FINISH_TOKEN})
            chosen = next(
                s for s in dispatchable if s["id"] == decision.next_subtask_id
            )
        return Command(
            goto=chosen["team"],
            update={
                "next": chosen["team"],
                "current_instruction": chosen["instruction"],
                "subtasks": state["subtasks"],
                "subtasks_planned": True,
                "_dispatched_id": chosen["id"],  # so the team node can mark it done
            },
        )

Les éléments moins importants peuvent être omis ; le chemin critique correspond à la condition if initiale qui transforme le message humain brut en tâches spécifiques à chaque équipe. Les tâches de type API sont acheminées vers l’équipe concernée par la demande ; les tâches de type recherche sont acheminées vers l’équipe de recherche.

Plan structuré en situation réelle

{
  "output": {
    "goto": "request_team",
    "graph": null,
    "resume": null,
    "update": {
      "_dispatched_id": "15c6885c",
      "current_instruction": "Check the API health/status. Report whether the API is healthy and available for use.",
      "next": "request_team",
      "subtasks": [
        {
          "id": "15c6885c",
          "instruction": "Check the API health/status. Report whether the API is healthy and available for use.",
          "status": "pending",
          "team": "request_team"
        },
        {
          "id": "01541ab0",
          "instruction": "Find relevant reliable news sites for latest stock-market-affecting news and return full Google News search URLs for each site, using query terms focused on current market-moving news and reliability.",
          "status": "pending",
          "team": "search_team"
        }
      ],
      "subtasks_planned": true
    }
  }
}

Avec cet objet dans cet état, chaque équipe ne reçoit que la partie qui lui appartient. Plus tard, des contrôles manuels peuvent valider les URL de nouvelles ou les appels API suggérés — ce qui est particulièrement important dans des domaines réglementés — sans annuler la décomposition. La leçon essentielle : les superviseurs ont besoin d’un plan structuré, et non d’un mégaphone partagé provenant de la ligne de chat initiale.

Pourquoi la diffusion non dirigée échoue

Lorsque chaque équipe subordonnée voit le même message diffusé par le mégaphone, chaque modèle invente sa propre interprétation. Le vocabulaire utilisé pour les vérifications de santé entre en conflit avec celui relatif au sujet. Les nouvelles financières par rapport aux technologies ne sont qu’un exemple ; tout prompt à plusieurs objectifs entraînera des fuites d’informations. Les sous-tâches structurées isolent le vocabulaire selon les responsabilités, de sorte que l’équipe chargée des recherches ne voit jamais le texte des requêtes API et que l’équipe chargée des demandes ne réécrit jamais un sujet de nouvelle.

Champs d’état qui doivent être transmis

En plus de la liste imbriquée des sous-tâches, conservez les identifiants du message initial émis par l’humain, l’état de chaque sous-tâche (en attente, déléguée, terminée, échouée) ainsi que le nom de l’équipe cible. Les superviseurs mettent à jour l’état au fur et à mesure que le travail est achevé, afin qu’une deuxième phase de planification puisse ignorer les éléments déjà terminés. Sans indication d’état, le superviseur principal pourrait réassigner à nouveau des tâches déjà effectuées lors de nouvelles tentatives.

Notes de production

• Préférez une sortie structurée ou des appels d’outils conformes au schéma du plan plutôt qu’un JSON libre sous forme de texte narratif.
• Enregistrez l’objet du plan aux côtés des traces générées par LangSmith afin que les opérateurs puissent comprendre pourquoi une équipe a été ignorée.
• Ajoutez HITL avant toute action ayant des effets secondaires irréversibles (appels API payants, publications externes), même après un décomposition réussie.
• Réplanifiez lorsque l’humain modifie le message original en cours d’exécution ; les sous-tâches obsolètes doivent être annulées explicitement.

La décomposition n’est pas une touche décorative optionnelle : c’est ce qui distingue un graphe de superviseur d’un comité confus.