Teilen Sie mehrzweckige menschliche Nachrichten in LangGraph-Unteraufgabenpläne auf.

Mehr-Agenten-Überwacher erhalten oft eine menschliche Nachricht, die mehrere Absichten enthält. Ohne einen expliziten Aufschlüsselungsschritt übernimmt jedes Team den gesamten Text und konzentriert sich auf das falsche Thema – beispielsweise eine API-Überprüfung, die eine Suchanfrage nach Nachrichten zu Märkten mit einer Anfrage zu Technologie vermischt.

Das Verwirrungsmuster

Ein leitender Supervisor steht über den Team-Supervisoren. Die menschliche Anfrage könnte darum bitten, (1) den Zustand der API zu überprüfen und (2) Such-URLs für Google News zu erstellen, die auf zuverlässige Quellen verweisen, die den Aktienmarkt beeinflussen. Ziel ist es, dass anfragespezifische Aufgaben an ein request_team weitergeleitet werden, das die API abfragt, während die URL-Erstellung einem search_team übertragen wird. In der Praxis kann die Formulierung zur Überprüfung des Zustands bei einer unveränderten Weiterleitung der gesamten Anfrage das Suchthema verfälschen – beispielsweise indem „API“ als Technologie-Thema statt als Finanznachricht interpretiert wird. Die Aufzeichnungen in LangSmith zeigen, dass derselbe unstrukturierte Datensatz jede Gruppe erreicht.

Dynamische Unteraufgabenplanung

Die Lösung besteht darin, die Aufteilung in strukturierte Unteraufgaben als eigenständige Aufgabe für LLMs zu betrachten: Der freiformige menschliche Text muss in einen strukturierten Plan umgewandelt werden, den der leitende Supervisor weitergeben kann. Da sich die Themen ändern, muss der Parser dynamisch bleiben und nicht auf starren Schlüsselwörtern beruhen.

Führen Sie typisierte Objekte für Unteraufgabe, Routing sowie einen Unteraufgabenplan ein, der eine Liste von Unteraufgaben enthält. Erweitern Sie den gemeinsamen Agentenzustand – das Dictionary, das vom leitenden Supervisor über die Teamsupervisoren bis zu den Arbeitern weitergegeben wird – sodass diese Felder bei jeder Weiterleitung mitübertragen werden.

Aufforderungen des leitenden Supervisors und Knoten

Auffordern Sie den leitenden Supervisor, nur die Planstruktur auszugeben:

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"

aktualisieren Sie den Knoten des leitenden Supervisors so, dass die erste Abzweigung den Plan erstellt, bevor die Delegation erfolgt:

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

Weniger relevante Schritte können weggelassen werden; der kritische Pfad ist die ursprüngliche if-Bedingung, die die rohe menschliche Nachricht in teambezogene Aufgaben umwandelt. API-ähnliche Aufgaben werden an das entsprechende Anfragen-Team weitergeleitet; suchbasierte Aufgaben an das Suchteam.

Gestrukturer Plan in der Praxis

Ein konkretes Objekt, das bei der Nachverfolgung erzeugt wird, sieht wie folgt aus:

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

Sobald dieses Objekt im gewünschten Zustand ist, erhält jedes Team nur den Teil, der ihm zusteht. Später können menschliche Prüfer vorgeschlagene News-URLs oder API-Aufrufe genehmigen – insbesondere in regulierten Bereichen – ohne die vorherige Aufteilung rückgängig zu machen. Die wichtige Lektion: Überwacher benötigen einen strukturierten Plan, nicht einfach nur eine gemeinsame Kommunikationsplattform wie die ursprüngliche Chat-Leiste.

Warum die ungerichtete Verteilung versagt

Da jedes untergeordnete Team dieselbe Nachricht erhält, entwickelt jedes Modell seine eigene Betonung. Das Vokabular zur Überprüfung der Qualität kollidiert mit dem Themenvokabular. Finanz- und Technologie-Nachrichten sind nur ein Beispiel; jede Anfrage mit mehreren Zielen führt zu Fehlern. Strukturierte Unteraufgaben isolieren das Vokabular je nach Verantwortungsbereich, sodass das Suchteam niemals die Formulierungen der API-Anfragen sieht und das Anfrage-Team niemals einen News-Thema umschreibt.

Zustandsfelder, die übertragen werden müssen

Neben der verschachtelten Liste der Unteraufgaben sollten Identifikatoren für die ursprüngliche menschliche Nachricht, den Status jeder Unteraufgabe (ausstehend, delegiert, abgeschlossen, fehlgeschlagen) sowie den Namen des Zielteams gespeichert werden. Aufsichtspersonen aktualisieren den Status, sobald die Arbeiten abgeschlossen sind, damit bei einer erneuten Planung fertige Einträge übersprungen werden können. Ohne Status könnte der leitende Aufsichtsperson bei erneuten Versuchen doppelte Arbeiten freigeben.

Produktionshinweise

• Bevorzugen Sie eine strukturierte Ausgabe bzw. Toolaufrufe gemäß dem Plan-Schema statt freien JSON-Textes in Prosaform.
• Protokollieren Sie das Plan-Objekt zusammen mit den LangSmith-Spuren, damit die Bediener erkennen können, warum ein Team übersprungen wurde.
• Fügen Sie HITL vor irreversiblen Nebeneffekten (bezahlte API-Aufrufe, ausgehende Beiträge) hinzu – auch dann, wenn die Dekomposition erfolgreich war.
• Planen Sie erneut, wenn der Mensch die ursprüngliche Nachricht während des Laufs bearbeitet; veraltete Unteraufgaben sollten explizit storniert werden.

Die Dekomposition ist keine optionale Verfeinerung – sie ist der Unterschied zwischen einem Überwachungsgraphen und einem verwirrten Ausschuss.