From 19a4b975040db3f37f5a0dcf3e215616c668929c Mon Sep 17 00:00:00 2001 From: James Hush Date: Thu, 24 Apr 2025 13:29:01 +0800 Subject: [PATCH] Example Add create context aggregator --- examples/foundational/99-open-ai-agent.py | 143 +++++++++++++++++++--- 1 file changed, 125 insertions(+), 18 deletions(-) diff --git a/examples/foundational/99-open-ai-agent.py b/examples/foundational/99-open-ai-agent.py index b480b4ff7..a0f12a6e9 100644 --- a/examples/foundational/99-open-ai-agent.py +++ b/examples/foundational/99-open-ai-agent.py @@ -5,20 +5,47 @@ # import os +from abc import ABC, abstractmethod +from dataclasses import field +from typing import Literal, Optional -from agents import Agent +import httpx from dotenv import load_dotenv from loguru import logger +from openai import BaseModel from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import ( + Frame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMMessagesFrame, + LLMTextFrame, + LLMUpdateSettingsFrame, +) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, + LLMUserAggregatorParams, +) +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, + OpenAILLMContextFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.services.ai_service import AIService from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.openai.llm import OpenAILLMService +from pipecat.services.openai.base_llm import BaseOpenAILLMService +from pipecat.services.openai.llm import ( + OpenAIAssistantContextAggregator, + OpenAIContextAggregatorPair, + OpenAILLMService, + OpenAIUserContextAggregator, +) from pipecat.transports.base_transport import TransportParams from pipecat.transports.network.small_webrtc import SmallWebRTCTransport from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection @@ -26,6 +53,100 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection load_dotenv(override=True) +class LlmMessage(BaseModel): + # ... + role: Literal["system", "user", "assistant", "tool"] + content: Optional[str] + + +class AgentResponse(BaseModel): + content: str + msgs: list[LlmMessage] = field(default_factory=list) + + +class BackendBase(ABC): + @abstractmethod + async def get_resp(self, messages: list[LlmMessage], extra_params) -> AgentResponse: + raise NotImplementedError("The method get_resp is not implemented.") + + +class CompassLLMService(AIService): + def __init__(self, backend: BackendBase): + super().__init__() + self.backend = backend + + def create_context_aggregator( + self, + context: OpenAILLMContext, + *, + user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), + ) -> OpenAIContextAggregatorPair: + """Create an instance of OpenAIContextAggregatorPair from an + OpenAILLMContext. Constructor keyword arguments for both the user and + assistant aggregators can be provided. + + Args: + context (OpenAILLMContext): The LLM context. + user_params (LLMUserAggregatorParams, optional): User aggregator parameters. + assistant_params (LLMAssistantAggregatorParams, optional): User aggregator parameters. + + Returns: + OpenAIContextAggregatorPair: A pair of context aggregators, one for + the user and one for the assistant, encapsulated in an + OpenAIContextAggregatorPair. + + """ + context.set_llm_adapter(self.get_llm_adapter()) + user = OpenAIUserContextAggregator(context, params=user_params) + assistant = OpenAIAssistantContextAggregator(context, params=assistant_params) + return OpenAIContextAggregatorPair(_user=user, _assistant=assistant) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + context = None + if isinstance(frame, OpenAILLMContextFrame): + context: OpenAILLMContext = frame.context + elif isinstance(frame, LLMMessagesFrame): + context = OpenAILLMContext.from_messages(frame.messages) + elif isinstance(frame, LLMUpdateSettingsFrame): + await self._update_settings(frame.settings) + else: + await self.push_frame(frame, direction) + + if context: + try: + await self.push_frame(LLMFullResponseStartFrame()) + await self.start_processing_metrics() + # await self._process_context(context) + + msgs = [] + for contmsg in context.messages: + msgs.append( + LlmMessage( + role=contmsg["role"], + content=contmsg["content"], + ) + ) + resp = await self.backend.get_resp( + msgs, + { + "conversation_id": "fake_conversation_id", + "user_id": "fake_user_id", + }, + ) + + context.add_messages(resp.msgs) + await self.push_frame(LLMTextFrame(resp.content)) + + except httpx.TimeoutException: + await self._call_event_handler("on_completion_timeout") + finally: + await self.stop_processing_metrics() + await self.push_frame(LLMFullResponseEndFrame()) + + async def run_bot(webrtc_connection: SmallWebRTCConnection): logger.info(f"Starting bot") @@ -55,22 +176,8 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): "content": "You are a helpful LLM in a WebRTC call. 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.", }, ] - agent = Agent( - name="Math Tutor", - instructions="You provide help with math problems. Explain your reasoning at each step and include examples", - # output_type="probably something here?", - # handoffs=[] - ) - # You can also do this to pass a context... - # agent = Agent[OpenAILLMContext] - # Not sure how useful this is, but it seems like a good idea to have - math_tool = agent.as_tool( - tool_name="math_tool", - tool_description="Call this tool whenever someone asks for help with math problems. You will be able to ask follow up questions to clarify the problem.", - ) - tools = ToolsSchema(standard_tools=[math_tool]) - context = OpenAILLMContext(messages=messages, tools=tools) + context = OpenAILLMContext(messages=messages) context_aggregator = llm.create_context_aggregator(context) pipeline = Pipeline(