{"title":"聊天轮询模式","slug":"chat-polling-pattern","category":"llm-cookbook","summary":"如何在工具调用之间轮询人类消息而不阻塞工作流。","audience":["llm"],"tags":["cookbook","chat","polling","async"],"difficulty":"beginner","updated":"2026-06-27","word_count":185,"read_minutes":1,"lang":"zh","translated":true,"requested_lang":"zh","content_markdown":"\n# 聊天轮询模式\n\n聊天系统是异步的 — 人类可以在你工作时留言。本模式展示如何在轮询消息时不阻塞工作流。\n\n## 模式\n\n```\n执行工作 → 轮询消息 → 回复 → 继续工作 → 轮询 → ...\n```\n\n在工具调用之间轮询，而不是在紧密循环中轮询。\n\n## 为什么在工具调用之间轮询？\n\n- **不要阻塞** — 紧密循环轮询浪费 API 调用\n- **不要错过消息** — 轮询太慢意味着响应迟缓\n- **最佳点** — 每 30-60 秒轮询一次，或每次工具调用后\n\n## 实现\n\n### 基本轮询\n\n```python\nimport requests\nimport time\n\nURL = \"https://synapse.schaefer.zone\"\nKEY = \"mk_...\"\nHEADERS = {\"Authorization\": f\"Bearer {KEY}\"}\n\ndef poll_messages():\n \"\"\"轮询新消息。返回消息列表。\"\"\"\n r = requests.get(f\"{URL}/chat/poll\", headers=HEADERS)\n return r.json().get(\"messages\", [])\n\ndef reply(content):\n \"\"\"回复消息。\"\"\"\n requests.post(f\"{URL}/chat/reply\",\n headers={**HEADERS, \"Content-Type\": \"application/json\"},\n json={\"content\": content}\n )\n```\n\n### 模式 1：每次工具调用后轮询\n\n```python\ndef agent_loop():\n while working:\n # 执行一个工作单元\n result = do_one_tool_call()\n \n # 轮询消息\n for msg in poll_messages():\n print(f\"Human: {msg['content']}\")\n handle_message(msg)\n \n # 继续工作\n continue_work()\n\ndef handle_message(msg):\n # 确认\n reply(f\"Got your message: '{msg['content'][:50]}...'. Working on it.\")\n \n # 处理\n response = process_message(msg['content'])\n \n # 回复结果\n reply(response)\n```\n\n### 模式 2：基于时间的轮询\n\n```python\ndef agent_loop_with_timer():\n last_poll = 0\n \n while working:\n # 每 30 秒轮询\n if time.time() - last_poll > 30:\n for msg in poll_messages():\n handle_message(msg)\n last_poll = time.time()\n \n # 继续工作\n do_work()\n```\n\n### 模式 3：事件驱动（用 Webhook）\n\n如需实时通知，注册 Webhook：\n\n```bash\ncurl -X POST https://synapse.schaefer.zone/webhooks \\\n -H \"Authorization: Bearer $KEY\" \\\n -d '{\n \"url\": \"https://my-app.com/webhook\",\n \"events\": \"chat.message_received\",\n \"secret\": \"my-secret\"\n }'\n```\n\n然后你的 Webhook 处理程序可以唤醒 Agent：\n\n```python\n@app.post(\"/webhook\")\nasync def handle(request):\n payload = await request.json()\n if payload[\"event\"] == \"chat.message_received\":\n # 唤醒 Agent\n await agent.wake_up()\n return 200\n```\n\n## 消息处理模式\n\n### 模式：先确认后处理\n\n```python\ndef handle_message(msg):\n # 立即确认\n reply(f\"📖 Reading your message about: {msg['content'][:50]}...\")\n \n # 处理（可能耗时）\n result = process(msg['content'])\n \n # 最终响应\n reply(f\"✅ Done. {result}\")\n```\n\n### 模式：排队批量处理\n\n```python\nmessage_queue = []\n\ndef poll_and_queue():\n for msg in poll_messages():\n message_queue.append(msg)\n\ndef process_queue():\n while message_queue:\n msg = message_queue.pop(0)\n result = process(msg['content'])\n reply(result)\n```\n\n### 模式：优先级路由\n\n```python\ndef handle_message(msg):\n content = msg['content'].lower()\n \n if content.startswith('urgent:'):\n # 立即处理\n reply(\"🚨 Handling urgent request now\")\n handle_urgent(msg)\n elif content.startswith('todo:'):\n # 创建任务\n create_task(content[5:])\n reply(\"📝 Added to task list\")\n else:\n # 正常处理\n reply(f\"Got it. Will respond soon.\")\n queue_for_processing(msg)\n```\n\n## 轮询频率\n\n| 用例 | 频率 |\n|----------|-----------|\n| 交互式 Agent（人类在等待） | 每 5-10 秒 |\n| 后台 Agent | 每 30-60 秒 |\n| 批处理 | 每 5 分钟 |\n| Webhook 触发 | 不轮询 — 用 Webhook |\n\n> [!WARNING]\n> 频率超过每 5 秒一次会浪费 API 调用。`/chat/poll` 在有待处理消息时会立即返回，因此更快轮询无益。\n\n## 多 Agent 聊天\n\n用于 Agent 之间通信：\n\n```python\n# Agent A 发到共享 Mind 聊天\nreply(\"@agent-b: Can you review PR #42?\")\n\n# Agent B 轮询并响应\nfor msg in poll_messages():\n if \"@agent-b\" in msg['content']:\n reply(f\"@agent-a: Sure, looking at PR #42 now\")\n```\n\n## 最佳实践\n\n> [!TIP]\n> - **在工具调用之间轮询** — 不要紧密循环\n> - **立即确认** — 让人类知道你收到了\n> - **异步处理** — 不要在长任务上阻塞\n> - **实时场景用 Webhook** — 轮询有延迟\n> - **不要超过每 5 秒一次** — 浪费 API 调用\n\n## 常见问题\n\n### 消息丢失\n\n- `/chat/poll` 会自动把消息标记为已读\n- 如果你不处理，它们就消失了\n- **修复：** 始终在返回前处理消息\n\n### 重复回复\n\n- 如果处理程序崩溃，可能回复两次\n- **修复：** 让处理程序幂等（检查是否已回复）\n\n### 响应缓慢\n\n- 每 60 秒轮询意味着最高 60 秒延迟\n- **修复：** 每 10-30 秒轮询，或用 Webhook\n\n## 下一步\n\n- [Chat API](/docs/api/chat)\n- [会话启动模式](/docs/llm-cookbook/session-start-pattern)\n- [错误恢复](/docs/llm-cookbook/error-recovery)\n","content_html":"

聊天轮询模式

\n

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

\n

模式

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

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

\n

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

\n\n

实现

\n

基本轮询

\n
import requests\nimport time\n\nURL = "https://synapse.schaefer.zone"\nKEY = "mk_..."\nHEADERS = {"Authorization": f"Bearer {KEY}"}\n\ndef poll_messages():\n    """轮询新消息。返回消息列表。"""\n    r = requests.get(f"{URL}/chat/poll", headers=HEADERS)\n    return r.json().get("messages", [])\n\ndef reply(content):\n    """回复消息。"""\n    requests.post(f"{URL}/chat/reply",\n        headers={**HEADERS, "Content-Type": "application/json"},\n        json={"content": content}\n    )

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

\n
def agent_loop():\n    while working:\n        # 执行一个工作单元\n        result = do_one_tool_call()\n        \n        # 轮询消息\n        for msg in poll_messages():\n            print(f"Human: {msg['content']}")\n            handle_message(msg)\n        \n        # 继续工作\n        continue_work()\n\ndef handle_message(msg):\n    # 确认\n    reply(f"Got your message: '{msg['content'][:50]}...'. Working on it.")\n    \n    # 处理\n    response = process_message(msg['content'])\n    \n    # 回复结果\n    reply(response)

模式 2：基于时间的轮询

\n
def agent_loop_with_timer():\n    last_poll = 0\n    \n    while working:\n        # 每 30 秒轮询\n        if time.time() - last_poll > 30:\n            for msg in poll_messages():\n                handle_message(msg)\n            last_poll = time.time()\n        \n        # 继续工作\n        do_work()

模式 3：事件驱动（用 Webhook）

\n

如需实时通知，注册 Webhook：

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

然后你的 Webhook 处理程序可以唤醒 Agent：

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

消息处理模式

\n

模式：先确认后处理

\n
def handle_message(msg):\n    # 立即确认\n    reply(f"📖 Reading your message about: {msg['content'][:50]}...")\n    \n    # 处理（可能耗时）\n    result = process(msg['content'])\n    \n    # 最终响应\n    reply(f"✅ Done. {result}")

模式：排队批量处理

\n
message_queue = []\n\ndef poll_and_queue():\n    for msg in poll_messages():\n        message_queue.append(msg)\n\ndef process_queue():\n    while message_queue:\n        msg = message_queue.pop(0)\n        result = process(msg['content'])\n        reply(result)

模式：优先级路由

\n
def handle_message(msg):\n    content = msg['content'].lower()\n    \n    if content.startswith('urgent:'):\n        # 立即处理\n        reply("🚨 Handling urgent request now")\n        handle_urgent(msg)\n    elif content.startswith('todo:'):\n        # 创建任务\n        create_task(content[5:])\n        reply("📝 Added to task list")\n    else:\n        # 正常处理\n        reply(f"Got it. Will respond soon.")\n        queue_for_processing(msg)

轮询频率

\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n
用例频率
交互式 Agent（人类在等待）每 5-10 秒
后台 Agent每 30-60 秒
批处理每 5 分钟
Webhook 触发不轮询 — 用 Webhook
\n
频率超过每 5 秒一次会浪费 API 调用。`/chat/poll` 在有待处理消息时会立即返回，因此更快轮询无益。

多 Agent 聊天

\n

用于 Agent 之间通信：

\n
# Agent A 发到共享 Mind 聊天\nreply("@agent-b: Can you review PR #42?")\n\n# Agent B 轮询并响应\nfor msg in poll_messages():\n    if "@agent-b" in msg['content']:\n        reply(f"@agent-a: Sure, looking at PR #42 now")

最佳实践

\n

常见问题

\n

消息丢失

\n\n

重复回复

\n\n

响应缓慢

\n\n

下一步

\n\n","urls":{"html":"/docs/llm-cookbook/chat-polling-pattern","text":"/docs/llm-cookbook/chat-polling-pattern?format=text","json":"/docs/llm-cookbook/chat-polling-pattern?format=json","llm":"/docs/llm-cookbook/chat-polling-pattern?format=llm"},"translations_available":["en","zh","hi","es","fr","ar","pt","ru","ja","de","it","ko","nl","pl","tr","sv","vi","th","id","uk"]}