Utilizziamo i cookie per migliorare la tua esperienza sul sito
Subagents i Hierarchie Agentów
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
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└─────────────────────────────────────────────────────────────┘Podstawowa implementacja Subagentów
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
29
30class Subagent(ABC):
31 """Bazowa klasa subagenta."""
32
33 def __init__(self, name: str, role: SubagentRole, model: str = "gpt-4o-mini"):
34 self.name = name
35 self.role = role
36 self.model = model
37 self.tools: list[dict] = []
38 self.memory: list[str] = []
39
40 @property
41 @abstractmethod
42 def system_prompt(self) -> str:
43 """Prompt systemowy subagenta."""
44 pass
45
46 def add_tool(self, tool: dict) -> None:
47 """Dodaje narzędzie do subagenta."""
48 self.tools.append(tool)
49
50 async def execute(self, task: SubagentTask) -> SubagentResult:
51 """Wykonuje zadanie."""
52 messages = [
53 {"role": "system", "content": self.system_prompt},
54 {"role": "user", "content": f"""
55Zadanie: {task.description}
56Kontekst: {task.context}
57Oczekiwany output: {task.expected_output}
58"""}
59 ]
60
61 response = client.chat.completions.create(
62 model=self.model,
63 messages=messages,
64 tools=self.tools if self.tools else None
65 )
66
67 output = response.choices[0].message.content
68 self.memory.append(f"Task: {task.description} -> Result: {output[:100]}...")
69
70 return SubagentResult(
71 success=True,
72 output=output,
73 metadata={"model": self.model, "role": self.role.value}
74 )Wyspecjalizowane Subagenty
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"""
37
38
39class ExecutorSubagent(Subagent):
40 """Subagent wykonawczy - realizuje zadania."""
41
42 @property
43 def system_prompt(self) -> str:
44 return """Jesteś wyspecjalizowanym agentem wykonawczym.
45
46Twoje zadania:
471. Realizacja zaplanowanych działań
482. Używanie narzędzi do wykonania zadań
493. Raportowanie postępów
504. Obsługa błędów
51
52Zawsze:
53- Wykonuj zadania krok po kroku
54- Weryfikuj wyniki każdej akcji
55- Raportuj problemy natychmiast"""
56
57
58class ValidatorSubagent(Subagent):
59 """Subagent walidujący - sprawdza jakość."""
60
61 @property
62 def system_prompt(self) -> str:
63 return """Jesteś wyspecjalizowanym agentem walidacyjnym.
64
65Twoje zadania:
661. Weryfikacja poprawności wyników
672. Sprawdzanie zgodności z wymaganiami
683. Identyfikacja błędów i niedociągnięć
694. Sugerowanie poprawek
70
71Zawsze:
72- Bądź krytyczny ale konstruktywny
73- Używaj konkretnych kryteriów
74- Proponuj rozwiązania problemów"""Orchestrator - Zarządzanie Subagentami
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 ])
23
24 async def plan_execution(self, task: str) -> List[SubagentTask]:
25 """Planuje wykonanie zadania przez odpowiednich subagentów."""
26 planning_prompt = f"""Masz do wykonania zadanie: {task}
27
28Dostępni subagenci:
29{self.get_available_subagents()}
30
31Zaplanuj wykonanie zadania. Dla każdego kroku określ:
321. Który subagent powinien wykonać krok
332. Opis zadania dla subagenta
343. Oczekiwany output
354. Priorytet (1-5)
36
37Odpowiedz w formacie JSON:
38[{{"subagent": "nazwa", "task": "opis", "expected_output": "...", "priority": 1}}, ...]"""
39
40 response = client.chat.completions.create(
41 model=self.model,
42 messages=[
43 {"role": "system", "content": "Jesteś planistą zadań. Odpowiadaj tylko w JSON."},
44 {"role": "user", "content": planning_prompt}
45 ],
46 response_format={"type": "json_object"}
47 )
48
49 import json
50 plan = json.loads(response.choices[0].message.content)
51
52 tasks = []
53 for step in plan.get("steps", plan):
54 tasks.append(SubagentTask(
55 description=step["task"],
56 context=task,
57 expected_output=step["expected_output"],
58 priority=step.get("priority", 1)
59 ))
60
61 return tasks
62
63 async def execute_task(self, task: str) -> str:
64 """Wykonuje zadanie używając subagentów."""
65 # 1. Planowanie
66 planned_tasks = await self.plan_execution(task)
67 print(f"Zaplanowano {len(planned_tasks)} kroków")
68
69 # 2. Wykonanie
70 results = []
71 for i, subtask in enumerate(planned_tasks):
72 # Znajdź odpowiedniego subagenta
73 subagent = self._select_subagent(subtask)
74
75 if subagent:
76 print(f"Krok {i+1}: {subagent.name} wykonuje: {subtask.description[:50]}...")
77 result = await subagent.execute(subtask)
78 results.append({
79 "step": i + 1,
80 "subagent": subagent.name,
81 "task": subtask.description,
82 "result": result.output
83 })
84
85 # 3. Synteza
86 final_result = await self._synthesize_results(task, results)
87
88 return final_result
89
90 def _select_subagent(self, task: SubagentTask) -> Optional[Subagent]:
91 """Wybiera najlepszego subagenta dla zadania."""
92 # Prosta heurystyka - można rozbudować o ML
93 keywords = {
94 "research": ["zbadaj", "znajdź", "wyszukaj", "informacje"],
95 "analyst": ["analizuj", "oceń", "porównaj", "wnioski"],
96 "executor": ["wykonaj", "zrób", "stwórz", "zaimplementuj"],
97 "validator": ["sprawdź", "zweryfikuj", "waliduj", "przetestuj"]
98 }
99
100 task_lower = task.description.lower()
101
102 for subagent_type, kws in keywords.items():
103 if any(kw in task_lower for kw in kws):
104 for agent in self.subagents.values():
105 if agent.role.value == subagent_type:
106 return agent
107
108 # Domyślnie pierwszy dostępny
109 return list(self.subagents.values())[0] if self.subagents else None
110
111 async def _synthesize_results(self, original_task: str, results: List[dict]) -> str:
112 """Syntezuje wyniki ze wszystkich subagentów."""
113 results_summary = "\n\n".join([
114 f"Krok {r['step']} ({r['subagent']}): {r['result']}"
115 for r in results
116 ])
117
118 synthesis_prompt = f"""Oryginalne zadanie: {original_task}
119
120Wyniki subagentów:
121{results_summary}
122
123Stwórz spójną, końcową odpowiedź na podstawie wyników wszystkich subagentów."""
124
125 response = client.chat.completions.create(
126 model=self.model,
127 messages=[
128 {"role": "system", "content": "Jesteś syntetykiem. Łączysz wyniki w spójną całość."},
129 {"role": "user", "content": synthesis_prompt}
130 ]
131 )
132
133 return response.choices[0].message.contentUżycie Orchestratora
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())Hierarchia Wielopoziomowa
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)
27
28 def _assess_complexity(self, task: str) -> int:
29 """Ocenia złożoność zadania (1-5)."""
30 # Prosta heurystyka
31 complexity_indicators = [
32 "kompletny", "szczegółowy", "wieloetapowy",
33 "kompleksowy", "rozbudowany"
34 ]
35 return sum(1 for ind in complexity_indicators if ind in task.lower()) + 1
36
37 def _select_sub_orchestrator(self, task: str) -> "HierarchicalOrchestrator":
38 """Wybiera pod-orchestratora dla zadania."""
39 # Można rozbudować o inteligentny wybór
40 return list(self.sub_orchestrators.values())[0]
41
42 async def _execute_locally(self, task: str) -> str:
43 """Wykonuje zadanie lokalnie."""
44 # Użyj własnych subagentów
45 results = []
46 for subagent in self.subagents.values():
47 subtask = SubagentTask(
48 description=task,
49 context=f"Level {self.level}",
50 expected_output="Rezultat zadania"
51 )
52 result = await subagent.execute(subtask)
53 results.append(result.output)
54
55 return "\n".join(results)
56
57
58# Przykład użycia hierarchii
59async def hierarchical_example():
60 # Główny orchestrator
61 main_orchestrator = HierarchicalOrchestrator("Main Orchestrator")
62
63 # Pod-orchestrator dla badań
64 research_orchestrator = HierarchicalOrchestrator("Research Team")
65 research_orchestrator.subagents["researcher"] = ResearchSubagent(
66 "researcher", SubagentRole.RESEARCHER
67 )
68
69 # Pod-orchestrator dla analizy
70 analysis_orchestrator = HierarchicalOrchestrator("Analysis Team")
71 analysis_orchestrator.subagents["analyst"] = AnalysisSubagent(
72 "analyst", SubagentRole.ANALYST
73 )
74
75 # Dodaj do hierarchii
76 main_orchestrator.add_sub_orchestrator(research_orchestrator)
77 main_orchestrator.add_sub_orchestrator(analysis_orchestrator)
78
79 # Wykonaj złożone zadanie
80 result = await main_orchestrator.delegate_task(
81 "Przeprowadź kompletne badanie rynku Safari w Afryce Wschodniej"
82 )
83 print(result)Komunikacja między Subagentami
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)
23
24class MessageBus:
25 """Szyna komunikacyjna dla agentów."""
26
27 def __init__(self):
28 self.messages: List[AgentMessage] = []
29 self.subscribers: Dict[str, List[callable]] = {}
30
31 def publish(self, message: AgentMessage) -> None:
32 """Publikuje wiadomość."""
33 self.messages.append(message)
34
35 # Powiadom subskrybentów
36 if message.receiver in self.subscribers:
37 for callback in self.subscribers[message.receiver]:
38 callback(message)
39
40 def subscribe(self, agent_name: str, callback: callable) -> None:
41 """Subskrybuje agenta do wiadomości."""
42 if agent_name not in self.subscribers:
43 self.subscribers[agent_name] = []
44 self.subscribers[agent_name].append(callback)
45
46 def get_messages_for(self, agent_name: str) -> List[AgentMessage]:
47 """Pobiera wiadomości dla agenta."""
48 return [m for m in self.messages if m.receiver == agent_name]
49
50
51class CommunicatingSubagent(Subagent):
52 """Subagent z możliwością komunikacji."""
53
54 def __init__(self, name: str, role: SubagentRole, message_bus: MessageBus):
55 super().__init__(name, role)
56 self.message_bus = message_bus
57 self.message_bus.subscribe(name, self._handle_message)
58 self.pending_responses: Dict[str, str] = {}
59
60 def _handle_message(self, message: AgentMessage) -> None:
61 """Obsługuje przychodzące wiadomości."""
62 if message.message_type == MessageType.QUERY:
63 # Odpowiedz na zapytanie
64 response = self._process_query(message.content)
65 self.send_message(
66 message.sender,
67 response,
68 MessageType.RESPONSE,
69 correlation_id=message.correlation_id
70 )
71 elif message.message_type == MessageType.RESPONSE:
72 # Zapisz odpowiedź
73 if message.correlation_id:
74 self.pending_responses[message.correlation_id] = message.content
75
76 def send_message(
77 self,
78 receiver: str,
79 content: str,
80 message_type: MessageType,
81 correlation_id: Optional[str] = None
82 ) -> None:
83 """Wysyła wiadomość do innego agenta."""
84 import uuid
85 message = AgentMessage(
86 sender=self.name,
87 receiver=receiver,
88 content=content,
89 message_type=message_type,
90 correlation_id=correlation_id or str(uuid.uuid4())
91 )
92 self.message_bus.publish(message)
93
94 def _process_query(self, query: str) -> str:
95 """Przetwarza zapytanie od innego agenta."""
96 # Implementacja zależna od roli
97 return f"Odpowiedź na: {query}"
98
99 async def ask_other_agent(self, agent_name: str, question: str) -> str:
100 """Pyta innego agenta i czeka na odpowiedź."""
101 import uuid
102 correlation_id = str(uuid.uuid4())
103
104 self.send_message(
105 agent_name,
106 question,
107 MessageType.QUERY,
108 correlation_id
109 )
110
111 # Czekaj na odpowiedź (z timeout)
112 import asyncio
113 for _ in range(10):
114 if correlation_id in self.pending_responses:
115 return self.pending_responses.pop(correlation_id)
116 await asyncio.sleep(0.1)
117
118 return "Timeout - brak odpowiedzi"Best Practices dla Subagentów
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"""
32
33class RobustOrchestrator(AgentOrchestrator):
34 """Orchestrator z obsługą błędów i fallback."""
35
36 async def execute_with_fallback(self, task: str) -> str:
37 """Wykonuje zadanie z obsługą błędów."""
38 try:
39 return await self.execute_task(task)
40 except Exception as e:
41 print(f"Błąd głównego wykonania: {e}")
42
43 # Fallback - prostsze podejście
44 return await self._simple_execution(task)
45
46 async def _simple_execution(self, task: str) -> str:
47 """Proste wykonanie jako fallback."""
48 response = client.chat.completions.create(
49 model=self.model,
50 messages=[
51 {"role": "system", "content": "Jesteś pomocnym asystentem."},
52 {"role": "user", "content": task}
53 ]
54 )
55 return response.choices[0].message.contentSubagenty 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!