通信プロトコル（メッセージバス）
メッセージバス（Redis、NATS、Kafka）でエージェントを疎結合にし、それぞれが独立してスケールおよび障害対応できるようにします。
「通信プロトコル（メッセージバス）」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
このレッスンの一部はまだ翻訳されておらず、英語で表示されています。
プロセス内の会話を超えて
小規模なシステムでは、エージェント同士がプロセス内で呼び出し合います。大規模なシステムでは、Redis、NATS、Kafka などのメッセージバスを介して分離します。
なぜ分離するのか
- エージェントを独立してスケールできます
- 障害をエージェントごとに分離できます
- 非同期ワークフローを実現できます（遅いエージェントをインラインで待つ必要がありません）
- デバッグのためにメッセージを再生できます
- 言語をまたいで利用できます：Python エージェントと Go エージェントの両方が購読できます
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
多くのエージェントがイベント（例：「user-question-received」）を知る必要がある場合に使います。
作業分配に使うキュー
各タスクを必ず1つのワーカーだけが処理する場合は、キュー（Redis BLPOP、RabbitMQ）を使います。
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
NATS はマイクロサービス向けに設計された軽量なメッセージブローカーです。サブミリ秒のレイテンシーと、組み込みの 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
Kafka は永続的で再生可能なログを追加します。監査と再生に最適です。
from confluent_kafka import Producer, Consumer
producer.produce('agent-events', key=task_id, value=json.dumps(task))
producer.flush()イベントスキーマ
すべてのメッセージ形式を Pydantic または Protobuf で定義します。スキーマがなければ、バスは混乱した状態になります。
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相関 ID
関連するすべてのメッセージに同じ correlation_id を持たせることで、サービスをまたいでタスクを追跡できます。
record = {'correlation_id': 'abc', 'task_id': 'def', 'span': 'llm_call'}
print(record)
冪等なハンドラー
メッセージは複数回配信されることがあります（at-least-once セマンティクス）。ハンドラーを冪等にし、同じ task_id のタスクが複数回実行されても結果が変わらないようにします。
デッドレターキュー
ハンドラーが繰り返し失敗した場合は、無限にループさせるのではなく、メッセージを DLQ に移して人が確認できるようにします。
if attempts > MAX_RETRIES:
r.rpush('queue:research:dlq', raw)
log.error('Sent to DLQ', extra={'task_id': task_id})可観測性
すべてのメッセージを追跡します。
- トレース ID をバス全体に引き継ぎます
- ログに task_id と correlation_id を含めます
- メトリクスとして、トピックごとのスループット、レイテンシー、エラー率を記録します
導入するタイミング
まずはシンプルに、プロセス内のエージェントから始めます。次の条件が整った場合にのみ、バスへ移行してください。
- 本番環境に5個以上のエージェントがある
- エージェントを独立してスケールする必要がある
- 複数のサービスにまたがるワークフローがある
At-Least-Once とは
「at-least-once 配信」は、ハンドラーに何を求めるでしょうか。
まとめ
単純なケースでは Redis、レイテンシー重視では NATS、永続性重視では Kafka を使って分離します。スキーマ、相関 ID、冪等なハンドラー、DLQ も用意します。
AI チューターと学ぶ AI Agents — 無料
ブラウザでリアルコードを書いて実行し、24/7 の AI チューターから瞬時にサポートを受け、ウェブまたはアプリで続きから学習できます。
- コース
- 60
- レッスン
- 239
よくある質問
「通信プロトコル（メッセージバス）」レッスンは無料ですか？
はい。「通信プロトコル（メッセージバス）」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「通信プロトコル（メッセージバス）」で何を学びますか？
メッセージバス（Redis、NATS、Kafka）でエージェントを疎結合にし、それぞれが独立してスケールおよび障害対応できるようにします。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「通信プロトコル（メッセージバス）」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。