Deep research (Qwen-Deep-Research)

import os
import dashscope

# これは、中国北部 2 (北京) リージョンの URL です。各リージョンの URL は異なります。
dashscope.base_http_api_url = 'https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1'

# API キーを構成します
# 設定されていない場合は、次の行を Model Studio API キー (形式: sk-xxx) に置き換えてください
API_KEY = os.getenv('DASHSCOPE_API_KEY')

def call_deep_research_model(messages, step_name):
    print(f"\n=== {step_name} ===")

    try:
        responses = dashscope.Generation.call(
            api_key=API_KEY,
            model="qwen-deep-research",
            messages=messages,
            # qwen-deep-research モデルは現在、ストリーミング出力のみをサポートしています
            stream=True
            # incremental_output=True 増分出力にはこのパラメーターを追加します
        )

        return process_responses(responses, step_name)

    except Exception as e:
        print(f"API 呼び出し中にエラーが発生しました: {e}")
        return ""

# フェーズのコンテンツを表示
def display_phase_content(phase, content, status):
    if content:
        print(f"\n[{phase}] {status}: {content}")
    else:
        print(f"\n[{phase}] {status}")

# レスポンスを処理
def process_responses(responses, step_name):
    current_phase = None
    phase_content = ""
    research_goal = ""
    web_sites = []
    references = []
    keepalive_shown = False  # KeepAlive プロンプトが表示されたかどうかを確認するフラグ

    for response in responses:
        # レスポンスのステータスコードを確認
        if hasattr(response, 'status_code') and response.status_code != 200:
            print(f"HTTP リターンコード: {response.status_code}")
            if hasattr(response, 'code'):
                print(f"エラーコード: {response.code}")
            if hasattr(response, 'message'):
                print(f"エラーメッセージ: {response.message}")
            print("詳細については、https://www.alibabacloud.com/help/model-studio/error-code をご参照ください")
            continue

        if hasattr(response, 'output') and response.output:
            message = response.output.get('message', {})
            phase = message.get('phase')
            content = message.get('content', '')
            status = message.get('status')
            extra = message.get('extra', {})

            # フェーズ変更の検出
            if phase != current_phase:
                if current_phase and phase_content:
                    # フェーズとステップ名に基づいて異なる完了説明を表示
                    if step_name == "ステップ 1: モデルのクエリ確認" and current_phase == "answer":
                        print(f"\n クエリ確認フェーズが完了しました")
                    else:
                        print(f"\n {current_phase} フェーズが完了しました")
                current_phase = phase
                phase_content = ""
                keepalive_shown = False  # KeepAlive プロンプトのフラグをリセット

                # フェーズとステップ名に基づいて異なる説明を表示
                if step_name == "ステップ 1: モデルのクエリ確認" and phase == "answer":
                    print(f"\n クエリ確認フェーズに入ります")
                else:
                    print(f"\n {phase} フェーズに入ります")

            # Answer フェーズでの参照情報の処理
            if phase == "answer":
                if extra.get('deep_research', {}).get('references'):
                    new_references = extra['deep_research']['references']
                    if new_references and new_references != references:  # 重複表示を避ける
                        references = new_references
                        print(f"\n   参考文献 ({len(references)}):")
                        for i, ref in enumerate(references, 1):
                            print(f"     {i}. {ref.get('title', 'タイトルなし')}")
                            if ref.get('url'):
                                print(f"        URL: {ref['url']}")
                            if ref.get('description'):
                                print(f"        説明: {ref['description'][:100]}...")
                            print()

            # WebResearch フェーズでの特別情報の処理
            # 注: qwen-deep-research-2025-12-15 モデルは、streamingQueries と streamingWebResult の代わりに
            # streamingThinking ステータスを使用します
            if phase == "WebResearch":
                if extra.get('deep_research', {}).get('research'):
                    research_info = extra['deep_research']['research']

                    # streamingThinking (スナップショットモデル) または streamingQueries (メインラインモデル) ステータスの処理
                    if status in ("streamingThinking", "streamingQueries"):
                        if 'researchGoal' in research_info:
                            goal = research_info['researchGoal']
                            if goal:
                                research_goal += goal
                                print(f"\n   調査目標: {goal}", end='', flush=True)

                    # streamingWebResult ステータス (メインラインモデル) の処理
                    # スナップショットモデルは、streamingThinking を使用してこのステータスをマージします
                    elif status == "streamingWebResult":
                        if 'webSites' in research_info:
                            sites = research_info['webSites']
                            if sites and sites != web_sites:  # 重複表示を避ける
                                web_sites = sites
                                print(f"\n   {len(sites)} 件の関連 Web サイトが見つかりました:")
                                for i, site in enumerate(sites, 1):
                                    print(f"     {i}. {site.get('title', 'タイトルなし')}")
                                    print(f"        説明: {site.get('description', '説明なし')[:100]}...")
                                    print(f"        URL: {site.get('url', 'リンクなし')}")
                                    if site.get('favicon'):
                                        print(f"        アイコン: {site['favicon']}")
                                    print()

                    # WebResultFinished ステータスの処理
                    elif status == "WebResultFinished":
                        print(f"\n   Web 検索が完了しました。{len(web_sites)} 件の参照ソースが見つかりました。")
                        if research_goal:
                            print(f"   調査目標: {research_goal}")

            # コンテンツを蓄積して表示
            if content:
                phase_content += content
                # コンテンツをリアルタイムで表示
                print(content, end='', flush=True)

            # フェーズのステータス変更を表示
            if status and status != "typing":
                print(f"\n   ステータス: {status}")

                # ステータスの説明を表示
                if status == "streamingThinking":
                    print("   → 調査タスクの分解と Web コンテンツの要約 (WebResearch フェーズ)")
                elif status == "streamingQueries":
                    print("   → 調査目標と検索クエリの生成 (WebResearch フェーズ)")
                elif status == "streamingWebResult":
                    print("   → 検索、Web ページの読み取り、コード実行の実行 (WebResearch フェーズ)")
                elif status == "WebResultFinished":
                    print("   → Web 検索フェーズが完了しました (WebResearch フェーズ)")

            # ステータスが finished の場合、トークン消費量を表示
            if status == "finished":
                if hasattr(response, 'usage') and response.usage:
                    usage = response.usage
                    print(f"\n    トークン消費量の統計:")
                    print(f"      入力トークン: {usage.get('input_tokens', 0)}")
                    print(f"      出力トークン: {usage.get('output_tokens', 0)}")
                    print(f"      リクエスト ID: {response.get('request_id', '不明')}")

            if phase == "KeepAlive":
                # KeepAlive フェーズに初めて入ったときのみプロンプトを表示
                if not keepalive_shown:
                    print("現在のステップが完了しました。次のステップの準備をしています。")
                    keepalive_shown = True
                continue

    if current_phase and phase_content:
        if step_name == "ステップ 1: モデルのクエリ確認" and current_phase == "answer":
            print(f"\n クエリ確認フェーズが完了しました")
        else:
            print(f"\n {current_phase} フェーズが完了しました")

    return phase_content

def main():
    # API キーを確認
    if not API_KEY:
        print("エラー: DASHSCOPE_API_KEY 環境変数が設定されていません")
        print("環境変数を設定するか、コード内の API_KEY 変数を直接変更してください")
        return

    print("ユーザーが対話を開始: 教育における人工知能の応用を調査")

    # ステップ 1: モデルのクエリ確認
    # モデルはユーザーの質問を分析し、調査の方向性を定義するために明確化の質問をします
    messages = [{'role': 'user', 'content': '教育における人工知能の応用を調査'}]
    step1_content = call_deep_research_model(messages, "ステップ 1: モデルのクエリ確認")

    # ステップ 2: 詳細な調査
    # ステップ 1 のクエリ確認に基づいて、モデルは完全な調査プロセスを実行します
    messages = [
        {'role': 'user', 'content': '教育における人工知能の応用を調査'},
        {'role': 'assistant', 'content': step1_content},  # モデルのクエリ確認コンテンツを含む
        {'role': 'user', 'content': '主に個別化学習とインテリジェント評価に焦点を当てています'}
    ]

    call_deep_research_model(messages, "ステップ 2: 詳細な調査")
    print("\n 調査が完了しました！")

if __name__ == "__main__":
    main()

import os
import dashscope

# これは、中国北部 2 (北京) リージョンの URL です。各リージョンの URL は異なります。
dashscope.base_http_api_url = 'https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1'

API_KEY = os.getenv('DASHSCOPE_API_KEY')

messages = [
    {
        "role": "user",
        "content": [
            {"image": "https://example.aliyuncs.com/example.png"},
            {"text": "このチャートのデータトレンドを分析し、主要な発見について詳細な調査を行ってください"}
        ]
    }
]

responses = dashscope.Generation.call(
    api_key=API_KEY,
    model="qwen-deep-research",
    messages=messages,
    stream=True
)

for response in responses:
    if hasattr(response, 'output') and response.output:
        message = response.output.get('message', {})
        content = message.get('content', '')
        if content:
            print(content, end='', flush=True)

import os
import dashscope

# これは、中国北部 2 (北京) リージョンの URL です。各リージョンの URL は異なります。
dashscope.base_http_api_url = 'https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1'

API_KEY = os.getenv('DASHSCOPE_API_KEY')

messages = [
    {
        "role": "user",
        "content": "ナレッジベースを使用して最近公開されたプロダクト更新のお知らせを検索し、調査レポートにまとめてください"
    }
]

responses = dashscope.Generation.call(
    api_key=API_KEY,
    model="qwen-deep-research-2025-12-15",
    messages=messages,
    stream=True,
    enable_feedback=False,
    research_tools=[{
        "type": "mcp",
        "server_label": "my-server",
        "server_url": "https://your-mcp-server.example.com/sse",
        "allowed_tools": ["search", "fetch"],
        "authentication": {
            "bearer": "your_jwt_token_here"
        }
    }]
)

for response in responses:
    if hasattr(response, 'output') and response.output:
        message = response.output.get('message', {})
        content = message.get('content', '')
        if content:
            print(content, end='', flush=True)