Разбіліце багатоцелевыя поведамленні людзей на планы падзеў 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, выходныя публікацыі) нават пасля успеху дэкомпазіі.
• Перапланаваць, калі чалавек рэдагуе выначальнае паведамленне пад час адработкі; застарэлыя падзябныя задачы трэба явна анулюваць.

Разбір на складовыя — гэта не проста допамінг, а разлік межы графа керавання і заплутанага комітэту.