From b04443be1a23926b08c803d230ecd746ee559b45 Mon Sep 17 00:00:00 2001 From: Alessandro <155005371+3clyp50@users.noreply.github.com> Date: Mon, 1 Jun 2026 01:32:23 +0200 Subject: [PATCH] Add native Responses API transport Route Agent Zero turns through a LiteLLM transport layer that prefers the Responses API while preserving chat-completions fallback for providers without compatible endpoints. Persist Responses metadata in history and agent state so provider-state continuation, local replay, native function-call execution, and stored-response cleanup survive normal chat workflows. Normalize prompt caching by provider: OpenAI and Azure use prompt_cache_key and prompt_cache_retention, while Anthropic, Gemini, Bedrock, OpenRouter, and compatible chat providers keep block-level cache_control breakpoints and cached tool definitions. --- agent.py | 549 ++++++++- helpers/history.py | 112 +- helpers/litellm_transport.py | 1676 ++++++++++++++++++++++++++ helpers/llm_result.py | 311 +++++ helpers/persist_chat.py | 69 ++ helpers/responses_tools.py | 231 ++++ models.py | 343 ++++-- tests/test_responses_architecture.py | 385 ++++++ tests/test_stream_tool_early_stop.py | 780 +++++++++++- 9 files changed, 4299 insertions(+), 157 deletions(-) create mode 100644 helpers/litellm_transport.py create mode 100644 helpers/llm_result.py create mode 100644 helpers/responses_tools.py create mode 100644 tests/test_responses_architecture.py diff --git a/agent.py b/agent.py index 149c900e3..8eef2906b 100644 --- a/agent.py +++ b/agent.py @@ -1,4 +1,4 @@ -import asyncio, random, string, threading +import asyncio, json, random, re, string, threading from collections import OrderedDict from dataclasses import dataclass, field @@ -32,6 +32,15 @@ from typing import Callable from helpers.localization import Localization from helpers import extension from helpers.errors import RepairableException, InterventionException, HandledException +from helpers.llm_result import ( + LLMResult, + RESPONSE_METADATA_KEY, + function_call_output_item, + metadata_from_llm_result, + result_from_metadata, +) +from helpers.litellm_transport import ResponsesTransport +from helpers.responses_tools import build_responses_function_tools, original_tool_name class AgentContextType(Enum): USER = "user" @@ -346,6 +355,9 @@ class Agent: DATA_NAME_SUPERIOR = "_superior" DATA_NAME_SUBORDINATE = "_subordinate" DATA_NAME_CTX_WINDOW = "ctx_window" + DATA_NAME_RESPONSES_STATE = "responses_state" + DATA_NAME_RESPONSES_TOOL_NAME_MAP = "responses_tool_name_map" + DATA_NAME_RESPONSES_COMPUTER_SESSION = "responses_computer_session_id" @extension.extensible def __init__( @@ -468,11 +480,12 @@ class Agent: return stop_response # call main LLM - agent_response, _reasoning = await self.call_chat_model( + llm_result = await self.call_chat_model_turn( messages=prompt, response_callback=stream_callback, reasoning_callback=reasoning_callback, ) + agent_response = llm_result.response await self.handle_intervention(agent_response) # Notify extensions to finalize their stream filters @@ -492,7 +505,12 @@ class Agent: ): # if assistant_response is the same as last message in history, let him know # Append the assistant's response to the history log_item = self.loop_data.params_temporary.get("log_item_generating") - self.hist_add_ai_response(agent_response, id=log_item.id if log_item else "") + assistant_message = self.hist_add_ai_response( + agent_response, + id=log_item.id if log_item else "", + llm_result=llm_result, + ) + self._remember_llm_result_state(llm_result, assistant_message) # Append warning message to the history warning_msg = self.read_prompt("fw.msg_repeat.md") wmsg = self.hist_add_warning(message=warning_msg) @@ -504,9 +522,16 @@ class Agent: else: # otherwise proceed with tool # Append the assistant's response to the history log_item = self.loop_data.params_temporary.get("log_item_generating") - self.hist_add_ai_response(agent_response, id=log_item.id if log_item else "") + assistant_message = self.hist_add_ai_response( + agent_response, + id=log_item.id if log_item else "", + llm_result=llm_result, + ) + self._remember_llm_result_state(llm_result, assistant_message) # process tools requested in agent message - tools_result = await self.process_tools(agent_response) + tools_result = await self.process_llm_result_tools( + llm_result + ) if tools_result: # final response of message loop available return tools_result # break the execution if the task is done @@ -664,7 +689,12 @@ class Agent: @extension.extensible def hist_add_message( - self, ai: bool, content: history.MessageContent, tokens: int = 0, id: str = "" + self, + ai: bool, + content: history.MessageContent, + tokens: int = 0, + id: str = "", + metadata: dict[str, Any] | None = None, ): self.last_message = Localization.get().now() # Allow extensions to process content before adding to history @@ -673,7 +703,11 @@ class Agent: "hist_add_before", self, content_data=content_data, ai=ai ) return self.history.add_message( - ai=ai, content=content_data["content"], tokens=tokens, id=id + ai=ai, + content=content_data["content"], + tokens=tokens, + id=id, + metadata=metadata, ) @extension.extensible @@ -706,10 +740,17 @@ class Agent: return msg @extension.extensible - def hist_add_ai_response(self, message: str, id: str = ""): + def hist_add_ai_response( + self, message: str, id: str = "", llm_result: LLMResult | None = None + ): self.loop_data.last_response = message content = self.parse_prompt("fw.ai_response.md", message=message) - return self.hist_add_message(True, content=content, id=id) + return self.hist_add_message( + True, + content=content, + id=id, + metadata=metadata_from_llm_result(llm_result), + ) @extension.extensible def hist_add_warning(self, message: history.MessageContent, id: str = ""): @@ -719,13 +760,28 @@ class Agent: @extension.extensible def hist_add_tool_result(self, tool_name: str, tool_result: str, **kwargs): msg_id = kwargs.pop("id", "") + responses_item = kwargs.pop("_responses_output_item", None) or kwargs.pop( + "responses_item", None + ) + metadata = ( + { + RESPONSE_METADATA_KEY: { + "input_items": [responses_item], + "output_items": [], + "mode": "responses", + "state": "provider", + } + } + if isinstance(responses_item, dict) + else None + ) data = { "tool_name": tool_name, "tool_result": tool_result, **kwargs, } extension.call_extensions_sync("hist_add_tool_result", self, data=data) - return self.hist_add_message(False, content=data, id=msg_id) + return self.hist_add_message(False, content=data, id=msg_id, metadata=metadata) def concat_messages( self, messages @@ -830,6 +886,170 @@ class Agent: return response, reasoning + @extension.extensible + async def call_chat_model_turn( + self, + messages: list[BaseMessage], + response_callback: Callable[[str, str], Awaitable[str | None]] | None = None, + reasoning_callback: Callable[[str, str], Awaitable[None]] | None = None, + background: bool = False, + explicit_caching: bool = True, + ) -> LLMResult: + model = self.get_chat_model() + model_kwargs = getattr(model, "kwargs", {}) if model else {} + if isinstance(model_kwargs, dict) and model_kwargs.get("responses_delete_on_chat_delete") is False: + self.set_data("responses_delete_on_chat_delete", False) + response_tools, name_map = build_responses_function_tools(self) + self.set_data(Agent.DATA_NAME_RESPONSES_TOOL_NAME_MAP, name_map) + + call_data = { + "model": model, + "messages": messages, + "response_callback": response_callback, + "reasoning_callback": reasoning_callback, + "background": background, + "explicit_caching": explicit_caching, + "a0_responses_function_tools": response_tools, + } + + previous_state = self._responses_state_for_model(model) + if previous_state: + history_counter = int(previous_state.get("history_counter", 0) or 0) + call_data["previous_response_id"] = previous_state.get("response_id", "") + call_data["responses_input_items"] = self._responses_input_items_since( + model, + history_counter, + ) + call_data["responses_local_input_items"] = ( + self._responses_static_prefix_items(model, messages) + + self._responses_input_items_since(model, 0) + ) + + await extension.call_extensions_async( + "chat_model_call_before", self, call_data=call_data + ) + + turn_kwargs = { + "a0_responses_function_tools": call_data.get( + "a0_responses_function_tools" + ), + "responses_local_input_items": call_data.get( + "responses_local_input_items" + ), + } + for key in ( + "responses_builtin_tools", + "responses_state", + "previous_response_id", + "responses_input_items", + ): + if call_data.get(key) is not None: + turn_kwargs[key] = call_data.get(key) + + llm_result = await call_data["model"].unified_turn( + messages=call_data["messages"], + reasoning_callback=call_data["reasoning_callback"], + response_callback=call_data["response_callback"], + rate_limiter_callback=( + self.rate_limiter_callback if not call_data["background"] else None + ), + explicit_caching=call_data["explicit_caching"], + **turn_kwargs, + ) + + downgraded = llm_result.capability.get("builtin_tool_downgrades") + if downgraded: + self.context.log.log( + type="info", + heading="Responses capability downgrade", + content=( + "Provider rejected Responses built-in tool(s); omitted: " + + ", ".join(str(item) for item in downgraded) + ), + ) + + await extension.call_extensions_async( + "chat_model_call_after", + self, + call_data=call_data, + response=llm_result.response, + reasoning=llm_result.reasoning, + ) + + return llm_result + + def _responses_state_for_model(self, model: Any) -> dict[str, Any]: + state = self.get_data(Agent.DATA_NAME_RESPONSES_STATE) + if not isinstance(state, dict): + return {} + provider_model_key = str(getattr(model, "model_name", "") or "") + if state.get("provider_model_key") != provider_model_key: + return {} + if not state.get("response_id"): + return {} + return state + + def _responses_input_items_since( + self, model: Any, sequence: int + ) -> list[dict[str, Any]]: + items: list[dict[str, Any]] = [] + for message in self.history.messages_since(sequence): + items.extend(self._responses_input_items_for_message(model, message)) + return items + + def _responses_input_items_for_message( + self, model: Any, message: history.Message + ) -> list[dict[str, Any]]: + result = result_from_metadata(message.metadata) + if result: + if message.ai and result.output_items: + return [item.to_dict() for item in result.output_items] + if not message.ai and result.input_items: + return [dict(item) for item in result.input_items] + + output = message.output() + langchain_messages = history.output_langchain(output) + if hasattr(model, "_convert_messages"): + converted = model._convert_messages(langchain_messages) + return ResponsesTransport.input_from_messages(converted) + return [] + + def _responses_static_prefix_items( + self, model: Any, messages: list[BaseMessage] + ) -> list[dict[str, Any]]: + prefix: list[BaseMessage] = [] + for message in messages: + if isinstance(message, SystemMessage): + prefix.append(message) + continue + break + if not prefix or not hasattr(model, "_convert_messages"): + return [] + converted = model._convert_messages(prefix) + return ResponsesTransport.input_from_messages(converted) + + def _remember_llm_result_state( + self, llm_result: LLMResult, history_message: history.Message + ) -> None: + if not llm_result.response_id: + return + current = self.get_data(Agent.DATA_NAME_RESPONSES_STATE) + response_ids = [] + if isinstance(current, dict) and isinstance(current.get("response_ids"), list): + response_ids = [str(item) for item in current["response_ids"] if item] + if llm_result.response_id not in response_ids: + response_ids.append(llm_result.response_id) + self.set_data( + Agent.DATA_NAME_RESPONSES_STATE, + { + "response_id": llm_result.response_id, + "previous_response_id": llm_result.previous_response_id, + "provider_model_key": llm_result.provider_model_key, + "history_counter": history_message.sequence, + "response_ids": response_ids, + }, + ) + @extension.extensible async def rate_limiter_callback( self, message: str, key: str, total: int, limit: int @@ -863,6 +1083,310 @@ class Agent: while self.context.paused: await asyncio.sleep(0.1) + async def process_llm_result_tools(self, llm_result: LLMResult): + await self._log_response_builtin_items(llm_result) + if llm_result.function_calls: + for function_call in llm_result.function_calls: + name_map = self.get_data(Agent.DATA_NAME_RESPONSES_TOOL_NAME_MAP) + tool_name = original_tool_name(function_call.name, name_map) + response_item_factory = lambda response, call=function_call: function_call_output_item( + call.call_id, + response.message, + ) + result = await self._execute_tool_request( + tool_name=tool_name, + tool_args=function_call.arguments, + message=llm_result.response, + raw_tool_name=tool_name, + responses_item_factory=response_item_factory, + ) + if result: + return result + return None + if llm_result.builtin_items and not llm_result.response: + return None + if ( + llm_result.mode == "responses" + and llm_result.response + and extract_tools.json_parse_dirty(llm_result.response) is None + ): + return llm_result.response + return await self.process_tools(llm_result.response) + + async def _execute_tool_request( + self, + tool_name: str, + tool_args: dict, + message: str, + raw_tool_name: str = "", + responses_item_factory: Callable[[Any], dict[str, Any]] | None = None, + ): + raw_tool_name = raw_tool_name or tool_name + tool_method = None + tool = None + + try: + import helpers.mcp_handler as mcp_helper + + mcp_tool_candidate = mcp_helper.MCPConfig.get_instance().get_tool( + self, tool_name + ) + if mcp_tool_candidate: + tool = mcp_tool_candidate + except ImportError: + PrintStyle( + background_color="black", font_color="yellow", padding=True + ).print("MCP helper module not found. Skipping MCP tool lookup.") + except Exception as e: + PrintStyle(background_color="black", font_color="red", padding=True).print( + f"Failed to get MCP tool '{tool_name}': {e}" + ) + + if not tool: + tool = self.get_tool( + name=tool_name, + method=tool_method, + args=tool_args, + message=message, + loop_data=self.loop_data, + ) + + if not tool: + error_detail = ( + f"Tool '{raw_tool_name}' not found or could not be initialized." + ) + wmsg = self.hist_add_warning(error_detail) + PrintStyle(font_color="red", padding=True).print(error_detail) + self.context.log.log( + type="warning", + content=f"{self.agent_name}: {error_detail}", + id=wmsg.id, + ) + return None + + self.loop_data.current_tool = tool # type: ignore + try: + await self.handle_intervention() + + await tool.before_execution(**tool_args) + await self.handle_intervention() + + await extension.call_extensions_async( + "tool_execute_before", + self, + tool_args=tool_args or {}, + tool_name=tool_name, + ) + + response = await tool.execute(**tool_args) + await self.handle_intervention() + + await extension.call_extensions_async( + "tool_execute_after", + self, + response=response, + tool_name=tool_name, + ) + + if responses_item_factory: + response.additional = { + **(response.additional or {}), + "_responses_output_item": responses_item_factory(response), + } + + await tool.after_execution(response) + await self.handle_intervention() + + if response.break_loop: + self._clear_responses_pending_state() + return response.message + finally: + self.loop_data.current_tool = None + return None + + async def _log_response_builtin_items(self, llm_result: LLMResult) -> None: + for item in llm_result.builtin_items: + if item.type == "computer_call": + await self._handle_responses_computer_call(item.data) + continue + if item.type == "mcp_approval_request": + self._handle_responses_mcp_approval_request(item.data) + continue + self.context.log.log( + type="info", + heading=f"Responses tool item: {item.type}", + content=json.dumps(item.data, ensure_ascii=False, default=str), + ) + + async def _handle_responses_computer_call(self, item: dict[str, Any]) -> None: + safety_checks = item.get("pending_safety_checks") or item.get("safety_checks") + if safety_checks: + message = ( + "Responses computer_call requested safety-check acknowledgement. " + "Agent Zero requires explicit user acknowledgement before executing it." + ) + output_item = { + "type": "computer_call_output", + "call_id": str(item.get("call_id") or item.get("id") or ""), + "output": {"type": "input_text", "text": message}, + } + self.hist_add_tool_result( + "computer_call", + message, + responses_item=output_item, + ) + self.context.log.log(type="warning", content=message) + return + + args = self._computer_call_args(item) + if not args: + message = "Responses computer_call action is unsupported by Agent Zero." + output_item = { + "type": "computer_call_output", + "call_id": str(item.get("call_id") or item.get("id") or ""), + "output": {"type": "input_text", "text": message}, + } + self.hist_add_tool_result( + "computer_call", + message, + responses_item=output_item, + ) + self.context.log.log(type="warning", content=message) + return + + if args.get("action") != "start_session" and not args.get("session_id"): + session_id = str( + self.get_data(Agent.DATA_NAME_RESPONSES_COMPUTER_SESSION) or "" + ) + if session_id: + args["session_id"] = session_id + + response_item_factory = lambda response: self._computer_call_output_item( + item, + response, + ) + result = await self._execute_tool_request( + tool_name="computer_use_remote", + tool_args=args, + message=json.dumps(item, ensure_ascii=False, default=str), + raw_tool_name="computer_call", + responses_item_factory=response_item_factory, + ) + _ = result + + def _handle_responses_mcp_approval_request(self, item: dict[str, Any]) -> None: + request_id = str( + item.get("approval_request_id") or item.get("id") or item.get("call_id") or "" + ) + message = ( + "Responses MCP approval request received. Agent Zero denied it because " + "provider-hosted MCP approval requires explicit user approval." + ) + output_item = { + "type": "mcp_approval_response", + "approval_request_id": request_id, + "approve": False, + } + self.hist_add_tool_result( + "mcp_approval_request", + message, + responses_item=output_item, + ) + self.context.log.log( + type="warning", + heading="Responses MCP approval required", + content=message, + ) + + def _computer_call_args(self, item: dict[str, Any]) -> dict[str, Any]: + action = item.get("action") + action_data = dict(action) if isinstance(action, dict) else {} + action_type = str( + action_data.get("type") + or action_data.get("action") + or item.get("action_type") + or "" + ).strip().lower() + args: dict[str, Any] = {} + + if action_type in {"screenshot", "capture"}: + args["action"] = "capture" + elif action_type in {"move", "mousemove"}: + args.update({"action": "move", "x": action_data.get("x"), "y": action_data.get("y")}) + elif action_type in {"click", "double_click"}: + args.update( + { + "action": "click", + "x": action_data.get("x"), + "y": action_data.get("y"), + "button": action_data.get("button", "left"), + "count": 2 if action_type == "double_click" else action_data.get("count", 1), + } + ) + elif action_type == "scroll": + args.update( + { + "action": "scroll", + "dx": action_data.get("dx", action_data.get("scroll_x", 0)), + "dy": action_data.get("dy", action_data.get("scroll_y", 0)), + } + ) + elif action_type in {"keypress", "key"}: + args.update( + { + "action": "key", + "keys": action_data.get("keys") or action_data.get("key"), + } + ) + elif action_type in {"type", "input_text"}: + args.update({"action": "type", "text": action_data.get("text", "")}) + else: + return {} + + session_id = item.get("session_id") or action_data.get("session_id") + if session_id: + args["session_id"] = session_id + return args + + def _computer_call_output_item( + self, source_item: dict[str, Any], response: Any + ) -> dict[str, Any]: + output: dict[str, Any] = { + "type": "input_text", + "text": str(getattr(response, "message", "") or ""), + } + additional = getattr(response, "additional", None) + raw_content = additional.get("raw_content") if isinstance(additional, dict) else None + if isinstance(raw_content, list): + for content in raw_content: + if not isinstance(content, dict): + continue + if content.get("type") != "image_url": + continue + image_url = content.get("image_url") + url = image_url.get("url") if isinstance(image_url, dict) else image_url + if url: + output = {"type": "input_image", "image_url": url} + break + + session_id_match = re_search_session_id(str(getattr(response, "message", "") or "")) + if session_id_match: + self.set_data(Agent.DATA_NAME_RESPONSES_COMPUTER_SESSION, session_id_match) + + return { + "type": "computer_call_output", + "call_id": str(source_item.get("call_id") or source_item.get("id") or ""), + "output": output, + } + + def _clear_responses_pending_state(self) -> None: + state = self.get_data(Agent.DATA_NAME_RESPONSES_STATE) + if isinstance(state, dict): + state = dict(state) + state.pop("response_id", None) + state.pop("previous_response_id", None) + self.set_data(Agent.DATA_NAME_RESPONSES_STATE, state) + @extension.extensible async def process_tools(self, msg: str): # search for tool usage requests in agent message @@ -1037,3 +1561,8 @@ class Agent: loop_data=loop_data, **kwargs, ) + + +def re_search_session_id(text: str) -> str: + match = re.search(r"session_id=([A-Za-z0-9_.:-]+)", text or "") + return match.group(1) if match else "" diff --git a/helpers/history.py b/helpers/history.py index 4fdefd179..c579c501d 100644 --- a/helpers/history.py +++ b/helpers/history.py @@ -40,9 +40,12 @@ MessageContent = Union[ ] -class OutputMessage(TypedDict): +class OutputMessage(TypedDict, total=False): ai: bool content: MessageContent + metadata: dict[str, Any] + id: str + sequence: int class Record: @@ -82,10 +85,20 @@ class Record: class Message(Record): - def __init__(self, ai: bool, content: MessageContent, tokens: int = 0, id: str = ""): + def __init__( + self, + ai: bool, + content: MessageContent, + tokens: int = 0, + id: str = "", + metadata: dict[str, Any] | None = None, + sequence: int = 0, + ): self.id = id or str(uuid.uuid4()) self.ai = ai self.content = content + self.metadata = metadata or {} + self.sequence = sequence self.summary: str = "" self.tokens: int = tokens or self.calculate_tokens() @@ -106,7 +119,15 @@ class Message(Record): return False def output(self): - return [OutputMessage(ai=self.ai, content=self.summary or self.content)] + return [ + OutputMessage( + ai=self.ai, + content=self.summary or self.content, + metadata=self.metadata, + id=self.id, + sequence=self.sequence, + ) + ] def output_langchain(self): return output_langchain(self.output()) @@ -120,6 +141,8 @@ class Message(Record): "id": self.id, "ai": self.ai, "content": self.content, + "metadata": self.metadata, + "sequence": self.sequence, "summary": self.summary, "tokens": self.tokens, } @@ -127,7 +150,13 @@ class Message(Record): @staticmethod def from_dict(data: dict, history: "History"): content = data.get("content", "Content lost") - msg = Message(ai=data["ai"], content=content, id=data.get("id", "")) + msg = Message( + ai=data["ai"], + content=content, + id=data.get("id", ""), + metadata=data.get("metadata", {}) if isinstance(data.get("metadata"), dict) else {}, + sequence=int(data.get("sequence", 0) or 0), + ) msg.summary = data.get("summary", "") msg.tokens = data.get("tokens", 0) return msg @@ -146,9 +175,22 @@ class Topic(Record): return sum(msg.get_tokens() for msg in self.messages) def add_message( - self, ai: bool, content: MessageContent, tokens: int = 0, id: str = "" + self, + ai: bool, + content: MessageContent, + tokens: int = 0, + id: str = "", + metadata: dict[str, Any] | None = None, + sequence: int = 0, ) -> Message: - msg = Message(ai=ai, content=content, tokens=tokens, id=id) + msg = Message( + ai=ai, + content=content, + tokens=tokens, + id=id, + metadata=metadata, + sequence=sequence, + ) self.messages.append(msg) return msg @@ -335,10 +377,22 @@ class History(Record): return self.current.get_tokens() def add_message( - self, ai: bool, content: MessageContent, tokens: int = 0, id: str = "" + self, + ai: bool, + content: MessageContent, + tokens: int = 0, + id: str = "", + metadata: dict[str, Any] | None = None, ) -> Message: self.counter += 1 - return self.current.add_message(ai, content=content, tokens=tokens, id=id) + return self.current.add_message( + ai, + content=content, + tokens=tokens, + id=id, + metadata=metadata, + sequence=self.counter, + ) def new_topic(self): if self.current.messages: @@ -353,6 +407,35 @@ class History(Record): result += self.current.output() return result + def messages_since(self, sequence: int) -> list[Message]: + return [ + message + for message in self.all_messages() + if int(message.sequence or 0) > int(sequence or 0) + ] + + def all_messages(self) -> list[Message]: + messages: list[Message] = [] + for bulk in self.bulks: + messages.extend(_messages_from_record(bulk)) + for topic in self.topics: + messages.extend(topic.messages) + messages.extend(self.current.messages) + return messages + + def latest_llm_result_for_model(self, provider_model_key: str): + from helpers.llm_result import result_from_metadata + + for message in reversed(self.all_messages()): + if not message.ai: + continue + result = result_from_metadata(message.metadata) + if not result: + continue + if result.provider_model_key == provider_model_key and result.response_id: + return result + return None + def trim_embeds(self, max_embeds: int) -> int: if max_embeds == -1: return 0 @@ -679,6 +762,19 @@ def _is_embedded_data(obj: object) -> bool: return isinstance(obj, Mapping) and obj.get("type") == "image_url" +def _messages_from_record(record: Record) -> list[Message]: + if isinstance(record, Message): + return [record] + if isinstance(record, Topic): + return list(record.messages) + if isinstance(record, Bulk): + messages: list[Message] = [] + for nested in record.records: + messages.extend(_messages_from_record(nested)) + return messages + return [] + + def _json_dumps(obj): return json.dumps(obj, ensure_ascii=False) diff --git a/helpers/litellm_transport.py b/helpers/litellm_transport.py new file mode 100644 index 000000000..3d7155296 --- /dev/null +++ b/helpers/litellm_transport.py @@ -0,0 +1,1676 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +import hashlib +import inspect +import json +from typing import Any, AsyncIterator, Iterator, Optional + +from litellm import ( + acompletion, + adelete_responses, + aresponses, + completion, + delete_responses, + responses, +) + +from helpers import images +from helpers.llm_result import LLMResult + + +ChatChunk = dict[str, str] + + +class TransportMode(Enum): + RESPONSES = "responses" + CHAT_COMPLETIONS = "chat_completions" + + +class TransportRecovery(Enum): + RAISE = "raise" + RETRY_RESPONSES = "retry_responses" + RETRY_LOCAL_RESPONSES = "retry_local_responses" + FALLBACK_TO_CHAT = "fallback_to_chat" + + +CHAT_COMPLETIONS_ALIASES = { + "chat", + "chat_completion", + "chat_completions", + "completion", + "completions", +} +RESPONSES_ALIASES = {"", "auto", "default", "response", "responses", "responses_api"} +RESPONSES_REASONING_EFFORTS = {"minimal", "low", "medium", "high"} +RESPONSES_REASONING_FALLBACK_EFFORT = "high" +NO_REASONING_EFFORT_ALIASES = {"", "0", "false", "no", "none", "off", "disabled"} +RESPONSES_UNSUPPORTED_CACHE: set[str] = set() +RESPONSES_STATE_UNSUPPORTED_CACHE: set[str] = set() +RESPONSES_BUILTIN_UNSUPPORTED_CACHE: dict[str, set[str]] = {} +OPENAI_RESPONSES_EXTRA_BODY_PARAMS = { + "context_management", + "prompt_cache_retention", +} +CACHE_CONTROL_PROMPT_PROVIDERS = { + "anthropic", + "bedrock", + "databricks", + "dashscope", + "gemini", + "gemini_api_oauth", + "minimax", + "openrouter", + "vertex_ai", + "vertexai", + "z_ai", + "zai", +} +OPENAI_PROMPT_CACHE_PROVIDERS = {"openai", "azure"} +RESPONSES_STATE_PROVIDER = "provider" +RESPONSES_STATE_LOCAL = "local" +RESPONSES_STATE_OFF = "off" +RESPONSES_STATES = { + RESPONSES_STATE_PROVIDER, + RESPONSES_STATE_LOCAL, + RESPONSES_STATE_OFF, +} + + +@dataclass +class TransportPolicy: + mode: TransportMode + allow_fallback: bool = True + retried_reasoning: bool = False + fallback_error: Exception | None = None + state_fallback_error: Exception | None = None + cache_key: str = "" + state: str = RESPONSES_STATE_PROVIDER + + @classmethod + def from_request( + cls, + model: str, + kwargs: dict[str, Any], + messages: list[dict[str, Any]] | None = None, + ) -> "TransportPolicy": + mode = cls._pop_mode(kwargs) + allow_fallback = _coerce_bool( + kwargs.pop("a0_responses_fallback", True), default=True + ) + cache_key = _responses_cache_key(model, kwargs) + state = _normalize_responses_state(kwargs.get("responses_state")) + + if mode is TransportMode.CHAT_COMPLETIONS: + _drop_responses_only_kwargs(kwargs) + return cls( + mode=mode, + allow_fallback=allow_fallback, + cache_key=cache_key, + state=RESPONSES_STATE_OFF, + ) + + if ( + state == RESPONSES_STATE_PROVIDER + and cache_key in RESPONSES_STATE_UNSUPPORTED_CACHE + ): + kwargs["responses_state"] = RESPONSES_STATE_LOCAL + state = RESPONSES_STATE_LOCAL + + _filter_unsupported_builtin_tools(kwargs, cache_key) + + if _should_preserve_cache_control_on_chat(model, kwargs, messages or []): + return cls( + mode=TransportMode.CHAT_COMPLETIONS, + allow_fallback=allow_fallback, + cache_key=cache_key, + state=RESPONSES_STATE_OFF, + ) + + if cache_key in RESPONSES_UNSUPPORTED_CACHE: + return cls( + mode=TransportMode.CHAT_COMPLETIONS, + allow_fallback=allow_fallback, + fallback_error=RuntimeError("Responses API previously failed"), + cache_key=cache_key, + state=RESPONSES_STATE_OFF, + ) + + return cls( + mode=TransportMode.RESPONSES, + allow_fallback=allow_fallback, + cache_key=cache_key, + state=state, + ) + + @staticmethod + def _pop_mode(kwargs: dict[str, Any]) -> TransportMode: + value = str(kwargs.pop("a0_api_mode", "responses") or "").lower().strip() + if value in CHAT_COMPLETIONS_ALIASES: + return TransportMode.CHAT_COMPLETIONS + if value in RESPONSES_ALIASES: + return TransportMode.RESPONSES + return TransportMode.RESPONSES + + @property + def using_responses(self) -> bool: + return self.mode is TransportMode.RESPONSES + + def recover(self, exc: Exception, *, got_any_chunk: bool) -> TransportRecovery: + if not self.using_responses or got_any_chunk: + return TransportRecovery.RAISE + if not self.retried_reasoning and _is_responses_reasoning_effort_error(exc): + self.retried_reasoning = True + return TransportRecovery.RETRY_RESPONSES + if ( + self.state == RESPONSES_STATE_PROVIDER + and _is_responses_state_unsupported_error(exc) + ): + self.state = RESPONSES_STATE_LOCAL + self.state_fallback_error = exc + if self.cache_key: + RESPONSES_STATE_UNSUPPORTED_CACHE.add(self.cache_key) + return TransportRecovery.RETRY_LOCAL_RESPONSES + if self.allow_fallback and _is_responses_not_supported_error(exc): + self.mode = TransportMode.CHAT_COMPLETIONS + self.fallback_error = exc + self.state = RESPONSES_STATE_OFF + if self.cache_key: + RESPONSES_UNSUPPORTED_CACHE.add(self.cache_key) + return TransportRecovery.FALLBACK_TO_CHAT + return TransportRecovery.RAISE + + +@dataclass +class LiteLLMTransport: + model: str + messages: list[dict[str, Any]] + kwargs: dict[str, Any] + stop: Optional[list[str]] = None + policy: TransportPolicy = field(init=False) + last_result: LLMResult | None = field(init=False, default=None) + last_request_state: str = field(init=False, default=RESPONSES_STATE_PROVIDER) + explicit_prompt_caching: bool = field(init=False, default=False) + + def __post_init__(self) -> None: + self.kwargs = _without_stream_kwarg(dict(self.kwargs)) + self.explicit_prompt_caching = _coerce_bool( + self.kwargs.pop("a0_explicit_prompt_caching", False), default=False + ) + if self.explicit_prompt_caching: + self.messages = apply_chat_prompt_cache_markers( + self.messages, + model=self.model, + kwargs=self.kwargs, + ) + self.policy = TransportPolicy.from_request( + self.model, + self.kwargs, + messages=self.messages, + ) + + def complete(self) -> ChatChunk: + while True: + try: + if self.policy.mode is TransportMode.CHAT_COMPLETIONS: + parsed = ChatCompletionsTransport.parse( + completion(**self._chat_request(stream=False)) + ) + self.last_result = self._llm_result_from_chat(parsed) + return parsed + request = self._responses_request(stream=False) + raw_response = responses(**request) + parsed = ResponsesTransport.parse_response(raw_response) + self.last_result = self._llm_result_from_response( + raw_response, request + ) + return parsed + except Exception as exc: + if self._recover(exc, got_any_chunk=False): + continue + raise + + async def acomplete(self) -> ChatChunk: + while True: + try: + if self.policy.mode is TransportMode.CHAT_COMPLETIONS: + parsed = ChatCompletionsTransport.parse( + await acompletion(**self._chat_request(stream=False)) + ) + self.last_result = self._llm_result_from_chat(parsed) + return parsed + request = self._responses_request(stream=False) + raw_response = await aresponses(**request) + parsed = ResponsesTransport.parse_response(raw_response) + self.last_result = self._llm_result_from_response( + raw_response, request + ) + return parsed + except Exception as exc: + if self._recover(exc, got_any_chunk=False): + continue + raise + + def stream(self) -> Iterator[ChatChunk]: + while True: + iterator = None + exhausted = False + got_any_chunk = False + try: + if self.policy.mode is TransportMode.CHAT_COMPLETIONS: + iterator = completion(**self._chat_request(stream=True)) + for chunk in iterator: + parsed = ChatCompletionsTransport.parse(chunk) + if _has_chunk_delta(parsed): + got_any_chunk = True + yield parsed + else: + request = self._responses_request(stream=True) + iterator = responses(**request) + parser = ResponsesEventParser() + for event in iterator: + parsed = parser.parse(event) + if _has_chunk_delta(parsed): + got_any_chunk = True + yield parsed + self.last_result = self._stream_result_from_parser( + parser, request + ) + exhausted = True + return + except Exception as exc: + if self._recover(exc, got_any_chunk=got_any_chunk): + continue + raise + finally: + if iterator is not None and not exhausted: + _close_sync_stream(iterator) + + async def astream(self) -> AsyncIterator[ChatChunk]: + while True: + iterator = None + exhausted = False + got_any_chunk = False + try: + if self.policy.mode is TransportMode.CHAT_COMPLETIONS: + iterator = await acompletion(**self._chat_request(stream=True)) + async for chunk in iterator: # type: ignore[union-attr] + parsed = ChatCompletionsTransport.parse(chunk) + if _has_chunk_delta(parsed): + got_any_chunk = True + yield parsed + else: + request = self._responses_request(stream=True) + iterator = await aresponses(**request) + parser = ResponsesEventParser() + async for event in iterator: # type: ignore[union-attr] + parsed = parser.parse(event) + if _has_chunk_delta(parsed): + got_any_chunk = True + yield parsed + self.last_result = self._stream_result_from_parser( + parser, request + ) + exhausted = True + return + except Exception as exc: + if self._recover(exc, got_any_chunk=got_any_chunk): + continue + raise + finally: + if iterator is not None and not exhausted: + await _close_async_stream(iterator) + + def _recover(self, exc: Exception, *, got_any_chunk: bool) -> bool: + if ( + self.policy.using_responses + and not got_any_chunk + and self.kwargs.get("responses_builtin_tools") + and _is_responses_builtin_tool_error(exc) + ): + downgraded = _builtin_tool_types(self.kwargs.get("responses_builtin_tools")) + if downgraded: + RESPONSES_BUILTIN_UNSUPPORTED_CACHE.setdefault( + self.policy.cache_key, set() + ).update(downgraded) + self.kwargs["_a0_responses_builtin_downgrades"] = sorted(downgraded) + self.kwargs["responses_builtin_tools"] = [] + return True + + recovery = self.policy.recover(exc, got_any_chunk=got_any_chunk) + if recovery is TransportRecovery.RETRY_RESPONSES: + self.kwargs["reasoning"] = { + "effort": RESPONSES_REASONING_FALLBACK_EFFORT + } + return True + if recovery is TransportRecovery.RETRY_LOCAL_RESPONSES: + self.kwargs["responses_state"] = RESPONSES_STATE_LOCAL + self.kwargs.pop("previous_response_id", None) + return True + return recovery is TransportRecovery.FALLBACK_TO_CHAT + + def _chat_request(self, *, stream: bool) -> dict[str, Any]: + chat_kwargs = ChatCompletionsTransport.prepare_kwargs( + self.kwargs, + fallback_error=self.policy.fallback_error, + model=self.model, + messages=self.messages, + explicit_prompt_caching=self.explicit_prompt_caching, + ) + request = { + "model": self.model, + "messages": ChatCompletionsTransport.prepare_messages( + self.messages, + model=self.model, + kwargs=chat_kwargs, + ), + "stream": stream, + **chat_kwargs, + } + if self.stop is not None: + request["stop"] = self.stop + return request + + def _responses_request(self, *, stream: bool) -> dict[str, Any]: + response_kwargs = ResponsesTransport.from_chat( + self.messages, + self.kwargs, + stop=self.stop, + model=self.model, + ) + self.last_request_state = _normalize_responses_state( + self.kwargs.get("responses_state") + ) + return { + "model": self.model, + "stream": stream, + **response_kwargs, + } + + def _llm_result_from_chat(self, parsed: ChatChunk) -> LLMResult: + return LLMResult.from_chat( + response=parsed["response_delta"], + reasoning=parsed["reasoning_delta"], + input_items=ResponsesTransport.input_from_messages(self.messages), + provider_model_key=self.model, + capability=self._capability_metadata(), + ) + + def _llm_result_from_response( + self, response: Any, request: dict[str, Any] + ) -> LLMResult: + return LLMResult.from_response( + response, + input_items=_as_list(request.get("input")), + previous_response_id=str(request.get("previous_response_id") or ""), + provider_model_key=self.model, + mode=TransportMode.RESPONSES.value, + state=self.last_request_state, + capability=self._capability_metadata(), + ) + + def _stream_result_from_parser( + self, parser: "ResponsesEventParser", request: dict[str, Any] + ) -> LLMResult | None: + if parser.completed_response is None: + return None + return self._llm_result_from_response(parser.completed_response, request) + + def _capability_metadata(self) -> dict[str, Any]: + return { + "mode": self.policy.mode.value, + "state": self.policy.state, + "cache_key": self.policy.cache_key, + "fallback_error": _exception_text(self.policy.fallback_error) + if self.policy.fallback_error + else "", + "state_fallback_error": _exception_text(self.policy.state_fallback_error) + if self.policy.state_fallback_error + else "", + "builtin_tool_downgrades": list( + self.kwargs.get("_a0_responses_builtin_downgrades") or [] + ), + } + + +class ChatCompletionsTransport: + @staticmethod + def prepare_messages( + messages: list[dict[str, Any]], + *, + model: str = "", + kwargs: dict[str, Any] | None = None, + ) -> list[dict[str, Any]]: + if _is_openai_prompt_cache_provider(model, kwargs or {}): + stripped = _without_cache_control(messages) + return stripped if isinstance(stripped, list) else messages + return messages + + @staticmethod + def prepare_kwargs( + kwargs: dict[str, Any], + fallback_error: Exception | None = None, + *, + model: str = "", + messages: list[dict[str, Any]] | None = None, + explicit_prompt_caching: bool = False, + ) -> dict[str, Any]: + chat_kwargs = dict(kwargs) + _drop_internal_transport_kwargs(chat_kwargs) + if not _has_tools(chat_kwargs.get("tools")): + chat_kwargs.pop("tool_choice", None) + chat_kwargs.pop("parallel_tool_calls", None) + if _is_openai_prompt_cache_provider(model, chat_kwargs): + _prepare_openai_prompt_cache_params( + chat_kwargs, + messages or [], + model=model, + ) + elif _supports_cache_control_markers(model, chat_kwargs) and ( + explicit_prompt_caching or _has_cache_control(messages or []) + ): + _apply_tool_cache_control(chat_kwargs) + if fallback_error is not None: + chat_kwargs.setdefault("drop_params", True) + return {key: value for key, value in chat_kwargs.items() if value is not None} + + @staticmethod + def parse(chunk: Any) -> ChatChunk: + choice = _first_choice(chunk) + delta = _get_value(choice, "delta") or {} + message = _get_value(choice, "message") or _get_value( + _get_value(choice, "model_extra") or {}, "message" + ) or {} + response_delta = _get_value(delta, "content") or _get_value( + message, "content" + ) or "" + reasoning_delta = _get_value(delta, "reasoning_content") or _get_value( + message, "reasoning_content" + ) or "" + return {"reasoning_delta": reasoning_delta, "response_delta": response_delta} + + +class ResponsesTransport: + @classmethod + def from_chat( + cls, + messages: list[dict[str, Any]], + kwargs: dict[str, Any], + stop: Optional[list[str]] = None, + model: str = "", + ) -> dict[str, Any]: + request = cls.prepare_kwargs(kwargs, stop=stop, model=model, messages=messages) + state = _normalize_responses_state(kwargs.get("responses_state")) + input_items = cls._select_input_items(kwargs, messages, state) + request["input"] = input_items or "" + cls.apply_state(request, kwargs, state=state) + return request + + @classmethod + def from_input( + cls, + input_items: list[dict[str, Any]], + kwargs: dict[str, Any], + stop: Optional[list[str]] = None, + model: str = "", + messages: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + request = cls.prepare_kwargs(kwargs, stop=stop, model=model, messages=messages) + state = _normalize_responses_state(kwargs.get("responses_state")) + request["input"] = list(input_items or []) or "" + cls.apply_state(request, kwargs, state=state) + return request + + @classmethod + def prepare_kwargs( + cls, + kwargs: dict[str, Any], + stop: Optional[list[str]] = None, + model: str = "", + messages: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + request = dict(kwargs) + response_function_tools = request.pop("a0_responses_function_tools", None) + response_builtin_tools = request.pop("responses_builtin_tools", None) + _drop_responses_only_kwargs(request) + _drop_legacy_transport_kwargs(request) + request.pop("stop", None) + + max_completion_tokens = request.pop("max_completion_tokens", None) + max_tokens = request.pop("max_tokens", None) + if "max_output_tokens" not in request: + request["max_output_tokens"] = max_completion_tokens or max_tokens + + reasoning_effort = request.pop("reasoning_effort", None) + if "reasoning" in request: + request["reasoning"] = cls.normalize_reasoning(request["reasoning"]) + elif reasoning_effort is not None: + request["reasoning"] = cls.normalize_reasoning(reasoning_effort) + + response_format = request.pop("response_format", None) + if response_format is not None: + text_param, text_format = cls.text_from_response_format(response_format) + if text_param is not None and "text" not in request: + request["text"] = text_param + if text_format is not None and "text_format" not in request: + request["text_format"] = text_format + + functions = request.pop("functions", None) + if functions and "tools" not in request: + request["tools"] = [ + {"type": "function", **function} + for function in functions + if isinstance(function, dict) + ] + + tools = cls.tools_from_chat(request.pop("tools", None)) + tools = cls.merge_response_tools( + tools, + response_function_tools=response_function_tools, + response_builtin_tools=response_builtin_tools, + ) + if _has_tools(tools): + request["tools"] = tools + else: + request.pop("tools", None) + + function_call = request.pop("function_call", None) + if function_call is not None and "tool_choice" not in request: + request["tool_choice"] = cls.tool_choice_from_function_call(function_call) + elif "tool_choice" in request: + request["tool_choice"] = cls.tool_choice_from_chat(request["tool_choice"]) + + if not _has_tools(request.get("tools")): + request.pop("tool_choice", None) + request.pop("parallel_tool_calls", None) + + cls.prepare_prompt_caching(request, messages or [], model=model) + + _ = stop + return {key: value for key, value in request.items() if value is not None} + + @classmethod + def _select_input_items( + cls, + kwargs: dict[str, Any], + messages: list[dict[str, Any]], + state: str, + ) -> list[dict[str, Any]]: + provider_items = kwargs.get("responses_input_items") + local_items = kwargs.get("responses_local_input_items") + previous_response_id = kwargs.get("previous_response_id") + + if ( + state == RESPONSES_STATE_PROVIDER + and previous_response_id + and isinstance(provider_items, list) + ): + return [dict(item) for item in provider_items if isinstance(item, dict)] + + if state == RESPONSES_STATE_LOCAL and isinstance(local_items, list): + return [dict(item) for item in local_items if isinstance(item, dict)] + + return cls.input_from_messages(messages) + + @staticmethod + def apply_state( + request: dict[str, Any], + kwargs: dict[str, Any], + *, + state: str, + ) -> None: + if state == RESPONSES_STATE_PROVIDER: + request.setdefault("store", True) + previous_response_id = str(kwargs.get("previous_response_id") or "") + if previous_response_id: + request["previous_response_id"] = previous_response_id + elif state == RESPONSES_STATE_LOCAL: + request.setdefault("store", False) + else: + request.setdefault("store", False) + + @classmethod + def merge_response_tools( + cls, + tools: Any, + *, + response_function_tools: Any = None, + response_builtin_tools: Any = None, + ) -> list[Any]: + merged: list[Any] = [] + if isinstance(tools, list): + merged.extend(tools) + elif tools: + merged.append(tools) + + for source in (response_function_tools, response_builtin_tools): + for tool in _as_list(source): + normalized = cls.normalize_response_tool(tool) + if normalized: + merged.append(normalized) + return merged + + @staticmethod + def normalize_response_tool(tool: Any) -> dict[str, Any] | None: + if isinstance(tool, str): + tool = {"type": tool} + if not isinstance(tool, dict): + return None + return dict(tool) + + @staticmethod + def prepare_prompt_caching( + request: dict[str, Any], + messages: list[dict[str, Any]], + model: str = "", + ) -> None: + if not _is_openai_prompt_cache_provider(model, request): + return + + _prepare_openai_prompt_cache_params(request, messages, model=model) + + for key in OPENAI_RESPONSES_EXTRA_BODY_PARAMS: + if key not in request: + continue + extra_body = request.get("extra_body") + if not isinstance(extra_body, dict): + extra_body = {} + extra_body.setdefault(key, request.pop(key)) + request["extra_body"] = extra_body + + @classmethod + def input_from_messages( + cls, messages: list[dict[str, Any]] + ) -> list[dict[str, Any]]: + response_input: list[dict[str, Any]] = [] + + for message in messages: + role = str(message.get("role") or "user") + content = message.get("content", "") + + if role == "tool": + response_input.append( + { + "type": "function_call_output", + "call_id": str(message.get("tool_call_id") or ""), + "output": _content_to_text(content), + } + ) + continue + + tool_calls = message.get("tool_calls") + if role == "assistant" and isinstance(tool_calls, list) and tool_calls: + if _has_real_content(content): + response_input.append( + { + "role": "assistant", + "content": cls.content_from_chat(content, role=role), + } + ) + response_input.extend(cls.tool_calls_from_chat(tool_calls)) + continue + + response_input.append( + { + "role": role + if role in {"user", "assistant", "system", "developer"} + else "user", + "content": cls.content_from_chat(content, role=role), + } + ) + + return response_input + + @classmethod + def content_from_chat(cls, content: Any, role: str = "user") -> Any: + content = images.prepare_content(content) + if not isinstance(content, list): + return content + return [ + converted + for item in content + if (converted := cls.content_part_from_chat(item, role=role)) is not None + ] + + @staticmethod + def content_part_from_chat(item: Any, role: str = "user") -> Any: + if not isinstance(item, dict): + return item + + item_type = item.get("type") + if item_type in {"input_text", "output_text", "input_image", "input_file"}: + return dict(item) + if item_type == "text": + return { + "type": "output_text" if role == "assistant" else "input_text", + "text": item.get("text", ""), + } + if item_type == "image_url": + image_url = item.get("image_url") + if isinstance(image_url, dict): + url = image_url.get("url", "") + detail = image_url.get("detail") + else: + url = image_url or "" + detail = item.get("detail") + result = {"type": "input_image", "image_url": url} + if detail: + result["detail"] = detail + return result + + return dict(item) + + @staticmethod + def tool_calls_from_chat(tool_calls: list[Any]) -> list[dict[str, Any]]: + response_input: list[dict[str, Any]] = [] + for tool_call in tool_calls: + if not isinstance(tool_call, dict): + continue + function = tool_call.get("function") or {} + if not isinstance(function, dict): + function = {} + response_input.append( + { + "type": "function_call", + "call_id": str(tool_call.get("id") or ""), + "id": str(tool_call.get("id") or ""), + "name": str(function.get("name") or tool_call.get("name") or ""), + "arguments": str(function.get("arguments") or ""), + "status": "completed", + } + ) + return response_input + + @staticmethod + def tools_from_chat(tools: Any) -> Any: + if not isinstance(tools, list): + return tools + response_tools: list[Any] = [] + for tool in tools: + if not isinstance(tool, dict): + response_tools.append(tool) + continue + if tool.get("type") == "function" and isinstance( + tool.get("function"), dict + ): + function = tool["function"] + response_tool = { + "type": "function", + "name": function.get("name", ""), + "description": function.get("description", ""), + "parameters": function.get("parameters", {}), + } + if "strict" in function: + response_tool["strict"] = function["strict"] + response_tools.append(response_tool) + else: + response_tools.append(dict(tool)) + return response_tools + + @staticmethod + def tool_choice_from_function_call(function_call: Any) -> Any: + if isinstance(function_call, str): + return function_call + if isinstance(function_call, dict) and function_call.get("name"): + return {"type": "function", "name": function_call["name"]} + return function_call + + @staticmethod + def tool_choice_from_chat(tool_choice: Any) -> Any: + if ( + isinstance(tool_choice, dict) + and tool_choice.get("type") == "function" + and isinstance(tool_choice.get("function"), dict) + ): + return {"type": "function", "name": tool_choice["function"].get("name", "")} + return tool_choice + + @staticmethod + def text_from_response_format(response_format: Any) -> tuple[Any, Any]: + if isinstance(response_format, type): + return None, response_format + if not isinstance(response_format, dict): + return response_format, None + + format_type = response_format.get("type") + if format_type == "json_schema": + schema = response_format.get("json_schema") or {} + return ( + { + "format": { + "type": "json_schema", + "name": schema.get("name", "response_schema"), + "schema": schema.get("schema", {}), + "strict": schema.get("strict", False), + } + }, + None, + ) + if format_type: + return {"format": {"type": format_type}}, None + return response_format, None + + @staticmethod + def normalize_reasoning(reasoning: Any) -> Any: + if isinstance(reasoning, dict): + normalized = dict(reasoning) + if "effort" in normalized: + effort = _normalize_reasoning_effort(normalized.get("effort")) + if effort is None: + normalized.pop("effort", None) + else: + normalized["effort"] = effort + return normalized or None + if reasoning is None: + return None + effort = _normalize_reasoning_effort(reasoning) + return {"effort": effort} if effort is not None else None + + @classmethod + def parse_response(cls, response: Any) -> ChatChunk: + response_delta = cls.output_text(response) + reasoning_delta = cls.reasoning_text(response) + if not response_delta: + response_delta = cls.function_calls_text(response) + return {"reasoning_delta": reasoning_delta, "response_delta": response_delta} + + @classmethod + def parse_event(cls, event: Any) -> ChatChunk: + return ResponsesEventParser().parse(event) + + @classmethod + def output_text(cls, response: Any) -> str: + output_text = _get_value(response, "output_text") + if isinstance(output_text, str): + return output_text + + pieces: list[str] = [] + for item in _as_list(_get_value(response, "output")): + if _get_value(item, "type") != "message": + continue + for block in _as_list(_get_value(item, "content")): + block_type = _get_value(block, "type") + if block_type in {"output_text", "text"}: + text = _get_value(block, "text") + if isinstance(text, str): + pieces.append(text) + elif block_type == "refusal": + refusal = _get_value(block, "refusal") + if isinstance(refusal, str): + pieces.append(refusal) + return "".join(pieces) + + @staticmethod + def reasoning_text(response: Any) -> str: + pieces: list[str] = [] + for item in _as_list(_get_value(response, "output")): + if _get_value(item, "type") != "reasoning": + continue + for block in _as_list(_get_value(item, "summary")): + text = _get_value(block, "text") or _get_value(block, "reasoning") + if isinstance(text, str): + pieces.append(text) + return "".join(pieces) + + @classmethod + def function_calls_text(cls, response: Any) -> str: + calls = [ + cls.function_call_object(item) + for item in _as_list(_get_value(response, "output")) + ] + calls = [call for call in calls if call] + if not calls: + return "" + if len(calls) == 1: + return json.dumps(calls[0]) + return json.dumps( + {"tool_name": "parallel_tool_calls", "tool_args": {"calls": calls}} + ) + + @classmethod + def function_call_text(cls, item: Any) -> str: + call = cls.function_call_object(item) + if not call: + return "" + return json.dumps(call) + + @staticmethod + def function_call_object(item: Any) -> dict[str, Any]: + if _get_value(item, "type") != "function_call": + return {} + name = _get_value(item, "name") + if not name: + return {} + raw_arguments = _get_value(item, "arguments") or "{}" + if isinstance(raw_arguments, str): + try: + args = json.loads(raw_arguments or "{}") + except Exception: + args = {"arguments": raw_arguments} + elif isinstance(raw_arguments, dict): + args = raw_arguments + else: + args = {"arguments": raw_arguments} + if not isinstance(args, dict): + args = {"arguments": args} + return { + "tool_name": str(name), + "tool_args": args, + } + + +class ResponsesEventParser: + """Stateful parser for Responses streaming events.""" + + def __init__(self) -> None: + self.function_calls: dict[str, dict[str, Any]] = {} + self.output_index_keys: dict[str, str] = {} + self.emitted_function_calls: set[str] = set() + self.seen_response_delta = False + self.seen_reasoning_delta = False + self.completed_response: Any = None + + def parse(self, event: Any) -> ChatChunk: + event_type = _get_value(event, "type") or "" + response_delta = "" + reasoning_delta = "" + + if event_type in { + "response.output_text.delta", + "response.refusal.delta", + "response.text.delta", + }: + response_delta = str(_get_value(event, "delta") or "") + elif event_type in { + "response.reasoning_summary_text.delta", + "response.reasoning_text.delta", + }: + reasoning_delta = str(_get_value(event, "delta") or "") + elif event_type == "response.output_item.added": + self._remember_function_call(_get_value(event, "item"), event) + elif event_type == "response.function_call_arguments.delta": + self._append_function_call_arguments(event) + elif event_type == "response.function_call_arguments.done": + response_delta = self._complete_function_call(event) + elif event_type == "response.output_item.done": + response_delta = self._complete_output_item(_get_value(event, "item"), event) + elif event_type == "response.completed": + response_delta, reasoning_delta = self._complete_response(event) + elif event_type == "response.failed": + raise RuntimeError(self._response_error_message(event)) + elif event_type == "error": + error = _get_value(event, "error") + message = _get_value(error, "message") or error + raise RuntimeError(str(message)) + + if response_delta: + self.seen_response_delta = True + if reasoning_delta: + self.seen_reasoning_delta = True + + return {"reasoning_delta": reasoning_delta, "response_delta": response_delta} + + def _remember_function_call(self, item: Any, event: Any) -> str: + if _get_value(item, "type") != "function_call": + return "" + key = self._event_key(event, item) + if not key: + return "" + current = self.function_calls.get(key, {}) + merged = {**current, **_object_to_dict(item)} + self.function_calls[key] = merged + output_index = _get_value(event, "output_index") + if output_index is not None: + self.output_index_keys[str(output_index)] = key + return key + + def _append_function_call_arguments(self, event: Any) -> None: + key = self._event_key(event) + if not key: + return + current = self.function_calls.setdefault(key, {"type": "function_call"}) + current["arguments"] = str(current.get("arguments") or "") + str( + _get_value(event, "delta") or "" + ) + + def _complete_function_call(self, event: Any) -> str: + key = self._event_key(event) + if not key: + return "" + current = self.function_calls.setdefault(key, {"type": "function_call"}) + if _get_value(event, "arguments") is not None: + current["arguments"] = _get_value(event, "arguments") + if _get_value(event, "name"): + current["name"] = _get_value(event, "name") + return self._emit_function_call(key, current) + + def _complete_output_item(self, item: Any, event: Any) -> str: + key = self._remember_function_call(item, event) + if not key: + return "" + return self._emit_function_call(key, self.function_calls[key]) + + def _complete_response(self, event: Any) -> tuple[str, str]: + self.completed_response = _get_value(event, "response") + if self.seen_response_delta or self.emitted_function_calls: + return "", "" + parsed = ResponsesTransport.parse_response(self.completed_response) + if self.seen_reasoning_delta: + parsed["reasoning_delta"] = "" + return parsed["response_delta"], parsed["reasoning_delta"] + + def _emit_function_call(self, key: str, item: Any) -> str: + if key in self.emitted_function_calls: + return "" + text = ResponsesTransport.function_call_text(item) + if text: + self.emitted_function_calls.add(key) + return text + + def _event_key(self, event: Any, item: Any = None) -> str: + key = _get_value(event, "item_id") or _get_value(item, "id") + if key: + return str(key) + output_index = _get_value(event, "output_index") + if output_index is not None: + output_key = self.output_index_keys.get(str(output_index)) + if output_key: + return output_key + return f"output:{output_index}" + return "" + + def _response_error_message(self, event: Any) -> str: + response = _get_value(event, "response") or {} + error = _get_value(response, "error") or _get_value(event, "error") + message = _get_value(error, "message") or error + return str(message or "Responses API request failed") + + +def clear_transport_capability_cache() -> None: + RESPONSES_UNSUPPORTED_CACHE.clear() + RESPONSES_STATE_UNSUPPORTED_CACHE.clear() + RESPONSES_BUILTIN_UNSUPPORTED_CACHE.clear() + + +def delete_stored_response_ids( + response_ids: list[str], **kwargs: Any +) -> list[tuple[str, str]]: + errors: list[tuple[str, str]] = [] + for response_id in response_ids: + try: + delete_responses(response_id=response_id, **kwargs) + except Exception as exc: + errors.append((response_id, _exception_text(exc))) + return errors + + +async def adelete_stored_response_ids( + response_ids: list[str], **kwargs: Any +) -> list[tuple[str, str]]: + errors: list[tuple[str, str]] = [] + for response_id in response_ids: + try: + await adelete_responses(response_id=response_id, **kwargs) + except Exception as exc: + errors.append((response_id, _exception_text(exc))) + return errors + + +def _coerce_bool(value: Any, default: bool = False) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off", "none"}: + return False + return bool(value) + + +def _responses_cache_key(model: str, kwargs: dict[str, Any]) -> str: + api_base = ( + kwargs.get("api_base") + or kwargs.get("base_url") + or kwargs.get("api_base_url") + or "" + ) + custom_provider = kwargs.get("custom_llm_provider") or "" + return "|".join(str(part) for part in (model, custom_provider, api_base)) + + +def _drop_legacy_transport_kwargs(kwargs: dict[str, Any]) -> None: + kwargs.pop("a0_api_mode", None) + kwargs.pop("a0_responses_fallback", None) + + +def _drop_responses_only_kwargs(kwargs: dict[str, Any]) -> None: + kwargs.pop("responses_state", None) + kwargs.pop("responses_delete_on_chat_delete", None) + kwargs.pop("responses_input_items", None) + kwargs.pop("responses_local_input_items", None) + kwargs.pop("previous_response_id", None) + kwargs.pop("_a0_responses_builtin_downgrades", None) + + +def _drop_internal_transport_kwargs(kwargs: dict[str, Any]) -> None: + _drop_legacy_transport_kwargs(kwargs) + kwargs.pop("a0_explicit_prompt_caching", None) + kwargs.pop("a0_responses_function_tools", None) + kwargs.pop("responses_builtin_tools", None) + _drop_responses_only_kwargs(kwargs) + + +def _normalize_responses_state(value: Any) -> str: + normalized = str(value or RESPONSES_STATE_PROVIDER).strip().lower() + return normalized if normalized in RESPONSES_STATES else RESPONSES_STATE_PROVIDER + + +def _filter_unsupported_builtin_tools(kwargs: dict[str, Any], cache_key: str) -> None: + unsupported = RESPONSES_BUILTIN_UNSUPPORTED_CACHE.get(cache_key) + if not unsupported: + return + tools = kwargs.get("responses_builtin_tools") + if not isinstance(tools, list): + return + filtered = [] + downgraded = [] + for tool in tools: + tool_type = _response_tool_type(tool) + if tool_type in unsupported: + downgraded.append(tool_type) + continue + filtered.append(tool) + if downgraded: + kwargs["responses_builtin_tools"] = filtered + kwargs["_a0_responses_builtin_downgrades"] = sorted(set(downgraded)) + + +def _builtin_tool_types(tools: Any) -> set[str]: + return { + tool_type + for tool in _as_list(tools) + if (tool_type := _response_tool_type(tool)) + } + + +def _response_tool_type(tool: Any) -> str: + if isinstance(tool, str): + return tool + if isinstance(tool, dict): + return str(tool.get("type") or "") + return "" + + +def apply_chat_prompt_cache_markers( + messages: list[dict[str, Any]], + *, + model: str = "", + kwargs: dict[str, Any] | None = None, +) -> list[dict[str, Any]]: + if not _supports_cache_control_markers(model, kwargs or {}): + return [dict(message) for message in messages] + + prepared = [_strip_message_cache_control(message) for message in messages] + for index in _prompt_cache_message_indexes(prepared): + prepared[index] = _message_with_cache_control(prepared[index]) + return prepared + + +def _prompt_cache_message_indexes(messages: list[dict[str, Any]]) -> list[int]: + indexes: list[int] = [] + + leading_context: list[int] = [] + for index, message in enumerate(messages): + role = str(message.get("role") or "") + if role in {"system", "developer"}: + leading_context.append(index) + continue + break + if leading_context: + indexes.append(leading_context[-1]) + + user_indexes = [ + index + for index, message in enumerate(messages) + if str(message.get("role") or "") == "user" + ] + indexes.extend(user_indexes[-2:]) + + deduplicated: list[int] = [] + for index in indexes: + if index not in deduplicated: + deduplicated.append(index) + return deduplicated[:3] + + +def _strip_message_cache_control(message: dict[str, Any]) -> dict[str, Any]: + result = dict(message) + result.pop("cache_control", None) + return result + + +def _message_with_cache_control(message: dict[str, Any]) -> dict[str, Any]: + result = dict(message) + result["content"] = _content_with_cache_control(result.get("content", "")) + return result + + +def _content_with_cache_control(content: Any) -> Any: + marker = _cache_control_marker() + if isinstance(content, list): + blocks = [_copy_content_block(block) for block in content] + if not blocks: + return [{"type": "text", "text": "", "cache_control": marker}] + for index in range(len(blocks) - 1, -1, -1): + block = blocks[index] + if isinstance(block, dict): + block["cache_control"] = marker + return blocks + if isinstance(block, str): + blocks[index] = { + "type": "text", + "text": block, + "cache_control": marker, + } + return blocks + return blocks + + if isinstance(content, dict): + block = dict(content) + block["cache_control"] = marker + return block + + return [ + { + "type": "text", + "text": _content_to_text(content), + "cache_control": marker, + } + ] + + +def _copy_content_block(block: Any) -> Any: + if isinstance(block, dict): + return dict(block) + if isinstance(block, list): + return [_copy_content_block(item) for item in block] + return block + + +def _apply_tool_cache_control(kwargs: dict[str, Any]) -> None: + tools = kwargs.get("tools") + if not isinstance(tools, list) or not tools or _has_cache_control(tools): + return + + prepared = [dict(tool) if isinstance(tool, dict) else tool for tool in tools] + for index in range(len(prepared) - 1, -1, -1): + tool = prepared[index] + if not isinstance(tool, dict): + continue + if tool.get("type") == "function" and isinstance(tool.get("function"), dict): + function = dict(tool["function"]) + function["cache_control"] = _cache_control_marker() + tool = dict(tool) + tool["function"] = function + prepared[index] = tool + else: + tool = dict(tool) + tool["cache_control"] = _cache_control_marker() + prepared[index] = tool + kwargs["tools"] = prepared + return + + +def _cache_control_marker() -> dict[str, str]: + return {"type": "ephemeral"} + + +def _should_preserve_cache_control_on_chat( + model: str, + kwargs: dict[str, Any], + messages: list[dict[str, Any]], +) -> bool: + if not (_has_cache_control(messages) or _has_cache_control(kwargs.get("tools"))): + return False + return not _is_native_responses_provider(model, kwargs) + + +def _is_native_responses_provider(model: str, kwargs: dict[str, Any]) -> bool: + api_base = _api_base(kwargs) + if "openrouter.ai" in api_base or "anthropic.com" in api_base: + return False + + provider = _normalized_provider(model, kwargs) + return provider in {"openai", "azure", "azure_ai", "xai"} + + +def _supports_cache_control_markers( + model: str, + kwargs: dict[str, Any], +) -> bool: + api_base = _api_base(kwargs) + if "openrouter.ai" in api_base or "anthropic.com" in api_base: + return True + provider = _normalized_provider(model, kwargs) + return provider in CACHE_CONTROL_PROMPT_PROVIDERS + + +def _is_openai_prompt_cache_provider(model: str, kwargs: dict[str, Any]) -> bool: + api_base = _api_base(kwargs) + provider = _normalized_provider(model, kwargs) + + if api_base: + if "api.openai.com" in api_base: + return provider in {"", "openai"} + if "openai.azure.com" in api_base: + return provider in {"", "azure", "openai"} + return False + + if not provider and str(model): + provider = "openai" + return provider in OPENAI_PROMPT_CACHE_PROVIDERS + + +def _api_base(kwargs: dict[str, Any]) -> str: + return str( + kwargs.get("api_base") + or kwargs.get("base_url") + or kwargs.get("api_base_url") + or "" + ).lower() + + +def _normalized_provider(model: str, kwargs: dict[str, Any]) -> str: + provider = str(kwargs.get("custom_llm_provider") or "").strip().lower() + if not provider and "/" in str(model): + provider = str(model).split("/", 1)[0].strip().lower() + return provider.replace("-", "_") + + +def _prepare_openai_prompt_cache_params( + request: dict[str, Any], + messages: list[dict[str, Any]], + *, + model: str = "", +) -> None: + if "prompt_cache_key" in request: + return + + sanitized_messages = _without_cache_control(messages) + sanitized_request = _without_cache_control(request) + prompt_cache_key = _default_prompt_cache_key( + model, + sanitized_messages if isinstance(sanitized_messages, list) else messages, + sanitized_request if isinstance(sanitized_request, dict) else request, + ) + if prompt_cache_key: + request["prompt_cache_key"] = prompt_cache_key + + +def _default_prompt_cache_key( + model: str, + messages: list[dict[str, Any]], + request: dict[str, Any], +) -> str: + material = _prompt_cache_key_material(messages, request) + if not material: + return "" + digest = hashlib.sha256( + json.dumps( + { + "model": model, + "material": material, + }, + sort_keys=True, + default=str, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest()[:32] + return f"a0-{digest}" + + +def _prompt_cache_key_material( + messages: list[dict[str, Any]], + request: dict[str, Any], +) -> dict[str, Any]: + material: dict[str, Any] = {} + + leading_messages: list[dict[str, Any]] = [] + for message in messages: + role = str(message.get("role") or "") + if role not in {"system", "developer"}: + break + leading_messages.append( + { + "role": role, + "content": message.get("content"), + } + ) + if leading_messages: + material["messages"] = leading_messages + + if request.get("instructions"): + material["instructions"] = request["instructions"] + if request.get("prompt"): + material["prompt"] = request["prompt"] + if request.get("tools"): + material["tools"] = request["tools"] + + return material + + +def _has_cache_control(value: Any) -> bool: + if isinstance(value, dict): + if value.get("cache_control") is not None: + return True + return any(_has_cache_control(item) for item in value.values()) + if isinstance(value, list): + return any(_has_cache_control(item) for item in value) + return False + + +def _without_cache_control(value: Any) -> Any: + if isinstance(value, dict): + return { + key: _without_cache_control(item) + for key, item in value.items() + if key != "cache_control" + } + if isinstance(value, list): + return [_without_cache_control(item) for item in value] + return value + + +def _object_to_dict(obj: Any) -> dict[str, Any]: + if isinstance(obj, dict): + return dict(obj) + if hasattr(obj, "model_dump"): + dumped = obj.model_dump() + return dict(dumped) if isinstance(dumped, dict) else {} + if hasattr(obj, "dict"): + dumped = obj.dict() + return dict(dumped) if isinstance(dumped, dict) else {} + return {} + + +def _normalize_reasoning_effort(effort: Any) -> str | None: + if isinstance(effort, str): + normalized = effort.strip().lower() + else: + normalized = str(effort).strip().lower() if effort is not None else "" + if normalized in RESPONSES_REASONING_EFFORTS: + return normalized + if normalized in NO_REASONING_EFFORT_ALIASES: + return None + return RESPONSES_REASONING_FALLBACK_EFFORT + + +def _is_responses_reasoning_effort_error(exc: Exception) -> bool: + text = _exception_text(exc).lower() + return ( + "response.reasoning.effort" in text + and "minimal" in text + and "high" in text + and "none" in text + ) + + +def _is_responses_not_supported_error(exc: Exception) -> bool: + text = _exception_text(exc).lower() + if any(marker in text for marker in ("429", "too many requests", "rate limit")): + return False + if "/v1/responses" in text and any( + marker in text for marker in ("404", "not found") + ): + return True + return any( + marker in text + for marker in ( + "responses api", + "does not support responses", + "not support responses", + "unsupportedparamserror", + "does not support parameters", + "no 'tools' defined while 'tool_choice' is specified", + ) + ) + + +def _is_responses_state_unsupported_error(exc: Exception) -> bool: + text = _exception_text(exc).lower() + if any(marker in text for marker in ("429", "too many requests", "rate limit")): + return False + if "404" in text and "/v1/responses/" in text: + return True + return any( + marker in text + for marker in ( + "previous_response_id", + "store", + "stored response", + "response not found", + "no response found", + "does not support response storage", + "doesn't support response storage", + "response storage is not supported", + "state is not supported", + ) + ) + + +def _is_responses_builtin_tool_error(exc: Exception) -> bool: + text = _exception_text(exc).lower() + if any(marker in text for marker in ("429", "too many requests", "rate limit")): + return False + return any( + marker in text + for marker in ( + "unsupported tool", + "unsupported tools", + "invalid tool", + "tool type", + "tools[", + "web_search", + "file_search", + "code_interpreter", + "image_generation", + "computer_use_preview", + "mcp", + ) + ) + + +def _exception_text(exc: Exception | None) -> str: + if exc is None: + return "" + parts = [exc.__class__.__name__, str(exc)] + cause = getattr(exc, "__cause__", None) + context = getattr(exc, "__context__", None) + if cause is not None: + parts.append(str(cause)) + if context is not None and context is not cause: + parts.append(str(context)) + return "\n".join(parts) + + +def _close_sync_stream(stream: Any) -> None: + for method_name in ("close", "aclose"): + close = getattr(stream, method_name, None) + if close is None: + continue + result = close() + if inspect.isawaitable(result): + result.close() + return + + +async def _close_async_stream(stream: Any) -> None: + for method_name in ("aclose", "close"): + close = getattr(stream, method_name, None) + if close is None: + continue + result = close() + if inspect.isawaitable(result): + await result + return + + +def _without_stream_kwarg(kwargs: dict[str, Any]) -> dict[str, Any]: + kwargs.pop("stream", None) + return kwargs + + +def _first_choice(chunk: Any) -> Any: + choices = _get_value(chunk, "choices") or [] + return choices[0] if choices else {} + + +def _get_value(obj: Any, key: str) -> Any: + if isinstance(obj, dict): + return obj.get(key) + return getattr(obj, key, None) + + +def _as_list(value: Any) -> list[Any]: + return value if isinstance(value, list) else [] + + +def _has_tools(tools: Any) -> bool: + if isinstance(tools, list): + return bool(tools) + return bool(tools) + + +def _has_chunk_delta(chunk: ChatChunk) -> bool: + return bool(chunk.get("response_delta") or chunk.get("reasoning_delta")) + + +def _has_real_content(content: Any) -> bool: + if content == "empty": + return False + if isinstance(content, str): + return bool(content.strip()) + if isinstance(content, list): + return len(content) > 0 + return content is not None + + +def _content_to_text(content: Any) -> str: + content = images.prepare_content(content) + if isinstance(content, str): + return content + if isinstance(content, list): + pieces: list[str] = [] + for item in content: + if isinstance(item, str): + pieces.append(item) + elif isinstance(item, dict): + text = item.get("text") + if isinstance(text, str): + pieces.append(text) + return "\n".join(pieces) + return "" if content is None else str(content) diff --git a/helpers/llm_result.py b/helpers/llm_result.py new file mode 100644 index 000000000..61e996c95 --- /dev/null +++ b/helpers/llm_result.py @@ -0,0 +1,311 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +import json +from typing import Any + + +RESPONSE_METADATA_KEY = "responses" +LOCAL_FUNCTION_TOOL_TYPES = {"function_call"} +TEXT_OUTPUT_TYPES = {"message"} +REASONING_OUTPUT_TYPES = {"reasoning"} + + +@dataclass +class ResponseItem: + type: str + data: dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_any(cls, item: Any) -> "ResponseItem": + data = object_to_dict(item) + return cls(type=str(data.get("type") or ""), data=data) + + def to_dict(self) -> dict[str, Any]: + return dict(self.data) + + +@dataclass +class ResponseFunctionCall: + name: str + arguments: dict[str, Any] + call_id: str + item_id: str = "" + raw: dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_item(cls, item: ResponseItem) -> "ResponseFunctionCall | None": + if item.type != "function_call": + return None + name = str(item.data.get("name") or "") + if not name: + return None + return cls( + name=name, + arguments=parse_arguments(item.data.get("arguments")), + call_id=str(item.data.get("call_id") or item.data.get("id") or ""), + item_id=str(item.data.get("id") or ""), + raw=dict(item.data), + ) + + +@dataclass +class LLMResult: + response: str = "" + reasoning: str = "" + response_id: str = "" + previous_response_id: str = "" + input_items: list[dict[str, Any]] = field(default_factory=list) + output_items: list[ResponseItem] = field(default_factory=list) + provider_model_key: str = "" + mode: str = "responses" + state: str = "provider" + usage: dict[str, Any] = field(default_factory=dict) + raw: dict[str, Any] = field(default_factory=dict) + capability: dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_dict(cls, data: dict[str, Any] | None) -> "LLMResult": + data = data or {} + return cls( + response=str(data.get("response") or ""), + reasoning=str(data.get("reasoning") or ""), + response_id=str(data.get("response_id") or ""), + previous_response_id=str(data.get("previous_response_id") or ""), + input_items=list(data.get("input_items") or []), + output_items=[ + ResponseItem.from_any(item) for item in data.get("output_items") or [] + ], + provider_model_key=str(data.get("provider_model_key") or ""), + mode=str(data.get("mode") or "responses"), + state=str(data.get("state") or "provider"), + usage=object_to_dict(data.get("usage") or {}), + raw=object_to_dict(data.get("raw") or {}), + capability=object_to_dict(data.get("capability") or {}), + ) + + @classmethod + def from_response( + cls, + response: Any, + *, + input_items: list[dict[str, Any]] | None = None, + previous_response_id: str = "", + provider_model_key: str = "", + mode: str = "responses", + state: str = "provider", + capability: dict[str, Any] | None = None, + ) -> "LLMResult": + raw = object_to_dict(response) + output_items = [ResponseItem.from_any(item) for item in as_list(raw.get("output"))] + result = cls( + response_id=str(raw.get("id") or ""), + previous_response_id=str( + raw.get("previous_response_id") or previous_response_id or "" + ), + input_items=list(input_items or []), + output_items=output_items, + provider_model_key=provider_model_key, + mode=mode, + state=state, + usage=object_to_dict(raw.get("usage") or {}), + raw=raw, + capability=dict(capability or {}), + ) + result.response = output_text(raw, output_items) + result.reasoning = reasoning_text(output_items) + if not result.response and result.function_calls: + result.response = result.function_calls_text() + return result + + @classmethod + def from_chat( + cls, + *, + response: str, + reasoning: str = "", + input_items: list[dict[str, Any]] | None = None, + provider_model_key: str = "", + capability: dict[str, Any] | None = None, + ) -> "LLMResult": + output_items = [] + if response: + output_items.append( + ResponseItem( + type="message", + data={ + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": response}], + }, + ) + ) + if reasoning: + output_items.insert( + 0, + ResponseItem( + type="reasoning", + data={ + "type": "reasoning", + "summary": [{"type": "summary_text", "text": reasoning}], + }, + ), + ) + return cls( + response=response, + reasoning=reasoning, + input_items=list(input_items or []), + output_items=output_items, + provider_model_key=provider_model_key, + mode="chat_completions", + state="off", + capability=dict(capability or {}), + ) + + @property + def function_calls(self) -> list[ResponseFunctionCall]: + calls: list[ResponseFunctionCall] = [] + for item in self.output_items: + call = ResponseFunctionCall.from_item(item) + if call: + calls.append(call) + return calls + + @property + def builtin_items(self) -> list[ResponseItem]: + return [ + item + for item in self.output_items + if item.type + and item.type not in TEXT_OUTPUT_TYPES + and item.type not in REASONING_OUTPUT_TYPES + and item.type not in LOCAL_FUNCTION_TOOL_TYPES + ] + + def function_calls_text(self) -> str: + calls = [ + {"tool_name": call.name, "tool_args": call.arguments} + for call in self.function_calls + ] + if not calls: + return "" + if len(calls) == 1: + return json.dumps(calls[0]) + return json.dumps( + {"tool_name": "parallel_tool_calls", "tool_args": {"calls": calls}} + ) + + def to_dict(self) -> dict[str, Any]: + return { + "response": self.response, + "reasoning": self.reasoning, + "response_id": self.response_id, + "previous_response_id": self.previous_response_id, + "input_items": self.input_items, + "output_items": [item.to_dict() for item in self.output_items], + "provider_model_key": self.provider_model_key, + "mode": self.mode, + "state": self.state, + "usage": self.usage, + "raw": self.raw, + "capability": self.capability, + } + + def metadata(self) -> dict[str, Any]: + return {RESPONSE_METADATA_KEY: self.to_dict()} + + +def function_call_output_item( + call_id: str, + output: str, + *, + acknowledged_safety_checks: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + item: dict[str, Any] = { + "type": "function_call_output", + "call_id": str(call_id or ""), + "output": output, + } + if acknowledged_safety_checks: + item["acknowledged_safety_checks"] = acknowledged_safety_checks + return item + + +def metadata_from_llm_result(result: LLMResult | None) -> dict[str, Any]: + return result.metadata() if result else {} + + +def result_from_metadata(metadata: dict[str, Any] | None) -> LLMResult | None: + if not isinstance(metadata, dict): + return None + data = metadata.get(RESPONSE_METADATA_KEY) + if not isinstance(data, dict): + return None + return LLMResult.from_dict(data) + + +def object_to_dict(obj: Any) -> dict[str, Any]: + if isinstance(obj, dict): + return dict(obj) + if hasattr(obj, "model_dump"): + dumped = obj.model_dump() + return dict(dumped) if isinstance(dumped, dict) else {} + if hasattr(obj, "dict"): + dumped = obj.dict() + return dict(dumped) if isinstance(dumped, dict) else {} + return {} + + +def as_list(value: Any) -> list[Any]: + return value if isinstance(value, list) else [] + + +def output_text(raw: dict[str, Any], output_items: list[ResponseItem]) -> str: + direct = raw.get("output_text") + if isinstance(direct, str): + return direct + pieces: list[str] = [] + for item in output_items: + if item.type != "message": + continue + for block in as_list(item.data.get("content")): + if not isinstance(block, dict): + continue + block_type = block.get("type") + if block_type in {"output_text", "text", "input_text"}: + text = block.get("text") + if isinstance(text, str): + pieces.append(text) + elif block_type == "refusal": + refusal = block.get("refusal") + if isinstance(refusal, str): + pieces.append(refusal) + return "".join(pieces) + + +def reasoning_text(output_items: list[ResponseItem]) -> str: + pieces: list[str] = [] + for item in output_items: + if item.type != "reasoning": + continue + for block in as_list(item.data.get("summary")): + if isinstance(block, dict): + text = block.get("text") or block.get("reasoning") + if isinstance(text, str): + pieces.append(text) + elif isinstance(block, str): + pieces.append(block) + return "".join(pieces) + + +def parse_arguments(raw_arguments: Any) -> dict[str, Any]: + if isinstance(raw_arguments, dict): + return raw_arguments + if isinstance(raw_arguments, str): + try: + parsed = json.loads(raw_arguments or "{}") + except Exception: + parsed = {"arguments": raw_arguments} + else: + parsed = {"arguments": raw_arguments} + return parsed if isinstance(parsed, dict) else {"arguments": parsed} diff --git a/helpers/persist_chat.py b/helpers/persist_chat.py index 086a68fa7..d64a9b5db 100644 --- a/helpers/persist_chat.py +++ b/helpers/persist_chat.py @@ -4,6 +4,7 @@ from typing import Any import uuid from agent import Agent, AgentConfig, AgentContext, AgentContextType from helpers import files, history +from helpers.litellm_transport import delete_stored_response_ids from helpers.localization import Localization import json from initialize import initialize_agent @@ -118,6 +119,7 @@ def export_json_chat(context: AgentContext): def remove_chat(ctxid): """Remove a chat or task context""" + _delete_provider_responses_for_chat(ctxid) path = get_chat_folder_path(ctxid) files.delete_dir(path) @@ -324,3 +326,70 @@ def _safe_json_serialize(obj, **kwargs): return False return json.dumps(obj, default=serializer, **kwargs) + + +def _delete_provider_responses_for_chat(ctxid: str) -> None: + try: + data = json.loads(files.read_file(_get_chat_file_path(ctxid))) + except Exception: + return + if _responses_delete_disabled(data): + return + response_ids = _collect_response_ids(data) + if not response_ids: + return + delete_stored_response_ids(response_ids) + + +def _responses_delete_disabled(data: dict[str, Any]) -> bool: + if data.get("responses_delete_on_chat_delete") is False: + return True + context_data = data.get("data") + if isinstance(context_data, dict) and context_data.get("responses_delete_on_chat_delete") is False: + return True + for agent_data in data.get("agents", []) or []: + if not isinstance(agent_data, dict): + continue + state = agent_data.get("data") + if isinstance(state, dict) and state.get("responses_delete_on_chat_delete") is False: + return True + return False + + +def _collect_response_ids(data: Any) -> list[str]: + found: list[str] = [] + seen: set[str] = set() + + def add(value: Any) -> None: + response_id = str(value or "").strip() + if response_id and response_id not in seen: + seen.add(response_id) + found.append(response_id) + + def walk(obj: Any) -> None: + if isinstance(obj, dict): + state = obj.get(Agent.DATA_NAME_RESPONSES_STATE) + if isinstance(state, dict): + add(state.get("response_id")) + for response_id in state.get("response_ids") or []: + add(response_id) + + metadata = obj.get("metadata") + if isinstance(metadata, dict): + responses = metadata.get("responses") + if isinstance(responses, dict): + add(responses.get("response_id")) + + for value in obj.values(): + walk(value) + elif isinstance(obj, list): + for value in obj: + walk(value) + elif isinstance(obj, str) and '"response_id"' in obj: + try: + walk(json.loads(obj)) + except Exception: + return + + walk(data) + return found diff --git a/helpers/responses_tools.py b/helpers/responses_tools.py new file mode 100644 index 000000000..a18f68baa --- /dev/null +++ b/helpers/responses_tools.py @@ -0,0 +1,231 @@ +from __future__ import annotations + +import hashlib +import json +import os +import re +from typing import Any + +from helpers import files, subagents + + +FUNCTION_NAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,64}$") +TOOL_PROMPT_PREFIX = "agent.system.tool." +TOOL_PROMPT_SUFFIX = ".md" +MAX_TOOL_DESCRIPTION_CHARS = 1024 + + +def build_responses_function_tools(agent: Any) -> tuple[list[dict[str, Any]], dict[str, str]]: + """Build permissive Responses function tools from A0 tool prompts and MCP schemas.""" + + tools: list[dict[str, Any]] = [] + name_map: dict[str, str] = {} + + for tool_name, prompt in _local_tool_prompts(agent): + native_name = _native_tool_name(tool_name) + name_map[native_name] = tool_name + tools.append( + { + "type": "function", + "name": native_name, + "description": _description_from_prompt(prompt, fallback=tool_name), + "parameters": _schema_from_prompt(prompt), + } + ) + + for tool_name, tool in _mcp_tools(agent): + native_name = _native_tool_name(tool_name) + name_map[native_name] = tool_name + tools.append( + { + "type": "function", + "name": native_name, + "description": _truncate(str(tool.get("description") or tool_name)), + "parameters": _schema_from_any(tool.get("input_schema")), + } + ) + + return _dedupe_tools(tools), name_map + + +def original_tool_name(native_name: str, name_map: dict[str, str] | None) -> str: + if not name_map: + return native_name + return name_map.get(native_name, native_name) + + +def _local_tool_prompts(agent: Any) -> list[tuple[str, str]]: + prompt_dirs = subagents.get_paths(agent, "prompts") + tool_files = files.get_unique_filenames_in_dirs( + prompt_dirs, f"{TOOL_PROMPT_PREFIX}*{TOOL_PROMPT_SUFFIX}" + ) + result: list[tuple[str, str]] = [] + for tool_file in tool_files: + basename = os.path.basename(tool_file) + tool_name = _tool_name_from_prompt_basename(basename) + if not tool_name: + continue + try: + prompt = agent.read_prompt(basename) + except Exception: + try: + prompt = files.read_file(tool_file) + except Exception: + prompt = "" + result.append((tool_name, prompt)) + return result + + +def _mcp_tools(agent: Any) -> list[tuple[str, dict[str, Any]]]: + try: + import helpers.mcp_handler as mcp_helper + + raw_tools = mcp_helper.MCPConfig.get_instance().get_tools() + except Exception: + return [] + + result: list[tuple[str, dict[str, Any]]] = [] + for entry in raw_tools or []: + if not isinstance(entry, dict): + continue + for tool_name, tool in entry.items(): + if isinstance(tool, dict): + result.append((str(tool_name), tool)) + return result + + +def _tool_name_from_prompt_basename(basename: str) -> str: + if not basename.startswith(TOOL_PROMPT_PREFIX) or not basename.endswith(TOOL_PROMPT_SUFFIX): + return "" + name = basename[len(TOOL_PROMPT_PREFIX) : -len(TOOL_PROMPT_SUFFIX)] + if not name or name in {"tools", "tools_vision"}: + return "" + return name + + +def _native_tool_name(tool_name: str) -> str: + if FUNCTION_NAME_PATTERN.fullmatch(tool_name): + return tool_name + slug = re.sub(r"[^A-Za-z0-9_-]+", "_", tool_name).strip("_") + digest = hashlib.sha1(tool_name.encode("utf-8")).hexdigest()[:8] + native = f"{slug[:52]}_{digest}" if slug else f"a0_tool_{digest}" + return native[:64] + + +def _description_from_prompt(prompt: str, *, fallback: str) -> str: + lines: list[str] = [] + in_fence = False + for raw_line in (prompt or "").splitlines(): + line = raw_line.strip() + if line.startswith("```"): + in_fence = not in_fence + continue + if in_fence or not line: + continue + if line.startswith("#"): + line = line.lstrip("#").strip() + if line.lower() == fallback.lower(): + continue + lines.append(line) + if sum(len(part) for part in lines) >= MAX_TOOL_DESCRIPTION_CHARS: + break + description = " ".join(lines).strip() or fallback + return _truncate(description) + + +def _schema_from_prompt(prompt: str) -> dict[str, Any]: + schema = _schema_from_embedded_json(prompt) + if schema: + return schema + return _schema_from_args_line(prompt) + + +def _schema_from_embedded_json(prompt: str) -> dict[str, Any]: + marker = "Input schema for tool_args:" + index = (prompt or "").find(marker) + if index == -1: + return {} + tail = prompt[index + len(marker) :].strip() + match = re.search(r"\{(?:[^{}]|(?R))*\}", tail, flags=re.DOTALL) if hasattr(re, "VERSION1") else None + candidate = match.group(0) if match else _balanced_json_object(tail) + if not candidate: + return {} + try: + return _schema_from_any(json.loads(candidate)) + except Exception: + return {} + + +def _schema_from_args_line(prompt: str) -> dict[str, Any]: + properties: dict[str, Any] = {} + for line in (prompt or "").splitlines(): + normalized = line.strip() + if "args:" not in normalized.lower() and "argument:" not in normalized.lower(): + continue + for name in re.findall(r"`([A-Za-z_][A-Za-z0-9_-]*)`", normalized): + properties.setdefault(name, {"type": "string"}) + if properties: + return { + "type": "object", + "properties": properties, + "additionalProperties": True, + } + return _permissive_schema() + + +def _schema_from_any(schema: Any) -> dict[str, Any]: + if isinstance(schema, dict): + normalized = dict(schema) + normalized.setdefault("type", "object") + normalized.setdefault("additionalProperties", True) + return normalized + return _permissive_schema() + + +def _permissive_schema() -> dict[str, Any]: + return {"type": "object", "additionalProperties": True} + + +def _balanced_json_object(text: str) -> str: + start = text.find("{") + if start == -1: + return "" + depth = 0 + in_string = False + escape = False + for index, char in enumerate(text[start:], start=start): + if in_string: + if escape: + escape = False + elif char == "\\": + escape = True + elif char == '"': + in_string = False + continue + if char == '"': + in_string = True + elif char == "{": + depth += 1 + elif char == "}": + depth -= 1 + if depth == 0: + return text[start : index + 1] + return "" + + +def _dedupe_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]: + seen: set[str] = set() + result: list[dict[str, Any]] = [] + for tool in tools: + name = str(tool.get("name") or "") + if not name or name in seen: + continue + seen.add(name) + result.append(tool) + return result + + +def _truncate(text: str) -> str: + if len(text) <= MAX_TOOL_DESCRIPTION_CHARS: + return text + return text[: MAX_TOOL_DESCRIPTION_CHARS - 3].rstrip() + "..." diff --git a/models.py b/models.py index ff2632a94..b81333a0a 100644 --- a/models.py +++ b/models.py @@ -14,19 +14,19 @@ from typing import ( TypedDict, ) -from litellm import completion, acompletion, embedding +from litellm import embedding import litellm import openai -from litellm.types.utils import ModelResponse from helpers import dotenv -from helpers import settings, dirty_json, images +from helpers import settings, images from helpers.dotenv import load_dotenv from helpers.providers import ModelType as ProviderModelType, get_provider_config from helpers.rate_limiter import RateLimiter from helpers.tokens import approximate_tokens -from helpers import dirty_json from helpers.extension import extensible # extensible: allows plugins to intercept get_api_key() +from helpers.litellm_transport import LiteLLMTransport, ResponsesTransport +from helpers.llm_result import LLMResult from langchain_core.language_models.chat_models import SimpleChatModel from langchain_core.outputs.chat_generation import ChatGenerationChunk @@ -155,6 +155,7 @@ class ChatChunk(TypedDict): response_delta: str reasoning_delta: str + class ChatGenerationResult: """Chat generation result object""" def __init__(self, chunk: ChatChunk|None = None): @@ -434,14 +435,6 @@ class LiteLLMChatWrapper(SimpleChatModel): result.append(message_dict) - if explicit_caching and result: - if result[0]["role"] == "system": - result[0]["cache_control"] = {"type": "ephemeral"} - for i in range(len(result) - 1, -1, -1): - if result[i]["role"] == "assistant": - result[i]["cache_control"] = {"type": "ephemeral"} - break - return result def _call( @@ -451,24 +444,20 @@ class LiteLLMChatWrapper(SimpleChatModel): run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> str: - import asyncio - configure_litellm() msgs = self._convert_messages(messages) # Apply rate limiting if configured apply_rate_limiter_sync(self.a0_model_conf, str(msgs)) - # Call the model - call_kwargs = _without_stream_kwarg( - _merge_litellm_call_kwargs(self.kwargs, kwargs) + call_kwargs = _merge_litellm_call_kwargs(self.kwargs, kwargs) + transport = LiteLLMTransport( + model=self.model_name, + messages=msgs, + kwargs=call_kwargs, + stop=stop, ) - resp = completion( - model=self.model_name, messages=msgs, stop=stop, **call_kwargs - ) - - # Parse output - parsed = _parse_chunk(resp) + parsed = transport.complete() output = ChatGenerationResult(parsed).output() return output["response_delta"] @@ -479,8 +468,6 @@ class LiteLLMChatWrapper(SimpleChatModel): run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> Iterator[ChatGenerationChunk]: - import asyncio - configure_litellm() msgs = self._convert_messages(messages) @@ -488,22 +475,15 @@ class LiteLLMChatWrapper(SimpleChatModel): apply_rate_limiter_sync(self.a0_model_conf, str(msgs)) result = ChatGenerationResult() - call_kwargs = _without_stream_kwarg( - _merge_litellm_call_kwargs(self.kwargs, kwargs) - ) - - for chunk in completion( + call_kwargs = _merge_litellm_call_kwargs(self.kwargs, kwargs) + transport = LiteLLMTransport( model=self.model_name, messages=msgs, - stream=True, + kwargs=call_kwargs, stop=stop, - **call_kwargs, - ): - # parse chunk - parsed = _parse_chunk(chunk) # chunk parsing - output = result.add_chunk(parsed) # chunk processing - - # Only yield chunks with non-None content + ) + for parsed in transport.stream(): + output = result.add_chunk(parsed) if output["response_delta"]: yield ChatGenerationChunk( message=AIMessageChunk(content=output["response_delta"]) @@ -523,23 +503,15 @@ class LiteLLMChatWrapper(SimpleChatModel): await apply_rate_limiter(self.a0_model_conf, str(msgs)) result = ChatGenerationResult() - call_kwargs = _without_stream_kwarg( - _merge_litellm_call_kwargs(self.kwargs, kwargs) - ) - - response = await acompletion( + call_kwargs = _merge_litellm_call_kwargs(self.kwargs, kwargs) + transport = LiteLLMTransport( model=self.model_name, messages=msgs, - stream=True, + kwargs=call_kwargs, stop=stop, - **call_kwargs, ) - async for chunk in response: # type: ignore - # parse chunk - parsed = _parse_chunk(chunk) # chunk parsing - output = result.add_chunk(parsed) # chunk processing - - # Only yield chunks with non-None content + async for parsed in transport.astream(): + output = result.add_chunk(parsed) if output["response_delta"]: yield ChatGenerationChunk( message=AIMessageChunk(content=output["response_delta"]) @@ -579,12 +551,19 @@ class LiteLLMChatWrapper(SimpleChatModel): ) # Prepare call kwargs and retry config (strip A0-only params before calling LiteLLM) - call_kwargs: dict[str, Any] = _without_stream_kwarg( - _merge_litellm_call_kwargs(self.kwargs, kwargs) + call_kwargs: dict[str, Any] = _merge_litellm_call_kwargs( + self.kwargs, kwargs ) + if explicit_caching: + call_kwargs["a0_explicit_prompt_caching"] = True max_retries: int = int(call_kwargs.pop("a0_retry_attempts", 2)) retry_delay_s: float = float(call_kwargs.pop("a0_retry_delay_seconds", 1.5)) stream = reasoning_callback is not None or response_callback is not None or tokens_callback is not None + transport = LiteLLMTransport( + model=self.model_name, + messages=msgs_conv, + kwargs=call_kwargs, + ) # results result = ChatGenerationResult() @@ -593,60 +572,45 @@ class LiteLLMChatWrapper(SimpleChatModel): while True: got_any_chunk = False try: - # call model - _completion = await acompletion( - model=self.model_name, - messages=msgs_conv, - stream=stream, - **call_kwargs, - ) - if stream: - # iterate over chunks stop_response: str | None = None - try: - async for chunk in _completion: # type: ignore - got_any_chunk = True - # parse chunk - parsed = _parse_chunk(chunk) - output = result.add_chunk(parsed) + async for parsed in transport.astream(): + got_any_chunk = True + output = result.add_chunk(parsed) - # collect reasoning delta and call callbacks - if output["reasoning_delta"]: - if reasoning_callback: - await reasoning_callback(output["reasoning_delta"], result.reasoning) - if tokens_callback: - await tokens_callback( - output["reasoning_delta"], - approximate_tokens(output["reasoning_delta"]), - ) - # Add output tokens to rate limiter if configured - if limiter: - limiter.add(output=approximate_tokens(output["reasoning_delta"])) - # collect response delta and call callbacks - if output["response_delta"]: - if response_callback: - stop_response = await response_callback( - output["response_delta"], result.response - ) - if tokens_callback: - await tokens_callback( - output["response_delta"], - approximate_tokens(output["response_delta"]), - ) - # Add output tokens to rate limiter if configured - if limiter: - limiter.add(output=approximate_tokens(output["response_delta"])) - if stop_response is not None: - result.response = stop_response - break - finally: - if stop_response is not None and hasattr(_completion, "aclose"): - await _completion.aclose() # type: ignore[attr-defined] + # collect reasoning delta and call callbacks + if output["reasoning_delta"]: + if reasoning_callback: + await reasoning_callback(output["reasoning_delta"], result.reasoning) + if tokens_callback: + await tokens_callback( + output["reasoning_delta"], + approximate_tokens(output["reasoning_delta"]), + ) + # Add output tokens to rate limiter if configured + if limiter: + limiter.add(output=approximate_tokens(output["reasoning_delta"])) + # collect response delta and call callbacks + if output["response_delta"]: + if response_callback: + stop_response = await response_callback( + output["response_delta"], result.response + ) + if tokens_callback: + await tokens_callback( + output["response_delta"], + approximate_tokens(output["response_delta"]), + ) + # Add output tokens to rate limiter if configured + if limiter: + limiter.add(output=approximate_tokens(output["response_delta"])) + if stop_response is not None: + result.response = stop_response + break # non-stream response else: - parsed = _parse_chunk(_completion) + parsed = await transport.acomplete() output = result.add_chunk(parsed) if limiter: if output["response_delta"]: @@ -666,6 +630,151 @@ class LiteLLMChatWrapper(SimpleChatModel): attempt += 1 await asyncio.sleep(retry_delay_s) + async def unified_turn( + self, + system_message="", + user_message="", + messages: List[BaseMessage] | None = None, + response_callback: Callable[[str, str], Awaitable[str | None]] | None = None, + reasoning_callback: Callable[[str, str], Awaitable[None]] | None = None, + tokens_callback: Callable[[str, int], Awaitable[None]] | None = None, + rate_limiter_callback: ( + Callable[[str, str, int, int], Awaitable[bool]] | None + ) = None, + explicit_caching: bool = False, + **kwargs: Any, + ) -> LLMResult: + """Canonical internal LLM turn with Responses metadata. + + Public plugin-facing callers should keep using ``unified_call``. Core + orchestration uses this method when it needs response ids, native output + items, and state/capability metadata. + """ + + configure_litellm() + + if not messages: + messages = [] + if system_message: + messages.insert(0, SystemMessage(content=system_message)) + if user_message: + messages.append(HumanMessage(content=user_message)) + + msgs_conv = self._convert_messages(messages, explicit_caching=explicit_caching) + + limiter = await apply_rate_limiter( + self.a0_model_conf, str(msgs_conv), rate_limiter_callback + ) + + call_kwargs: dict[str, Any] = _merge_litellm_call_kwargs( + self.kwargs, kwargs + ) + if explicit_caching: + call_kwargs["a0_explicit_prompt_caching"] = True + max_retries: int = int(call_kwargs.pop("a0_retry_attempts", 2)) + retry_delay_s: float = float(call_kwargs.pop("a0_retry_delay_seconds", 1.5)) + stream = ( + reasoning_callback is not None + or response_callback is not None + or tokens_callback is not None + ) + transport = LiteLLMTransport( + model=self.model_name, + messages=msgs_conv, + kwargs=call_kwargs, + ) + + result = ChatGenerationResult() + + attempt = 0 + while True: + got_any_chunk = False + try: + if stream: + stop_response: str | None = None + async for parsed in transport.astream(): + got_any_chunk = True + output = result.add_chunk(parsed) + + if output["reasoning_delta"]: + if reasoning_callback: + await reasoning_callback( + output["reasoning_delta"], result.reasoning + ) + if tokens_callback: + await tokens_callback( + output["reasoning_delta"], + approximate_tokens(output["reasoning_delta"]), + ) + if limiter: + limiter.add( + output=approximate_tokens( + output["reasoning_delta"] + ) + ) + + if output["response_delta"]: + if response_callback: + stop_response = await response_callback( + output["response_delta"], result.response + ) + if tokens_callback: + await tokens_callback( + output["response_delta"], + approximate_tokens(output["response_delta"]), + ) + if limiter: + limiter.add( + output=approximate_tokens( + output["response_delta"] + ) + ) + if ( + stop_response is not None + and not transport.policy.using_responses + ): + result.response = stop_response + break + if stop_response is not None: + result.response = stop_response + else: + parsed = await transport.acomplete() + output = result.add_chunk(parsed) + if limiter: + if output["response_delta"]: + limiter.add( + output=approximate_tokens(output["response_delta"]) + ) + if output["reasoning_delta"]: + limiter.add( + output=approximate_tokens(output["reasoning_delta"]) + ) + + llm_result = transport.last_result or LLMResult.from_chat( + response=result.output()["response_delta"], + reasoning=result.output()["reasoning_delta"], + input_items=ResponsesTransport.input_from_messages(msgs_conv), + provider_model_key=self.model_name, + capability=transport._capability_metadata(), + ) + if result.output()["response_delta"]: + llm_result.response = result.output()["response_delta"] + if result.output()["reasoning_delta"]: + llm_result.reasoning = result.output()["reasoning_delta"] + return llm_result + + except Exception as e: + import asyncio + + if ( + got_any_chunk + or not _is_transient_litellm_error(e) + or attempt >= max_retries + ): + raise + attempt += 1 + await asyncio.sleep(retry_delay_s) + class LiteLLMEmbeddingWrapper(Embeddings): model_name: str @@ -820,38 +929,6 @@ def _get_litellm_embedding( ) -def _parse_chunk(chunk: Any) -> ChatChunk: - delta = chunk["choices"][0].get("delta", {}) - message = chunk["choices"][0].get("message", {}) or chunk["choices"][0].get( - "model_extra", {} - ).get("message", {}) - response_delta = ( - delta.get("content", "") - if isinstance(delta, dict) - else getattr(delta, "content", "") - ) or ( - message.get("content", "") - if isinstance(message, dict) - else getattr(message, "content", "") - ) or "" - reasoning_delta = ( - delta.get("reasoning_content", "") - if isinstance(delta, dict) - else getattr(delta, "reasoning_content", "") - ) or ( - message.get("reasoning_content", "") - if isinstance(message, dict) - else getattr(message, "reasoning_content", "") - ) or "" - - return ChatChunk(reasoning_delta=reasoning_delta, response_delta=response_delta) - - -def _without_stream_kwarg(kwargs: dict[str, Any]) -> dict[str, Any]: - kwargs.pop("stream", None) - return kwargs - - def _adjust_call_args(provider_name: str, model_name: str, kwargs: dict): diff --git a/tests/test_responses_architecture.py b/tests/test_responses_architecture.py new file mode 100644 index 000000000..03c0ae54e --- /dev/null +++ b/tests/test_responses_architecture.py @@ -0,0 +1,385 @@ +import sys +from pathlib import Path + +import pytest +from langchain_core.messages import HumanMessage + + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +import models +from agent import Agent, AgentConfig, LoopData +from helpers import history, litellm_transport +from helpers.log import Log +from helpers.llm_result import LLMResult, result_from_metadata +from helpers.persist_chat import _collect_response_ids +from helpers.tool import Response + + +@pytest.fixture(autouse=True) +def _clear_transport_capability_cache(): + litellm_transport.clear_transport_capability_cache() + + +class _AsyncEventStream: + def __init__(self, events: list[dict]): + self.events = events + self.index = 0 + self.closed = False + + def __aiter__(self): + return self + + async def __anext__(self): + if self.index >= len(self.events): + raise StopAsyncIteration + event = self.events[self.index] + self.index += 1 + return event + + async def aclose(self): + self.closed = True + + +def test_llm_result_round_trips_responses_metadata(): + result = LLMResult.from_response( + { + "id": "resp_123", + "usage": {"input_tokens": 10}, + "output": [ + {"type": "reasoning", "summary": [{"text": "because"}]}, + { + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": "lookup", + "arguments": '{"q":"a0"}', + }, + { + "type": "web_search_call", + "id": "ws_1", + "status": "completed", + }, + ], + }, + input_items=[{"role": "user", "content": "question"}], + previous_response_id="resp_prev", + provider_model_key="openai/gpt-5.4", + ) + + loaded = result_from_metadata(result.metadata()) + + assert loaded is not None + assert loaded.response_id == "resp_123" + assert loaded.previous_response_id == "resp_prev" + assert loaded.function_calls[0].name == "lookup" + assert loaded.function_calls[0].arguments == {"q": "a0"} + assert loaded.builtin_items[0].type == "web_search_call" + + +def test_history_serializes_metadata_and_migrates_old_messages(): + class DummyAgent: + pass + + hist = history.History(DummyAgent()) + result = LLMResult.from_response( + {"id": "resp_1", "output": [{"type": "message", "content": [{"type": "output_text", "text": "ok"}]}]}, + provider_model_key="openai/gpt-5.4", + ) + + message = hist.add_message(True, "ok", metadata=result.metadata()) + restored = history.deserialize_history(hist.serialize(), DummyAgent()) + + restored_message = restored.all_messages()[0] + assert restored_message.sequence == message.sequence + assert result_from_metadata(restored_message.metadata).response_id == "resp_1" + + old = history.Message.from_dict({"_cls": "Message", "ai": False, "content": "old"}, restored) + assert old.metadata == {} + assert old.sequence == 0 + + +def test_responses_provider_state_uses_previous_response_and_new_items(): + new_items = [{"type": "function_call_output", "call_id": "call_1", "output": "done"}] + local_items = [{"role": "user", "content": "full replay"}] + + request = litellm_transport.ResponsesTransport.from_chat( + [{"role": "user", "content": "ignored while continuing provider state"}], + { + "previous_response_id": "resp_1", + "responses_input_items": new_items, + "responses_local_input_items": local_items, + }, + model="openai/gpt-5.4", + ) + + assert request["store"] is True + assert request["previous_response_id"] == "resp_1" + assert request["input"] == new_items + + local_request = litellm_transport.ResponsesTransport.from_chat( + [{"role": "user", "content": "ignored"}], + { + "responses_state": "local", + "previous_response_id": "resp_1", + "responses_input_items": new_items, + "responses_local_input_items": local_items, + }, + model="openai/gpt-5.4", + ) + + assert local_request["store"] is False + assert "previous_response_id" not in local_request + assert local_request["input"] == local_items + + +@pytest.mark.asyncio +async def test_transport_retries_provider_state_as_local_replay(monkeypatch): + calls: list[dict] = [] + + async def fake_aresponses(*args, **kwargs): + calls.append(kwargs) + if len(calls) == 1: + raise RuntimeError("previous_response_id is not supported by this provider") + return { + "id": "resp_local", + "output": [ + { + "type": "message", + "content": [{"type": "output_text", "text": "ok"}], + } + ], + } + + monkeypatch.setattr(litellm_transport, "aresponses", fake_aresponses) + + transport = litellm_transport.LiteLLMTransport( + model="openai/gpt-5.4", + messages=[{"role": "user", "content": "new"}], + kwargs={ + "previous_response_id": "resp_1", + "responses_input_items": [{"role": "user", "content": "new"}], + "responses_local_input_items": [{"role": "user", "content": "full"}], + }, + ) + + parsed = await transport.acomplete() + + assert parsed["response_delta"] == "ok" + assert calls[0]["store"] is True + assert calls[0]["previous_response_id"] == "resp_1" + assert calls[1]["store"] is False + assert "previous_response_id" not in calls[1] + assert calls[1]["input"] == [{"role": "user", "content": "full"}] + assert transport.last_result.response_id == "resp_local" + + +@pytest.mark.asyncio +async def test_transport_downgrades_unsupported_builtin_tools(monkeypatch): + calls: list[dict] = [] + + async def fake_aresponses(*args, **kwargs): + calls.append(kwargs) + if len(calls) == 1: + raise RuntimeError("unsupported tool type: web_search") + return { + "id": "resp_no_builtin", + "output": [ + { + "type": "message", + "content": [{"type": "output_text", "text": "ok"}], + } + ], + } + + monkeypatch.setattr(litellm_transport, "aresponses", fake_aresponses) + + transport = litellm_transport.LiteLLMTransport( + model="openai/gpt-5.4", + messages=[{"role": "user", "content": "new"}], + kwargs={"responses_builtin_tools": [{"type": "web_search"}]}, + ) + + parsed = await transport.acomplete() + + assert parsed["response_delta"] == "ok" + assert calls[0]["tools"] == [{"type": "web_search"}] + assert "tools" not in calls[1] + assert transport.last_result.capability["builtin_tool_downgrades"] == [ + "web_search" + ] + + next_transport = litellm_transport.LiteLLMTransport( + model="openai/gpt-5.4", + messages=[{"role": "user", "content": "again"}], + kwargs={"responses_builtin_tools": [{"type": "web_search"}]}, + ) + request = next_transport._responses_request(stream=False) + assert "tools" not in request + + +@pytest.mark.asyncio +async def test_unified_turn_keeps_stream_open_to_capture_response_id(monkeypatch): + stream = _AsyncEventStream( + [ + { + "type": "response.output_item.added", + "output_index": 0, + "item": { + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": "lookup", + "arguments": "", + }, + }, + { + "type": "response.function_call_arguments.done", + "item_id": "fc_1", + "output_index": 0, + "name": "lookup", + "arguments": '{"q":"a0"}', + }, + { + "type": "response.completed", + "response": { + "id": "resp_1", + "output": [ + { + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": "lookup", + "arguments": '{"q":"a0"}', + } + ], + }, + }, + ] + ) + + async def fake_aresponses(*args, **kwargs): + return stream + + async def fake_rate_limiter(*args, **kwargs): + return None + + monkeypatch.setattr(litellm_transport, "aresponses", fake_aresponses) + monkeypatch.setattr(models, "apply_rate_limiter", fake_rate_limiter) + + wrapper = models.LiteLLMChatWrapper( + model="test-model", + provider="openai", + model_config=None, + ) + + async def response_callback(chunk: str, full: str): + return full + + result = await wrapper.unified_turn( + messages=[HumanMessage(content="hi")], + response_callback=response_callback, + ) + + assert stream.index == 3 + assert stream.closed is False + assert result.response_id == "resp_1" + assert result.function_calls[0].call_id == "call_1" + + +def test_collect_response_ids_from_agent_state_and_history_metadata(): + payload = { + "agents": [ + { + "data": { + "responses_state": { + "response_id": "resp_latest", + "response_ids": ["resp_old", "resp_latest"], + } + }, + "history": '{"current":{"messages":[{"metadata":{"responses":{"response_id":"resp_history"}}}]}}', + } + ] + } + + assert _collect_response_ids(payload) == [ + "resp_latest", + "resp_old", + "resp_history", + ] + + +@pytest.mark.asyncio +async def test_agent_executes_native_responses_function_call_and_records_output(): + class DummyContext: + paused = False + log = Log() + + def get_data(self, key, recursive=True): + return None + + class DummyTool: + name = "lookup" + progress = "" + + def __init__(self, agent): + self.agent = agent + + async def before_execution(self, **kwargs): + self.args = kwargs + + async def execute(self, **kwargs): + return Response(message=f"done:{kwargs['q']}", break_loop=False) + + async def after_execution(self, response): + self.agent.hist_add_tool_result( + self.name, + response.message, + **(response.additional or {}), + ) + + agent = object.__new__(Agent) + agent.data = {Agent.DATA_NAME_RESPONSES_TOOL_NAME_MAP: {}} + agent.context = DummyContext() + agent.config = AgentConfig(mcp_servers="") + agent.loop_data = LoopData() + agent.history = history.History(agent) + agent.intervention = None + agent.agent_name = "A0" + agent.number = 0 + + def get_tool(**kwargs): + return DummyTool(agent) + + agent.get_tool = get_tool + + result = LLMResult.from_response( + { + "id": "resp_1", + "output": [ + { + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": "lookup", + "arguments": '{"q":"a0"}', + } + ], + }, + provider_model_key="openai/gpt-5.4", + ) + + assert await Agent.process_llm_result_tools(agent, result) is None + + recorded = agent.history.all_messages()[0] + metadata = result_from_metadata(recorded.metadata) + assert recorded.content["tool_result"] == "done:a0" + assert metadata.input_items == [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": "done:a0", + } + ] diff --git a/tests/test_stream_tool_early_stop.py b/tests/test_stream_tool_early_stop.py index e01d5c1fb..696ace7c1 100644 --- a/tests/test_stream_tool_early_stop.py +++ b/tests/test_stream_tool_early_stop.py @@ -2,6 +2,7 @@ import sys from pathlib import Path import pytest +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage PROJECT_ROOT = Path(__file__).resolve().parents[1] @@ -10,16 +11,27 @@ if str(PROJECT_ROOT) not in sys.path: import models from helpers import extract_tools +from helpers import litellm_transport + + +@pytest.fixture(autouse=True) +def _clear_transport_capability_cache(): + litellm_transport.clear_transport_capability_cache() def _chunk(content: str) -> dict: return {"choices": [{"delta": {"content": content}, "message": {}}]} +def _response_event(delta: str) -> dict: + return {"type": "response.output_text.delta", "delta": delta} + + class _AsyncChunkStream: def __init__(self, chunks: list[dict]): self._chunks = chunks self.index = 0 + self.closed = False def __aiter__(self): return self @@ -31,6 +43,24 @@ class _AsyncChunkStream: self.index += 1 return chunk + async def aclose(self): + self.closed = True + + +class _FailingAsyncChunkStream: + def __init__(self, exc: Exception): + self.exc = exc + self.closed = False + + def __aiter__(self): + return self + + async def __anext__(self): + raise self.exc + + async def aclose(self): + self.closed = True + def test_extract_json_root_string_returns_canonical_snapshot(): text = ( @@ -94,22 +124,24 @@ def test_litellm_global_kwargs_merge_defaults_and_config(monkeypatch): async def test_unified_call_stops_after_canonical_root_snapshot(monkeypatch): stream = _AsyncChunkStream( [ - _chunk( + {"type": "response.created"}, + _response_event( '{"tool_name":"response","tool_args":{"text":"hello"}} trailing text' ), - _chunk(" unreachable"), + _response_event(" unreachable"), ] ) - async def fake_acompletion(*args, **kwargs): + async def fake_aresponses(*args, **kwargs): assert kwargs["stream"] is True - assert kwargs["drop_params"] is True + assert kwargs["input"] == "" + assert kwargs["store"] is True return stream async def fake_rate_limiter(*args, **kwargs): return None - monkeypatch.setattr(models, "acompletion", fake_acompletion) + monkeypatch.setattr(litellm_transport, "aresponses", fake_aresponses) monkeypatch.setattr(models, "apply_rate_limiter", fake_rate_limiter) monkeypatch.setattr( models.settings, @@ -139,6 +171,742 @@ async def test_unified_call_stops_after_canonical_root_snapshot(monkeypatch): assert response == '{"tool_name":"response","tool_args":{"text":"hello"}}' assert reasoning == "" - assert stream.index == 1 + assert stream.index == 2 + assert stream.closed is True assert len(seen) == 1 assert seen[0][1] == '{"tool_name":"response","tool_args":{"text":"hello"}} trailing text' + + +@pytest.mark.asyncio +async def test_unified_call_closes_responses_stream_when_callback_raises(monkeypatch): + stream = _AsyncChunkStream([_response_event("interrupt me")]) + + class ExpectedIntervention(Exception): + pass + + async def fake_aresponses(*args, **kwargs): + assert kwargs["stream"] is True + return stream + + async def fake_rate_limiter(*args, **kwargs): + return None + + monkeypatch.setattr(litellm_transport, "aresponses", fake_aresponses) + monkeypatch.setattr(models, "apply_rate_limiter", fake_rate_limiter) + + wrapper = models.LiteLLMChatWrapper( + model="test-model", + provider="openai", + model_config=None, + ) + + async def response_callback(chunk: str, full: str): + raise ExpectedIntervention() + + with pytest.raises(ExpectedIntervention): + await wrapper.unified_call( + messages=[], + response_callback=response_callback, + ) + + assert stream.closed is True + + +@pytest.mark.asyncio +async def test_chat_completions_escape_hatch_still_uses_acompletion(monkeypatch): + stream = _AsyncChunkStream([_chunk("hello")]) + calls: list[str] = [] + + async def fake_acompletion(*args, **kwargs): + calls.append("chat") + assert kwargs["stream"] is True + assert "a0_api_mode" not in kwargs + return stream + + async def fake_aresponses(*args, **kwargs): + raise AssertionError("Responses path should not be used") + + async def fake_rate_limiter(*args, **kwargs): + return None + + monkeypatch.setattr(litellm_transport, "acompletion", fake_acompletion) + monkeypatch.setattr(litellm_transport, "aresponses", fake_aresponses) + monkeypatch.setattr(models, "apply_rate_limiter", fake_rate_limiter) + + wrapper = models.LiteLLMChatWrapper( + model="test-model", + provider="openai", + model_config=None, + a0_api_mode="chat_completions", + ) + + async def response_callback(chunk: str, full: str): + return None + + response, reasoning = await wrapper.unified_call( + messages=[], + response_callback=response_callback, + ) + + assert response == "hello" + assert reasoning == "" + assert calls == ["chat"] + + +@pytest.mark.asyncio +async def test_unified_call_retries_responses_with_high_reasoning(monkeypatch): + validation_error = ValueError( + "1 validation error for ResponseCreatedEvent\n" + "response.reasoning.effort\n" + "Input should be 'minimal', 'low', 'medium' or 'high' " + "[type=literal_error, input_value='none', input_type=str]" + ) + failing_stream = _FailingAsyncChunkStream(validation_error) + working_stream = _AsyncChunkStream([_response_event("ok")]) + calls: list[dict] = [] + + async def fake_aresponses(*args, **kwargs): + calls.append(kwargs) + if len(calls) == 1: + return failing_stream + return working_stream + + async def fake_rate_limiter(*args, **kwargs): + return None + + monkeypatch.setattr(litellm_transport, "aresponses", fake_aresponses) + monkeypatch.setattr(models, "apply_rate_limiter", fake_rate_limiter) + + wrapper = models.LiteLLMChatWrapper( + model="gpt-5.4", + provider="openai", + model_config=None, + ) + + async def response_callback(chunk: str, full: str): + return None + + response, reasoning = await wrapper.unified_call( + messages=[], + response_callback=response_callback, + ) + + assert response == "ok" + assert reasoning == "" + assert failing_stream.closed is True + assert len(calls) == 2 + assert "reasoning" not in calls[0] + assert calls[1]["reasoning"] == {"effort": "high"} + + +@pytest.mark.asyncio +async def test_unified_call_falls_back_to_chat_when_responses_endpoint_missing( + monkeypatch, +): + calls: list[str] = [] + + async def fake_aresponses(*args, **kwargs): + calls.append("responses") + raise RuntimeError( + "Client error '404 Not Found' for url " + "'https://llm.agent-zero.ai/v1/responses'" + ) + + async def fake_acompletion(*args, **kwargs): + calls.append("chat") + assert kwargs["stream"] is True + assert kwargs["drop_params"] is True + assert "tool_choice" not in kwargs + assert "parallel_tool_calls" not in kwargs + return _AsyncChunkStream([_chunk("fallback")]) + + async def fake_rate_limiter(*args, **kwargs): + return None + + monkeypatch.setattr(litellm_transport, "aresponses", fake_aresponses) + monkeypatch.setattr(litellm_transport, "acompletion", fake_acompletion) + monkeypatch.setattr(models, "apply_rate_limiter", fake_rate_limiter) + + wrapper = models.LiteLLMChatWrapper( + model="claude-opus-4.7", + provider="openai", + model_config=None, + tool_choice="auto", + parallel_tool_calls=True, + ) + + async def response_callback(chunk: str, full: str): + return None + + response, reasoning = await wrapper.unified_call( + messages=[], + response_callback=response_callback, + ) + + assert response == "fallback" + assert reasoning == "" + assert calls == ["responses", "chat"] + + response, reasoning = await wrapper.unified_call( + messages=[], + response_callback=response_callback, + ) + + assert response == "fallback" + assert reasoning == "" + assert calls == ["responses", "chat", "chat"] + + +@pytest.mark.asyncio +async def test_unified_call_preserves_cache_control_with_chat_for_non_native_responses( + monkeypatch, +): + calls: list[str] = [] + + async def fake_aresponses(*args, **kwargs): + raise AssertionError("cache_control should keep Anthropic-family calls on chat") + + async def fake_acompletion(*args, **kwargs): + calls.append("chat") + assert kwargs["stream"] is True + messages = kwargs["messages"] + assert "cache_control" not in messages[0] + assert messages[0]["content"][-1]["cache_control"] == { + "type": "ephemeral" + } + assert messages[1]["content"][-1]["cache_control"] == { + "type": "ephemeral" + } + assert "cache_control" not in messages[2] + assert messages[3]["content"][-1]["cache_control"] == { + "type": "ephemeral" + } + return _AsyncChunkStream([_chunk("cached")]) + + async def fake_rate_limiter(*args, **kwargs): + return None + + monkeypatch.setattr(litellm_transport, "aresponses", fake_aresponses) + monkeypatch.setattr(litellm_transport, "acompletion", fake_acompletion) + monkeypatch.setattr(models, "apply_rate_limiter", fake_rate_limiter) + + wrapper = models.LiteLLMChatWrapper( + model="claude-sonnet-4-5", + provider="anthropic", + model_config=None, + ) + + async def response_callback(chunk: str, full: str): + return None + + response, reasoning = await wrapper.unified_call( + messages=[ + SystemMessage(content="static instructions"), + HumanMessage(content="question"), + AIMessage(content="previous answer"), + HumanMessage(content="follow up"), + ], + response_callback=response_callback, + explicit_caching=True, + ) + + assert response == "cached" + assert reasoning == "" + assert calls == ["chat"] + + +def test_responses_request_translates_messages_and_params(): + messages = [ + {"role": "system", "content": "You are precise."}, + { + "role": "user", + "content": [ + {"type": "text", "text": "Inspect this."}, + { + "type": "image_url", + "image_url": {"url": "https://example.test/a.png"}, + }, + ], + }, + { + "role": "assistant", + "content": "empty", + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "lookup", "arguments": '{"q":"a0"}'}, + } + ], + }, + {"role": "tool", "tool_call_id": "call_1", "content": "done"}, + ] + kwargs = { + "max_tokens": 42, + "reasoning_effort": "high", + "response_format": { + "type": "json_schema", + "json_schema": { + "name": "answer", + "schema": {"type": "object"}, + "strict": True, + }, + }, + "tools": [ + { + "type": "function", + "function": { + "name": "lookup", + "description": "Search", + "parameters": {"type": "object"}, + "strict": True, + }, + } + ], + } + + request = litellm_transport.ResponsesTransport.from_chat(messages, kwargs) + + assert "instructions" not in request + assert request["store"] is True + assert request["max_output_tokens"] == 42 + assert request["reasoning"] == {"effort": "high"} + assert request["text"] == { + "format": { + "type": "json_schema", + "name": "answer", + "schema": {"type": "object"}, + "strict": True, + } + } + assert request["tools"] == [ + { + "type": "function", + "name": "lookup", + "description": "Search", + "parameters": {"type": "object"}, + "strict": True, + } + ] + assert request["input"] == [ + {"role": "system", "content": "You are precise."}, + { + "role": "user", + "content": [ + {"type": "input_text", "text": "Inspect this."}, + { + "type": "input_image", + "image_url": "https://example.test/a.png", + }, + ], + }, + { + "type": "function_call", + "call_id": "call_1", + "id": "call_1", + "name": "lookup", + "arguments": '{"q":"a0"}', + "status": "completed", + }, + {"type": "function_call_output", "call_id": "call_1", "output": "done"}, + ] + + +def test_responses_request_normalizes_reasoning_and_orphan_tool_choice(): + request = litellm_transport.ResponsesTransport.from_chat( + [], + { + "reasoning_effort": "none", + "tools": [], + "tool_choice": "auto", + "parallel_tool_calls": True, + }, + ) + + assert "reasoning" not in request + assert "tools" not in request + assert "tool_choice" not in request + assert "parallel_tool_calls" not in request + + request = litellm_transport.ResponsesTransport.from_chat( + [], + {"reasoning": {"effort": "xhigh"}}, + ) + + assert request["reasoning"] == {"effort": "high"} + + request = litellm_transport.ResponsesTransport.from_chat( + [], + {"reasoning_effort": "off"}, + ) + + assert "reasoning" not in request + + +def test_responses_request_adds_openai_prompt_cache_key_for_static_prefix(): + request = litellm_transport.ResponsesTransport.from_chat( + [ + {"role": "system", "content": "stable system prompt"}, + {"role": "user", "content": "dynamic question"}, + ], + { + "tools": [ + { + "type": "function", + "function": { + "name": "lookup", + "description": "Search", + "parameters": {"type": "object"}, + }, + } + ], + }, + model="openai/gpt-5.4", + ) + + assert request["prompt_cache_key"].startswith("a0-") + assert len(request["prompt_cache_key"]) == 35 + assert "stable system prompt" not in request["prompt_cache_key"] + + request_again = litellm_transport.ResponsesTransport.from_chat( + [ + {"role": "system", "content": "stable system prompt"}, + {"role": "user", "content": "different dynamic question"}, + ], + { + "tools": [ + { + "type": "function", + "function": { + "name": "lookup", + "description": "Search", + "parameters": {"type": "object"}, + }, + } + ], + }, + model="openai/gpt-5.4", + ) + + assert request_again["prompt_cache_key"] == request["prompt_cache_key"] + + +def test_responses_request_respects_explicit_prompt_cache_and_retention(): + request = litellm_transport.ResponsesTransport.from_chat( + [{"role": "system", "content": "stable system prompt"}], + { + "prompt_cache_key": "user-provided-key", + "prompt_cache_retention": "24h", + "extra_body": {"prompt_cache_retention": "in_memory"}, + }, + model="openai/gpt-5.4", + ) + + assert request["prompt_cache_key"] == "user-provided-key" + assert "prompt_cache_retention" not in request + assert request["extra_body"]["prompt_cache_retention"] == "in_memory" + + +def test_responses_request_adds_azure_prompt_cache_params(): + request = litellm_transport.ResponsesTransport.from_chat( + [{"role": "system", "content": "stable system prompt"}], + {"prompt_cache_retention": "24h"}, + model="azure/gpt-4.1", + ) + + assert request["prompt_cache_key"].startswith("a0-") + assert "prompt_cache_retention" not in request + assert request["extra_body"]["prompt_cache_retention"] == "24h" + + +def test_responses_request_does_not_add_openai_cache_key_to_custom_api_base(): + request = litellm_transport.ResponsesTransport.from_chat( + [{"role": "system", "content": "stable system prompt"}], + {"api_base": "https://llm.agent-zero.ai/v1"}, + model="openai/gpt-5.4", + ) + + assert "prompt_cache_key" not in request + + +def test_chat_kwargs_add_openai_prompt_cache_key_for_chat_completions(): + kwargs = litellm_transport.ChatCompletionsTransport.prepare_kwargs( + {"max_tokens": 10}, + model="openai/gpt-5.4", + messages=[ + {"role": "system", "content": "stable system prompt"}, + {"role": "user", "content": "dynamic question"}, + ], + ) + + assert kwargs["prompt_cache_key"].startswith("a0-") + assert kwargs["max_tokens"] == 10 + + +def test_chat_messages_strip_cache_control_for_openai_prompt_cache(): + messages = [ + { + "role": "system", + "cache_control": {"type": "ephemeral"}, + "content": [ + { + "type": "text", + "text": "stable system prompt", + "cache_control": {"type": "ephemeral"}, + } + ], + } + ] + + prepared = litellm_transport.ChatCompletionsTransport.prepare_messages( + messages, + model="openai/gpt-5.4", + kwargs={}, + ) + + assert "cache_control" not in prepared[0] + assert "cache_control" not in prepared[0]["content"][0] + assert messages[0]["content"][0]["cache_control"] == {"type": "ephemeral"} + + +def test_chat_kwargs_mark_cached_tools_for_cache_control_providers(): + kwargs = litellm_transport.ChatCompletionsTransport.prepare_kwargs( + { + "tools": [ + { + "type": "function", + "function": { + "name": "lookup", + "description": "Search", + "parameters": {"type": "object"}, + }, + } + ], + }, + model="anthropic/claude-sonnet-4-5", + messages=[ + { + "role": "system", + "content": [ + { + "type": "text", + "text": "static instructions", + "cache_control": {"type": "ephemeral"}, + } + ], + } + ], + explicit_prompt_caching=True, + ) + + assert kwargs["tools"][0]["function"]["cache_control"] == { + "type": "ephemeral" + } + + +def test_chat_kwargs_strip_orphan_tool_choice_and_enable_fallback_drop_params(): + kwargs = litellm_transport.ChatCompletionsTransport.prepare_kwargs( + { + "tool_choice": "auto", + "parallel_tool_calls": True, + "max_tokens": 10, + }, + fallback_error=RuntimeError("This model does not support Responses API"), + ) + + assert kwargs["max_tokens"] == 10 + assert kwargs["drop_params"] is True + assert "tool_choice" not in kwargs + assert "parallel_tool_calls" not in kwargs + + +def test_cache_control_policy_keeps_native_responses_first(): + messages = [ + { + "role": "system", + "content": "static instructions", + "cache_control": {"type": "ephemeral"}, + } + ] + + openai_policy = litellm_transport.TransportPolicy.from_request( + "openai/gpt-5.4", + {}, + messages=messages, + ) + anthropic_policy = litellm_transport.TransportPolicy.from_request( + "anthropic/claude-sonnet-4-5", + {}, + messages=messages, + ) + + assert openai_policy.mode is litellm_transport.TransportMode.RESPONSES + assert anthropic_policy.mode is litellm_transport.TransportMode.CHAT_COMPLETIONS + + +def test_responses_fallback_does_not_mask_rate_limits(): + exc = RuntimeError( + "RateLimitError: 429 Too Many Requests for url " + "https://api.openai.com/v1/responses" + ) + + policy = litellm_transport.TransportPolicy( + mode=litellm_transport.TransportMode.RESPONSES + ) + + assert ( + policy.recover(exc, got_any_chunk=False) + is litellm_transport.TransportRecovery.RAISE + ) + + +def test_responses_response_parser_extracts_text_reasoning_and_function_calls(): + text_response = { + "output": [ + {"type": "reasoning", "summary": [{"text": "because"}]}, + { + "type": "message", + "content": [{"type": "output_text", "text": "answer"}], + }, + ] + } + + parsed = litellm_transport.ResponsesTransport.parse_response(text_response) + + assert parsed == {"response_delta": "answer", "reasoning_delta": "because"} + + tool_response = { + "output": [ + { + "type": "function_call", + "name": "lookup", + "arguments": '{"q":"a0"}', + } + ] + } + + parsed_tool = litellm_transport.ResponsesTransport.parse_response(tool_response) + + assert extract_tools.json_parse_dirty(parsed_tool["response_delta"]) == { + "tool_name": "lookup", + "tool_args": {"q": "a0"}, + } + + +def test_responses_stream_parser_accumulates_function_call_arguments(): + parser = litellm_transport.ResponsesEventParser() + + assert parser.parse( + { + "type": "response.output_item.added", + "output_index": 0, + "item": { + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": "lookup", + "arguments": "", + }, + } + ) == {"reasoning_delta": "", "response_delta": ""} + assert parser.parse( + { + "type": "response.function_call_arguments.delta", + "item_id": "fc_1", + "output_index": 0, + "delta": '{"q":', + } + ) == {"reasoning_delta": "", "response_delta": ""} + + parsed = parser.parse( + { + "type": "response.function_call_arguments.done", + "item_id": "fc_1", + "output_index": 0, + "name": "lookup", + "arguments": '{"q":"a0"}', + } + ) + + assert extract_tools.json_parse_dirty(parsed["response_delta"]) == { + "tool_name": "lookup", + "tool_args": {"q": "a0"}, + } + assert parser.parse( + { + "type": "response.output_item.done", + "output_index": 0, + "item": { + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": "lookup", + "arguments": '{"q":"a0"}', + }, + } + ) == {"reasoning_delta": "", "response_delta": ""} + + +def test_responses_stream_parser_uses_completed_response_when_no_deltas_arrive(): + parser = litellm_transport.ResponsesEventParser() + + parsed = parser.parse( + { + "type": "response.completed", + "response": { + "output": [ + { + "type": "message", + "content": [{"type": "output_text", "text": "done"}], + } + ] + }, + } + ) + + assert parsed == {"reasoning_delta": "", "response_delta": "done"} + + +def test_responses_stream_parser_handles_refusal_and_failed_events(): + parser = litellm_transport.ResponsesEventParser() + + assert parser.parse( + {"type": "response.refusal.delta", "delta": "no"} + ) == {"reasoning_delta": "", "response_delta": "no"} + + with pytest.raises(RuntimeError, match="policy"): + parser.parse( + { + "type": "response.failed", + "response": {"error": {"message": "policy"}}, + } + ) + + +def test_responses_response_parser_groups_parallel_function_calls(): + response = { + "output": [ + { + "type": "function_call", + "name": "lookup", + "arguments": '{"q":"a0"}', + }, + { + "type": "function_call", + "name": "rank", + "arguments": '{"limit":2}', + }, + ] + } + + parsed = litellm_transport.ResponsesTransport.parse_response(response) + + assert extract_tools.json_parse_dirty(parsed["response_delta"]) == { + "tool_name": "parallel_tool_calls", + "tool_args": { + "calls": [ + {"tool_name": "lookup", "tool_args": {"q": "a0"}}, + {"tool_name": "rank", "tool_args": {"limit": 2}}, + ] + }, + }