【第3回】GPT-6 Astra API Python 実践入門 — ストリーミング応答とファイル・画像入力の実践活用

※この記事にはプロモーションが含まれます

前回の記事では、GPT-6 Astra API を Python から呼び出す基本部分を紹介しました。今回はその続きで、「返答が全部出るまで待つ」状態から抜け出すためのストリーミングと非同期処理、それから画像やPDFを投げる方法を整理していきます。

Astra は推論が重めのモデルなので、普通に呼び出すと返事が来るまでそこそこ待たされます。ターミナルを見つめて固まった経験がある人は多いはず……というわけで、ストリーミングを先にやっておきたかったのが今回の動機です。

この記事でわかること

• Responses API でストリーミング応答を受け取る基本的な書き方
• AsyncOpenAI と asyncio で複数リクエストを並行して流す方法（同時実行数の制限つき）
• 画像（URL / base64）とPDFを入力として渡す方法
• ストリーミングやマルチモーダル入力で気をつけたいコストと制約

前提：GPT-6 Astra API の仕様をおさらい

先に、今回関係するところだけ仕様を確認しておきます。

• モデルIDは gpt-6-astra。Responses API から使う形が基本です。
• コンテキストウィンドウは約105万トークン、最大出力は128,000トークン
• 入力はテキストに加えて画像やPDFなどのファイルも扱える、とされています。ストリーミングやツール呼び出し（関数呼び出し）なども含めて、どこまでを「Astraの対応」と呼ぶかは機能の粒度が混ざりやすいので、困ったらResponses APIのリファレンスも併読するのがよさそうです。
• reasoning effort は low / medium / high / xhigh / max の5段階
• 標準料金は入力100万トークンあたり10ドル、出力100万トークンあたり50ドル（キャッシュされた入力など条件によって変わります）

ちなみに Fast mode というのもあって、標準の最大2倍の速度で動くかわりに料金も2倍になるそうです。「待ち時間が嫌なら Fast mode でいいのでは？」とも思いましたが、出力50ドルの2倍はなかなか勇気がいります。体感の待ち時間を減らすだけなら、ストリーミングのほうが財布に優しいです。

ストリーミング応答の基本（同期版）

Responses API では stream=True を付けると、レスポンスが「イベント」の列として順番に届きます。テキストの断片は response.output_text.delta というイベントで来るので、それを拾って順に表示すればOKです。

from openai import OpenAI

client = OpenAI()

stream = client.responses.create(
    model="gpt-6-astra",
    input="AWS Lambdaのコールドスタート対策を3つ、短く教えて",
    reasoning={"effort": "low"},
    stream=True,
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)  # flushしないとまとめて表示されることがある
    elif event.type == "response.completed":
        usage = event.response.usage
        print(f"\n\n--- input: {usage.input_tokens} / output: {usage.output_tokens}")

ポイントは、最後の response.completed イベントでレスポンス全体や usage が取れることが多い、という点です。ストリーミングにするとログ用の usage を取り逃しがちなので、ここは押さえておきたいところ。

イベントの種類はほかにも、response.created、推論サマリーの delta、ツール呼び出し関連など色々あります。最初は全部 print してみて、どんなイベントがどの順番で来るかを眺めるのが一番わかりやすいかもしれません。Astra は推論を挟むぶん、最初のテキスト delta が来るまで少し間があることも頭に入れておくとよさそうです。

reasoning effort とストリーミングの関係

ここは正直、自分もまだ感覚をつかみきれていません。effort を上げると、最初の1文字が出るまでの時間が伸びやすいはずです。チャットUIのように「とにかく早く何か表示したい」用途なら low〜medium、バッチで品質重視なら high 以上、という使い分けが無難かなと思っています。

非同期処理で複数リクエストを並行実行する

次は非同期処理です。たとえば「質問を10個まとめて投げたい」とき、同期のまま for で回すと、1件ずつ律儀に待つことになります。Astra は1件あたりの応答時間が長めなので、ここを並行化できるかどうかで体感がかなり変わります。

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()
sem = asyncio.Semaphore(3)

async def ask(question: str) -> str:
    async with sem:
        stream = await client.responses.create(
            model="gpt-6-astra",
            input=question,
            reasoning={"effort": "low"},
            stream=True,
        )
        parts = []
        async for event in stream:
            if event.type == "response.output_text.delta":
                parts.append(event.delta)
            elif event.type == "error":
                raise RuntimeError(event.message)
        return "".join(parts)

async def main():
    questions = [
        "DynamoDBのパーティションキー設計のコツは？",
        "SQSの可視性タイムアウトとは？",
        "EventBridgeのスケジュール式の書き方は？",
    ]
    results = await asyncio.gather(*(ask(q) for q in questions), return_exceptions=True)
    for q, r in zip(questions, results):
        print(f"Q: {q}\nA: {r}\n")

asyncio.run(main())

ハマりやすいのは await client.responses.create(...) の部分です。非同期クライアントの場合、まず await してストリームオブジェクトを受け取り、そのあと async for で回します。await を忘れると「coroutine は async iterable じゃないよ」と怒られます。

Semaphore で同時実行数を絞っているのはレート制限対策です。何も考えずに gather で100件投げると、レート制限に一気に引っかかります。いくつまで並べられるかは、アカウントの契約や制限によって違うようです。

実際のAPIを呼ばずに「Semaphore で絞りつつ並行でストリームを読む」動きだけを確認する場合、ダミーのストリームで再現するのが手っ取り早いです。

import asyncio
import random
import time

async def fake_stream(text: str):
    for ch in text:
        await asyncio.sleep(random.uniform(0.01, 0.03))
        yield {"type": "response.output_text.delta", "delta": ch}
    yield {"type": "response.completed"}

async def run_one(name: str, text: str, sem: asyncio.Semaphore):
    async with sem:
        start = time.perf_counter()
        buf = []
        async for event in fake_stream(text):
            if event["type"] == "response.output_text.delta":
                buf.append(event["delta"])
            elif event["type"] == "response.completed":
                break
        elapsed = time.perf_counter() - start
        return name, "".join(buf), elapsed

async def main():
    random.seed(0)
    sem = asyncio.Semaphore(2)
    jobs = {
        "A": "Lambdaのコスト集計",
        "B": "DynamoDBの設計メモ",
        "C": "S3のライフサイクル",
    }
    t0 = time.perf_counter()
    results = await asyncio.gather(*(run_one(k, v, sem) for k, v in jobs.items()))
    for name, text, sec in results:
        print(f"[{name}] {text} ({sec:.2f}s)")
    print(f"total: {time.perf_counter() - t0:.2f}s")

asyncio.run(main())

▲実際にこのブログの裏側で実行してみた結果です

同時実行数を2にしているので、AとBが先に並行で走り、Cは空きが出てから始まります。合計時間は、3件の所要時間を全部足した値より短くなるはずです。本物のAPIでも構造はこれとほぼ同じです。

画像入力：URLとbase64の2通り

画像は input_image として、テキストと並べて content に入れます。公開URLがあればそのまま渡せますが、ローカルの画像なら base64 の data URL にするのが手っ取り早いです。

import base64
from pathlib import Path
from openai import OpenAI

client = OpenAI()

def to_data_url(path: str, mime: str = "image/png") -> str:
    b64 = base64.b64encode(Path(path).read_bytes()).decode()
    return f"data:{mime};base64,{b64}"

stream = client.responses.create(
    model="gpt-6-astra",
    input=[{
        "role": "user",
        "content": [
            {"type": "input_text", "text": "このコストグラフから、増えているサービスと考えられる原因を挙げて"},
            {"type": "input_image", "image_url": to_data_url("cost_dashboard.png"), "detail": "high"},
        ],
    }],
    stream=True,
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)

ここで想定しているのは、Lambda で AWS の日次コストを集計して Web ページに出すところまで作ったものの、表示部分を数か月ほど放置しているページのスクリーンショットです。表示を作り込むより、スクショを投げて解説させたほうが早いのでは……と一瞬思いましたが、それだと完全に負けな気もします。

画像も入力トークンとしてカウントされるので、detail を high にすると高くつく可能性があります。細かい文字を読ませたいときだけ high にする、くらいの使い分けがよさそうです。

動画は直接は渡せない

ちなみに動画（mp4）は、少なくとも現時点の Responses API のネイティブ入力としては未対応のようです。回避策としては、一定間隔で切り出したフレームを input_image として並べ、プロンプトにタイムスタンプを書き込む、といったやり方が現実的です。画像の枚数が増えるとトークンも一気に増えるので、そこは注意が必要です。

PDF・ファイル入力

PDF は input_file で渡します。方法は2つあって、Files API に一度アップロードして file_id で参照するか、base64 で直接埋め込むかです。

from openai import OpenAI

client = OpenAI()

with open("aws_invoice.pdf", "rb") as f:
    uploaded = client.files.create(file=f, purpose="user_data")

response = client.responses.create(
    model="gpt-6-astra",
    input=[{
        "role": "user",
        "content": [
            {"type": "input_file", "file_id": uploaded.id},
            {"type": "input_text", "text": "この請求書で金額が大きい上位3サービスを表にして"},
        ],
    }],
)
print(response.output_text)

base64 で直接渡す場合は、{"type": "input_file", "filename": "aws_invoice.pdf", "file_data": "data:application/pdf;base64,..."} のような形になります。同じファイルを何度も使うなら file_id 方式、使い捨てなら base64 方式、という使い分けが自分の中ではしっくりきています。

コンテキストが約105万トークンあるので、長いPDFもそのまま入れられそうに見えます。ただ、入力1Mトークンで10ドルという料金を考えると、雑に全部突っ込むのはおすすめしづらいです。また、条件によって料金テーブルが変わるので、長文PDFを突っ込む前に公式の料金ページで確認したほうが安心です。

ハマりそうなポイント

ちゃんと整理しきれていないので、箇条書きで並べておきます。

• Lambda でストリーミングしたい場合：Lambda のレスポンスストリーミングは、マネージドランタイムだと Node.js が中心だったはずです。Python でやるなら、Lambda Web Adapter を使うなど一工夫いりそうです。ここはまだ試せていないので要確認
• 途中で接続が切れたとき：ストリーミング中にエラーになると、それまでの delta だけが手元に残ります。リトライするなら「最初からやり直す」前提で組むのが無難かなと
• Enterprise のワークスペース：提供開始時点では Astra がデフォルトでオフになっていたらしく、モデルが見つからないエラーが出たらプロジェクト設定を確認
• 画像＋非同期の組み合わせ：並行で画像を大量に投げると、レート制限とコストの両方が一気に膨らみます。Semaphore は小さめから始めたほうが安心

もっときれいなエラーハンドリングの書き方があるのかな、と思いつつまだ手を付けていません。

PR（アフィリエイト広告）

そういえば、GPT-6 Astra APIのストリーミング処理でスクリーンショットを大量に保存してたら画質の粗さが気になってきて、調べてたらAiarty Image Enhancerというツールを見つけました。非同期処理の合間にちょっと試してみようかなと思ってます。

まとめ

ストリーミング応答は stream=True を付けて delta イベントを拾うだけ、非同期処理は AsyncOpenAI と Semaphore を組み合わせるだけで、基本形は意外とシンプルです。画像とPDFも content に input_image / input_file を足す形なので、ストリーミングとそのまま組み合わせられます。

ツール呼び出しまわり（特にAstraで追加されたと言われている要素）は、自分もまだ追いきれていないので、一次情報（公式ドキュメント）を見ながら詰めていくのがよさそうです。