Wiele rąk, jeden pióro: siedem zasad orkiestracji agentów AI
Siedem zasad zarządzania agentami AI: kiedy zwycięża prosty proces pracy, dlaczego orkiestrator nie posiada narzędzi, powrót podagenta zapisanego w formie tekstu oraz warunki zatrzymania.
Źródła: Many Hands, One Pen: Orchestrating Agents Without Losing the Plot
Wyobraźmy sobie Marynarkę Wojenną USA z 1942 roku, która borykała się z problemem koordynacji, który wiele mówi o współczesnym oprogramowaniu. Okręty wojenne były wyposażone w radar, sonar, odbiory radiowe oraz ludzkich obserwatorów, a każdy raport docierał jednocześnie na mostek w losowej kolejności. Kapitan próbujący prowadzić bitwę tonął w nieprzetworzonych sygnałach.
Odpowiedzią Marynarki było pomieszczenie bez okien, zwane Centrum Informacji Bojowej. Każdy raport z czujników był zaznaczany na jednym wspólnym wykresie przy użyciu ustalonego formatu. Operatorzy, którzy odczytywali pomiary, nie wydawali żadnych rozkazów; ich zadaniem było utrzymywanie dokładności wykresu, podczas gdy jeden oficer sam decydował o kolejnym ruchu okrętu. Informacje docierały szeroko rozproszone, decyzje skupiały się w jednym punkcie, a standardowy format raportów umożliwiał porównywanie poszczególnych danych.
Systemy wieloagentowe wciąż odkrywają ten „pokój”. Etap, w którym „dodanie kolejnego agenta” było rozwiązaniem wszystkich problemów, już minął, a rok 2026 w dużej mierze upłynął na mierzeniu kosztów tego rozwiązania. Jedynym wzorcem, który przetrwał, jest ten, który przyjąła marynarka wojenna: praca równoległa, raportowanie w uporządkowanym formacie, podejmowanie decyzji w jednym miejscu oraz jasne zdefiniowanie kryteriów ukończenia zadania.
Korzystna perspektywa: większość problemów wieloagentowych w rzeczywistości nie dotyczy samych agentów. Są to problemy kontekstowe oraz problemy przepływu sterowania w przebraniu agentów. Przebieg zadania zazwyczaj idzie nie tak z powodu podjęcia przez dwa agenty niekompatybilnych decyzji, których żaden z nich nie mógł dostrzec, z powodu utraty znaczenia informacji po przekroczeniu pewnej granicy, albo dlatego, że nikt nie zapisał, jak powinno wyglądać ukończenie zadania.
Następuje siedem praktyk, z których każda wynika z błędów w systemach już wdrożonych w praktyce. Razem pomagają ocenić, kiedy zwykły przepływ pracy jest lepszy od agenta, kiedy orkiestrator musi sfinansować swoje tokeny, jak dzielić pracę według granic kontekstu, dlaczego orkiestrator powinien nie ingerować w narzędzia, jak zapewnić subagentom umowę zwrotu, którą można zweryfikować, oraz jak celowo zakończyć proces, a nie z powodu kaprysu modelu.
Używaj zwykłych przepływów pracy jako standardu; autonomię trzeba zdobyć
Co idzie nie tak. Weźmy proces zwrotu pieniędzy z czterema możliwymi wynikami, z których każdy jest już objęty pisemną zasadą. Zespół otacza go autonomicznym pętlą, sześcioma narzędziami oraz ciągiem celów. Nagle ticket z wtorku podąża inną ścieżką niż ten z poniedziałku, a każdy jego przeprowadzenie kosztuje od dziesięciu do pięćdziesięciu razy więcej niż w starym kodzie. Dokładność pozostaje niezmienna, ponieważ żaden element tego procesu nie jest na tyle niejednoznaczny, by wymagał zastosowania inteligencji.
Rozwiązanie. Najpierw narysuj schemat przepływu. Wszystko, co można zaplanować z wyprzedzeniem, należy do zwykłego przepływu sterowania, więc prosty kod obsługuje każdą możliwą gałąź. Model powinien być wywoływany tylko w tym kroku, w którym dane wejściowe rzeczywiście mogą być interpretowane różnie. Agent powinien uzyskać uprawnienia do obsługi danej gałęzi tylko wtedy, gdy udowodniono, że nie da się jej sformułować w słowach.
REFUND_LIMIT = 50.00
def llm_step(prompt: str) -> str:
"""Called only where the written rules run out."""
return model.complete(prompt).strip().lower()
def handle_refund(ticket):
if ticket.days_since_purchase > 30:
return deny(ticket, reason="outside window")
if ticket.amount <= REFUND_LIMIT:
return approve(ticket) # a rule, not a judgement call
intent = llm_step(
f"Classify as fraud, defect, or remorse:\n{ticket.body}"
)
if intent == "fraud":
return escalate(ticket, queue="risk")
if intent == "defect":
return approve(ticket)
return route_to_human(ticket)
Problem. Można przejść w zbyt skomplikowany labirynt nawarstwionych warunków, które w tle kodują zasady oceny, których modele tak naprawdę nie mogą stosować. Gdy pisze się piąty warunek dotyczący tekstu swobodnego, to właśnie ta niejednoznaczność jest powodem istnienia takich modeli.
Pomijaj orkiestratora, gdy problem jest już znany
Co idzie nie tak. Proces generowania raportu składający się z czterech kroków zostaje otoczony pętlą orkiestratora ograniczoną do piętnastu iteracji. Każda iteracja poświęca czas na planowanie kolejnych działań, a każde takie rozważanie prowadzi do tej samej odpowiedzi, którą już ustalono: pobrać, wyekstrahować, podsumować, sformatować. To oznacza piętnaście takich procesów rozumowania, po kilka tysięcy tokenów każdy, aby ponownie odkryć listę, którą można by wpisać w pięciu liniach.
Rozwiązanie. Zastosuj szybką zasadę przed napisaniem jakiegokolwiek kodu. Gdy lista podzadań jest znana przed rozpoczęciem wykonywania, uruchom ją jako prosty pętla z ustalonym rozdzieleniem zadań i zachowaj tokeny do innych celów. Orkiestracja oparta na modelach nadaje się tam, gdzie sam podział zadań jest nieznany lub musi ewoluować w oparciu o wyniki wcześniejszych kroków.
# Before: up to 15 planning calls to rediscover a fixed list
def run_orchestrated(doc):
state = {"doc": doc}
for _ in range(15):
decision = orchestrator.plan(state) # 1 LLM call per pass
if decision.action == "finalize":
break
state = WORKERS[decision.worker](state)
return state
# After: 0 planning calls, same four workers, same result
PIPELINE = [fetch, extract, summarize, format_report]
def run_static(doc):
state = {"doc": doc}
for step in PIPELINE: # 0 LLM calls here
state = step(state)
return state
Problem. Ta rada ma swoją drugą stronę. System, który co miesiąc tworzy nowy przypadek specjalny, lub w którym wynik drugiego kroku faktycznie decyduje o tym, czy zostanie wykonywany trzeci krok, naprawdę planuje. Usunięcie takiego planisty jedynie przenosi to rozgałęzienie do jakiegoś pliku konfiguracyjnego, który nikt nie będzie pielęgnował.
Dziel agenty według kontekstu, a nie według tytułu stanowiska
Co idzie nie tak. Wersja diagramu organizacyjnego systemu wielu agentów wygląda schludnie: planista, implementator, tester, recenzent – praca nad poszczególnymi funkcjami w kolejności. W praktyce każda przekazanie obowiązków wiąże się z założeniami, których nikt nie zapisał. Tester sprawdza zachowanie, którego implementator wcale nie zamierzał, a recenzent sprzeciwia się wyborowi dokonanemu celowo przez planistę. Płacisz cztery razy za efekt degradacji wynikającej z takiego podejścia, a jeden agent wykonywający wszystkie cztery zadania osiągnąłby lepsze rezultaty.
Rozwiązanie. Należy przeciąć wzdłuż linii kontekstowych. Zanim dodamy drugiego agenta, należy zadać pytanie projektowi: czy ta podzadanie może zostać zakończona w oknie kontekstowym, które nie ma nic wspólnego z oknem sąsiedniego? Jeśli odpowiedź brzmi „nie”, oba należą do jednego agenta, i to właśnie dlatego agent tworzący funkcjonalność powinien również napisać dla niej testy.
from dataclasses import dataclass
@dataclass
class Subtask:
name: str
needs: set[str] # facts this subtask must read
produces: set[str] # facts this subtask decides
def should_split(a: Subtask, b: Subtask) -> bool:
"""Split only when neither side needs what the other decides."""
shared = (a.needs & b.produces) | (b.needs & a.produces)
return not shared
implement = Subtask("implement", {"spec"}, {"api_shape", "error_semantics"})
test = Subtask("test", {"spec", "api_shape", "error_semantics"}, {"cases"})
assert not should_split(implement, test) # one agent writes both
Problem. Naprawdę izolowane podzadania są rzadsze, niż sugeruje schemat organizacyjny, więc test często wspomina o „jednym agencie”, a to jest właściwa odpowiedź, a nie brak wyobraźni. Dzielenie zadań ma sens w przypadku prac wymagających szerokiego zakresu, takich jak czytanie pięcioma agentami pięciu niepowiązanych ze sobą źródeł, ponieważ decyzja jednego nie zmienia tego, czego potrzebuje inny.
Niech orkiestrator wyprowadza decyzję i nigdy nie wywołuje narzędzia
Co idzie nie tak. Orkiestrator otrzymuje narzędzie do wyszukiwania w sytuacjach awaryjnych. Wkrótce przestaje przydzielać zadania i zaczyna samodzielnie odpowiadać na podzadania. Surowe wyniki z narzędzia gromadzą się w jego oknie, przez co planowanie staje się coraz gorsze, właśnie w momencie, gdy czas trwania zadania się wydłuża, a planowanie ma największe znaczenie. Utrudnia się również śledzenie, ponieważ nie jest jasne, który agent stworzył jakie ustalenie.
Rozwiązanie. Nie powinny przychodzić żadne dodatkowe plany, więc nie należy dawać koordynatorowi żadnych narzędzi. Wysyła on tylko jedną zwięzłą, sformatowaną decyzję: delegować, przeplanować lub zakończyć, wraz z wybranym pracownikiem i tekstem zadania. Otaczający go pętla odczytuje tę decyzję, uruchamia wskazanego pracownika i przechowuje wynik w wspólnym stanie, gotowy do następnej decyzji.
from typing import Literal
from pydantic import BaseModel
class OrchestratorDecision(BaseModel):
next_action: Literal["delegate", "replan", "finalize"]
target_worker: str | None = None
task_description: str | None = None
reasoning: str
def orchestrator_step(state):
d = decide(state) # model has zero tools attached
if d.next_action == "finalize":
return finalize(state, d.reasoning)
if d.next_action == "replan":
return state.reset_plan(d.reasoning)
worker = WORKERS[d.target_worker] # workers own every tool
result = worker.run(d.task_description)
return state.record(d.target_worker, result)
Problem. Bez narzędzi koordynator widzi tylko to, co przysyłają pracownicy. Pracownik, który zwraca zwięzły podsumowanie, pozostawia go bez możliwości planowania. Zamiast dostępu do narzędzi w czasie rzeczywistym, oferuj okno tylko do odczytu stanu współdzielonego oraz upewnij się, że dane zwracane przez pracowników są wystarczająco ustrukturyzowane, aby można było na ich podstawie planować.
Wymagaj od każdego podagenta przestrzegania ustalonego kontraktu zwracania danych
Co idzie nie tak. Weźmy podagenta zajmującego się badaniami, który zwraca 2500-słowny dziennik wszystkich prób, włączając te, które zakończyły się niepowodzeniem. Pomnóż to przez siedmiu pracowników – około 20 000 tokenów trafia do okna nadrzędnego agenta, zanim rozpocznie się jakakolwiek syntezę. Delegacja miała chronić to okno. W rzeczywistości eksploracja jest realizowana dwukrotnie, a nadrzędny agent realizuje instrukcje jeszcze gorzej, właśnie wtedy, gdy jest to najważniejsze.
Rozwiązanie. Traktuj powrót jako granicę API, przy czym każdy podagent ma swoją specyfikację odpowiedzi. Jego wiadomość kończąca to JSON zawierający nazwane pola: ustalenia, źródła, nierozwiązane pytania oraz status zakończenia. Sprawdź ją po przybyciu i odrzuć wszystko, co jest niepoprawnie sformatowane lub zbyt długie, aby można je było ponownie wysłać – w ten sposób nigdy nie dotrą one do agenta nadrzędnego.
from pydantic import BaseModel, Field, ValidationError
class SubagentResult(BaseModel):
findings: list[str] = Field(max_length=5) # hard cap, one sentence each
sources: list[str]
open_questions: list[str] = []
completion_status: str # complete | partial | blocked
def parse_or_retry(worker, task, attempts=2):
for _ in range(attempts):
raw = worker.run(task, response_format=SubagentResult)
try:
return SubagentResult.model_validate_json(raw)
except ValidationError as err:
task = f"{task}\n\nRejected: {err}\nReturn only JSON in the schema."
raise RuntimeError(f"{worker.name} returned no valid result")
Problem. Schematy określają jedynie minimalną wartość, nic więcej. Pracownik nadal może umieścić akapit liczący 400 słów w jednej ciągu danych i dostarczyć identyczny, niepotrzebnie rozbudowany tekst w formacie JSON. Należy ograniczyć zarówno liczbę elementów, jak i długość każdego pola, a w tekście odrzucenia podać nazwę pola, które przekroczyło limit, aby próba ponowna wiedziała, co naprawić.
Paralelizacja odczytów i kierowanie zapisów przez jeden agent
Co idzie nie tak. Dwa agenty pracują jednocześnie nad tym samym elementem i oba zgłaszają sukces. Jeden narysował tło w stylu Super Mario, a drugi sprite Flappy Bird, przy czym oba przechodzą testy napisane pod kątem ich własnych wizji projektu. Połączenie tych wyników nie może rozstrzygnąć wyborów, o których żaden z agentów nie wiedział, więc jeden lub oba wyniki zostają odrzucone.
Rozwiązanie. Czytanie może odbywać się równolegle, natomiast pisanie nie. Ogranicz każdego pracownika do listy narzędzi tylko do odczytu, co umożliwia rozszerzenie fazy zbierania danych w zależności od potrzeb, i nigdy nie udzielaj pracownikowi żadnego narzędzia do zapisu. Wtedy artefakt jest wytwarzany w jednej, bezwarunkowej wywołaniu, która otrzymuje wyniki wszystkich pracowników jednocześnie.
import asyncio
READ_TOOLS = ["search_repo", "read_file", "fetch_docs"]
async def gather_context(subtasks):
workers = [Agent(name=t.name, tools=READ_TOOLS) for t in subtasks]
return await asyncio.gather(
*(w.run(t.prompt) for w, t in zip(workers, subtasks))
)
async def build_feature(spec, subtasks):
findings = await gather_context(subtasks) # wide, parallel, read-only
writer = Agent(name="writer", tools=["write_file", "apply_patch"])
return await writer.run(spec, context=findings) # one writer, one pass
Problem. Ten jeden pisarz stanowi wąskie gardło i musi jednocześnie przechowywać wyniki wszystkich pracowników, więc napotkasz ograniczenie ilości tokenów znacznie wcześniej niż limit logiczny. Jeśli artefakt rzeczywiście dzieli się czysto, jak w przypadku trzech plików, które nie mają ze sobą żadnego interfejsu, przypisz każdemu plikowi osobnego pisarza i zachowuj tę samą dyscyplinę w każdym z nich.
Zapisz warunki zakończenia zapisu
Co idzie nie tak. Proces się kończy, nie pojawia się błąd, ale praca pozostaje niedokończona. Analiza 1 600 śladów działania wielu agentów w siedmiu frameworkach wykazała, że 12,4% awarii wynika z tego, że agenci nie znają swoich warunków zatrzymania, a kolejne 6,2% dotyczy przedwczesnego zatrzymania się. Niemal jedna awaria na pięć objawia się więc jako uporządkowane zakończenie z częściowo ukończonym wynikiem.
Rozwiązanie. Zamień linię mety na coś konkretnego, a nie na zwykłe uczucie. Przydziel każdemu zadaniu wyraźną listę kontrolną jego ukończenia oraz umieść kontrolkę zatwierdzenia, opartą na pierwotnym celu, na jedynym ścieżce prowadzącej do stanu końcowego. W tych samych badaniach stwierdzono, że stosowanie bramy weryfikacyjnej zwiększyło skuteczność ChatDev o 9,4%, natomiast sprawdzanie celów wysokiego poziomu — o 15,6%.
CHECKLIST = [
"every requested section exists",
"each claim cites a retrieved source",
"open questions are listed, or explicitly none",
]
def verify_complete(objective, checklist, output) -> bool:
for item in checklist:
verdict = judge(f"Objective: {objective}\nCheck: {item}\n\n{output}")
if not verdict.passed:
log.info("termination blocked by: %s", item)
return False
return judge(f"Does this satisfy the objective?\n{objective}\n\n{output}").passed
def finish(state):
if verify_complete(state.objective, CHECKLIST, state.draft):
return state.done()
return state.keep_working(reason="checklist not satisfied")
Problem. Jeśli pozwolisz modelowi produkcyjnemu oceniać własną listę kontrolną, on po prostu zatwierdzi swoją pracę. Staraj się używać elementów mechanicznych, gdzie to możliwe – np. obecności pliku, wypełnionego pola lub poprawnego źródła. Brama, która nigdy się nie otwiera, stanowi sam w sobie rodzaj awarii, dlatego ogranicz liczbę prób i wyświetl błąd po ich wyczerpaniu.
Od czego zacząć
Jeśli możesz zmienić tylko jedną rzecz, wybierz najtańsze rozwiązanie: zdefiniuj na piśmie, co oznacza ukończenie zadania. Lista kontrolna w połączeniu z krokiem weryfikacji przed osiągnięciem stanu końcowego zajmuje zaledwie kilka godzin, działa zarówno dla jednego agenta, jak i całej floty, i eliminuje całą klasę ukrytych awarii. Uczyń to zadaniem na ten tydzień.
Następnie przychodzi krok usuwania. Przyjrzyj się swojemu orkiestratorowi i sprawdź, czy jego opcje sprowadzają się do ustalonej listy oraz funkcji agregacji. Jeśli tak, usuń go – zużycie tokenów i opóźnienia spadną, a wyniki pozostaną stabilne. Ponieważ łatwo go cofnąć, stanowi bezpieczny pierwszy krok.
Chroń cięższe narzędzia do momentu, gdy staną się konieczne. Kontrakty zwrotne typu „typed” są opłacalne przy około pięciu pracownikach i wzwyż, a lista uprawnień typu „read-only” ma znaczenie, gdy dwa agenty naprawdę piszą jednocześnie. Wyjątkiem jest orkiestrator bez narzędzi: niemal bezkosztowy na początku, ale trudny do wdrożenia później.
Rozwiązania, które przetrwają do 2026 roku, nie charakteryzują się największą liczbą agentów. W takich rozwiązaniach połączenia między agentami są najmniejsze i najbardziej oczywiste w projekcie. Mniej osób, mocniejsze kontrakty, jeden długopis.
Literatura pokrewna
- Agenci kodujące równolegle z użyciem worktree’ów Git, dyrygenta oraz Antigravity 2.0 — Jak agent koordynujący, worktree’y Git dla poszczególnych agentów oraz ścieżki oparte na problemach umożliwiają stworzenie kilkunastu funkcji SRE w ciągu około pięciu godzin, plus wskazówki dotyczące pytań HITL i zasad higieny łączenia zmian.
- RAG oparty na agentach z Agno Teams: przepisywanie zapytań, pętla krytyki i Qdrant — Zespół koordynujący przepisuje zapytania, powtarza proces wyszukiwania za pomocą krytyka i udziela odpowiedzi wyłącznie na podstawie zatwierdzonego kontekstu, wykorzystując w Qdrant hybrydowe algorytmy wyszukiwania typu dense, BM25 oraz ColBERT.