Розділіть багатоцільові людські повідомлення на плани підзавдань LangGraph

Керувачі з багатьма агентами часто отримують одне повідомлення від людини, яке містить кілька намірів. Без чіткого кроку розбирання кожна команда отримує повний текст та присвячує увагу неправильній темі — наприклад, перевірка стану API затьмарює запит на пошук новин про ринки замість технологій.

Патерн плутанини

Головний керівник знаходиться над керівниками команд. У людському повідомленні може бути прохання (1) перевірити стан API та (2) створити URL-адреси пошуку в Google News для надійних джерел, які впливають на акції. Мета полягає у тому, щоб робота, орієнтована на запити, надходила до request_team, який перевіряє API, тоді як пошук URL — до search_team. На практиці, коли все повідомлення передається без змін, формулювання щодо перевірки стану може спотворити тему пошуку (наприклад, розглядати „API“ як технологічну новину замість фінансової). Дані LangSmith показують, що однаковий неструктурований вміст надходить до кожної команди.

Динамічне планування підзавдань

Рішення полягає у тому, щоб розглядати розкладку завдань як основну функцію LLM: перетворювати вільноформатний текст людини на структурований план підзавдань, який може бути доручений головному керівнику. Оскільки теми змінюються, парсер має залишатися динамічним, а не просто механізмом розподілу за ключовими словами.

Введіть об’єкти типу Підзавдання, Маршрутизація та План підзавдань, який містить список підзавдань. Розширте спільний стан агента — словник, який передається від головного керівника через керівників команд та працівників — щоб ці поля супроводжували кожен крок обробки.

Запити головного керівника та вузол

Запитайте головного керівника видати лише структуру плану:

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"

Оновіть вузол головного керівника так, щоб перша гілка створювала план перед делегуванням:

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

Н менш важливі елементи можна опустити; критичний шлях — це початкова умова if, яка перетворює первинне повідомлення людини на роботу для кожної команди. Завдання у форматі API направляються до відповідної команди; завдання у форматі пошуку — до команди з пошуком.

Структурований план у реальних умовах

Конкретний об’єкт, створений під час відстеження, виглядає так:

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

У такому стані об’єкта кожна команда отримує лише ту частину, яка їй належить. Пізніше люди, які контролюють процес, можуть схвалити запропоновані URL новин чи виклики API — що особливо важливо у регульованих сферах — не скасовуючи при цьому процес розбиття на частини. Важливий урок: керівникам потрібен типований план, а не спільний „мегафон“ початкової лінії чату.

Чому некероване розповсюдження інформації не спрацьовує

Коли кожна підпорядкована команда бачить одне й те саме повідомлення, кожна модель створює власний акцент. Лексика перевірки стану зіткнеться з лексикою теми. Новини про фінанси та технології — це лише один приклад; будь-який запит з кількома цілями призведе до витоку інформації. Типовані підзавдання ізолюють лексику за обов’язками, тож команда з пошуку ніколи не бачить формулювань API, а команда з обробки запитів ніколи не переписує тему новин.

Поля стану, які мають передаватися

Окрім вкладених списків підзавдань, необхідно зберігати ідентифікатори початкового повідомлення від людини, статус кожного підзавдання (очікується, делеговано, виконано, зазнало невдачі) та назву цільової команди. Керівники оновлюють статус після завершення роботи, щоб під час наступного планування можна було пропустити вже виконані завдання. Без статусу головний керівник може знову видати дублікати завдань під час спроб їх виконання.

Примітки до використання

• Краще використовувати структурований формат вихідних даних чи інструменти для зберігання схеми плану, ніж вільний JSON у прозовому форматі.
• Записуйте об’єкт плану разом із записами LangSmith, щоб оператори могли бачити, чому певна команда була пропущена.
• Додавайте позначку HITL перед завданнями з незворотними наслідками (оплачувані виклики API, надсилання повідомлень) навіть у разі успішного розкладання завдань.
• Плануйте заново, якщо людина редагує початкове повідомлення під час виконання; застарілі підзавдання слід явно скасовувати.

Розкладання не є додатковою опцією — це різниця між графом керування та заплутаним комітетом.