Wskazówki praktyczne: Twój graf agentów nie powinien znajdować się w Pythonie: kompilowanie

Poniższe notatki przedstawiają praktyczny plan działania dotyczący tematu „Twój graf agentów nie pasuje do Pythona: kompilowanie procesu wieloagentowego z pojedynczego pliku YAML”. Nacisk kładziony jest na umowy, sprawdzania oraz miejsca zastępcze dla kodu, a nie na motywacyjne aspekty. Podczas przechodzenia przez etap przeglądu najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno prawidłowy przebieg działania, jak i ścieżkę naprawczą. Próby ponowne, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później w celu udoskonalenia.

Problem, o którym nikt cię nie ostrzega

Problem, przed którym nikt nie ostrzega, sprawdza się najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustal wartości interpretera oraz pliku blokującego zależności przed omawianiem pętli. Różnice między laptopem a środowiskiem CI to najczęstsza, niewidoczna przyczyna awarii w demonstracjach API.

Jak wygląda proces pracy w formie danych

Faza „Jak wygląda proces pracy” funkcjonuje najlepiej, gdy traktowana jest jako mierzalna powierzchnia do analizy. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć przypadki częściowego ukończenia bez żadnych informacji. Ustal stałe wartości interpretera oraz pliku blokującego zależności przed nauczeniem pętli. Rozbieżności pomiędzy laptopem a środowiskiem CI są najczęstszą przyczyną ukrytych awarii w demonstracjach API.

entry: entry_agent
exit: exit

guardrails:
  - Reject queries that are outside the application's domain.
  - Reject queries about the system, agents, design, or internal workings.state_schema:
  query:
    type: str
    description: "User query or current message."
  chat_history:
    type: list
    annotated_with: add_messages
    description: "Conversation history between user and system."
  result:
    type: dict
    description: "Result from the processing agent."agents:
  - name: agent_one
    kind: function
    impl: your_package.agents.agent_one.agent_one_fn  - name: agent_two
    kind: function
    impl: your_package.agents.agent_two.agent_two_fnworkflow:
  nodes:
    - id: agent_one
      agent: agent_one
      writes: [query, result]
      next: decision_router    - id: decision_router
      kind: router
      router:
        impl: your_package.agents.routers.route_after_agent_one
        reads: [result]
        edges:
          agent_two: agent_two
          human_agent: human_agent

Podstęp nr 1: Generowanie klasy stanu w czasie wykonywania na podstawie schematu

Sztuczka nr 1: Najlepiej działa generowanie swojej środowiska testowego, gdy traktuje się je jako coś mierzalnego. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy środowisko przechodzi z wersji demonstracyjnej do współdzielonych środowisk. Ustal parametry interpretera oraz pliku blokującego zależności przed nauczeniem pętli. Różnice między laptopem a środowiskiem CI to najczęstsza przyczyna ukrytych awarii w demonstracjach API. Sztuczka nr 1: Najlepiej działa generowanie swojej środowiska testowego, gdy traktuje się je jako coś mierzalnego. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Dokumentuj zarówno prawidłowy przebieg działania, jak i ścieżkę naprawczą. Powtórne próby, mechanizmy kontroli ludzkiej oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.

# your_package/orchestrator/schema.py
annotations = {}
for key, value in state_schema.items():
    type_str = value.get("type", "str")
    # Convert YAML string to Python type
    py_type = eval(type_str)
    if value.get("annotated_with") == "add_messages":
        py_type = Annotated[list, {}]
    annotations[key] = py_type

spec = Spec(
    ...
    state=TypedDict("State", annotations),   # <- dynamic class, born at boot
    ...
)

Podstęp nr 2: Agenci odwoływane za pomocą ciągu z kropkami, rozwiązywane przez importlib

W fazie odwoływania się do agentów w ramach Podstępu nr 2 należy najpierw zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia, zanim zmieni się kod. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Lepiej używać małych, testowalnych jednostek niż rozbudowanych skryptów. Gdy dany krok zawiedzie, powód awarii powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną strukturę przepływu. Należy oddzielić budowę klienta od pętli komunikacji, aby można było wymieniać dostawców bez konieczności przepisywania maszyny stanu rozmowy.

impl: your_package.agents.agent_one.agent_one_fn

# your_package/orchestrator/schema.py
def _import_from_path(dotted: str) -> Callable[..., Any]:
    """Import a callable from a dotted path like 'package.module.function'."""
    if not dotted or "." not in dotted:
        raise ImportError(f"Invalid impl path: {dotted!r}")
    mod_path, attr = dotted.rsplit(".", 1)
    mod = importlib.import_module(mod_path)
    fn = getattr(mod, attr)
    if not callable(fn):
        raise TypeError(f"Imported object is not callable: {dotted}")
    return fn

def agent_impl_map(spec: Spec) -> Dict[str, Optional[Callable]]:
    """Map agent name -> callable (or None if impl missing)."""
    return {a.name: _import_from_path(a.impl) if a.impl else None
            for a in spec.agents}

Podstęp nr 3: Kompilator — węzły YAML stają się węzłami grafu

Dla etapu kompilacji w Triku 3 należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy artefaktom, zdefiniuj sprawdzenia sukcesu i odrzuć ciche, częściowe ukończenie zadania. Oddziel konstrukcję klienta od pętli komunikatów, aby można było zmieniać dostawców bez konieczności przepisywania maszyny stanu rozmowy.

# your_package/orchestrator/runner.py
def build(self):
    graph = StateGraph(state_schema=self.spec.state)   # our generated TypedDict

    def _add_task_node(node):
        async def _node(state: Dict[str, Any]) -> Dict[str, Any]:
            res = await self._call_agent(node.agent, state, node.id)
            if getattr(node, "writes", None):
                if isinstance(res, dict):
                    # Only let the node write the keys it declared in YAML
                    filtered = {k: v for k, v in res.items() if k in node.writes}
                    return filtered or res
                key = node.writes[0]
                return {key: res}
            return res
        graph.add_node(node.id, _node)    # Build every node
    for node in self.spec.workflow.nodes:
        if getattr(node, "router", None):
            _add_router_node(node)
        else:
            _add_task_node(node)    graph.set_entry_point(entry)    # Inline "next:" edges from YAML become static edges
    for node in self.spec.workflow.nodes:
        if getattr(node, "next", None):
            graph.add_edge(node.id, node.next)    # Terminal nodes wire to END
    for node in self.spec.workflow.nodes:
        if getattr(node, "terminal", False):
            graph.add_edge(node.id, END)    self._runnable = graph.compile(checkpointer=self.checkpoint)
    return self

Routery: warunkowe rozgałęzianie jako tabela wyszukiwania

Dla warunkowego rozgałęziania w Routersach jako etapu należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Zapisuj czas trwania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Oddziel konstrukcję klienta od pętli komunikatów, aby można było zmieniać dostawców bez konieczności przepisywania maszyny stanu rozmowy. Dla warunkowego rozgałęziania w Routersach jako etapu należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Zdokumentuj razem ścieżkę pomyślną oraz ścieżkę naprawczą. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości nieudanych są częścią produktu, a nie opóźnieniami.

Trochę polskiego.

# your_package/orchestrator/runner.py
def _add_router_node(node):
    router = self.router_fns[node.id]

    def _router_fn():
        def _f(state):
            out = router(state)
            # Routers may return either a label, or (state_updates, label)
            if isinstance(out, tuple):
                updates, label = out
                if isinstance(updates, dict):
                    for k, v in updates.items():
                        state[k] = v
            else:
                label = out
            return label
        return _f    graph.add_node(node.id, lambda s: {})
    graph.add_conditional_edges(node.id, _router_fn(), node.router.edges)

Podstęp nr 4: Adaptacyjna funkcja wywoławcza — agenci mogą wpisać dowolny podpis

Podczas pracy nad etapem adaptacyjnym zgodnie z Podstępem nr 4 najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Zapisuj ID żądania, ID modelu oraz opóźnienie przy każdym wywołaniu. Bez tych informacji przerywane błędy dostawcy wyglądają jak błędy aplikacji.

# your_package/orchestrator/runner.py
async def _adapt_and_call(self, fn, state, node_id):
    """
    Adaptively call agent functions so implementations receive what they expect:
    - def agent(**kwargs):        → pass **state (+ inject 'query' if missing)
    - def agent(query, **kwargs): → pass query=..., plus any **extra
    - def agent(state):           → pass state
    - def agent(query):           → pass query
    - def agent():                → call without args
    """
    sig = inspect.signature(fn)
    params = sig.parameters
    has_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD
                     for p in params.values())

    kwargs = {}
    if has_var_kw:
        kwargs.update(state)
    if "state" in params:
        kwargs["state"] = state
    if "query" in params or has_var_kw:
        kwargs.setdefault("query", self._fallback_query(state))    # A lone positional 'query' → call it positionally
    if (len(params) == 1
            and next(iter(params.keys())) == "query"):
        return await _maybe_await(fn(self._fallback_query(state)))    res = fn(**kwargs)
    return await res if hasattr(res, "__await__") else res

Podstęp nr 5: Hot-swapping węzła na sesję (udział człowieka)

Gdy pracujesz nad techniką nr 5 – szybką wymianę etapu – najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć możliwość cichego, częściowego ukończenia zadania. Zapisuj identyfikator żądania, identyfikator modelu oraz czas opóźnienia przy każdej próbie połączenia. Bez tych informacji przerywane błędy dostawcy wyglądają jak błędy aplikacji.

# your_package/services/session_service.py (paraphrased)
if websocket is not None:
    session_handler = SessionHandler(websocket, user_id=user_id, session_id=session_id, ...)
    _runner.agent_fns["human_agent"] = _import_from_function(
        make_input_method(session_handler)
    )

Co naprawdę daje ci ta architektura

Gdy przechodzisz przez etap „Co tak naprawdę oznacza ta architektura”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Obok wyników funkcjonalnych zapisz czas wykonywania oraz koszt tokena lub zapytania. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Zapisuj ID żądania, ID modelu oraz opóźnienie przy każdym wywołaniu. Bez takich informacji przerywane błędy dostawcy wyglądają jak błędy aplikacji. Gdy przechodzisz przez etap „Co tak naprawdę oznacza ta architektura”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Dokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych wywołań, mechanizmy ludzkiej kontroli oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.

Główne wnioski

Faza przygotowania wyników działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustal parametry interpretera oraz plik blokujący zależności przed omawianiem pętli. Rozbieżności między laptopem a środowiskiem CI to najczęstsza przyczyna niewidocznych problemów w demonstracjach API.

Lista kontrolna operacyjna

W fazie listy kontrolnej operacyjnej zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia pracy przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanej punktacji kontrolnej, bez konieczności zgadywania ukrytego stanu systemu.

Zachowaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności przeglądania całej struktury.

Rozdziel budowę klienta od pętli przekazywania wiadomości, aby można było zmieniać dostawców bez konieczności przepisywania maszyny stanu rozmowy.

Utwórz punkt kontrolny po kosztownych krokach. System powinien unikać ponownego naliczania opłat za tę samą operację z LLM, gdy operator spróbuje ponownie wykonać późniejszy etap.

Zdefiniuj wersje zależności i zapisz hash obrazu, który został użyty do wykonania demonstracji. Reprodukowalność jest ważniejsza od wiedzy przekazywanej ustnie.

Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy plikom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania.

Zanim wdrożysz tę architekturę, zamroź wersje, utwórz dokładny zapis dla kluczowych etapów realizacji oraz potwierdź kroki odwracania zmian. Środowiska współdzielone wymagają ograniczeń szybkości, weryfikacji przynależności użytkowników oraz wyraźnego właściciela odpowiedzialnego za rotację haseł. Wolimy nudną niezawodność od pomysłowych, jednorazowych demonstracji.

Uwaga dotycząca wersji 2822ea5988ca: unikaj przechowywania kluczy dostawcy w repozytorium, ustaw ograniczenie liczby tokenów na sesję oraz przechowuj zapisy obok plików testowych, aby późniejsze zmiany modeli pozostały porównywalne.