from __future__ import annotations import os from typing import Any, Generic, Callable, Iterator, cast, overload from typing_extensions import Literal, TypeVar import rich import httpx import pytest from respx import MockRouter from pydantic import BaseModel from inline_snapshot import ( external, snapshot, outsource, # pyright: ignore[reportUnknownVariableType] ) import openai from openai import OpenAI, AsyncOpenAI from openai._utils import consume_sync_iterator, assert_signatures_in_sync from openai._compat import model_copy from openai.types.chat import ChatCompletionChunk from openai.lib.streaming.chat import ( ContentDoneEvent, ChatCompletionStream, ChatCompletionStreamEvent, ChatCompletionStreamState, ChatCompletionStreamManager, ParsedChatCompletionSnapshot, ) from openai.lib._parsing._completions import ResponseFormatT from ..utils import print_obj, get_snapshot_value from ...conftest import base_url _T = TypeVar("_T") # all the snapshots in this file are auto-generated from the live API # # you can update them with # # `OPENAI_LIVE=1 pytest --inline-snapshot=fix -p no:xdist -o addopts=""` @pytest.mark.respx(base_url=base_url) def test_parse_nothing(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "What's the weather like in SF?", }, ], ), content_snapshot=snapshot(external("e2aad469b71d*.bin")), mock_client=client, respx_mock=respx_mock, ) assert print_obj(listener.stream.get_final_completion().choices, monkeypatch) == snapshot( """\ [ ParsedChoice[NoneType]( finish_reason='stop', index=0, logprobs=None, message=ParsedChatCompletionMessage[NoneType]( annotations=None, audio=None, content="I'm unable to provide real-time weather updates. To get the current weather in San Francisco, I recommend checking a reliable weather website or a weather app.", function_call=None, parsed=None, refusal=None, role='assistant', tool_calls=None ) ) ] """ ) assert print_obj(listener.get_event_by_type("content.done"), monkeypatch) == snapshot( """\ ContentDoneEvent[NoneType]( content="I'm unable to provide real-time weather updates. To get the current weather in San Francisco, I recommend checking a reliable weather website or a weather app.", parsed=None, type='content.done' ) """ ) @pytest.mark.respx(base_url=base_url) def test_parse_pydantic_model(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: class Location(BaseModel): city: str temperature: float units: Literal["c", "f"] done_snapshots: list[ParsedChatCompletionSnapshot] = [] def on_event(stream: ChatCompletionStream[Location], event: ChatCompletionStreamEvent[Location]) -> None: if event.type == "content.done": done_snapshots.append(model_copy(stream.current_completion_snapshot, deep=True)) listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "What's the weather like in SF?", }, ], response_format=Location, ), content_snapshot=snapshot(external("7e5ea4d12e7c*.bin")), mock_client=client, respx_mock=respx_mock, on_event=on_event, ) assert len(done_snapshots) == 1 assert isinstance(done_snapshots[0].choices[0].message.parsed, Location) for event in reversed(listener.events): if event.type == "content.delta": data = cast(Any, event.parsed) assert isinstance(data["city"], str), data assert isinstance(data["temperature"], (int, float)), data assert isinstance(data["units"], str), data break else: rich.print(listener.events) raise AssertionError("Did not find a `content.delta` event") assert print_obj(listener.stream.get_final_completion(), monkeypatch) == snapshot( """\ ParsedChatCompletion[Location]( choices=[ ParsedChoice[Location]( finish_reason='stop', index=0, logprobs=None, message=ParsedChatCompletionMessage[Location]( annotations=None, audio=None, content='{"city":"San Francisco","temperature":61,"units":"f"}', function_call=None, parsed=Location(city='San Francisco', temperature=61.0, units='f'), refusal=None, role='assistant', tool_calls=None ) ) ], created=1727346169, id='chatcmpl-ABfw1e5abtU8OwGr15vOreYVb2MiF', model='gpt-4o-2024-08-06', object='chat.completion', service_tier=None, system_fingerprint='fp_5050236cbd', usage=CompletionUsage( completion_tokens=14, completion_tokens_details=CompletionTokensDetails( accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=0, rejected_prediction_tokens=None ), prompt_tokens=79, prompt_tokens_details=None, total_tokens=93 ) ) """ ) assert print_obj(listener.get_event_by_type("content.done"), monkeypatch) == snapshot( """\ ContentDoneEvent[Location]( content='{"city":"San Francisco","temperature":61,"units":"f"}', parsed=Location(city='San Francisco', temperature=61.0, units='f'), type='content.done' ) """ ) @pytest.mark.respx(base_url=base_url) def test_parse_pydantic_model_multiple_choices( client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch ) -> None: class Location(BaseModel): city: str temperature: float units: Literal["c", "f"] listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "What's the weather like in SF?", }, ], n=3, response_format=Location, ), content_snapshot=snapshot(external("a491adda08c3*.bin")), mock_client=client, respx_mock=respx_mock, ) assert [e.type for e in listener.events] == snapshot( [ "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.delta", "chunk", "content.done", "chunk", "content.done", "chunk", "content.done", "chunk", ] ) assert print_obj(listener.stream.get_final_completion().choices, monkeypatch) == snapshot( """\ [ ParsedChoice[Location]( finish_reason='stop', index=0, logprobs=None, message=ParsedChatCompletionMessage[Location]( annotations=None, audio=None, content='{"city":"San Francisco","temperature":65,"units":"f"}', function_call=None, parsed=Location(city='San Francisco', temperature=65.0, units='f'), refusal=None, role='assistant', tool_calls=None ) ), ParsedChoice[Location]( finish_reason='stop', index=1, logprobs=None, message=ParsedChatCompletionMessage[Location]( annotations=None, audio=None, content='{"city":"San Francisco","temperature":61,"units":"f"}', function_call=None, parsed=Location(city='San Francisco', temperature=61.0, units='f'), refusal=None, role='assistant', tool_calls=None ) ), ParsedChoice[Location]( finish_reason='stop', index=2, logprobs=None, message=ParsedChatCompletionMessage[Location]( annotations=None, audio=None, content='{"city":"San Francisco","temperature":59,"units":"f"}', function_call=None, parsed=Location(city='San Francisco', temperature=59.0, units='f'), refusal=None, role='assistant', tool_calls=None ) ) ] """ ) @pytest.mark.respx(base_url=base_url) def test_parse_max_tokens_reached(client: OpenAI, respx_mock: MockRouter) -> None: class Location(BaseModel): city: str temperature: float units: Literal["c", "f"] with pytest.raises(openai.LengthFinishReasonError): _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "What's the weather like in SF?", }, ], max_tokens=1, response_format=Location, ), content_snapshot=snapshot(external("4cc50a6135d2*.bin")), mock_client=client, respx_mock=respx_mock, ) @pytest.mark.respx(base_url=base_url) def test_parse_pydantic_model_refusal(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: class Location(BaseModel): city: str temperature: float units: Literal["c", "f"] listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "How do I make anthrax?", }, ], response_format=Location, ), content_snapshot=snapshot(external("173417d55340*.bin")), mock_client=client, respx_mock=respx_mock, ) assert print_obj(listener.get_event_by_type("refusal.done"), monkeypatch) == snapshot("""\ RefusalDoneEvent(refusal="I'm sorry, I can't assist with that request.", type='refusal.done') """) assert print_obj(listener.stream.get_final_completion().choices, monkeypatch) == snapshot( """\ [ ParsedChoice[Location]( finish_reason='stop', index=0, logprobs=None, message=ParsedChatCompletionMessage[Location]( annotations=None, audio=None, content=None, function_call=None, parsed=None, refusal="I'm sorry, I can't assist with that request.", role='assistant', tool_calls=None ) ) ] """ ) @pytest.mark.respx(base_url=base_url) def test_content_logprobs_events(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "Say foo", }, ], logprobs=True, ), content_snapshot=snapshot(external("83b060bae42e*.bin")), mock_client=client, respx_mock=respx_mock, ) assert print_obj([e for e in listener.events if e.type.startswith("logprobs")], monkeypatch) == snapshot("""\ [ LogprobsContentDeltaEvent( content=[ ChatCompletionTokenLogprob(bytes=[70, 111, 111], logprob=-0.0025094282, token='Foo', top_logprobs=[]) ], snapshot=[ ChatCompletionTokenLogprob(bytes=[70, 111, 111], logprob=-0.0025094282, token='Foo', top_logprobs=[]) ], type='logprobs.content.delta' ), LogprobsContentDeltaEvent( content=[ChatCompletionTokenLogprob(bytes=[33], logprob=-0.26638845, token='!', top_logprobs=[])], snapshot=[ ChatCompletionTokenLogprob(bytes=[70, 111, 111], logprob=-0.0025094282, token='Foo', top_logprobs=[]), ChatCompletionTokenLogprob(bytes=[33], logprob=-0.26638845, token='!', top_logprobs=[]) ], type='logprobs.content.delta' ), LogprobsContentDoneEvent( content=[ ChatCompletionTokenLogprob(bytes=[70, 111, 111], logprob=-0.0025094282, token='Foo', top_logprobs=[]), ChatCompletionTokenLogprob(bytes=[33], logprob=-0.26638845, token='!', top_logprobs=[]) ], type='logprobs.content.done' ) ] """) assert print_obj(listener.stream.get_final_completion().choices, monkeypatch) == snapshot("""\ [ ParsedChoice[NoneType]( finish_reason='stop', index=0, logprobs=ChoiceLogprobs( content=[ ChatCompletionTokenLogprob(bytes=[70, 111, 111], logprob=-0.0025094282, token='Foo', top_logprobs=[]), ChatCompletionTokenLogprob(bytes=[33], logprob=-0.26638845, token='!', top_logprobs=[]) ], refusal=None ), message=ParsedChatCompletionMessage[NoneType]( annotations=None, audio=None, content='Foo!', function_call=None, parsed=None, refusal=None, role='assistant', tool_calls=None ) ) ] """) @pytest.mark.respx(base_url=base_url) def test_refusal_logprobs_events(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: class Location(BaseModel): city: str temperature: float units: Literal["c", "f"] listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "How do I make anthrax?", }, ], logprobs=True, response_format=Location, ), content_snapshot=snapshot(external("569c877e6942*.bin")), mock_client=client, respx_mock=respx_mock, ) assert print_obj([e.type for e in listener.events if e.type.startswith("logprobs")], monkeypatch) == snapshot("""\ [ 'logprobs.refusal.delta', 'logprobs.refusal.delta', 'logprobs.refusal.delta', 'logprobs.refusal.delta', 'logprobs.refusal.delta', 'logprobs.refusal.delta', 'logprobs.refusal.delta', 'logprobs.refusal.delta', 'logprobs.refusal.delta', 'logprobs.refusal.delta', 'logprobs.refusal.delta', 'logprobs.refusal.done' ] """) assert print_obj(listener.stream.get_final_completion().choices, monkeypatch) == snapshot("""\ [ ParsedChoice[Location]( finish_reason='stop', index=0, logprobs=ChoiceLogprobs( content=None, refusal=[ ChatCompletionTokenLogprob(bytes=[73, 39, 109], logprob=-0.0012038043, token="I'm", top_logprobs=[]), ChatCompletionTokenLogprob( bytes=[32, 118, 101, 114, 121], logprob=-0.8438816, token=' very', top_logprobs=[] ), ChatCompletionTokenLogprob( bytes=[32, 115, 111, 114, 114, 121], logprob=-3.4121115e-06, token=' sorry', top_logprobs=[] ), ChatCompletionTokenLogprob(bytes=[44], logprob=-3.3809047e-05, token=',', top_logprobs=[]), ChatCompletionTokenLogprob( bytes=[32, 98, 117, 116], logprob=-0.038048144, token=' but', top_logprobs=[] ), ChatCompletionTokenLogprob(bytes=[32, 73], logprob=-0.0016109125, token=' I', top_logprobs=[]), ChatCompletionTokenLogprob( bytes=[32, 99, 97, 110, 39, 116], logprob=-0.0073532974, token=" can't", top_logprobs=[] ), ChatCompletionTokenLogprob( bytes=[32, 97, 115, 115, 105, 115, 116], logprob=-0.0020837625, token=' assist', top_logprobs=[] ), ChatCompletionTokenLogprob( bytes=[32, 119, 105, 116, 104], logprob=-0.00318354, token=' with', top_logprobs=[] ), ChatCompletionTokenLogprob( bytes=[32, 116, 104, 97, 116], logprob=-0.0017186158, token=' that', top_logprobs=[] ), ChatCompletionTokenLogprob(bytes=[46], logprob=-0.57687104, token='.', top_logprobs=[]) ] ), message=ParsedChatCompletionMessage[Location]( annotations=None, audio=None, content=None, function_call=None, parsed=None, refusal="I'm very sorry, but I can't assist with that.", role='assistant', tool_calls=None ) ) ] """) @pytest.mark.respx(base_url=base_url) def test_parse_pydantic_tool(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: class GetWeatherArgs(BaseModel): city: str country: str units: Literal["c", "f"] = "c" listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "What's the weather like in Edinburgh?", }, ], tools=[ openai.pydantic_function_tool(GetWeatherArgs), ], ), content_snapshot=snapshot(external("c6aa7e397b71*.bin")), mock_client=client, respx_mock=respx_mock, ) assert print_obj(listener.stream.current_completion_snapshot.choices, monkeypatch) == snapshot( """\ [ ParsedChoice[object]( finish_reason='tool_calls', index=0, logprobs=None, message=ParsedChatCompletionMessage[object]( annotations=None, audio=None, content=None, function_call=None, parsed=None, refusal=None, role='assistant', tool_calls=[ ParsedFunctionToolCall( function=ParsedFunction( arguments='{"city":"Edinburgh","country":"UK","units":"c"}', name='GetWeatherArgs', parsed_arguments=GetWeatherArgs(city='Edinburgh', country='UK', units='c') ), id='call_c91SqDXlYFuETYv8mUHzz6pp', index=0, type='function' ) ] ) ) ] """ ) assert print_obj(listener.stream.get_final_completion().choices, monkeypatch) == snapshot( """\ [ ParsedChoice[NoneType]( finish_reason='tool_calls', index=0, logprobs=None, message=ParsedChatCompletionMessage[NoneType]( annotations=None, audio=None, content=None, function_call=None, parsed=None, refusal=None, role='assistant', tool_calls=[ ParsedFunctionToolCall( function=ParsedFunction( arguments='{"city":"Edinburgh","country":"UK","units":"c"}', name='GetWeatherArgs', parsed_arguments=GetWeatherArgs(city='Edinburgh', country='UK', units='c') ), id='call_c91SqDXlYFuETYv8mUHzz6pp', index=0, type='function' ) ] ) ) ] """ ) @pytest.mark.respx(base_url=base_url) def test_parse_multiple_pydantic_tools(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: class GetWeatherArgs(BaseModel): """Get the temperature for the given country/city combo""" city: str country: str units: Literal["c", "f"] = "c" class GetStockPrice(BaseModel): ticker: str exchange: str listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "What's the weather like in Edinburgh?", }, { "role": "user", "content": "What's the price of AAPL?", }, ], tools=[ openai.pydantic_function_tool(GetWeatherArgs), openai.pydantic_function_tool( GetStockPrice, name="get_stock_price", description="Fetch the latest price for a given ticker" ), ], ), content_snapshot=snapshot(external("f82268f2fefd*.bin")), mock_client=client, respx_mock=respx_mock, ) assert print_obj(listener.stream.current_completion_snapshot.choices, monkeypatch) == snapshot( """\ [ ParsedChoice[object]( finish_reason='tool_calls', index=0, logprobs=None, message=ParsedChatCompletionMessage[object]( annotations=None, audio=None, content=None, function_call=None, parsed=None, refusal=None, role='assistant', tool_calls=[ ParsedFunctionToolCall( function=ParsedFunction( arguments='{"city": "Edinburgh", "country": "GB", "units": "c"}', name='GetWeatherArgs', parsed_arguments=GetWeatherArgs(city='Edinburgh', country='GB', units='c') ), id='call_JMW1whyEaYG438VE1OIflxA2', index=0, type='function' ), ParsedFunctionToolCall( function=ParsedFunction( arguments='{"ticker": "AAPL", "exchange": "NASDAQ"}', name='get_stock_price', parsed_arguments=GetStockPrice(exchange='NASDAQ', ticker='AAPL') ), id='call_DNYTawLBoN8fj3KN6qU9N1Ou', index=1, type='function' ) ] ) ) ] """ ) completion = listener.stream.get_final_completion() assert print_obj(completion.choices[0].message.tool_calls, monkeypatch) == snapshot( """\ [ ParsedFunctionToolCall( function=ParsedFunction( arguments='{"city": "Edinburgh", "country": "GB", "units": "c"}', name='GetWeatherArgs', parsed_arguments=GetWeatherArgs(city='Edinburgh', country='GB', units='c') ), id='call_JMW1whyEaYG438VE1OIflxA2', index=0, type='function' ), ParsedFunctionToolCall( function=ParsedFunction( arguments='{"ticker": "AAPL", "exchange": "NASDAQ"}', name='get_stock_price', parsed_arguments=GetStockPrice(exchange='NASDAQ', ticker='AAPL') ), id='call_DNYTawLBoN8fj3KN6qU9N1Ou', index=1, type='function' ) ] """ ) @pytest.mark.respx(base_url=base_url) def test_parse_strict_tools(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "What's the weather like in SF?", }, ], tools=[ { "type": "function", "function": { "name": "get_weather", "parameters": { "type": "object", "properties": { "city": {"type": "string"}, "state": {"type": "string"}, }, "required": [ "city", "state", ], "additionalProperties": False, }, "strict": True, }, } ], ), content_snapshot=snapshot(external("a247c49c5fcd*.bin")), mock_client=client, respx_mock=respx_mock, ) assert print_obj(listener.stream.current_completion_snapshot.choices, monkeypatch) == snapshot( """\ [ ParsedChoice[object]( finish_reason='tool_calls', index=0, logprobs=None, message=ParsedChatCompletionMessage[object]( annotations=None, audio=None, content=None, function_call=None, parsed=None, refusal=None, role='assistant', tool_calls=[ ParsedFunctionToolCall( function=ParsedFunction( arguments='{"city":"San Francisco","state":"CA"}', name='get_weather', parsed_arguments={'city': 'San Francisco', 'state': 'CA'} ), id='call_CTf1nWJLqSeRgDqaCG27xZ74', index=0, type='function' ) ] ) ) ] """ ) @pytest.mark.respx(base_url=base_url) def test_non_pydantic_response_format(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "What's the weather like in SF? Give me any JSON back", }, ], response_format={"type": "json_object"}, ), content_snapshot=snapshot(external("d61558011839*.bin")), mock_client=client, respx_mock=respx_mock, ) assert print_obj(listener.stream.get_final_completion().choices, monkeypatch) == snapshot( """\ [ ParsedChoice[NoneType]( finish_reason='stop', index=0, logprobs=None, message=ParsedChatCompletionMessage[NoneType]( annotations=None, audio=None, content='\\n {\\n "location": "San Francisco, CA",\\n "weather": {\\n "temperature": "18°C",\\n "condition": "Partly Cloudy",\\n "humidity": "72%",\\n "windSpeed": "15 km/h",\\n "windDirection": "NW"\\n },\\n "forecast": [\\n {\\n "day": "Monday",\\n "high": "20°C",\\n "low": "14°C",\\n "condition": "Sunny"\\n },\\n {\\n "day": "Tuesday",\\n "high": "19°C",\\n "low": "15°C",\\n "condition": "Mostly Cloudy"\\n },\\n {\\n "day": "Wednesday",\\n "high": "18°C",\\n "low": "14°C",\\n "condition": "Cloudy"\\n }\\n ]\\n }\\n', function_call=None, parsed=None, refusal=None, role='assistant', tool_calls=None ) ) ] """ ) @pytest.mark.respx(base_url=base_url) def test_allows_non_strict_tools_but_no_parsing( client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch ) -> None: listener = _make_stream_snapshot_request( lambda c: c.chat.completions.stream( model="gpt-4o-2024-08-06", messages=[{"role": "user", "content": "what's the weather in NYC?"}], tools=[ { "type": "function", "function": { "name": "get_weather", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}}, }, } ], ), content_snapshot=snapshot(external("2018feb66ae1*.bin")), mock_client=client, respx_mock=respx_mock, ) assert print_obj(listener.get_event_by_type("tool_calls.function.arguments.done"), monkeypatch) == snapshot("""\ FunctionToolCallArgumentsDoneEvent( arguments='{"city":"New York City"}', index=0, name='get_weather', parsed_arguments=None, type='tool_calls.function.arguments.done' ) """) assert print_obj(listener.stream.get_final_completion().choices, monkeypatch) == snapshot( """\ [ ParsedChoice[NoneType]( finish_reason='tool_calls', index=0, logprobs=None, message=ParsedChatCompletionMessage[NoneType]( annotations=None, audio=None, content=None, function_call=None, parsed=None, refusal=None, role='assistant', tool_calls=[ ParsedFunctionToolCall( function=ParsedFunction( arguments='{"city":"New York City"}', name='get_weather', parsed_arguments=None ), id='call_4XzlGBLtUe9dy3GVNV4jhq7h', index=0, type='function' ) ] ) ) ] """ ) @pytest.mark.respx(base_url=base_url) def test_chat_completion_state_helper(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: state = ChatCompletionStreamState() def streamer(client: OpenAI) -> Iterator[ChatCompletionChunk]: stream = client.chat.completions.create( model="gpt-4o-2024-08-06", messages=[ { "role": "user", "content": "What's the weather like in SF?", }, ], stream=True, ) for chunk in stream: state.handle_chunk(chunk) yield chunk _make_raw_stream_snapshot_request( streamer, content_snapshot=snapshot(external("e2aad469b71d*.bin")), mock_client=client, respx_mock=respx_mock, ) assert print_obj(state.get_final_completion().choices, monkeypatch) == snapshot( """\ [ ParsedChoice[NoneType]( finish_reason='stop', index=0, logprobs=None, message=ParsedChatCompletionMessage[NoneType]( annotations=None, audio=None, content="I'm unable to provide real-time weather updates. To get the current weather in San Francisco, I recommend checking a reliable weather website or a weather app.", function_call=None, parsed=None, refusal=None, role='assistant', tool_calls=None ) ) ] """ ) @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) def test_stream_method_in_sync(sync: bool, client: OpenAI, async_client: AsyncOpenAI) -> None: checking_client: OpenAI | AsyncOpenAI = client if sync else async_client assert_signatures_in_sync( checking_client.chat.completions.create, checking_client.chat.completions.stream, exclude_params={"response_format", "stream"}, ) class StreamListener(Generic[ResponseFormatT]): def __init__(self, stream: ChatCompletionStream[ResponseFormatT]) -> None: self.stream = stream self.events: list[ChatCompletionStreamEvent[ResponseFormatT]] = [] def __iter__(self) -> Iterator[ChatCompletionStreamEvent[ResponseFormatT]]: for event in self.stream: self.events.append(event) yield event @overload def get_event_by_type(self, event_type: Literal["content.done"]) -> ContentDoneEvent[ResponseFormatT] | None: ... @overload def get_event_by_type(self, event_type: str) -> ChatCompletionStreamEvent[ResponseFormatT] | None: ... def get_event_by_type(self, event_type: str) -> ChatCompletionStreamEvent[ResponseFormatT] | None: return next((e for e in self.events if e.type == event_type), None) def _make_stream_snapshot_request( func: Callable[[OpenAI], ChatCompletionStreamManager[ResponseFormatT]], *, content_snapshot: Any, respx_mock: MockRouter, mock_client: OpenAI, on_event: Callable[[ChatCompletionStream[ResponseFormatT], ChatCompletionStreamEvent[ResponseFormatT]], Any] | None = None, ) -> StreamListener[ResponseFormatT]: live = os.environ.get("OPENAI_LIVE") == "1" if live: def _on_response(response: httpx.Response) -> None: # update the content snapshot assert outsource(response.read()) == content_snapshot respx_mock.stop() client = OpenAI( http_client=httpx.Client( event_hooks={ "response": [_on_response], } ) ) else: respx_mock.post("/chat/completions").mock( return_value=httpx.Response( 200, content=get_snapshot_value(content_snapshot), headers={"content-type": "text/event-stream"}, ) ) client = mock_client with func(client) as stream: listener = StreamListener(stream) for event in listener: if on_event: on_event(stream, event) if live: client.close() return listener def _make_raw_stream_snapshot_request( func: Callable[[OpenAI], Iterator[ChatCompletionChunk]], *, content_snapshot: Any, respx_mock: MockRouter, mock_client: OpenAI, ) -> None: live = os.environ.get("OPENAI_LIVE") == "1" if live: def _on_response(response: httpx.Response) -> None: # update the content snapshot assert outsource(response.read()) == content_snapshot respx_mock.stop() client = OpenAI( http_client=httpx.Client( event_hooks={ "response": [_on_response], } ) ) else: respx_mock.post("/chat/completions").mock( return_value=httpx.Response( 200, content=get_snapshot_value(content_snapshot), headers={"content-type": "text/event-stream"}, ) ) client = mock_client stream = func(client) consume_sync_iterator(stream) if live: client.close()