Adapters API Reference¶

class PydanticAIAdapter:
    """Wrap a pydantic-ai Agent to conform to the agent callable contract.

    Args:
        agent: A pydantic-ai ``Agent`` instance.

    Example:
        ```python
        from pydantic_ai import Agent
        from pytest_agent_eval.adapters.pydantic_ai import PydanticAIAdapter

        my_agent = Agent("openai:gpt-4o", system_prompt="You are helpful.")

        @pytest.fixture
        def llm_eval_agent():
            return PydanticAIAdapter(my_agent)
        ```
    """

    def __init__(self, agent: AnyAgent) -> None:
        """Store the pydantic-ai agent to delegate calls to."""
        if not hasattr(agent, "run"):
            raise TypeError(
                f"PydanticAIAdapter expects a pydantic-ai Agent with an async .run() method, "
                f"got {type(agent).__name__}."
            )
        self._agent = agent

    async def __call__(self, history: History) -> AgentReply:
        """Run the agent and normalise output to (reply, tool_calls)."""
        user_msg = history[-1].content if history else ""
        message_history = _to_model_messages(history[:-1], _static_system_prompts(self._agent))
        result = await self._agent.run(user_msg, message_history=message_history or None)

        tool_calls = [
            ToolCall(part.tool_name, part.args_as_dict())
            for msg in result.all_messages()
            for part in getattr(msg, "parts", [])
            if _is_tool_call_part(part)
        ]

        reply = result.output if isinstance(result.output, str) else str(result.output)
        return AgentReply(reply, tool_calls)

class LangChainAdapter:
    """Wrap a LangChain Runnable to conform to the agent callable contract.

    Expects the runnable to accept ``{"messages": [...]}`` and return an
    ``AIMessage`` or object with a ``content`` attribute.

    Args:
        runnable: A LangChain Runnable (e.g. a compiled graph or chain).

    Example:
        ```python
        from pytest_agent_eval.adapters.langchain import LangChainAdapter

        @pytest.fixture
        def llm_eval_agent():
            return LangChainAdapter(my_langchain_graph)
        ```
    """

    def __init__(self, runnable: LangChainRunnable) -> None:
        """Store the LangChain runnable to delegate calls to."""
        # The real Runnable type is on the parameter, so a type checker rejects a wrong
        # object at the call site. The guard is for callers without one: it names the extra
        # to install, which an assignability error does not.
        if not hasattr(runnable, "ainvoke"):
            raise TypeError(
                f"LangChainAdapter expects a LangChain Runnable with an .ainvoke() method, "
                f"got {type(runnable).__name__}. Wrap a compiled graph or chain, and make sure "
                "the extra is installed: pip install 'pytest-agent-eval[langchain]'"
            )
        self._runnable = runnable

    async def __call__(self, history: History) -> AgentReply:
        """Run the runnable and normalise output to (reply, tool_calls)."""
        # Plain dicts, not Messages: langchain_core.convert_to_messages() raises
        # NotImplementedError on a Mapping that is not a dict.
        result = await self._runnable.ainvoke({"messages": [m.to_dict() for m in history]})

        # `ainvoke` returns `object`, so reading the attribute is the one untyped hop.
        # Absent *and* None both mean "no tools were called"; LangChain produces either.
        if hasattr(result, "content"):
            return AgentReply(str(result.content), _tool_calls(getattr(result, "tool_calls", []) or []))
        last = _last_message(result)
        if last is not None:
            content = str(getattr(last, "content", ""))
            return AgentReply(content, _tool_calls(getattr(last, "tool_calls", []) or []))
        return AgentReply(str(result), [])

class OpenAIAdapter:
    """Wrap an AsyncOpenAI client to conform to the agent callable contract.

    Args:
        client: An ``openai.AsyncOpenAI`` or ``openai.AsyncAzureOpenAI`` instance.
        model: Model name to use for completions (e.g. ``"gpt-4o"``).
        system_prompt: Optional system prompt prepended to every call.

    Example:
        ```python
        from openai import AsyncOpenAI
        from pytest_agent_eval.adapters.openai import OpenAIAdapter

        @pytest.fixture
        def llm_eval_agent():
            client = AsyncOpenAI()
            return OpenAIAdapter(client, model="gpt-4o")
        ```
    """

    def __init__(
        self,
        client: AsyncOpenAI,
        model: str,
        system_prompt: str | None = None,
    ) -> None:
        """Store the OpenAI client, model name, and optional system prompt."""
        # The real SDK type, imported under TYPE_CHECKING, and the sole adapter that cannot
        # use a Protocol: `create` is overloaded (streaming vs not), which no hand-rolled
        # structural type can express. The guard stays for callers with no type checker —
        # it is what turns a missing extra into a message naming it.
        if not hasattr(client, "chat"):
            raise TypeError(
                f"OpenAIAdapter expects an AsyncOpenAI-compatible client with .chat.completions, "
                f"got {type(client).__name__}. Make sure the extra is installed: "
                "pip install 'pytest-agent-eval[openai]'"
            )
        self._client = client
        self._model = model
        self._system_prompt = system_prompt

    async def __call__(self, history: History) -> AgentReply:
        """Run a chat completion and normalise to (reply, tool_calls)."""
        messages: list[ChatCompletionMessageParam] = []
        if self._system_prompt:
            messages.append({"role": "system", "content": self._system_prompt})
        messages.extend(_as_param(m) for m in history)

        response = await self._client.chat.completions.create(
            model=self._model,
            messages=messages,
        )
        message = response.choices[0].message
        reply = message.content or ""
        return AgentReply(reply, [_tool_call(tc) for tc in (message.tool_calls or [])])

class SmolagentsAdapter:
    """Wrap a smolagents agent to conform to the agent callable contract.

    Duck-typed: works with any object exposing ``.run(task, reset=...)`` and
    ``.memory.steps``. Smolagents's sync ``run`` is offloaded with
    ``asyncio.to_thread`` so the event loop stays responsive.

    Args:
        agent: A smolagents agent (e.g. ``ToolCallingAgent``, ``CodeAgent``).
        include_internal_tools: When ``True``, smolagents-internal pseudo-tools
            (``python_interpreter``, ``final_answer``) are included in the
            returned tool-call list. Defaults to ``False``.

    Example:
        ```python
        from smolagents import ToolCallingAgent, InferenceClientModel
        from pytest_agent_eval.adapters.smolagents import SmolagentsAdapter

        model = InferenceClientModel(model_id="meta-llama/Llama-3.3-70B-Instruct")
        agent = ToolCallingAgent(tools=[...], model=model)

        @pytest.fixture
        def llm_eval_agent():
            return SmolagentsAdapter(agent)
        ```
    """

    def __init__(self, agent: SmolagentsAgent, *, include_internal_tools: bool = False) -> None:
        """Store the smolagents agent and the internal-tool filter setting."""
        # The Protocol is on the parameter, so a type checker rejects a wrong object at the
        # call site. The guard is for callers without one: it names the extra to install,
        # which an assignability error does not.
        if not hasattr(agent, "run") or not hasattr(agent, "memory"):
            raise TypeError(
                f"SmolagentsAdapter expects a smolagents agent with .run() and .memory.steps, "
                f"got {type(agent).__name__}. Make sure the extra is installed: "
                "pip install 'pytest-agent-eval[smolagents]'"
            )
        self._agent = agent
        self._include_internal_tools = include_internal_tools

    async def __call__(self, history: History) -> AgentReply:
        """Run the agent against the latest user message and return (reply, tool_calls)."""
        user_msg = history[-1].content if history else ""
        reset = len(history) == 1
        prev = len(self._agent.memory.steps)
        result = await asyncio.to_thread(self._agent.run, user_msg, reset=reset)
        new_steps = self._agent.memory.steps[prev:] if not reset else self._agent.memory.steps
        calls = [
            ToolCall(tc.name, coerce_args(getattr(tc, "arguments", None)))
            for step in new_steps
            for tc in getattr(step, "tool_calls", None) or []
        ]
        if not self._include_internal_tools:
            calls = [c for c in calls if c not in _INTERNAL_TOOLS]
        return AgentReply(str(result), calls)

class LiveKitAdapter:
    """Voice adapter: streams a WAV per turn into a fresh LiveKit ``AgentSession``.

    The user supplies a ``session_factory`` callable that returns a fresh
    ``(AgentSession, Agent)`` pair on every invocation — one pair per turn.
    The adapter attaches a :class:`WavFileAudioInput` from the turn's
    ``audio:`` field, captures every executed tool call via
    ``function_tools_executed``, and accumulates the assistant transcript via
    ``conversation_item_added``.

    Args:
        session_factory: Returns a fresh ``(AgentSession, Agent)`` per call.
        sample_rate: WAV sample rate in Hz (must match the input file). Default
            24 kHz, the OpenAI Realtime native rate.
        frame_ms: Frame size in milliseconds. Default 20 ms.
        grace_period_s: Seconds to wait after the WAV drains before closing
            the session — gives the model time to fire trailing tool calls.
        timeout_s: Maximum seconds to wait for WAV exhaustion before forcibly
            closing the session.

    Example:
        ```python
        from livekit.agents.voice import Agent, AgentSession
        from livekit.plugins import openai
        from pytest_agent_eval.adapters.livekit import LiveKitAdapter

        def make_session():
            session = AgentSession(llm=openai.realtime.RealtimeModel())
            agent = Agent(instructions="...", tools=[...])
            return session, agent

        @pytest.fixture
        def llm_eval_agent():
            return LiveKitAdapter(make_session)
        ```
    """

    def __init__(
        self,
        session_factory: SessionFactory,
        *,
        sample_rate: int = 24_000,
        frame_ms: int = 20,
        grace_period_s: float = 8.0,
        timeout_s: float = 30.0,
    ) -> None:
        """Store the session factory and streaming/event-capture knobs."""
        self._session_factory = session_factory
        self._sample_rate = sample_rate
        self._frame_ms = frame_ms
        self._grace_period_s = grace_period_s
        self._timeout_s = timeout_s
        for name in _QUIET_LOGGERS:
            logging.getLogger(name).setLevel(logging.WARNING)

    async def __call__(self, history: History) -> AgentReply:
        """Stream the WAV on the last user turn and return ``(reply, tool_calls)``."""
        if not history or history[-1].role != "user":
            raise ValueError("LiveKitAdapter: history must end with a user turn")
        audio_path_raw = history[-1].audio
        if not audio_path_raw:
            raise ValueError(
                "LiveKitAdapter requires Turn.audio — the last user turn has no audio path. "
                "Run `python -m pytest_agent_eval.synthesize_audio` to generate fixtures."
            )

        wav_path = Path(audio_path_raw)
        if not wav_path.exists():
            raise FileNotFoundError(
                f"LiveKitAdapter: WAV fixture missing at {wav_path}. "
                "Run `python -m pytest_agent_eval.synthesize_audio` to generate it."
            )

        session, agent = self._session_factory()

        tool_calls: list[ToolCall] = []
        reply_chunks: list[str] = []

        # The reads stay `getattr` with a default even though the events are typed:
        # livekit's payloads vary by version and by which model fired them, and
        # `event.item` is itself a union whose members differ.
        def _on_function_tools_executed(event: FunctionToolsExecutedEvent) -> None:
            """Record every tool call livekit reports as executed on this turn."""
            for fc in getattr(event, "function_calls", []) or []:
                name = getattr(fc, "name", "") or ""
                if name:
                    tool_calls.append(ToolCall(name, coerce_args(getattr(fc, "arguments", None))))

        def _on_conversation_item_added(event: ConversationItemAddedEvent) -> None:
            """Accumulate the assistant's transcript as livekit appends conversation items."""
            item = getattr(event, "item", None)
            if item is None:
                return
            if getattr(item, "role", None) != "assistant":
                return
            text = getattr(item, "text_content", None)
            if not text:
                content = getattr(item, "content", None) or []
                text = "".join(c for c in content if isinstance(c, str))
            if text:
                reply_chunks.append(text)

        session.on("function_tools_executed", _on_function_tools_executed)
        session.on("conversation_item_added", _on_conversation_item_added)

        wav_input = WavFileAudioInput(
            wav_path,
            sample_rate=self._sample_rate,
            frame_ms=self._frame_ms,
        )
        session.input.audio = wav_input

        try:
            await session.start(agent)
            try:
                await asyncio.wait_for(wav_input.wait_for_exhaustion(), timeout=self._timeout_s)
            except TimeoutError:
                logger.warning("LiveKitAdapter: timed out waiting for WAV exhaustion")
            await asyncio.sleep(self._grace_period_s)
        finally:
            try:
                await wav_input.aclose()
            except Exception:
                logger.debug("LiveKitAdapter: wav_input.aclose raised", exc_info=True)
            try:
                await session.aclose()
            except Exception:
                logger.debug("LiveKitAdapter: session.aclose raised", exc_info=True)

        return AgentReply("".join(reply_chunks), tool_calls)