Workflow

VoiceWorkflowBase

基础: ABC

一个语音工作流的基础类。您必须实现 run 方法。 “工作流”是指您希望执行的任何代码，它接收转录文本并产生将被文本转语音模型转换为语音的文本。在大多数情况下，您将创建 Agent 并使用 Runner.run_streamed() 运行它们，返回流中的一些或所有文本事件。您可以使用 VoiceWorkflowHelper 类来帮助从流中提取文本事件。如果您有一个简单的工作流，它只有一个起始代理且没有自定义逻辑，则可以直接使用 SingleAgentVoiceWorkflow。

源代码位于 src/agents/voice/workflow.py

class VoiceWorkflowBase(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.abstractmethod
    def run(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.
        """
        pass

    async def on_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.
        """
        return
        yield

run abstractmethod

run(transcription: str) -> AsyncIterator[str]

运行语音工作流。您将收到一个输入转录文本，并且必须产生将被告知用户的文本。您可以在这里运行任何逻辑。在大多数情况下，最终逻辑将涉及调用 Runner.run_streamed() 并产生流中的任何文本事件。

源代码位于 src/agents/voice/workflow.py

@abc.abstractmethod
def run(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.
    """
    pass

on_start async

on_start() -> AsyncIterator[str]

可选方法，在接收到任何用户输入之前运行。可用于通过TTS传递问候语或指令。默认情况下不执行任何操作。

源代码位于 src/agents/voice/workflow.py

async def on_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.
    """
    return
    yield

VoiceWorkflowHelper

源代码位于 src/agents/voice/workflow.py

class VoiceWorkflowHelper:
    @classmethod
    async def stream_text_from(cls, result: RunResultStreaming) -> AsyncIterator[str]:
        """Wraps a `RunResultStreaming` object and yields text events from the stream."""
        async for event in result.stream_events():
            if (
                event.type == "raw_response_event"
                and event.data.type == "response.output_text.delta"
            ):
                yield event.data.delta

stream_text_from async classmethod

stream_text_from(
    result: RunResultStreaming,
) -> AsyncIterator[str]

封装一个 RunResultStreaming 对象并产生流中的文本事件。

源代码位于 src/agents/voice/workflow.py

@classmethod
async def stream_text_from(cls, result: RunResultStreaming) -> AsyncIterator[str]:
    """Wraps a `RunResultStreaming` object and yields text events from the stream."""
    async for event in result.stream_events():
        if (
            event.type == "raw_response_event"
            and event.data.type == "response.output_text.delta"
        ):
            yield event.data.delta

SingleAgentWorkflowCallbacks

源代码位于 src/agents/voice/workflow.py

class SingleAgentWorkflowCallbacks:
    def on_run(self, workflow: SingleAgentVoiceWorkflow, transcription: str) -> None:
        """Called when the workflow is run."""
        pass

on_run

on_run(
    workflow: SingleAgentVoiceWorkflow, transcription: str
) -> None

在运行工作流时调用。

源代码位于 src/agents/voice/workflow.py

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

SingleAgentVoiceWorkflow

继承自：VoiceWorkflowBase

一个简单的语音工作流，它运行单个代理。每个转录和结果都会添加到输入历史记录中。对于更复杂的工作流（例如，多次调用 Runner、自定义消息历史记录、自定义逻辑、自定义配置），请子类化 VoiceWorkflowBase 并实现您自己的逻辑。

源代码位于 src/agents/voice/workflow.py

class SingleAgentVoiceWorkflow(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 = agent
        self._callbacks = callbacks

    async def run(self, transcription: str) -> AsyncIterator[str]:
        if self._callbacks:
            self._callbacks.on_run(self, transcription)

        # Add the transcription to the input history
        self._input_history.append(
            {
                "role": "user",
                "content": transcription,
            }
        )

        # Run the agent
        result = Runner.run_streamed(self._current_agent, self._input_history)

        # Stream the text from the result
        async for chunk in VoiceWorkflowHelper.stream_text_from(result):
            yield chunk

        # Update the input history and current agent
        self._input_history = result.to_input_list()
        self._current_agent = result.last_agent

__init__

__init__(
    agent: Agent[Any],
    callbacks: SingleAgentWorkflowCallbacks | None = None,
)

创建一个新的单代理语音工作流。

参数

名称	类型	描述	默认
agent	Agent[Any]	要运行的代理。	required
callbacks	SingleAgentWorkflowCallbacks | None	在工作流期间调用的可选回调函数。	None

源代码位于 src/agents/voice/workflow.py

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 = agent
    self._callbacks = callbacks

on_start async

on_start() -> AsyncIterator[str]

可选方法，在接收到任何用户输入之前运行。可用于通过TTS传递问候语或指令。默认情况下不执行任何操作。

源代码位于 src/agents/voice/workflow.py

async def on_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.
    """
    return
    yield