통신 프로토콜(메시지 버스)

프로세스 내부 대화를 넘어서

분리해야 하는 이유

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']))

발행/구독을 통한 브로드캐스팅

작업 분배를 위한 큐

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

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

from confluent_kafka import Producer, Consumer

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

이벤트 스키마

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

상관관계 식별자

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

멱등성 핸들러

배달 불능 큐

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

관찰 가능성

도입 시점

최소 한 번 전달이란

복습