wehub-resource-sync
1168 lines
39 KiB
Python
1168 lines
39 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections.abc import AsyncIterator
|
|
from typing import Any, cast
|
|
|
|
import httpx
|
|
import pytest
|
|
from openai import APIConnectionError, APIStatusError, AsyncOpenAI, omit
|
|
from openai.types.chat.chat_completion import ChatCompletion, Choice, ChoiceLogprobs
|
|
from openai.types.chat.chat_completion_chunk import ChatCompletionChunk
|
|
from openai.types.chat.chat_completion_message import ChatCompletionMessage
|
|
from openai.types.chat.chat_completion_message_custom_tool_call import (
|
|
ChatCompletionMessageCustomToolCall,
|
|
Custom,
|
|
)
|
|
from openai.types.chat.chat_completion_message_tool_call import ( # type: ignore[attr-defined]
|
|
ChatCompletionMessageFunctionToolCall,
|
|
Function,
|
|
)
|
|
from openai.types.chat.chat_completion_token_logprob import (
|
|
ChatCompletionTokenLogprob,
|
|
TopLogprob,
|
|
)
|
|
from openai.types.completion_usage import (
|
|
CompletionUsage,
|
|
PromptTokensDetails,
|
|
)
|
|
from openai.types.responses import (
|
|
Response,
|
|
ResponseFunctionToolCall,
|
|
ResponseOutputMessage,
|
|
ResponseOutputRefusal,
|
|
ResponseOutputText,
|
|
)
|
|
from openai.types.shared import Reasoning
|
|
|
|
from agents import (
|
|
Agent,
|
|
ModelResponse,
|
|
ModelRetryAdviceRequest,
|
|
ModelSettings,
|
|
ModelTracing,
|
|
OpenAIChatCompletionsModel,
|
|
OpenAIProvider,
|
|
Runner,
|
|
__version__,
|
|
generation_span,
|
|
)
|
|
from agents.exceptions import UserError
|
|
from agents.models._retry_runtime import provider_managed_retries_disabled
|
|
from agents.models.chatcmpl_helpers import HEADERS_OVERRIDE, ChatCmplHelpers
|
|
from agents.models.fake_id import FAKE_RESPONSES_ID
|
|
|
|
|
|
def _minimal_chat_completion(content: str = "ok") -> ChatCompletion:
|
|
return ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[
|
|
Choice(
|
|
index=0,
|
|
finish_reason="stop",
|
|
message=ChatCompletionMessage(role="assistant", content=content),
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
async def _run_chat_completions_model_with_custom_base_url(
|
|
model_settings: ModelSettings | None = None,
|
|
) -> dict[str, Any]:
|
|
class DummyCompletions:
|
|
def __init__(self) -> None:
|
|
self.kwargs: dict[str, Any] = {}
|
|
|
|
async def create(self, **kwargs: Any) -> Any:
|
|
self.kwargs = kwargs
|
|
return ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[
|
|
Choice(
|
|
index=0,
|
|
finish_reason="stop",
|
|
message=ChatCompletionMessage(role="assistant", content="ok"),
|
|
)
|
|
],
|
|
)
|
|
|
|
class DummyClient:
|
|
def __init__(self, completions: DummyCompletions) -> None:
|
|
self.chat = type("_Chat", (), {"completions": completions})()
|
|
self.base_url = httpx.URL("https://custom.example.test/v1/")
|
|
|
|
completions = DummyCompletions()
|
|
model = OpenAIChatCompletionsModel(
|
|
model="gpt-4",
|
|
openai_client=DummyClient(completions), # type: ignore[arg-type]
|
|
)
|
|
agent = Agent(name="test", model=model, model_settings=model_settings or ModelSettings())
|
|
|
|
await Runner.run(agent, "hi")
|
|
|
|
return completions.kwargs
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_with_text_message(monkeypatch) -> None:
|
|
"""
|
|
When the model returns a ChatCompletionMessage with plain text content,
|
|
`get_response` should produce a single `ResponseOutputMessage` containing
|
|
a `ResponseOutputText` with that content, and a `Usage` populated from
|
|
the completion's usage.
|
|
"""
|
|
msg = ChatCompletionMessage(role="assistant", content="Hello")
|
|
choice = Choice(index=0, finish_reason="stop", message=msg)
|
|
chat = ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[choice],
|
|
usage=CompletionUsage(
|
|
completion_tokens=5,
|
|
prompt_tokens=7,
|
|
total_tokens=12,
|
|
# completion_tokens_details left blank to test default
|
|
prompt_tokens_details=PromptTokensDetails.model_validate(
|
|
{"cached_tokens": 3, "cache_write_tokens": 4}
|
|
),
|
|
),
|
|
)
|
|
|
|
async def patched_fetch_response(self, *args, **kwargs):
|
|
return chat
|
|
|
|
monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
|
|
model = OpenAIProvider(use_responses=False).get_model("gpt-4")
|
|
resp: ModelResponse = await model.get_response(
|
|
system_instructions=None,
|
|
input="",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
# Should have produced exactly one output message with one text part
|
|
assert isinstance(resp, ModelResponse)
|
|
assert len(resp.output) == 1
|
|
assert isinstance(resp.output[0], ResponseOutputMessage)
|
|
msg_item = resp.output[0]
|
|
assert len(msg_item.content) == 1
|
|
assert isinstance(msg_item.content[0], ResponseOutputText)
|
|
assert msg_item.content[0].text == "Hello"
|
|
# Usage should be preserved from underlying ChatCompletion.usage
|
|
assert resp.usage.input_tokens == 7
|
|
assert resp.usage.output_tokens == 5
|
|
assert resp.usage.total_tokens == 12
|
|
assert resp.usage.input_tokens_details.cached_tokens == 3
|
|
assert getattr(resp.usage.input_tokens_details, "cache_write_tokens", None) == 4
|
|
assert resp.usage.output_tokens_details.reasoning_tokens == 0
|
|
assert resp.response_id is None
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("previous_response_id", "conversation_id", "expected_param"),
|
|
[
|
|
("resp_123", None, "previous_response_id"),
|
|
(None, "conv_123", "conversation_id"),
|
|
],
|
|
)
|
|
async def test_get_response_warns_and_ignores_server_managed_conversation_state_by_default(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
caplog: pytest.LogCaptureFixture,
|
|
previous_response_id: str | None,
|
|
conversation_id: str | None,
|
|
expected_param: str,
|
|
) -> None:
|
|
called = False
|
|
|
|
async def patched_fetch_response(self, *args, **kwargs):
|
|
nonlocal called
|
|
called = True
|
|
return _minimal_chat_completion()
|
|
|
|
monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
|
|
model = OpenAIProvider(use_responses=False).get_model("gpt-4")
|
|
caplog.set_level(logging.WARNING, logger="openai.agents")
|
|
|
|
await model.get_response(
|
|
system_instructions=None,
|
|
input="",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=previous_response_id,
|
|
conversation_id=conversation_id,
|
|
prompt=None,
|
|
)
|
|
|
|
assert expected_param in caplog.text
|
|
assert "Ignoring unsupported server-managed conversation state" in caplog.text
|
|
assert called is True
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_warns_and_ignores_prompt_by_default(
|
|
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
captured_prompt: Any = None
|
|
|
|
async def patched_fetch_response(self, *args, **kwargs):
|
|
nonlocal captured_prompt
|
|
captured_prompt = kwargs.get("prompt")
|
|
return _minimal_chat_completion()
|
|
|
|
monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
|
|
model = OpenAIProvider(use_responses=False).get_model("gpt-4")
|
|
caplog.set_level(logging.WARNING, logger="openai.agents")
|
|
|
|
await model.get_response(
|
|
system_instructions=None,
|
|
input="",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=cast(Any, {"id": "pmpt_123"}),
|
|
)
|
|
|
|
assert "Reusable prompts are only supported by the Responses API" in caplog.text
|
|
assert "Ignoring `prompt`" in caplog.text
|
|
assert captured_prompt is None
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("previous_response_id", "conversation_id", "expected_param"),
|
|
[
|
|
("resp_123", None, "previous_response_id"),
|
|
(None, "conv_123", "conversation_id"),
|
|
],
|
|
)
|
|
async def test_get_response_rejects_server_managed_conversation_state_in_strict_mode(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
previous_response_id: str | None,
|
|
conversation_id: str | None,
|
|
expected_param: str,
|
|
) -> None:
|
|
called = False
|
|
|
|
async def patched_fetch_response(self, *args, **kwargs):
|
|
nonlocal called
|
|
called = True
|
|
raise AssertionError("_fetch_response should not be called")
|
|
|
|
monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
|
|
model = OpenAIProvider(
|
|
use_responses=False,
|
|
strict_feature_validation=True,
|
|
).get_model("gpt-4")
|
|
|
|
with pytest.raises(UserError, match="server-managed conversation state") as exc_info:
|
|
await model.get_response(
|
|
system_instructions=None,
|
|
input="",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=previous_response_id,
|
|
conversation_id=conversation_id,
|
|
prompt=None,
|
|
)
|
|
|
|
assert expected_param in str(exc_info.value)
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_rejects_prompt_in_strict_mode(monkeypatch) -> None:
|
|
async def patched_fetch_response(self, *args, **kwargs):
|
|
raise AssertionError("_fetch_response should not run when prompt is unsupported")
|
|
|
|
monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
|
|
model = OpenAIProvider(
|
|
use_responses=False,
|
|
strict_feature_validation=True,
|
|
).get_model("gpt-4")
|
|
|
|
with pytest.raises(UserError, match="Reusable prompts"):
|
|
await model.get_response(
|
|
system_instructions=None,
|
|
input="",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=cast(Any, {"id": "pmpt_123"}),
|
|
)
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_rejects_non_text_tool_output_in_strict_mode() -> None:
|
|
class DummyCompletions:
|
|
async def create(self, **kwargs: Any) -> Any:
|
|
raise AssertionError("chat.completions.create should not run")
|
|
|
|
class DummyClient:
|
|
def __init__(self) -> None:
|
|
self.chat = type("_Chat", (), {"completions": DummyCompletions()})()
|
|
self.base_url = httpx.URL("http://fake")
|
|
|
|
model = OpenAIChatCompletionsModel(
|
|
model="gpt-4",
|
|
openai_client=DummyClient(), # type: ignore[arg-type]
|
|
strict_feature_validation=True,
|
|
)
|
|
|
|
with pytest.raises(UserError, match="cannot be empty or contain only non-text content"):
|
|
await model.get_response(
|
|
system_instructions=None,
|
|
input=[
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_image",
|
|
"output": [
|
|
{
|
|
"type": "input_image",
|
|
"image_url": "https://example.com/image.png",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_warns_and_sends_placeholder_for_non_text_tool_output(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
class DummyCompletions:
|
|
def __init__(self) -> None:
|
|
self.kwargs: dict[str, Any] = {}
|
|
|
|
async def create(self, **kwargs: Any) -> Any:
|
|
self.kwargs = kwargs
|
|
return _minimal_chat_completion()
|
|
|
|
class DummyClient:
|
|
def __init__(self) -> None:
|
|
self.completions = DummyCompletions()
|
|
self.chat = type("_Chat", (), {"completions": self.completions})()
|
|
self.base_url = httpx.URL("http://fake")
|
|
|
|
client = DummyClient()
|
|
model = OpenAIChatCompletionsModel(
|
|
model="gpt-4",
|
|
openai_client=client, # type: ignore[arg-type]
|
|
)
|
|
|
|
with caplog.at_level(logging.WARNING, logger="openai.agents"):
|
|
await model.get_response(
|
|
system_instructions=None,
|
|
input=[
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_image",
|
|
"output": [
|
|
{
|
|
"type": "input_image",
|
|
"image_url": "https://example.com/image.png",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
assert client.completions.kwargs["messages"] == [
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": "call_image",
|
|
"content": "[tool output omitted]",
|
|
}
|
|
]
|
|
assert "Replacing the tool output with a placeholder" in caplog.text
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_attaches_logprobs(monkeypatch) -> None:
|
|
msg = ChatCompletionMessage(role="assistant", content="Hi!")
|
|
choice = Choice(
|
|
index=0,
|
|
finish_reason="stop",
|
|
message=msg,
|
|
logprobs=ChoiceLogprobs(
|
|
content=[
|
|
ChatCompletionTokenLogprob(
|
|
token="Hi",
|
|
logprob=-0.5,
|
|
bytes=[1],
|
|
top_logprobs=[TopLogprob(token="Hi", logprob=-0.5, bytes=[1])],
|
|
),
|
|
ChatCompletionTokenLogprob(
|
|
token="!",
|
|
logprob=-0.1,
|
|
bytes=[2],
|
|
top_logprobs=[TopLogprob(token="!", logprob=-0.1, bytes=[2])],
|
|
),
|
|
]
|
|
),
|
|
)
|
|
chat = ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[choice],
|
|
usage=None,
|
|
)
|
|
|
|
async def patched_fetch_response(self, *args, **kwargs):
|
|
return chat
|
|
|
|
monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
|
|
model = OpenAIProvider(use_responses=False).get_model("gpt-4")
|
|
resp: ModelResponse = await model.get_response(
|
|
system_instructions=None,
|
|
input="",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
assert len(resp.output) == 1
|
|
assert isinstance(resp.output[0], ResponseOutputMessage)
|
|
text_part = resp.output[0].content[0]
|
|
assert isinstance(text_part, ResponseOutputText)
|
|
assert text_part.logprobs is not None
|
|
assert [lp.token for lp in text_part.logprobs] == ["Hi", "!"]
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_with_refusal(monkeypatch) -> None:
|
|
"""
|
|
When the model returns a ChatCompletionMessage with a `refusal` instead
|
|
of normal `content`, `get_response` should produce a single
|
|
`ResponseOutputMessage` containing a `ResponseOutputRefusal` part.
|
|
"""
|
|
msg = ChatCompletionMessage(role="assistant", refusal="No thanks")
|
|
choice = Choice(index=0, finish_reason="stop", message=msg)
|
|
chat = ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[choice],
|
|
usage=None,
|
|
)
|
|
|
|
async def patched_fetch_response(self, *args, **kwargs):
|
|
return chat
|
|
|
|
monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
|
|
model = OpenAIProvider(use_responses=False).get_model("gpt-4")
|
|
resp: ModelResponse = await model.get_response(
|
|
system_instructions=None,
|
|
input="",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
assert len(resp.output) == 1
|
|
assert isinstance(resp.output[0], ResponseOutputMessage)
|
|
refusal_part = resp.output[0].content[0]
|
|
assert isinstance(refusal_part, ResponseOutputRefusal)
|
|
assert refusal_part.refusal == "No thanks"
|
|
# With no usage from the completion, usage defaults to zeros.
|
|
assert resp.usage.requests == 0
|
|
assert resp.usage.input_tokens == 0
|
|
assert resp.usage.output_tokens == 0
|
|
assert resp.usage.input_tokens_details.cached_tokens == 0
|
|
assert resp.usage.output_tokens_details.reasoning_tokens == 0
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_with_tool_call(monkeypatch) -> None:
|
|
"""
|
|
If the ChatCompletionMessage includes one or more tool_calls, `get_response`
|
|
should append corresponding `ResponseFunctionToolCall` items after the
|
|
assistant message item with matching name/arguments.
|
|
"""
|
|
tool_call = ChatCompletionMessageFunctionToolCall(
|
|
id="call-id",
|
|
type="function",
|
|
function=Function(name="do_thing", arguments="{'x':1}"),
|
|
)
|
|
msg = ChatCompletionMessage(role="assistant", content="Hi", tool_calls=[tool_call])
|
|
choice = Choice(index=0, finish_reason="stop", message=msg)
|
|
chat = ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[choice],
|
|
usage=None,
|
|
)
|
|
|
|
async def patched_fetch_response(self, *args, **kwargs):
|
|
return chat
|
|
|
|
monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
|
|
model = OpenAIProvider(use_responses=False).get_model("gpt-4")
|
|
resp: ModelResponse = await model.get_response(
|
|
system_instructions=None,
|
|
input="",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
# Expect a message item followed by a function tool call item.
|
|
assert len(resp.output) == 2
|
|
assert isinstance(resp.output[0], ResponseOutputMessage)
|
|
fn_call_item = resp.output[1]
|
|
assert isinstance(fn_call_item, ResponseFunctionToolCall)
|
|
assert fn_call_item.call_id == "call-id"
|
|
assert fn_call_item.name == "do_thing"
|
|
assert fn_call_item.arguments == "{'x':1}"
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_rejects_custom_tool_call_in_strict_mode(monkeypatch) -> None:
|
|
tool_call = ChatCompletionMessageCustomToolCall(
|
|
id="tool1",
|
|
type="custom",
|
|
custom=Custom(name="raw_tool", input="payload"),
|
|
)
|
|
msg = ChatCompletionMessage(role="assistant", tool_calls=[tool_call])
|
|
chat = ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[Choice(index=0, finish_reason="tool_calls", message=msg)],
|
|
usage=None,
|
|
)
|
|
|
|
async def patched_fetch_response(self, *args, **kwargs):
|
|
return chat
|
|
|
|
monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
|
|
model = OpenAIProvider(use_responses=False, strict_feature_validation=True).get_model("gpt-4")
|
|
|
|
with pytest.raises(UserError, match="Custom tool calls are not supported"):
|
|
await model.get_response(
|
|
system_instructions=None,
|
|
input="",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
|
|
def test_get_client_disables_provider_managed_retries_on_runner_retry() -> None:
|
|
class DummyChatCompletionsClient:
|
|
def __init__(self) -> None:
|
|
self.base_url = httpx.URL("https://api.openai.com/v1/")
|
|
self.chat = type("ChatNamespace", (), {"completions": object()})()
|
|
self.with_options_calls: list[dict[str, Any]] = []
|
|
|
|
def with_options(self, **kwargs):
|
|
self.with_options_calls.append(kwargs)
|
|
return self
|
|
|
|
client = DummyChatCompletionsClient()
|
|
model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=client) # type: ignore[arg-type]
|
|
|
|
assert cast(object, model._get_client()) is client
|
|
with provider_managed_retries_disabled(True):
|
|
assert cast(object, model._get_client()) is client
|
|
|
|
assert client.with_options_calls == [{"max_retries": 0}]
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_with_no_message(monkeypatch) -> None:
|
|
"""If the model returns no message, get_response should return an empty output."""
|
|
msg = ChatCompletionMessage(role="assistant", content="ignored")
|
|
choice = Choice(index=0, finish_reason="content_filter", message=msg)
|
|
choice.message = None # type: ignore[assignment]
|
|
chat = ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[choice],
|
|
usage=None,
|
|
)
|
|
|
|
async def patched_fetch_response(self, *args, **kwargs):
|
|
return chat
|
|
|
|
monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
|
|
model = OpenAIProvider(use_responses=False).get_model("gpt-4")
|
|
resp: ModelResponse = await model.get_response(
|
|
system_instructions=None,
|
|
input="",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
assert resp.output == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_response_non_stream(monkeypatch) -> None:
|
|
"""
|
|
Verify that `_fetch_response` builds the correct OpenAI API call when not
|
|
streaming and returns the ChatCompletion object directly. We supply a
|
|
dummy ChatCompletion through a stubbed OpenAI client and inspect the
|
|
captured kwargs.
|
|
"""
|
|
|
|
# Dummy completions to record kwargs
|
|
class DummyCompletions:
|
|
def __init__(self) -> None:
|
|
self.kwargs: dict[str, Any] = {}
|
|
|
|
async def create(self, **kwargs: Any) -> Any:
|
|
self.kwargs = kwargs
|
|
return chat
|
|
|
|
class DummyClient:
|
|
def __init__(self, completions: DummyCompletions) -> None:
|
|
self.chat = type("_Chat", (), {"completions": completions})()
|
|
self.base_url = httpx.URL("http://fake")
|
|
|
|
msg = ChatCompletionMessage(role="assistant", content="ignored")
|
|
choice = Choice(index=0, finish_reason="stop", message=msg)
|
|
chat = ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[choice],
|
|
)
|
|
completions = DummyCompletions()
|
|
dummy_client = DummyClient(completions)
|
|
model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=dummy_client) # type: ignore
|
|
# Execute the private fetch with a system instruction and simple string input.
|
|
with generation_span(disabled=True) as span:
|
|
result = await model._fetch_response(
|
|
system_instructions="sys",
|
|
input="hi",
|
|
model_settings=ModelSettings(
|
|
reasoning=Reasoning(effort="xhigh"),
|
|
prompt_cache_retention="24h",
|
|
prompt_cache_options={"mode": "explicit", "ttl": "30m"},
|
|
),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
span=span,
|
|
tracing=ModelTracing.DISABLED,
|
|
stream=False,
|
|
)
|
|
assert result is chat
|
|
# Ensure expected args were passed through to OpenAI client.
|
|
kwargs = completions.kwargs
|
|
assert kwargs["stream"] is omit
|
|
assert kwargs["store"] is omit
|
|
assert kwargs["model"] == "gpt-4"
|
|
assert kwargs["messages"][0]["role"] == "system"
|
|
assert kwargs["messages"][0]["content"] == "sys"
|
|
assert kwargs["messages"][1]["role"] == "user"
|
|
# Defaults for optional fields become the omit sentinel
|
|
assert kwargs["tools"] is omit
|
|
assert kwargs["tool_choice"] is omit
|
|
assert kwargs["response_format"] is omit
|
|
assert kwargs["stream_options"] is omit
|
|
assert kwargs["reasoning_effort"] == "xhigh"
|
|
assert kwargs["prompt_cache_retention"] == "24h"
|
|
assert kwargs["prompt_cache_options"] == {"mode": "explicit", "ttl": "30m"}
|
|
|
|
|
|
def test_chat_completions_warns_once_for_responses_only_reasoning_settings(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
model = OpenAIChatCompletionsModel(
|
|
model="gpt-5.6-sol",
|
|
openai_client=cast(Any, object()),
|
|
)
|
|
model_settings = ModelSettings(
|
|
reasoning=Reasoning(mode="pro", effort="max", context="all_turns")
|
|
)
|
|
caplog.set_level(logging.WARNING, logger="openai.agents")
|
|
|
|
model._handle_unsupported_reasoning_settings(model_settings)
|
|
model._handle_unsupported_reasoning_settings(model_settings)
|
|
|
|
assert caplog.text.count("Ignoring unsupported reasoning settings") == 1
|
|
assert "reasoning.mode" in caplog.text
|
|
assert "reasoning.context" in caplog.text
|
|
|
|
|
|
def test_chat_completions_rejects_responses_only_reasoning_settings_in_strict_mode() -> None:
|
|
model = OpenAIChatCompletionsModel(
|
|
model="gpt-5.6-sol",
|
|
openai_client=cast(Any, object()),
|
|
strict_feature_validation=True,
|
|
)
|
|
|
|
with pytest.raises(UserError, match="reasoning.mode"):
|
|
model._handle_unsupported_reasoning_settings(
|
|
ModelSettings(reasoning=Reasoning(mode="pro", context="all_turns"))
|
|
)
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_custom_base_url_prompt_cache_key_uses_model_settings_only() -> None:
|
|
default_kwargs = await _run_chat_completions_model_with_custom_base_url()
|
|
explicit_kwargs = await _run_chat_completions_model_with_custom_base_url(
|
|
model_settings=ModelSettings(extra_args={"prompt_cache_key": "cache-key"})
|
|
)
|
|
|
|
assert "prompt_cache_key" not in default_kwargs
|
|
assert explicit_kwargs["prompt_cache_key"] == "cache-key"
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_extra_args_prompt_cache_options_allowed_when_direct_field_is_omitted() -> None:
|
|
prompt_cache_options = {"mode": "explicit", "ttl": "30m"}
|
|
|
|
kwargs = await _run_chat_completions_model_with_custom_base_url(
|
|
model_settings=ModelSettings(extra_args={"prompt_cache_options": prompt_cache_options})
|
|
)
|
|
|
|
assert kwargs["prompt_cache_options"] == prompt_cache_options
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_prompt_cache_options_rejected() -> None:
|
|
with pytest.raises(TypeError, match="multiple values.*prompt_cache_options"):
|
|
await _run_chat_completions_model_with_custom_base_url(
|
|
model_settings=ModelSettings(
|
|
prompt_cache_options={"mode": "explicit", "ttl": "30m"},
|
|
extra_args={"prompt_cache_options": {"mode": "implicit"}},
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
async def test_get_response_accepts_raw_chat_completions_image_content() -> None:
|
|
"""
|
|
Raw Chat Completions content parts should be accepted on the SDK input path
|
|
when using the Chat Completions backend.
|
|
"""
|
|
|
|
class DummyCompletions:
|
|
def __init__(self) -> None:
|
|
self.kwargs: dict[str, Any] = {}
|
|
|
|
async def create(self, **kwargs: Any) -> Any:
|
|
self.kwargs = kwargs
|
|
return chat
|
|
|
|
class DummyClient:
|
|
def __init__(self, completions: DummyCompletions) -> None:
|
|
self.chat = type("_Chat", (), {"completions": completions})()
|
|
self.base_url = httpx.URL("https://api.openai.com/v1/")
|
|
|
|
msg = ChatCompletionMessage(role="assistant", content="ok")
|
|
choice = Choice(index=0, finish_reason="stop", message=msg)
|
|
chat = ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[choice],
|
|
usage=None,
|
|
)
|
|
completions = DummyCompletions()
|
|
dummy_client = DummyClient(completions)
|
|
model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=dummy_client) # type: ignore[arg-type]
|
|
|
|
await model.get_response(
|
|
system_instructions=None,
|
|
input=[
|
|
# Cast the fixture because the raw chat-style alias is intentionally outside the
|
|
# canonical TypedDict shape that mypy expects for ordinary SDK inputs.
|
|
cast(
|
|
Any,
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "text", "text": "What is in this image?"},
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {
|
|
"url": "data:image/png;base64,AAAA",
|
|
"detail": "high",
|
|
},
|
|
},
|
|
],
|
|
},
|
|
)
|
|
],
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
assert completions.kwargs["messages"] == [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "text", "text": "What is in this image?"},
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {
|
|
"url": "data:image/png;base64,AAAA",
|
|
"detail": "high",
|
|
},
|
|
},
|
|
],
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_response_stream(monkeypatch) -> None:
|
|
"""
|
|
When `stream=True`, `_fetch_response` should return a bare `Response`
|
|
object along with the underlying async stream. The OpenAI client call
|
|
should include `stream_options` to request usage-delimited chunks.
|
|
"""
|
|
|
|
async def event_stream() -> AsyncIterator[ChatCompletionChunk]:
|
|
if False: # pragma: no cover
|
|
yield # pragma: no cover
|
|
|
|
class DummyCompletions:
|
|
def __init__(self) -> None:
|
|
self.kwargs: dict[str, Any] = {}
|
|
|
|
async def create(self, **kwargs: Any) -> Any:
|
|
self.kwargs = kwargs
|
|
return event_stream()
|
|
|
|
class DummyClient:
|
|
def __init__(self, completions: DummyCompletions) -> None:
|
|
self.chat = type("_Chat", (), {"completions": completions})()
|
|
self.base_url = httpx.URL("http://fake")
|
|
|
|
completions = DummyCompletions()
|
|
dummy_client = DummyClient(completions)
|
|
model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=dummy_client) # type: ignore
|
|
with generation_span(disabled=True) as span:
|
|
response, stream = await model._fetch_response(
|
|
system_instructions=None,
|
|
input="hi",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
span=span,
|
|
tracing=ModelTracing.DISABLED,
|
|
stream=True,
|
|
)
|
|
# Check OpenAI client was called for streaming
|
|
assert completions.kwargs["stream"] is True
|
|
assert completions.kwargs["store"] is omit
|
|
assert completions.kwargs["stream_options"] is omit
|
|
# Response is a proper openai Response
|
|
assert isinstance(response, Response)
|
|
assert response.id == FAKE_RESPONSES_ID
|
|
assert response.model == "gpt-4"
|
|
assert response.object == "response"
|
|
assert response.output == []
|
|
# We returned the async iterator produced by our dummy.
|
|
assert hasattr(stream, "__aiter__")
|
|
|
|
|
|
def test_store_param():
|
|
"""Should default to True for OpenAI API calls, and False otherwise."""
|
|
|
|
model_settings = ModelSettings()
|
|
client = AsyncOpenAI()
|
|
assert ChatCmplHelpers.get_store_param(client, model_settings) is True, (
|
|
"Should default to True for OpenAI API calls"
|
|
)
|
|
|
|
model_settings = ModelSettings(store=False)
|
|
assert ChatCmplHelpers.get_store_param(client, model_settings) is False, (
|
|
"Should respect explicitly set store=False"
|
|
)
|
|
|
|
model_settings = ModelSettings(store=True)
|
|
assert ChatCmplHelpers.get_store_param(client, model_settings) is True, (
|
|
"Should respect explicitly set store=True"
|
|
)
|
|
|
|
|
|
def test_clean_gemini_tool_call_id_removes_thought_suffix() -> None:
|
|
assert (
|
|
ChatCmplHelpers.clean_gemini_tool_call_id(
|
|
"call_123__thought__signature",
|
|
model="gemini-2.5-pro",
|
|
)
|
|
== "call_123"
|
|
)
|
|
|
|
|
|
def test_get_retry_advice_uses_openai_headers() -> None:
|
|
request = httpx.Request("POST", "https://api.openai.com/v1/chat/completions")
|
|
response = httpx.Response(
|
|
429,
|
|
request=request,
|
|
headers={
|
|
"x-should-retry": "true",
|
|
"retry-after-ms": "500",
|
|
"x-request-id": "req_123",
|
|
},
|
|
json={"error": {"code": "rate_limit"}},
|
|
)
|
|
error = APIStatusError(
|
|
"rate limited", response=response, body={"error": {"code": "rate_limit"}}
|
|
)
|
|
model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=cast(Any, object()))
|
|
|
|
advice = model.get_retry_advice(
|
|
ModelRetryAdviceRequest(
|
|
error=error,
|
|
attempt=1,
|
|
stream=False,
|
|
)
|
|
)
|
|
|
|
assert advice is not None
|
|
assert advice.suggested is True
|
|
assert advice.retry_after == 0.5
|
|
assert advice.replay_safety == "safe"
|
|
assert advice.normalized is not None
|
|
assert advice.normalized.error_code == "rate_limit"
|
|
assert advice.normalized.status_code == 429
|
|
assert advice.normalized.request_id == "req_123"
|
|
|
|
|
|
def test_get_retry_advice_keeps_stateful_transport_failures_ambiguous() -> None:
|
|
model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=cast(Any, object()))
|
|
error = APIConnectionError(
|
|
message="connection error",
|
|
request=httpx.Request("POST", "https://api.openai.com/v1/chat/completions"),
|
|
)
|
|
|
|
advice = model.get_retry_advice(
|
|
ModelRetryAdviceRequest(
|
|
error=error,
|
|
attempt=1,
|
|
stream=False,
|
|
previous_response_id="resp_prev",
|
|
)
|
|
)
|
|
|
|
assert advice is not None
|
|
assert advice.suggested is True
|
|
assert advice.replay_safety is None
|
|
assert advice.normalized is not None
|
|
assert advice.normalized.is_network_error is True
|
|
|
|
|
|
def test_get_retry_advice_marks_stateful_http_failures_replay_safe() -> None:
|
|
request = httpx.Request("POST", "https://api.openai.com/v1/chat/completions")
|
|
response = httpx.Response(
|
|
429,
|
|
request=request,
|
|
json={"error": {"code": "rate_limit"}},
|
|
)
|
|
error = APIStatusError(
|
|
"rate limited", response=response, body={"error": {"code": "rate_limit"}}
|
|
)
|
|
model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=cast(Any, object()))
|
|
|
|
advice = model.get_retry_advice(
|
|
ModelRetryAdviceRequest(
|
|
error=error,
|
|
attempt=1,
|
|
stream=False,
|
|
previous_response_id="resp_prev",
|
|
)
|
|
)
|
|
|
|
assert advice is not None
|
|
assert advice.suggested is True
|
|
assert advice.replay_safety == "safe"
|
|
assert advice.normalized is not None
|
|
assert advice.normalized.status_code == 429
|
|
|
|
|
|
def test_get_client_disables_provider_managed_retries_when_requested() -> None:
|
|
class DummyClient:
|
|
def __init__(self):
|
|
self.calls: list[dict[str, int]] = []
|
|
|
|
def with_options(self, **kwargs):
|
|
self.calls.append(kwargs)
|
|
return "retry-client"
|
|
|
|
client = DummyClient()
|
|
model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=cast(Any, client))
|
|
|
|
assert cast(object, model._get_client()) is client
|
|
|
|
with provider_managed_retries_disabled(True):
|
|
assert cast(object, model._get_client()) == "retry-client"
|
|
|
|
assert client.calls == [{"max_retries": 0}]
|
|
|
|
|
|
@pytest.mark.allow_call_model_methods
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("override_ua", [None, "test_user_agent"])
|
|
async def test_user_agent_header_chat_completions(override_ua):
|
|
called_kwargs: dict[str, Any] = {}
|
|
expected_ua = override_ua or f"Agents/Python {__version__}"
|
|
|
|
class DummyCompletions:
|
|
async def create(self, **kwargs):
|
|
nonlocal called_kwargs
|
|
called_kwargs = kwargs
|
|
msg = ChatCompletionMessage(role="assistant", content="Hello")
|
|
choice = Choice(index=0, finish_reason="stop", message=msg)
|
|
return ChatCompletion(
|
|
id="resp-id",
|
|
created=0,
|
|
model="fake",
|
|
object="chat.completion",
|
|
choices=[choice],
|
|
usage=None,
|
|
)
|
|
|
|
class DummyChatClient:
|
|
def __init__(self):
|
|
self.chat = type("_Chat", (), {"completions": DummyCompletions()})()
|
|
self.base_url = "https://api.openai.com"
|
|
|
|
model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=DummyChatClient()) # type: ignore
|
|
|
|
if override_ua is not None:
|
|
token = HEADERS_OVERRIDE.set({"User-Agent": override_ua})
|
|
else:
|
|
token = None
|
|
|
|
try:
|
|
await model.get_response(
|
|
system_instructions=None,
|
|
input="hi",
|
|
model_settings=ModelSettings(),
|
|
tools=[],
|
|
output_schema=None,
|
|
handoffs=[],
|
|
tracing=ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
)
|
|
finally:
|
|
if token is not None:
|
|
HEADERS_OVERRIDE.reset(token)
|
|
|
|
assert "extra_headers" in called_kwargs
|
|
assert called_kwargs["extra_headers"]["User-Agent"] == expected_ua
|
|
|
|
client = AsyncOpenAI(base_url="http://www.notopenai.com")
|
|
model_settings = ModelSettings()
|
|
assert ChatCmplHelpers.get_store_param(client, model_settings) is None, (
|
|
"Should default to None for non-OpenAI API calls"
|
|
)
|
|
|
|
model_settings = ModelSettings(store=False)
|
|
assert ChatCmplHelpers.get_store_param(client, model_settings) is False, (
|
|
"Should respect explicitly set store=False"
|
|
)
|
|
|
|
model_settings = ModelSettings(store=True)
|
|
assert ChatCmplHelpers.get_store_param(client, model_settings) is True, (
|
|
"Should respect explicitly set store=True"
|
|
)
|