Python SDK 快速开始

安装

pip install shannon-sdk

第一个任务

from shannon import ShannonClient
client = ShannonClient(base_url="http://localhost:8080")

handle = client.submit_task("以单个词返回法国的首都。")
final = client.wait(handle.task_id, timeout=60)
print(final.result)

流式传输 (SSE)

from shannon import ShannonClient, EventType
client = ShannonClient()

h = client.submit_task(
  "起草一份约 120 字的关于近期三个量子计算里程碑的执行摘要。以 Markdown 段落格式输出。"
)
for e in client.stream(h.workflow_id, types=[EventType.LLM_OUTPUT, EventType.WORKFLOW_COMPLETED]):
  print(e.type, e.message)
  if e.type == EventType.WORKFLOW_COMPLETED:
    break

模型/模式选择（可选）

在提交时控制成本和提供商选择：

handle = client.submit_task(
  "将段落总结为三个重点关注收入趋势的要点。输出：Markdown 列表。",
  model_tier="small",                # small|medium|large
  # model_override="gpt-5-nano-2025-08-07",
  # provider_override="openai",      # 或 anthropic、google 等
  mode="simple",                      # simple|standard|complex|supervisor
)
print(client.wait(handle.task_id).result)

集群模式

使用 force_swarm 强制多智能体集群执行：

handle = client.submit_task(
  "调研并对比排名前三的云服务商在 AI 工作负载上的表现。",
  force_swarm=True,
)
result = client.wait(handle.task_id)
print(result.result)

异步版本

import asyncio
from shannon import AsyncShannonClient

async def main():
  async with AsyncShannonClient(base_url="http://localhost:8080") as client:
    h = await client.submit_task(
      "对"我喜欢这个产品！"进行情感分类（positive|neutral|negative）。返回 JSON {label, score}。"
    )
    f = await client.wait(h.task_id)
    print(f.result)

asyncio.run(main())

任务控制（暂停/恢复）

控制长时间运行的任务，支持暂停、恢复和状态检查：

from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")

# 提交一个长时间运行的研究任务
handle = client.submit_task("研究 2025 年 AI 趋势并提供详细分析")

# 在下一个检查点暂停任务
client.pause_task(handle.task_id, reason="查看中间结果")

# 检查控制状态
state = client.get_control_state(handle.task_id)
print(f"是否暂停: {state.is_paused}")
print(f"暂停时间: {state.paused_at}")
print(f"暂停原因: {state.pause_reason}")

# 恢复执行
client.resume_task(handle.task_id, reason="审查后继续")

# 等待完成
result = client.wait(handle.task_id)
print(result.result)

CLI 用法：

# 暂停正在运行的任务
shannon pause task-123 --reason "暂停以供审查"

# 检查控制状态
shannon control-state task-123

# 恢复任务
shannon resume task-123 --reason "继续执行"

人机协同审查

与需要人工审查才能继续的工作流进行交互：

from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")

# 获取工作流的当前审查状态
state = client.get_review_state("workflow-123")
print(f"状态: {state.status}")        # "reviewing" 或 "approved"
print(f"轮次: {state.round}")
print(f"当前计划: {state.current_plan}")

# 查看对话轮次
for r in state.rounds:
    print(f"  [{r.role}] {r.message}")

# 提交反馈以完善计划
updated = client.submit_review_feedback(
    "workflow-123",
    "请添加网络异常的错误处理。",
    version=state.version,  # 乐观并发控制
)
print(f"更新后的轮次: {updated.round}")

# 批准计划，让工作流继续执行
result = client.approve_review("workflow-123", version=updated.version)
print(f"审查状态: {result['status']}")

CLI 用法：

# 获取审查状态
shannon review-get workflow-123

# 提交反馈
shannon review-feedback workflow-123 "添加错误处理"

# 批准
shannon review-approve workflow-123

技能

列出并查看可用的技能：

from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")

# 列出所有技能
skills = client.list_skills()
for s in skills:
    print(f"{s.name} v{s.version} [{s.category}] - {s.description}")

# 按类别过滤
coding_skills = client.list_skills(category="coding")

# 获取特定技能的详细信息
detail = client.get_skill("web_search")
print(f"作者: {detail.author}")
print(f"所需工具: {detail.requires_tools}")
print(f"内容: {detail.content}")

# 获取技能的所有版本
versions = client.get_skill_versions("web_search")
for v in versions:
    print(f"  {v.name} v{v.version}")

CLI 用法：

# 列出技能
shannon skills-list

# 获取技能详情
shannon skill-get web_search

# 获取技能版本
shannon skill-versions web_search

工具 API

列出、查看和直接执行工具：

from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")

# 列出可用工具
tools = client.list_tools()
for tool in tools:
    print(f"{tool.name}: {tool.description}")

# 获取工具详情
detail = client.get_tool("web_search")
print(detail.parameters)

# 直接执行工具
result = client.execute_tool("web_search", arguments={"query": "Shannon AI"})
print(result.output)

CLI 用法：

# 列出工具
shannon tools-list

# 获取工具详情
shannon tool-get web_search

# 执行工具
shannon tool-exec web_search --arguments '{"query": "Shannon AI"}'

智能体 API

列出、查看和执行确定性智能体：

from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")

# 列出确定性智能体
agents = client.list_agents()

# 获取智能体详情
agent = client.get_agent(agents[0].id)
print(f"{agent.name}: {agent.description}")

# 执行智能体
execution = client.execute_agent(agent.id, {"query": "test input"})
status = execution.wait(timeout=60)
print(status.result)

# 向正在运行的集群发送后续消息
result = client.send_swarm_message(workflow_id, "Please focus on cost analysis")

CLI 用法：

# 列出智能体
shannon agents-list

# 获取智能体详情
shannon agent-get AGENT_ID

# 执行智能体
shannon agent-exec AGENT_ID --input '{"query": "test input"}'

# 向集群发送后续消息
shannon swarm-message WORKFLOW_ID "Please focus on cost analysis"

OpenAI 兼容接口

使用 OpenAI 兼容的聊天补全端点：

from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")

# 列出可用模型
models = client.list_openai_models()

# 聊天补全（非流式）
completion = client.create_chat_completion(
    messages=[{"role": "user", "content": "Hello!"}],
    model="gpt-4o-mini",
)
print(completion.choices[0].message.content)

# 流式聊天补全
for chunk in client.stream_chat_completion(
    messages=[{"role": "user", "content": "Tell me a story"}]
):
    if chunk.choices and chunk.choices[0].delta:
        print(chunk.choices[0].delta.content, end="")

文件访问

访问会话工作区文件和用户记忆文件：

from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")

# 列出会话的工作区文件
files = client.list_session_files(session_id)
for f in files:
    print(f"{f.name} ({f.size_bytes} bytes)")

# 下载文件
content = client.download_session_file(session_id, "report.md")
print(content.content)

# 列出和下载记忆文件
memory_files = client.list_memory_files()
memory = client.download_memory_file("profile.md")

CLI 用法：

# 列出会话工作区文件
shannon session-files SESSION_ID

# 下载工作区文件
shannon session-file-get SESSION_ID report.md

# 列出记忆文件
shannon memory-files

# 下载记忆文件
shannon memory-file-get profile.md

CLI

# 使用模型选择提交
python -m shannon.cli --base-url http://localhost:8080 \
  submit "总结" --model-tier small --mode simple --wait

# 使用集群模式提交
python -m shannon.cli --base-url http://localhost:8080 \
  submit "调研云服务商" --swarm --wait

Python SDK

​安装

​第一个任务

​流式传输 (SSE)

​模型/模式选择（可选）

​集群模式

​异步版本

​任务控制（暂停/恢复）

​人机协同审查

​技能

​工具 API

​智能体 API

​OpenAI 兼容接口

​文件访问

​CLI

安装

第一个任务

流式传输 (SSE)

模型/模式选择（可选）

集群模式

异步版本

任务控制（暂停/恢复）

人机协同审查

技能

工具 API

智能体 API

OpenAI 兼容接口

文件访问

CLI