Spaces:
Running
Running
| from __future__ import annotations | |
| import abc | |
| import copy | |
| from dataclasses import dataclass | |
| from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar, Union | |
| from openai.types.responses import ( | |
| Response, | |
| ResponseComputerToolCall, | |
| ResponseFileSearchToolCall, | |
| ResponseFunctionToolCall, | |
| ResponseFunctionWebSearch, | |
| ResponseInputItemParam, | |
| ResponseOutputItem, | |
| ResponseOutputMessage, | |
| ResponseOutputRefusal, | |
| ResponseOutputText, | |
| ResponseStreamEvent, | |
| ) | |
| from openai.types.responses.response_code_interpreter_tool_call import ( | |
| ResponseCodeInterpreterToolCall, | |
| ) | |
| from openai.types.responses.response_input_item_param import ( | |
| ComputerCallOutput, | |
| FunctionCallOutput, | |
| LocalShellCallOutput, | |
| McpApprovalResponse, | |
| ) | |
| from openai.types.responses.response_output_item import ( | |
| ImageGenerationCall, | |
| LocalShellCall, | |
| McpApprovalRequest, | |
| McpCall, | |
| McpListTools, | |
| ) | |
| from openai.types.responses.response_reasoning_item import ResponseReasoningItem | |
| from pydantic import BaseModel | |
| from typing_extensions import TypeAlias | |
| from .exceptions import AgentsException, ModelBehaviorError | |
| from .usage import Usage | |
| if TYPE_CHECKING: | |
| from .agent import Agent | |
| TResponse = Response | |
| """A type alias for the Response type from the OpenAI SDK.""" | |
| TResponseInputItem = ResponseInputItemParam | |
| """A type alias for the ResponseInputItemParam type from the OpenAI SDK.""" | |
| TResponseOutputItem = ResponseOutputItem | |
| """A type alias for the ResponseOutputItem type from the OpenAI SDK.""" | |
| TResponseStreamEvent = ResponseStreamEvent | |
| """A type alias for the ResponseStreamEvent type from the OpenAI SDK.""" | |
| T = TypeVar("T", bound=Union[TResponseOutputItem, TResponseInputItem]) | |
| class RunItemBase(Generic[T], abc.ABC): | |
| agent: Agent[Any] | |
| """The agent whose run caused this item to be generated.""" | |
| raw_item: T | |
| """The raw Responses item from the run. This will always be a either an output item (i.e. | |
| `openai.types.responses.ResponseOutputItem` or an input item | |
| (i.e. `openai.types.responses.ResponseInputItemParam`). | |
| """ | |
| def to_input_item(self) -> TResponseInputItem: | |
| """Converts this item into an input item suitable for passing to the model.""" | |
| if isinstance(self.raw_item, dict): | |
| # We know that input items are dicts, so we can ignore the type error | |
| return self.raw_item # type: ignore | |
| elif isinstance(self.raw_item, BaseModel): | |
| # All output items are Pydantic models that can be converted to input items. | |
| return self.raw_item.model_dump(exclude_unset=True) # type: ignore | |
| else: | |
| raise AgentsException(f"Unexpected raw item type: {type(self.raw_item)}") | |
| class MessageOutputItem(RunItemBase[ResponseOutputMessage]): | |
| """Represents a message from the LLM.""" | |
| raw_item: ResponseOutputMessage | |
| """The raw response output message.""" | |
| type: Literal["message_output_item"] = "message_output_item" | |
| class HandoffCallItem(RunItemBase[ResponseFunctionToolCall]): | |
| """Represents a tool call for a handoff from one agent to another.""" | |
| raw_item: ResponseFunctionToolCall | |
| """The raw response function tool call that represents the handoff.""" | |
| type: Literal["handoff_call_item"] = "handoff_call_item" | |
| class HandoffOutputItem(RunItemBase[TResponseInputItem]): | |
| """Represents the output of a handoff.""" | |
| raw_item: TResponseInputItem | |
| """The raw input item that represents the handoff taking place.""" | |
| source_agent: Agent[Any] | |
| """The agent that made the handoff.""" | |
| target_agent: Agent[Any] | |
| """The agent that is being handed off to.""" | |
| type: Literal["handoff_output_item"] = "handoff_output_item" | |
| ToolCallItemTypes: TypeAlias = Union[ | |
| ResponseFunctionToolCall, | |
| ResponseComputerToolCall, | |
| ResponseFileSearchToolCall, | |
| ResponseFunctionWebSearch, | |
| McpCall, | |
| ResponseCodeInterpreterToolCall, | |
| ImageGenerationCall, | |
| LocalShellCall, | |
| ] | |
| """A type that represents a tool call item.""" | |
| class ToolCallItem(RunItemBase[ToolCallItemTypes]): | |
| """Represents a tool call e.g. a function call or computer action call.""" | |
| raw_item: ToolCallItemTypes | |
| """The raw tool call item.""" | |
| type: Literal["tool_call_item"] = "tool_call_item" | |
| class ToolCallOutputItem( | |
| RunItemBase[Union[FunctionCallOutput, ComputerCallOutput, LocalShellCallOutput]] | |
| ): | |
| """Represents the output of a tool call.""" | |
| raw_item: FunctionCallOutput | ComputerCallOutput | LocalShellCallOutput | |
| """The raw item from the model.""" | |
| output: Any | |
| """The output of the tool call. This is whatever the tool call returned; the `raw_item` | |
| contains a string representation of the output. | |
| """ | |
| type: Literal["tool_call_output_item"] = "tool_call_output_item" | |
| class ReasoningItem(RunItemBase[ResponseReasoningItem]): | |
| """Represents a reasoning item.""" | |
| raw_item: ResponseReasoningItem | |
| """The raw reasoning item.""" | |
| type: Literal["reasoning_item"] = "reasoning_item" | |
| class MCPListToolsItem(RunItemBase[McpListTools]): | |
| """Represents a call to an MCP server to list tools.""" | |
| raw_item: McpListTools | |
| """The raw MCP list tools call.""" | |
| type: Literal["mcp_list_tools_item"] = "mcp_list_tools_item" | |
| class MCPApprovalRequestItem(RunItemBase[McpApprovalRequest]): | |
| """Represents a request for MCP approval.""" | |
| raw_item: McpApprovalRequest | |
| """The raw MCP approval request.""" | |
| type: Literal["mcp_approval_request_item"] = "mcp_approval_request_item" | |
| class MCPApprovalResponseItem(RunItemBase[McpApprovalResponse]): | |
| """Represents a response to an MCP approval request.""" | |
| raw_item: McpApprovalResponse | |
| """The raw MCP approval response.""" | |
| type: Literal["mcp_approval_response_item"] = "mcp_approval_response_item" | |
| RunItem: TypeAlias = Union[ | |
| MessageOutputItem, | |
| HandoffCallItem, | |
| HandoffOutputItem, | |
| ToolCallItem, | |
| ToolCallOutputItem, | |
| ReasoningItem, | |
| MCPListToolsItem, | |
| MCPApprovalRequestItem, | |
| MCPApprovalResponseItem, | |
| ] | |
| """An item generated by an agent.""" | |
| class ModelResponse: | |
| output: list[TResponseOutputItem] | |
| """A list of outputs (messages, tool calls, etc) generated by the model""" | |
| usage: Usage | |
| """The usage information for the response.""" | |
| response_id: str | None | |
| """An ID for the response which can be used to refer to the response in subsequent calls to the | |
| model. Not supported by all model providers. | |
| If using OpenAI models via the Responses API, this is the `response_id` parameter, and it can | |
| be passed to `Runner.run`. | |
| """ | |
| def to_input_items(self) -> list[TResponseInputItem]: | |
| """Convert the output into a list of input items suitable for passing to the model.""" | |
| # We happen to know that the shape of the Pydantic output items are the same as the | |
| # equivalent TypedDict input items, so we can just convert each one. | |
| # This is also tested via unit tests. | |
| return [it.model_dump(exclude_unset=True) for it in self.output] # type: ignore | |
| class ItemHelpers: | |
| def extract_last_content(cls, message: TResponseOutputItem) -> str: | |
| """Extracts the last text content or refusal from a message.""" | |
| if not isinstance(message, ResponseOutputMessage): | |
| return "" | |
| last_content = message.content[-1] | |
| if isinstance(last_content, ResponseOutputText): | |
| return last_content.text | |
| elif isinstance(last_content, ResponseOutputRefusal): | |
| return last_content.refusal | |
| else: | |
| raise ModelBehaviorError(f"Unexpected content type: {type(last_content)}") | |
| def extract_last_text(cls, message: TResponseOutputItem) -> str | None: | |
| """Extracts the last text content from a message, if any. Ignores refusals.""" | |
| if isinstance(message, ResponseOutputMessage): | |
| last_content = message.content[-1] | |
| if isinstance(last_content, ResponseOutputText): | |
| return last_content.text | |
| return None | |
| def input_to_new_input_list( | |
| cls, input: str | list[TResponseInputItem] | |
| ) -> list[TResponseInputItem]: | |
| """Converts a string or list of input items into a list of input items.""" | |
| if isinstance(input, str): | |
| return [ | |
| { | |
| "content": input, | |
| "role": "user", | |
| } | |
| ] | |
| return copy.deepcopy(input) | |
| def text_message_outputs(cls, items: list[RunItem]) -> str: | |
| """Concatenates all the text content from a list of message output items.""" | |
| text = "" | |
| for item in items: | |
| if isinstance(item, MessageOutputItem): | |
| text += cls.text_message_output(item) | |
| return text | |
| def text_message_output(cls, message: MessageOutputItem) -> str: | |
| """Extracts all the text content from a single message output item.""" | |
| text = "" | |
| for item in message.raw_item.content: | |
| if isinstance(item, ResponseOutputText): | |
| text += item.text | |
| return text | |
| def tool_call_output_item( | |
| cls, tool_call: ResponseFunctionToolCall, output: str | |
| ) -> FunctionCallOutput: | |
| """Creates a tool call output item from a tool call and its output.""" | |
| return { | |
| "call_id": tool_call.call_id, | |
| "output": output, | |
| "type": "function_call_output", | |
| } | |