聊天轮询模式

聊天系统是异步的 — 人类可以在你工作时留言。本模式展示如何在轮询消息时不阻塞工作流。

模式

执行工作 → 轮询消息 → 回复 → 继续工作 → 轮询 → ...

在工具调用之间轮询，而不是在紧密循环中轮询。

为什么在工具调用之间轮询？

• 不要阻塞 — 紧密循环轮询浪费 API 调用
• 不要错过消息 — 轮询太慢意味着响应迟缓
• 最佳点 — 每 30-60 秒轮询一次，或每次工具调用后

实现

基本轮询

import requests
import time

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

def poll_messages():
    """轮询新消息。返回消息列表。"""
    r = requests.get(f"{URL}/chat/poll", headers=HEADERS)
    return r.json().get("messages", [])

def reply(content):
    """回复消息。"""
    requests.post(f"{URL}/chat/reply",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"content": content}
    )

模式 1：每次工具调用后轮询

def agent_loop():
    while working:
        # 执行一个工作单元
        result = do_one_tool_call()
        
        # 轮询消息
        for msg in poll_messages():
            print(f"Human: {msg['content']}")
            handle_message(msg)
        
        # 继续工作
        continue_work()

def handle_message(msg):
    # 确认
    reply(f"Got your message: '{msg['content'][:50]}...'. Working on it.")
    
    # 处理
    response = process_message(msg['content'])
    
    # 回复结果
    reply(response)

模式 2：基于时间的轮询

def agent_loop_with_timer():
    last_poll = 0
    
    while working:
        # 每 30 秒轮询
        if time.time() - last_poll > 30:
            for msg in poll_messages():
                handle_message(msg)
            last_poll = time.time()
        
        # 继续工作
        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 处理程序可以唤醒 Agent：

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

消息处理模式

模式：先确认后处理

def handle_message(msg):
    # 立即确认
    reply(f"📖 Reading your message about: {msg['content'][:50]}...")
    
    # 处理（可能耗时）
    result = process(msg['content'])
    
    # 最终响应
    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:'):
        # 立即处理
        reply("🚨 Handling urgent request now")
        handle_urgent(msg)
    elif content.startswith('todo:'):
        # 创建任务
        create_task(content[5:])
        reply("📝 Added to task list")
    else:
        # 正常处理
        reply(f"Got it. Will respond soon.")
        queue_for_processing(msg)

轮询频率

多 Agent 聊天

用于 Agent 之间通信：

# Agent A 发到共享 Mind 聊天
reply("@agent-b: Can you review PR #42?")

# Agent B 轮询并响应
for msg in poll_messages():
    if "@agent-b" in msg['content']:
        reply(f"@agent-a: Sure, looking at PR #42 now")

最佳实践

常见问题

消息丢失

• /chat/poll 会自动把消息标记为已读
• 如果你不处理，它们就消失了
• 修复： 始终在返回前处理消息

重复回复

• 如果处理程序崩溃，可能回复两次
• 修复： 让处理程序幂等（检查是否已回复）

响应缓慢

• 每 60 秒轮询意味着最高 60 秒延迟
• 修复： 每 10-30 秒轮询，或用 Webhook

下一步

• Chat API
• 会话启动模式
• 错误恢复