1314151617181920212223242526272829303132333435363738394041

classVoiceWorkflowBase(abc.ABC):"""    A base class for a voice workflow. You must implement the `run` method. A "workflow" is any    code you want, that receives a transcription and yields text that will be turned into speech    by a text-to-speech model.    In most cases, you'll create `Agent`s and use `Runner.run_streamed()` to run them, returning    some or all of the text events from the stream. You can use the `VoiceWorkflowHelper` class to    help with extracting text events from the stream.    If you have a simple workflow that has a single starting agent and no custom logic, you can    use `SingleAgentVoiceWorkflow` directly.    """@abc.abstractmethoddefrun(self,transcription:str)->AsyncIterator[str]:"""        Run the voice workflow. You will receive an input transcription, and must yield text that        will be spoken to the user. You can run whatever logic you want here. In most cases, the        final logic will involve calling `Runner.run_streamed()` and yielding any text events from        the stream.        """passasyncdefon_start(self)->AsyncIterator[str]:"""        Optional method that runs before any user input is received. Can be used        to deliver a greeting or instruction via TTS. Defaults to doing nothing.        """returnyield

44454647484950515253

classVoiceWorkflowHelper:@classmethodasyncdefstream_text_from(cls,result:RunResultStreaming)->AsyncIterator[str]:"""Wraps a `RunResultStreaming` object and yields text events from the stream."""asyncforeventinresult.stream_events():if(event.type=="raw_response_event"andevent.data.type=="response.output_text.delta"):yieldevent.data.delta

56575859

classSingleAgentWorkflowCallbacks:defon_run(self,workflow:SingleAgentVoiceWorkflow,transcription:str)->None:"""Called when the workflow is run."""pass

 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99100101

classSingleAgentVoiceWorkflow(VoiceWorkflowBase):"""A simple voice workflow that runs a single agent. Each transcription and result is added to    the input history.    For more complex workflows (e.g. multiple Runner calls, custom message history, custom logic,    custom configs), subclass `VoiceWorkflowBase` and implement your own logic.    """def__init__(self,agent:Agent[Any],callbacks:SingleAgentWorkflowCallbacks|None=None):"""Create a new single agent voice workflow.        Args:            agent: The agent to run.            callbacks: Optional callbacks to call during the workflow.        """self._input_history:list[TResponseInputItem]=[]self._current_agent=agentself._callbacks=callbacksasyncdefrun(self,transcription:str)->AsyncIterator[str]:ifself._callbacks:self._callbacks.on_run(self,transcription)# Add the transcription to the input historyself._input_history.append({"role":"user","content":transcription,})# Run the agentresult=Runner.run_streamed(self._current_agent,self._input_history)# Stream the text from the resultasyncforchunkinVoiceWorkflowHelper.stream_text_from(result):yieldchunk# Update the input history and current agentself._input_history=result.to_input_list()self._current_agent=result.last_agent