Function example_data_analysis_flow

scripts/main.py:526–586

数据分析完整流程示例

()

def example_data_analysis_flow():
    """数据分析完整流程示例"""
    print("=" * 60)
    print("示例2: 数据分析完整流程")
    print("=" * 60)

    host = os.environ.get("RACCOON_API_HOST", "")
    token = os.environ.get("RACCOON_API_TOKEN", "")
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}",
    }

    print("\n[步骤1] 创建会话...")
    resp = requests.post(
        f"{host}/api/open/office/v2/sessions",
        headers=headers,
        json={"name": "Python示例会话"},
        timeout=10,
    )
    session_data = resp.json().get("data", {})
    session_id = session_data.get("id")
    if not session_id:
        print("创建会话失败")
        return
    print(f"  会话ID: {session_id}")

    print("\n[步骤2] 上传临时文件...")
    batch_id = str(uuid.uuid4())
    csv_content = "name,age,score\nAlice,25,88\nBob,30,92\nCarol,28,85\nDave,35,95\nEve,22,78"
    upload_headers = {k: v for k, v in headers.items() if k != "Content-Type"}
    resp = requests.post(
        f"{host}/api/open/office/v2/sessions/default_session/{batch_id}/files",
        headers=upload_headers,
        files={"file": ("sample_data.csv", csv_content.encode(), "text/csv")},
        timeout=30,
    )
    upload_result = resp.json().get("data", {})
    file_list = upload_result.get("file_list", [])
    if not file_list:
        print("上传文件失败")
        return
    file_id = file_list[0]["id"]
    print(f"  文件ID: {file_id}, 文件名: {file_list[0]['file_name']}")

    print("\n[步骤3] 发起对话分析...")
    chat_data = {
        "content": "帮我分析这份数据，画一个成绩分布柱状图",
        "verbose": True,
        "enable_web_search": False,
        "deep_think": False,
        "message_uuid": str(uuid.uuid4()),
        "upload_file_id": [file_id],
    }
    stream_request_example(
        host, headers, "POST",
        f"/api/open/office/v2/sessions/{session_id}/chat/conversations",
        chat_data,