gpt-6-astra

curl --request POST \
  --url https://www.yunqiai.chat/v1/responses \
  --header "Authorization: Bearer YOUR_API_KEY" \
  --header "Content-Type: application/json" \
  --data '{
    "model": "gpt-6-astra",
    "input": "用三句话解释量子计算",
    "max_output_tokens": 1024,
    "stream": false,
    "reasoning": {
      "effort": "low"
    }
  }'

{
  "model": "gpt-6-astra",
  "input": [
    {
      "role": "user",
      "content": [
        {
          "type": "input_text",
          "text": "描述这张图片"
        },
        {
          "type": "input_image",
          "image_url": "data:image/png;base64,BASE64_IMAGE_DATA",
          "detail": "low"
        }
      ]
    }
  ],
  "max_output_tokens": 256
}

import base64
import os
from pathlib import Path

import requests

file_path = Path("report.pdf")
file_data = base64.b64encode(file_path.read_bytes()).decode("ascii")
payload = {
    "model": "gpt-6-astra",
    "input": [{
        "role": "user",
        "content": [
            {
                "type": "input_file",
                "filename": file_path.name,
                "file_data": f"data:application/pdf;base64,{file_data}",
            },
            {"type": "input_text", "text": "提取报告中的结论与关键数字"},
        ],
    }],
}

response = requests.post(
    "https://www.yunqiai.chat/v1/responses",
    headers={
        "Authorization": f"Bearer {os.environ['YUNQIAI_API_KEY']}",
        "Content-Type": "application/json",
    },
    json=payload,
    timeout=(10, 180),
)
response.encoding = "utf-8"
if not response.ok:
    request_id = response.headers.get("x-request-id") or response.headers.get("request-id")
    raise RuntimeError(f"HTTP {response.status_code}; request_id={request_id}; body={response.text[:2000]}")
result = response.json()

if result.get("error") or result.get("status") != "completed":
    raise RuntimeError(f"File response not completed: {result.get('id')}; {result.get('status')}; {result.get('error')}; {result.get('incomplete_details')}")
texts = [
    part["text"]
    for item in result.get("output", [])
    if item.get("type") == "message"
    for part in item.get("content", [])
    if part.get("type") == "output_text"
]
if not any(text.strip() for text in texts):
    raise RuntimeError(f"No file answer in response: {result.get('id')}")
print("\n".join(texts))

import json
import os

import requests

def iter_sse_data(response):
    data_lines = []
    for line in response.iter_lines(decode_unicode=True):
        if line == "":
            if data_lines:
                data = "\n".join(data_lines)
                data_lines = []
                if data:
                    yield data
            continue
        if line.startswith(":"):
            continue
        field, _, value = line.partition(":")
        if value.startswith(" "):
            value = value[1:]
        if field == "data":
            data_lines.append(value)
    # A frame without its terminating blank line is incomplete.


payload = {
    "model": "gpt-6-astra",
    "input": "用三句话解释量子计算",
    "stream": True,
}

with requests.post(
    "https://www.yunqiai.chat/v1/responses",
    headers={
        "Authorization": f"Bearer {os.environ['YUNQIAI_API_KEY']}",
        "Content-Type": "application/json",
    },
    json=payload,
    stream=True,
    timeout=(10, 180),
) as response:
    response.encoding = "utf-8"
    if not response.ok:
        request_id = response.headers.get("x-request-id") or response.headers.get("request-id")
        raise RuntimeError(f"HTTP {response.status_code}; request_id={request_id}; body={response.text[:2000]}")
    completed = False
    tool_calls = {}

    for data in iter_sse_data(response):
        if data == "[DONE]":
            break
        event = json.loads(data)
        event_type = event.get("type")
        if event_type == "response.output_text.delta":
            print(event.get("delta", ""), end="", flush=True)
        elif event_type == "response.output_item.added":
            item = event.get("item") or {}
            if item.get("type") == "function_call":
                item_id = item.get("id", str(event.get("output_index", 0)))
                tool_calls[item_id] = {
                    "call_id": item.get("call_id", ""),
                    "name": item.get("name", ""),
                    "arguments": item.get("arguments", ""),
                }
        elif event_type == "response.function_call_arguments.delta":
            item_id = event.get("item_id", str(event.get("output_index", 0)))
            current = tool_calls.setdefault(item_id, {"call_id": "", "name": "", "arguments": ""})
            current["arguments"] += event.get("delta", "")
        elif event_type == "response.function_call_arguments.done":
            item_id = event.get("item_id", str(event.get("output_index", 0)))
            current = tool_calls.setdefault(item_id, {"call_id": "", "name": "", "arguments": ""})
            current["arguments"] = event.get("arguments", current["arguments"])
        elif event_type == "response.output_item.done":
            item = event.get("item") or {}
            if item.get("type") == "function_call":
                item_id = item.get("id", str(event.get("output_index", 0)))
                current = tool_calls.setdefault(item_id, {"call_id": "", "name": "", "arguments": ""})
                current["call_id"] = item.get("call_id", current["call_id"])
                current["name"] = item.get("name", current["name"])
                current["arguments"] = item.get("arguments", current["arguments"])
        elif event_type == "response.completed":
            for item in (event.get("response") or {}).get("output", []):
                if item.get("type") != "function_call":
                    continue
                item_id = item.get("id", str(item.get("output_index", 0)))
                current = tool_calls.setdefault(item_id, {"call_id": "", "name": "", "arguments": ""})
                current.update({
                    "call_id": item.get("call_id", current["call_id"]),
                    "name": item.get("name", current["name"]),
                    "arguments": item.get("arguments", current["arguments"]),
                })
            completed = True
        elif event_type in {"response.failed", "response.incomplete"}:
            final_response = event.get("response", {})
            error = final_response.get("error") or {}
            detail = error.get("message") or final_response.get("incomplete_details") or event_type
            raise RuntimeError(str(detail))
        elif event_type == "error":
            error = event.get("error") or event
            code = error.get("code") or error.get("type") or "stream_error"
            raise RuntimeError(f"{code}: {error.get('message', 'unknown error')}")

if not completed:
    raise RuntimeError("stream ended before response.completed")

print()
for call in tool_calls.values():
    arguments = json.loads(call["arguments"] or "{}")
    print("tool", call["call_id"], call["name"], arguments)

{
  "error": {
    "message": "请求参数无效",
    "type": "invalid_request_error",
    "param": "model",
    "code": "invalid_model"
  }
}

import os
import requests

url = "https://www.yunqiai.chat/v1/responses"
headers = {"Authorization": f"Bearer {os.environ['YUNQIAI_API_KEY']}"}
payload = {"model": "gpt-6-astra", "input": "你好", "max_output_tokens": 1024}

# A POST may have reached the server even when the client times out.
# Do not automatically replay it or switch to a different model.
response = requests.post(url, headers=headers, json=payload, timeout=(10, 180))
request_id = response.headers.get("x-request-id") or response.headers.get("request-id")
if not response.ok:
    raise RuntimeError(
        f"HTTP {response.status_code}; request_id={request_id or 'unavailable'}; "
        "check the redacted error response before starting a new request"
    )
print(response.json())

{
  "id": "resp_01JY7X",
  "object": "response",
  "status": "completed",
  "model": "gpt-6-astra",
  "output": [
    {
      "id": "msg_01JY7X",
      "type": "message",
      "status": "completed",
      "role": "assistant",
      "content": [
        {
          "type": "output_text",
          "annotations": [],
          "text": "量子计算利用量子态处理信息……"
        }
      ]
    }
  ],
  "usage": {
    "input_tokens": 16,
    "output_tokens": 48,
    "total_tokens": 64
  }
}