AI Agents · Lekcja

Protokoły komunikacji (magistrale komunikatów)

Rozdziel agentów za pomocą magistrali komunikatów (Redis, NATS, Kafka), aby mogli niezależnie się skalować i ulegać awariom.

Protokoły komunikacji (magistrale komunikatów) to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 4 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.

Części tej lekcji nie zostały jeszcze przetłumaczone i są wyświetlane po angielsku.

Poza komunikacją w procesie

W małych systemach agenci wywołują się wzajemnie w ramach jednego procesu. W większych systemach należy ich rozdzielić za pomocą brokera wiadomości — Redis, NATS lub Kafka.

Po co rozdzielać?

Agenci mogą być skalowani niezależnie
Awarie są izolowane dla poszczególnych agentów
Asynchroniczne przepływy pracy (nie trzeba synchronicznie czekać na wolnych agentów)
Możliwość ponownego odtwarzania wiadomości na potrzeby debugowania
Obsługa wielu języków: agent w Pythonie i agent w Go mogą subskrybować te same wiadomości

Simple Bus: Redis Pub/Sub

import redis
r = redis.Redis()

# Publisher
r.publish('agent.research.task', json.dumps({'task_id': 'abc', 'query': '...'}))

# Subscriber
p = r.pubsub()
p.subscribe('agent.research.task')
for msg in p.listen():
    if msg['type'] == 'message':
        handle_task(json.loads(msg['data']))

Pub/Sub do rozgłaszania

Należy używać go, gdy wiele agentów może chcieć otrzymać informacje o zdarzeniu (np. „user-question-received”).

Kolejki do dystrybucji zadań

Należy użyć kolejki (Redis BLPOP, RabbitMQ), gdy dokładnie jeden worker powinien obsłużyć każde zadanie:

import json, queue

r = queue.Queue()

# Producer
r.put(json.dumps({'task_id': 'abc', 'kind': 'research'}))

def process(task):
    print('processing', task)

# Worker
while not r.empty():
    raw = r.get()
    task = json.loads(raw)
    process(task)

NATS dla szybkości

NATS to lekki broker wiadomości przeznaczony dla mikrousług. Opóźnienie poniżej milisekundy i wbudowany mechanizm request/reply:

import nats
nc = await nats.connect('nats://localhost:4222')

await nc.publish('agent.research', json.dumps(task).encode())

# Request/reply
response = await nc.request('agent.research', payload, timeout=10)

Kafka dla trwałości

Kafka dodaje trwałe dzienniki z możliwością odtwarzania. Idealne rozwiązanie do audytu i ponownego odtwarzania:

from confluent_kafka import Producer, Consumer

producer.produce('agent-events', key=task_id, value=json.dumps(task))
producer.flush()

Schematy zdarzeń

Należy definiować format każdej wiadomości za pomocą Pydantic lub Protobuf. Bez schematów broker szybko staje się chaotyczny:

class ResearchTask(BaseModel):
    task_id: str
    user_id: str
    query: str
    deadline: datetime

class ResearchResult(BaseModel):
    task_id: str
    findings: list[str]
    duration_ms: int

Identyfikatory korelacji

Każda powiązana wiadomość zawiera ten sam correlation_id, dzięki czemu można śledzić zadanie w całej infrastrukturze usług:

record = {'correlation_id': 'abc', 'task_id': 'def', 'span': 'llm_call'}
print(record)

Idempotentne handlery

Wiadomości mogą być dostarczane więcej niż raz (semantyka co najmniej jednokrotnego dostarczenia). Handlery muszą być idempotentne — zadanie o tym samym task_id jest przetwarzane tylko raz.

Kolejka dead-letter

Gdy handler wielokrotnie kończy działanie błędem, należy przenieść wiadomość do DLQ w celu weryfikacji przez człowieka, zamiast zapętlać proces w nieskończoność:

if attempts > MAX_RETRIES:
    r.rpush('queue:research:dlq', raw)
    log.error('Sent to DLQ', extra={'task_id': task_id})

Obserwowalność

Należy śledzić każdą wiadomość:

Identyfikatory śledzenia przepływają przez brokera
Logi zawierają task_id i correlation_id
Metryki: przepustowość, opóźnienie i współczynnik błędów dla każdego tematu

Kiedy wdrożyć

Należy zacząć od prostego rozwiązania — agentów działających w procesie. Na brokera należy przejść dopiero wtedy, gdy występuje:

co najmniej 5 agentów na produkcji
potrzeba niezależnego skalowania agentów
przepływy pracy obejmujące wiele usług

Co najmniej raz?

Czego wymaga od handlerów „co najmniej jednokrotne dostarczenie”?

Podsumowanie

W prostych przypadkach należy rozdzielać komponenty za pomocą Redis, dla niskich opóźnień używać NATS, a dla trwałości — Kafki. Schematy, identyfikatory korelacji, idempotentne handlery i DLQ.

Często zadawane pytania

Czy lekcja „Protokoły komunikacji (magistrale komunikatów)” jest bezpłatna?

Tak — pełny tekst „Protokoły komunikacji (magistrale komunikatów)” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.

Co nauczysz się w „Protokoły komunikacji (magistrale komunikatów)”?

Rozdziel agentów za pomocą magistrali komunikatów (Redis, NATS, Kafka), aby mogli niezależnie się skalować i ulegać awariom. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Agents?

Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 4 z 4.

Ile czasu zajmuje lekcja „Protokoły komunikacji (magistrale komunikatów)”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?

Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

Multi-agent oparty na rozmowie (AutoGen)
Hierarchiczni nadzorcy (koordynator + wykonawcy)
Role i specjalizacje agentów
Protokoły komunikacji (magistrale komunikatów)

← Powrót do AI Agents