Protocolli di comunicazione (message bus)
Disaccoppi gli agenti con un message bus, come Redis, NATS o Kafka, così potranno scalare e subire guasti in modo indipendente.
Protocolli di comunicazione (message bus) è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
Parti di questa lezione non sono ancora state tradotte e vengono mostrate in inglese.
Oltre le conversazioni in-process
Nei sistemi piccoli, gli agenti si chiamano tra loro in-process. Nei sistemi più grandi, li disaccoppi tramite un message bus — Redis, NATS o Kafka.
Perché disaccoppiare?
- Gli agenti possono scalare in modo indipendente
- I guasti restano isolati per agente
- Flussi di lavoro asincroni (non è necessario attendere inline gli agenti lenti)
- È possibile riprodurre i messaggi per il debugging
- Compatibilità tra linguaggi: un agente Python e uno Go possono entrambi sottoscriversi
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 per la trasmissione a tutti
Lo usi quando più agenti potrebbero voler ricevere informazioni su un evento (ad esempio "user-question-received").
Code per distribuire il lavoro
Usi una coda (Redis BLPOP, RabbitMQ) quando ogni attività deve essere gestita da un solo worker:
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 per la velocità
NATS è un message broker leggero progettato per i microservizi. Offre una latenza inferiore al millisecondo e il pattern request/reply integrato:
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 per la persistenza
Kafka aggiunge log persistenti e riproducibili. È perfetto per l'audit e la riproduzione:
from confluent_kafka import Producer, Consumer
producer.produce('agent-events', key=task_id, value=json.dumps(task))
producer.flush()Schemi degli eventi
Definisca la struttura di ogni messaggio con Pydantic o Protobuf. Senza schemi, il bus diventa ingestibile:
class ResearchTask(BaseModel):
task_id: str
user_id: str
query: str
deadline: datetime
class ResearchResult(BaseModel):
task_id: str
findings: list[str]
duration_ms: intID di correlazione
Ogni messaggio correlato include lo stesso correlation_id, così può tracciare un'attività attraverso tutti i servizi:
record = {'correlation_id': 'abc', 'task_id': 'def', 'span': 'llm_call'}
print(record)
Handler idempotenti
I messaggi possono essere consegnati più di una volta (semantica at-least-once). Renda gli handler idempotenti: lo stesso task_id deve essere elaborato una sola volta.
Coda dei messaggi non recapitabili
Quando un handler fallisce ripetutamente, sposti il messaggio in una DLQ per la revisione umana invece di riprovare all'infinito:
if attempts > MAX_RETRIES:
r.rpush('queue:research:dlq', raw)
log.error('Sent to DLQ', extra={'task_id': task_id})Osservabilità
Tracci ogni messaggio:
- Gli ID di traccia attraversano il bus
- I log includono task_id e correlation_id
- Metriche: throughput, latenza e tasso di errore per topic
Quando adottarlo
Inizi in modo semplice, con agenti in-process. Passi a un bus solo quando dispone di:
- Almeno 5 agenti in produzione
- La necessità di scalare gli agenti in modo indipendente
- Flussi di lavoro che coinvolgono più servizi
Almeno una volta?
Che cosa richiede la "consegna almeno una volta" ai suoi handler?
Riepilogo
Disaccoppi con Redis nei casi semplici, NATS quando conta la latenza e Kafka quando serve la persistenza. Schemi, ID di correlazione, handler idempotenti e DLQ.
Domande Frequenti
La lezione «Protocolli di comunicazione (message bus)» è gratuita?
Sì — il testo completo di «Protocolli di comunicazione (message bus)» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Protocolli di comunicazione (message bus)»?
Disaccoppi gli agenti con un message bus, come Redis, NATS o Kafka, così potranno scalare e subire guasti in modo indipendente. Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.
Quanto tempo richiede la lezione «Protocolli di comunicazione (message bus)»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Multi-agente basato sulla conversazione (AutoGen)
- Supervisori gerarchici (orchestratore + worker)
- Ruoli e specializzazioni degli agenti
- Protocolli di comunicazione (message bus)