From b17165b7ea7a8af6eefe9180439fdaed6d8edf9b Mon Sep 17 00:00:00 2001 From: James Hush Date: Thu, 8 May 2025 14:33:59 +0800 Subject: [PATCH] Save progress --- examples/foundational/99-open-ai-agent.py | 133 +++++++++++++--------- examples/foundational/agent.py | 122 ++++++++++++++++++++ 2 files changed, 204 insertions(+), 51 deletions(-) create mode 100644 examples/foundational/agent.py diff --git a/examples/foundational/99-open-ai-agent.py b/examples/foundational/99-open-ai-agent.py index a0f12a6e9..b4b481016 100644 --- a/examples/foundational/99-open-ai-agent.py +++ b/examples/foundational/99-open-ai-agent.py @@ -4,15 +4,18 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import argparse import os from abc import ABC, abstractmethod from dataclasses import field -from typing import Literal, Optional +from typing import List, Literal, Optional import httpx +from agents import Agent, Runner from dotenv import load_dotenv from loguru import logger -from openai import BaseModel +from openai import AsyncStream, BaseModel +from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer @@ -70,10 +73,27 @@ class BackendBase(ABC): raise NotImplementedError("The method get_resp is not implemented.") -class CompassLLMService(AIService): - def __init__(self, backend: BackendBase): - super().__init__() - self.backend = backend +class ChoiceDelta(BaseModel): + content: Optional[str] = None + """The contents of the chunk message.""" + + +class Choice(BaseModel): + delta: ChoiceDelta + """The contents of the chunk message.""" + + index: int + """The index of the choice in the list of choices.""" + + +class CustomLLMService(BaseOpenAILLMService): + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._client = Agent( + name="Assistant agent", + instructions="Respond with haikus.", + # tools=[get_weather], + ) def create_context_aggregator( self, @@ -82,7 +102,9 @@ class CompassLLMService(AIService): user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), ) -> OpenAIContextAggregatorPair: - """Create an instance of OpenAIContextAggregatorPair from an + """Create an instance of OpenAIContextAggregatorPair. + + from an OpenAILLMContext. Constructor keyword arguments for both the user and assistant aggregators can be provided. @@ -102,52 +124,63 @@ class CompassLLMService(AIService): 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) + def create_client( + self, + api_key=None, + base_url=None, + organization=None, + project=None, + default_headers=None, + **kwargs, + ): + return Agent( + name="Assistant agent", + instructions="Respond with haikus.", + # tools=[get_weather], + ) - 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) + async def get_chat_completions( + self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] + ) -> AsyncStream[ChatCompletionChunk]: + # self._client.tools = context.tools + logger.info(f"get_chat_completions: {self._client}") - if context: - try: - await self.push_frame(LLMFullResponseStartFrame()) - await self.start_processing_metrics() - # await self._process_context(context) + result = Runner.run_streamed( + # context=context, + starting_agent=self._client, + input="Tell a joke about pirates.", # messages + # --- + # no func tool + # input="give me a 2 sentences about life", + ) - msgs = [] - for contmsg in context.messages: - msgs.append( - LlmMessage( - role=contmsg["role"], - content=contmsg["content"], - ) + logger.info(f"get_chat_completions: {result}") + + if result is None: + logger.error("Runner.run_streamed returned None") + return + + async for event in result.stream_events(): + # if not event.type == "raw_response_event": + # break + if event.type == "raw_response_event": + if event.data.type == "response.output_text.delta": + delta = ChoiceDelta(content=event.data.delta) + choice = Choice(delta=delta, index=event.data.output_index) + + converted_message = ChatCompletionChunk( + id=event.data.item_id, + choices=[choice], ) - resp = await self.backend.get_resp( - msgs, - { - "conversation_id": "fake_conversation_id", - "user_id": "fake_user_id", - }, - ) + return converted_message + else: + break - 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()) + # chunks = await self._client.chat.completions.create(**params) + # return chunks -async def run_bot(webrtc_connection: SmallWebRTCConnection): +async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace): logger.info(f"Starting bot") transport = SmallWebRTCTransport( @@ -155,9 +188,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): params=TransportParams( audio_in_enabled=True, audio_out_enabled=True, - vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), - vad_audio_passthrough=True, ), ) @@ -168,7 +199,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + llm = CustomLLMService(model="gpt-4.1", api_key=os.getenv("OPENAI_API_KEY")) messages = [ { @@ -206,8 +237,8 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): async def on_client_connected(transport, client): logger.info(f"Client connected") # Kick off the conversation. - messages.append({"role": "system", "content": "Please introduce yourself to the user."}) - await task.queue_frames([context_aggregator.user().get_context_frame()]) + # messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + # await task.queue_frames([context_aggregator.user().get_context_frame()]) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): diff --git a/examples/foundational/agent.py b/examples/foundational/agent.py new file mode 100644 index 000000000..a0ee738de --- /dev/null +++ b/examples/foundational/agent.py @@ -0,0 +1,122 @@ +import asyncio +import logging +import os +from datetime import datetime + +from agents import ( + Agent, + FunctionTool, + HandoffOutputItem, + ItemHelpers, + MessageOutputItem, + RunContextWrapper, + Runner, + ToolCallItem, + ToolCallOutputItem, + function_tool, + set_default_openai_api, + set_default_openai_client, + set_tracing_disabled, + trace, +) +from httpx import get + + +@function_tool +async def get_weather(location: str) -> str: + """Fetch the weather for today. + + Args: + location: The location to fetch the weather for. + """ + return f"{location} is sunny" + + +system_prompt = """ + you are a helpful assistant for a real estate brokerage AI assistant. +""" + + +bot = Agent( + name="Assistant agent", + instructions=system_prompt, + # tools=[get_weather], +) + + +async def main(): + # res = await Runner.run( + # starting_agent=bot, + # input="What is the weather today?", + # ) + # print(res) + + result = Runner.run_streamed( + starting_agent=bot, + # --- + # with func tool + input="Tell a joke about pirates.", + # --- + # no func tool + # input="give me a 2 sentences about life", + ) + + final = [] + async for event in result.stream_events(): + # We'll ignore the raw responses event deltas + name = getattr(event, "name", None) + # print(f"Event: {event.type} - name {name}") + # print(event) + # continue + if event.type == "raw_response_event": + if event.data.type == "response.output_text.delta": + final += event.data.delta + + print(f"raw resp: {event}") + # When the agent updates, print that + elif event.type == "agent_updated_stream_event": + print(f"Agent updated: {event.new_agent.name}") + continue + # When items are generated, print them + elif event.type == "run_item_stream_event": + if event.item.type == "tool_call_item": + print("-- Tool was called") + elif event.item.type == "tool_call_output_item": + print(f"-- Tool output: {event.item.output}") + elif event.item.type == "message_output_item": + print(f"-- Message output:\n {ItemHelpers.text_message_output(event.item)}") + else: + print(f"-- Unknown item type: {event.item.type}") + pass # Ignore other event types + else: + print(f"-- Unknown out item type: {event.item.type}") + + print(f"----------------------") + + print(f"FinalFinalFinal: {''.join(final)}") + + +if __name__ == "__main__": + asyncio.run(main()) + + +# no func tool: +# +# Event: agent_updated_stream_event - name None +# Event: raw_response_event - name None +# ... +# Event: raw_response_event - name None +# Event: run_item_stream_event - name message_output_created + +# with func tool: +# +# Event: agent_updated_stream_event - name None +# Event: raw_response_event - name None +# ... +# Event: raw_response_event - name None +# Event: run_item_stream_event - name tool_called +# Event: run_item_stream_event - name tool_output +# Event: raw_response_event - name None +# ... +# Event: raw_response_event - name None +# Event: run_item_stream_event - name message_output_created