例 - Shannon

この例のコレクションは増加しています。今後も定期的に使用例が追加されます。

基本的な例

モデル選択

from shannon import ShannonClient

client = ShannonClient(base_url="http://localhost:8080")
handle = client.submit_task(
  "収益トレンドに焦点を当てて段落を3つの箇条書きに要約してください。出力形式: Markdownリスト。",
  model_tier="small",
  # model_override="gpt-5-nano-2025-08-07",
  # provider_override="openai",
  mode="simple",
)
print(client.wait(handle.task_id).result)

簡単な質問応答

from shannon import ShannonClient

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

# 簡単な質問をする
handle = client.submit_task(query="フランスの首都はどこですか？")
final = client.wait(handle.task_id)
print(f"回答: {final.result}")

データ分析

from shannon import ShannonClient

client = ShannonClient()

# 現実的なデータ分析
handle = client.submit_task(
    query=(
        "以下の売上データに基づいて計算してください:\n"
        "1. 製品ごとの総収益\n"
        "2. 月ごとの成長率\n"
        "3. 総収益が最も高い上位3つを返す\n\n"
        "データ:\n"
        "ProductA: 1月=$10000, 2月=$12000\n"
        "ProductB: 1月=$8000, 2月=$9500\n"
        "ProductC: 1月=$15000, 2月=$14000\n"
        "ProductD: 1月=$5000, 2月=$6000\n\n"
        "フォーマット: JSON配列 [{product, revenue, mom_growth_pct}]"
    )
)

result = client.wait(handle.task_id)
print(result.result)

高度な例

マルチステップワークフロー

"""Shannon SDKを使用したマルチステップワークフローの例"""
from shannon import ShannonClient

def main():
    client = ShannonClient()
    session_id = "quarterly-analysis-demo"

    print("=" * 60)
    print("マルチステップワークフローの例")
    print("=" * 60)

    # ステップ1: 初期データクエリ
    print("\n[ステップ 1] Q4データを読み込み中...")
    h1 = client.submit_task(
        query="1000 + 500 はいくつですか？これはQ4の収益（1000）と費用（500）を表します。",
        session_id=session_id
    )
    result1 = client.wait(h1.task_id)
    print(f"結果: {result1.result}")

    # ステップ2: 前のコンテキストに基づく分析
    print("\n[ステップ 2] トレンドを分析中...")
    h2 = client.submit_task(
        query="あなたが計算したQ4の数字に基づいて、利益率は何パーセントですか？",
        session_id=session_id
    )
    result2 = client.wait(h2.task_id)
    print(f"結果: {result2.result}")

    # ステップ3: すべての前のコンテキストを使用した要約
    print("\n[ステップ 3] エグゼクティブサマリーを作成中...")
    h3 = client.submit_task(
        query="Q4の財務分析を2文で要約してください。",
        session_id=session_id
    )
    result3 = client.wait(h3.task_id)
    print(f"結果: {result3.result}")

    print("\n" + "=" * 60)
    print("✅ マルチステップワークフローが完了しました！")
    print(f"セッションID: {session_id}")
    print("=" * 60)

if __name__ == "__main__":
    main()

並列処理

"""Shannon SDKを使用した並列処理の例"""
import asyncio
from shannon import AsyncShannonClient

async def main():
    print("=" * 60)
    print("並列処理の例")
    print("=" * 60)

    async with AsyncShannonClient() as client:
        # 並列処理するトピック
        topics = ["AI", "量子コンピューティング", "バイオテクノロジー"]

        print(f"\n[ステップ 1] {len(topics)} タスクを並列で提出中...")

        # すべてのタスクを同時に提出（ノンブロッキング）
        tasks = [
            client.submit_task(
                query=f"{topic}の1文要約とその主な応用を教えてください。"
            )
            for topic in topics
        ]

        # すべての提出が完了するのを待つ
        handles = await asyncio.gather(*tasks)
        print(f"✅ {len(handles)} タスクがすべて提出されました")

        print("\n[ステップ 2] すべての結果を待機中...")

        # すべてのタスクが並列で完了するのを待つ
        results = await asyncio.gather(
            *[client.wait(h.task_id) for h in handles]
        )
        print(f"✅ {len(results)} タスクがすべて完了しました")

        print("\n[ステップ 3] 結果:")
        print("-" * 60)

        # 結果を表示
        for topic, result in zip(topics, results):
            print(f"\n📌 {topic}:")
            print(f"   {result.result}")

            # メタデータが利用可能な場合は安全にアクセス
            metadata = []
            if hasattr(result, 'metadata') and result.metadata:
                model = result.metadata.get('model_used') or result.metadata.get('model')
                if model:
                    metadata.append(f"モデル: {model}")
                tokens = result.metadata.get('total_tokens')
                if tokens:
                    metadata.append(f"トークン: {tokens}")

            if metadata:
                print(f"   ({', '.join(metadata)})")

        print("\n" + "=" * 60)
        print("✅ 並列処理が完了しました！")
        print(f"{len(topics)} トピックを同時に処理しました")
        print("=" * 60)

if __name__ == "__main__":
    asyncio.run(main())

フィルターを使用したストリーミング

"""フィルター付きストリーミングの例"""
from shannon import ShannonClient, EventType

def main():
    print("=" * 60)
    print("フィルターを使用したストリーミングの例")
    print("=" * 60)

    client = ShannonClient()

    print("\n[ステップ 1] タスクを送信中...")
    handle = client.submit_task(
        query=(
            "AWS、Azure、GCPの価格モデルを3文で比較してください。 "
            "生成される出力をストリームします。"
        )
    )
    print(f"✅ タスクが送信されました: {handle.task_id}")

    print("\n[ステップ 2] フィルターを使用したストリーミング [LLM_OUTPUT, WORKFLOW_COMPLETED]...")
    print("-" * 60)

    event_count = 0
    llm_outputs = []

    # LLM_OUTPUT と WORKFLOW_COMPLETED イベントのみをストリーム
    for event in client.stream(
        handle.workflow_id,
        types=[EventType.LLM_OUTPUT, EventType.WORKFLOW_COMPLETED]
    ):
        event_count += 1
        print(f"[{event_count}] {event.type}")

        if event.type == EventType.LLM_OUTPUT:
            print(f"    コンテンツ: {event.message[:80]}...")
            llm_outputs.append(event.message)

        if event.type == EventType.WORKFLOW_COMPLETED:
            print(f"    ステータス: {event.message}")
            break

    print("\n" + "-" * 60)
    print(f"✅ {event_count} 件のフィルタリングされたイベントを受信")
    print(f"✅ {len(llm_outputs)} 件の LLM 出力をキャプチャ")

    print("\n[ステップ 3] 完全な結果:")
    print("-" * 60)
    final = client.wait(handle.task_id)
    print(final.result)

    print("\n" + "=" * 60)
    print("✅ フィルターを使用したストリーミングが完了しました！")
    print("=" * 60)

if __name__ == "__main__":
    main()

イベントタイプの完全なリストは、イベントタイプカタログを参照してください。

利用可能なイベントタイプ

LLM_OUTPUT — LLM の応答
LLM_PARTIAL — ストリーミングトークン
WORKFLOW_COMPLETED — タスク完了
AGENT_THINKING — エージェントの推論
AGENT_STARTED — エージェントが作業を開始
AGENT_COMPLETED — エージェントが完了
PROGRESS — 進捗更新
ERROR_OCCURRED — エラー

一般的なフィルター

最終出力のみ: types=[EventType.LLM_OUTPUT, EventType.WORKFLOW_COMPLETED]
リアルタイムストリーミング: types=[EventType.LLM_PARTIAL, EventType.WORKFLOW_COMPLETED]
エージェントの監視: types=[EventType.AGENT_STARTED, EventType.AGENT_COMPLETED]
すべて (フィルターなし): types=None (またはパラメータを省略)

タスク制御の例

長時間実行されるタスクの一時停止と再開

"""タスク制御 (一時停止/再開) の例"""
from shannon import ShannonClient
import time

def main():
    print("=" * 60)
    print("タスク制御 (一時停止/再開) の例")
    print("=" * 60)

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

    # 長時間実行される研究タスクを送信
    print("\n[ステップ 1] 長時間実行されるタスクを送信中...")
    handle = client.submit_task(
        query="量子コンピューティングとAI統合の最新の進展を調査し、詳細な分析を提供してください。"
    )
    print(f"✅ タスクが送信されました: {handle.task_id}")

    # 少し実行させる
    print("\n[ステップ 2] タスクを5秒間実行させる...")
    time.sleep(5)

    # タスクを一時停止
    print("\n[ステップ 3] タスクを一時停止中...")
    success = client.pause_task(handle.task_id, reason="中間結果のレビュー")
    if success:
        print("✅ 一時停止信号が送信されました")

    # 制御状態を確認
    print("\n[ステップ 4] 制御状態を確認中...")
    state = client.get_control_state(handle.task_id)
    print(f"一時停止中: {state.is_paused}")
    print(f"キャンセル済み: {state.is_cancelled}")
    if state.paused_at:
        print(f"一時停止時刻: {state.paused_at}")
        print(f"一時停止理由: {state.pause_reason}")
        print(f"一時停止した人: {state.paused_by}")

    # 再開前に少し待つ
    print("\n[ステップ 5] 再開前に3秒待機中...")
    time.sleep(3)

    # タスクを再開
    print("\n[ステップ 6] タスクを再開中...")
    success = client.resume_task(handle.task_id, reason="レビュー後に続行")
    if success:
        print("✅ 再開信号が送信されました")

    # 完了を待つ
    print("\n[ステップ 7] タスクの完了を待機中...")
    result = client.wait(handle.task_id)
    print(f"\n✅ タスクが完了しました！")
    print(f"ステータス: {result.status}")
    print(f"\n結果プレビュー: {result.result[:200]}...")

    print("\n" + "=" * 60)
    print("✅ タスク制御の例が完了しました！")
    print("=" * 60)

if __name__ == "__main__":
    main()

制御信号のエラーハンドリング

"""一時停止/再開操作のエラーハンドリング"""
from shannon import ShannonClient
from shannon.errors import ShannonError

def main():
    client = ShannonClient()

    task_id = "task-123"

    # 一時停止を試みる
    try:
        client.pause_task(task_id, reason="レビューが必要")
        print("✅ タスクが正常に一時停止されました")
    except ShannonError as e:
        # 特定のエラーケースを処理
        if "完了したタスクを一時停止できません" in str(e):
            print("⚠️  タスクはすでに完了しているため、一時停止できません")
        elif "タスクはすでに一時停止しています" in str(e):
            print("⚠️  タスクはすでに一時停止しています")
        elif "タスクが見つかりません" in str(e):
            print("❌ タスクが見つかりません")
        else:
            print(f"❌ エラー: {e}")

    # 再開を試みる
    try:
        client.resume_task(task_id, reason="実行を続ける")
        print("✅ タスクが正常に再開されました")
    except ShannonError as e:
        if "タスクは一時停止されていません" in str(e):
            print("⚠️  タスクは一時停止されていないため、再開できません")
        elif "完了したタスクを再開できません" in str(e):
            print("⚠️  タスクはすでに完了しているため、再開できません")
        else:
            print(f"❌ エラー: {e}")

if __name__ == "__main__":
    main()

スウォームモード

マルチエージェント実行の強制

"""スウォームモードの例（Shannon SDK）"""
from shannon import ShannonClient

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

# スウォームモードで送信し、複数のエージェントを連携させる
handle = client.submit_task(
    query="MLワークロード向けのAWS、Azure、GCPを調査・比較してください。料金、GPU可用性、マネージドサービスをカバーしてください。",
    force_swarm=True,
)

result = client.wait(handle.task_id)
print(result.result)

# モデルと使用量情報を確認
if result.usage:
    print(f"合計トークン: {result.usage.get('total_tokens')}")
    print(f"コスト: ${result.usage.get('cost_usd', 0):.6f}")

ヒューマン・イン・ザ・ループレビューの例

インタラクティブなレビューワークフロー

"""ヒューマン・イン・ザ・ループ レビューの例"""
from shannon import ShannonClient

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

# レビューが必要なワークフローがすでに実行中であると仮定
workflow_id = "workflow-123"

# 現在のレビュー状態を確認
state = client.get_review_state(workflow_id)
print(f"レビューステータス: {state.status}")
print(f"ラウンド: {state.round}, バージョン: {state.version}")

if state.current_plan:
    print(f"現在のプラン:\n{state.current_plan}")

# 会話履歴を表示
for r in state.rounds:
    print(f"  [{r.role}] {r.message}")

# まだレビュー中の場合、フィードバックを提供
if state.status == "reviewing":
    updated = client.submit_review_feedback(
        workflow_id,
        "入力バリデーションとユニットテストを追加してください。",
        version=state.version,
    )
    print(f"フィードバックが送信されました。現在のラウンド: {updated.round}")

    # 更新されたプランを承認
    result = client.approve_review(workflow_id, version=updated.version)
    print(f"承認済み: {result['status']}")

楽観的同時実行制御を使用したレビュー

"""レビュー中のバージョン競合の処理"""
from shannon import ShannonClient, ValidationError

client = ShannonClient()

workflow_id = "workflow-456"

try:
    state = client.get_review_state(workflow_id)
    updated = client.submit_review_feedback(
        workflow_id,
        "全体的に良いですが、些細なスタイルの変更が必要です。",
        version=state.version,
    )
    client.approve_review(workflow_id, version=updated.version)
except ValidationError as e:
    if "409" in str(e.code):
        print("バージョン競合: 別のユーザーがレビューを変更しました。更新して再試行してください。")
    else:
        raise

スキルの例

スキルの閲覧と確認

"""スキル閲覧の例"""
from shannon import ShannonClient

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

# 利用可能なすべてのスキルを一覧表示
skills = client.list_skills()
print(f"{len(skills)} 個のスキルが見つかりました:\n")
for s in skills:
    flag = " [危険]" if s.dangerous else ""
    status = "有効" if s.enabled else "無効"
    print(f"  {s.name} v{s.version} ({s.category}) - {status}{flag}")
    print(f"    {s.description}")

# カテゴリでフィルタリング
print("\n--- コーディングスキル ---")
coding = client.list_skills(category="coding")
for s in coding:
    print(f"  {s.name}: {s.description}")

# 詳細情報を取得
detail = client.get_skill("web_search")
print(f"\nスキル詳細: {detail.name} v{detail.version}")
print(f"  著者: {detail.author}")
print(f"  必要なツール: {detail.requires_tools}")
print(f"  必要なロール: {detail.requires_role}")
print(f"  最大バジェット: {detail.budget_max}")

# バージョン履歴を取得
versions = client.get_skill_versions("web_search")
print(f"\nweb_search のバージョン:")
for v in versions:
    print(f"  v{v.version} - {v.description}")

スケジュール管理の例

スケジュールタスクの作成と管理

"""スケジュール管理の例"""
from shannon import ShannonClient

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

# 平日の毎朝9時（UTC）に実行される調査タスクを作成
result = client.create_schedule(
    name="デイリーAIニュース",
    cron_expression="0 9 * * 1-5",
    task_query="最新のAI研究ニュースを要約してください",
    task_context={"force_research": "true", "research_strategy": "quick"},
    timezone="UTC",
    max_budget_per_run_usd=0.50,
)
schedule_id = result["schedule_id"]
print(f"スケジュールが作成されました: {schedule_id}")

# すべてのスケジュールを一覧表示
schedules, total = client.list_schedules()
print(f"\n{total} 個のスケジュール:")
for s in schedules:
    print(f"  {s.name}: {s.cron_expression} ({s.status})")
    print(f"    実行回数: 合計 {s.total_runs}, 成功 {s.successful_runs}, 失敗 {s.failed_runs}")

# スケジュールの詳細を取得
sched = client.get_schedule(schedule_id)
print(f"\nスケジュール: {sched.name}")
print(f"  次回実行: {sched.next_run_at}")
print(f"  前回実行: {sched.last_run_at}")

# 実行履歴を表示
runs, _ = client.get_schedule_runs(schedule_id)
for run in runs:
    print(f"  {run.triggered_at}: {run.status} - ${run.total_cost_usd:.4f}")

# スケジュールを一時停止（例: メンテナンス中）
client.pause_schedule(schedule_id, reason="メンテナンスウィンドウ")

# スケジュールを再開
client.resume_schedule(schedule_id)

# スケジュール設定を更新
client.update_schedule(
    schedule_id,
    cron_expression="0 8 * * 1-5",  # 8時に変更
    max_budget_per_run_usd=0.75,
)

# 不要になったら削除
# client.delete_schedule(schedule_id)

ツール直接実行

ツールの一覧表示、詳細確認、実行

"""ツール直接実行の例"""
import json
from shannon import ShannonClient

def main():
    client = ShannonClient(base_url="http://localhost:8080")
    try:
        # 利用可能なツールを一覧表示
        tools = client.list_tools()
        print(f"{len(tools)} 個のツールが見つかりました")
        for tool in tools[:10]:
            print(f"  - {tool.name}: {tool.description}")

        if not tools:
            return

        # 特定のツールの詳細情報を取得
        target = tools[0].name
        detail = client.get_tool(target)
        print(f"\nツール: {detail.name}")
        print(f"カテゴリ: {detail.category}")
        print(f"説明: {detail.description}")
        print(f"パラメータ: {json.dumps(detail.parameters, indent=2)}")

        # ツールを実行
        result = client.execute_tool(target, arguments={"expression": "6 * 7"})
        print(f"\n成功: {result.success}")
        if result.text:
            print(f"テキスト: {result.text}")
        if result.output is not None:
            print(f"出力: {result.output}")
        if result.error:
            print(f"エラー: {result.error}")
    finally:
        client.close()

if __name__ == "__main__":
    main()

確定的エージェント実行

エージェントの実行とスウォームフォローアップ

"""確定的エージェント実行の例"""
import json
from shannon import ShannonClient

def main():
    client = ShannonClient(base_url="http://localhost:8080")
    try:
        # 利用可能なエージェントを一覧表示
        agents = client.list_agents()
        print(f"エージェント数: {len(agents)}")
        for agent in agents[:10]:
            print(f"  - {agent.id}: {agent.name} ({agent.tool})")

        if not agents:
            return

        # エージェントの詳細を取得
        agent = client.get_agent(agents[0].id)
        print(f"\nエージェント: {agent.id}")
        print(f"名前: {agent.name}")
        print(f"説明: {agent.description}")
        print(f"入力スキーマ: {json.dumps(agent.input_schema, indent=2)}")

        # エージェントを実行
        execution = client.execute_agent(
            agent.id,
            {"query": "最近のトレンドを分析"},
        )
        print(f"\nタスクID: {execution.task_id}")
        print(f"ワークフローID: {execution.workflow_id}")
        print(f"ステータス: {execution.status}")

        # 結果を待つ
        final = execution.wait(timeout=120)
        print(f"\n結果: {final.result}")

        # オプション: スウォームワークフローにフォローアップメッセージを送信
        # result = client.send_swarm_message(execution.workflow_id, "コスト分析に焦点を当ててください")
        # print(f"スウォームフォローアップ受理: {result.success}")
    finally:
        client.close()

if __name__ == "__main__":
    main()

OpenAI互換チャット補完

非ストリーミングとストリーミング

"""OpenAI互換チャット補完の例"""
from shannon import ShannonClient, OpenAIChatMessage

def main():
    client = ShannonClient(base_url="http://localhost:8080")
    try:
        # 利用可能なモデルを一覧表示
        models = client.list_openai_models()
        print(f"OpenAI互換モデル: {len(models)}")
        for entry in models[:10]:
            print(f"  - {entry.id}")

        messages = [OpenAIChatMessage(role="user", content="Shannonの強みを2文で要約してください。")]

        # 非ストリーミング補完
        completion = client.create_chat_completion(
            messages,
            model="shannon-chat",
        )
        print(f"\nレスポンス: {completion.choices[0].message.content}")
        if completion.usage:
            print(f"トークン数: {completion.usage.total_tokens}")

        # ストリーミング補完
        print("\nストリーミングレスポンス:")
        for chunk in client.stream_chat_completion(
            [OpenAIChatMessage(role="user", content="3文で物語を教えてください。")],
            model="shannon-chat",
            include_usage=True,
        ):
            if chunk.choices and chunk.choices[0].delta and chunk.choices[0].delta.content:
                print(chunk.choices[0].delta.content, end="", flush=True)
        print()
    finally:
        client.close()

if __name__ == "__main__":
    main()

セッションファイルとメモリ

ワークスペースファイルとメモリファイルの一覧表示とダウンロード

"""セッションファイルとメモリの例"""
from shannon import ShannonClient

def main():
    client = ShannonClient(base_url="http://localhost:8080")
    session_id = "my-session"

    try:
        # メモリファイルを一覧表示
        memory_files = client.list_memory_files()
        print(f"メモリファイル: {len(memory_files)}")
        for entry in memory_files[:10]:
            print(f"  - {entry.path} ({entry.size_bytes} bytes)")

        # メモリファイルをダウンロード
        if memory_files:
            downloaded = client.download_memory_file(memory_files[0].path)
            print(f"\nメモリファイルの内容プレビュー:")
            print(downloaded.content[:400])

        # セッションワークスペースファイルを一覧表示
        session_files = client.list_session_files(session_id)
        print(f"\nセッション {session_id} のファイル: {len(session_files)}")
        for entry in session_files[:10]:
            print(f"  - {entry.path} ({entry.size_bytes} bytes)")

        # セッションファイルをダウンロード
        if session_files:
            downloaded = client.download_session_file(session_id, session_files[0].path)
            print(f"\nセッションファイルの内容プレビュー:")
            print(downloaded.content[:400])
    finally:
        client.close()

if __name__ == "__main__":
    main()

エラーハンドリングの例

完全なパターンとフルリトライロジックの例については、専用ページをご覧ください: エラーハンドリング。

Python SDK

例

基本的な例

モデル選択

簡単な質問応答

データ分析

高度な例

マルチステップワークフロー

並列処理

フィルターを使用したストリーミング

タスク制御の例

長時間実行されるタスクの一時停止と再開

制御信号のエラーハンドリング

スウォームモード

マルチエージェント実行の強制

ヒューマン・イン・ザ・ループレビューの例

インタラクティブなレビューワークフロー

楽観的同時実行制御を使用したレビュー

スキルの例

スキルの閲覧と確認

スケジュール管理の例

スケジュールタスクの作成と管理

ツール直接実行

ツールの一覧表示、詳細確認、実行

確定的エージェント実行

エージェントの実行とスウォームフォローアップ

OpenAI互換チャット補完

非ストリーミングとストリーミング

セッションファイルとメモリ

ワークスペースファイルとメモリファイルの一覧表示とダウンロード

エラーハンドリングの例

次のステップ

Async Usage

Error Handling

​基本的な例

​モデル選択

​簡単な質問応答

​データ分析

​高度な例

​マルチステップワークフロー

​並列処理

​フィルターを使用したストリーミング

​タスク制御の例

​長時間実行されるタスクの一時停止と再開

​制御信号のエラーハンドリング

​スウォームモード

​マルチエージェント実行の強制

​ヒューマン・イン・ザ・ループ レビューの例

​インタラクティブなレビューワークフロー

​楽観的同時実行制御を使用したレビュー

​スキルの例

​スキルの閲覧と確認

​スケジュール管理の例

​スケジュールタスクの作成と管理

​ツール直接実行

​ツールの一覧表示、詳細確認、実行

​確定的エージェント実行

​エージェントの実行とスウォームフォローアップ

​OpenAI互換チャット補完

​非ストリーミングとストリーミング

​セッションファイルとメモリ

​ワークスペースファイルとメモリファイルの一覧表示とダウンロード

​エラーハンドリングの例

​次のステップ

Async Usage

Error Handling

基本的な例

モデル選択

簡単な質問応答

データ分析

高度な例

マルチステップワークフロー

並列処理

フィルターを使用したストリーミング

タスク制御の例

長時間実行されるタスクの一時停止と再開

制御信号のエラーハンドリング

スウォームモード

マルチエージェント実行の強制

ヒューマン・イン・ザ・ループレビューの例

インタラクティブなレビューワークフロー

楽観的同時実行制御を使用したレビュー

スキルの例

スキルの閲覧と確認

スケジュール管理の例

スケジュールタスクの作成と管理

ツール直接実行

ツールの一覧表示、詳細確認、実行

確定的エージェント実行

エージェントの実行とスウォームフォローアップ

OpenAI互換チャット補完

非ストリーミングとストリーミング

セッションファイルとメモリ

ワークスペースファイルとメモリファイルの一覧表示とダウンロード

エラーハンドリングの例

次のステップ