OpenAIのRealtimeAPIをチャットで利用して理解を深める！Python版！

# はじめに
* Pythonのopenaiライブラリにおいて、12月17日リリースの ver. 1.58 で、RealtimeAPIがサポートされました
* これに伴い早速、前回Node.jsで書いた以下の記事の内容を、Pythonで作り直したので、だれかの助けになればと記事化したいと思います
    * [OpenAIのRealtimeAPIをチャットで利用して理解を深める！](https://qiita.com/tuneyukkie/items/b81aa06f21575b3f77eb)

# 実行例
* 今回は、実行のイベントを色分けで区別して表示するようにしています
* 色分けはおおまかに、以下としています
    * Conversationアイテムは黄色系、
    * Sessionは紫、
    * Responseは緑色


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/fbd031b1-3bf6-c2b8-200a-d74420249b6d.png)

# シーケンス図

```mermaid
sequenceDiagram
    participant User as ユーザー
    participant Client as Pythonクライアント
    participant OpenAI as OpenAI Realtime API

    User->>Client: アプリ実行
    Client->>OpenAI: client.beta.realtime.connect(WebSocket接続要求)
    OpenAI-->>Client: session.created<br>（セッション作成）
    Client->>OpenAI: session.update<br>（ツール登録: getCurrentWeather）
    OpenAI-->>Client: session.updated<br>（ツール登録成功）

    Note over Client: ユーザー入力を待機

    Note over Client: Function Call を呼ぶチャットの流れ
    User->>Client: メッセージ入力<br>("東京の天気を知りたいんだけど")
    Client->>OpenAI: conversation.item.create<br>（ユーザー入力送信）
    OpenAI-->>Client: conversation.item.created<br>（ユーザー入力送信完了）
    Client->>OpenAI: response.create<br>（レスポンス生成リクエスト）
    OpenAI-->>Client: response.created<br>（レスポンス進行中）
    OpenAI-->>Client: response.output_item.added<br>（関数呼び出し: getCurrentWeather）

    loop 関数引数の送信
        OpenAI-->>Client: response.function_call_arguments.delta<br>("{\"location\":\"")
        OpenAI-->>Client: response.function_call_arguments.delta<br>("東京\"}")
    end

    OpenAI-->>Client: response.function_call_arguments.done<br>（関数引数確定: {"location":"東京"}）
    Client->>Client: getCurrentWeather 実行<br>（ローカルで処理）
    Client->>OpenAI: conversation.item.create<br>（関数結果送信: "Weather:sunny, Temperature:23, wind:10"）
    Client->>OpenAI: response.create<br>（次のレスポンス生成リクエスト）

    loop Deltaレスポンス送信
        OpenAI-->>Client: response.text.delta<br>("東京, は, 晴れ, で, 気温, は, 23度, です！")<br>※小分け(stream的)に到着
    end

    OpenAI-->>Client: response.text.done<br>（完全なレスポンス通知：東京は晴れで気温は23度です！）
    OpenAI-->>Client: response.output_item.done<br>（レスポンス生成完了）
    Client->>User: レスポンス表示<br>("東京は晴れで気温は23度です！")
```

# ソースコード
* 必要パッケージは以下
* openaiについては、最新版を入れるために、`pip install --upgrade openai`とやっておいたほうがよいかも
```
pip install asyncio aioconsole colorama openai
```

```python
import os
import asyncio
import logging
import json

from openai import AsyncOpenAI
from aioconsole import ainput

from colorama import Fore, Style, init

# Windows環境でも色が表示されるように初期化
init(autoreset=True)

OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]

# イベントタイプごとの色付け
EVENT_COLORS = {
    'session.created': Fore.MAGENTA,
    'session.updated': Style.BRIGHT + Fore.MAGENTA,
    'rate_limits.updated': Style.BRIGHT + Fore.MAGENTA,
    'conversation.item.created': Fore.YELLOW,
    'response.created': Fore.GREEN,
    'response.output_item.added': Style.BRIGHT + Fore.GREEN,
    'response.content_part.added': Style.BRIGHT + Fore.GREEN,
    'response.content_part.done': Style.BRIGHT + Fore.GREEN,
    'response.output_item.done': Style.BRIGHT + Fore.GREEN,
    'response.function_call_arguments.delta': Fore.BLUE,
    'response.function_call_arguments.done': Style.BRIGHT + Fore.BLUE,
    'response.text.delta': Fore.GREEN,
    'response.done': Fore.CYAN,
    'response.text.done': Fore.YELLOW,
}

# ダミーの天気情報取得関数
def get_current_weather(location):
    if not location:
        logging.error("Missing location argument")
        return {"error": "Location not provided"}
    return {
        "location": location,
        "weather": "Sunny",
        "temperature": "23degrees",
        "wind_speed": "10meter",
    }

# WebSocket イベントを処理するメインハンドラ
async def handle_websocket_events(connection):
    async for event in connection:
        try:
            event_type = event.type

            if event_type in EVENT_COLORS:
                color = EVENT_COLORS[event_type]
                print(f"{color}{event_type}: {event}{Style.RESET_ALL}")

            if event_type == "response.function_call_arguments.done":
                await handle_function_call(event, connection)

            elif event_type == "response.text.delta":
                print(f"{Fore.YELLOW}Delta: {event.delta}{Style.RESET_ALL}", end="\n")


        except Exception as e:
            logging.error(f"Error handling event: {e}")
            break

# Function Call の処理
async def handle_function_call(event, connection):
    print("Handling function call...")
    try:
        data = json.loads(event.arguments)
        location = data.get("location")
        result = get_current_weather(location)
        await connection.conversation.item.create(item={
            "type": "function_call_output",
            "call_id": event.call_id,
            "output": json.dumps(result),
        })
        await connection.response.create()
    except Exception as e:
        logging.error(f"Error handling function call: {e}")

# ユーザー入力の処理
async def process_user_input(connection):
    # Session Update完了まで3秒程度待つ
    await asyncio.sleep(3)
    while True:
        user_input = await ainput(f"{Fore.CYAN}You: {Style.RESET_ALL}")
        if not user_input.strip():
            continue  # 空文字の場合は次の入力を促す
        if user_input.lower() in ["exit", "quit", "q"]:
            print(f"{Fore.RED}Exiting chat...{Style.RESET_ALL}")
            await connection.close()
            break

        try:
            await connection.conversation.item.create(item={
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": user_input}]
            })
            await connection.response.create()
        except Exception as e:
            logging.error(f"Error sending user input: {e}")

        # 1秒間の待機を追加
        await asyncio.sleep(1)


# メイン処理
async def main():
    client = AsyncOpenAI(api_key=OPENAI_API_KEY)
    async with client.beta.realtime.connect(model="gpt-4o-mini-realtime-preview-2024-12-17") as connection:
        print("Connected to WebSocket.")

        # セッション設定
        await connection.session.update(session={
            "modalities": ["text"],
            "tools": [
                {
                    "type": "function",
                    "name": "getCurrentWeather",
                    "description": "指定された都市の現在の天気を取得します。",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "location": {"type": "string", "description": "都市名"},
                        },
                        "required": ["location"],
                    },
                },
            ],
            "tool_choice": "auto",
        })

        # イベント処理とユーザー入力を並行実行
        await asyncio.gather(
            handle_websocket_events(connection),
            process_user_input(connection),
        )

if __name__ == "__main__":
    asyncio.run(main())
```

## 補足説明
* コンソールでのユーザー入力を受け付けるため、Websocketのイベントハンドラと、ユーザー入力は並列動作させています
```
        # イベント処理とユーザー入力を並行実行
        await asyncio.gather(
            handle_websocket_events(connection),
            process_user_input(connection),
        )
```

# さいごに
* 今後ますますニーズが高まっていくと思われる、RealtimeAPIを利用して、みなさんもハッピーAIライフをお過ごしください！