跳转到主要内容

​
概述

推荐使用 SSE 获取简单可靠的实时更新。若环境需要 WS，可使用可选的 WebSocket 辅助方法。

​
SSE（推荐）

from shannon import ShannonClient, EventType
client = ShannonClient()

h = client.submit_task(
  "起草一份约 120 字的关于 Q4 营收驱动与风险的执行摘要。流式传输部分输出；最后一行以‘结论：’结尾。"
)
for e in client.stream(h.workflow_id, types=[EventType.LLM_OUTPUT, EventType.WORKFLOW_COMPLETED]):
  print(f"{e.type}: {e.message}")
  if e.type == EventType.WORKFLOW_COMPLETED:
    break
从上一个事件恢复： 
last_id = None
for e in client.stream(h.workflow_id, last_event_id=last_id):
  # ...
  last_id = e.id
注意：
  • last_event_id 接受 Redis 流 ID（优先）或数字序号。SDK 的 event.id 会在存在时选择 stream_id，否则回退到 seq。

​
WebSocket（可选）

需要 pip install websockets。 
from shannon import ShannonClient
client = ShannonClient()

h = client.submit_task(
  "生成一份约 120 字的关于近期可再生能源政策变化及其对成本影响的简报。流式传输部分输出。"
)
for e in client.stream_ws(h.workflow_id):
  print(e.type, e.message)
  if e.type == "WORKFLOW_COMPLETED":
    break

​
常见事件类型

  • WORKFLOW_STARTED / WORKFLOW_COMPLETED
  • LLM_PROMPT / LLM_PARTIAL / LLM_OUTPUT
  • TOOL_INVOKED / TOOL_OBSERVATION
  • APPROVAL_REQUESTED / APPROVAL_DECISION
  • ERROR_OCCURRED
提示：使用 EventType 枚举或原始字符串进行过滤。