日々の開発Tips

【GPT4o対応】OpenAI API のPythonサンプルコードと出力例

Qualiteg プロダクト開発部

2024年6月5日 — 4 min read

今回は、OpenAI の API 利用サンプルコードをご紹介します。

OpenAI API は OpenAI純正のAPI のみならず、vLLMなど他の推論エンジンでも OpenAI 準拠のAPIサーバーが公開されており、LLMサービングAPIのデファクトとなりつつありますので、コーディングのお作法をおさえておきましょう。

OpenAI の GPT シリーズのAPIにアクセスするための、シンプルなサンプルコードは以下のようになります。生成結果をストリーミングで逐次受信してみましょう。

サンプルコード:クイックスタート

import asyncio
import os
import traceback

from openai import AsyncOpenAI


async def main() -> None:
    try:
        # モデル名を指定
        # model="gpt-4-turbo" # $10.00/MTok for input ,$30.00/MTok for output
        # model="gpt-4o" # $5.00/MTok for input ,$15.00/MTok for output
        model = "gpt-3.5-turbo-0125"  #

        # 環境変数からAPIキーを取得
        api_key = "your api key"
        
        client = AsyncOpenAI(
            api_key=api_key
        )

        stream = await client.chat.completions.create(
            model=model,
            stream=True,
            messages=[
                {"role": "system", "content": "あなたは誠実な日本語アシスタントです"},
                {"role": "user", "content": "こんにちは"}
            ],
            stream_options={"include_usage": True},  # usage(in,outのトークン数) を出力
        )

        async for chunk in stream:
            print(f"chunk__{chunk}")
    except Exception as e:
        print(f"予期せぬエラーが発生しました: {e}\n{traceback.format_exc()}")
    finally:
        pass


asyncio.run(main())

出力例

ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='', function_call=None, role='assistant', tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='こんにちは', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='！', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='何', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='か', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='お', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='手', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='伝', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='い', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='で', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='き', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='ます', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='か', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='？', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=None, function_call=None, role=None, tool_calls=None), finish_reason='stop', index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=CompletionUsage(completion_tokens=14, prompt_tokens=31, total_tokens=45))

サンプルコード:ストリーミングされたチャンク内容をパースする

chunkをパースして、中身の各種データを取得してみましょう

import asyncio
import traceback

from openai import AsyncOpenAI


async def main() -> None:
    try:
        model = "gpt-3.5-turbo-0125"
        api_key = "your api key"
        client = AsyncOpenAI(api_key=api_key)

        stream = await client.chat.completions.create(
            model=model,
            stream=True,
            messages=[
                {"role": "system", "content": "あなたは誠実な日本語アシスタントです"},
                {"role": "user", "content": "こんにちは"}
            ],
            stream_options={"include_usage": True},  # usage を出力する
        )

        first_chunk = None
        last_chunk = None
        finish_reason = None
        full_content = ""
        role = None
        created = None
        model = None
        completion_id = None

        async for chunk in stream:

            object_type = chunk.object

            if object_type == "chat.completion.chunk":

                if first_chunk is None:
                    # 初回チャンクのとき
                    first_chunk = chunk

                    model = first_chunk.model
                    created = first_chunk.created
                    completion_id = first_chunk.id

                    if chunk.choices:
                        first_choice = chunk.choices[0]
                        role = first_choice.delta.role

                    # 初回チャンクで取得できる情報
                    print(f"completion_id: {completion_id}")
                    print(f"created: {created}")

                    # 初回チャンクのみで取得できる情報
                    print(f"model: {model}")
                    print(f"role: {role}")

                    print("streaming text：", end="", flush=True)

                last_chunk = chunk

                if chunk.choices:
                    first_choice = chunk.choices[0]

                    if first_choice.delta.content:
                        # 今イテレーションで生成されたテキスト
                        delta_str = first_choice.delta.content

                        print(delta_str, end="", flush=True)  # 生成されたテキストを逐次出力する

                        full_content += delta_str  # 全体テキストに追記

                    if finish_reason is None:
                        finish_reason = first_choice.finish_reason

        print()

        if last_chunk:
            # 最終チャンクのデータを処理

            usage = last_chunk.usage

            print(f"Full Content: {full_content}")
            print(f"Finish Reason: {finish_reason}")

            if usage:
                print(f"ttl tokens: {usage.total_tokens}")
                print(f"num input tokens:: {usage.prompt_tokens}")
                print(f"num output tokens: {usage.completion_tokens}")
            else:
                print("Usage information not available")

    except Exception as e:
        print(f"予期せぬエラーが発生しました: {e}\n{traceback.format_exc()}")
    finally:
        pass


asyncio.run(main())

実行結果

completion_id: chatcmpl-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
created: 123456789
model: gpt-3.5-turbo-0125
role: assistant
streaming text：こんにちは！どのようにお手伝いしましょうか？
Full Content: こんにちは！どのようにお手伝いしましょうか？
Finish Reason: stop
ttl tokens: 51
num input tokens:: 31
num output tokens: 20

大企業のAIセキュリティを支える基盤技術 - 今こそ理解するActive Directory 第2回ドメイン環境の構築

こんにちは、今回はシリーズ第２回ドメイン環境の構築 - 検証環境の構築手順について解説いたします！連載の構成第1章：基本概念の理解 - Active DirectoryとKerberos/NTLM認証の基礎【★今回です★】第2章：ドメイン環境の構築 - 検証環境の構築手順第3章：クライアントとサーバーのドメイン参加 - ドメイン参加の詳細手順第4章：プロキシサーバーと統合Windows認証第5章：ブラウザ設定と認証 - 各ブラウザでの設定方法第6章：トラブルシューティング - よくある問題と解決方法第7章：セキュリティとベストプラクティス - 本番環境での考慮事項第8章：実践的な構成例 - AIセキュリティツールとの統合事例第2章：ドメイン環境の構築 2.1 ドメイン名の設計 2.1.1 ドメイン名の命名規則 Active Directoryを構築する際、

AIがよく間違える「クロージャ問題」の本質と対策

こんにちは！本日は「クロージャ問題」に関する話題となります。 Pythonでループ内に関数を定義したことはありますか？もしあるなら、あれれ？な挙動に遭遇したことがあるかもしれません。本稿では、Pythonプログラマーなら一度は経験する「クロージャ問題」について、初心者にもわかりやすく解説してみたいとおもいますクロージャとは何か？そもそも　”クロージャ”　とは何でしょうか。クロージャ（closure）とは、関数が自分の定義されたスコープの変数を覚えて持ち運ぶ仕組みのことです。もう少し分解すると、次の2つがポイントとなります 1. 内側の関数が、外側の関数の変数を使える 2. 外側の関数が終了しても、その変数は生き続ける普通の関数とクロージャ―を使った関数を比較してみましょう普通の関数との比較まずは普通の関数から、 def add(x, y): return x + y print(add(3, 5)) # 8 print(add(3, 7)

フリーランスHub様にQualiteg Blogをご紹介いただきました

この度、フリーランス向け案件検索サービス「フリーランスHub」様の特集記事「トレンドをキャッチアップ！AIに関する情報が得られるメディア・ブログまとめ」にて、弊社が運営する「Qualiteg Blog」をご紹介いただきました。掲載記事についてフリーランスHub様の記事では、AI技術の最前線で活躍するエンジニアや開発者の方々に向けて、価値ある情報源となるメディア・ブログが厳選して紹介されています。その中で、Qualiteg Blogを「AI技術の専門知識を実践的なビジネス活用につなげる貴重な情報源」として取り上げていただきました。特に以下の点を評価いただいております * 実践的なビジネス活用事例の提供 AI新規事業創出や事業選定方法など、経営者やビジネスリーダーが直面する課題への具体的な解決策 * 技術的な深掘りコンテンツリップシンク技術など、実際のサービスで使用されている技術の開発現場目線での詳細な解説 * 多様な情報発信代表執筆記事、AIトピックス、講演会動画など、幅広いフォーマットでの情報提供今後も価値ある情報発

PyTorchの重いCUDA処理を非同期化したらメモリリークした話と、その解決策

こんにちは！Qualitegプロダクト開発部です！今回は同期メソッドを非同期メソッド(async)化しただけなのに、思わぬメモリリーク※に見舞われたお話です。深層学習モデルを使った動画処理システムを開発していた時のことです。「処理の進捗をリアルタイムでWebSocketで通知したい」という要件があり、「単にasync/awaitを使えばいいだけでしょ？」と軽く考えていたら、思わぬ落とし穴にはまりました。プロ仕様のGPUを使っていたにも関わらず、メモリ不足でクラッシュしてしまいました。この記事では、その原因と解決策、そして学んだ教訓を詳しく共有したいと思います。同じような問題に直面している方の参考になれば幸いです。 ※ 厳密には「メモリリーク」ではなく「メモリの解放遅延」ですが、実用上の影響は同じなので、この記事では便宜上「メモリリーク」と表現します。背景：なぜ進捗通知は非同期である必要があるのかモダンなWebアプリケーションの要求最近のWebアプリケーション開発では、ユーザー体験を向上させるため、長時間かかる処理の進捗をリアルタイムで表示することが