From b086fbafe6e920731ef8f934d1d2005b107303f2 Mon Sep 17 00:00:00 2001 From: James Hush Date: Tue, 16 Sep 2025 16:20:30 +0800 Subject: [PATCH] feat: Add OpenAI Agents SDK integration service - Create new OpenAIAgentService that integrates OpenAI Agents SDK with Pipecat - Support for agent loops, handoffs, guardrails, and session management - Add streaming and non-streaming response modes - Include comprehensive tool integration and error handling - Add optional dependency for openai-agents package - Create foundational examples showing basic usage and agent handoffs - Add comprehensive tests with mocked dependencies - Include detailed documentation and README Key features: - Real-time streaming responses compatible with Pipecat pipelines - Agent handoffs for specialized task delegation - Tool calling with automatic schema generation - Input/output guardrails for safety and validation - Session context management for conversation continuity - Built-in tracing and monitoring integration Examples: - 45-openai-agent-basic.py: Basic agent with weather and trivia tools - 46-openai-agent-handoffs.py: Multi-agent system with specialist handoffs --- AGENTS.md | 285 +++++++++++++ .../foundational/45-openai-agent-basic.py | 161 ++++++++ .../foundational/46-openai-agent-handoffs.py | 254 ++++++++++++ pyproject.toml | 1 + src/pipecat/services/openai_agent/README.md | 209 ++++++++++ src/pipecat/services/openai_agent/__init__.py | 11 + .../services/openai_agent/agent_service.py | 390 ++++++++++++++++++ tests/test_openai_agent_service.py | 286 +++++++++++++ 8 files changed, 1597 insertions(+) create mode 100644 AGENTS.md create mode 100644 examples/foundational/45-openai-agent-basic.py create mode 100644 examples/foundational/46-openai-agent-handoffs.py create mode 100644 src/pipecat/services/openai_agent/README.md create mode 100644 src/pipecat/services/openai_agent/__init__.py create mode 100644 src/pipecat/services/openai_agent/agent_service.py create mode 100644 tests/test_openai_agent_service.py diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 000000000..3025c1413 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,285 @@ +# AGENTS.md + +## Project Overview + +Pipecat is an open-source Python framework for building real-time voice and multimodal conversational AI agents. The codebase is organized around a pipeline architecture where data flows through connected services (STT → LLM → TTS). + +## Development Environment Setup + +### Prerequisites +- **Minimum Python Version:** 3.10 +- **Recommended Python Version:** 3.12 +- **Package Manager:** uv (recommended) or pip + +### Setup Commands + +```bash +# Clone the repository +git clone https://github.com/pipecat-ai/pipecat.git +cd pipecat + +# Install dependencies with uv (recommended) +uv sync --group dev --all-extras \ + --no-extra gstreamer \ + --no-extra krisp \ + --no-extra local \ + --no-extra ultravox + +# Or with pip +pip install -e ".[dev]" + +# Install pre-commit hooks +uv run pre-commit install + +# Set up environment variables +cp env.example .env +``` + +## Build and Test Commands + +### Running Tests +```bash +# Run all tests +uv run pytest + +# Run specific test file +uv run pytest tests/test_name.py + +# Run tests with coverage +uv run pytest --cov=pipecat --cov-report=html +``` + +### Code Quality +```bash +# Format code (required before commits) +uv run ruff format + +# Lint code +uv run ruff check + +# Type checking +uv run mypy src/pipecat + +# Run pre-commit checks manually +uv run pre-commit run --all-files +``` + +### Documentation +```bash +# Build API documentation +cd docs/api +./build-docs.sh + +# Build docs manually +sphinx-build -b html . _build/html -W --keep-going +``` + +## Code Style Guidelines + +### Python Standards +- **Formatting:** Strict PEP 8 via Ruff +- **Docstrings:** Google-style format +- **Type Hints:** Required for all public APIs +- **Import Organization:** Automated via Ruff + +### Docstring Conventions +- **Classes:** Describe purpose + `__init__` with complete `Args:` section +- **Dataclasses:** Use `Parameters:` section, no `__init__` docstring +- **Methods:** Include `Args:` and `Returns:` sections +- **Properties:** Must have `Returns:` section +- **Examples:** Use `Examples:` section with `::` syntax + +### File Organization +``` +src/pipecat/ # Main package +├── processors/ # Frame processors +├── services/ # AI service integrations +├── transports/ # Communication layers +├── frames/ # Data frame definitions +└── pipeline/ # Pipeline orchestration + +examples/foundational/ # Step-by-step tutorials +tests/ # Test suite +``` + +## Testing Instructions + +### Test Structure +- **Unit Tests:** Test individual components in isolation +- **Integration Tests:** Test service interactions +- **Example Tests:** Validate foundational examples work + +### Adding Tests +```bash +# Test naming convention +test__.py + +# Run specific test pattern +uv run pytest -k "test_pipeline" + +# Run with debugging +uv run pytest -s -vv tests/test_name.py::test_function +``` + +### Pre-commit Requirements +All commits must pass: +- Ruff formatting +- Ruff linting +- Type checking +- Basic test suite + +## Dependency Management + +### Using uv (Recommended) +```bash +# Add runtime dependency +uv add package-name + +# Add optional dependency +uv add --optional service package-name + +# Add development dependency +uv add --group dev package-name + +# Update lockfile +uv lock + +# Sync dependencies +uv sync +``` + +### Important Notes +- **Always commit both `pyproject.toml` and `uv.lock` together** +- **Never manually edit `uv.lock`** - it's auto-generated +- **Use extras for optional service dependencies** (e.g., `[openai]`, `[cartesia]`) + +## Project Structure Guidelines + +### Service Integration +When adding new AI services: +1. Create service class in `src/pipecat/services//` +2. Follow existing patterns (e.g., STTService, LLMService) +3. Add to appropriate extras in `pyproject.toml` +4. Include tests in `tests/` +5. Add documentation examples + +### Frame Processing +For custom processors: +1. Inherit from `FrameProcessor` +2. Implement `process_frame()` method. ALWAYS explicitly call `await super().process_frame(frame, direction)` at the top of this method. +3. Handle frame direction (FrameDirection.UPSTREAM/DOWNSTREAM) +4. Add proper type hints and docstrings + +### Transport Implementation +For new transport layers: +1. Inherit from `BaseTransport` +2. Implement required abstract methods +3. Handle connection lifecycle +4. Support both input and output streams + +## Security Considerations + +### API Keys +- **Never commit API keys** to the repository +- **Use environment variables** for all secrets +- **Reference `env.example`** for required variables +- **Use `.env` files** for local development + +### Input Validation +- **Validate all external inputs** (audio, text, API responses) +- **Sanitize user data** before processing +- **Handle rate limiting** for external services +- **Implement proper timeout handling** + +## Performance Guidelines + +### Memory Management +- **Clean up resources** in transport disconnection handlers +- **Use async context managers** for service connections +- **Implement proper frame lifecycle** management + +### Latency Optimization +- **Choose appropriate STT services** for latency requirements +- **Use streaming TTS** when possible +- **Implement connection pooling** for HTTP services +- **Consider WebRTC** for real-time applications + +## Common Patterns + +### Error Handling +```python +@transport.event_handler("on_error") +async def on_error(transport, error): + logger.error(f"Transport error: {error}") + + # Shutdown the pipeline + await task.queue_frame(EndFrame()) + +``` + +### Service Configuration +```python +# Use environment variables for configuration +service = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY", ""), + model="gpt-4o", + params={"temperature": 0.7} +) +``` + +### Pipeline Assembly +```python +pipeline = Pipeline([ + transport.input(), + stt_service, + context_aggregator.user(), + llm_service, + tts_service, + transport.output(), + context_aggregator.assistant(), +]) +``` + +## Commit and PR Guidelines + +### Commit Message Format +``` +(): + +[optional body] + +[optional footer] +``` + +Types: `feat`, `fix`, `docs`, `style`, `refactor`, `test`, `chore` + +### PR Requirements +- **All tests must pass** +- **Code must be properly formatted** (Ruff) +- **Include appropriate tests** for new functionality +- **Update documentation** if needed +- **Reference related issues** in description + +### Review Process +1. Automated checks must pass +2. Manual code review by maintainers +3. Documentation review for user-facing changes +4. Integration testing for service additions + +## Troubleshooting + +### Common Issues +- **Import errors:** Run `uv sync` to ensure dependencies are installed +- **Test failures:** Check environment variables in `.env` +- **Format errors:** Run `uv run ruff format` before committing +- **Type errors:** Ensure all public methods have type hints + +### Development Tips +- **Use foundational examples** as starting points for testing +- **Check existing services** for integration patterns +- **Run tests frequently** during development +- **Use IDE integration** for Ruff formatting + +### Getting Help +- **Documentation:** [docs.pipecat.ai](https://docs.pipecat.ai) +- **Issues:** [GitHub Issues](https://github.com/pipecat-ai/pipecat/issues) diff --git a/examples/foundational/45-openai-agent-basic.py b/examples/foundational/45-openai-agent-basic.py new file mode 100644 index 000000000..cff0e1cfd --- /dev/null +++ b/examples/foundational/45-openai-agent-basic.py @@ -0,0 +1,161 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +""" +Basic OpenAI Agent service example. + +This example demonstrates how to use the OpenAI Agents SDK within a Pipecat +pipeline to create an interactive agent with tool calling capabilities. + +Requirements: +- OpenAI API key +- OpenAI Agents SDK: pip install openai-agents +""" + +import os +import random + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.frames.frames import EndFrame, TextFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +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.openai_agent.agent_service import OpenAIAgentService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +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), +} + + +def get_weather_tool(): + """Example tool function for weather information.""" + + def get_weather(location: str) -> str: + """Get the current weather for a location. + + Args: + location: The city or location to get weather for. + + 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." + + return get_weather + + +def get_random_fact_tool(): + """Example tool function for random facts.""" + + def get_random_fact() -> str: + """Get a random interesting fact. + + Returns: + A random fact string. + """ + facts = [ + "Honey never spoils. Archaeologists have found edible honey in ancient Egyptian tombs.", + "A group of flamingos is called a 'flamboyance'.", + "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) + + return get_random_fact + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info("Starting OpenAI Agent bot") + + # Set up TTS for voice output + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY", ""), + voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ) + + # Create tools for the agent + tools = [ + get_weather_tool(), + get_random_fact_tool(), + ] + + # Initialize the OpenAI Agent service + agent_service = OpenAIAgentService( + name="Assistant", + instructions="""You are a helpful assistant with access to weather information and random facts. + You can: + - Check weather for any location using the get_weather tool + - Share interesting facts using the get_random_fact tool + - Have natural conversations + + Be friendly, informative, and engaging in your responses.""", + tools=tools, + api_key=os.getenv("OPENAI_API_KEY"), + streaming=True, + ) + + # Create the processing pipeline + pipeline = Pipeline( + [ + agent_service, + tts, + transport.output(), + ] + ) + + task = PipelineTask( + pipeline, + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + # Send an initial greeting when client connects + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info("Client connected, sending greeting") + await task.queue_frames( + [ + TextFrame( + "Hello! I'm an AI assistant powered by the OpenAI Agents SDK. " + "I can help you with weather information, share interesting facts, " + "or just have a conversation. What would you like to know?" + ), + EndFrame(), + ] + ) + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/examples/foundational/46-openai-agent-handoffs.py b/examples/foundational/46-openai-agent-handoffs.py new file mode 100644 index 000000000..022a2a1d8 --- /dev/null +++ b/examples/foundational/46-openai-agent-handoffs.py @@ -0,0 +1,254 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +""" +Advanced OpenAI Agent service example with handoffs. + +This example demonstrates how to use multiple agents with handoffs in the +OpenAI Agents SDK within a Pipecat pipeline, showcasing agent orchestration +and specialization. + +Requirements: +- OpenAI API key +- OpenAI Agents SDK: pip install openai-agents +""" + +import os +import random +from typing import Any, Dict + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.frames.frames import EndFrame, TextFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +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.openai_agent.agent_service import OpenAIAgentService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +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), +} + + +def create_weather_tools(): + """Create weather-related tools.""" + + def get_weather(location: str) -> str: + """Get current weather for a location.""" + 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." + + def get_forecast(location: str, days: int = 3) -> str: + """Get weather forecast for multiple days.""" + forecast = [] + for i in range(days): + conditions = ["sunny", "cloudy", "rainy", "snowy"] + temp = random.randint(-5, 30) + condition = random.choice(conditions) + day = "today" if i == 0 else f"in {i} day{'s' if i > 1 else ''}" + forecast.append(f"{day.capitalize()}: {condition}, {temp}°C") + return f"Weather forecast for {location}:\n" + "\n".join(forecast) + + return [get_weather, get_forecast] + + +def create_trivia_tools(): + """Create trivia and fact tools.""" + + def get_random_fact() -> str: + """Get a random interesting fact.""" + facts = [ + "Honey never spoils. Archaeologists have found edible honey in ancient Egyptian tombs.", + "A group of flamingos is called a 'flamboyance'.", + "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.", + "Wombat poop is cube-shaped.", + "A shrimp's heart is in its head.", + "It's impossible to hum while holding your nose.", + ] + return random.choice(facts) + + def get_science_fact() -> str: + """Get a random science fact.""" + facts = [ + "The speed of light in a vacuum is approximately 299,792,458 meters per second.", + "DNA stands for Deoxyribonucleic Acid.", + "The human brain uses about 20% of the body's total energy.", + "There are more possible games of chess than atoms in the observable universe.", + "A single bolt of lightning contains enough energy to toast 100,000 slices of bread.", + ] + return random.choice(facts) + + return [get_random_fact, get_science_fact] + + +def create_math_tools(): + """Create math calculation tools.""" + + def calculate(expression: str) -> str: + """Safely calculate a mathematical expression.""" + try: + # Only allow basic math operations for safety + allowed_chars = set("0123456789+-*/.() ") + if not all(c in allowed_chars for c in expression): + return "Sorry, I can only calculate basic math expressions with +, -, *, /, and parentheses." + + result = eval(expression) + return f"{expression} = {result}" + except Exception as e: + return f"Error calculating '{expression}': {str(e)}" + + def generate_math_problem() -> str: + """Generate a random math problem.""" + operations = ["+", "-", "*"] + a = random.randint(1, 20) + b = random.randint(1, 20) + op = random.choice(operations) + + if op == "+": + answer = a + b + elif op == "-": + answer = a - b + else: # multiplication + answer = a * b + + return f"Here's a math problem for you: {a} {op} {b} = ?" + + return [calculate, generate_math_problem] + + +async def create_specialist_agents(): + """Create specialized agents for different domains.""" + + # Weather specialist agent + weather_agent = OpenAIAgentService( + name="Weather Specialist", + instructions="""You are a weather specialist. You provide detailed weather information, + forecasts, and weather-related advice. Use your tools to get accurate weather data. + Be informative and helpful about weather conditions and what they might mean for + outdoor activities.""", + tools=create_weather_tools(), + api_key=os.getenv("OPENAI_API_KEY"), + streaming=True, + ) + + # Trivia specialist agent + trivia_agent = OpenAIAgentService( + name="Trivia Master", + instructions="""You are a trivia and facts specialist. You love sharing interesting + facts, trivia, and educational content. Use your tools to provide fascinating + information and engage users with fun facts. Make learning enjoyable!""", + tools=create_trivia_tools(), + api_key=os.getenv("OPENAI_API_KEY"), + streaming=True, + ) + + # Math specialist agent + math_agent = OpenAIAgentService( + name="Math Helper", + instructions="""You are a mathematics specialist. You help with calculations, + math problems, and mathematical concepts. Use your tools to solve problems + and generate practice questions. Make math accessible and fun!""", + tools=create_math_tools(), + api_key=os.getenv("OPENAI_API_KEY"), + streaming=True, + ) + + return weather_agent, trivia_agent, math_agent + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info("Starting OpenAI Agent bot with handoffs") + + # Set up TTS for voice output + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY", ""), + voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ) + + # Create specialist agents + weather_agent, trivia_agent, math_agent = await create_specialist_agents() + + # Create the main triage agent that can hand off to specialists + triage_agent = OpenAIAgentService( + name="Assistant Coordinator", + instructions="""You are a helpful assistant coordinator. Your role is to understand + what the user needs and direct them to the right specialist: + + - For weather questions, forecasts, or outdoor activity planning -> Weather Specialist + - For interesting facts, trivia, or educational content -> Trivia Master + - For calculations, math problems, or mathematical help -> Math Helper + + If the request doesn't clearly fit a specialist, you can handle general conversation + yourself. Always be friendly and explain when you're connecting them to a specialist.""", + handoffs=[weather_agent.agent, trivia_agent.agent, math_agent.agent], + api_key=os.getenv("OPENAI_API_KEY"), + streaming=True, + ) + + # Create the processing pipeline (using just the triage agent) + # Note: In a real implementation, you might want to handle handoffs + # by switching the active agent in the pipeline dynamically + pipeline = Pipeline( + [ + triage_agent, + tts, + transport.output(), + ] + ) + + task = PipelineTask( + pipeline, + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + # Send an initial greeting when client connects + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info("Client connected, sending greeting") + await task.queue_frames( + [ + TextFrame( + "Hello! I'm your AI assistant coordinator. I work with a team of specialists " + "who can help you with different topics:\n\n" + "🌤️ Weather Specialist - for weather information and forecasts\n" + "🧠 Trivia Master - for interesting facts and trivia\n" + "🔢 Math Helper - for calculations and math problems\n\n" + "What would you like help with today?" + ), + EndFrame(), + ] + ) + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/pyproject.toml b/pyproject.toml index 8c86cbf1c..918e3ac8f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -83,6 +83,7 @@ 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" ] openrouter = [] perplexity = [] diff --git a/src/pipecat/services/openai_agent/README.md b/src/pipecat/services/openai_agent/README.md new file mode 100644 index 000000000..e38ed64a6 --- /dev/null +++ b/src/pipecat/services/openai_agent/README.md @@ -0,0 +1,209 @@ +# OpenAI Agents SDK Integration + +This service integrates the [OpenAI Agents SDK](https://openai.github.io/openai-agents-python/) with Pipecat, enabling powerful agentic workflows with features like: + +- **Agent loops** with tool calling and response streaming +- **Handoffs** between specialized agents +- **Guardrails** for input/output validation +- **Sessions** with automatic conversation history +- **Built-in tracing** and monitoring + +## Installation + +Install the OpenAI Agents SDK dependency: + +```bash +pip install "pipecat-ai[openai-agent]" +# or +uv add "pipecat-ai[openai-agent]" +``` + +## Basic Usage + +```python +from pipecat.services.openai_agent import OpenAIAgentService + +# Create a simple agent +agent_service = OpenAIAgentService( + name="Assistant", + instructions="You are a helpful assistant.", + api_key=os.getenv("OPENAI_API_KEY"), + streaming=True, +) + +# Use in a pipeline +pipeline = Pipeline([ + transport.input(), + stt, + agent_service, + tts, + transport.output(), +]) +``` + +## Features + +### Tool Integration + +```python +def get_weather(location: str) -> str: + """Get weather for a location.""" + return f"Weather in {location}: sunny, 22°C" + +agent_service = OpenAIAgentService( + name="Weather Assistant", + instructions="Help users with weather information.", + tools=[get_weather], + api_key=os.getenv("OPENAI_API_KEY"), +) +``` + +### Agent Handoffs + +```python +# Create specialized agents +weather_agent = OpenAIAgentService( + name="Weather Specialist", + instructions="Provide weather information and forecasts.", + tools=[get_weather, get_forecast], +) + +trivia_agent = OpenAIAgentService( + name="Trivia Master", + instructions="Share interesting facts and trivia.", + tools=[get_random_fact], +) + +# Create coordinator that can hand off to specialists +coordinator = OpenAIAgentService( + name="Coordinator", + instructions="Route users to the right specialist.", + handoffs=[weather_agent.agent, trivia_agent.agent], +) +``` + +### Guardrails + +```python +from agents import InputGuardrail, GuardrailFunctionOutput + +async def content_filter(ctx, agent, input_data): + # Check input for appropriate content + if is_inappropriate(input_data): + return GuardrailFunctionOutput( + tripwire_triggered=True, + output_info="Content not allowed" + ) + return GuardrailFunctionOutput(tripwire_triggered=False) + +agent_service = OpenAIAgentService( + name="Safe Assistant", + instructions="You are a helpful and safe assistant.", + input_guardrails=[InputGuardrail(guardrail_function=content_filter)], +) +``` + +### Session Management + +```python +agent_service = OpenAIAgentService( + name="Personal Assistant", + instructions="Remember user preferences and context.", + session_config={ + "user_id": "user_123", + "memory_enabled": True, + } +) + +# Update session context dynamically +agent_service.update_session_context({ + "user_preferences": {"language": "en", "style": "formal"} +}) +``` + +## Configuration Options + +### Basic Parameters + +- `name`: Agent identifier for handoffs and tracing +- `instructions`: System prompt defining agent behavior +- `api_key`: OpenAI API key (or use `OPENAI_API_KEY` env var) +- `streaming`: Enable real-time token streaming (default: True) + +### Advanced Configuration + +- `tools`: List of callable functions for the agent to use +- `handoffs`: List of other agents this agent can transfer to +- `input_guardrails`: Input validation and filtering +- `output_guardrails`: Output validation and filtering +- `model_config`: Model settings (model, temperature, etc.) +- `session_config`: Session and memory configuration + +### Model Configuration + +```python +agent_service = OpenAIAgentService( + name="Precise Assistant", + instructions="Provide accurate, concise responses.", + model_config={ + "model": "gpt-4o", + "temperature": 0.1, + "max_tokens": 150, + } +) +``` + +## Examples + +See the foundational examples: + +- [`45-openai-agent-basic.py`](../examples/foundational/45-openai-agent-basic.py) - Basic agent with tools +- [`46-openai-agent-handoffs.py`](../examples/foundational/46-openai-agent-handoffs.py) - Multi-agent system with handoffs + +## Methods + +### Core Methods + +- `update_agent_config()` - Update instructions and model settings +- `add_tool()` - Add new tools dynamically +- `add_handoff_agent()` - Add handoff destinations +- `get_session_context()` - Get current session state +- `update_session_context()` - Update session variables + +### Lifecycle Methods + +Inherited from `AIService`: +- `start()` - Initialize the agent +- `stop()` - Clean up resources +- `cancel()` - Cancel ongoing operations + +## Integration with Pipecat + +The service processes `TextFrame` inputs and generates: +- `LLMFullResponseStartFrame` - Response beginning +- `LLMTextFrame` - Streaming text tokens (if streaming enabled) +- `LLMFullResponseEndFrame` - Response completion + +This integrates seamlessly with Pipecat's conversation pipeline and context aggregators. + +## Error Handling + +The service includes robust error handling for: +- Missing API keys or SDK installation +- Agent processing failures +- Network connectivity issues +- Malformed tool responses + +Errors are emitted as `ErrorFrame` objects in the pipeline. + +## Requirements + +- OpenAI API key +- `openai-agents` package +- Python 3.10+ + +## Limitations + +- Currently supports OpenAI models only (via Agents SDK) +- Handoffs work within individual requests (no cross-request state) +- Real-time voice features require additional setup \ No newline at end of file diff --git a/src/pipecat/services/openai_agent/__init__.py b/src/pipecat/services/openai_agent/__init__.py new file mode 100644 index 000000000..1cae2e8d3 --- /dev/null +++ b/src/pipecat/services/openai_agent/__init__.py @@ -0,0 +1,11 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""OpenAI Agents SDK service for Pipecat integration.""" + +from .agent_service import OpenAIAgentService + +__all__ = ["OpenAIAgentService"] diff --git a/src/pipecat/services/openai_agent/agent_service.py b/src/pipecat/services/openai_agent/agent_service.py new file mode 100644 index 000000000..16917e293 --- /dev/null +++ b/src/pipecat/services/openai_agent/agent_service.py @@ -0,0 +1,390 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""OpenAI Agents SDK integration service. + +Provides integration with the OpenAI Agents SDK for building agentic 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 + +from loguru import logger + +try: + from agents import Agent, InputGuardrail, OutputGuardrail, Runner + from agents.result import RunResult, RunResultStreaming + from agents.stream_events import StreamEvent +except ImportError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use OpenAI Agents SDK, you need to `pip install openai-agents`. " + "Also, set `OPENAI_API_KEY` environment variable." + ) + raise Exception(f"Missing module: {e}") + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + ErrorFrame, + Frame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, + StartFrame, + TextFrame, + UserImageRawFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.ai_service import AIService + + +class OpenAIAgentService(AIService): + """OpenAI Agents SDK service for Pipecat. + + Integrates the OpenAI Agents SDK with Pipecat's pipeline architecture, + enabling advanced agentic workflows with features like handoffs, guardrails, + sessions, and tools within real-time conversational AI applications. + + The service processes text input frames and generates streaming responses + using the agent's configured capabilities. + """ + + def __init__( + self, + *, + 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, + model_config: Optional[Dict[str, Any]] = None, + session_config: Optional[Dict[str, Any]] = None, + api_key: Optional[str] = None, + streaming: bool = True, + **kwargs, + ): + """Initialize the OpenAI Agent service. + + Args: + agent: Pre-configured Agent instance. If provided, other agent configuration + parameters will be ignored. + name: Name of the agent for identification and handoffs. + instructions: System instructions that define the agent's behavior. + handoffs: List of other agents this agent can hand off to. + tools: List of callable functions the agent can use as tools. + input_guardrails: List of input validation guardrails. + output_guardrails: List of output validation guardrails. + model_config: Configuration for the underlying language model. + session_config: Configuration for session management. + api_key: OpenAI API key. If not provided, will use OPENAI_API_KEY env var. + streaming: Whether to use streaming responses for real-time output. + **kwargs: Additional arguments passed to the parent AIService. + """ + super().__init__(**kwargs) + + # Set up API key + if api_key: + os.environ["OPENAI_API_KEY"] = api_key + elif not os.getenv("OPENAI_API_KEY"): + logger.warning("No OpenAI API key provided. Set OPENAI_API_KEY environment variable.") + + # Create or use existing agent + if agent: + self._agent = agent + else: + self._agent = Agent( + name=name, + instructions=instructions, + handoffs=handoffs or [], + tools=tools or [], + input_guardrails=input_guardrails or [], + output_guardrails=output_guardrails or [], + model_config=model_config, + **kwargs, + ) + + self._streaming = streaming + self._session_config = session_config or {} + self._current_session = None + self._accumulated_text = "" + self._processing_task: Optional[asyncio.Task] = None + + # Set model name for metrics + if model_config and "model" in model_config: + self.set_model_name(model_config["model"]) + else: + self.set_model_name("gpt-4o") # Default model + + logger.info(f"Initialized OpenAI Agent service: {self._agent.name}") + + @property + def agent(self) -> Agent: + """Get the underlying OpenAI Agent. + + Returns: + The configured Agent instance. + """ + return self._agent + + def update_agent_config( + self, + *, + instructions: Optional[str] = None, + model_config: Optional[Dict[str, Any]] = None, + **kwargs, + ): + """Update agent configuration dynamically. + + Args: + instructions: New system instructions for the agent. + model_config: Updated model configuration. + **kwargs: Additional agent configuration parameters. + """ + if instructions: + self._agent.instructions = instructions + logger.info(f"Updated agent instructions for {self._agent.name}") + + if model_config: + self._agent.model_config = model_config + if "model" in model_config: + self.set_model_name(model_config["model"]) + logger.info(f"Updated model config for {self._agent.name}") + + async def start(self, frame: StartFrame): + """Start the OpenAI Agent service. + + Initializes the agent session and prepares for processing. + + Args: + frame: The start frame containing initialization parameters. + """ + logger.info(f"Starting OpenAI Agent service: {self._agent.name}") + await super().start(frame) + + async def stop(self, frame: EndFrame): + """Stop the OpenAI Agent service. + + Cleans up resources and ends the current session. + + Args: + frame: The end frame. + """ + logger.info(f"Stopping OpenAI Agent service: {self._agent.name}") + + # Cancel any ongoing processing + if self._processing_task and not self._processing_task.done(): + self._processing_task.cancel() + try: + await self._processing_task + except asyncio.CancelledError: + pass + + await super().stop(frame) + + async def cancel(self, frame: CancelFrame): + """Cancel the OpenAI Agent service. + + Cancels any ongoing operations. + + Args: + frame: The cancel frame. + """ + logger.info(f"Cancelling OpenAI Agent service: {self._agent.name}") + + # Cancel any ongoing processing + if self._processing_task and not self._processing_task.done(): + self._processing_task.cancel() + + await super().cancel(frame) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames and handle agent interactions. + + Processes text input frames by running them through the OpenAI Agent + and streams the results back as LLM frames. + + Args: + frame: The frame to process. + direction: The direction of frame processing. + """ + await super().process_frame(frame, direction) + + if isinstance(frame, TextFrame): + # Process text input through the agent + if self._processing_task and not self._processing_task.done(): + logger.warning("Already processing a request, cancelling previous task") + self._processing_task.cancel() + try: + await self._processing_task + except asyncio.CancelledError: + pass + + self._processing_task = asyncio.create_task(self._process_agent_request(frame.text)) + + async def _process_agent_request(self, input_text: str): + """Process an agent request and stream the results. + + Args: + input_text: The user input text to process. + """ + try: + logger.debug(f"Processing agent request: {input_text}") + + # Start the LLM response + await self.push_frame(LLMFullResponseStartFrame()) + + if self._streaming: + await self._process_streaming_response(input_text) + else: + await self._process_non_streaming_response(input_text) + + # End the LLM response + await self.push_frame(LLMFullResponseEndFrame()) + + except Exception as e: + logger.error(f"Error processing agent request: {e}") + await self.push_error(ErrorFrame(f"Agent processing error: {e}")) + + async def _process_streaming_response(self, input_text: str): + """Process a streaming agent response. + + Args: + input_text: The user input text to process. + """ + try: + # Run the agent with streaming + result: RunResultStreaming = Runner.run_streamed( + self._agent, input_text, context=self._session_config + ) + + # Process the stream events + 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)) + + elif event.type == "run_item_stream_event": + # Handle completed items + if event.item.type == "message_output_item": + # Get the complete message text + message_text = self._extract_message_text(event.item) + if message_text and message_text != self._accumulated_text: + # Send any new text that wasn't already streamed + new_text = message_text[len(self._accumulated_text) :] + if new_text: + await self.push_frame(LLMTextFrame(text=new_text)) + self._accumulated_text = message_text + + elif event.item.type == "tool_call_item": + logger.debug(f"Tool called: {event.item.tool_name}") + + elif event.item.type == "tool_call_output_item": + logger.debug(f"Tool output: {event.item.output}") + + elif event.type == "agent_updated_stream_event": + logger.debug(f"Agent updated: {event.new_agent.name}") + + # Reset accumulated text for next request + self._accumulated_text = "" + + except Exception as e: + logger.error(f"Error in streaming response: {e}") + raise + + async def _process_non_streaming_response(self, input_text: str): + """Process a non-streaming agent response. + + Args: + input_text: The user input text to process. + """ + try: + # Run the agent without streaming + result: RunResult = await Runner.run( + self._agent, input_text, context=self._session_config + ) + + # Send the final output + if result.final_output: + await self.push_frame(LLMTextFrame(text=result.final_output)) + + except Exception as e: + logger.error(f"Error in non-streaming response: {e}") + raise + + def _extract_message_text(self, item) -> str: + """Extract text from a message output item. + + Args: + item: The message output item from the agent. + + Returns: + The extracted text content. + """ + try: + # Handle different message item formats + if hasattr(item, "content"): + if isinstance(item.content, str): + return item.content + elif isinstance(item.content, list): + # Extract text from content array + text_parts = [] + for content_part in item.content: + if isinstance(content_part, dict) and content_part.get("type") == "text": + text_parts.append(content_part.get("text", "")) + elif isinstance(content_part, str): + text_parts.append(content_part) + return "".join(text_parts) + + # Fallback: try to get text through string conversion + return str(item) + + except Exception as e: + logger.warning(f"Could not extract text from message item: {e}") + return "" + + async def add_tool(self, tool_function: Callable): + """Add a tool function to the agent. + + Args: + tool_function: A callable function 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}") + + async def add_handoff_agent(self, agent: Agent): + """Add a handoff agent. + + Args: + agent: Another Agent instance 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}") + + def get_session_context(self) -> Dict[str, Any]: + """Get the current session context. + + Returns: + Dictionary containing the current session context. + """ + return self._session_config.copy() + + def update_session_context(self, context: Dict[str, Any]): + """Update the session context. + + Args: + context: Dictionary of context updates to apply. + """ + self._session_config.update(context) + logger.debug(f"Updated session context for agent {self._agent.name}") diff --git a/tests/test_openai_agent_service.py b/tests/test_openai_agent_service.py new file mode 100644 index 000000000..b12c146ac --- /dev/null +++ b/tests/test_openai_agent_service.py @@ -0,0 +1,286 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Tests for OpenAI Agent service.""" + +import asyncio +import os +import sys +import unittest.mock +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# Add src to path for testing +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, + StartFrame, + TextFrame, +) +from pipecat.processors.frame_processor import FrameDirection + + +class MockAgent: + """Mock Agent for testing.""" + + def __init__(self, name="Test Agent", instructions="Test instructions"): + self.name = name + self.instructions = instructions + self.tools = [] + self.handoffs = [] + + +class MockRunResult: + """Mock RunResult for testing.""" + + def __init__(self, final_output="Test response"): + self.final_output = final_output + + +class MockStreamEvent: + """Mock StreamEvent for testing.""" + + def __init__(self, event_type, data=None, item=None): + self.type = event_type + self.data = data + self.item = item + + +class MockMessageItem: + """Mock message item for testing.""" + + def __init__(self, content="Test content"): + self.type = "message_output_item" + self.content = content + + +class MockRunner: + """Mock Runner for testing.""" + + @staticmethod + async def run(agent, input_text, context=None): + return MockRunResult("Mocked response") + + @staticmethod + def run_streamed(agent, input_text, context=None): + class MockStreamResult: + async def stream_events(self): + yield MockStreamEvent("raw_response_event", data=MagicMock(delta="Test ")) + yield MockStreamEvent("raw_response_event", data=MagicMock(delta="response")) + yield MockStreamEvent( + "run_item_stream_event", item=MockMessageItem("Test response") + ) + + return MockStreamResult() + + +@pytest.fixture +def mock_openai_agents(): + """Mock the OpenAI Agents SDK imports.""" + with patch.dict( + "sys.modules", + { + "agents": MagicMock(), + "agents.stream_events": MagicMock(), + "agents.result": MagicMock(), + }, + ): + # Mock the classes and functions we need + mock_agent = MagicMock() + mock_agent.return_value = MockAgent() + + mock_runner = MagicMock() + mock_runner.run = AsyncMock(return_value=MockRunResult()) + mock_runner.run_streamed = MagicMock(return_value=MockRunner.run_streamed(None, None)) + + with ( + patch("pipecat.services.openai_agent.agent_service.Agent", mock_agent), + patch("pipecat.services.openai_agent.agent_service.Runner", mock_runner), + ): + yield { + "Agent": mock_agent, + "Runner": mock_runner, + } + + +@pytest.mark.asyncio +async def test_openai_agent_service_init(mock_openai_agents): + """Test OpenAI Agent service initialization.""" + from pipecat.services.openai_agent.agent_service import OpenAIAgentService + + service = OpenAIAgentService( + name="Test Agent", instructions="Test instructions", api_key="test-key", streaming=True + ) + + assert service.agent.name == "Test Agent" + assert service._streaming is True + + +@pytest.mark.asyncio +async def test_openai_agent_service_process_text_frame_streaming(mock_openai_agents): + """Test processing text frame with streaming enabled.""" + from pipecat.services.openai_agent.agent_service import OpenAIAgentService + + service = OpenAIAgentService( + name="Test Agent", instructions="Test instructions", api_key="test-key", streaming=True + ) + + # Mock the push_frame method to capture output + output_frames = [] + + async def mock_push_frame(frame, direction=FrameDirection.DOWNSTREAM): + output_frames.append(frame) + + service.push_frame = mock_push_frame + + # Process a text frame + text_frame = TextFrame("Hello, agent!") + await service.process_frame(text_frame, FrameDirection.DOWNSTREAM) + + # Wait a bit for async processing + await asyncio.sleep(0.1) + + # Check that appropriate frames were generated + assert len(output_frames) > 0 + assert any(isinstance(frame, LLMFullResponseStartFrame) for frame in output_frames) + + +@pytest.mark.asyncio +async def test_openai_agent_service_process_text_frame_non_streaming(mock_openai_agents): + """Test processing text frame with streaming disabled.""" + from pipecat.services.openai_agent.agent_service import OpenAIAgentService + + service = OpenAIAgentService( + name="Test Agent", instructions="Test instructions", api_key="test-key", streaming=False + ) + + # Mock the push_frame method to capture output + output_frames = [] + + async def mock_push_frame(frame, direction=FrameDirection.DOWNSTREAM): + output_frames.append(frame) + + service.push_frame = mock_push_frame + + # Process a text frame + text_frame = TextFrame("Hello, agent!") + await service.process_frame(text_frame, FrameDirection.DOWNSTREAM) + + # Wait a bit for async processing + await asyncio.sleep(0.1) + + # Check that appropriate frames were generated + assert len(output_frames) > 0 + + +@pytest.mark.asyncio +async def test_openai_agent_service_update_config(mock_openai_agents): + """Test updating agent configuration.""" + from pipecat.services.openai_agent.agent_service import OpenAIAgentService + + service = OpenAIAgentService( + name="Test Agent", instructions="Test instructions", api_key="test-key" + ) + + # Update configuration + service.update_agent_config( + instructions="Updated instructions", model_config={"model": "gpt-4o", "temperature": 0.7} + ) + + assert service.agent.instructions == "Updated instructions" + assert service.agent.model_config["model"] == "gpt-4o" + + +@pytest.mark.asyncio +async def test_openai_agent_service_session_context(mock_openai_agents): + """Test session context management.""" + from pipecat.services.openai_agent.agent_service import OpenAIAgentService + + service = OpenAIAgentService( + name="Test Agent", + instructions="Test instructions", + api_key="test-key", + session_config={"user_id": "test-user"}, + ) + + # Get initial context + context = service.get_session_context() + assert context["user_id"] == "test-user" + + # Update context + service.update_session_context({"session_id": "test-session"}) + + updated_context = service.get_session_context() + assert updated_context["user_id"] == "test-user" + assert updated_context["session_id"] == "test-session" + + +@pytest.mark.asyncio +async def test_openai_agent_service_add_tools(mock_openai_agents): + """Test adding tools to the agent.""" + from pipecat.services.openai_agent.agent_service import OpenAIAgentService + + service = OpenAIAgentService( + name="Test Agent", instructions="Test instructions", api_key="test-key" + ) + + # Define a test tool + def test_tool(): + return "test result" + + # Add the tool + await service.add_tool(test_tool) + + # Check if tool was added (this depends on the mock implementation) + assert hasattr(service.agent, "tools") + + +@pytest.mark.asyncio +async def test_openai_agent_service_lifecycle(mock_openai_agents): + """Test service lifecycle methods.""" + from pipecat.frames.frames import CancelFrame, EndFrame, StartFrame + from pipecat.services.openai_agent.agent_service import OpenAIAgentService + + service = OpenAIAgentService( + name="Test Agent", instructions="Test instructions", api_key="test-key" + ) + + # Test start + start_frame = StartFrame() + await service.start(start_frame) + + # Test cancel + cancel_frame = CancelFrame() + await service.cancel(cancel_frame) + + # Test stop + end_frame = EndFrame() + await service.stop(end_frame) + + +def test_openai_agent_service_import_error(): + """Test that import error is handled gracefully.""" + # Mock the import to fail + with patch.dict("sys.modules", {"agents": None}): + with pytest.raises(Exception) as exc_info: + # This should trigger the import error + import importlib + + import pipecat.services.openai_agent.agent_service + + importlib.reload(pipecat.services.openai_agent.agent_service) + + assert "Missing module" in str(exc_info.value) + + +if __name__ == "__main__": + pytest.main([__file__])