diff --git a/README.md b/README.md index 1fabef6a81..5f2c066cdb 100644 --- a/README.md +++ b/README.md @@ -396,7 +396,7 @@ Basic text based example: ```py import asyncio -from openai import AsyncOpenAI +from openai import AsyncOpenAI, OpenAIError async def main(): @@ -422,7 +422,13 @@ async def main(): print() elif event.type == "response.done": + if event.response.status != "completed": + raise OpenAIError("The Realtime response did not complete") break + elif event.type == "error": + raise OpenAIError("Realtime returned an error; inspect the event in your application") + else: + raise OpenAIError("Realtime disconnected before response.done") asyncio.run(main()) diff --git a/examples/live/audio_transcript.py b/examples/live/audio_transcript.py index ce952ee343..d3d9ad054d 100644 --- a/examples/live/audio_transcript.py +++ b/examples/live/audio_transcript.py @@ -4,7 +4,9 @@ python examples/live/audio_transcript.py speech.wav Use a 16-bit mono WAV recorded at 24 kHz. Audio is paced in real time. This -example sends silence after the recording so the assistant can finish replying. +example sends silence for the listen window after the recording, then requests +session closure and waits for confirmation. The window does not guarantee that +the assistant has finished replying. """ from __future__ import annotations @@ -49,16 +51,23 @@ def render(segment: TranscriptSegment) -> None: event = await asyncio.wait_for(connection.recv(), timeout=30) if event.type == "error": raise OpenAIError("Live rejected session startup; inspect your model access and configuration") + if event.type == "session.closed": + raise OpenAIError("Live session ended before startup") if event.type == "session.started": break + closing = False + async def receive() -> None: async for event in connection: if event.type == "error": raise OpenAIError("Live returned an error while processing the audio") await transcript.push(event) if event.type == "session.closed": + if not closing or event.reason != "close_requested": + raise OpenAIError("Live session ended without confirming the requested close") return + raise OpenAIError("Live disconnected before session.closed") receiver = asyncio.create_task(receive()) try: @@ -74,7 +83,9 @@ async def receive() -> None: await asyncio.sleep(len(chunk) / (24000 * 2)) if receiver.done(): await receiver - await transcript.close() + closing = True + await connection.session.close() + await asyncio.wait_for(receiver, timeout=30) finally: receiver.cancel() await asyncio.gather(receiver, return_exceptions=True) diff --git a/examples/realtime/azure_realtime.py b/examples/realtime/azure_realtime.py index a8bcf4c536..b0b044559c 100644 --- a/examples/realtime/azure_realtime.py +++ b/examples/realtime/azure_realtime.py @@ -20,60 +20,64 @@ async def main() -> None: Enter "q" to quit the conversation. """ - credential = DefaultAzureCredential() - token_provider = get_bearer_token_provider(credential, "https://cognitiveservices.azure.com/.default") - token = await token_provider() - - # The endpoint of your Azure OpenAI resource is required. You can set it in the AZURE_OPENAI_ENDPOINT - # environment variable. - # You can find it in the Microsoft Foundry portal in the Overview page of your Azure OpenAI resource. - # Example: https://{your-resource}.openai.azure.com - endpoint = os.environ["AZURE_OPENAI_ENDPOINT"] - - # The deployment name of the model you want to use is required. You can set it in the AZURE_OPENAI_DEPLOYMENT_NAME - # environment variable. - # You can find it in the Foundry portal in the "Models + endpoints" page of your Azure OpenAI resource. - # Example: gpt-realtime - deployment_name = os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"] - - base_url = endpoint.replace("https://", "wss://").rstrip("/") + "/openai/v1" - - # The APIs are compatible with the OpenAI client library. - # You can use the OpenAI client library to access the Azure OpenAI APIs. - # Make sure to set the baseURL and apiKey to use the Azure OpenAI endpoint and token. - client = AsyncOpenAI(websocket_base_url=base_url, api_key=token) - async with client.realtime.connect( - model=deployment_name, - ) as connection: - await connection.session.update( - session={ - "output_modalities": ["text"], - "model": deployment_name, - "type": "realtime", - } - ) - while True: - user_input = input("Enter a message: ") - if user_input == "q": - break - - await connection.conversation.item.create( - item={ - "type": "message", - "role": "user", - "content": [{"type": "input_text", "text": user_input}], - } + async with DefaultAzureCredential() as credential: + token_provider = get_bearer_token_provider(credential, "https://cognitiveservices.azure.com/.default") + token = await token_provider() + + # The endpoint of your Azure OpenAI resource is required. You can set it in the AZURE_OPENAI_ENDPOINT + # environment variable. + # You can find it in the Microsoft Foundry portal in the Overview page of your Azure OpenAI resource. + # Example: https://{your-resource}.openai.azure.com + endpoint = os.environ["AZURE_OPENAI_ENDPOINT"] + + # The deployment name of the model you want to use is required. You can set it in the AZURE_OPENAI_DEPLOYMENT_NAME + # environment variable. + # You can find it in the Foundry portal in the "Models + endpoints" page of your Azure OpenAI resource. + # Example: gpt-realtime + deployment_name = os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"] + + base_url = endpoint.replace("https://", "wss://").rstrip("/") + "/openai/v1" + + # The APIs are compatible with the OpenAI client library. + # You can use the OpenAI client library to access the Azure OpenAI APIs. + # Make sure to set the baseURL and apiKey to use the Azure OpenAI endpoint and token. + async with ( + AsyncOpenAI(websocket_base_url=base_url, api_key=token) as client, + client.realtime.connect(model=deployment_name) as connection, + ): + await connection.session.update( + session={ + "output_modalities": ["text"], + "model": deployment_name, + "type": "realtime", + }, ) - await connection.response.create() - async for event in connection: - if event.type == "response.output_text.delta": - print(event.delta, flush=True, end="") - elif event.type == "response.output_text.done": - print() - elif event.type == "response.done": + while True: + user_input = input("Enter a message: ") + if user_input == "q": break - await credential.close() + await connection.conversation.item.create( + item={ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": user_input}], + } + ) + await connection.response.create() + async for event in connection: + if event.type == "response.output_text.delta": + print(event.delta, flush=True, end="") + elif event.type == "response.output_text.done": + print() + elif event.type == "error": + raise RuntimeError(f"Realtime error: {event.error.type} ({event.error.code})") + elif event.type == "response.done": + if event.response.status != "completed": + raise RuntimeError(f"Response did not complete successfully: {event.response.status}") + break + else: + raise RuntimeError("WebSocket closed before the response completed.") asyncio.run(main()) diff --git a/examples/responses/multi_agent_websocket.py b/examples/responses/multi_agent_websocket.py index a6390419de..cb7d990fbd 100644 --- a/examples/responses/multi_agent_websocket.py +++ b/examples/responses/multi_agent_websocket.py @@ -1,6 +1,6 @@ from __future__ import annotations -from openai import OpenAI +from openai import OpenAI, OpenAIError from openai.types.beta import BetaResponseOutputItem PROMPT = """Compare these fictional proposals on speed, cost, reliability, and security. @@ -40,7 +40,13 @@ def agent_name(item: BetaResponseOutputItem) -> str: print(f"{separator}━━━ {role}: {name} ━━━\n") labeled_items.add(event.output_index) print(event.delta, end="", flush=True) - if event.type == "response.completed": + if event.type == "response.completed" and event.agent is None: break + if ( + event.type == "response.failed" or event.type == "response.incomplete" or event.type == "error" + ) and event.agent is None: + raise OpenAIError(f"Multi-agent request ended with {event.type}") + else: + raise OpenAIError("Connection closed before the multi-agent response completed") print() diff --git a/tests/lib/live/test_audio_transcript_example.py b/tests/lib/live/test_audio_transcript_example.py new file mode 100644 index 0000000000..e5846a93a4 --- /dev/null +++ b/tests/lib/live/test_audio_transcript_example.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import json +import wave +import asyncio +from pathlib import Path + +import pytest +from websockets.exceptions import ConnectionClosedOK +from websockets.sync.server import ServerConnection + +from openai import OpenAIError +from examples.live.audio_transcript import run + +from ..responses.test_websocket_session import script_server + + +@pytest.mark.parametrize("stage", ["startup", "audio", "requested"]) +@pytest.mark.parametrize( + "reason", ["close_requested", "expired", "content", "remote_hangup", "connection_lost", "error", "eof"] +) +async def test_audio_example_requires_its_requested_finalization( + stage: str, reason: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + wav_path = tmp_path / "fixture.wav" + with wave.open(str(wav_path), "wb") as wav: + wav.setparams((1, 2, 24000, 0, "NONE", "not compressed")) + wav.writeframes(bytes(24000 * 2 // 10)) + + sent_closed = False + received_close = False + audio_messages = 0 + + def script(socket: ServerConnection) -> None: + nonlocal sent_closed, received_close, audio_messages + + def send_closed() -> None: + nonlocal sent_closed + sent_closed = True + if reason == "eof": + socket.close() + elif reason == "error": + socket.send('{"type": "error", "event_id": "failure", "message": "Synthetic test error"}') + else: + socket.send(json.dumps({"type": "session.closed", "event_id": "done", "reason": reason})) + + assert json.loads(socket.recv(timeout=3))["type"] == "session.start" + if stage == "startup": + send_closed() + else: + socket.send('{"type": "session.started", "event_id": "started", "session": {"id": "fixture"}}') + # Leave the connection open after the terminal event. The client must + # reject that event immediately, rather than waiting for transport EOF. + try: + while True: + event = json.loads(socket.recv(timeout=3)) + if event["type"] == "session.input_audio.append": + audio_messages += 1 + if stage == "audio" and not sent_closed: + send_closed() + elif event["type"] == "session.close": + received_close = True + if stage == "requested": + # Final transcript data after close was requested must + # reach the caller before the confirmed result returns. + socket.send( + json.dumps( + { + "type": "session.output_transcript.delta", + "event_id": "text", + "delta": "Final fixture transcript.", + "start_ms": 0, + "end_ms": 100, + } + ) + ) + send_closed() + except ConnectionClosedOK: + return + + monkeypatch.setenv("OPENAI_API_KEY", "fake-transcript-fixture-key") + monkeypatch.setenv("NO_PROXY", "127.0.0.1") + monkeypatch.setenv("no_proxy", "127.0.0.1") + # The sample constructs its own client; restrict this fixture to loopback. + for name in ("ALL_PROXY", "all_proxy", "HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy"): + monkeypatch.delenv(name, raising=False) + with script_server(script) as url: + monkeypatch.setenv("OPENAI_BASE_URL", url) + if stage == "requested" and reason == "close_requested": + result = await asyncio.wait_for(run(wav_path, model="fixture-model", listen_seconds=0), timeout=1) + assert any(segment.text == "Final fixture transcript." for segment in result) + else: + # recv() can report transport EOF directly while awaiting startup; + # all terminal paths must fail instead of returning partial text. + with pytest.raises((OpenAIError, ConnectionClosedOK)): + await asyncio.wait_for(run(wav_path, model="fixture-model", listen_seconds=0), timeout=1) + assert sent_closed + assert received_close == (stage == "requested") + if stage == "startup": + assert audio_messages == 0 + elif stage == "requested": + assert audio_messages == 5 + else: + assert 1 <= audio_messages <= 5