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

【GPT4o対応】OpenAI API のPythonサンプルコードと出力例
Photo by Fotis Fotopoulos / Unsplash

今回は、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

Read more

大企業のAIセキュリティを支える基盤技術 - 今こそ理解するActive Directory 第6回 よくある問題と解決方法

大企業のAIセキュリティを支える基盤技術 - 今こそ理解するActive Directory 第6回 よくある問題と解決方法

こんにちは、今回はシリーズ第6回トラブルシューティング - よくある問題と解決方法 について解説いたします! さて、前回(第5回)は、統合Windows認証がブラウザでどのように動作するかを解説しました。 「イントラネットゾーン」という概念を理解することで、同じサーバーでもURLの書き方(NetBIOS名、FQDN、IPアドレス)によって認証動作が変わる理由が明確になったかと思います。また、Chrome/Firefoxではデフォルトで統合認証が無効になっている理由と、グループポリシーによる一括設定方法も学びました。 しかし、設定が完璧なはずなのに「なぜかうまく動かない」という場面は、実際の現場では必ず訪れます。 「最近、ファイルサーバーへのアクセスが遅い」「金曜日は使えたのに、月曜日の朝にログインできない」「特定のサービスだけKerberosが失敗する」——これらはヘルプデスクに日々寄せられる典型的な問い合わせです。 原因はKerberosの失敗、時刻のずれ、SPNの設定ミス、DNS関連の問題など多岐にわたりますが、体系的にトラブルシューティングすることで必ず解決できます。

By Qualiteg コンサルティング, Qualiteg AIセキュリティチーム
AIエージェントを"事業に載せる"ために【第2回】AIエージェントの責任分解はなぜ難しいのか

AIエージェントを"事業に載せる"ために【第2回】AIエージェントの責任分解はなぜ難しいのか

— AI導入を"事業に載せる"ために、いま設計すべきこと(全3回) こんにちは!Qualitegコンサルティングチームです! 前回(第1回)では、Replit/Lemkin事件とDeloitte豪州政府報告書問題を通じて、AIエージェント導入の課題がモデル性能ではなく「権限・監査・責任の設計不在」にあることを見ました。 では、実際に事故が起きたとき、責任は誰が負うのでしょうか。第2回となる本記事では、法務・契約・組織の3つの観点から、AIエージェントの責任分解がなぜ難しいのかを構造的に整理します。 結論を先に言えば、法務だけでも契約だけでも組織論だけでも足りません。この3つを接続して設計しなければ、AIエージェントの責任分解は実務上機能しません。 1. 法的フレームワーク:複数の法理論が並走している AIエージェントが損害を出したとき、どの法理論で責任が問われるかについて、現時点でグローバルなコンセンサスは形成されていません。 Clifford Chanceの論考は、この状況の根本的な難しさを整理しています。法律は歴史的に、有害な行為がいつどのように発生したかを特定でき

By Qualiteg コンサルティング
AIエージェントを"事業に載せる"ために【第1回】

AIエージェントを"事業に載せる"ために【第1回】

AI導入事故は何を示しているのか — AI導入を"事業に載せる"ために、いま設計すべきこと(全3回) こんにちは!Qualitegコンサルティングチームです! AIエージェントを導入する企業が増える一方で、 「試してみる」段階から「事業に載せる」段階へ進める難しさ が、はっきり見え始めています。 本シリーズでは、AIエージェント導入を技術論だけでなく、責任分解・監査可能性・契約・運用統制を含む業務設計の問題として整理します。 全3回を通じて、「AIが賢いかどうか」ではなく、「AIを業務に載せるために何を設計するか」を考えていきます。 第1回となる本記事では、2025年に起きた2つの事例を出発点に、なぜいま「責任設計」が問題になっているのかを見ていきます。 上図は、本シリーズ全体で扱う論点の全体像です。 AIエージェントの導入は、技術的なモデル選定だけでは完結せず、権限設計、契約、監査、品質監視、保険、異常時対応まで含めた設計が必要になります。 第1回ではまず、なぜこうした設計が求められるようになったのかを、実際の事例から見ていきたいとおもいます なお、本シリー

By Qualiteg コンサルティング
PII検出の混同行列では見えないもの ― 認識器間衝突と統合テスト

PII検出の混同行列では見えないもの ― 認識器間衝突と統合テスト

こんにちは!Qualiteg研究部です! 個人情報(PII: Personally Identifiable Information)の自動検出は、テキスト中から特定の表現を抽出し、それがどの種類のPIIに当たるかを判定する問題として捉えることができます。 電話番号、人名、口座番号、金額表現など、検出対象のPIIタイプが増えるにつれて、単一の手法ではカバーしきれなくなり、性質の異なる複数の認識器(Recognizer)を組み合わせるマルチレイヤー構成が採用されるのが一般的です。 本稿で想定しているのは、ユーザーが海外製LLMにチャットを送信する直前に、その内容に個人情報や機密情報が含まれていないかをリアルタイムに検査するユースケースです。 この場面では、検出精度だけでなく、送信体験を損ねない速度が不可欠です。 高精度なLLMやBERT系モデル、NERベースの手法は有力ですが、送信前チェックの第一層として常時適用するには、レイテンシやコストの面で不利になることがあります。 そのため、本システムでは、正規表現、辞書、軽量なルールベース認識器を組み合わせた超高速な第一層を設け、そ

By Qualiteg 研究部, Qualiteg AIセキュリティチーム