This is working

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

@@ -17,17 +17,19 @@ Requirements:
 
 import os
 import random
-from typing import Any
+from typing import Any, List
 
 # Import agents SDK for tools and agent creation
 from agents import Agent, function_tool
 from dotenv import load_dotenv
 from loguru import logger
+from openai.types.chat import ChatCompletionMessageParam
 
-from pipecat.frames.frames import EndFrame, TextFrame
+from pipecat.frames.frames import LLMRunFrame, TextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
@@ -145,14 +147,27 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         streaming=True,
     )
 
-    # Create the processing pipeline
+    # Set up conversation context with initial system message
+    messages: List[ChatCompletionMessageParam] = [
+        {
+            "role": "system",
+            "content": "You are a helpful assistant with access to weather information and random facts. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = agent_service.create_context_aggregator(context)
+
+    # Create the processing pipeline with context aggregators
     pipeline = Pipeline(
         [
-            transport.input(),  # Receive audio input
+            transport.input(),  # Transport user input
-            stt,  # Convert speech to text
+            stt,  # Speech to text
-            agent_service,  # Process with OpenAI Agent
+            context_aggregator.user(),  # User responses
-            tts,  # Convert text to speech
+            agent_service,  # OpenAI Agent processing
-            transport.output(),  # Send audio output
+            tts,  # Text to speech
+            transport.output(),  # Transport bot output
+            context_aggregator.assistant(),  # Assistant spoken responses
         ]
     )
 
@@ -165,17 +180,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     @transport.event_handler("on_client_connected")
     async def on_client_connected(transport, client):
         logger.info("Client connected, sending greeting")
-        await task.queue_frames(
+        # Kick off the conversation by adding system message and running LLM
-            [
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-                TextFrame(
+        await task.queue_frames([LLMRunFrame()])
-                    "Hello! I'm an AI assistant powered by the OpenAI Agents SDK. "
+
-                    "I can help you with weather information, share interesting facts, "
+    @transport.event_handler("on_client_disconnected")
-                    "or just have a conversation. What would you like to know?"
+    async def on_client_disconnected(transport, client):
-                ),
+        logger.info("Client disconnected")
-                # Don't send EndFrame() here - that closes the pipeline!
+        await task.cancel()
-                # The conversation should continue after the greeting
-            ]
-        )
 
     runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
     await runner.run(task)

src/pipecat/services/openai_agent/agent_service.py

@@ -13,6 +13,7 @@ guardrails, sessions, and tools from the OpenAI Agents SDK.
 
 import asyncio
 import os
+from dataclasses import dataclass
 from typing import (
     Any,
     Awaitable,
@@ -53,6 +54,16 @@ from pipecat.frames.frames import (
     TextFrame,
     UserImageRawFrame,
 )
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantAggregatorParams,
+    LLMAssistantContextAggregator,
+    LLMUserAggregatorParams,
+    LLMUserContextAggregator,
+)
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
+)
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_service import AIService
 
@@ -77,6 +88,35 @@ class AgentLike(Protocol):
         ...
 
 
+@dataclass
+class OpenAIAgentContextAggregatorPair:
+    """Pair of OpenAI Agent context aggregators for user and assistant messages.
+
+    Parameters:
+        _user: User context aggregator for processing user messages.
+        _assistant: Assistant context aggregator for processing assistant messages.
+    """
+
+    _user: "OpenAIAgentUserContextAggregator"
+    _assistant: "OpenAIAgentAssistantContextAggregator"
+
+    def user(self) -> "OpenAIAgentUserContextAggregator":
+        """Get the user context aggregator.
+
+        Returns:
+            The user context aggregator instance.
+        """
+        return self._user
+
+    def assistant(self) -> "OpenAIAgentAssistantContextAggregator":
+        """Get the assistant context aggregator.
+
+        Returns:
+            The assistant context aggregator instance.
+        """
+        return self._assistant
+
+
 class OpenAIAgentService(AIService):
     """OpenAI Agents SDK service for Pipecat.
 
@@ -179,6 +219,32 @@ class OpenAIAgentService(AIService):
         """
         return self._agent
 
+    def create_context_aggregator(
+        self,
+        context: OpenAILLMContext,
+        *,
+        user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
+        assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
+    ) -> OpenAIAgentContextAggregatorPair:
+        """Create OpenAI-specific context aggregators for agent interactions.
+
+        Creates a pair of context aggregators optimized for OpenAI Agent interactions,
+        including support for function calls, tool usage, and conversation management.
+
+        Args:
+            context: The LLM context to create aggregators for.
+            user_params: Parameters for user message aggregation.
+            assistant_params: Parameters for assistant message aggregation.
+
+        Returns:
+            OpenAIAgentContextAggregatorPair: A pair of context aggregators, one for
+            the user and one for the assistant, encapsulated in an
+            OpenAIAgentContextAggregatorPair.
+        """
+        user = OpenAIAgentUserContextAggregator(context, params=user_params)
+        assistant = OpenAIAgentAssistantContextAggregator(context, params=assistant_params)
+        return OpenAIAgentContextAggregatorPair(_user=user, _assistant=assistant)
+
     def update_agent_config(
         self,
         *,
@@ -241,7 +307,7 @@ class OpenAIAgentService(AIService):
     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
+        Processes OpenAILLMContextFrame and TextFrame by running them through the OpenAI Agent
         and streams the results back as LLM frames.
 
         Args:
@@ -250,8 +316,36 @@ class OpenAIAgentService(AIService):
         """
         await super().process_frame(frame, direction)
 
-        if isinstance(frame, TextFrame):
+        if isinstance(frame, OpenAILLMContextFrame):
-            # Process text input through the agent directly
+            # Process context frame through the agent
+            try:
+                await self.push_frame(LLMFullResponseStartFrame())
+                # Extract the latest user message from the context
+                messages = frame.context.get_messages()
+                if messages:
+                    # Get the last user message
+                    for message in reversed(messages):
+                        if message.get("role") == "user":
+                            content = message.get("content", "")
+                            if isinstance(content, list):
+                                # Extract text from content array
+                                text_parts = []
+                                for part in content:
+                                    if isinstance(part, dict) and part.get("type") == "text":
+                                        text_parts.append(part.get("text", ""))
+                                user_input = " ".join(text_parts)
+                            else:
+                                user_input = str(content)
+
+                            if user_input.strip():
+                                await self._process_agent_request(user_input)
+                                break
+                await self.push_frame(LLMFullResponseEndFrame())
+            except Exception as e:
+                logger.error(f"Error processing agent context: {e}")
+                await self.push_error(ErrorFrame(f"Agent processing error: {e}"))
+        elif isinstance(frame, TextFrame):
+            # Process text input through the agent directly (for backwards compatibility)
             try:
                 await self.push_frame(LLMFullResponseStartFrame())
                 await self._process_agent_request(frame.text)
@@ -450,3 +544,24 @@ class OpenAIAgentService(AIService):
         """
         self._session_config.update(context)
         logger.debug(f"Updated session context for agent {self._agent.name}")
+
+
+class OpenAIAgentUserContextAggregator(LLMUserContextAggregator):
+    """OpenAI Agent-specific user context aggregator.
+
+    Handles aggregation of user messages for OpenAI Agent services.
+    Inherits all functionality from the base LLMUserContextAggregator.
+    """
+
+    pass
+
+
+class OpenAIAgentAssistantContextAggregator(LLMAssistantContextAggregator):
+    """OpenAI Agent-specific assistant context aggregator.
+
+    Handles aggregation of assistant messages for OpenAI Agent services,
+    with specialized support for OpenAI's function calling format,
+    tool usage tracking, and agent interaction management.
+    """
+
+    pass