From 0421d97954e381bdbd6edc54cf640c23cdd76a2d Mon Sep 17 00:00:00 2001 From: James Hush Date: Wed, 17 Sep 2025 11:09:03 +0800 Subject: [PATCH] Save changes --- .../foundational/45-openai-agent-basic.py | 3 +- .../services/openai_agent/agent_service.py | 113 ++++++++++++++---- 2 files changed, 89 insertions(+), 27 deletions(-) diff --git a/examples/foundational/45-openai-agent-basic.py b/examples/foundational/45-openai-agent-basic.py index 17bc91e79..6d7456b7e 100644 --- a/examples/foundational/45-openai-agent-basic.py +++ b/examples/foundational/45-openai-agent-basic.py @@ -17,6 +17,7 @@ Requirements: import os import random +from typing import Any # Import agents SDK for tools and agent creation from agents import Agent, function_tool @@ -119,7 +120,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ) # Create tools for the agent - tools = [ + tools: list[Any] = [ get_weather, get_random_fact, ] diff --git a/src/pipecat/services/openai_agent/agent_service.py b/src/pipecat/services/openai_agent/agent_service.py index 42c2e9a35..7f6e5d742 100644 --- a/src/pipecat/services/openai_agent/agent_service.py +++ b/src/pipecat/services/openai_agent/agent_service.py @@ -6,19 +6,31 @@ """OpenAI Agents SDK integration service. -Provides integration with the OpenAI Agents SDK for building agentic AI applications +Provides integration with the OpenAI Agents SDK for building AI applications within Pipecat pipelines. This service allows leveraging agent loops, handoffs, guardrails, sessions, and tools from the OpenAI Agents SDK. """ import asyncio import os -from typing import Any, Awaitable, Callable, Dict, List, Optional, Union, override +from typing import ( + Any, + Awaitable, + Callable, + Dict, + List, + Optional, + Protocol, + Sequence, + Union, + override, + runtime_checkable, +) from loguru import logger try: - from agents import Agent, InputGuardrail, OutputGuardrail, Runner + from agents import Agent, InputGuardrail, OutputGuardrail, Runner, Tool from agents.result import RunResult, RunResultStreaming from agents.stream_events import StreamEvent except ImportError as e: @@ -45,6 +57,26 @@ from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_service import AIService +@runtime_checkable +class ToolLike(Protocol): + """Protocol for tool-like objects.""" + + def __call__(self, *args: Any, **kwargs: Any) -> Any: + """Tool call interface.""" + ... + + +@runtime_checkable +class AgentLike(Protocol): + """Protocol for agent-like objects.""" + + name: str + + def __call__(self, *args: Any, **kwargs: Any) -> Any: + """Agent call interface.""" + ... + + class OpenAIAgentService(AIService): """OpenAI Agents SDK service for Pipecat. @@ -61,11 +93,11 @@ class OpenAIAgentService(AIService): *, agent: Optional[Agent] = None, name: str = "Assistant", - instructions: str = "You are a helpful assistant.", - handoffs: Optional[List[Agent]] = None, - tools: Optional[List[Callable]] = None, - input_guardrails: Optional[List[InputGuardrail]] = None, - output_guardrails: Optional[List[OutputGuardrail]] = None, + instructions: Union[str, Sequence[str]] = "You are a helpful assistant.", + handoffs: Optional[Sequence[AgentLike]] = None, + tools: Optional[Sequence[ToolLike]] = None, + input_guardrails: Optional[Sequence[InputGuardrail]] = None, + output_guardrails: Optional[Sequence[OutputGuardrail]] = None, model_config: Optional[Dict[str, Any]] = None, session_config: Optional[Dict[str, Any]] = None, api_key: Optional[str] = None, @@ -101,13 +133,27 @@ class OpenAIAgentService(AIService): if agent: self._agent = agent else: + # Convert sequences to lists and handle string instructions + agent_handoffs: List[Any] = list(handoffs) if handoffs else [] + agent_tools: List[Any] = list(tools) if tools else [] + agent_input_guardrails: List[Any] = list(input_guardrails) if input_guardrails else [] + agent_output_guardrails: List[Any] = ( + list(output_guardrails) if output_guardrails else [] + ) + + # Handle instructions - convert sequence to string if needed + if isinstance(instructions, str): + agent_instructions = instructions + else: + agent_instructions = " ".join(str(instr) for instr in instructions) + self._agent = Agent( name=name, - instructions=instructions, - handoffs=handoffs or [], - tools=tools or [], - input_guardrails=input_guardrails or [], - output_guardrails=output_guardrails or [], + instructions=agent_instructions, + handoffs=agent_handoffs, + tools=agent_tools, + input_guardrails=agent_input_guardrails, + output_guardrails=agent_output_guardrails, model=model_config.get("model", "gpt-4o") if model_config else "gpt-4o", ) @@ -153,7 +199,8 @@ class OpenAIAgentService(AIService): logger.info(f"Updated agent instructions for {self._agent.name}") if model_config: - self._agent.model_config = model_config + # Note: OpenAI Agents SDK handles model configuration during agent creation + # We can't update model_config after agent is created, but we can update our model name if "model" in model_config: self.set_model_name(model_config["model"]) logger.info(f"Updated model config for {self._agent.name}") @@ -270,8 +317,11 @@ class OpenAIAgentService(AIService): async for event in result.stream_events(): if event.type == "raw_response_event": # Handle token-by-token streaming - if hasattr(event.data, "delta") and event.data.delta: - await self.push_frame(LLMTextFrame(text=event.data.delta)) + # Only check for delta on events that are known to have it + if hasattr(event.data, "delta") and getattr(event.data, "delta", None): + delta_text = getattr(event.data, "delta", "") + if delta_text: + await self.push_frame(LLMTextFrame(text=delta_text)) elif event.type == "run_item_stream_event": # Handle completed items @@ -286,10 +336,13 @@ class OpenAIAgentService(AIService): self._accumulated_text = message_text elif event.item.type == "tool_call_item": - logger.debug(f"Tool called: {event.item.tool_name}") + # Use getattr for safe attribute access + tool_name = getattr(event.item, "tool_name", "unknown") + logger.debug(f"Tool called: {tool_name}") elif event.item.type == "tool_call_output_item": - logger.debug(f"Tool output: {event.item.output}") + output = getattr(event.item, "output", "no output") + logger.debug(f"Tool output: {output}") elif event.type == "agent_updated_stream_event": logger.debug(f"Agent updated: {event.new_agent.name}") @@ -352,25 +405,33 @@ class OpenAIAgentService(AIService): logger.warning(f"Could not extract text from message item: {e}") return "" - async def add_tool(self, tool_function: Callable): + async def add_tool(self, tool_function: ToolLike): """Add a tool function to the agent. Args: - tool_function: A callable function to add as a tool. + tool_function: A callable function or Tool object to add as a tool. """ if hasattr(self._agent, "tools"): - self._agent.tools.append(tool_function) - logger.info(f"Added tool {tool_function.__name__} to agent {self._agent.name}") + # Cast to Any to handle the type variance issue + tools_list: List[Any] = self._agent.tools + tools_list.append(tool_function) + tool_name = getattr( + tool_function, "__name__", getattr(tool_function, "name", "unknown") + ) + logger.info(f"Added tool {tool_name} to agent {self._agent.name}") - async def add_handoff_agent(self, agent: Agent): + async def add_handoff_agent(self, agent: AgentLike): """Add a handoff agent. Args: - agent: Another Agent instance that this agent can hand off to. + agent: Another Agent instance or handoff object that this agent can hand off to. """ if hasattr(self._agent, "handoffs"): - self._agent.handoffs.append(agent) - logger.info(f"Added handoff agent {agent.name} to agent {self._agent.name}") + # Cast to Any to handle the type variance issue + handoffs_list: List[Any] = self._agent.handoffs + handoffs_list.append(agent) + agent_name = getattr(agent, "name", "unknown") + logger.info(f"Added handoff agent {agent_name} to agent {self._agent.name}") def get_session_context(self) -> Dict[str, Any]: """Get the current session context.