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)

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)

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)

# 暂停正在运行的任务
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']}")

# 获取审查状态
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}")

# 列出技能
shannon skills-list

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

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

# 使用模型选择提交
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