Budowanie agentów AI w produkcji ze stanowym grafem orkiestracji

Gdy agent ma więcej niż dwa narzędzia, pętla ReAct przestaje być architekturą i zaczyna być zobowiązaniem. Latencja się kumuluje, tryby awarii mnożą, i nie ma uczciwego miejsca, gdzie można narysować granicę dla retry.

Zmiana, którą wprowadziliśmy (i którą polecam każdemu zespołowi) to odejście od modelowania agenta jako pętli na rzecz modelowania go jako maszyny stanów, gdzie każde przejście ma nazwę, każdy stan można checkpointować, a każde wywołanie narzędzia jest węzłem, nie efektem ubocznym.

W terminologii LangGraph: graf jest kontraktem. Model jest uczestnikiem grafu, nie jego właścicielem. To pojedyncze odwrócenie pozwala robić wszystko, czego agent w produkcji faktycznie potrzebuje, pauzować, wznawiać, fan-outować, przekazywać do człowieka, odtwarzać wczorajszą sesję w regression suite.

Dlaczego pętle zawodzą na skali

Kanoniczna pętla ReAct jest elegancka w demach: obserwuj, myśl, działaj, powtarzaj. W produkcji akumuluje tryby awarii, które nasilają się z każdym kolejnym narzędziem.

Gdy wywołanie narzędzia się nie powiedzie, pętla nie ma naturalnego miejsca na wyznaczenie granicy. Dodajesz guard max_steps. Potem ktoś z zespołu go podwyższy "tylko ten jeden raz". Trzy miesiące później twoje p99 latency to 45 sekund i nikt nie wie dlaczego, rzetelne testy na produkcji mają swoje ograniczenia. Gdy klient zgłasza buga, chcesz odtworzyć dokładną sekwencję decyzji, a pętla bez nazwanych checkpointów daje ci co najwyżej log. Graf daje ci wznawialny snapshot. Pętla nie potrafi też zatrzymać się i czekać na człowieka: graf może przerwać w dowolnym nazwanym węźle i czekać w nieskończoność na input, a potem wznowić dokładnie z tego miejsca.

Topologia grafu

# jedyna pętla należy do runtim'u. agent jest grafem.
graph = StateGraph(AgentState)

graph.add_node("plan",     planner)
graph.add_node("retrieve", retriever)
graph.add_node("act",      tool_runner)
graph.add_node("reflect",  critic)

graph.add_edge(START, "plan")
graph.add_conditional_edges("plan",
    route=lambda s: "retrieve" if s.needs_context else "act")
graph.add_edge("retrieve", "act")
graph.add_conditional_edges("act",
    route=lambda s: END if s.done else "reflect")
graph.add_edge("reflect", "plan")

app = graph.compile(checkpointer=PostgresSaver(dsn))

Checkpointing z Postgres

Używamy PostgresSaver jako checkpointera. Każde przejście zapisuje wiersz do agent_checkpoints(thread_id, step, state_json, created_at). Po crashu runner podejmuje od ostatniego ukończonego kroku bez powtarzania czegokolwiek przed nim. Każdy pośredni stan można inspekcjonować lub odtworzyć z zamockowanymi narzędziami do testów regresyjnych, a każdy wiersz kroku zawiera liczniki tokenów, więc atrybucja kosztów jest dokładna zamiast Chłopskiego Rozumu as a Service na koniec miesiąca.

Zapis do Postgres dodaje ~3ms na krok. Dla agenta z 20 krokami to 60ms. Akceptowalne.

Czego nie można checkpointować

Streamingowych outputów narzędzi. Jeśli twoje narzędzie streamuje dane do użytkownika w trakcie grafu, nie możesz bezpiecznie powtórzyć tego przejścia bez ponownego wyzwolenia streamu. Rozwiązaliśmy to oznaczając węzły streamingowe jako non_resumable i uruchamiając je od nowa przy każdym replayu.

Obserwowalność

Każde wykonanie grafu emituje ustrukturyzowane zdarzenie per krok: {thread_id, step_name, input_tokens, output_tokens, latency_ms, tool_calls, error}. Wysyłamy je do time-series store i budujemy dashboardy per agent per tydzień. Gdy nowe narzędzie spowalnia medianę czasu kroku, widzimy to w jeden dzień, nie w jeden sprint.