跳过内容

RealtimeAgent

基础类: AgentBase, Generic[TContext]

一个专门的代理实例，旨在在 RealtimeSession 中使用以构建语音代理。由于此代理的性质，不支持常规 Agent 实例支持的一些配置选项。例如：- 不支持 model 选择，因为所有 RealtimeAgent 都将由 RealtimeSession 内的相同模型处理。- 不支持 modelSettings，因为所有 RealtimeAgent 都将由 RealtimeSession 内的相同模型处理。- 不支持 outputType，因为 RealtimeAgent 不支持结构化输出。- 不支持 toolUseBehavior，因为所有 RealtimeAgent 都将由 RealtimeSession 内的相同模型处理。- 可以在 Agent 级别配置 voice；但是，在 RealtimeSession 中的第一个代理说完后，无法更改它。

有关与 Agents 共享的基本参数，请参阅 AgentBase。

源代码在 src/agents/realtime/agent.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 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
 99
100
101
102
@dataclass
class RealtimeAgent(AgentBase, Generic[TContext]):
    """A specialized agent instance that is meant to be used within a `RealtimeSession` to build
    voice agents. Due to the nature of this agent, some configuration options are not supported
    that are supported by regular `Agent` instances. For example:
    - `model` choice is not supported, as all RealtimeAgents will be handled by the same model
      within a `RealtimeSession`.
    - `modelSettings` is not supported, as all RealtimeAgents will be handled by the same model
      within a `RealtimeSession`.
    - `outputType` is not supported, as RealtimeAgents do not support structured outputs.
    - `toolUseBehavior` is not supported, as all RealtimeAgents will be handled by the same model
      within a `RealtimeSession`.
    - `voice` can be configured on an `Agent` level; however, it cannot be changed after the first
      agent within a `RealtimeSession` has spoken.

    See `AgentBase` for base parameters that are shared with `Agent`s.
    """

    instructions: (
        str
        | Callable[
            [RunContextWrapper[TContext], RealtimeAgent[TContext]],
            MaybeAwaitable[str],
        ]
        | None
    ) = None
    """The instructions for the agent. Will be used as the "system prompt" when this agent is
    invoked. Describes what the agent should do, and how it responds.

    Can either be a string, or a function that dynamically generates instructions for the agent. If
    you provide a function, it will be called with the context and the agent instance. It must
    return a string.
    """

    prompt: Prompt | None = None
    """A prompt object. Prompts allow you to dynamically configure the instructions, tools
    and other config for an agent outside of your code. Only usable with OpenAI models.
    """

    handoffs: list[RealtimeAgent[Any] | Handoff[TContext, RealtimeAgent[Any]]] = field(
        default_factory=list
    )
    """Handoffs are sub-agents that the agent can delegate to. You can provide a list of handoffs,
    and the agent can choose to delegate to them if relevant. Allows for separation of concerns and
    modularity.
    """

    output_guardrails: list[OutputGuardrail[TContext]] = field(default_factory=list)
    """A list of checks that run on the final output of the agent, after generating a response.
    Runs only if the agent produces a final output.
    """

    hooks: RealtimeAgentHooks | None = None
    """A class that receives callbacks on various lifecycle events for this agent.
    """

    def clone(self, **kwargs: Any) -> RealtimeAgent[TContext]:
        """Make a copy of the agent, with the given arguments changed. For example, you could do:
        ```
        new_agent = agent.clone(instructions="New instructions")
        ```
        """
        return dataclasses.replace(self, **kwargs)

    async def get_system_prompt(self, run_context: RunContextWrapper[TContext]) -> str | None:
        """Get the system prompt for the agent."""
        if isinstance(self.instructions, str):
            return self.instructions
        elif callable(self.instructions):
            if inspect.iscoroutinefunction(self.instructions):
                return await cast(Awaitable[str], self.instructions(run_context, self))
            else:
                return cast(str, self.instructions(run_context, self))
        elif self.instructions is not None:
            logger.error(f"Instructions must be a string or a function, got {self.instructions}")

        return None

instructions 类属性 实例属性 

instructions: (
    str
    | Callable[
        [
            RunContextWrapper[TContext],
            RealtimeAgent[TContext],
        ],
        MaybeAwaitable[str],
    ]
    | None
) = None

代理的指令。在调用此代理时，将用作“系统提示”。描述了代理应该做什么以及如何响应。

可以是字符串，也可以是动态生成代理指令的函数。如果您提供一个函数，它将使用上下文和代理实例调用。它必须返回一个字符串。

prompt 类属性 实例属性 

prompt: Prompt | None = None

一个提示对象。提示允许您在代码之外动态配置代理的指令、工具和其他配置。仅适用于 OpenAI 模型。

handoffs 类属性 实例属性 

handoffs: list[
    RealtimeAgent[Any]
    | Handoff[TContext, RealtimeAgent[Any]]
] = field(default_factory=list)

代理可以委托的子代理。您可以提供一个交接列表，并且代理可以选择委托给它们（如果相关）。允许关注点分离和模块化。

output_guardrails 类属性 实例属性 

output_guardrails: list[OutputGuardrail[TContext]] = field(
    default_factory=list
)

在代理生成响应后运行在代理的最终输出上的检查列表。仅当代理生成最终输出时才运行。

hooks 类属性 实例属性 

hooks: RealtimeAgentHooks | None = None

一个类，用于接收此代理的各种生命周期事件的回调。

name instance-attribute 

name: str

代理的名称。

handoff_description 类属性 实例属性 

handoff_description: str | None = None

代理的描述。当代理用作交接时使用，以便 LLM 知道它做什么以及何时调用它。

tools 类属性 实例属性 

tools: list[Tool] = field(default_factory=list)

代理可以使用的工具列表。

mcp_servers 类属性 实例属性 

mcp_servers: list[MCPServer] = field(default_factory=list)

代理可以使用的 模型上下文协议 服务器列表。每次代理运行时，它都会将来自这些服务器的工具包含在可用工具列表中。

注意：您需要管理这些服务器的生命周期。具体来说，您必须在将其传递给代理之前调用 server.connect()，并在不再需要服务器时调用 server.cleanup()。

mcp_config 类属性 实例属性 

mcp_config: MCPConfig = field(
    default_factory=lambda: MCPConfig()
)

MCP 服务器的配置。

clone 

clone(**kwargs: Any) -> RealtimeAgent[TContext]

复制代理，并更改给定的参数。例如，您可以

new_agent = agent.clone(instructions="New instructions")

源代码在 src/agents/realtime/agent.py 
82
83
84
85
86
87
88
def clone(self, **kwargs: Any) -> RealtimeAgent[TContext]:
    """Make a copy of the agent, with the given arguments changed. For example, you could do:
    ```
    new_agent = agent.clone(instructions="New instructions")
    ```
    """
    return dataclasses.replace(self, **kwargs)

get_system_prompt 异步 

get_system_prompt(
    run_context: RunContextWrapper[TContext],
) -> str | None

获取代理的系统提示。

源代码在 src/agents/realtime/agent.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
async def get_system_prompt(self, run_context: RunContextWrapper[TContext]) -> str | None:
    """Get the system prompt for the agent."""
    if isinstance(self.instructions, str):
        return self.instructions
    elif callable(self.instructions):
        if inspect.iscoroutinefunction(self.instructions):
            return await cast(Awaitable[str], self.instructions(run_context, self))
        else:
            return cast(str, self.instructions(run_context, self))
    elif self.instructions is not None:
        logger.error(f"Instructions must be a string or a function, got {self.instructions}")

    return None

get_mcp_tools 异步 

get_mcp_tools(
    run_context: RunContextWrapper[TContext],
) -> list[Tool]

从 MCP 服务器获取可用的工具。

源代码在 src/agents/agent.py 
127
128
129
130
131
132
async def get_mcp_tools(self, run_context: RunContextWrapper[TContext]) -> list[Tool]:
    """Fetches the available tools from the MCP servers."""
    convert_schemas_to_strict = self.mcp_config.get("convert_schemas_to_strict", False)
    return await MCPUtil.get_all_function_tools(
        self.mcp_servers, convert_schemas_to_strict, run_context, self
    )

get_all_tools 异步 

get_all_tools(
    run_context: RunContextWrapper[TContext],
) -> list[Tool]

所有代理工具，包括 MCP 工具和函数工具。

源代码在 src/agents/agent.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
async def get_all_tools(self, run_context: RunContextWrapper[TContext]) -> list[Tool]:
    """All agent tools, including MCP tools and function tools."""
    mcp_tools = await self.get_mcp_tools(run_context)

    async def _check_tool_enabled(tool: Tool) -> bool:
        if not isinstance(tool, FunctionTool):
            return True

        attr = tool.is_enabled
        if isinstance(attr, bool):
            return attr
        res = attr(run_context, self)
        if inspect.isawaitable(res):
            return bool(await res)
        return bool(res)

    results = await asyncio.gather(*(_check_tool_enabled(t) for t in self.tools))
    enabled: list[Tool] = [t for t, ok in zip(self.tools, results) if ok]
    return [*mcp_tools, *enabled]