交接

HandoffInputFilter 模块属性

HandoffInputFilter: TypeAlias = Callable[
    [HandoffInputData], MaybeAwaitable[HandoffInputData]
]

过滤传递给下一个代理的输入数据的一个函数。

HandoffHistoryMapper 模块属性

HandoffHistoryMapper: TypeAlias = Callable[
    [list[TResponseInputItem]], list[TResponseInputItem]
]

将先前的对话记录映射到嵌套摘要负载的一个函数。

HandoffInputData 数据类

源代码在 src/agents/handoffs/__init__.py

@dataclass(frozen=True)
class HandoffInputData:
    input_history: str | tuple[TResponseInputItem, ...]
    """
    The input history before `Runner.run()` was called.
    """

    pre_handoff_items: tuple[RunItem, ...]
    """
    The items generated before the agent turn where the handoff was invoked.
    """

    new_items: tuple[RunItem, ...]
    """
    The new items generated during the current agent turn, including the item that triggered the
    handoff and the tool output message representing the response from the handoff output.
    """

    run_context: RunContextWrapper[Any] | None = None
    """
    The run context at the time the handoff was invoked. Note that, since this property was added
    later on, it is optional for backwards compatibility.
    """

    def clone(self, **kwargs: Any) -> HandoffInputData:
        """
        Make a copy of the handoff input data, with the given arguments changed. For example, you
        could do:

        ```
        new_handoff_input_data = handoff_input_data.clone(new_items=())
        ```
        """

        return dataclasses_replace(self, **kwargs)

input_history 实例属性

input_history: str | tuple[TResponseInputItem, ...]

在调用 Runner.run() 之前，输入历史记录。

pre_handoff_items 实例属性

pre_handoff_items: tuple[RunItem, ...]

在发起 handoff 的 agent 轮次之前生成的项目。

new_items 实例属性

new_items: tuple[RunItem, ...]

在当前 agent 轮次中生成的新项目，包括触发 handoff 的项目以及代表 handoff 输出响应的工具输出消息。

run_context 类属性 实例属性

run_context: RunContextWrapper[Any] | None = None

调用 handoff 时运行上下文。请注意，由于此属性是后来添加的，因此为了向后兼容，它是可选的。

clone

clone(**kwargs: Any) -> HandoffInputData

制作 handoff 输入数据的副本，并更改给定的参数。例如，您可以

new_handoff_input_data = handoff_input_data.clone(new_items=())

源代码在 src/agents/handoffs/__init__.py

def clone(self, **kwargs: Any) -> HandoffInputData:
    """
    Make a copy of the handoff input data, with the given arguments changed. For example, you
    could do:

    ```
    new_handoff_input_data = handoff_input_data.clone(new_items=())
    ```
    """

    return dataclasses_replace(self, **kwargs)

Handoff 数据类

基类：Generic[TContext, TAgent]

Handoff 是指一个代理将任务委托给另一个代理。

例如，在客户支持场景中，您可能有一个“分诊代理”，它确定哪个代理应该处理用户的请求，以及专门处理不同领域的子代理，例如计费、帐户管理等。

源代码在 src/agents/handoffs/__init__.py

@dataclass
class Handoff(Generic[TContext, TAgent]):
    """A handoff is when an agent delegates a task to another agent.

    For example, in a customer support scenario you might have a "triage agent" that determines
    which agent should handle the user's request, and sub-agents that specialize in different areas
    like billing, account management, etc.
    """

    tool_name: str
    """The name of the tool that represents the handoff."""

    tool_description: str
    """The description of the tool that represents the handoff."""

    input_json_schema: dict[str, Any]
    """The JSON schema for the handoff input. Can be empty if the handoff does not take an input."""

    on_invoke_handoff: Callable[[RunContextWrapper[Any], str], Awaitable[TAgent]]
    """The function that invokes the handoff.

    The parameters passed are: (1) the handoff run context, (2) the arguments from the LLM as a
    JSON string (or an empty string if ``input_json_schema`` is empty). Must return an agent.
    """

    agent_name: str
    """The name of the agent that is being handed off to."""

    input_filter: HandoffInputFilter | None = None
    """A function that filters the inputs that are passed to the next agent.

    By default, the new agent sees the entire conversation history. In some cases, you may want to
    filter inputs (for example, to remove older inputs or remove tools from existing inputs). The
    function receives the entire conversation history so far, including the input item that
    triggered the handoff and a tool call output item representing the handoff tool's output. You
    are free to modify the input history or new items as you see fit. The next agent that runs will
    receive ``handoff_input_data.all_items``. IMPORTANT: in streaming mode, we will not stream
    anything as a result of this function. The items generated before will already have been
    streamed.
    """

    nest_handoff_history: bool | None = None
    """Override the run-level ``nest_handoff_history`` behavior for this handoff only."""

    strict_json_schema: bool = True
    """Whether the input JSON schema is in strict mode. We strongly recommend setting this to True
    because it increases the likelihood of correct JSON input."""

    is_enabled: bool | Callable[[RunContextWrapper[Any], AgentBase[Any]], MaybeAwaitable[bool]] = (
        True
    )
    """Whether the handoff is enabled.

    Either a bool or a callable that takes the run context and agent and returns whether the
    handoff is enabled. You can use this to dynamically enable or disable a handoff based on your
    context or state.
    """

    def get_transfer_message(self, agent: AgentBase[Any]) -> str:
        return json.dumps({"assistant": agent.name})

    @classmethod
    def default_tool_name(cls, agent: AgentBase[Any]) -> str:
        return _transforms.transform_string_function_style(f"transfer_to_{agent.name}")

    @classmethod
    def default_tool_description(cls, agent: AgentBase[Any]) -> str:
        return (
            f"Handoff to the {agent.name} agent to handle the request. "
            f"{agent.handoff_description or ''}"
        )

tool_name 实例属性

tool_name: str

The name of the tool that represents the handoff.

tool_description 实例属性

tool_description: str

The description of the tool that represents the handoff.

input_json_schema 实例属性

input_json_schema: dict[str, Any]

handoff 输入的 JSON 模式。如果 handoff 没有输入，则可以为空。

on_invoke_handoff 实例属性

on_invoke_handoff: Callable[
    [RunContextWrapper[Any], str], Awaitable[TAgent]
]

调用 handoff 的函数。

传递的参数是：（1）handoff 运行上下文，（2）来自 LLM 的参数作为 JSON 字符串（如果 input_json_schema 为空，则为空字符串）。必须返回一个代理。

agent_name 实例属性

agent_name: str

The name of the agent that is being handed off to.

input_filter 类属性 实例属性

input_filter: HandoffInputFilter | None = None

过滤传递给下一个代理的输入的函数。

默认情况下，新的代理会看到整个对话历史记录。在某些情况下，您可能希望过滤输入（例如，删除较早的输入或删除现有输入中的工具）。该函数接收到到目前为止的整个对话历史记录，包括触发 handoff 的输入项以及表示 handoff 工具输出的工具调用输出项。您可以根据需要修改输入历史记录或新项目。运行的下一个代理将收到 handoff_input_data.all_items。重要提示：在流式传输模式下，我们将不会对该函数的结果进行任何流式传输。之前生成的项目已经流式传输。

nest_handoff_history 类属性 实例属性

nest_handoff_history: bool | None = None

覆盖 run 级别的 nest_handoff_history 行为，仅针对此 handoff。

strict_json_schema 类属性 实例属性

strict_json_schema: bool = True

输入 JSON 模式是否处于严格模式。我们强烈建议将其设置为 True，因为它增加了正确 JSON 输入的可能性。

is_enabled 类属性 实例属性

is_enabled: (
    bool
    | Callable[
        [RunContextWrapper[Any], AgentBase[Any]],
        MaybeAwaitable[bool],
    ]
) = True

handoff 是否启用。

一个布尔值或一个接受运行上下文和代理并返回 handoff 是否启用的可调用对象。您可以使用此方法根据您的上下文或状态动态启用或禁用 handoff。

default_handoff_history_mapper

default_handoff_history_mapper(
    transcript: list[TResponseInputItem],
) -> list[TResponseInputItem]

返回一个总结对话记录的助手消息。

源代码在 src/agents/handoffs/history.py

def default_handoff_history_mapper(
    transcript: list[TResponseInputItem],
) -> list[TResponseInputItem]:
    """Return a single assistant message summarizing the transcript."""

    summary_message = _build_summary_message(transcript)
    return [summary_message]

get_conversation_history_wrappers

get_conversation_history_wrappers() -> tuple[str, str]

返回当前用于嵌套对话摘要的起始/结束标记。

源代码在 src/agents/handoffs/history.py

def get_conversation_history_wrappers() -> tuple[str, str]:
    """Return the current start/end markers used for the nested conversation summary."""

    return (_conversation_history_start, _conversation_history_end)

nest_handoff_history

nest_handoff_history(
    handoff_input_data: HandoffInputData,
    *,
    history_mapper: HandoffHistoryMapper | None = None,
) -> HandoffInputData

总结下一个代理的先前对话记录。

源代码在 src/agents/handoffs/history.py

def nest_handoff_history(
    handoff_input_data: HandoffInputData,
    *,
    history_mapper: HandoffHistoryMapper | None = None,
) -> HandoffInputData:
    """Summarize the previous transcript for the next agent."""

    normalized_history = _normalize_input_history(handoff_input_data.input_history)
    flattened_history = _flatten_nested_history_messages(normalized_history)
    pre_items_as_inputs = [
        _run_item_to_plain_input(item) for item in handoff_input_data.pre_handoff_items
    ]
    new_items_as_inputs = [_run_item_to_plain_input(item) for item in handoff_input_data.new_items]
    transcript = flattened_history + pre_items_as_inputs + new_items_as_inputs

    mapper = history_mapper or default_handoff_history_mapper
    history_items = mapper(transcript)
    filtered_pre_items = tuple(
        item
        for item in handoff_input_data.pre_handoff_items
        if _get_run_item_role(item) != "assistant"
    )

    return handoff_input_data.clone(
        input_history=tuple(deepcopy(item) for item in history_items),
        pre_handoff_items=filtered_pre_items,
    )

reset_conversation_history_wrappers

reset_conversation_history_wrappers() -> None

恢复默认的 <CONVERSATION HISTORY> 标记。

源代码在 src/agents/handoffs/history.py

def reset_conversation_history_wrappers() -> None:
    """Restore the default ``<CONVERSATION HISTORY>`` markers."""

    global _conversation_history_start, _conversation_history_end
    _conversation_history_start = _DEFAULT_CONVERSATION_HISTORY_START
    _conversation_history_end = _DEFAULT_CONVERSATION_HISTORY_END

set_conversation_history_wrappers

set_conversation_history_wrappers(
    *, start: str | None = None, end: str | None = None
) -> None

覆盖包装生成的对话摘要的标记。

传递 None 以保持任一侧不变。

源代码在 src/agents/handoffs/history.py

def set_conversation_history_wrappers(
    *,
    start: str | None = None,
    end: str | None = None,
) -> None:
    """Override the markers that wrap the generated conversation summary.

    Pass ``None`` to leave either side unchanged.
    """

    global _conversation_history_start, _conversation_history_end
    if start is not None:
        _conversation_history_start = start
    if end is not None:
        _conversation_history_end = end

handoff

handoff(
    agent: Agent[TContext],
    *,
    tool_name_override: str | None = None,
    tool_description_override: str | None = None,
    input_filter: Callable[
        [HandoffInputData], HandoffInputData
    ]
    | None = None,
    nest_handoff_history: bool | None = None,
    is_enabled: bool
    | Callable[
        [RunContextWrapper[Any], Agent[Any]],
        MaybeAwaitable[bool],
    ] = True,
) -> Handoff[TContext, Agent[TContext]]

handoff(
    agent: Agent[TContext],
    *,
    on_handoff: OnHandoffWithInput[THandoffInput],
    input_type: type[THandoffInput],
    tool_description_override: str | None = None,
    tool_name_override: str | None = None,
    input_filter: Callable[
        [HandoffInputData], HandoffInputData
    ]
    | None = None,
    nest_handoff_history: bool | None = None,
    is_enabled: bool
    | Callable[
        [RunContextWrapper[Any], Agent[Any]],
        MaybeAwaitable[bool],
    ] = True,
) -> Handoff[TContext, Agent[TContext]]

handoff(
    agent: Agent[TContext],
    *,
    on_handoff: OnHandoffWithoutInput,
    tool_description_override: str | None = None,
    tool_name_override: str | None = None,
    input_filter: Callable[
        [HandoffInputData], HandoffInputData
    ]
    | None = None,
    nest_handoff_history: bool | None = None,
    is_enabled: bool
    | Callable[
        [RunContextWrapper[Any], Agent[Any]],
        MaybeAwaitable[bool],
    ] = True,
) -> Handoff[TContext, Agent[TContext]]

handoff(
    agent: Agent[TContext],
    tool_name_override: str | None = None,
    tool_description_override: str | None = None,
    on_handoff: OnHandoffWithInput[THandoffInput]
    | OnHandoffWithoutInput
    | None = None,
    input_type: type[THandoffInput] | None = None,
    input_filter: Callable[
        [HandoffInputData], HandoffInputData
    ]
    | None = None,
    nest_handoff_history: bool | None = None,
    is_enabled: bool
    | Callable[
        [RunContextWrapper[Any], Agent[TContext]],
        MaybeAwaitable[bool],
    ] = True,
) -> Handoff[TContext, Agent[TContext]]

从一个代理创建一个 handoff。

参数

名称	类型	描述	默认
agent	Agent[TContext]	要 handoff 的代理。	required
tool_name_override	str | None	表示 handoff 的工具名称的可选覆盖。	None
tool_description_override	str | None	表示 handoff 的工具描述的可选覆盖。	None
on_handoff	OnHandoffWithInput[THandoffInput] | OnHandoffWithoutInput | None	调用 handoff 时运行的函数。	None
input_type	type[THandoffInput] | None	handoff 的输入类型。如果提供，则输入将根据此类型进行验证。仅当您传递接受输入的函数时才相关。	None
input_filter	Callable[[HandoffInputData], HandoffInputData] | None	过滤传递给下一个代理的输入的函数。	None
nest_handoff_history	bool | None	RunConfig 级别的 nest_handoff_history 标志的可选覆盖。如果为 None，我们将回退到运行的配置。	None
is_enabled	bool | Callable[[RunContextWrapper[Any], Agent[TContext]], MaybeAwaitable[bool]]	handoff 是否启用。可以是布尔值或接受运行上下文和代理并返回 handoff 是否启用的可调用对象。禁用的 handoff 在运行时对 LLM 隐藏。	True

源代码在 src/agents/handoffs/__init__.py

def handoff(
    agent: Agent[TContext],
    tool_name_override: str | None = None,
    tool_description_override: str | None = None,
    on_handoff: OnHandoffWithInput[THandoffInput] | OnHandoffWithoutInput | None = None,
    input_type: type[THandoffInput] | None = None,
    input_filter: Callable[[HandoffInputData], HandoffInputData] | None = None,
    nest_handoff_history: bool | None = None,
    is_enabled: bool
    | Callable[[RunContextWrapper[Any], Agent[TContext]], MaybeAwaitable[bool]] = True,
) -> Handoff[TContext, Agent[TContext]]:
    """Create a handoff from an agent.

    Args:
        agent: The agent to handoff to.
        tool_name_override: Optional override for the name of the tool that represents the handoff.
        tool_description_override: Optional override for the description of the tool that
            represents the handoff.
        on_handoff: A function that runs when the handoff is invoked.
        input_type: The type of the input to the handoff. If provided, the input will be validated
            against this type. Only relevant if you pass a function that takes an input.
        input_filter: A function that filters the inputs that are passed to the next agent.
        nest_handoff_history: Optional override for the RunConfig-level ``nest_handoff_history``
            flag. If ``None`` we fall back to the run's configuration.
        is_enabled: Whether the handoff is enabled. Can be a bool or a callable that takes the run
            context and agent and returns whether the handoff is enabled. Disabled handoffs are
            hidden from the LLM at runtime.
    """

    assert (on_handoff and input_type) or not (on_handoff and input_type), (
        "You must provide either both on_handoff and input_type, or neither"
    )
    type_adapter: TypeAdapter[Any] | None
    if input_type is not None:
        assert callable(on_handoff), "on_handoff must be callable"
        sig = inspect.signature(on_handoff)
        if len(sig.parameters) != 2:
            raise UserError("on_handoff must take two arguments: context and input")

        type_adapter = TypeAdapter(input_type)
        input_json_schema = type_adapter.json_schema()
    else:
        type_adapter = None
        input_json_schema = {}
        if on_handoff is not None:
            sig = inspect.signature(on_handoff)
            if len(sig.parameters) != 1:
                raise UserError("on_handoff must take one argument: context")

    async def _invoke_handoff(
        ctx: RunContextWrapper[Any], input_json: str | None = None
    ) -> Agent[TContext]:
        if input_type is not None and type_adapter is not None:
            if input_json is None:
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="Handoff function expected non-null input, but got None",
                        data={"details": "input_json is None"},
                    )
                )
                raise ModelBehaviorError("Handoff function expected non-null input, but got None")

            validated_input = _json.validate_json(
                json_str=input_json,
                type_adapter=type_adapter,
                partial=False,
            )
            input_func = cast(OnHandoffWithInput[THandoffInput], on_handoff)
            if inspect.iscoroutinefunction(input_func):
                await input_func(ctx, validated_input)
            else:
                input_func(ctx, validated_input)
        elif on_handoff is not None:
            no_input_func = cast(OnHandoffWithoutInput, on_handoff)
            if inspect.iscoroutinefunction(no_input_func):
                await no_input_func(ctx)
            else:
                no_input_func(ctx)

        return agent

    tool_name = tool_name_override or Handoff.default_tool_name(agent)
    tool_description = tool_description_override or Handoff.default_tool_description(agent)

    # Always ensure the input JSON schema is in strict mode. If needed, we can make this
    # configurable in the future.
    input_json_schema = ensure_strict_json_schema(input_json_schema)

    async def _is_enabled(ctx: RunContextWrapper[Any], agent_base: AgentBase[Any]) -> bool:
        from ..agent import Agent

        assert callable(is_enabled), "is_enabled must be callable here"
        assert isinstance(agent_base, Agent), "Can't handoff to a non-Agent"
        result = is_enabled(ctx, agent_base)
        if inspect.isawaitable(result):
            return await result
        return bool(result)

    return Handoff(
        tool_name=tool_name,
        tool_description=tool_description,
        input_json_schema=input_json_schema,
        on_invoke_handoff=_invoke_handoff,
        input_filter=input_filter,
        nest_handoff_history=nest_handoff_history,
        agent_name=agent.name,
        is_enabled=_is_enabled if callable(is_enabled) else is_enabled,
    )