diff --git a/go/api/v1alpha2/modelconfig_types.go b/go/api/v1alpha2/modelconfig_types.go index 19cc53e4f0..78c50bf46a 100644 --- a/go/api/v1alpha2/modelconfig_types.go +++ b/go/api/v1alpha2/modelconfig_types.go @@ -215,7 +215,7 @@ type OpenAIConfig struct { TokenExchange *TokenExchangeConfig `json:"tokenExchange,omitempty"` } -// OpenAIAPIFormat selects the OpenAI HTTP API shape used by the Go ADK runtime. +// OpenAIAPIFormat selects the OpenAI HTTP API shape used by the ADK runtime. // +kubebuilder:validation:Enum=chatCompletions;responses type OpenAIAPIFormat string diff --git a/python/packages/kagent-adk/src/kagent/adk/models/_openai.py b/python/packages/kagent-adk/src/kagent/adk/models/_openai.py index 9d26157775..56d160782a 100644 --- a/python/packages/kagent-adk/src/kagent/adk/models/_openai.py +++ b/python/packages/kagent-adk/src/kagent/adk/models/_openai.py @@ -2,6 +2,7 @@ import base64 import json +import logging import os from functools import cached_property from typing import TYPE_CHECKING, Any, AsyncGenerator, Iterable, Literal, Optional @@ -29,8 +30,24 @@ from openai.types.chat.chat_completion_message_tool_call_param import ( Function as ToolCallFunction, ) +from openai.types.responses import ( + EasyInputMessageParam, + FunctionToolParam, + Response, + ResponseFunctionToolCall, + ResponseFunctionToolCallParam, + ResponseInputImageParam, + ResponseInputItemParam, + ResponseInputTextParam, + ResponseOutputMessage, + ResponseUsage, +) +from openai.types.responses.response_input_item_param import FunctionCallOutput +from openai.types.responses.response_input_message_content_list_param import ( + ResponseInputMessageContentListParam, +) from openai.types.shared_params import FunctionDefinition, FunctionParameters -from pydantic import Field +from pydantic import Field, model_validator from ._ssl import KAgentTLSMixin from ._token_source import GDCHTokenSource @@ -38,6 +55,15 @@ if TYPE_CHECKING: from google.adk.models.llm_request import LlmRequest +logger = logging.getLogger(__name__) + +# OpenAI API format (ModelConfig openAI.apiFormat); chatCompletions is the default. +OpenAIAPIFormat = Literal["chatCompletions", "responses"] +OPENAI_API_FORMAT_RESPONSES = "responses" + +# Sampling parameters the Responses API does not accept. +_RESPONSES_UNSUPPORTED_PARAMS = ("frequency_penalty", "presence_penalty", "n", "seed") + def _convert_role_to_openai(role: Optional[str]) -> str: """Convert google.genai role to OpenAI role.""" @@ -89,6 +115,41 @@ def _thought_signatures_by_tool_call_id(contents: list[types.Content]) -> dict[s return thought_signatures +def _function_responses_by_tool_call_id(contents: list[types.Content]) -> dict[str, FunctionResponse]: + """Index function responses by tool call id.""" + function_responses: dict[str, FunctionResponse] = {} + for content in contents: + for part in content.parts or []: + if part.function_response: + function_responses[part.function_response.id or "call_1"] = part.function_response + + return function_responses + + +def _partition_content_parts( + content: types.Content, +) -> tuple[list[str], list[FunctionCall], list[types.Blob]]: + """Split a Content's parts into text, function calls and inline image blobs.""" + text_parts: list[str] = [] + function_calls: list[FunctionCall] = [] + image_blobs: list[types.Blob] = [] + + for part in content.parts or []: + if part.text: + text_parts.append(part.text) + elif part.function_call: + function_calls.append(part.function_call) + elif ( + part.inline_data + and part.inline_data.data + and part.inline_data.mime_type + and part.inline_data.mime_type.startswith("image") + ): + image_blobs.append(part.inline_data) + + return text_parts, function_calls, image_blobs + + def _build_function_call_part( *, name: str, @@ -116,6 +177,24 @@ def _build_function_call_part( return part +def _blob_data_uri(blob: types.Blob) -> str: + """Render an inline data blob as a base64 data URI.""" + return f"data:{blob.mime_type};base64,{base64.b64encode(blob.data or b'').decode()}" + + +def _extract_function_response_content(func_response: FunctionResponse) -> str: + """Extract text content from a genai FunctionResponse for the model to consume.""" + if isinstance(func_response.response, str): + return func_response.response + if func_response.response and "content" in func_response.response: + content_list = func_response.response["content"] + if len(content_list) > 0: + return "\n".join(item["text"] for item in content_list if "text" in item) + elif func_response.response and "result" in func_response.response: + return str(func_response.response["result"]) + return "" + + def _convert_content_to_openai_messages( contents: list[types.Content], system_instruction: Optional[str] = None ) -> list[ChatCompletionMessageParam]: @@ -128,40 +207,18 @@ def _convert_content_to_openai_messages( messages.append(system_message) # First pass: collect all function responses to match with tool calls - all_function_responses: dict[str, FunctionResponse] = {} + all_function_responses = _function_responses_by_tool_call_id(contents) thought_signatures = _thought_signatures_by_tool_call_id(contents) - for content in contents: - for part in content.parts or []: - if part.function_response: - tool_call_id = part.function_response.id or "call_1" - all_function_responses[tool_call_id] = part.function_response for content in contents: role = _convert_role_to_openai(content.role) - # Separate different types of parts - text_parts: list[str] = [] - function_calls: list[FunctionCall] = [] - function_responses: list[FunctionResponse] = [] - image_parts = [] + text_parts, function_calls, image_blobs = _partition_content_parts(content) + image_parts: list[ChatCompletionContentPartImageParam] = [ + {"type": "image_url", "image_url": {"url": _blob_data_uri(blob)}} for blob in image_blobs + ] - for part in content.parts or []: - if part.text: - text_parts.append(part.text) - elif part.function_call: - function_calls.append(part.function_call) - elif part.function_response: - function_responses.append(part.function_response) - elif part.inline_data and part.inline_data.mime_type and part.inline_data.mime_type.startswith("image"): - if part.inline_data.data: - image_data = base64.b64encode(part.inline_data.data).decode() - image_part: ChatCompletionContentPartImageParam = { - "type": "image_url", - "image_url": {"url": f"data:{part.inline_data.mime_type};base64,{image_data}"}, - } - image_parts.append(image_part) - - # Function responses are now handled together with function calls + # Function responses are handled together with function calls # This ensures proper pairing and prevents orphaned tool messages # Handle function calls (assistant messages with tool_calls) @@ -186,21 +243,10 @@ def _convert_content_to_openai_messages( # Check if we have a response for this tool call if tool_call_id in all_function_responses: - func_response = all_function_responses[tool_call_id] - content = "" - if isinstance(func_response.response, str): - content = func_response.response - elif func_response.response and "content" in func_response.response: - content_list = func_response.response["content"] - if len(content_list) > 0: - content = "\n".join(item["text"] for item in content_list if "text" in item) - elif func_response.response and "result" in func_response.response: - content = func_response.response["result"] - tool_message = ChatCompletionToolMessageParam( role="tool", tool_call_id=tool_call_id, - content=content, + content=_extract_function_response_content(all_function_responses[tool_call_id]), ) if extra_content := _openai_extra_content_for_thought_signature( thought_signatures.get(tool_call_id) @@ -278,38 +324,43 @@ def _update_type_string(value_dict: dict[str, Any]): _update_type_string(value) -def _convert_tools_to_openai(tools: list[types.Tool]) -> list[ChatCompletionToolParam]: - """Convert google.genai Tools to OpenAI tools format.""" - openai_tools: list[ChatCompletionToolParam] = [] - +def _iter_function_declarations(tools: list[types.Tool]) -> Iterable[types.FunctionDeclaration]: + """Yield every function declaration across a list of genai Tools.""" for tool in tools: - if tool.function_declarations: - for func_decl in tool.function_declarations: - # Build function definition - function_def = FunctionDefinition( - name=func_decl.name or "", - description=func_decl.description or "", - ) + for func_decl in tool.function_declarations or []: + yield func_decl - # Always include parameters field, even if empty - properties = {} - required = [] - if func_decl.parameters: - if func_decl.parameters.properties: - for prop_name, prop_schema in func_decl.parameters.properties.items(): - value_dict = prop_schema.model_dump(exclude_none=True) - _update_type_string(value_dict) - properties[prop_name] = value_dict +def _function_parameters_schema(func_decl: types.FunctionDeclaration) -> FunctionParameters: + """Build the JSON schema parameters object for a function declaration.""" + properties: dict[str, Any] = {} + required: list[str] = [] - if func_decl.parameters.required: - required = func_decl.parameters.required + if func_decl.parameters: + if func_decl.parameters.properties: + for prop_name, prop_schema in func_decl.parameters.properties.items(): + value_dict = prop_schema.model_dump(exclude_none=True) + _update_type_string(value_dict) + properties[prop_name] = value_dict - function_def["parameters"] = {"type": "object", "properties": properties, "required": required} + if func_decl.parameters.required: + required = func_decl.parameters.required + + # Always include the parameters field, even if empty. + return {"type": "object", "properties": properties, "required": required} + + +def _convert_tools_to_openai(tools: list[types.Tool]) -> list[ChatCompletionToolParam]: + """Convert google.genai Tools to OpenAI tools format.""" + openai_tools: list[ChatCompletionToolParam] = [] - # Create the tool param - openai_tool = ChatCompletionToolParam(type="function", function=function_def) - openai_tools.append(openai_tool) + for func_decl in _iter_function_declarations(tools): + function_def = FunctionDefinition( + name=func_decl.name or "", + description=func_decl.description or "", + parameters=_function_parameters_schema(func_decl), + ) + openai_tools.append(ChatCompletionToolParam(type="function", function=function_def)) return openai_tools @@ -365,12 +416,183 @@ def _convert_openai_response_to_llm_response(response: ChatCompletion) -> LlmRes return LlmResponse(content=content, usage_metadata=usage_metadata, finish_reason=finish_reason) +def _extract_system_instruction(llm_request: LlmRequest) -> Optional[str]: + """Extract the system instruction text from an LlmRequest's config, if any.""" + if not (llm_request.config and llm_request.config.system_instruction): + return None + + system_instruction = llm_request.config.system_instruction + if isinstance(system_instruction, str): + return system_instruction + if hasattr(system_instruction, "parts"): + text_parts = [] + parts = getattr(system_instruction, "parts", []) + if parts: + for part in parts: + if hasattr(part, "text") and part.text: + text_parts.append(part.text) + return "\n".join(text_parts) + return None + + +def _convert_content_to_responses_input(contents: list[types.Content]) -> list[ResponseInputItemParam]: + """Convert google.genai Content list to OpenAI Responses API input items.""" + input_items: list[ResponseInputItemParam] = [] + + # First pass: collect all function responses to match with tool calls + all_function_responses = _function_responses_by_tool_call_id(contents) + + for content in contents: + role = _convert_role_to_openai(content.role) + if role == "system": + continue + + text_parts, function_calls, image_blobs = _partition_content_parts(content) + image_parts = [ + ResponseInputImageParam(type="input_image", detail="auto", image_url=_blob_data_uri(blob)) + for blob in image_blobs + ] + + # Handle function calls (assistant tool calls + their outputs) + if function_calls and role == "assistant": + if text_parts: + input_items.append(EasyInputMessageParam(role="assistant", content="\n".join(text_parts))) + + for func_call in function_calls: + tool_call_id = func_call.id or "call_1" + input_items.append( + ResponseFunctionToolCallParam( + type="function_call", + call_id=tool_call_id, + name=func_call.name or "", + arguments=json.dumps(func_call.args) if func_call.args else "{}", + ) + ) + + if tool_call_id in all_function_responses: + output = _extract_function_response_content(all_function_responses[tool_call_id]) + else: + # If no response is available, create a placeholder response + # This prevents the OpenAI API error + output = "No response available for this function call." + input_items.append(FunctionCallOutput(type="function_call_output", call_id=tool_call_id, output=output)) + continue + + # Handle regular text/image messages (only if no function calls) + if text_parts or image_parts: + if image_parts: + message_content: ResponseInputMessageContentListParam = [ + ResponseInputTextParam(type="input_text", text=t) for t in text_parts + ] + message_content.extend(image_parts) + input_items.append(EasyInputMessageParam(role=role, content=message_content)) + else: + input_items.append(EasyInputMessageParam(role=role, content="\n".join(text_parts))) + + return input_items + + +def _convert_tools_to_responses(tools: list[types.Tool]) -> list[FunctionToolParam]: + """Convert google.genai Tools to OpenAI Responses API tools format.""" + responses_tools: list[FunctionToolParam] = [] + + for func_decl in _iter_function_declarations(tools): + responses_tools.append( + FunctionToolParam( + type="function", + name=func_decl.name or "", + description=func_decl.description or "", + parameters=_function_parameters_schema(func_decl), + strict=False, + ) + ) + + return responses_tools + + +def _responses_usage_to_genai( + usage: Optional[ResponseUsage], +) -> Optional[types.GenerateContentResponseUsageMetadata]: + """Convert a Responses API usage block to genai usage metadata.""" + if usage is None: + return None + return types.GenerateContentResponseUsageMetadata( + prompt_token_count=usage.input_tokens, + candidates_token_count=usage.output_tokens, + total_token_count=usage.total_tokens, + ) + + +def _responses_status_to_finish_reason(status: Optional[str]) -> types.FinishReason: + """Map a Responses API response status to a genai finish reason.""" + if status == "incomplete": + return types.FinishReason.MAX_TOKENS + if status == "failed": + return types.FinishReason.OTHER + return types.FinishReason.STOP + + +def _responses_error_message(code: Optional[str], message: Optional[str]) -> str: + """Render a Responses API error, keeping the upstream code when present.""" + message = message or "OpenAI responses request failed" + return f"{code}: {message}" if code else message + + +def _responses_failure_llm_response(response: Optional[Response]) -> Optional[LlmResponse]: + """Build an error LlmResponse for a failed Responses API response, else None.""" + if response is None or response.status != "failed": + return None + + error = response.error + return LlmResponse( + error_code="API_ERROR", + error_message=_responses_error_message( + getattr(error, "code", None), + getattr(error, "message", None), + ), + ) + + +def _responses_output_message_texts(item: ResponseOutputMessage) -> list[str]: + """Collect the text content of a Responses API output message.""" + return [text for output_content in item.content if (text := getattr(output_content, "text", None))] + + +def _responses_tool_call_part(item: ResponseFunctionToolCall) -> types.Part: + """Build a genai function-call part from a Responses API tool call.""" + try: + args = json.loads(item.arguments) if item.arguments else {} + except json.JSONDecodeError: + args = {} + return _build_function_call_part(name=item.name, args=args, tool_call_id=item.call_id or item.id or "call_1") + + +def _convert_responses_output_to_llm_response(response: Response) -> LlmResponse: + """Convert an OpenAI Responses API response to LlmResponse.""" + parts: list[types.Part] = [] + + for item in response.output: + if isinstance(item, ResponseOutputMessage): + parts.extend(types.Part.from_text(text=text) for text in _responses_output_message_texts(item)) + elif isinstance(item, ResponseFunctionToolCall): + parts.append(_responses_tool_call_part(item)) + + content = types.Content(role="model", parts=parts) + + return LlmResponse( + content=content, + usage_metadata=_responses_usage_to_genai(response.usage), + finish_reason=_responses_status_to_finish_reason(response.status), + ) + + class BaseOpenAI(KAgentTLSMixin, BaseLlm): """Base class for OpenAI-compatible models.""" model: str api_key: Optional[str] = Field(default=None, exclude=True) base_url: Optional[str] = None + api_format: Optional[OpenAIAPIFormat] = None frequency_penalty: Optional[float] = None default_headers: Optional[dict[str, str]] = None max_tokens: Optional[int] = None @@ -389,6 +611,18 @@ class BaseOpenAI(KAgentTLSMixin, BaseLlm): # GDCH token exchange: refreshes a short-lived bearer token before each model call. token_exchange: Optional[GDCHTokenSource] = Field(default=None, exclude=True) + @model_validator(mode="after") + def _warn_on_ignored_responses_params(self) -> "BaseOpenAI": + if self.api_format == OPENAI_API_FORMAT_RESPONSES: + ignored = [name for name in _RESPONSES_UNSUPPORTED_PARAMS if getattr(self, name) is not None] + if ignored: + logger.warning( + "Ignoring %s for model %s: not supported by the OpenAI Responses API", + ", ".join(ignored), + self.model, + ) + return self + def set_passthrough_key(self, token: str) -> None: if self.api_key != token: self.api_key = token @@ -437,20 +671,20 @@ async def generate_content_async( yield LlmResponse(error_message=f"Failed to refresh token-exchange credential: {exc}") return + if self.api_format == OPENAI_API_FORMAT_RESPONSES: + generator = self._generate_content_responses_async(llm_request, stream) + else: + generator = self._generate_content_completions_async(llm_request, stream) + + async for response in generator: + yield response + + async def _generate_content_completions_async( + self, llm_request: LlmRequest, stream: bool + ) -> AsyncGenerator[LlmResponse, None]: + """Generate content using the OpenAI Chat Completions API (/v1/chat/completions).""" # Convert messages - system_instruction = None - if llm_request.config and llm_request.config.system_instruction: - if isinstance(llm_request.config.system_instruction, str): - system_instruction = llm_request.config.system_instruction - elif hasattr(llm_request.config.system_instruction, "parts"): - # Handle Content type system instruction - text_parts = [] - parts = getattr(llm_request.config.system_instruction, "parts", []) - if parts: - for part in parts: - if hasattr(part, "text") and part.text: - text_parts.append(part.text) - system_instruction = "\n".join(text_parts) + system_instruction = _extract_system_instruction(llm_request) messages = _convert_content_to_openai_messages(llm_request.contents, system_instruction) @@ -607,6 +841,112 @@ async def generate_content_async( except Exception as e: yield LlmResponse(error_code="API_ERROR", error_message=str(e)) + async def _generate_content_responses_async( + self, llm_request: LlmRequest, stream: bool + ) -> AsyncGenerator[LlmResponse, None]: + """Generate content using the OpenAI Responses API (/v1/responses).""" + system_instruction = _extract_system_instruction(llm_request) + input_items = _convert_content_to_responses_input(llm_request.contents) + + kwargs: dict[str, Any] = { + "model": llm_request.model or self.model, + "input": input_items, + } + if system_instruction: + kwargs["instructions"] = system_instruction + + if self.temperature is not None: + kwargs["temperature"] = self.temperature + # Responses uses max_output_tokens (same semantics as max_completion_tokens). + if self.max_completion_tokens: + kwargs["max_output_tokens"] = self.max_completion_tokens + elif self.max_tokens: + kwargs["max_output_tokens"] = self.max_tokens + if self.top_p is not None: + kwargs["top_p"] = self.top_p + if self.reasoning_effort is not None: + kwargs["reasoning"] = {"effort": self.reasoning_effort} + + if llm_request.config and llm_request.config.tools: + genai_tools = [tool for tool in llm_request.config.tools if hasattr(tool, "function_declarations")] + if genai_tools: + responses_tools = _convert_tools_to_responses(genai_tools) + if responses_tools: + kwargs["tools"] = responses_tools + kwargs["tool_choice"] = "auto" + + try: + if stream: + aggregated_text = "" + status: Optional[str] = None + usage: Optional[ResponseUsage] = None + tool_calls: dict[str, types.Part] = {} + tool_call_order: list[str] = [] + + async for event in await self._client.responses.create(stream=True, **kwargs): + event_type = getattr(event, "type", None) + if event_type == "response.output_text.delta": + delta = event.delta + if not delta: + continue + aggregated_text += delta + yield LlmResponse( + content=types.Content(role="model", parts=[types.Part.from_text(text=delta)]), + partial=True, + turn_complete=False, + ) + elif event_type == "response.output_item.done": + item = event.item + if isinstance(item, ResponseFunctionToolCall): + call_id = item.call_id or item.id or "call_1" + if call_id not in tool_calls: + tool_call_order.append(call_id) + tool_calls[call_id] = _responses_tool_call_part(item) + elif isinstance(item, ResponseOutputMessage) and not aggregated_text: + # Endpoints that emit no text deltas still report the text here. + aggregated_text += "".join(_responses_output_message_texts(item)) + elif event_type == "response.completed": + usage = event.response.usage + status = event.response.status + elif event_type == "response.incomplete": + usage = event.response.usage + status = "incomplete" + elif event_type == "response.failed": + yield _responses_failure_llm_response(getattr(event, "response", None)) or LlmResponse( + error_code="API_ERROR", error_message="OpenAI responses stream failed" + ) + return + elif event_type == "error": + # Bare error events carry the message directly, not a Response. + yield LlmResponse( + error_code="API_ERROR", + error_message=_responses_error_message( + getattr(event, "code", None), + getattr(event, "message", None), + ), + ) + return + + final_parts: list[types.Part] = [] + if aggregated_text: + final_parts.append(types.Part.from_text(text=aggregated_text)) + for call_id in tool_call_order: + final_parts.append(tool_calls[call_id]) + + yield LlmResponse( + content=types.Content(role="model", parts=final_parts), + partial=False, + turn_complete=True, + finish_reason=_responses_status_to_finish_reason(status), + usage_metadata=_responses_usage_to_genai(usage), + ) + else: + response = await self._client.responses.create(stream=False, **kwargs) + yield _responses_failure_llm_response(response) or _convert_responses_output_to_llm_response(response) + + except Exception as e: + yield LlmResponse(error_code="API_ERROR", error_message=str(e)) + class OpenAI(BaseOpenAI): """OpenAI model implementation.""" diff --git a/python/packages/kagent-adk/src/kagent/adk/types.py b/python/packages/kagent-adk/src/kagent/adk/types.py index e8c7ec8671..81ec0f183e 100644 --- a/python/packages/kagent-adk/src/kagent/adk/types.py +++ b/python/packages/kagent-adk/src/kagent/adk/types.py @@ -23,6 +23,7 @@ from kagent.adk.models._ollama import create_ollama_llm from kagent.adk.models._openai import AzureOpenAI as OpenAIAzure from kagent.adk.models._openai import OpenAI as OpenAINative +from kagent.adk.models._openai import OpenAIAPIFormat from kagent.adk.models._ssl import create_ssl_context from kagent.adk.sandbox_code_executer import SandboxedLocalCodeExecutor from kagent.adk.tools.ask_user_tool import AskUserTool @@ -273,6 +274,7 @@ class TokenExchangeConfig(BaseModel): class OpenAI(BaseLLM): base_url: str | None = None + api_format: OpenAIAPIFormat | None = None frequency_penalty: float | None = None max_tokens: int | None = None max_completion_tokens: int | None = Field(default=None, ge=1) @@ -682,6 +684,7 @@ def _create_llm_from_model_config(model_config: ModelUnion): temperature=model_config.temperature, timeout=model_config.timeout, top_p=model_config.top_p, + api_format=model_config.api_format, token_exchange=token_exchange, **_transport_kwargs(model_config), ) diff --git a/python/packages/kagent-adk/tests/unittests/models/test_openai_responses.py b/python/packages/kagent-adk/tests/unittests/models/test_openai_responses.py new file mode 100644 index 0000000000..207fdd039c --- /dev/null +++ b/python/packages/kagent-adk/tests/unittests/models/test_openai_responses.py @@ -0,0 +1,629 @@ +import logging +from unittest import mock + +import pytest +from google.adk.models.llm_request import LlmRequest +from google.adk.models.llm_response import LlmResponse +from google.genai import types +from google.genai.types import Content, Part +from openai.types.chat import ChatCompletion, ChatCompletionMessage +from openai.types.chat.chat_completion import Choice +from openai.types.responses import ( + Response, + ResponseCompletedEvent, + ResponseError, + ResponseErrorEvent, + ResponseFailedEvent, + ResponseFunctionToolCall, + ResponseOutputItemDoneEvent, + ResponseOutputMessage, + ResponseOutputText, + ResponseTextDeltaEvent, + ResponseUsage, +) +from pydantic import ValidationError + +from kagent.adk.models import OpenAI +from kagent.adk.models._openai import ( + _convert_content_to_responses_input, + _convert_responses_output_to_llm_response, + _convert_tools_to_responses, +) +from kagent.adk.types import OpenAI as OpenAIModelConfig +from kagent.adk.types import _create_llm_from_model_config + + +def _make_response(output, status="completed", usage=None, error=None) -> Response: + return Response( + id="resp_1", + created_at=0, + model="gpt-4o", + object="response", + output=output, + parallel_tool_calls=True, + tool_choice="auto", + tools=[], + status=status, + usage=usage, + error=error, + ) + + +@pytest.fixture +def llm_request(): + return LlmRequest( + model="gpt-4o", + contents=[Content(role="user", parts=[Part.from_text(text="Hello")])], + config=types.GenerateContentConfig( + temperature=0.1, + response_modalities=[types.Modality.TEXT], + system_instruction="You are a helpful assistant", + ), + ) + + +@pytest.fixture +def responses_llm(): + return OpenAI(model="gpt-4o", type="openai", api_key="fake", api_format="responses", temperature=0.1) + + +async def _collect_kwargs(llm, llm_request, response): + """Run a non-streaming call against a mocked client and return the create() kwargs.""" + with mock.patch.object(llm, "_client") as mock_client: + + async def mock_coro(*args, **kwargs): + return response + + mock_client.responses.create.return_value = mock_coro() + + _ = [resp async for resp in llm.generate_content_async(llm_request, stream=False)] + + mock_client.responses.create.assert_called_once() + _, kwargs = mock_client.responses.create.call_args + return kwargs + + +def test_create_llm_from_model_config_passes_through_api_format(): + model_config = OpenAIModelConfig(model="gpt-4o", type="openai", api_format="responses") + llm = _create_llm_from_model_config(model_config) + assert llm.api_format == "responses" + + +def test_create_llm_from_model_config_defaults_api_format_to_none(): + model_config = OpenAIModelConfig(model="gpt-4o", type="openai") + llm = _create_llm_from_model_config(model_config) + assert llm.api_format is None + + +def test_model_config_rejects_unknown_api_format(): + with pytest.raises(ValidationError): + OpenAIModelConfig(model="gpt-4o", type="openai", api_format="Responses") + + +def test_convert_content_to_responses_input_user_message(): + input_items = _convert_content_to_responses_input([Content(role="user", parts=[Part.from_text(text="hello")])]) + assert len(input_items) == 1 + assert input_items[0]["role"] == "user" + assert input_items[0]["content"] == "hello" + + +def test_convert_content_to_responses_input_skips_system_role(): + input_items = _convert_content_to_responses_input( + [Content(role="system", parts=[Part.from_text(text="be helpful")])] + ) + assert input_items == [] + + +def test_convert_content_to_responses_input_function_call_and_output_paired(): + fc_part = Part.from_function_call(name="add", args={"a": 1, "b": 2}) + fc_part.function_call.id = "call_1" + fr_part = Part.from_function_response(name="add", response={"result": "3"}) + fr_part.function_response.id = "call_1" + + input_items = _convert_content_to_responses_input( + [ + Content(role="model", parts=[fc_part]), + Content(role="user", parts=[fr_part]), + ] + ) + + assert len(input_items) == 2 + call_item, output_item = input_items + assert call_item["type"] == "function_call" + assert call_item["call_id"] == "call_1" + assert call_item["name"] == "add" + assert output_item["type"] == "function_call_output" + assert output_item["call_id"] == "call_1" + assert output_item["output"] == "3" + + +def test_convert_content_to_responses_input_function_call_without_response(): + fc_part = Part.from_function_call(name="add", args={"a": 1}) + fc_part.function_call.id = "call_1" + + input_items = _convert_content_to_responses_input([Content(role="model", parts=[fc_part])]) + + assert len(input_items) == 2 + assert input_items[1]["output"] == "No response available for this function call." + + +def test_convert_content_to_responses_input_stringifies_non_string_result(): + fc_part = Part.from_function_call(name="add", args={"a": 1, "b": 2}) + fc_part.function_call.id = "call_1" + fr_part = Part.from_function_response(name="add", response={"result": 3}) + fr_part.function_response.id = "call_1" + + input_items = _convert_content_to_responses_input( + [ + Content(role="model", parts=[fc_part]), + Content(role="user", parts=[fr_part]), + ] + ) + + assert input_items[1]["output"] == "3" + + +def test_convert_content_to_responses_input_multimodal_user_message(): + input_items = _convert_content_to_responses_input( + [ + Content( + role="user", + parts=[ + Part.from_text(text="what is this?"), + Part.from_bytes(data=b"\x89PNG", mime_type="image/png"), + ], + ) + ] + ) + + assert len(input_items) == 1 + text_item, image_item = input_items[0]["content"] + assert text_item == {"type": "input_text", "text": "what is this?"} + assert image_item["type"] == "input_image" + assert image_item["image_url"].startswith("data:image/png;base64,") + + +def test_convert_tools_to_responses(): + tool = types.Tool( + function_declarations=[ + types.FunctionDeclaration( + name="get_weather", + description="Gets the weather.", + parameters=types.Schema( + type=types.Type.OBJECT, + properties={"location": types.Schema(type=types.Type.STRING)}, + required=["location"], + ), + ) + ] + ) + + result = _convert_tools_to_responses([tool]) + + assert len(result) == 1 + assert result[0]["type"] == "function" + assert result[0]["name"] == "get_weather" + assert result[0]["parameters"]["properties"]["location"]["type"] == "string" + assert result[0]["parameters"]["required"] == ["location"] + + +def test_convert_responses_output_to_llm_response_text(): + response = _make_response( + output=[ + ResponseOutputMessage( + id="msg_1", + content=[ResponseOutputText(type="output_text", text="Hi there!", annotations=[])], + role="assistant", + status="completed", + type="message", + ) + ], + usage=ResponseUsage( + input_tokens=10, + input_tokens_details={"cache_write_tokens": 0, "cached_tokens": 0}, + output_tokens=5, + output_tokens_details={"reasoning_tokens": 0}, + total_tokens=15, + ), + ) + + llm_response = _convert_responses_output_to_llm_response(response) + + assert llm_response.content.parts[0].text == "Hi there!" + assert llm_response.finish_reason == types.FinishReason.STOP + assert llm_response.usage_metadata.prompt_token_count == 10 + assert llm_response.usage_metadata.candidates_token_count == 5 + + +def test_convert_responses_output_to_llm_response_function_call(): + response = _make_response( + output=[ + ResponseFunctionToolCall( + type="function_call", + call_id="call_1", + name="get_weather", + arguments='{"location": "NYC"}', + ) + ], + ) + + llm_response = _convert_responses_output_to_llm_response(response) + + fc = llm_response.content.parts[0].function_call + assert fc.name == "get_weather" + assert fc.args == {"location": "NYC"} + assert fc.id == "call_1" + + +def test_convert_responses_output_to_llm_response_incomplete_status(): + response = _make_response(output=[], status="incomplete") + llm_response = _convert_responses_output_to_llm_response(response) + assert llm_response.finish_reason == types.FinishReason.MAX_TOKENS + + +@pytest.mark.asyncio +async def test_generate_content_async_uses_responses_api(responses_llm, llm_request): + response = _make_response( + output=[ + ResponseOutputMessage( + id="msg_1", + content=[ResponseOutputText(type="output_text", text="Hi there!", annotations=[])], + role="assistant", + status="completed", + type="message", + ) + ], + ) + + with mock.patch.object(responses_llm, "_client") as mock_client: + + async def mock_coro(*args, **kwargs): + return response + + mock_client.responses.create.return_value = mock_coro() + + results = [resp async for resp in responses_llm.generate_content_async(llm_request, stream=False)] + + assert len(results) == 1 + assert isinstance(results[0], LlmResponse) + assert results[0].content.parts[0].text == "Hi there!" + + mock_client.responses.create.assert_called_once() + assert mock_client.chat.completions.create.call_count == 0 + _, kwargs = mock_client.responses.create.call_args + assert kwargs["model"] == "gpt-4o" + assert kwargs["instructions"] == "You are a helpful assistant" + assert kwargs["temperature"] == 0.1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("api_format", [None, "chatCompletions"]) +async def test_generate_content_async_defaults_to_chat_completions(api_format, llm_request): + llm = OpenAI(model="gpt-4o", type="openai", api_key="fake", api_format=api_format) + + completion = ChatCompletion( + id="chatcmpl_1", + created=0, + model="gpt-4o", + object="chat.completion", + choices=[ + Choice( + finish_reason="stop", + index=0, + message=ChatCompletionMessage(role="assistant", content="Hi there!"), + ) + ], + ) + + with mock.patch.object(llm, "_client") as mock_client: + + async def mock_coro(*args, **kwargs): + return completion + + mock_client.chat.completions.create.return_value = mock_coro() + + results = [resp async for resp in llm.generate_content_async(llm_request, stream=False)] + + assert results[0].content.parts[0].text == "Hi there!" + mock_client.chat.completions.create.assert_called_once() + assert mock_client.responses.create.call_count == 0 + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_sends_reasoning_effort(llm_request): + llm = OpenAI(model="gpt-5", type="openai", api_key="fake", api_format="responses", reasoning_effort="high") + + kwargs = await _collect_kwargs(llm, llm_request, _make_response(output=[])) + + assert kwargs["reasoning"] == {"effort": "high"} + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_max_completion_tokens_wins(llm_request): + llm = OpenAI( + model="gpt-4o", + type="openai", + api_key="fake", + api_format="responses", + max_tokens=1024, + max_completion_tokens=2048, + ) + + kwargs = await _collect_kwargs(llm, llm_request, _make_response(output=[])) + + assert kwargs["max_output_tokens"] == 2048 + assert "max_tokens" not in kwargs + assert "max_completion_tokens" not in kwargs + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_falls_back_to_max_tokens(llm_request): + llm = OpenAI(model="gpt-4o", type="openai", api_key="fake", api_format="responses", max_tokens=1024) + + kwargs = await _collect_kwargs(llm, llm_request, _make_response(output=[])) + + assert kwargs["max_output_tokens"] == 1024 + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_sends_tools(responses_llm, llm_request): + llm_request.config.tools = [ + types.Tool( + function_declarations=[ + types.FunctionDeclaration( + name="get_weather", + description="Gets the weather.", + parameters=types.Schema( + type=types.Type.OBJECT, + properties={"location": types.Schema(type=types.Type.STRING)}, + required=["location"], + ), + ) + ] + ) + ] + + kwargs = await _collect_kwargs(responses_llm, llm_request, _make_response(output=[])) + + assert kwargs["tool_choice"] == "auto" + assert [tool["name"] for tool in kwargs["tools"]] == ["get_weather"] + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_ignored_params_warn(caplog): + with caplog.at_level(logging.WARNING, logger="kagent.adk.models._openai"): + OpenAI(model="gpt-5", type="openai", api_key="fake", api_format="responses", seed=7, n=2) + + assert "seed" in caplog.text + assert "n" in caplog.text + + +def test_responses_params_do_not_warn_for_chat_completions(caplog): + with caplog.at_level(logging.WARNING, logger="kagent.adk.models._openai"): + OpenAI(model="gpt-4o", type="openai", api_key="fake", seed=7) + + assert caplog.text == "" + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_streaming(responses_llm, llm_request): + events = [ + ResponseTextDeltaEvent( + content_index=0, + delta="Hi ", + item_id="msg_1", + logprobs=[], + output_index=0, + sequence_number=1, + type="response.output_text.delta", + ), + ResponseTextDeltaEvent( + content_index=0, + delta="there!", + item_id="msg_1", + logprobs=[], + output_index=0, + sequence_number=2, + type="response.output_text.delta", + ), + ResponseOutputItemDoneEvent( + item=ResponseFunctionToolCall( + type="function_call", + call_id="call_1", + name="get_weather", + arguments='{"location": "NYC"}', + ), + output_index=1, + sequence_number=3, + type="response.output_item.done", + ), + ResponseCompletedEvent( + response=_make_response( + output=[], + usage=ResponseUsage( + input_tokens=1, + input_tokens_details={"cache_write_tokens": 0, "cached_tokens": 0}, + output_tokens=2, + output_tokens_details={"reasoning_tokens": 0}, + total_tokens=3, + ), + ), + sequence_number=4, + type="response.completed", + ), + ] + + with mock.patch.object(responses_llm, "_client") as mock_client: + + async def mock_stream_gen_func(*args, **kwargs): + async def gen(): + for event in events: + yield event + + return gen() + + mock_client.responses.create.side_effect = mock_stream_gen_func + + results = [resp async for resp in responses_llm.generate_content_async(llm_request, stream=True)] + + partials = [r for r in results if r.partial] + assert [p.content.parts[0].text for p in partials] == ["Hi ", "there!"] + + final = results[-1] + assert final.partial is False + assert final.content.parts[0].text == "Hi there!" + fc = final.content.parts[1].function_call + assert fc.name == "get_weather" + assert fc.args == {"location": "NYC"} + assert final.usage_metadata.prompt_token_count == 1 + assert final.usage_metadata.candidates_token_count == 2 + + +async def _stream_results(llm, llm_request, events): + with mock.patch.object(llm, "_client") as mock_client: + + async def mock_stream_gen_func(*args, **kwargs): + async def gen(): + for event in events: + yield event + + return gen() + + mock_client.responses.create.side_effect = mock_stream_gen_func + + return [resp async for resp in llm.generate_content_async(llm_request, stream=True)] + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_streaming_recovers_text_without_deltas(responses_llm, llm_request): + events = [ + ResponseOutputItemDoneEvent( + item=ResponseOutputMessage( + id="msg_1", + content=[ResponseOutputText(type="output_text", text="Hi there!", annotations=[])], + role="assistant", + status="completed", + type="message", + ), + output_index=0, + sequence_number=1, + type="response.output_item.done", + ), + ResponseCompletedEvent( + response=_make_response(output=[]), + sequence_number=2, + type="response.completed", + ), + ] + + results = await _stream_results(responses_llm, llm_request, events) + + assert len(results) == 1 + assert results[0].content.parts[0].text == "Hi there!" + assert results[0].finish_reason == types.FinishReason.STOP + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_streaming_surfaces_failure(responses_llm, llm_request): + events = [ + ResponseFailedEvent( + response=_make_response( + output=[], + status="failed", + error=ResponseError(code="server_error", message="upstream exploded"), + ), + sequence_number=1, + type="response.failed", + ), + ] + + results = await _stream_results(responses_llm, llm_request, events) + + assert len(results) == 1 + assert results[0].error_code == "API_ERROR" + assert results[0].error_message == "server_error: upstream exploded" + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_streaming_surfaces_bare_error_event(responses_llm, llm_request): + events = [ + ResponseErrorEvent( + code="rate_limit_exceeded", + message="slow down", + sequence_number=1, + type="error", + ), + ] + + results = await _stream_results(responses_llm, llm_request, events) + + assert len(results) == 1 + assert results[0].error_code == "API_ERROR" + assert results[0].error_message == "rate_limit_exceeded: slow down" + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_surfaces_failed_status(responses_llm, llm_request): + response = _make_response( + output=[], + status="failed", + error=ResponseError(code="invalid_prompt", message="prompt was rejected"), + ) + + with mock.patch.object(responses_llm, "_client") as mock_client: + + async def mock_coro(*args, **kwargs): + return response + + mock_client.responses.create.return_value = mock_coro() + + results = [resp async for resp in responses_llm.generate_content_async(llm_request, stream=False)] + + assert len(results) == 1 + assert results[0].error_code == "API_ERROR" + assert results[0].error_message == "invalid_prompt: prompt was rejected" + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_failed_status_without_error_payload(responses_llm, llm_request): + response = _make_response(output=[], status="failed") + + with mock.patch.object(responses_llm, "_client") as mock_client: + + async def mock_coro(*args, **kwargs): + return response + + mock_client.responses.create.return_value = mock_coro() + + results = [resp async for resp in responses_llm.generate_content_async(llm_request, stream=False)] + + assert results[0].error_code == "API_ERROR" + assert results[0].error_message == "OpenAI responses request failed" + + +@pytest.mark.asyncio +async def test_generate_content_async_responses_completed_status_is_not_an_error(responses_llm, llm_request): + response = _make_response( + output=[ + ResponseOutputMessage( + id="msg_1", + content=[ResponseOutputText(type="output_text", text="Hi there!", annotations=[])], + role="assistant", + status="completed", + type="message", + ) + ], + ) + + with mock.patch.object(responses_llm, "_client") as mock_client: + + async def mock_coro(*args, **kwargs): + return response + + mock_client.responses.create.return_value = mock_coro() + + results = [resp async for resp in responses_llm.generate_content_async(llm_request, stream=False)] + + assert results[0].error_code is None + assert results[0].error_message is None + assert results[0].content.parts[0].text == "Hi there!"