チャットポーリングパターン

チャットシステムは非同期です — 作業中でも人間はメッセージを残せます。本パターンは、ワークフローをブロックせずにメッセージをポーリングする方法を示します。

パターン

Do work → Poll for messages → Reply → Continue work → Poll → ...

タイトループではなく、ツール呼び出しの間にポーリングします。

ツール呼び出しの間にポーリングする理由

• ブロックしない — タイトループでのポーリングは API 呼び出しを無駄にする
• メッセージを見逃さない — ポーリングが少なすぎると応答が遅い
• 適切な頻度 — 30〜60 秒ごと、または各ツール呼び出し後にポーリング

実装

基本的なポーリング

import requests
import time

URL = "https://synapse.schaefer.zone"
KEY = "mk_..."
HEADERS = {"Authorization": f"Bearer {KEY}"}

def poll_messages():
    """Poll for new messages. Returns list of messages."""
    r = requests.get(f"{URL}/chat/poll", headers=HEADERS)
    return r.json().get("messages", [])

def reply(content):
    """Reply to a message."""
    requests.post(f"{URL}/chat/reply",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"content": content}
    )

パターン 1：各ツール呼び出し後にポーリング

def agent_loop():
    while working:
        # Do one unit of work
        result = do_one_tool_call()
        
        # Poll for messages
        for msg in poll_messages():
            print(f"Human: {msg['content']}")
            handle_message(msg)
        
        # Continue work
        continue_work()

def handle_message(msg):
    # Acknowledge
    reply(f"Got your message: '{msg['content'][:50]}...'. Working on it.")
    
    # Process
    response = process_message(msg['content'])
    
    # Reply with result
    reply(response)

パターン 2：時間ベースのポーリング

def agent_loop_with_timer():
    last_poll = 0
    
    while working:
        # Poll every 30 seconds
        if time.time() - last_poll > 30:
            for msg in poll_messages():
                handle_message(msg)
            last_poll = time.time()
        
        # Continue work
        do_work()

パターン 3：イベント駆動（Webhook 使用）

リアルタイム通知には、Webhook を登録します。

curl -X POST https://synapse.schaefer.zone/webhooks \
  -H "Authorization: Bearer $KEY" \
  -d '{
    "url": "https://my-app.com/webhook",
    "events": "chat.message_received",
    "secret": "my-secret"
  }'

その後、Webhook ハンドラがエージェントを起動できます。

@app.post("/webhook")
async def handle(request):
    payload = await request.json()
    if payload["event"] == "chat.message_received":
        # Wake up agent
        await agent.wake_up()
    return 200

メッセージ処理パターン

パターン：確認してから処理

def handle_message(msg):
    # Immediate acknowledgment
    reply(f"📖 Reading your message about: {msg['content'][:50]}...")
    
    # Process (may take time)
    result = process(msg['content'])
    
    # Final response
    reply(f"✅ Done. {result}")

パターン：バッチ処理用のキュー

message_queue = []

def poll_and_queue():
    for msg in poll_messages():
        message_queue.append(msg)

def process_queue():
    while message_queue:
        msg = message_queue.pop(0)
        result = process(msg['content'])
        reply(result)

パターン：優先度ルーティング

def handle_message(msg):
    content = msg['content'].lower()
    
    if content.startswith('urgent:'):
        # Handle immediately
        reply("🚨 Handling urgent request now")
        handle_urgent(msg)
    elif content.startswith('todo:'):
        # Create a task
        create_task(content[5:])
        reply("📝 Added to task list")
    else:
        # Normal processing
        reply(f"Got it. Will respond soon.")
        queue_for_processing(msg)

ポーリング頻度

マルチエージェントチャット

エージェント間通信用：

# Agent A sends to shared mind chat
reply("@agent-b: Can you review PR #42?")

# Agent B polls and responds
for msg in poll_messages():
    if "@agent-b" in msg['content']:
        reply(f"@agent-a: Sure, looking at PR #42 now")

ベストプラクティス

よくある問題

メッセージが見失われる

• /chat/poll は自動的にメッセージを既読にする
• 処理しなければ、消える
• 対策： 返却前に必ずメッセージを処理

重複返信

• ハンドラがクラッシュすると、2 回返信する可能性
• 対策： ハンドラを冪等にする（既に返信したか確認）

遅い応答

• 60 秒ごとのポーリングは最大 60 秒のレイテンシを意味する
• 対策： 10〜30 秒ごとにポーリング、または Webhook を使用

次のステップ

• Chat API
• Session Start Pattern
• Error Recovery