Utilizziamo i cookie per migliorare la tua esperienza sul sito
Multi-Agent Systems
Systemy multi-agentowe to architektura, gdzie wiele wyspecjalizowanych agentów AI współpracuje przy rozwiązywaniu złożonych zadań. To jak stado zwierząt, gdzie każdy osobnik ma swoją rolę - łowcy, obserwatorzy, obrońcy!
Podstawy Multi-Agent
1from dataclasses import dataclass
2from enum import Enum
3from abc import ABC, abstractmethod
4from openai import OpenAI
5
6class AgentRole(Enum):
7 """Role agentów w systemie."""
8 RESEARCHER = "researcher" # Zbiera informacje
9 ANALYZER = "analyzer" # Analizuje dane
10 WRITER = "writer" # Pisze treści
11 CRITIC = "critic" # Recenzuje i poprawia
12 COORDINATOR = "coordinator" # Koordynuje pracę
13
14@dataclass
15class AgentMessage:
16 """Wiadomość między agentami."""
17 sender: str
18 receiver: str
19 content: str
20 message_type: str = "task"
21
22class BaseAgent(ABC):
23 """Bazowa klasa agenta."""
24
25 def __init__(self, name: str, role: AgentRole, model: str = "gpt-4o-mini"):
26 self.name = name
27 self.role = role
28 self.model = model
29 self.client = OpenAI()
30 self.memory: list[AgentMessage] = []
31
32 @property
33 @abstractmethod
34 def system_prompt(self) -> str:
35 """Prompt systemowy agenta."""
36 pass
37
38 def process(self, message: AgentMessage) -> str:
39 """Przetwarza wiadomość i zwraca odpowiedź."""
40 self.memory.append(message)
41
42 response = self.client.chat.completions.create(
43 model=self.model,
44 messages=[
45 {"role": "system", "content": self.system_prompt},
46 {"role": "user", "content": message.content}
47 ]
48 )
49
50 return response.choices[0].message.contentImplementacja agentów specjalistycznych
1class ResearcherAgent(BaseAgent):
2 """Agent badawczy - zbiera informacje."""
3
4 @property
5 def system_prompt(self) -> str:
6 return """Jesteś agentem badawczym. Twoja rola to:
71. Zbieranie informacji na zadany temat
82. Identyfikacja kluczowych faktów
93. Weryfikacja źródeł
104. Strukturyzowanie wiedzy
11
12Zawsze podawaj źródła i dziel informacje na kategorie."""
13
14class AnalyzerAgent(BaseAgent):
15 """Agent analityczny - analizuje dane."""
16
17 @property
18 def system_prompt(self) -> str:
19 return """Jesteś agentem analitycznym. Twoja rola to:
201. Analiza dostarczonych danych
212. Identyfikacja wzorców i trendów
223. Formułowanie wniosków
234. Ocena ryzyka i szans
24
25Bądź precyzyjny i podawaj konkretne liczby."""
26
27class WriterAgent(BaseAgent):
28 """Agent pisarski - tworzy treści."""
29
30 @property
31 def system_prompt(self) -> str:
32 return """Jesteś agentem pisarskim. Twoja rola to:
331. Tworzenie angażujących treści
342. Dostosowanie stylu do odbiorcy
353. Strukturyzowanie tekstu
364. Dbanie o klarowność przekazu
37
38Pisz zwięźle i na temat."""
39
40class CriticAgent(BaseAgent):
41 """Agent krytyczny - recenzuje i poprawia."""
42
43 @property
44 def system_prompt(self) -> str:
45 return """Jesteś agentem krytycznym. Twoja rola to:
461. Ocena jakości treści
472. Identyfikacja błędów i niedociągnięć
483. Sugerowanie ulepszeń
494. Weryfikacja faktów
50
51Bądź konstruktywny, ale szczery."""Orkiestracja agentów
1from typing import Optional
2
3class AgentOrchestrator:
4 """Koordynator pracy agentów."""
5
6 def __init__(self):
7 self.agents: dict[str, BaseAgent] = {}
8 self.conversation_history: list[AgentMessage] = []
9
10 def register_agent(self, agent: BaseAgent) -> None:
11 """Rejestruje agenta w systemie."""
12 self.agents[agent.name] = agent
13 print(f"Zarejestrowano agenta: {agent.name} ({agent.role.value})")
14
15 def send_message(
16 self,
17 sender: str,
18 receiver: str,
19 content: str
20 ) -> str:
21 """Wysyła wiadomość między agentami."""
22 if receiver not in self.agents:
23 raise ValueError(f"Agent {receiver} nie istnieje!")
24
25 message = AgentMessage(
26 sender=sender,
27 receiver=receiver,
28 content=content
29 )
30
31 self.conversation_history.append(message)
32
33 response = self.agents[receiver].process(message)
34
35 # Zapisz odpowiedź
36 response_message = AgentMessage(
37 sender=receiver,
38 receiver=sender,
39 content=response,
40 message_type="response"
41 )
42 self.conversation_history.append(response_message)
43
44 return response
45
46 def run_pipeline(self, task: str) -> dict[str, str]:
47 """Uruchamia pipeline przetwarzania."""
48 results = {}
49
50 # 1. Researcher zbiera informacje
51 research = self.send_message("user", "researcher",
52 f"Zbierz informacje na temat: {task}")
53 results["research"] = research
54
55 # 2. Analyzer analizuje
56 analysis = self.send_message("researcher", "analyzer",
57 f"Przeanalizuj te informacje:\n{research}")
58 results["analysis"] = analysis
59
60 # 3. Writer tworzy treść
61 draft = self.send_message("analyzer", "writer",
62 f"Na podstawie analizy napisz artykuł:\n{analysis}")
63 results["draft"] = draft
64
65 # 4. Critic recenzuje
66 review = self.send_message("writer", "critic",
67 f"Zrecenzuj ten artykuł:\n{draft}")
68 results["review"] = review
69
70 return results
71
72# Użycie
73orchestrator = AgentOrchestrator()
74orchestrator.register_agent(ResearcherAgent("researcher", AgentRole.RESEARCHER))
75orchestrator.register_agent(AnalyzerAgent("analyzer", AgentRole.ANALYZER))
76orchestrator.register_agent(WriterAgent("writer", AgentRole.WRITER))
77orchestrator.register_agent(CriticAgent("critic", AgentRole.CRITIC))
78
79results = orchestrator.run_pipeline("Przyszłość sztucznej inteligencji w 2025")Komunikacja asynchroniczna
1import asyncio
2from typing import Callable
3
4class AsyncAgentSystem:
5 """Asynchroniczny system agentów."""
6
7 def __init__(self):
8 self.agents: dict[str, BaseAgent] = {}
9 self.message_queue: asyncio.Queue = asyncio.Queue()
10 self.results: dict[str, str] = {}
11
12 async def process_messages(self) -> None:
13 """Przetwarza wiadomości z kolejki."""
14 while True:
15 message = await self.message_queue.get()
16
17 if message.content == "STOP":
18 break
19
20 agent = self.agents.get(message.receiver)
21 if agent:
22 response = agent.process(message)
23 self.results[message.receiver] = response
24
25 self.message_queue.task_done()
26
27 async def run_parallel_tasks(self, tasks: list[tuple[str, str]]) -> dict:
28 """Uruchamia zadania równolegle."""
29 # Dodaj zadania do kolejki
30 for agent_name, task in tasks:
31 await self.message_queue.put(
32 AgentMessage("system", agent_name, task)
33 )
34
35 # Dodaj sygnał końca
36 await self.message_queue.put(
37 AgentMessage("system", "STOP", "STOP")
38 )
39
40 # Uruchom przetwarzanie
41 await self.process_messages()
42
43 return self.results
44
45# Przykład równoległego przetwarzania
46async def main():
47 system = AsyncAgentSystem()
48 # ... rejestracja agentów ...
49
50 tasks = [
51 ("researcher", "Zbadaj temat X"),
52 ("analyzer", "Przeanalizuj dane Y"),
53 ]
54
55 results = await system.run_parallel_tasks(tasks)
56 print(results)
57
58asyncio.run(main())Wzorce współpracy agentów
1class CollaborationPatterns:
2 """Wzorce współpracy agentów."""
3
4 @staticmethod
5 def chain(agents: list[BaseAgent], initial_input: str) -> str:
6 """Chain - każdy agent przekazuje output kolejnemu."""
7 current_output = initial_input
8
9 for agent in agents:
10 message = AgentMessage("system", agent.name, current_output)
11 current_output = agent.process(message)
12
13 return current_output
14
15 @staticmethod
16 def debate(agent1: BaseAgent, agent2: BaseAgent, topic: str, rounds: int = 3) -> list[str]:
17 """Debate - agenci dyskutują na temat."""
18 conversation = []
19
20 current = f"Dyskusja na temat: {topic}. Przedstaw swoje stanowisko."
21
22 for i in range(rounds):
23 # Agent 1
24 response1 = agent1.process(
25 AgentMessage(agent2.name, agent1.name, current)
26 )
27 conversation.append(f"{agent1.name}: {response1}")
28
29 # Agent 2 odpowiada
30 response2 = agent2.process(
31 AgentMessage(agent1.name, agent2.name, response1)
32 )
33 conversation.append(f"{agent2.name}: {response2}")
34 current = response2
35
36 return conversation
37
38 @staticmethod
39 def consensus(agents: list[BaseAgent], question: str) -> str:
40 """Consensus - agenci dochodzą do wspólnego wniosku."""
41 opinions = []
42
43 # Zbierz opinie
44 for agent in agents:
45 opinion = agent.process(
46 AgentMessage("system", agent.name, question)
47 )
48 opinions.append(f"{agent.name}: {opinion}")
49
50 # Synteza (użyj jednego z agentów)
51 synthesis_prompt = f"""Pytanie: {question}
52
53Opinie agentów:
54{chr(10).join(opinions)}
55
56Sformułuj wspólny konsensus na podstawie tych opinii."""
57
58 return agents[0].process(
59 AgentMessage("system", agents[0].name, synthesis_prompt)
60 )Systemy multi-agentowe otwierają drzwi do złożonych aplikacji AI. W następnej lekcji poznasz CrewAI - framework do tworzenia zespołów agentów!