Save progress

examples/foundational/45-openai-agent-basic.py

@@ -18,6 +18,8 @@ Requirements:
 import os
 import random
 
+# Import agents SDK for tools and agent creation
+from agents import Agent, function_tool
 from dotenv import load_dotenv
 from loguru import logger
 
@@ -28,6 +30,7 @@ from pipecat.pipeline.task import PipelineTask
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.openai_agent.agent_service import OpenAIAgentService
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
@@ -37,32 +40,46 @@ load_dotenv(override=True)
 
 # Transport configuration
 transport_params = {
-    "daily": lambda: DailyParams(audio_out_enabled=True),
-    "twilio": lambda: FastAPIWebsocketParams(audio_out_enabled=True),
-    "webrtc": lambda: TransportParams(audio_out_enabled=True),
+    "daily": lambda: DailyParams(audio_out_enabled=True, audio_in_enabled=True),
+    "twilio": lambda: FastAPIWebsocketParams(audio_out_enabled=True, audio_in_enabled=True),
+    "webrtc": lambda: TransportParams(audio_out_enabled=True, audio_in_enabled=True),
 }
 
 
-def get_weather_tool():
-    """Example tool function for weather information."""
+@function_tool
+def get_weather(location: str) -> str:
+    """Get the current weather for a location.
 
-    def get_weather(location: str) -> str:
-        """Get the current weather for a location.
+    Args:
+        location: The location to get weather for
 
-        Args:
-            location: The city or location to get weather for.
+    Returns:
+        A weather description string
+    """
+    # Mock weather data - in real usage, integrate with weather API
+    weather_data = {
+        "San Francisco": "Foggy, 65°F",
+        "New York": "Sunny, 72°F",
+        "London": "Rainy, 59°F",
+        "Tokyo": "Partly cloudy, 68°F",
+    }
+    return weather_data.get(location, f"Weather data not available for {location}")
 
-        Returns:
-            A weather description string.
-        """
-        # Simulate weather data
-        conditions = ["sunny", "cloudy", "rainy", "snowy", "windy"]
-        temp = random.randint(-10, 35)
-        condition = random.choice(conditions)
 
-        return f"The weather in {location} is {condition} with a temperature of {temp}°C."
+@function_tool
+def get_random_fact() -> str:
+    """Get a random interesting fact.
 
-    return get_weather
+    Returns:
+        A random fact string
+    """
+    facts = [
+        "Honey never spoils. Archaeologists have found edible honey in ancient Egyptian tombs.",
+        "Octopuses have three hearts and blue blood.",
+        "The Great Wall of China isn't visible from space with the naked eye.",
+        "Bananas are berries, but strawberries aren't.",
+    ]
+    return random.choice(facts)
 
 
 def get_random_fact_tool():
@@ -89,6 +106,12 @@ def get_random_fact_tool():
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info("Starting OpenAI Agent bot")
 
+    # Set up STT for speech recognition
+    stt = DeepgramSTTService(
+        api_key=os.getenv("DEEPGRAM_API_KEY", ""),
+        model="nova-2",
+    )
+
     # Set up TTS for voice output
     tts = CartesiaTTSService(
         api_key=os.getenv("CARTESIA_API_KEY", ""),
@@ -97,12 +120,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     # Create tools for the agent
     tools = [
-        get_weather_tool(),
-        get_random_fact_tool(),
+        get_weather,
+        get_random_fact,
     ]
 
-    # Initialize the OpenAI Agent service
-    agent_service = OpenAIAgentService(
+    # Create the agent with tools
+    agent = Agent(
         name="Assistant",
         instructions="""You are a helpful assistant with access to weather information and random facts. 
         You can:
@@ -112,6 +135,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         
         Be friendly, informative, and engaging in your responses.""",
         tools=tools,
+    )
+
+    # Initialize the OpenAI Agent service with the pre-configured agent
+    agent_service = OpenAIAgentService(
+        agent=agent,
         api_key=os.getenv("OPENAI_API_KEY"),
         streaming=True,
     )
@@ -119,9 +147,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     # Create the processing pipeline
     pipeline = Pipeline(
         [
-            agent_service,
-            tts,
-            transport.output(),
+            transport.input(),  # Receive audio input
+            stt,  # Convert speech to text
+            agent_service,  # Process with OpenAI Agent
+            tts,  # Convert text to speech
+            transport.output(),  # Send audio output
         ]
     )
 

pyproject.toml

@@ -34,7 +34,7 @@ dependencies = [
     "pyloudnorm~=0.1.1",
     "resampy~=0.4.3",
     "soxr~=0.5.0",
-    "openai>=1.74.0,<=1.99.1",
+    "openai>=1.74.0,<2.0.0",
     # Pinning numba to resolve package dependencies
     "numba==0.61.2",
     "wait_for2>=0.4.1; python_version<'3.12'",
@@ -74,7 +74,7 @@ langchain = [ "langchain~=0.3.20", "langchain-community~=0.3.20", "langchain-ope
 livekit = [ "livekit~=0.22.0", "livekit-api~=0.8.2", "tenacity>=8.2.3,<10.0.0" ]
 lmnt = [ "websockets>=13.1,<15.0" ]
 local = [ "pyaudio~=0.2.14" ]
-mcp = [ "mcp[cli]~=1.9.4" ]
+mcp = [ "mcp[cli]>=1.11.0,<2.0.0" ]
 mem0 = [ "mem0ai~=0.1.94" ]
 mistral = []
 mlx-whisper = [ "mlx-whisper~=0.4.2" ]
@@ -83,8 +83,8 @@ nim = []
 neuphonic = [ "websockets>=13.1,<15.0" ]
 noisereduce = [ "noisereduce~=3.0.3" ]
 openai = [ "websockets>=13.1,<15.0" ]
-openai-agent = [ "openai-agents~=1.0.0" ]
-openpipe = [ "openpipe~=4.50.0" ]
+openai-agent = [ "openai-agents~=0.3.0" ]
+# openpipe = [ "openpipe~=4.50.0" ]  # Temporarily disabled due to openai version conflict
 openrouter = []
 perplexity = []
 playht = [ "websockets>=13.1,<15.0" ]

src/pipecat/services/openai_agent/README.md

@@ -206,4 +206,4 @@ Errors are emitted as `ErrorFrame` objects in the pipeline.
 
 - Currently supports OpenAI models only (via Agents SDK)
 - Handoffs work within individual requests (no cross-request state)
-- Real-time voice features require additional setup
+- Real-time voice features require additional setup

src/pipecat/services/openai_agent/agent_service.py

@@ -13,7 +13,7 @@ guardrails, sessions, and tools from the OpenAI Agents SDK.
 
 import asyncio
 import os
-from typing import Any, Awaitable, Callable, Dict, List, Optional, Union
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Union, override
 
 from loguru import logger
 
@@ -108,8 +108,7 @@ class OpenAIAgentService(AIService):
                 tools=tools or [],
                 input_guardrails=input_guardrails or [],
                 output_guardrails=output_guardrails or [],
-                model_config=model_config,
-                **kwargs,
+                model=model_config.get("model", "gpt-4o") if model_config else "gpt-4o",
             )
 
         self._streaming = streaming
@@ -141,7 +140,7 @@ class OpenAIAgentService(AIService):
         instructions: Optional[str] = None,
         model_config: Optional[Dict[str, Any]] = None,
         **kwargs,
-    ):
+    ) -> None:
         """Update agent configuration dynamically.
 
         Args:
@@ -206,7 +205,8 @@ class OpenAIAgentService(AIService):
 
         await super().cancel(frame)
 
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
+    @override
+    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
         """Process frames and handle agent interactions.
 
         Processes text input frames by running them through the OpenAI Agent

test_openai_agent.py

@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+
+"""Simple test script for OpenAI Agent service."""
+
+import asyncio
+import os
+from unittest.mock import MagicMock, patch
+
+# Mock the OpenAI API key for testing
+os.environ["OPENAI_API_KEY"] = "test-key-for-testing"
+
+from pipecat.frames.frames import TextFrame
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.openai_agent import OpenAIAgentService
+
+
+async def test_basic_functionality():
+    """Test basic OpenAI Agent service functionality."""
+    print("🧪 Testing OpenAI Agent Service...")
+
+    # Create a simple weather tool for testing
+    def get_weather(location: str) -> str:
+        """Get weather for a location."""
+        return f"The weather in {location} is sunny and 22°C."
+
+    try:
+        # Create the service
+        print("📋 Creating OpenAI Agent service...")
+        service = OpenAIAgentService(
+            name="Test Assistant",
+            instructions="You are a helpful test assistant.",
+            tools=[get_weather],
+            api_key="test-key",
+            streaming=True,
+        )
+
+        print(f"✅ Service created successfully!")
+        print(f"   - Agent name: {service.agent.name}")
+        print(f"   - Model name: {service.model_name}")
+        print(f"   - Streaming enabled: {service._streaming}")
+
+        # Test basic configuration
+        print("⚙️  Testing configuration updates...")
+        service.update_agent_config(
+            instructions="Updated test instructions",
+            model_config={"model": "gpt-4o", "temperature": 0.5},
+        )
+
+        print(f"✅ Configuration updated!")
+        print(f"   - New instructions: {service.agent.instructions}")
+        print(f"   - New model: {service.model_name}")
+
+        # Test session context
+        print("💾 Testing session context...")
+        service.update_session_context({"user_id": "test-user", "session": "test-session"})
+        context = service.get_session_context()
+
+        print(f"✅ Session context managed!")
+        print(f"   - Context keys: {list(context.keys())}")
+
+        # Test adding tools
+        print("🔧 Testing tool management...")
+
+        def get_time() -> str:
+            """Get current time."""
+            return "The current time is 3:00 PM."
+
+        await service.add_tool(get_time)
+        print(f"✅ Tool added successfully!")
+
+        print("\n🎉 All basic functionality tests passed!")
+        return True
+
+    except Exception as e:
+        print(f"❌ Test failed with error: {e}")
+        return False
+
+
+async def test_frame_processing():
+    """Test frame processing with mocked responses."""
+    print("\n🔄 Testing frame processing...")
+
+    try:
+        # Mock the Runner to avoid actual API calls
+        with patch("pipecat.services.openai_agent.agent_service.Runner") as mock_runner:
+            # Set up mock responses
+            mock_stream_result = MagicMock()
+
+            # Mock stream events
+            async def mock_stream_events():
+                # Simulate streaming response
+                yield MagicMock(type="raw_response_event", data=MagicMock(delta="Hello "))
+                yield MagicMock(type="raw_response_event", data=MagicMock(delta="from "))
+                yield MagicMock(type="raw_response_event", data=MagicMock(delta="agent!"))
+
+                # Simulate completed message
+                mock_item = MagicMock()
+                mock_item.type = "message_output_item"
+                mock_item.content = "Hello from agent!"
+                yield MagicMock(type="run_item_stream_event", item=mock_item)
+
+            mock_stream_result.stream_events.return_value = mock_stream_events()
+            mock_runner.run_streamed.return_value = mock_stream_result
+
+            # Create service with mocked runner
+            service = OpenAIAgentService(
+                name="Test Assistant",
+                instructions="You are a helpful test assistant.",
+                api_key="test-key",
+                streaming=True,
+            )
+
+            # Collect output frames
+            output_frames = []
+
+            async def mock_push_frame(frame, direction=FrameDirection.DOWNSTREAM):
+                output_frames.append(frame)
+                print(f"   📤 Frame: {type(frame).__name__}")
+                if hasattr(frame, "text"):
+                    print(f"      Text: '{frame.text}'")
+
+            service.push_frame = mock_push_frame
+
+            # Process a text frame
+            print("📝 Processing text frame...")
+            text_frame = TextFrame("Hello, how are you?")
+            await service.process_frame(text_frame, FrameDirection.DOWNSTREAM)
+
+            # Wait for async processing
+            await asyncio.sleep(0.2)
+
+            print(f"✅ Frame processing completed!")
+            print(f"   - Generated {len(output_frames)} output frames")
+
+            # Check if we got expected frame types
+            frame_types = [type(frame).__name__ for frame in output_frames]
+            print(f"   - Frame types: {frame_types}")
+
+            return True
+
+    except Exception as e:
+        print(f"❌ Frame processing test failed: {e}")
+        return False
+
+
+async def main():
+    """Run all tests."""
+    print("🚀 Starting OpenAI Agent Service Tests\n")
+
+    try:
+        # Run basic functionality tests
+        basic_test = await test_basic_functionality()
+
+        # Run frame processing tests
+        frame_test = await test_frame_processing()
+
+        # Summary
+        print(f"\n📊 Test Results:")
+        print(f"   - Basic functionality: {'✅ PASS' if basic_test else '❌ FAIL'}")
+        print(f"   - Frame processing: {'✅ PASS' if frame_test else '❌ FAIL'}")
+
+        if basic_test and frame_test:
+            print(f"\n🎉 All tests passed! The OpenAI Agent service is working correctly.")
+        else:
+            print(f"\n⚠️  Some tests failed. Please check the output above.")
+
+    except Exception as e:
+        print(f"❌ Test suite failed with error: {e}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

test_simple_agent.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+
+import asyncio
+import os
+
+from loguru import logger
+
+# Test the actual agents package API
+try:
+    from agents import Agent, run
+
+    # Create a simple agent
+    agent = Agent(
+        name="test-agent",
+        instructions="You are a helpful assistant.",
+    )
+
+    print("✅ Agent created successfully!")
+    print(f"Agent name: {agent.name}")
+
+    # Test a simple conversation
+    async def test_agent():
+        result = await run(agent, "Hello, how are you?")
+        print(f"Agent response: {result}")
+
+    # Run the test
+    asyncio.run(test_agent())
+
+except Exception as e:
+    print(f"❌ Error: {e}")
+    import traceback
+
+    traceback.print_exc()