深入研究（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 Key
# 若沒有配置環境變數，請用百鍊API Key將下行替換為：API_KEY = "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/zh/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 == "第一步：模型反問確認" 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 == "第一步：模型反問確認" and phase == "answer":
                    print(f"\n 進入模型反問階段")
                else:
                    print(f"\n 進入 {phase} 階段")

            # 處理Answer階段的references資訊
            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模型使用streamingThinking狀態
            # 替代streamingQueries和streamingWebResult
            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)} 個相關網站:")
                                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   網路搜尋完成，共找到 {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("   → 正在拆解研究任務並總結網頁內容（WebResearch階段）")
                elif status == "streamingQueries":
                    print("   → 正在產生研究目標和搜尋查詢（WebResearch階段）")
                elif status == "streamingWebResult":
                    print("   → 正在執行搜尋、網頁閱讀和代碼執行（WebResearch階段）")
                elif status == "WebResultFinished":
                    print("   → 網路搜尋階段完成（WebResearch階段）")

            # 當狀態為finished時，顯示token消耗情況
            if status == "finished":
                if hasattr(response, 'usage') and response.usage:
                    usage = response.usage
                    print(f"\n    Token消耗統計:")
                    print(f"      輸入tokens: {usage.get('input_tokens', 0)}")
                    print(f"      輸出tokens: {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 == "第一步：模型反問確認" and current_phase == "answer":
            print(f"\n 模型反問階段完成")
        else:
            print(f"\n {current_phase} 階段完成")

    return phase_content

def main():
    # 檢查API Key
    if not API_KEY:
        print("錯誤：未設定 DASHSCOPE_API_KEY 環境變數")
        print("請設定環境變數或直接在代碼中修改 API_KEY 變數")
        return

    print("使用者發起對話：研究一下人工智慧在教育中的應用")

    # 第一步：模型反問確認
    # 模型會分析使用者問題，提出細化問題來明確研究方向
    messages = [{'role': 'user', 'content': '研究一下人工智慧在教育中的應用'}]
    step1_content = call_deep_research_model(messages, "第一步：模型反問確認")

    # 第二步：深入研究
    # 基於第一步的反問內容，模型會執行完整的研究流程
    messages = [
        {'role': 'user', 'content': '研究一下人工智慧在教育中的應用'},
        {'role': 'assistant', 'content': step1_content},  # 包含模型的反問內容
        {'role': 'user', 'content': '我主要關注個人化學習和智能評估這兩個方面'}
    ]

    call_deep_research_model(messages, "第二步：深入研究")
    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)