Subagents i Hierarchie Agentów

Wyprawa rośnie: ktoś bada trasę, ktoś liczy zapasy, ktoś sprawdza pozwolenia, ktoś pilnuje jakości. Gdy każdy z nich rozmawia z każdym, po tygodniu nikt nie wie, kto co ustalił. Potrzebujesz kierownika wyprawy, który rozdziela zadania i zbiera raporty.

Subagenty to wzorzec architektoniczny, gdzie główny agent (orchestrator) deleguje zadania do wyspecjalizowanych podagentów. To jak struktura organizacyjna firmy - CEO deleguje zadania do managerów, którzy delegują dalej.

Architektura Subagentów

Schemat pokazuje trzy piętra hierarchii, od kierownika po narzędzia:

1┌─────────────────────────────────────────────────────────────┐
2│                   Agent Hierarchy                           │
3├─────────────────────────────────────────────────────────────┤
4│                                                             │
5│                    ┌───────────────┐                        │
6│                    │  Orchestrator │                        │
7│                    │   (Main AI)   │                        │
8│                    └───────┬───────┘                        │
9│                            │                                │
10│           ┌────────────────┼────────────────┐               │
11│           │                │                │               │
12│           ▼                ▼                ▼               │
13│    ┌────────────┐   ┌────────────┐   ┌────────────┐        │
14│    │ Research   │   │ Analysis   │   │ Execution  │        │
15│    │ Subagent   │   │ Subagent   │   │ Subagent   │        │
16│    └─────┬──────┘   └─────┬──────┘   └─────┬──────┘        │
17│          │                │                │                │
18│    ┌─────┴─────┐    ┌─────┴─────┐    ┌─────┴─────┐         │
19│    │  Tools    │    │  Tools    │    │  Tools    │         │
20│    └───────────┘    └───────────┘    └───────────┘         │
21│                                                             │
22└─────────────────────────────────────────────────────────────┘

Na górze orchestrator, który koordynuje pracę i deleguje zadania. Niżej subagenci specjaliści, a na dole narzędzia, których używają. Orchestrator nigdy nie sięga po narzędzia bezpośrednio.

Podstawowa implementacja Subagentów

Najpierw typy danych: role jako Enum oraz zadanie i wynik jako @dataclass:

1from openai import OpenAI
2from abc import ABC, abstractmethod
3from dataclasses import dataclass
4from typing import Optional
5from enum import Enum
6
7client = OpenAI()
8
9class SubagentRole(Enum):
10    RESEARCHER = "researcher"
11    ANALYST = "analyst"
12    EXECUTOR = "executor"
13    VALIDATOR = "validator"
14
15@dataclass
16class SubagentTask:
17    """Zadanie dla subagenta."""
18    description: str
19    context: str
20    expected_output: str
21    priority: int = 1
22
23@dataclass
24class SubagentResult:
25    """Wynik pracy subagenta."""
26    success: bool
27    output: str
28    metadata: dict

SubagentTask niesie opis, kontekst i oczekiwany wynik, czyli wszystko, czego specjalista potrzebuje, żeby nie pytać kierownika o szczegóły.

Bazowa klasa Subagent dziedziczy po ABC i wymusza system_prompt przez @abstractmethod, tak jak BaseAgent z lekcji o systemach multi-agentowych:

1class Subagent(ABC):
2    """Bazowa klasa subagenta."""
3
4    def __init__(self, name: str, role: SubagentRole, model: str = "gpt-4o-mini"):
5        self.name = name
6        self.role = role
7        self.model = model
8        self.tools: list[dict] = []
9        self.memory: list[str] = []
10
11    @property
12    @abstractmethod
13    def system_prompt(self) -> str:
14        """Prompt systemowy subagenta."""
15        pass
16
17    def add_tool(self, tool: dict) -> None:
18        """Dodaje narzędzie do subagenta."""
19        self.tools.append(tool)

Każdy subagent ma własną listę narzędzi i pamięć. add_tool dopisuje definicję narzędzia w formacie OpenAI, a memory zbiera skróty wykonanych zadań.

Metoda execute składa prompt z trzech pól zadania i woła model:

1    async def execute(self, task: SubagentTask) -> SubagentResult:
2        """Wykonuje zadanie."""
3        messages = [
4            {"role": "system", "content": self.system_prompt},
5            {"role": "user", "content": f"""
6Zadanie: {task.description}
7Kontekst: {task.context}
8Oczekiwany output: {task.expected_output}
9"""}
10        ]
11
12        response = client.chat.completions.create(
13            model=self.model,
14            messages=messages,
15            tools=self.tools if self.tools else None
16        )
17
18        output = response.choices[0].message.content
19        self.memory.append(f"Task: {task.description} -> Result: {output[:100]}...")
20
21        return SubagentResult(
22            success=True,
23            output=output,
24            metadata={"model": self.model, "role": self.role.value}
25        )

Metoda jest async, ale używa synchronicznego klienta, więc blokuje pętlę zdarzeń. W produkcji użyj AsyncOpenAI i await. Uwaga też na narzędzia: gdy model zamiast tekstu wywoła narzędzie, content będzie None i output[:100] zgłosi błąd.

Wyspecjalizowane Subagenty

Specjaliści, podobnie jak wcześniej, różnią się tylko promptem. Badacz i analityk:

1class ResearchSubagent(Subagent):
2    """Subagent badawczy - zbiera informacje."""
3
4    @property
5    def system_prompt(self) -> str:
6        return """Jesteś wyspecjalizowanym agentem badawczym.
7
8Twoje zadania:
91. Zbieranie informacji z dostępnych źródeł
102. Weryfikacja faktów
113. Strukturyzowanie wiedzy
124. Identyfikacja luk informacyjnych
13
14Zawsze:
15- Podawaj źródła informacji
16- Zaznaczaj niepewności
17- Proponuj dalsze kierunki badań"""
18
19
20class AnalysisSubagent(Subagent):
21    """Subagent analityczny - analizuje dane."""
22
23    @property
24    def system_prompt(self) -> str:
25        return """Jesteś wyspecjalizowanym agentem analitycznym.
26
27Twoje zadania:
281. Analiza dostarczonych danych
292. Identyfikacja wzorców i trendów
303. Formułowanie wniosków
314. Ocena ryzyka i szans
32
33Zawsze:
34- Używaj danych liczbowych
35- Przedstawiaj alternatywne interpretacje
36- Podkreślaj kluczowe odkrycia"""

Każdy prompt mówi, co subagent robi i czego ma zawsze pilnować. Sekcja "Zawsze" działa jak regulamin obozu dla danego specjalisty.

Wykonawca i walidator domykają zespół:

1class ExecutorSubagent(Subagent):
2    """Subagent wykonawczy - realizuje zadania."""
3
4    @property
5    def system_prompt(self) -> str:
6        return """Jesteś wyspecjalizowanym agentem wykonawczym.
7
8Twoje zadania:
91. Realizacja zaplanowanych działań
102. Używanie narzędzi do wykonania zadań
113. Raportowanie postępów
124. Obsługa błędów
13
14Zawsze:
15- Wykonuj zadania krok po kroku
16- Weryfikuj wyniki każdej akcji
17- Raportuj problemy natychmiast"""
18
19
20class ValidatorSubagent(Subagent):
21    """Subagent walidujący - sprawdza jakość."""
22
23    @property
24    def system_prompt(self) -> str:
25        return """Jesteś wyspecjalizowanym agentem walidacyjnym.
26
27Twoje zadania:
281. Weryfikacja poprawności wyników
292. Sprawdzanie zgodności z wymaganiami
303. Identyfikacja błędów i niedociągnięć
314. Sugerowanie poprawek
32
33Zawsze:
34- Bądź krytyczny ale konstruktywny
35- Używaj konkretnych kryteriów
36- Proponuj rozwiązania problemów"""

Walidator to recenzent wyników pozostałych, dzięki czemu błąd jednego specjalisty ma szansę zostać wyłapany.

Orchestrator - Zarządzanie Subagentami

Orchestrator trzyma rejestr subagentów i historię zadań:

1from typing import Dict, List
2import asyncio
3
4class AgentOrchestrator:
5    """Główny orchestrator zarządzający subagentami."""
6
7    def __init__(self, model: str = "gpt-4o-mini"):
8        self.model = model
9        self.subagents: Dict[str, Subagent] = {}
10        self.task_history: List[dict] = []
11
12    def register_subagent(self, subagent: Subagent) -> None:
13        """Rejestruje subagenta."""
14        self.subagents[subagent.name] = subagent
15        print(f"Zarejestrowano subagenta: {subagent.name} ({subagent.role.value})")
16
17    def get_available_subagents(self) -> str:
18        """Zwraca opis dostępnych subagentów."""
19        return "\n".join([
20            f"- {name}: {agent.role.value}"
21            for name, agent in self.subagents.items()
22        ])

get_available_subagents tworzy tekstową listę specjalistów, którą za chwilę pokażemy modelowi.

Kroki orchestratora idą zawsze w tej kolejności: plan_execution, wybór subagenta, subagent.execute i synthesize_results. Planowanie prosi model o plan w JSON:

1    async def plan_execution(self, task: str) -> List[SubagentTask]:
2        """Planuje wykonanie zadania przez odpowiednich subagentów."""
3        planning_prompt = f"""Masz do wykonania zadanie: {task}
4
5Dostępni subagenci:
6{self.get_available_subagents()}
7
8Zaplanuj wykonanie zadania. Dla każdego kroku określ:
91. Który subagent powinien wykonać krok
102. Opis zadania dla subagenta
113. Oczekiwany output
124. Priorytet (1-5)
13
14Odpowiedz w formacie JSON:
15{{"steps": [{{"subagent": "nazwa", "task": "opis", "expected_output": "...", "priority": 1}}, ...]}}"""
16
17        response = client.chat.completions.create(
18            model=self.model,
19            messages=[
20                {"role": "system", "content": "Jesteś planistą zadań. Odpowiadaj tylko w JSON."},
21                {"role": "user", "content": planning_prompt}
22            ],
23            response_format={"type": "json_object"}
24        )
25
26        import json
27        plan = json.loads(response.choices[0].message.content)
28
29        tasks = []
30        for step in plan.get("steps", plan):
31            tasks.append(SubagentTask(
32                description=step["task"],
33                context=task,
34                expected_output=step["expected_output"],
35                priority=step.get("priority", 1)
36            ))
37
38        return tasks

Tryb response_format={"type": "json_object"} gwarantuje poprawny obiekt JSON, ale nie tablicę. Dlatego prompt prosi o obiekt z kluczem steps, a kod czyta plan.get("steps", plan).

execute_task spina planowanie, wykonanie i syntezę:

1    async def execute_task(self, task: str) -> str:
2        """Wykonuje zadanie używając subagentów."""
3        # 1. Planowanie
4        planned_tasks = await self.plan_execution(task)
5        print(f"Zaplanowano {len(planned_tasks)} kroków")
6
7        # 2. Wykonanie
8        results = []
9        for i, subtask in enumerate(planned_tasks):
10            # Znajdź odpowiedniego subagenta
11            subagent = self._select_subagent(subtask)
12
13            if subagent:
14                print(f"Krok {i+1}: {subagent.name} wykonuje: {subtask.description[:50]}...")
15                result = await subagent.execute(subtask)
16                results.append({
17                    "step": i + 1,
18                    "subagent": subagent.name,
19                    "task": subtask.description,
20                    "result": result.output
21                })
22
23        # 3. Synteza
24        final_result = await self._synthesize_results(task, results)
25
26        return final_result

Wyniki trafiają do listy słowników, więc synteza wie, który subagent co zrobił.

Wybór subagenta to prosta heurystyka słów kluczowych:

1    def _select_subagent(self, task: SubagentTask) -> Optional[Subagent]:
2        """Wybiera najlepszego subagenta dla zadania."""
3        # Prosta heurystyka - można rozbudować o ML
4        keywords = {
5            "researcher": ["zbadaj", "znajdź", "wyszukaj", "informacje"],
6            "analyst": ["analizuj", "oceń", "porównaj", "wnioski"],
7            "executor": ["wykonaj", "zrób", "stwórz", "zaimplementuj"],
8            "validator": ["sprawdź", "zweryfikuj", "waliduj", "przetestuj"]
9        }
10
11        task_lower = task.description.lower()
12
13        for subagent_type, kws in keywords.items():
14            if any(kw in task_lower for kw in kws):
15                for agent in self.subagents.values():
16                    if agent.role.value == subagent_type:
17                        return agent
18
19        # Domyślnie pierwszy dostępny
20        return list(self.subagents.values())[0] if self.subagents else None

Klucze słownika muszą równać się wartościom ról, dlatego pierwszy klucz to "researcher", a nie "research", bo inaczej badacz nigdy by nie został wybrany. Pole subagent z planu jest tu pomijane, co warto poprawić.

Na końcu synteza łączy wyniki w jedną odpowiedź:

1    async def _synthesize_results(self, original_task: str, results: List[dict]) -> str:
2        """Syntezuje wyniki ze wszystkich subagentów."""
3        results_summary = "\n\n".join([
4            f"Krok {r['step']} ({r['subagent']}): {r['result']}"
5            for r in results
6        ])
7
8        synthesis_prompt = f"""Oryginalne zadanie: {original_task}
9
10Wyniki subagentów:
11{results_summary}
12
13Stwórz spójną, końcową odpowiedź na podstawie wyników wszystkich subagentów."""
14
15        response = client.chat.completions.create(
16            model=self.model,
17            messages=[
18                {"role": "system", "content": "Jesteś syntetykiem. Łączysz wyniki w spójną całość."},
19                {"role": "user", "content": synthesis_prompt}
20            ]
21        )
22
23        return response.choices[0].message.content

To trzecie wywołanie modelu w jednym zadaniu, obok planowania i pracy subagentów.

Użycie Orchestratora

Rejestrujemy czterech specjalistów i zlecamy planowanie Safari:

1import asyncio
2
3async def main():
4    # Tworzenie orchestratora
5    orchestrator = AgentOrchestrator()
6
7    # Rejestracja subagentów
8    orchestrator.register_subagent(
9        ResearchSubagent("safari_researcher", SubagentRole.RESEARCHER)
10    )
11    orchestrator.register_subagent(
12        AnalysisSubagent("data_analyst", SubagentRole.ANALYST)
13    )
14    orchestrator.register_subagent(
15        ExecutorSubagent("task_executor", SubagentRole.EXECUTOR)
16    )
17    orchestrator.register_subagent(
18        ValidatorSubagent("quality_checker", SubagentRole.VALIDATOR)
19    )
20
21    # Wykonanie złożonego zadania
22    result = await orchestrator.execute_task(
23        "Przygotuj kompletny plan Safari w Serengeti na 7 dni dla 4 osób"
24    )
25
26    print("\n" + "="*50)
27    print("WYNIK KOŃCOWY:")
28    print("="*50)
29    print(result)
30
31asyncio.run(main())

asyncio.run uruchamia główną korutynę, a orchestrator sam planuje, deleguje i składa wynik. O tym, który specjalista dostanie krok, decydują słowa w opisach wygenerowanych przez planistę, a nie w samym zadaniu.

Hierarchia Wielopoziomowa

Duża wyprawa ma kierowników zespołów. Orchestrator może mieć pod sobą innych orchestratorów:

1class HierarchicalOrchestrator:
2    """Orchestrator z wielopoziomową hierarchią."""
3
4    def __init__(self, name: str, level: int = 0):
5        self.name = name
6        self.level = level
7        self.subagents: Dict[str, Subagent] = {}
8        self.sub_orchestrators: Dict[str, "HierarchicalOrchestrator"] = {}
9
10    def add_sub_orchestrator(self, orchestrator: "HierarchicalOrchestrator") -> None:
11        """Dodaje pod-orchestratora (dla głębszej hierarchii)."""
12        orchestrator.level = self.level + 1
13        self.sub_orchestrators[orchestrator.name] = orchestrator
14
15    async def delegate_task(self, task: str) -> str:
16        """Deleguje zadanie do odpowiedniego poziomu hierarchii."""
17        # Sprawdź czy zadanie wymaga pod-orchestratora
18        complexity = self._assess_complexity(task)
19
20        if complexity > 3 and self.sub_orchestrators:
21            # Deleguj do pod-orchestratora
22            sub_orch = self._select_sub_orchestrator(task)
23            return await sub_orch.delegate_task(task)
24        else:
25            # Wykonaj przez własnych subagentów
26            return await self._execute_locally(task)

delegate_task przekazuje zadanie niżej, gdy ocena złożoności przekracza 3.

Metody pomocnicze oceniają złożoność i wykonują zadanie lokalnie:

1    def _assess_complexity(self, task: str) -> int:
2        """Ocenia złożoność zadania (1-5)."""
3        # Prosta heurystyka
4        complexity_indicators = [
5            "kompletny", "szczegółowy", "wieloetapowy",
6            "kompleksowy", "rozbudowany"
7        ]
8        return sum(1 for ind in complexity_indicators if ind in task.lower()) + 1
9
10    def _select_sub_orchestrator(self, task: str) -> "HierarchicalOrchestrator":
11        """Wybiera pod-orchestratora dla zadania."""
12        # Można rozbudować o inteligentny wybór
13        return list(self.sub_orchestrators.values())[0]
14
15    async def _execute_locally(self, task: str) -> str:
16        """Wykonuje zadanie lokalnie."""
17        # Użyj własnych subagentów
18        results = []
19        for subagent in self.subagents.values():
20            subtask = SubagentTask(
21                description=task,
22                context=f"Level {self.level}",
23                expected_output="Rezultat zadania"
24            )
25            result = await subagent.execute(subtask)
26            results.append(result.output)
27
28        return "\n".join(results)

Złożoność to liczba słów-wskaźników plus jeden, a _select_sub_orchestrator na razie zawsze bierze pierwszy zespół.

Przykład buduje dwa zespoły pod głównym orchestratorem:

1# Przykład użycia hierarchii
2async def hierarchical_example():
3    # Główny orchestrator
4    main_orchestrator = HierarchicalOrchestrator("Main Orchestrator")
5
6    # Pod-orchestrator dla badań
7    research_orchestrator = HierarchicalOrchestrator("Research Team")
8    research_orchestrator.subagents["researcher"] = ResearchSubagent(
9        "researcher", SubagentRole.RESEARCHER
10    )
11
12    # Pod-orchestrator dla analizy
13    analysis_orchestrator = HierarchicalOrchestrator("Analysis Team")
14    analysis_orchestrator.subagents["analyst"] = AnalysisSubagent(
15        "analyst", SubagentRole.ANALYST
16    )
17
18    # Dodaj do hierarchii
19    main_orchestrator.add_sub_orchestrator(research_orchestrator)
20    main_orchestrator.add_sub_orchestrator(analysis_orchestrator)
21
22    # Wykonaj złożone zadanie
23    result = await main_orchestrator.delegate_task(
24        "Przeprowadź kompletne badanie rynku Safari w Afryce Wschodniej"
25    )
26    print(result)

Znajdź pułapkę: zadanie zawiera za mało słów-wskaźników, więc zostaje u głównego orchestratora, który nie ma własnych subagentów i zwraca pusty tekst. Obniż próg albo daj mu subagenta.

Komunikacja między Subagentami

Czasem specjalista musi dopytać kolegę. Najpierw typy wiadomości:

1from dataclasses import dataclass, field
2from typing import Optional, List
3from datetime import datetime
4from enum import Enum
5
6class MessageType(Enum):
7    TASK = "task"
8    RESULT = "result"
9    QUERY = "query"
10    RESPONSE = "response"
11    ERROR = "error"
12
13@dataclass
14class AgentMessage:
15    """Wiadomość między agentami."""
16    sender: str
17    receiver: str
18    content: str
19    message_type: MessageType
20    timestamp: datetime = field(default_factory=datetime.now)
21    correlation_id: Optional[str] = None
22    metadata: dict = field(default_factory=dict)

correlation_id łączy pytanie z odpowiedzią, a field(default_factory=...) daje każdej wiadomości własny znacznik czasu.

Message Bus to szyna, przez którą agenci rozmawiają bez bezpośrednich zależności:

1class MessageBus:
2    """Szyna komunikacyjna dla agentów."""
3
4    def __init__(self):
5        self.messages: List[AgentMessage] = []
6        self.subscribers: Dict[str, List[callable]] = {}
7
8    def publish(self, message: AgentMessage) -> None:
9        """Publikuje wiadomość."""
10        self.messages.append(message)
11
12        # Powiadom subskrybentów
13        if message.receiver in self.subscribers:
14            for callback in self.subscribers[message.receiver]:
15                callback(message)
16
17    def subscribe(self, agent_name: str, callback: callable) -> None:
18        """Subskrybuje agenta do wiadomości."""
19        if agent_name not in self.subscribers:
20            self.subscribers[agent_name] = []
21        self.subscribers[agent_name].append(callback)
22
23    def get_messages_for(self, agent_name: str) -> List[AgentMessage]:
24        """Pobiera wiadomości dla agenta."""
25        return [m for m in self.messages if m.receiver == agent_name]

Agent subskrybuje swoje imię, a publish wywołuje callbacki odbiorcy. Nadawca nie musi znać obiektu odbiorcy, wystarczy mu nazwa, i to właśnie oznacza luźne powiązanie.

Subagent z komunikacją odpowiada na zapytania i wysyła wiadomości:

1class CommunicatingSubagent(Subagent):
2    """Subagent z możliwością komunikacji."""
3
4    def __init__(self, name: str, role: SubagentRole, message_bus: MessageBus):
5        super().__init__(name, role)
6        self.message_bus = message_bus
7        self.message_bus.subscribe(name, self._handle_message)
8        self.pending_responses: Dict[str, str] = {}
9
10    def _handle_message(self, message: AgentMessage) -> None:
11        """Obsługuje przychodzące wiadomości."""
12        if message.message_type == MessageType.QUERY:
13            # Odpowiedz na zapytanie
14            response = self._process_query(message.content)
15            self.send_message(
16                message.sender,
17                response,
18                MessageType.RESPONSE,
19                correlation_id=message.correlation_id
20            )
21        elif message.message_type == MessageType.RESPONSE:
22            # Zapisz odpowiedź
23            if message.correlation_id:
24                self.pending_responses[message.correlation_id] = message.content
25
26    def send_message(
27        self,
28        receiver: str,
29        content: str,
30        message_type: MessageType,
31        correlation_id: Optional[str] = None
32    ) -> None:
33        """Wysyła wiadomość do innego agenta."""
34        import uuid
35        message = AgentMessage(
36            sender=self.name,
37            receiver=receiver,
38            content=content,
39            message_type=message_type,
40            correlation_id=correlation_id or str(uuid.uuid4())
41        )
42        self.message_bus.publish(message)
43
44    def _process_query(self, query: str) -> str:
45        """Przetwarza zapytanie od innego agenta."""
46        # Implementacja zależna od roli
47        return f"Odpowiedź na: {query}"

Odpowiedź wraca z tym samym correlation_id, więc pytający wie, na co to odpowiedź.

Pytanie z oczekiwaniem na odpowiedź:

1    async def ask_other_agent(self, agent_name: str, question: str) -> str:
2        """Pyta innego agenta i czeka na odpowiedź."""
3        import uuid
4        correlation_id = str(uuid.uuid4())
5
6        self.send_message(
7            agent_name,
8            question,
9            MessageType.QUERY,
10            correlation_id
11        )
12
13        # Czekaj na odpowiedź (z timeout)
14        import asyncio
15        for _ in range(10):
16            if correlation_id in self.pending_responses:
17                return self.pending_responses.pop(correlation_id)
18            await asyncio.sleep(0.1)
19
20        return "Timeout - brak odpowiedzi"

Szyna wywołuje callbacki synchronicznie, więc odpowiedź zwykle czeka już przy pierwszym sprawdzeniu, a limit to sekunda.

Best Practices dla Subagentów

Zasady architektury subagentów zebrane w jednym miejscu:

1"""
2Best Practices dla architektury subagentów:
3
41. SEPARATION OF CONCERNS
5   - Każdy subagent ma jedną, jasno określoną odpowiedzialność
6   - Unikaj "god agents" robiących wszystko
7
82. LOOSE COUPLING
9   - Subagenci komunikują się przez message bus
10   - Nie ma bezpośrednich zależności między subagentami
11
123. SINGLE SOURCE OF TRUTH
13   - Orchestrator jest jedynym źródłem prawdy o stanie zadania
14   - Subagenci raportują wyniki do orchestratora
15
164. GRACEFUL DEGRADATION
17   - System działa nawet gdy jeden subagent zawiedzie
18   - Implementuj fallback strategies
19
205. OBSERVABILITY
21   - Loguj wszystkie interakcje między agentami
22   - Implementuj metryki i monitoring
23
246. SCALABILITY
25   - Subagenci powinni być bezstanowi (stateless)
26   - Łatwo dodać więcej instancji tego samego typu
27
287. TESTING
29   - Testuj subagentów w izolacji
30   - Testuj integrację orchestratora z subagentami
31"""

Zwróć uwagę na punkt o bezstanowości: nasze subagenty mają memory, więc przy skalowaniu przenieś ją do zewnętrznego magazynu.

Ostatni element to orchestrator odporny na błędy:

1class RobustOrchestrator(AgentOrchestrator):
2    """Orchestrator z obsługą błędów i fallback."""
3
4    async def execute_with_fallback(self, task: str) -> str:
5        """Wykonuje zadanie z obsługą błędów."""
6        try:
7            return await self.execute_task(task)
8        except Exception as e:
9            print(f"Błąd głównego wykonania: {e}")
10
11            # Fallback - prostsze podejście
12            return await self._simple_execution(task)
13
14    async def _simple_execution(self, task: str) -> str:
15        """Proste wykonanie jako fallback."""
16        response = client.chat.completions.create(
17            model=self.model,
18            messages=[
19                {"role": "system", "content": "Jesteś pomocnym asystentem."},
20                {"role": "user", "content": task}
21            ]
22        )
23        return response.choices[0].message.content

Gdy pełny pipeline zawiedzie, fallback pyta model bezpośrednio. Polecam od początku projektować hierarchię płasko, a kolejne piętra dodawać dopiero przy prawdziwej potrzebie.

Subagenty i hierarchie agentów to potężne wzorce architektoniczne, które pozwalają budować złożone, skalowalne systemy AI. Gratulacje - poznałeś zaawansowane techniki AI! W ostatnim module stworzysz kompleksowy projekt łączący wszystkie zdobyte umiejętności!

Zapamiętaj: dobry orchestrator to kierownik wyprawy, który rozdziela zadania, zbiera raporty i sam nie niesie wszystkich plecaków.