genealogy-app/test_ai_stream.py

import requests
import json
import os

api_key = "a1800657-9212-4afe-9b7c-b49f015c54d3"
api_url = "https://ark.cn-beijing.volces.com/api/v3/responses"

prompt = "你看见了什么？"
image_url = "https://ark-project.tos-cn-beijing.volces.com/doc_image/ark_demo_img_1.png"

payload = {
    "model": "doubao-seed-1-8-251228",
    "stream": True,
    "input": [
        {
            "role": "user",
            "content": [
                {
                    "type": "input_image",
                    "image_url": image_url
                },
                {
                    "type": "input_text",
                    "text": prompt
                }
            ]
        }
    ]
}

headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

print("开始请求 AI 接口...")
try:
    with requests.post(api_url, json=payload, headers=headers, stream=True, timeout=30) as r:
        print(f"状态码: {r.status_code}")
        if r.status_code != 200:
            print(f"错误内容: {r.text}")
        else:
            print("开始接收流式数据:")
            for line in r.iter_lines():
                if line:
                    line_str = line.decode('utf-8')
                    print(f"收到行: {line_str}")
                    if line_str.startswith('data: '):
                        json_str = line_str[6:]
                        if json_str.strip() == '[DONE]':
                            print("流结束")
                            break
                        try:
                            chunk = json.loads(json_str)
                            if 'choices' in chunk and len(chunk['choices']) > 0:
                                delta = chunk['choices'][0].get('delta', {})
                                if 'content' in delta:
                                    print(f"内容片段: {delta['content']}", end="", flush=True)
                        except Exception as e:
                            print(f"\n解析错误: {e}")
            print("\n请求完成")
except Exception as e:
    print(f"请求发生异常: {e}")