wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ok its kinda working

Browse Source
This commit is contained in:
James Hush
2025-09-17 11:29:11 +08:00
parent 0421d97954
commit 50a2a0dc86
1 changed files with 16 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
src/pipecat/services/openai_agent/agent_service.py
View File

@@ -161,7 +161,6 @@ class OpenAIAgentService(AIService):
self._session_config = session_config or {} self._session_config = session_config or {}
self._current_session = None self._current_session = None
self._accumulated_text = "" self._accumulated_text = ""
self._processing_task: Optional[asyncio.Task] = None
# Set model name for metrics # Set model name for metrics
if model_config and "model" in model_config: if model_config and "model" in model_config:
@@ -225,15 +224,6 @@ class OpenAIAgentService(AIService):
frame: The end frame. frame: The end frame.
""" """
logger.info(f"Stopping OpenAI Agent service: {self._agent.name}") 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) await super().stop(frame)
async def cancel(self, frame: CancelFrame): async def cancel(self, frame: CancelFrame):
@@ -245,11 +235,6 @@ class OpenAIAgentService(AIService):
frame: The cancel frame. frame: The cancel frame.
""" """
logger.info(f"Cancelling OpenAI Agent service: {self._agent.name}") 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) await super().cancel(frame)
@override @override
@@ -266,16 +251,17 @@ class OpenAIAgentService(AIService):
await super().process_frame(frame, direction) await super().process_frame(frame, direction)
if isinstance(frame, TextFrame): if isinstance(frame, TextFrame):
# Process text input through the agent # Process text input through the agent directly
if self._processing_task and not self._processing_task.done(): try:
logger.warning("Already processing a request, cancelling previous task") await self.push_frame(LLMFullResponseStartFrame())
self._processing_task.cancel() await self._process_agent_request(frame.text)
try: await self.push_frame(LLMFullResponseEndFrame())
await self._processing_task except Exception as e:
except asyncio.CancelledError: logger.error(f"Error processing agent request: {e}")
pass await self.push_error(ErrorFrame(f"Agent processing error: {e}"))
else:
self._processing_task = asyncio.create_task(self._process_agent_request(frame.text)) # For frames we don't handle, pass them through with direction
await self.push_frame(frame, direction)
async def _process_agent_request(self, input_text: str): async def _process_agent_request(self, input_text: str):
"""Process an agent request and stream the results. """Process an agent request and stream the results.
@@ -283,23 +269,12 @@ class OpenAIAgentService(AIService):
Args: Args:
input_text: The user input text to process. input_text: The user input text to process.
""" """
try: logger.debug(f"Processing agent request: {input_text}")
logger.debug(f"Processing agent request: {input_text}")
# Start the LLM response if self._streaming:
await self.push_frame(LLMFullResponseStartFrame()) await self._process_streaming_response(input_text)
else:
if self._streaming: await self._process_non_streaming_response(input_text)
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): async def _process_streaming_response(self, input_text: str):
"""Process a streaming agent response. """Process a streaming agent response.