Multi-Agent Systems

Poproś jeden model, żeby zebrał informacje, przeanalizował je, napisał artykuł i jeszcze go skrytykował. Dostaniesz tekst, w którym wszystko jest trochę, a nic porządnie, bo jeden prompt ciągnie model w czterech kierunkach naraz. Na sawannie też nikt nie robi wszystkiego sam.

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. Każdy agent dostaje własny prompt systemowy i robi jedną rzecz dobrze.

Podstawy Multi-Agent

Najpierw ustalmy słownik. Role zapiszemy jako Enum, czyli zamknięty zbiór nazwanych stałych, a wiadomość między agentami jako @dataclass:

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"

AgentMessage pamięta nadawcę, odbiorcę, treść i typ wiadomości. Domyślny typ to "task", więc przy zwykłym zleceniu nie trzeba go podawać.

Teraz bazowa klasa agenta. Dziedziczy po ABC, a właściwość system_prompt jest oznaczona @abstractmethod, więc każda klasa potomna musi ją zdefiniować:

1class BaseAgent(ABC):
2    """Bazowa klasa agenta."""
3
4    def __init__(self, name: str, role: AgentRole, model: str = "gpt-4o-mini"):
5        self.name = name
6        self.role = role
7        self.model = model
8        self.client = OpenAI()
9        self.memory: list[AgentMessage] = []
10
11    @property
12    @abstractmethod
13    def system_prompt(self) -> str:
14        """Prompt systemowy agenta."""
15        pass
16
17    def process(self, message: AgentMessage) -> str:
18        """Przetwarza wiadomość i zwraca odpowiedź."""
19        self.memory.append(message)
20
21        response = self.client.chat.completions.create(
22            model=self.model,
23            messages=[
24                {"role": "system", "content": self.system_prompt},
25                {"role": "user", "content": message.content}
26            ]
27        )
28
29        return response.choices[0].message.content

Metoda process zapisuje wiadomość w pamięci agenta i wysyła do modelu dwie wiadomości: prompt systemowy z rolą i treść zadania. Nie da się utworzyć obiektu samego BaseAgent, Python zgłosi TypeError, bo brakuje mu implementacji system_prompt.

Implementacja agentów specjalistycznych

Agenci specjaliści różnią się wyłącznie promptem. Pierwsza para to badacz i analityk:

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."""

Każda klasa nadpisuje tylko system_prompt, a cała logika wywołania modelu pozostaje w BaseAgent. Tak wygląda dziedziczenie w praktyce: wspólny kod raz, różnice w podklasach.

Druga para to pisarz i krytyk:

1class WriterAgent(BaseAgent):
2    """Agent pisarski - tworzy treści."""
3
4    @property
5    def system_prompt(self) -> str:
6        return """Jesteś agentem pisarskim. Twoja rola to:
71. Tworzenie angażujących treści
82. Dostosowanie stylu do odbiorcy
93. Strukturyzowanie tekstu
104. Dbanie o klarowność przekazu
11
12Pisz zwięźle i na temat."""
13
14class CriticAgent(BaseAgent):
15    """Agent krytyczny - recenzuje i poprawia."""
16
17    @property
18    def system_prompt(self) -> str:
19        return """Jesteś agentem krytycznym. Twoja rola to:
201. Ocena jakości treści
212. Identyfikacja błędów i niedociągnięć
223. Sugerowanie ulepszeń
234. Weryfikacja faktów
24
25Bądź konstruktywny, ale szczery."""

Researcher zbiera informacje, Analyzer je analizuje, Writer tworzy treść, a Critic recenzuje i poprawia. To typowe role w systemie multi-agent. Zwróć uwagę, że rola "Destroyer, który usuwa pliki" nigdy tu nie występuje: agent ma pomagać, a nie niszczyć.

Orkiestracja agentów

Ktoś musi przekazywać wiadomości między członkami stada. Tę rolę pełni orkiestrator, który rejestruje agentów i pośredniczy w każdej rozmowie:

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

send_message sprawdza, czy odbiorca istnieje, zapisuje wiadomość w historii, wywołuje process i zapisuje też odpowiedź. Dzięki temu cała rozmowa zostaje w jednym miejscu.

Metoda run_pipeline układa agentów w łańcuch, w którym wynik jednego staje się wejściem następnego:

1    def run_pipeline(self, task: str) -> dict[str, str]:
2        """Uruchamia pipeline przetwarzania."""
3        results = {}
4
5        # 1. Researcher zbiera informacje
6        research = self.send_message("user", "researcher",
7            f"Zbierz informacje na temat: {task}")
8        results["research"] = research
9
10        # 2. Analyzer analizuje
11        analysis = self.send_message("researcher", "analyzer",
12            f"Przeanalizuj te informacje:\n{research}")
13        results["analysis"] = analysis
14
15        # 3. Writer tworzy treść
16        draft = self.send_message("analyzer", "writer",
17            f"Na podstawie analizy napisz artykuł:\n{analysis}")
18        results["draft"] = draft
19
20        # 4. Critic recenzuje
21        review = self.send_message("writer", "critic",
22            f"Zrecenzuj ten artykuł:\n{draft}")
23        results["review"] = review
24
25        return results

Kolejność jest stała: Researcher zbiera informacje, Analyzer analizuje dane, Writer tworzy treść, Critic recenzuje. Słownik results zachowuje wynik każdego etapu, a nie tylko ostatni.

Pozostało zarejestrować czterech agentów i uruchomić pipeline:

1# Użycie
2orchestrator = AgentOrchestrator()
3orchestrator.register_agent(ResearcherAgent("researcher", AgentRole.RESEARCHER))
4orchestrator.register_agent(AnalyzerAgent("analyzer", AgentRole.ANALYZER))
5orchestrator.register_agent(WriterAgent("writer", AgentRole.WRITER))
6orchestrator.register_agent(CriticAgent("critic", AgentRole.CRITIC))
7
8results = orchestrator.run_pipeline("Przyszłość sztucznej inteligencji w 2025")

Każdy agent jest zapisany pod swoją nazwą, więc send_message("user", "researcher", ...) trafia dokładnie tam, gdzie trzeba.

Komunikacja asynchroniczna

Agenci nie muszą rozmawiać wyłącznie po kolei. Można im zostawiać wiadomości w kolejce asyncio.Queue, jak notatki przy wodopoju:

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()

Pętla process_messages czeka na wiadomość słowem await, obsługuje ją i kończy pracę, gdy dostanie sygnał "STOP".

Metoda run_parallel_tasks wkłada zadania do kolejki i uruchamia przetwarzanie:

1    async def run_parallel_tasks(self, tasks: list[tuple[str, str]]) -> dict:
2        """Uruchamia zadania równolegle."""
3        # Dodaj zadania do kolejki
4        for agent_name, task in tasks:
5            await self.message_queue.put(
6                AgentMessage("system", agent_name, task)
7            )
8
9        # Dodaj sygnał końca
10        await self.message_queue.put(
11            AgentMessage("system", "STOP", "STOP")
12        )
13
14        # Uruchom przetwarzanie
15        await self.process_messages()
16
17        return self.results
18
19# Przykład równoległego przetwarzania
20async def main():
21    system = AsyncAgentSystem()
22    # ... rejestracja agentów ...
23
24    tasks = [
25        ("researcher", "Zbadaj temat X"),
26        ("analyzer", "Przeanalizuj dane Y"),
27    ]
28
29    results = await system.run_parallel_tasks(tasks)
30    print(results)
31
32asyncio.run(main())

Uczciwie: mimo nazwy te zadania nie wykonują się równolegle. Jest jeden konsument kolejki, a agent.process to zwykłe, blokujące wywołanie. Prawdziwą równoległość dałby asynchroniczny klient AsyncOpenAI i asyncio.gather albo asyncio.to_thread. Kolejka przydaje się jednak do rozdzielenia nadawców od odbiorców.

Narzędzia agenta

Agent, który tylko pisze, jest jak tropiciel bez lornetki. Narzędzia pozwalają mu działać: przeszukać bazę, pobrać pogodę, policzyć koszt. W LangChain funkcję zamienia się w narzędzie dekoratorem @tool:

1from langchain_core.tools import tool
2
3@tool
4def search_db(query: str) -> list:
5    """Przeszukuje bazę obserwacji zwierząt."""
6    return [f"Obserwacja pasująca do: {query}"]
7
8print(search_db.name)           # search_db
9print(search_db.invoke("lwy"))  # ['Obserwacja pasująca do: lwy']

Docstring nie jest tu ozdobą: model czyta go jako opis narzędzia, a bez niego dekorator zgłosi błąd. Typy parametrów mówią modelowi, jakie argumenty podać. Agent z narzędziami pracuje w pętli: planuje działanie, wykonuje akcję, obserwuje wynik i robi refleksję, czy poprawić kurs.

Wzorce współpracy agentów

Na koniec trzy klasyczne wzorce. Chain to łańcuch, w którym każdy agent przekazuje wynik kolejnemu:

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

To uogólnienie naszego run_pipeline: lista agentów może mieć dowolną długość.

Debate to dyskusja dwóch agentów przez kilka rund:

1    @staticmethod
2    def debate(agent1: BaseAgent, agent2: BaseAgent, topic: str, rounds: int = 3) -> list[str]:
3        """Debate - agenci dyskutują na temat."""
4        conversation = []
5
6        current = f"Dyskusja na temat: {topic}. Przedstaw swoje stanowisko."
7
8        for i in range(rounds):
9            # Agent 1
10            response1 = agent1.process(
11                AgentMessage(agent2.name, agent1.name, current)
12            )
13            conversation.append(f"{agent1.name}: {response1}")
14
15            # Agent 2 odpowiada
16            response2 = agent2.process(
17                AgentMessage(agent1.name, agent2.name, response1)
18            )
19            conversation.append(f"{agent2.name}: {response2}")
20            current = response2
21
22        return conversation

Każda odpowiedź jednego agenta staje się wejściem drugiego, a lista conversation zapisuje całą wymianę zdań.

Consensus zbiera opinie wszystkich i prosi jednego agenta o syntezę:

1    @staticmethod
2    def consensus(agents: list[BaseAgent], question: str) -> str:
3        """Consensus - agenci dochodzą do wspólnego wniosku."""
4        opinions = []
5
6        # Zbierz opinie
7        for agent in agents:
8            opinion = agent.process(
9                AgentMessage("system", agent.name, question)
10            )
11            opinions.append(f"{agent.name}: {opinion}")
12
13        # Synteza (użyj jednego z agentów)
14        synthesis_prompt = f"""Pytanie: {question}
15
16Opinie agentów:
17{chr(10).join(opinions)}
18
19Sformułuj wspólny konsensus na podstawie tych opinii."""
20
21        return agents[0].process(
22            AgentMessage("system", agents[0].name, synthesis_prompt)
23        )

chr(10) to znak nowej linii, użyty zamiast "\n" wewnątrz f-stringa. Polecam zaczynać od prostego łańcucha, a debatę i konsensus dokładać dopiero tam, gdzie jedna perspektywa naprawdę nie wystarcza, bo każda runda to kolejne płatne wywołania modelu.

Systemy multi-agentowe otwierają drzwi do złożonych aplikacji AI. W następnej lekcji poznasz CrewAI - framework do tworzenia zespołów agentów, który daje te wzorce w gotowej postaci!

Zapamiętaj: dobry system agentów działa jak stado, w którym każdy zna swoją rolę, a przewodnik pilnuje kierunku.