Swarm 工作流 API

概述

Swarm 工作流通过在任务上下文中设置 force_swarm: true 来触发。它使用与所有其他工作流相同的 POST /api/v1/tasks 端点——无需单独的端点。 Swarm 模式将查询分解为子任务，生成持久化智能体并行工作，支持智能体间消息传递，并将结果综合为统一响应。

提交 Swarm 任务

端点

POST http://localhost:8080/api/v1/tasks

请求体

{
  "query": "Your complex multi-faceted task",
  "session_id": "optional-session-id",
  "context": {
    "force_swarm": true
  }
}

Swarm 专用上下文参数

参数	类型	默认值	描述
`force_swarm`	boolean	`false`	必需，触发 Swarm 工作流
`model_tier`	string	（自动）	智能体执行的模型级别：`small`、`medium`、`large`

所有标准任务参数（session_id、mode、model_tier、model_override、provider_override）均适用于 Swarm 任务。

force_swarm 标志必须设置在 context 对象内，而非顶级参数。同时服务端配置中必须启用 Swarm（features.yaml 中 workflows.swarm.enabled: true）。

示例：基本 Swarm 任务

curl -X POST http://localhost:8080/api/v1/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "query": "Compare AI chip markets across US, Japan, and South Korea",
    "session_id": "swarm-demo",
    "context": {
      "force_swarm": true
    }
  }'

from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")
handle = client.submit_task(
    "Compare AI chip markets across US, Japan, and South Korea",
    force_swarm=True,
    session_id="swarm-demo",
)

import httpx

response = httpx.post(
    "http://localhost:8080/api/v1/tasks",
    json={
        "query": "Compare AI chip markets across US, Japan, and South Korea",
        "session_id": "swarm-demo",
        "context": {"force_swarm": True},
    },
)
task = response.json()
print(f"Task ID: {task['task_id']}")

响应

{
  "task_id": "task-abc123...",
  "status": "STATUS_CODE_OK",
  "message": "Task submitted successfully. Session: swarm-demo",
  "created_at": "2025-11-10T10:00:00Z"
}

响应头：

X-Workflow-ID：Temporal 工作流标识符（与 task_id 相同）
X-Session-ID：会话标识符

提交 + 流式传输

使用合并端点一步提交并获取流式传输 URL：

POST http://localhost:8080/api/v1/tasks/stream

curl -s -X POST http://localhost:8080/api/v1/tasks/stream \
  -H "Content-Type: application/json" \
  -d '{
    "query": "Analyze competitive landscape of major cloud AI platforms",
    "context": { "force_swarm": true }
  }' | jq

from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")
handle, stream_url = client.submit_and_stream(
    "Analyze competitive landscape of major cloud AI platforms",
    force_swarm=True,
)
print(f"Stream URL: {stream_url}")

响应（201 Created）：

{
  "workflow_id": "task-def456...",
  "task_id": "task-def456...",
  "stream_url": "/api/v1/stream/sse?workflow_id=task-def456..."
}

监控 Swarm 进度

SSE 事件流

GET http://localhost:8080/api/v1/stream/sse?workflow_id={workflow_id}

Swarm 专用事件

事件类型	`agent_id`	描述
`WORKFLOW_STARTED`	`swarm-supervisor`	Swarm 工作流已初始化
`PROGRESS`（规划）	`swarm-supervisor`	任务分解进行中
`LEAD_DECISION`	`swarm-lead`	Lead 做出协调决策（生成、分配、修订等）
`TASKLIST_UPDATED`	`swarm-lead`	任务依赖图已变更
`TEAM_STATUS`	`swarm-lead`	团队组成变更（智能体生成或关闭）
`AGENT_STARTED`	智能体名称（如 `takao`）	单个智能体开始执行
`PROGRESS`（迭代）	智能体名称	智能体迭代进度
`AGENT_COMPLETED`	智能体名称	单个智能体已完成
`PROGRESS`（综合）	`swarm-supervisor`	正在合并所有智能体的结果
`WORKFLOW_COMPLETED`	`swarm-supervisor`	最终综合完成

SSE 输出示例

data: {"type":"WORKFLOW_STARTED","agent_id":"swarm-supervisor","message":"Assigning a team of agents","timestamp":"..."}

data: {"type":"PROGRESS","agent_id":"swarm-supervisor","message":"Planning approach","timestamp":"..."}

data: {"type":"PROGRESS","agent_id":"swarm-supervisor","message":"Assigning 3 agents","timestamp":"..."}

data: {"type":"AGENT_STARTED","agent_id":"takao","message":"Agent takao started","timestamp":"..."}

data: {"type":"PROGRESS","agent_id":"takao","message":"Agent takao progress: iteration 1/25, action: tool_call","timestamp":"..."}

data: {"type":"AGENT_COMPLETED","agent_id":"takao","message":"Agent takao completed","timestamp":"..."}

data: {"type":"PROGRESS","agent_id":"swarm-supervisor","message":"Combining findings from 3 agents","timestamp":"..."}

data: {"type":"WORKFLOW_COMPLETED","agent_id":"swarm-supervisor","message":"All done","timestamp":"..."}

任务状态响应

GET http://localhost:8080/api/v1/tasks/{task_id}

Swarm 元数据

当 Swarm 工作流完成时，状态响应包含 Swarm 专用元数据：

{
  "task_id": "task-abc123...",
  "workflow_id": "task-abc123...",
  "status": "TASK_STATUS_COMPLETED",
  "result": "## AI Chip Market Comparison\n\n...",
  "query": "Compare AI chip markets across US, Japan, and South Korea",
  "session_id": "swarm-demo",
  "mode": "standard",
  "model_used": "claude-haiku-4-5-20251001",
  "provider": "anthropic",
  "metadata": {
    "workflow_type": "swarm",
    "total_agents": 3,
    "total_tokens": 598235,
    "model_breakdown": [
      {
        "model": "claude-haiku-4-5-20251001",
        "provider": "anthropic",
        "executions": 38,
        "tokens": 297524,
        "cost_usd": 0.372
      },
      {
        "model": "shannon_web_search",
        "provider": "shannon-scraper",
        "executions": 12,
        "tokens": 90000,
        "cost_usd": 0.048
      }
    ]
  },
  "usage": {
    "input_tokens": 289164,
    "output_tokens": 309071,
    "total_tokens": 598235,
    "estimated_cost": 0.780
  }
}

元数据字段

字段	类型	描述
`metadata.workflow_type`	string	Swarm 工作流始终为 `"swarm"`
`metadata.total_agents`	integer	参与的智能体总数（初始 + 动态）
`metadata.total_tokens`	integer	所有智能体消耗的总 Token 数
`metadata.model_breakdown`	array	每个模型的执行摘要
`metadata.model_breakdown[].model`	string	模型标识符
`metadata.model_breakdown[].provider`	string	提供商名称
`metadata.model_breakdown[].executions`	integer	使用该模型的 LLM 调用次数
`metadata.model_breakdown[].tokens`	integer	该模型消耗的 Token 数
`metadata.model_breakdown[].cost_usd`	number	预估成本（美元）

服务端配置

Swarm 参数在 config/features.yaml 的 workflows.swarm 下配置：

workflows:
  swarm:
    enabled: true                     # 启用/禁用 Swarm 路由
    max_agents: 10                    # 智能体总数上限（初始 + 动态）
    max_iterations_per_agent: 25      # 每个智能体的最大推理-行动循环次数
    agent_timeout_seconds: 1800       # 每个智能体超时（30 分钟）
    max_messages_per_agent: 20        # 每个智能体的 P2P 消息上限
    workspace_snippet_chars: 800      # 提示中每个工作区条目的最大字符数
    workspace_max_entries: 5          # 每个主题显示的最近条目数
    max_total_llm_calls: 200          # 全局 LLM 调用预算
    max_total_tokens: 1000000         # 全局 Token 预算（1M）
    max_wall_clock_minutes: 30        # 最大挂钟时间

参数	默认值	描述
`enabled`	`true`	必须为 `true` 才能使 `force_swarm` 生效
`max_agents`	`10`	总上限（含动态生成的辅助智能体）
`max_iterations_per_agent`	`25`	每个智能体的迭代限制
`agent_timeout_seconds`	`1800`	每个智能体 30 分钟超时
`max_messages_per_agent`	`20`	防止 P2P 消息泛滥
`workspace_snippet_chars`	`800`	控制工作区上下文的 Token 消耗
`workspace_max_entries`	`5`	限制智能体提示中每个主题的工作区条目
`max_total_llm_calls`	`200`	所有智能体的最大 LLM 调用总数
`max_total_tokens`	`1000000`	所有智能体的最大 Token 消耗总数
`max_wall_clock_minutes`	`30`	整个 Swarm 的最大挂钟时间

错误处理和回退

部分失败

如果部分智能体失败但至少一个成功，Swarm 工作流仍会使用成功智能体的输出生成结果。如果所有智能体都失败，响应将包含错误信息：

{
  "task_id": "task-xyz...",
  "status": "TASK_STATUS_COMPLETED",
  "result": "",
  "error": "All 3 agents failed — no results to synthesize",
  "metadata": {
    "workflow_type": "swarm",
    "total_agents": 3,
    "total_tokens": 45200,
    "cost_usd": 0.058
  }
}

自动回退

如果整个 Swarm 工作流失败（分解错误、所有智能体失败等），Shannon 会自动回退到标准工作流路由（DAG 或 Supervisor）。force_swarm 标志会从上下文中移除以防止递归失败。

提交任务

POST /api/v1/tasks（完整参考）

获取状态

GET /api/v1/tasks/

流式事件

SSE 事件流

取消任务

POST /api/v1/tasks//cancel

入门指南

任务

流式

智能体

会话

Swarm 集群

技能

定时任务

人机协同审查

Channels

Tools

Daemon WebSocket

参考

Swarm 工作流 API

概述

提交 Swarm 任务

端点

请求体

Swarm 专用上下文参数

示例：基本 Swarm 任务

响应

提交 + 流式传输

监控 Swarm 进度

SSE 事件流

Swarm 专用事件

SSE 输出示例

任务状态响应

Swarm 元数据

元数据字段

服务端配置

错误处理和回退

部分失败

自动回退

相关端点

提交任务

获取状态

流式事件

取消任务

​概述

​提交 Swarm 任务

​端点

​请求体

​Swarm 专用上下文参数

​示例：基本 Swarm 任务

​响应

​提交 + 流式传输

​监控 Swarm 进度

​SSE 事件流

​Swarm 专用事件

​SSE 输出示例

​任务状态响应

​Swarm 元数据

​元数据字段

​服务端配置

​错误处理和回退

​部分失败

​自动回退

​相关端点

提交任务

获取状态

流式事件

取消任务

概述

提交 Swarm 任务

端点

请求体

Swarm 专用上下文参数

示例：基本 Swarm 任务

响应

提交 + 流式传输

监控 Swarm 进度

SSE 事件流

Swarm 专用事件

SSE 输出示例

任务状态响应

Swarm 元数据

元数据字段

服务端配置

错误处理和回退

部分失败

自动回退

相关端点