test/_helpers.py

Rama main · réplica de lectura del repositorio original.

# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0

from ag2.events import BaseEvent, ModelReasoning, ProviderReplay
from ag2.tools import tool


class DurableReasoning(ModelReasoning, ProviderReplay):
    """Provider reasoning item that must be replayed, like OpenAIReasoningEvent.

    Reasoning is transient by default; a provider that persists its item opts out
    (see ``ag2.config.openai.events``). Redeclared here so tests outside a
    provider package can exercise the anchor case without importing its SDK.
    """

    __transient__ = False
    __replay_role__ = "anchor"


class ProviderTurnState(BaseEvent, ProviderReplay):
    """Provider-native object standing in for a whole assistant turn.

    Like ``XAIAssistantEvent``: the only way to rebuild a turn carrying
    ``tool_calls`` for a provider whose SDK cannot construct one from primitives.
    Redeclared here so tests outside a provider package can exercise the turn
    case without importing its SDK.
    """

    __replay_role__ = "turn"


@tool
def lookup() -> str:
    """Look something up."""
    return "42"


class ScriptedHuman:
    """A ``hitl_hook`` that gives ``answer`` to every question and counts them."""

    def __init__(self, answer: str) -> None:
        self.answer = answer
        self.questions = 0

    def __call__(self, request: object) -> str:
        self.questions += 1
        return self.answer