Break multi-intent human messages into LangGraph subtask plans

Multi-agent supervisors often receive one human message that packs several intents. Without an explicit decomposition step, every team inherits the full text and latches onto the wrong topic—for example an API health check that contaminates a news-search query about markets versus technology.

The confusion pattern

A lead supervisor sits above team supervisors. The human message might ask to (1) check API health and (2) produce Google News search URLs for reliable sources that affect equities. The intent is that request-oriented work goes to a request_team that polls an API, while URL discovery goes to a search_team. In practice, when the entire message is forwarded unchanged, the health-check wording can distort the search topic (for example treating “API” as a technology beat instead of financial news). Traces in LangSmith show the same undirected payload reaching each team.

Dynamic subtask planning

The fix is to treat decomposition as a first-class LLM job: parse the free-form human text into a structured subtask plan the lead supervisor can delegate. Because topics change, the parser must stay dynamic rather than a brittle keyword router.

Introduce typed objects for Subtask, Routing, and a Subtask plan that nests a list of subtasks. Extend the shared agent state—the dictionary passed from the lead supervisor down through team supervisors and workers—so those fields travel with every hop.

Lead supervisor prompts and node

Prompt the lead supervisor to emit only the plan structure:

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"

Update the lead supervisor node so the first branch builds the plan before delegation:

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
            },
        )

Less relevant plumbing can be elided; the critical path is the initial if that turns the raw human message into per-team work. API-shaped tasks route to the request team; search-shaped tasks route to the search team.

Structured plan in the wild

A concrete object produced in tracing looks like:

{
  "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
    }
  }
}

With that object in state, each team receives only the slice it owns. Later, human-in-the-loop gates can approve suggested news URLs or API calls—especially important in regulated domains—without undoing the decomposition. The durable lesson: supervisors need a typed plan, not a shared megaphone of the original chat line.

Why undirected fan-out fails

When every subordinate team sees the same megaphone message, each model invents its own emphasis. Health-check vocabulary collides with topic vocabulary. Financial versus technology news is only one example; any multi-intent prompt will leak. Typed subtasks isolate vocabulary per responsibility so the search team never sees API probe wording and the request team never rewrites a news topic.

State fields that must travel

Beyond the nested subtask list, keep identifiers for the originating human message, per-subtask status (pending, delegated, done, failed), and the target team name. Supervisors update status as work completes so a second planning pass can skip finished items. Without status, the lead supervisor may re-emit duplicate work on retries.

Production notes

• Prefer structured output / tool calling for the plan schema over free JSON in prose.
• Log the plan object beside LangSmith traces so operators can see why a team was skipped.
• Add HITL before irreversible side effects (paid API calls, outbound posts) even after decomposition succeeds.
• Re-plan when the human edits the original message mid-run; stale subtasks should be cancelled explicitly.

Decomposition is not optional polish—it is the difference between a supervisor graph and a confused committee.