karenina.adapters.langchain_deep_agents

langchain_deep_agents

LangChain Deep Agents adapter for natively agentic evaluation.

This adapter provides AgentPort, LLMPort, and ParserPort implementations using LangChain Deep Agents (create_deep_agent). It enables provider-agnostic agentic evaluation with built-in planning, context management, and subagent orchestration.

Requires: pip install deepagents langchain-mcp-adapters

Adapter classes

• DeepAgentsAgentAdapter: Agent loops via create_deep_agent with MCP support
• DeepAgentsLLMAdapter: Simple LLM invocation via single-turn agent
• DeepAgentsParserAdapter: Structured output parsing

Utilities

• DeepAgentsMessageConverter: Convert between unified Message and LangGraph types
• check_deep_agents_available: Check if deepagents is installed
• convert_mcp_to_tools: Convert MCPServerConfig to LangChain tools
• extract_deep_agents_usage: Extract UsageMetadata from agent results
• deep_agents_messages_to_raw_trace: Format messages as raw trace string
• deep_agents_messages_to_trace_messages: Structured trace_messages with per-call usage

Classes

DeepAgentsAgentAdapter

Agent adapter using LangChain Deep Agents' create_deep_agent.

This adapter implements the AgentPort Protocol for agent execution with built-in planning tools, filesystem operations, subagent delegation, and context management. Uses create_deep_agent() which returns a compiled LangGraph graph.

The adapter handles: - Message conversion from unified Message to prompt string - Model initialization via init_chat_model - Agent creation and invocation via LangGraph - Dual trace output (raw_trace string and trace_messages list) - Usage metadata extraction from AIMessage response_metadata - Recursion limit detection from LangGraph state

Global concurrency cap: max_concurrent_requests (GlobalLLMLimiter) gates single-turn LLM adapter calls and langchain-based agent model calls only. This agent's internal model calls run outside that cap. They were never semaphore-gated, so this is the honest current state rather than a regression.

Example

config = ModelConfig( ... id="test", ... model_name="claude-sonnet-4-20250514", ... model_provider="anthropic", ... interface="langchain_deep_agents", ... ) adapter = DeepAgentsAgentAdapter(config) result = await adapter.arun( ... messages=[Message.user("What files are in /tmp?")], ... config=AgentConfig(max_turns=10), ... ) print(result.final_response)

Source code in src/karenina/adapters/langchain_deep_agents/agent.py

class DeepAgentsAgentAdapter:
    """Agent adapter using LangChain Deep Agents' create_deep_agent.

    This adapter implements the AgentPort Protocol for agent execution with
    built-in planning tools, filesystem operations, subagent delegation,
    and context management. Uses create_deep_agent() which returns a
    compiled LangGraph graph.

    The adapter handles:
    - Message conversion from unified Message to prompt string
    - Model initialization via init_chat_model
    - Agent creation and invocation via LangGraph
    - Dual trace output (raw_trace string and trace_messages list)
    - Usage metadata extraction from AIMessage response_metadata
    - Recursion limit detection from LangGraph state

    Global concurrency cap: ``max_concurrent_requests`` (GlobalLLMLimiter)
    gates single-turn LLM adapter calls and langchain-based agent model
    calls only. This agent's internal model calls run outside that cap.
    They were never semaphore-gated, so this is the honest current state
    rather than a regression.

    Example:
        >>> config = ModelConfig(
        ...     id="test",
        ...     model_name="claude-sonnet-4-20250514",
        ...     model_provider="anthropic",
        ...     interface="langchain_deep_agents",
        ... )
        >>> adapter = DeepAgentsAgentAdapter(config)
        >>> result = await adapter.arun(
        ...     messages=[Message.user("What files are in /tmp?")],
        ...     config=AgentConfig(max_turns=10),
        ... )
        >>> print(result.final_response)
    """

    def __init__(self, model_config: ModelConfig) -> None:
        """Initialize the Deep Agents adapter.

        Args:
            model_config: Configuration specifying model, provider, and interface.
        """
        self._config = model_config
        self._converter = DeepAgentsMessageConverter()

    @property
    def capabilities(self) -> PortCapabilities:
        """Declare adapter capabilities.

        Returns:
            PortCapabilities implied by the configured DeepAgents backend.
        """
        return get_agent_runtime_capabilities(self._config)

    def _build_backend(self, workspace_path: Any) -> Any:
        """Create the DeepAgents backend configured for this model."""

        backend = get_deepagents_backend(self._config)
        access_mode = get_agent_runtime_access_mode(self._config)
        read_max_bytes = _runtime_int_option(self._config, "read_max_bytes", 0)
        if backend == CONTAINER_BACKEND:
            if workspace_path is None:
                raise AdapterUnavailableError(
                    "agent_runtime backend='container' requires an AgentConfig.workspace_path",
                    reason="missing_workspace",
                )
            from .docker_backend import ContainerSandboxBackend

            container_backend = ContainerSandboxBackend(
                root_dir=workspace_path,
                container_config=get_container_runtime_config(self._config),
                timeout=_runtime_int_option(self._config, "execute_timeout", 120),
                max_output_bytes=_runtime_int_option(self._config, "execute_max_output_bytes", 100_000),
            )
            if access_mode == "read_only":
                from .read_only_backend import ReadOnlyBackend

                return ReadOnlyBackend(container_backend, read_max_bytes=read_max_bytes)
            return container_backend

        if backend == "local_shell":
            if workspace_path is None:
                raise AdapterUnavailableError(
                    "agent_runtime backend='local_shell' requires an AgentConfig.workspace_path",
                    reason="missing_workspace",
                )
            from deepagents.backends import LocalShellBackend

            logger.warning(
                "Using unsafe DeepAgents LocalShellBackend with root_dir=%s",
                workspace_path,
            )
            local_backend = LocalShellBackend(
                root_dir=str(workspace_path),
                virtual_mode=True,
                timeout=_runtime_int_option(self._config, "execute_timeout", 120),
                max_output_bytes=_runtime_int_option(self._config, "execute_max_output_bytes", 100_000),
                inherit_env=True,
            )
            if access_mode == "read_only":
                from .read_only_backend import ReadOnlyBackend

                return ReadOnlyBackend(local_backend, read_max_bytes=read_max_bytes)
            return local_backend

        from deepagents.backends import FilesystemBackend

        if workspace_path:
            logger.info("Using FilesystemBackend with root_dir=%s", workspace_path)
            filesystem_backend = FilesystemBackend(root_dir=str(workspace_path), virtual_mode=True)
            if access_mode == "read_only":
                from .read_only_backend import ReadOnlyBackend

                return ReadOnlyBackend(filesystem_backend, read_max_bytes=read_max_bytes)
            return filesystem_backend

        logger.info("Using FilesystemBackend with default root (cwd)")
        filesystem_backend = FilesystemBackend(virtual_mode=True)
        if access_mode == "read_only":
            from .read_only_backend import ReadOnlyBackend

            return ReadOnlyBackend(filesystem_backend, read_max_bytes=read_max_bytes)
        return filesystem_backend

    def _build_raw_trace(
        self,
        lc_messages: list[Any],
        subgraph_results: dict[tuple[Any, ...], dict[str, Any]],
        *,
        limit_reached: bool,
        timeout_reached: bool,
    ) -> str:
        """Build a raw trace, including any streamed subgraph state."""

        raw_trace = deep_agents_messages_to_raw_trace(lc_messages) if lc_messages else ""
        subgraph_trace_parts = []
        for namespace, subgraph_result in subgraph_results.items():
            subgraph_messages = subgraph_result.get("messages", [])
            subgraph_trace = deep_agents_messages_to_raw_trace(subgraph_messages)
            if subgraph_trace:
                namespace_text = " / ".join(str(part) for part in namespace)
                subgraph_trace_parts.append(f"--- Subgraph Trace: {namespace_text} ---\n{subgraph_trace}")
        if subgraph_trace_parts:
            parts = [part for part in [raw_trace, *subgraph_trace_parts] if part]
            raw_trace = "\n\n".join(parts).strip()
        if limit_reached:
            raw_trace += "\n\n[Note: Recursion limit reached, partial response shown]"
        if timeout_reached:
            raw_trace += "\n\n[Note: Wall-clock timeout reached, partial response shown]"
        return raw_trace.strip()

    def _extract_final_response(self, lc_messages: list[Any]) -> str:
        """Extract final text response from the last AIMessage.

        Args:
            lc_messages: List of LangGraph BaseMessage objects.

        Returns:
            The final text response string.
        """
        from langchain_core.messages import AIMessage

        for msg in reversed(lc_messages):
            if isinstance(msg, AIMessage):
                if isinstance(msg.content, str) and msg.content:
                    return msg.content
                if isinstance(msg.content, list):
                    text_parts = []
                    for block in msg.content:
                        if isinstance(block, str):
                            text_parts.append(block)
                        elif isinstance(block, dict) and block.get("type") == "text":
                            text_parts.append(block["text"])
                    if text_parts:
                        return "\n".join(text_parts)

        return "[No final response extracted]"

    def _count_turns(self, lc_messages: list[Any]) -> int:
        """Count the number of agent turns (AIMessage instances).

        Args:
            lc_messages: List of LangGraph BaseMessage objects.

        Returns:
            Number of AIMessage instances in the conversation.
        """
        from langchain_core.messages import AIMessage

        return sum(1 for msg in lc_messages if isinstance(msg, AIMessage))

    async def arun(
        self,
        messages: list[Message],
        tools: list[Tool] | None = None,
        mcp_servers: dict[str, MCPServerConfig] | None = None,
        config: AgentConfig | None = None,
        _partial_result_store: _PartialResultStore | None = None,
    ) -> AgentResult:
        """Execute an agent loop with optional tools and MCP servers.

        Args:
            messages: Initial conversation messages.
            tools: Optional list of Tool definitions the agent can invoke.
            mcp_servers: Optional dict of MCP server configurations.
            config: Optional AgentConfig for execution parameters.

        Returns:
            AgentResult with final response, traces, usage, and metadata.

        Raises:
            AgentExecutionError: If the agent fails during execution.
            AgentTimeoutError: If execution exceeds the timeout.
            AgentResponseError: If the response is malformed or invalid.
        """
        global _create_deep_agent  # noqa: PLW0603
        if _create_deep_agent is None:
            from deepagents import create_deep_agent as _cda

            _create_deep_agent = _cda

        config = config or AgentConfig()

        # Convert messages to prompt string and extract system prompt
        prompt_string = self._converter.to_prompt_string(messages)
        system_prompt = self._converter.extract_system_prompt(messages)

        # Use config system_prompt as fallback
        if not system_prompt and config.system_prompt:
            system_prompt = config.system_prompt
        elif not system_prompt and self._config.system_prompt:
            system_prompt = self._config.system_prompt

        # Initialize model. Agent loops need SDK-level retries at the model
        # call boundary; retrying the whole agent can repeat workspace writes.
        retry_policy = self._config.retry_policy or RetryPolicy()
        chat_model = create_chat_model(
            self._config,
            max_retries=retry_policy.derive_sdk_max_retries(),
        )

        # Build agent kwargs
        agent_kwargs: dict[str, Any] = {"model": chat_model}
        if system_prompt:
            agent_kwargs["system_prompt"] = system_prompt

        agent_kwargs["backend"] = self._build_backend(config.workspace_path)

        # Pass through extra config to create_deep_agent
        if config.extra:
            for key, value in config.extra.items():
                if key not in ("model", "system_prompt", "backend"):
                    agent_kwargs[key] = value

        # Convert MCP servers to LangChain tools and combine with explicit tools
        all_tools: list[Any] = []
        if tools:
            all_tools.extend(tools)

        # Use AsyncExitStack for persistent MCP sessions. Sessions stay alive
        # for all tool calls during agent execution. Exceptions are captured
        # inside the block and re-raised after clean exit, because MCP session
        # cleanup can wrap errors in ExceptionGroup if an exception propagates
        # through the exit stack.
        result: dict[str, Any] = {}
        subgraph_results: dict[tuple[Any, ...], dict[str, Any]] = {}
        limit_reached = False
        timeout_reached = False
        deferred_error: Exception | None = None

        async with AsyncExitStack() as exit_stack:
            # Convert MCP servers to LangChain tools
            if mcp_servers:
                from .mcp import convert_mcp_to_tools

                try:
                    mcp_tools = await convert_mcp_to_tools(mcp_servers, exit_stack)
                    all_tools.extend(mcp_tools)
                    logger.info(
                        "Loaded %d MCP tools from %d servers",
                        len(mcp_tools),
                        len(mcp_servers),
                    )
                except Exception as e:
                    logger.warning("Failed to load MCP tools: %s", e)
                    deferred_error = AgentExecutionError(f"Failed to initialize MCP tools: {e}")
                    deferred_error.__cause__ = e

            if deferred_error is None:
                # Pass tools to agent if any were collected
                if all_tools:
                    agent_kwargs["tools"] = all_tools

                # Create the agent
                agent = _create_deep_agent(**agent_kwargs)

                # Build invocation input
                invoke_input: dict[str, Any] = {
                    "messages": [{"role": "user", "content": prompt_string}],
                }

                # LangGraph config for recursion limit
                # Each tool call + response = 2 steps, so double max_turns
                langgraph_config: RunnableConfig = {
                    "recursion_limit": config.max_turns * 2,
                }

                # Execute agent
                async def execute_agent() -> None:
                    nonlocal result, subgraph_results, limit_reached
                    async for chunk in agent.astream(
                        invoke_input,
                        config=langgraph_config,
                        stream_mode="values",
                        subgraphs=True,
                    ):
                        namespace: tuple[Any, ...] = ()
                        state = chunk
                        if isinstance(chunk, tuple) and len(chunk) == 2:
                            namespace_raw, state = chunk
                            namespace = tuple(namespace_raw)
                        if isinstance(state, dict):
                            if namespace:
                                subgraph_results[namespace] = state
                            else:
                                result = state
                    if result.get("is_last_step", False):
                        limit_reached = True

                try:
                    if config.timeout:
                        await asyncio.wait_for(execute_agent(), timeout=config.timeout)
                    else:
                        await execute_agent()

                except TimeoutError:
                    timeout_reached = True
                    logger.warning("Agent timed out after %ss; returning partial trace", config.timeout)
                    if _partial_result_store is not None:
                        partial_lc_messages = result.get("messages", [])
                        raw_trace = self._build_raw_trace(
                            partial_lc_messages,
                            subgraph_results,
                            limit_reached=limit_reached,
                            timeout_reached=True,
                        )
                        if partial_lc_messages:
                            partial_result = AgentResult(
                                final_response=self._extract_final_response(partial_lc_messages),
                                raw_trace=raw_trace,
                                trace_messages=self._converter.from_provider(partial_lc_messages),
                                usage=extract_deep_agents_usage(partial_lc_messages, model=self._config.model_name),
                                turns=self._count_turns(partial_lc_messages),
                                limit_reached=limit_reached,
                                session_id=None,
                                actual_model=extract_actual_model(partial_lc_messages) or self._config.model_name,
                                timeout_reached=True,
                            )
                        else:
                            partial_result = AgentResult(
                                final_response="[Agent stopped before producing a final response]",
                                raw_trace=raw_trace or "[Note: Agent stopped before producing trace messages]",
                                trace_messages=[],
                                usage=UsageMetadata(model=self._config.model_name),
                                turns=0,
                                limit_reached=limit_reached,
                                session_id=None,
                                actual_model=self._config.model_name,
                                timeout_reached=True,
                            )
                        _partial_result_store.set(partial_result)
                except Exception as e:
                    mapped_error, was_limit = wrap_deep_agents_error(e)
                    if was_limit:
                        limit_reached = True
                        logger.warning("Agent hit turn limit: %s", e)
                    else:
                        deferred_error = mapped_error
                        deferred_error.__cause__ = e

        # exit_stack closed: MCP sessions cleaned up

        if deferred_error is not None:
            raise deferred_error

        # Extract messages from result
        lc_messages: list[Any] = result.get("messages", [])

        if not lc_messages and not limit_reached and not timeout_reached:
            raise AgentResponseError("No messages received from Deep Agents")

        # If limit was reached but no messages, return a partial result
        if not lc_messages and (limit_reached or timeout_reached):
            raw_trace = self._build_raw_trace(
                lc_messages,
                subgraph_results,
                limit_reached=limit_reached,
                timeout_reached=timeout_reached,
            )
            return AgentResult(
                final_response="[Agent stopped before producing a final response]",
                raw_trace=raw_trace or "[Note: Agent stopped before producing trace messages]",
                trace_messages=[],
                usage=UsageMetadata(model=self._config.model_name),
                turns=0,
                limit_reached=limit_reached,
                session_id=None,
                actual_model=self._config.model_name,
                timeout_reached=timeout_reached,
            )

        # Build raw_trace (legacy string format)
        raw_trace = self._build_raw_trace(
            lc_messages,
            subgraph_results,
            limit_reached=limit_reached,
            timeout_reached=timeout_reached,
        )

        # Build trace_messages (structured format)
        trace_messages = self._converter.from_provider(lc_messages)

        # Extract final response
        final_response = self._extract_final_response(lc_messages)

        # Extract usage
        usage = extract_deep_agents_usage(lc_messages, model=self._config.model_name)

        # Count turns
        turns = self._count_turns(lc_messages)

        # Extract actual model
        actual_model = extract_actual_model(lc_messages) or self._config.model_name

        return AgentResult(
            final_response=final_response,
            raw_trace=raw_trace,
            trace_messages=trace_messages,
            usage=usage,
            turns=turns,
            limit_reached=limit_reached,
            session_id=None,
            actual_model=actual_model,
            timeout_reached=timeout_reached,
        )

    def run(
        self,
        messages: list[Message],
        tools: list[Tool] | None = None,
        mcp_servers: dict[str, MCPServerConfig] | None = None,
        config: AgentConfig | None = None,
    ) -> AgentResult:
        """Synchronous wrapper for arun().

        Args:
            messages: Initial conversation messages.
            tools: Optional list of Tool definitions.
            mcp_servers: Optional MCP server configurations.
            config: Optional AgentConfig for execution parameters.

        Returns:
            AgentResult from the agent execution.

        Raises:
            AgentExecutionError: If the agent fails during execution.
            AgentTimeoutError: If execution exceeds the timeout.
            AgentResponseError: If the response is malformed or invalid.
        """
        timeout = config.timeout if config and config.timeout else 600
        partial_result_store = _PartialResultStore()
        return cast(
            AgentResult,
            _run_in_fresh_loop(
                self.arun,
                messages,
                tools,
                mcp_servers,
                config,
                partial_result_store,
                timeout=timeout,
                timeout_result=partial_result_store.get,
            ),
        )

    async def aclose(self) -> None:
        """Close underlying resources.

        Deep Agents manages its own cleanup via LangGraph's compiled graph,
        so this is a no-op. Provided for interface consistency with other
        adapters that do require cleanup.
        """

Attributes

capabilities property

capabilities: PortCapabilities

Declare adapter capabilities.

Returns:

Type	Description
PortCapabilities	PortCapabilities implied by the configured DeepAgents backend.

Functions

__init__

__init__(model_config: ModelConfig) -> None

Parameters:

Name	Type	Description	Default
model_config	ModelConfig	Configuration specifying model, provider, and interface.	required

Source code in src/karenina/adapters/langchain_deep_agents/agent.py

def __init__(self, model_config: ModelConfig) -> None:
    """Initialize the Deep Agents adapter.

    Args:
        model_config: Configuration specifying model, provider, and interface.
    """
    self._config = model_config
    self._converter = DeepAgentsMessageConverter()

aclose async

aclose() -> None

Close underlying resources.

Deep Agents manages its own cleanup via LangGraph's compiled graph, so this is a no-op. Provided for interface consistency with other adapters that do require cleanup.

Source code in src/karenina/adapters/langchain_deep_agents/agent.py

async def aclose(self) -> None:
    """Close underlying resources.

    Deep Agents manages its own cleanup via LangGraph's compiled graph,
    so this is a no-op. Provided for interface consistency with other
    adapters that do require cleanup.
    """

arun async

arun(
    messages: list[Message],
    tools: list[Tool] | None = None,
    mcp_servers: dict[str, MCPServerConfig] | None = None,
    config: AgentConfig | None = None,
    _partial_result_store: _PartialResultStore
    | None = None,
) -> AgentResult

Execute an agent loop with optional tools and MCP servers.

Parameters:

Name	Type	Description	Default
messages	list[Message]	Initial conversation messages.	required
tools	list[Tool] | None	Optional list of Tool definitions the agent can invoke.	None
mcp_servers	dict[str, MCPServerConfig] | None	Optional dict of MCP server configurations.	None
config	AgentConfig | None	Optional AgentConfig for execution parameters.	None

Returns:

Type	Description
AgentResult	AgentResult with final response, traces, usage, and metadata.

Raises:

Type	Description
AgentExecutionError	If the agent fails during execution.
AgentTimeoutError	If execution exceeds the timeout.
AgentResponseError	If the response is malformed or invalid.

Source code in src/karenina/adapters/langchain_deep_agents/agent.py

async def arun(
    self,
    messages: list[Message],
    tools: list[Tool] | None = None,
    mcp_servers: dict[str, MCPServerConfig] | None = None,
    config: AgentConfig | None = None,
    _partial_result_store: _PartialResultStore | None = None,
) -> AgentResult:
    """Execute an agent loop with optional tools and MCP servers.

    Args:
        messages: Initial conversation messages.
        tools: Optional list of Tool definitions the agent can invoke.
        mcp_servers: Optional dict of MCP server configurations.
        config: Optional AgentConfig for execution parameters.

    Returns:
        AgentResult with final response, traces, usage, and metadata.

    Raises:
        AgentExecutionError: If the agent fails during execution.
        AgentTimeoutError: If execution exceeds the timeout.
        AgentResponseError: If the response is malformed or invalid.
    """
    global _create_deep_agent  # noqa: PLW0603
    if _create_deep_agent is None:
        from deepagents import create_deep_agent as _cda

        _create_deep_agent = _cda

    config = config or AgentConfig()

    # Convert messages to prompt string and extract system prompt
    prompt_string = self._converter.to_prompt_string(messages)
    system_prompt = self._converter.extract_system_prompt(messages)

    # Use config system_prompt as fallback
    if not system_prompt and config.system_prompt:
        system_prompt = config.system_prompt
    elif not system_prompt and self._config.system_prompt:
        system_prompt = self._config.system_prompt

    # Initialize model. Agent loops need SDK-level retries at the model
    # call boundary; retrying the whole agent can repeat workspace writes.
    retry_policy = self._config.retry_policy or RetryPolicy()
    chat_model = create_chat_model(
        self._config,
        max_retries=retry_policy.derive_sdk_max_retries(),
    )

    # Build agent kwargs
    agent_kwargs: dict[str, Any] = {"model": chat_model}
    if system_prompt:
        agent_kwargs["system_prompt"] = system_prompt

    agent_kwargs["backend"] = self._build_backend(config.workspace_path)

    # Pass through extra config to create_deep_agent
    if config.extra:
        for key, value in config.extra.items():
            if key not in ("model", "system_prompt", "backend"):
                agent_kwargs[key] = value

    # Convert MCP servers to LangChain tools and combine with explicit tools
    all_tools: list[Any] = []
    if tools:
        all_tools.extend(tools)

    # Use AsyncExitStack for persistent MCP sessions. Sessions stay alive
    # for all tool calls during agent execution. Exceptions are captured
    # inside the block and re-raised after clean exit, because MCP session
    # cleanup can wrap errors in ExceptionGroup if an exception propagates
    # through the exit stack.
    result: dict[str, Any] = {}
    subgraph_results: dict[tuple[Any, ...], dict[str, Any]] = {}
    limit_reached = False
    timeout_reached = False
    deferred_error: Exception | None = None

    async with AsyncExitStack() as exit_stack:
        # Convert MCP servers to LangChain tools
        if mcp_servers:
            from .mcp import convert_mcp_to_tools

            try:
                mcp_tools = await convert_mcp_to_tools(mcp_servers, exit_stack)
                all_tools.extend(mcp_tools)
                logger.info(
                    "Loaded %d MCP tools from %d servers",
                    len(mcp_tools),
                    len(mcp_servers),
                )
            except Exception as e:
                logger.warning("Failed to load MCP tools: %s", e)
                deferred_error = AgentExecutionError(f"Failed to initialize MCP tools: {e}")
                deferred_error.__cause__ = e

        if deferred_error is None:
            # Pass tools to agent if any were collected
            if all_tools:
                agent_kwargs["tools"] = all_tools

            # Create the agent
            agent = _create_deep_agent(**agent_kwargs)

            # Build invocation input
            invoke_input: dict[str, Any] = {
                "messages": [{"role": "user", "content": prompt_string}],
            }

            # LangGraph config for recursion limit
            # Each tool call + response = 2 steps, so double max_turns
            langgraph_config: RunnableConfig = {
                "recursion_limit": config.max_turns * 2,
            }

            # Execute agent
            async def execute_agent() -> None:
                nonlocal result, subgraph_results, limit_reached
                async for chunk in agent.astream(
                    invoke_input,
                    config=langgraph_config,
                    stream_mode="values",
                    subgraphs=True,
                ):
                    namespace: tuple[Any, ...] = ()
                    state = chunk
                    if isinstance(chunk, tuple) and len(chunk) == 2:
                        namespace_raw, state = chunk
                        namespace = tuple(namespace_raw)
                    if isinstance(state, dict):
                        if namespace:
                            subgraph_results[namespace] = state
                        else:
                            result = state
                if result.get("is_last_step", False):
                    limit_reached = True

            try:
                if config.timeout:
                    await asyncio.wait_for(execute_agent(), timeout=config.timeout)
                else:
                    await execute_agent()

            except TimeoutError:
                timeout_reached = True
                logger.warning("Agent timed out after %ss; returning partial trace", config.timeout)
                if _partial_result_store is not None:
                    partial_lc_messages = result.get("messages", [])
                    raw_trace = self._build_raw_trace(
                        partial_lc_messages,
                        subgraph_results,
                        limit_reached=limit_reached,
                        timeout_reached=True,
                    )
                    if partial_lc_messages:
                        partial_result = AgentResult(
                            final_response=self._extract_final_response(partial_lc_messages),
                            raw_trace=raw_trace,
                            trace_messages=self._converter.from_provider(partial_lc_messages),
                            usage=extract_deep_agents_usage(partial_lc_messages, model=self._config.model_name),
                            turns=self._count_turns(partial_lc_messages),
                            limit_reached=limit_reached,
                            session_id=None,
                            actual_model=extract_actual_model(partial_lc_messages) or self._config.model_name,
                            timeout_reached=True,
                        )
                    else:
                        partial_result = AgentResult(
                            final_response="[Agent stopped before producing a final response]",
                            raw_trace=raw_trace or "[Note: Agent stopped before producing trace messages]",
                            trace_messages=[],
                            usage=UsageMetadata(model=self._config.model_name),
                            turns=0,
                            limit_reached=limit_reached,
                            session_id=None,
                            actual_model=self._config.model_name,
                            timeout_reached=True,
                        )
                    _partial_result_store.set(partial_result)
            except Exception as e:
                mapped_error, was_limit = wrap_deep_agents_error(e)
                if was_limit:
                    limit_reached = True
                    logger.warning("Agent hit turn limit: %s", e)
                else:
                    deferred_error = mapped_error
                    deferred_error.__cause__ = e

    # exit_stack closed: MCP sessions cleaned up

    if deferred_error is not None:
        raise deferred_error

    # Extract messages from result
    lc_messages: list[Any] = result.get("messages", [])

    if not lc_messages and not limit_reached and not timeout_reached:
        raise AgentResponseError("No messages received from Deep Agents")

    # If limit was reached but no messages, return a partial result
    if not lc_messages and (limit_reached or timeout_reached):
        raw_trace = self._build_raw_trace(
            lc_messages,
            subgraph_results,
            limit_reached=limit_reached,
            timeout_reached=timeout_reached,
        )
        return AgentResult(
            final_response="[Agent stopped before producing a final response]",
            raw_trace=raw_trace or "[Note: Agent stopped before producing trace messages]",
            trace_messages=[],
            usage=UsageMetadata(model=self._config.model_name),
            turns=0,
            limit_reached=limit_reached,
            session_id=None,
            actual_model=self._config.model_name,
            timeout_reached=timeout_reached,
        )

    # Build raw_trace (legacy string format)
    raw_trace = self._build_raw_trace(
        lc_messages,
        subgraph_results,
        limit_reached=limit_reached,
        timeout_reached=timeout_reached,
    )

    # Build trace_messages (structured format)
    trace_messages = self._converter.from_provider(lc_messages)

    # Extract final response
    final_response = self._extract_final_response(lc_messages)

    # Extract usage
    usage = extract_deep_agents_usage(lc_messages, model=self._config.model_name)

    # Count turns
    turns = self._count_turns(lc_messages)

    # Extract actual model
    actual_model = extract_actual_model(lc_messages) or self._config.model_name

    return AgentResult(
        final_response=final_response,
        raw_trace=raw_trace,
        trace_messages=trace_messages,
        usage=usage,
        turns=turns,
        limit_reached=limit_reached,
        session_id=None,
        actual_model=actual_model,
        timeout_reached=timeout_reached,
    )

run

run(
    messages: list[Message],
    tools: list[Tool] | None = None,
    mcp_servers: dict[str, MCPServerConfig] | None = None,
    config: AgentConfig | None = None,
) -> AgentResult

Synchronous wrapper for arun().

Parameters:

Name	Type	Description	Default
messages	list[Message]	Initial conversation messages.	required
tools	list[Tool] | None	Optional list of Tool definitions.	None
mcp_servers	dict[str, MCPServerConfig] | None	Optional MCP server configurations.	None
config	AgentConfig | None	Optional AgentConfig for execution parameters.	None

Returns:

Type	Description
AgentResult	AgentResult from the agent execution.

Raises:

Type	Description
AgentExecutionError	If the agent fails during execution.
AgentTimeoutError	If execution exceeds the timeout.
AgentResponseError	If the response is malformed or invalid.

Source code in src/karenina/adapters/langchain_deep_agents/agent.py

def run(
    self,
    messages: list[Message],
    tools: list[Tool] | None = None,
    mcp_servers: dict[str, MCPServerConfig] | None = None,
    config: AgentConfig | None = None,
) -> AgentResult:
    """Synchronous wrapper for arun().

    Args:
        messages: Initial conversation messages.
        tools: Optional list of Tool definitions.
        mcp_servers: Optional MCP server configurations.
        config: Optional AgentConfig for execution parameters.

    Returns:
        AgentResult from the agent execution.

    Raises:
        AgentExecutionError: If the agent fails during execution.
        AgentTimeoutError: If execution exceeds the timeout.
        AgentResponseError: If the response is malformed or invalid.
    """
    timeout = config.timeout if config and config.timeout else 600
    partial_result_store = _PartialResultStore()
    return cast(
        AgentResult,
        _run_in_fresh_loop(
            self.arun,
            messages,
            tools,
            mcp_servers,
            config,
            partial_result_store,
            timeout=timeout,
            timeout_result=partial_result_store.get,
        ),
    )

DeepAgentsLLMAdapter

LLM adapter using LangChain's init_chat_model for single-turn calls.

This adapter implements the LLMPort Protocol for simple LLM invocation without agent loops. Uses the LangChain model directly for efficiency.

Example

config = ModelConfig( ... id="test", ... model_name="claude-sonnet-4-20250514", ... model_provider="anthropic", ... interface="langchain_deep_agents", ... ) adapter = DeepAgentsLLMAdapter(config) response = await adapter.ainvoke([Message.user("Hello!")]) print(response.content)

Source code in src/karenina/adapters/langchain_deep_agents/llm.py

class DeepAgentsLLMAdapter:
    """LLM adapter using LangChain's init_chat_model for single-turn calls.

    This adapter implements the LLMPort Protocol for simple LLM invocation
    without agent loops. Uses the LangChain model directly for efficiency.

    Example:
        >>> config = ModelConfig(
        ...     id="test",
        ...     model_name="claude-sonnet-4-20250514",
        ...     model_provider="anthropic",
        ...     interface="langchain_deep_agents",
        ... )
        >>> adapter = DeepAgentsLLMAdapter(config)
        >>> response = await adapter.ainvoke([Message.user("Hello!")])
        >>> print(response.content)
    """

    def __init__(
        self,
        model_config: ModelConfig,
        *,
        _structured_schema: type[BaseModel] | None = None,
    ) -> None:
        """Initialize the Deep Agents LLM adapter.

        Args:
            model_config: Configuration specifying model, provider, and interface.
            _structured_schema: Internal; schema for structured output mode.
        """
        self._config = model_config
        self._converter = DeepAgentsMessageConverter()
        self._structured_schema = _structured_schema

        retry_policy = model_config.retry_policy or RetryPolicy()
        self._retry_executor = RetryExecutor(retry_policy, ErrorRegistry())

    @property
    def capabilities(self) -> PortCapabilities:
        """Declare adapter capabilities.

        Returns:
            PortCapabilities with system_prompt, structured_output, and streaming support.
        """
        return PortCapabilities(
            supports_system_prompt=True,
            supports_structured_output=True,
            supports_streaming=True,
        )

    async def ainvoke(self, messages: list[Message]) -> LLMResponse:
        """Invoke the LLM asynchronously with automatic retry for transient errors.

        Converts karenina Messages to LangChain format, invokes the model
        through RetryExecutor (per-category retry budgets for connection,
        timeout, rate limit, and server errors), and converts the response
        back.

        Args:
            messages: List of messages forming the conversation.

        Returns:
            LLMResponse containing the generated content and usage metadata.

        Raises:
            The last transient exception if all retries are exhausted, or
            immediately for permanent errors.
        """
        from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

        # Build LangChain message list
        lc_messages: list[Any] = []
        for msg in messages:
            text = msg.text or ""
            if msg.role.value == "system":
                lc_messages.append(SystemMessage(content=text))
            elif msg.role.value == "user":
                lc_messages.append(HumanMessage(content=text))
            elif msg.role.value == "assistant":
                lc_messages.append(AIMessage(content=text))

        # Create the model once, outside the retried callable, so that
        # retries repeat the API call rather than model construction.
        chat_model = create_chat_model(self._config)

        # Apply structured output if configured
        if self._structured_schema is not None:
            chat_model = chat_model.with_structured_output(self._structured_schema)

        # Invoke through RetryExecutor with a wall-clock guard per attempt.
        # GlobalLLMLimiter gating happens per attempt inside
        # _ainvoke_with_timeout, so retry backoff never holds a permit.
        # Sync invoke() reaches the wire only through this method, so it
        # is covered without a sync acquire.
        response = await self._retry_executor.aexecute_with_timeout(
            self._ainvoke_with_timeout,
            chat_model,
            lc_messages,
            timeout=self._config.request_timeout,
        )

        # Extract content
        if self._structured_schema is not None:
            # Structured output returns a Pydantic model or dict
            if isinstance(response, BaseModel):
                content = response.model_dump_json()
            elif isinstance(response, dict):
                import json

                content = json.dumps(response)
            else:
                content = str(response)
        elif isinstance(response, AIMessage):
            content = response.content if isinstance(response.content, str) else str(response.content)
        else:
            content = str(response)

        # Extract usage
        usage = UsageMetadata(model=self._config.model_name)
        if isinstance(response, AIMessage):
            usage_meta = getattr(response, "usage_metadata", None)
            if usage_meta and isinstance(usage_meta, dict):
                usage = UsageMetadata(
                    input_tokens=usage_meta.get("input_tokens", 0),
                    output_tokens=usage_meta.get("output_tokens", 0),
                    total_tokens=usage_meta.get("input_tokens", 0) + usage_meta.get("output_tokens", 0),
                    model=self._config.model_name,
                )

        return LLMResponse(content=content, usage=usage, raw=response)

    async def _ainvoke_with_timeout(
        self,
        model: Any,
        lc_messages: list[Any],
        *,
        timeout: float | None = None,
    ) -> Any:
        """Call ``model.ainvoke`` under a wall-clock timeout as a guardrail.

        LangChain's ``ainvoke`` can stall internally without ever raising,
        and the httpx ``request_timeout`` does not catch every such case,
        so this karenina-layer ``asyncio.wait_for`` enforces a hard
        per-attempt wall-clock budget. Mirrors the langchain adapter.

        A fired timeout raises a stock ``asyncio.TimeoutError``, which
        ``ErrorRegistry`` classifies as ``TIMEOUT`` via the built-in MRO
        check. The configured timeout retry budget then applies inside
        ``RetryExecutor.aexecute_with_timeout``. Note that
        ``RetryPolicy.timeout_escalation`` only extends this guard on
        TIMEOUT retries. The underlying SDK client's own timeout stays
        pinned at ``request_timeout`` from construction, so the SDK may
        cut the request before an escalated guard fires.

        When ``timeout`` is None, falls back to ``self._config.request_timeout``.
        When that is also None, the call is made without any wrapper.

        Each attempt holds one GlobalLLMLimiter permit for the wire call
        only (uniform per-attempt policy), so retry backoff sleeps never
        hold a permit. The permit wait itself is NOT bounded by the
        attempt timeout: a saturated limiter delays the attempt rather
        than timing it out, matching the legacy semaphore semantics.

        Args:
            model: The LangChain model exposing ``ainvoke``.
            lc_messages: Provider-formatted messages to send.
            timeout: Optional per-call wall-clock timeout in seconds.
                When None, falls back to ``self._config.request_timeout``.

        Returns:
            The raw model response object.

        Raises:
            asyncio.TimeoutError: If the call exceeds the effective timeout.
        """
        async with get_global_llm_limiter().borrow():
            if timeout is None:
                timeout = self._config.request_timeout
            if timeout is None:
                return await model.ainvoke(lc_messages)
            return await asyncio.wait_for(model.ainvoke(lc_messages), timeout=timeout)

    @with_llm_semaphore
    def invoke(self, messages: list[Message]) -> LLMResponse:
        """Invoke the LLM synchronously.

        Args:
            messages: List of messages forming the conversation.

        Returns:
            LLMResponse containing the generated content and usage metadata.
        """
        from karenina.benchmark.verification.executor import get_async_portal

        portal = get_async_portal()
        if portal is not None:
            return portal.call(self.ainvoke, messages)

        try:
            asyncio.get_running_loop()
            # Fresh thread with the caller's context propagated, so
            # track_retries telemetry survives the dispatch.
            thread_timeout = compute_sync_wrapper_timeout(
                self._config.request_timeout,
                floor=DEEP_AGENTS_SYNC_WRAPPER_FLOOR,
                retry_policy=self._config.retry_policy,
            )
            return run_coro_in_thread(self.ainvoke, messages, timeout=thread_timeout)

        except RuntimeError:
            return asyncio.run(self.ainvoke(messages))

    def with_structured_output(
        self,
        schema: type[BaseModel],
        *,
        max_retries: int | None = None,
    ) -> DeepAgentsLLMAdapter:
        """Return a new adapter configured for structured output.

        Args:
            schema: A Pydantic model class defining the output structure.
            max_retries: Not supported by this adapter. A warning is emitted
                if a non-None value is provided. See
                karenina.ports.llm.LLMPort.with_structured_output for the
                cross-adapter contract (LangChain and Claude Tool respect
                max_retries, this adapter and claude_agent_sdk warn and
                ignore).

        Returns:
            A new DeepAgentsLLMAdapter configured with the schema.
        """
        if max_retries is not None:
            logger.warning(
                "max_retries=%d ignored by langchain_deep_agents adapter. "
                "Validation-feedback retries are not implemented here. "
                "Transient errors are retried via RetryExecutor regardless.",
                max_retries,
            )
        return DeepAgentsLLMAdapter(
            self._config,
            _structured_schema=schema,
        )

    def astream(self, messages: list[Message]) -> AbstractAsyncContextManager[StreamingLLMResponse]:
        """Stream LLM response using LangChain's cross-provider astream.

        Stream ESTABLISHMENT (opening the LangChain astream and fetching
        the first chunk) routes through RetryExecutor, so transient
        connection-setup failures (including langchain-openai's cross
        event loop pooled-connection errors) retry with per-category
        budgets. Each retry recreates the stream from scratch. Once
        established, the stream is never retried mid-flight.

        Args:
            messages: List of unified Message objects.

        Returns:
            Async context manager yielding a StreamingLLMResponse that
            can be iterated for text chunks.
        """
        return self._astream_impl(messages, establishment_retry=True)

    @asynccontextmanager
    async def _astream_impl(
        self,
        messages: list[Message],
        *,
        establishment_retry: bool,
    ) -> AsyncIterator[StreamingLLMResponse]:  # noqa: ANN202
        """Shared astream body with switchable establishment retries.

        The public astream keeps establishment retries on. The
        stream_invoke path (via _astream_with_timeout) turns them off
        because its outer RetryExecutor already retries the whole
        attempt, and nesting the two would amplify worst-case connection
        attempts multiplicatively.
        """
        from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

        lc_messages: list[Any] = []
        for msg in messages:
            text = msg.text or ""
            if msg.role.value == "system":
                lc_messages.append(SystemMessage(content=text))
            elif msg.role.value == "user":
                lc_messages.append(HumanMessage(content=text))
            elif msg.role.value == "assistant":
                lc_messages.append(AIMessage(content=text))

        chat_model = create_chat_model(self._config)
        response = StreamingLLMResponse()

        async def _establish(*, timeout: float | None = None) -> tuple[AsyncIterator[Any], Any | None]:
            """One stream-establishment attempt: open astream, await the first chunk.

            Each attempt holds one GlobalLLMLimiter permit for the
            establishment only, released before the stream is consumed:
            max_concurrent_requests caps concurrent request setups, not
            concurrent open streams. Borrowing per attempt (instead of
            around the retry loop) releases the permit during backoff.
            """
            async with get_global_llm_limiter().borrow():
                iterator = chat_model.astream(lc_messages).__aiter__()
                try:
                    if timeout is None:
                        first = await iterator.__anext__()
                    else:
                        first = await asyncio.wait_for(iterator.__anext__(), timeout=timeout)
                except StopAsyncIteration:
                    return iterator, None
                except BaseException:
                    # Close the half-opened stream before surfacing the failure
                    # (or cancellation) so no "async generator was never closed"
                    # debris is left behind.
                    aclose = getattr(iterator, "aclose", None)
                    if aclose is not None:
                        with contextlib.suppress(Exception):
                            await aclose()
                    raise
                return iterator, first

        if establishment_retry:
            iterator, first_chunk = await self._retry_executor.aexecute_with_timeout(
                _establish,
                timeout=self._config.request_timeout,
            )
        else:
            iterator, first_chunk = await _establish(timeout=self._config.request_timeout)

        def _handle_chunk(chunk: Any) -> str:
            """Extract text and capture usage metadata from a chunk."""
            text = chunk.content if isinstance(chunk.content, str) else ""
            if hasattr(chunk, "usage_metadata") and chunk.usage_metadata:
                response.usage = extract_usage_from_chunk(chunk, model_name=self._config.model_name)
            return text

        async def _chunk_generator() -> AsyncIterator[str]:  # noqa: ANN202
            """Yield text chunks from LangChain's astream, extracting usage from final chunk."""
            if first_chunk is not None:
                text = _handle_chunk(first_chunk)
                if text:
                    yield text
            async for chunk in iterator:
                text = _handle_chunk(chunk)
                if text:
                    yield text

        response._set_chunk_source(_chunk_generator())
        yield response
        response.is_complete = True

    async def _astream_with_timeout(self, messages: list[Message], timeout: float | None) -> LLMResponse:
        """Stream with wall-clock timeout, returning accumulated content.

        The timeout window covers stream ESTABLISHMENT as well as the
        drain: the stream is opened inside asyncio.timeout, so an
        establishment stall surfaces as StreamingTimeoutError (with empty
        partial content) within about ``timeout`` seconds even when
        request_timeout is unset. Establishment retries are disabled here
        because the caller (stream_invoke) already wraps each attempt in
        RetryExecutor.

        Args:
            messages: List of unified Message objects.
            timeout: Wall-clock timeout in seconds. None means no timeout.

        Returns:
            LLMResponse with accumulated content.

        Raises:
            StreamingTimeoutError: If establishment plus drain exceed the
                wall-clock timeout.
        """
        from karenina.exceptions import StreamingTimeoutError

        stream_cm = self._astream_impl(messages, establishment_retry=False)
        sr: StreamingLLMResponse | None = None
        try:
            try:
                async with asyncio.timeout(timeout):
                    sr = await stream_cm.__aenter__()
                    async for _chunk in sr:
                        pass
            except TimeoutError:
                raise StreamingTimeoutError(
                    f"Streaming timed out after {timeout}s",
                    partial_content=sr.accumulated_content if sr is not None else "",
                ) from None
        finally:
            if sr is not None:
                await stream_cm.__aexit__(None, None, None)

        return LLMResponse(
            content=sr.accumulated_content,
            usage=sr.usage,
            raw=None,
        )

    @with_llm_semaphore
    def stream_invoke(self, messages: list[Message], timeout: float | None = None) -> LLMResponse:
        """Stream with wall-clock timeout synchronously.

        Uses streaming internally so that partial content can be captured on
        timeout. Returns the same LLMResponse type as invoke(). Retries via
        RetryExecutor on transient errors (including StreamingTimeoutError
        with zero content, classified as RATE_LIMIT for queue congestion).
        The inner stream-establishment retry used by the bare astream path
        is disabled on this path, so this outer executor is the single
        retry layer and connection attempts are not amplified
        multiplicatively.

        Args:
            messages: List of unified Message objects.
            timeout: Wall-clock timeout in seconds. None means no timeout.

        Returns:
            LLMResponse with accumulated content.

        Raises:
            StreamingTimeoutError: If retries are exhausted and the stream
                still exceeds the wall-clock timeout.
        """
        result: LLMResponse = self._retry_executor.execute_with_timeout(
            self._stream_invoke_once, messages, timeout=timeout
        )
        return result

    def _stream_invoke_once(self, messages: list[Message], timeout: float | None = None) -> LLMResponse:
        """Single stream_invoke attempt (no retry). Called by RetryExecutor."""
        from karenina.benchmark.verification.executor import get_async_portal

        portal = get_async_portal()

        if portal is not None:
            return portal.call(self._astream_with_timeout, messages, timeout)

        try:
            asyncio.get_running_loop()
            return run_coro_in_thread(
                self._astream_with_timeout,
                messages,
                timeout,
                timeout=(timeout or 300) + 30,
            )

        except RuntimeError:
            return asyncio.run(self._astream_with_timeout(messages, timeout))

    async def aclose(self) -> None:
        """Close underlying resources.

        No resources to clean up: the LangChain model is created fresh
        per ainvoke() call. Provided for interface consistency with other
        adapters.
        """

Attributes

capabilities property

capabilities: PortCapabilities

Declare adapter capabilities.

Returns:

Type	Description
PortCapabilities	PortCapabilities with system_prompt, structured_output, and streaming support.

Functions

__init__

__init__(
    model_config: ModelConfig,
    *,
    _structured_schema: type[BaseModel] | None = None,
) -> None

Parameters:

Name	Type	Description	Default
model_config	ModelConfig	Configuration specifying model, provider, and interface.	required
_structured_schema	type[BaseModel] | None	Internal; schema for structured output mode.	None

Source code in src/karenina/adapters/langchain_deep_agents/llm.py

def __init__(
    self,
    model_config: ModelConfig,
    *,
    _structured_schema: type[BaseModel] | None = None,
) -> None:
    """Initialize the Deep Agents LLM adapter.

    Args:
        model_config: Configuration specifying model, provider, and interface.
        _structured_schema: Internal; schema for structured output mode.
    """
    self._config = model_config
    self._converter = DeepAgentsMessageConverter()
    self._structured_schema = _structured_schema

    retry_policy = model_config.retry_policy or RetryPolicy()
    self._retry_executor = RetryExecutor(retry_policy, ErrorRegistry())

aclose async

aclose() -> None

Close underlying resources.

No resources to clean up: the LangChain model is created fresh per ainvoke() call. Provided for interface consistency with other adapters.

Source code in src/karenina/adapters/langchain_deep_agents/llm.py

async def aclose(self) -> None:
    """Close underlying resources.

    No resources to clean up: the LangChain model is created fresh
    per ainvoke() call. Provided for interface consistency with other
    adapters.
    """

ainvoke async

ainvoke(messages: list[Message]) -> LLMResponse

Invoke the LLM asynchronously with automatic retry for transient errors.

Converts karenina Messages to LangChain format, invokes the model through RetryExecutor (per-category retry budgets for connection, timeout, rate limit, and server errors), and converts the response back.

Parameters:

Name	Type	Description	Default
messages	list[Message]	List of messages forming the conversation.	required

Returns:

Type	Description
LLMResponse	LLMResponse containing the generated content and usage metadata.

Source code in src/karenina/adapters/langchain_deep_agents/llm.py

async def ainvoke(self, messages: list[Message]) -> LLMResponse:
    """Invoke the LLM asynchronously with automatic retry for transient errors.

    Converts karenina Messages to LangChain format, invokes the model
    through RetryExecutor (per-category retry budgets for connection,
    timeout, rate limit, and server errors), and converts the response
    back.

    Args:
        messages: List of messages forming the conversation.

    Returns:
        LLMResponse containing the generated content and usage metadata.

    Raises:
        The last transient exception if all retries are exhausted, or
        immediately for permanent errors.
    """
    from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

    # Build LangChain message list
    lc_messages: list[Any] = []
    for msg in messages:
        text = msg.text or ""
        if msg.role.value == "system":
            lc_messages.append(SystemMessage(content=text))
        elif msg.role.value == "user":
            lc_messages.append(HumanMessage(content=text))
        elif msg.role.value == "assistant":
            lc_messages.append(AIMessage(content=text))

    # Create the model once, outside the retried callable, so that
    # retries repeat the API call rather than model construction.
    chat_model = create_chat_model(self._config)

    # Apply structured output if configured
    if self._structured_schema is not None:
        chat_model = chat_model.with_structured_output(self._structured_schema)

    # Invoke through RetryExecutor with a wall-clock guard per attempt.
    # GlobalLLMLimiter gating happens per attempt inside
    # _ainvoke_with_timeout, so retry backoff never holds a permit.
    # Sync invoke() reaches the wire only through this method, so it
    # is covered without a sync acquire.
    response = await self._retry_executor.aexecute_with_timeout(
        self._ainvoke_with_timeout,
        chat_model,
        lc_messages,
        timeout=self._config.request_timeout,
    )

    # Extract content
    if self._structured_schema is not None:
        # Structured output returns a Pydantic model or dict
        if isinstance(response, BaseModel):
            content = response.model_dump_json()
        elif isinstance(response, dict):
            import json

            content = json.dumps(response)
        else:
            content = str(response)
    elif isinstance(response, AIMessage):
        content = response.content if isinstance(response.content, str) else str(response.content)
    else:
        content = str(response)

    # Extract usage
    usage = UsageMetadata(model=self._config.model_name)
    if isinstance(response, AIMessage):
        usage_meta = getattr(response, "usage_metadata", None)
        if usage_meta and isinstance(usage_meta, dict):
            usage = UsageMetadata(
                input_tokens=usage_meta.get("input_tokens", 0),
                output_tokens=usage_meta.get("output_tokens", 0),
                total_tokens=usage_meta.get("input_tokens", 0) + usage_meta.get("output_tokens", 0),
                model=self._config.model_name,
            )

    return LLMResponse(content=content, usage=usage, raw=response)

astream

astream(
    messages: list[Message],
) -> AbstractAsyncContextManager[StreamingLLMResponse]

Stream LLM response using LangChain's cross-provider astream.

Stream ESTABLISHMENT (opening the LangChain astream and fetching the first chunk) routes through RetryExecutor, so transient connection-setup failures (including langchain-openai's cross event loop pooled-connection errors) retry with per-category budgets. Each retry recreates the stream from scratch. Once established, the stream is never retried mid-flight.

Parameters:

Name	Type	Description	Default
messages	list[Message]	List of unified Message objects.	required

Returns:

Type	Description
AbstractAsyncContextManager[StreamingLLMResponse]	Async context manager yielding a StreamingLLMResponse that
AbstractAsyncContextManager[StreamingLLMResponse]	can be iterated for text chunks.

Source code in src/karenina/adapters/langchain_deep_agents/llm.py

def astream(self, messages: list[Message]) -> AbstractAsyncContextManager[StreamingLLMResponse]:
    """Stream LLM response using LangChain's cross-provider astream.

    Stream ESTABLISHMENT (opening the LangChain astream and fetching
    the first chunk) routes through RetryExecutor, so transient
    connection-setup failures (including langchain-openai's cross
    event loop pooled-connection errors) retry with per-category
    budgets. Each retry recreates the stream from scratch. Once
    established, the stream is never retried mid-flight.

    Args:
        messages: List of unified Message objects.

    Returns:
        Async context manager yielding a StreamingLLMResponse that
        can be iterated for text chunks.
    """
    return self._astream_impl(messages, establishment_retry=True)

invoke

invoke(messages: list[Message]) -> LLMResponse

Invoke the LLM synchronously.

Parameters:

Name	Type	Description	Default
messages	list[Message]	List of messages forming the conversation.	required

Returns:

Type	Description
LLMResponse	LLMResponse containing the generated content and usage metadata.

Source code in src/karenina/adapters/langchain_deep_agents/llm.py

@with_llm_semaphore
def invoke(self, messages: list[Message]) -> LLMResponse:
    """Invoke the LLM synchronously.

    Args:
        messages: List of messages forming the conversation.

    Returns:
        LLMResponse containing the generated content and usage metadata.
    """
    from karenina.benchmark.verification.executor import get_async_portal

    portal = get_async_portal()
    if portal is not None:
        return portal.call(self.ainvoke, messages)

    try:
        asyncio.get_running_loop()
        # Fresh thread with the caller's context propagated, so
        # track_retries telemetry survives the dispatch.
        thread_timeout = compute_sync_wrapper_timeout(
            self._config.request_timeout,
            floor=DEEP_AGENTS_SYNC_WRAPPER_FLOOR,
            retry_policy=self._config.retry_policy,
        )
        return run_coro_in_thread(self.ainvoke, messages, timeout=thread_timeout)

    except RuntimeError:
        return asyncio.run(self.ainvoke(messages))

stream_invoke

stream_invoke(
    messages: list[Message], timeout: float | None = None
) -> LLMResponse

Stream with wall-clock timeout synchronously.

Uses streaming internally so that partial content can be captured on timeout. Returns the same LLMResponse type as invoke(). Retries via RetryExecutor on transient errors (including StreamingTimeoutError with zero content, classified as RATE_LIMIT for queue congestion). The inner stream-establishment retry used by the bare astream path is disabled on this path, so this outer executor is the single retry layer and connection attempts are not amplified multiplicatively.

Parameters:

Name	Type	Description	Default
messages	list[Message]	List of unified Message objects.	required
timeout	float | None	Wall-clock timeout in seconds. None means no timeout.	None

Returns:

Type	Description
LLMResponse	LLMResponse with accumulated content.

Raises:

Type	Description
StreamingTimeoutError	If retries are exhausted and the stream still exceeds the wall-clock timeout.

Source code in src/karenina/adapters/langchain_deep_agents/llm.py

@with_llm_semaphore
def stream_invoke(self, messages: list[Message], timeout: float | None = None) -> LLMResponse:
    """Stream with wall-clock timeout synchronously.

    Uses streaming internally so that partial content can be captured on
    timeout. Returns the same LLMResponse type as invoke(). Retries via
    RetryExecutor on transient errors (including StreamingTimeoutError
    with zero content, classified as RATE_LIMIT for queue congestion).
    The inner stream-establishment retry used by the bare astream path
    is disabled on this path, so this outer executor is the single
    retry layer and connection attempts are not amplified
    multiplicatively.

    Args:
        messages: List of unified Message objects.
        timeout: Wall-clock timeout in seconds. None means no timeout.

    Returns:
        LLMResponse with accumulated content.

    Raises:
        StreamingTimeoutError: If retries are exhausted and the stream
            still exceeds the wall-clock timeout.
    """
    result: LLMResponse = self._retry_executor.execute_with_timeout(
        self._stream_invoke_once, messages, timeout=timeout
    )
    return result

with_structured_output

with_structured_output(
    schema: type[BaseModel],
    *,
    max_retries: int | None = None,
) -> DeepAgentsLLMAdapter

Return a new adapter configured for structured output.

Parameters:

Name	Type	Description	Default
schema	type[BaseModel]	A Pydantic model class defining the output structure.	required
max_retries	int | None	Not supported by this adapter. A warning is emitted if a non-None value is provided. See karenina.ports.llm.LLMPort.with_structured_output for the cross-adapter contract (LangChain and Claude Tool respect max_retries, this adapter and claude_agent_sdk warn and ignore).	None

Returns:

Type	Description
DeepAgentsLLMAdapter	A new DeepAgentsLLMAdapter configured with the schema.

Source code in src/karenina/adapters/langchain_deep_agents/llm.py

def with_structured_output(
    self,
    schema: type[BaseModel],
    *,
    max_retries: int | None = None,
) -> DeepAgentsLLMAdapter:
    """Return a new adapter configured for structured output.

    Args:
        schema: A Pydantic model class defining the output structure.
        max_retries: Not supported by this adapter. A warning is emitted
            if a non-None value is provided. See
            karenina.ports.llm.LLMPort.with_structured_output for the
            cross-adapter contract (LangChain and Claude Tool respect
            max_retries, this adapter and claude_agent_sdk warn and
            ignore).

    Returns:
        A new DeepAgentsLLMAdapter configured with the schema.
    """
    if max_retries is not None:
        logger.warning(
            "max_retries=%d ignored by langchain_deep_agents adapter. "
            "Validation-feedback retries are not implemented here. "
            "Transient errors are retried via RetryExecutor regardless.",
            max_retries,
        )
    return DeepAgentsLLMAdapter(
        self._config,
        _structured_schema=schema,
    )

DeepAgentsMessageConverter

Convert between karenina's unified Message and LangGraph message types.

Deep Agents accepts messages as dicts with role/content keys for invocation, and returns LangGraph BaseMessage subclasses in results.

Source code in src/karenina/adapters/langchain_deep_agents/messages.py

class DeepAgentsMessageConverter:
    """Convert between karenina's unified Message and LangGraph message types.

    Deep Agents accepts messages as dicts with role/content keys for invocation,
    and returns LangGraph BaseMessage subclasses in results.
    """

    def to_prompt_string(self, messages: list[Message]) -> str:
        """Convert user/assistant messages to a prompt string.

        System messages are excluded (use extract_system_prompt instead).

        Args:
            messages: List of karenina Message objects.

        Returns:
            Concatenated prompt string from non-system messages.
        """
        parts = []
        for msg in messages:
            if msg.role == Role.SYSTEM:
                continue
            text = msg.text
            if text:
                parts.append(text)
        return "\n\n".join(parts)

    def extract_system_prompt(self, messages: list[Message]) -> str | None:
        """Extract system prompt from messages.

        Args:
            messages: List of karenina Message objects.

        Returns:
            Combined system prompt text, or None if no system messages.
        """
        system_parts = []
        for msg in messages:
            if msg.role == Role.SYSTEM:
                text = msg.text
                if text:
                    system_parts.append(text)
        return "\n\n".join(system_parts) if system_parts else None

    def to_langchain_messages(self, messages: list[Message]) -> list[dict[str, str]]:
        """Convert karenina messages to LangGraph-compatible dicts.

        Args:
            messages: List of karenina Message objects.

        Returns:
            List of message dicts with role and content keys.
        """
        result = []
        for msg in messages:
            if msg.role == Role.SYSTEM:
                continue
            role_map = {
                Role.USER: "user",
                Role.ASSISTANT: "assistant",
                Role.TOOL: "tool",
            }
            role = role_map.get(msg.role, "user")
            result.append({"role": role, "content": msg.text or ""})
        return result

    def from_provider(self, lc_messages: list[Any]) -> list[Message]:
        """Convert LangGraph BaseMessage list to karenina Messages.

        Args:
            lc_messages: List of LangGraph message objects.

        Returns:
            List of karenina Message objects.
        """
        from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage

        result: list[Message] = []

        for msg in lc_messages:
            if isinstance(msg, SystemMessage):
                result.append(Message.system(msg.content if isinstance(msg.content, str) else str(msg.content)))

            elif isinstance(msg, HumanMessage):
                result.append(Message.user(msg.content if isinstance(msg.content, str) else str(msg.content)))

            elif isinstance(msg, AIMessage):
                converted = self._convert_ai_message(msg)
                # Attach per-call usage so it propagates into
                # template.trace_messages[*].usage_metadata via to_dict().
                # Reuse the extraction helper from trace.py to keep the two
                # paths consistent.
                from karenina.adapters.langchain_deep_agents.trace import (
                    _extract_ai_usage_metadata,
                )

                converted.usage_metadata = _extract_ai_usage_metadata(msg)
                result.append(converted)

            elif isinstance(msg, ToolMessage):
                content = msg.content if isinstance(msg.content, str) else str(msg.content)
                tool_call_id = getattr(msg, "tool_call_id", "") or ""
                result.append(
                    Message(
                        role=Role.TOOL,
                        content=[ToolResultContent(tool_use_id=tool_call_id, content=content)],
                    )
                )

            else:
                logger.debug("Skipping unknown message type: %s", type(msg).__name__)

        return result

    def _convert_ai_message(self, msg: Any) -> Message:
        """Convert a LangGraph AIMessage to a karenina Message.

        Handles text content, structured content blocks, tool calls, and
        chain-of-thought reasoning. Reasoning is sourced from the same
        locations as ``trace.deep_agents_messages_to_raw_trace``:
        ``additional_kwargs["reasoning_content"]``, ``additional_kwargs
        ["reasoning"]``, or content-list entries with ``type == "thinking"``.
        Putting it on the unified Message preserves parity with the CSDK
        adapter, which already emits ``ThinkingContent`` here so downstream
        consumers walking ``Message.content`` see reasoning identically
        across adapters.

        Args:
            msg: LangGraph AIMessage instance.

        Returns:
            Karenina Message with Role.ASSISTANT.
        """
        content_blocks: list[Content] = []

        # Place ThinkingContent first so the block order matches CSDK
        # (thinking -> text -> tool_use), which downstream rendering relies on.
        additional = getattr(msg, "additional_kwargs", None) or {}
        if isinstance(additional, dict):
            for key in ("reasoning_content", "reasoning"):
                value = additional.get(key)
                if isinstance(value, str) and value.strip():
                    content_blocks.append(ThinkingContent(thinking=value))
                    break

        if isinstance(msg.content, str) and msg.content:
            content_blocks.append(TextContent(text=msg.content))
        elif isinstance(msg.content, list):
            for block in msg.content:
                if isinstance(block, str):
                    content_blocks.append(TextContent(text=block))
                elif isinstance(block, dict):
                    if block.get("type") == "text":
                        content_blocks.append(TextContent(text=block["text"]))
                    elif block.get("type") == "tool_use":
                        content_blocks.append(
                            ToolUseContent(
                                id=block.get("id", ""),
                                name=block.get("name", ""),
                                input=block.get("input", {}),
                            )
                        )
                    elif block.get("type") == "thinking":
                        thinking_text = block.get("thinking") or ""
                        if isinstance(thinking_text, str) and thinking_text.strip():
                            # Anthropic-style thinking block in content list:
                            # prefer this over an additional_kwargs duplicate.
                            content_blocks = [b for b in content_blocks if not isinstance(b, ThinkingContent)]
                            content_blocks.insert(
                                0,
                                ThinkingContent(
                                    thinking=thinking_text,
                                    signature=block.get("signature"),
                                ),
                            )

        if hasattr(msg, "tool_calls") and msg.tool_calls:
            for tc in msg.tool_calls:
                content_blocks.append(
                    ToolUseContent(
                        id=tc.get("id", ""),
                        name=tc.get("name", ""),
                        input=tc.get("args", {}),
                    )
                )

        if content_blocks:
            return Message(role=Role.ASSISTANT, content=content_blocks)
        return Message.assistant("")

Functions

extract_system_prompt

extract_system_prompt(
    messages: list[Message],
) -> str | None

Extract system prompt from messages.

Parameters:

Name	Type	Description	Default
messages	list[Message]	List of karenina Message objects.	required

Returns:

Type	Description
str | None	Combined system prompt text, or None if no system messages.

Source code in src/karenina/adapters/langchain_deep_agents/messages.py

def extract_system_prompt(self, messages: list[Message]) -> str | None:
    """Extract system prompt from messages.

    Args:
        messages: List of karenina Message objects.

    Returns:
        Combined system prompt text, or None if no system messages.
    """
    system_parts = []
    for msg in messages:
        if msg.role == Role.SYSTEM:
            text = msg.text
            if text:
                system_parts.append(text)
    return "\n\n".join(system_parts) if system_parts else None

from_provider

from_provider(lc_messages: list[Any]) -> list[Message]

Convert LangGraph BaseMessage list to karenina Messages.

Parameters:

Name	Type	Description	Default
lc_messages	list[Any]	List of LangGraph message objects.	required

Returns:

Type	Description
list[Message]	List of karenina Message objects.

Source code in src/karenina/adapters/langchain_deep_agents/messages.py

def from_provider(self, lc_messages: list[Any]) -> list[Message]:
    """Convert LangGraph BaseMessage list to karenina Messages.

    Args:
        lc_messages: List of LangGraph message objects.

    Returns:
        List of karenina Message objects.
    """
    from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage

    result: list[Message] = []

    for msg in lc_messages:
        if isinstance(msg, SystemMessage):
            result.append(Message.system(msg.content if isinstance(msg.content, str) else str(msg.content)))

        elif isinstance(msg, HumanMessage):
            result.append(Message.user(msg.content if isinstance(msg.content, str) else str(msg.content)))

        elif isinstance(msg, AIMessage):
            converted = self._convert_ai_message(msg)
            # Attach per-call usage so it propagates into
            # template.trace_messages[*].usage_metadata via to_dict().
            # Reuse the extraction helper from trace.py to keep the two
            # paths consistent.
            from karenina.adapters.langchain_deep_agents.trace import (
                _extract_ai_usage_metadata,
            )

            converted.usage_metadata = _extract_ai_usage_metadata(msg)
            result.append(converted)

        elif isinstance(msg, ToolMessage):
            content = msg.content if isinstance(msg.content, str) else str(msg.content)
            tool_call_id = getattr(msg, "tool_call_id", "") or ""
            result.append(
                Message(
                    role=Role.TOOL,
                    content=[ToolResultContent(tool_use_id=tool_call_id, content=content)],
                )
            )

        else:
            logger.debug("Skipping unknown message type: %s", type(msg).__name__)

    return result

to_langchain_messages

to_langchain_messages(
    messages: list[Message],
) -> list[dict[str, str]]

Convert karenina messages to LangGraph-compatible dicts.

Parameters:

Name	Type	Description	Default
messages	list[Message]	List of karenina Message objects.	required

Returns:

Type	Description
list[dict[str, str]]	List of message dicts with role and content keys.

Source code in src/karenina/adapters/langchain_deep_agents/messages.py

def to_langchain_messages(self, messages: list[Message]) -> list[dict[str, str]]:
    """Convert karenina messages to LangGraph-compatible dicts.

    Args:
        messages: List of karenina Message objects.

    Returns:
        List of message dicts with role and content keys.
    """
    result = []
    for msg in messages:
        if msg.role == Role.SYSTEM:
            continue
        role_map = {
            Role.USER: "user",
            Role.ASSISTANT: "assistant",
            Role.TOOL: "tool",
        }
        role = role_map.get(msg.role, "user")
        result.append({"role": role, "content": msg.text or ""})
    return result

to_prompt_string

to_prompt_string(messages: list[Message]) -> str

Convert user/assistant messages to a prompt string.

System messages are excluded (use extract_system_prompt instead).

Parameters:

Name	Type	Description	Default
messages	list[Message]	List of karenina Message objects.	required

Returns:

Type	Description
str	Concatenated prompt string from non-system messages.

Source code in src/karenina/adapters/langchain_deep_agents/messages.py

def to_prompt_string(self, messages: list[Message]) -> str:
    """Convert user/assistant messages to a prompt string.

    System messages are excluded (use extract_system_prompt instead).

    Args:
        messages: List of karenina Message objects.

    Returns:
        Concatenated prompt string from non-system messages.
    """
    parts = []
    for msg in messages:
        if msg.role == Role.SYSTEM:
            continue
        text = msg.text
        if text:
            parts.append(text)
    return "\n\n".join(parts)

DeepAgentsParserAdapter

Parser adapter using LangChain's structured output for data extraction.

Implements the ParserPort Protocol by using with_structured_output() on the LangChain model. Falls back to JSON extraction from text if structured output is not available.

Example

from pydantic import BaseModel, Field class Answer(BaseModel): ... gene: str = Field(description="Gene name") parser = DeepAgentsParserAdapter(config) result = await parser.aparse_to_pydantic(messages, Answer) print(result.parsed.gene)

Source code in src/karenina/adapters/langchain_deep_agents/parser.py

class DeepAgentsParserAdapter:
    """Parser adapter using LangChain's structured output for data extraction.

    Implements the ParserPort Protocol by using with_structured_output()
    on the LangChain model. Falls back to JSON extraction from text if
    structured output is not available.

    Example:
        >>> from pydantic import BaseModel, Field
        >>> class Answer(BaseModel):
        ...     gene: str = Field(description="Gene name")
        >>> parser = DeepAgentsParserAdapter(config)
        >>> result = await parser.aparse_to_pydantic(messages, Answer)
        >>> print(result.parsed.gene)
    """

    def __init__(self, model_config: ModelConfig) -> None:
        """Initialize the Deep Agents Parser adapter.

        Args:
            model_config: Configuration specifying model, provider, and interface.
        """
        self._config = model_config
        self._converter = DeepAgentsMessageConverter()

        retry_policy = model_config.retry_policy or RetryPolicy()
        self._retry_executor = RetryExecutor(retry_policy, ErrorRegistry())

    @property
    def capabilities(self) -> PortCapabilities:
        """Declare adapter capabilities.

        Returns:
            PortCapabilities with system_prompt=True and structured_output=True.
        """
        return PortCapabilities(
            supports_system_prompt=True,
            supports_structured_output=True,
        )

    async def aparse_to_pydantic(
        self,
        messages: list[Any],
        schema: type[T],
    ) -> ParsePortResult[T]:
        """Parse pre-assembled prompt messages into a Pydantic model.

        Uses LangChain's with_structured_output() to constrain the LLM.
        Falls back to JSON extraction if structured output fails.

        Args:
            messages: Pre-assembled prompt messages (system + user).
            schema: A Pydantic model class defining the expected structure.

        Returns:
            ParsePortResult containing the parsed model and usage metadata.

        Raises:
            ParseError: If the LLM fails to produce valid structured data.
        """
        from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

        # Build LangChain message list
        lc_messages: list[Any] = []
        for msg in messages:
            text = msg.text or ""
            if msg.role.value == "system":
                lc_messages.append(SystemMessage(content=text))
            elif msg.role.value == "user":
                lc_messages.append(HumanMessage(content=text))
            elif msg.role.value == "assistant":
                lc_messages.append(AIMessage(content=text))

        # Create the model once, outside the retried callable, so that
        # retries repeat the API call rather than model construction.
        chat_model = create_chat_model(self._config)

        try:
            # Use include_raw=True to get the AIMessage alongside the parsed output.
            # This is critical for usage tracking: without it, with_structured_output
            # returns only the parsed dict/model, losing the AIMessage.response_metadata
            # where token counts live.
            structured_model = chat_model.with_structured_output(schema, include_raw=True)
            # This parser calls the raw LangChain model directly (it does
            # not delegate to an LLM adapter ainvoke), so it borrows its
            # own GlobalLLMLimiter permit per attempt inside
            # _ainvoke_with_timeout.
            raw_response = await self._retry_executor.aexecute_with_timeout(
                self._ainvoke_with_timeout,
                structured_model,
                lc_messages,
                timeout=self._config.request_timeout,
            )
        except Exception as e:
            if ErrorRegistry().classify(e).is_retryable():
                # A transient error here means the structured attempt
                # already exhausted its retry budget inside RetryExecutor.
                # Re-raise instead of paying a second budget on the text
                # fallback: a transient outage is not a missing
                # structured-output capability.
                raise
            logger.warning("Structured output failed, falling back to text extraction: %s", e)
            # Some OpenAI-compatible transports retain loop-affine connection
            # state after a structured-output failure. Use a fresh model for
            # the text fallback so it cannot inherit a stale async pool.
            fallback_model = create_chat_model(self._config)
            response = await self._retry_executor.aexecute_with_timeout(
                self._ainvoke_with_timeout,
                fallback_model,
                lc_messages,
                timeout=self._config.request_timeout,
            )
            return self._extract_from_text(response, schema)

        # include_raw=True returns {"raw": AIMessage, "parsed": dict/model, "parsing_error": ...}
        parsed_output = raw_response.get("parsed") if isinstance(raw_response, dict) else raw_response
        raw_msg = raw_response.get("raw") if isinstance(raw_response, dict) else None
        usage = self._extract_usage_from_response(raw_msg) if raw_msg else UsageMetadata(model=self._config.model_name)

        if isinstance(parsed_output, schema):
            return ParsePortResult(parsed=parsed_output, usage=usage)

        if isinstance(parsed_output, dict):
            try:
                parsed = schema.model_validate(parsed_output)
                return ParsePortResult(parsed=parsed, usage=usage)
            except Exception as e:
                raise ParseError(f"Failed to validate structured output: {e}") from e

        if isinstance(parsed_output, BaseModel):
            try:
                parsed = schema.model_validate(parsed_output.model_dump())
                return ParsePortResult(parsed=parsed, usage=usage)
            except Exception as e:
                raise ParseError(f"Failed to convert structured output to target schema: {e}") from e

        raise ParseError(f"Unexpected response type from structured output: {type(parsed_output).__name__}")

    async def _ainvoke_with_timeout(
        self,
        model: Any,
        lc_messages: list[Any],
        *,
        timeout: float | None = None,
    ) -> Any:
        """Call ``model.ainvoke`` under a wall-clock timeout as a guardrail.

        Mirrors the langchain adapter's helper: a fired timeout raises a
        stock ``asyncio.TimeoutError``, which ``ErrorRegistry`` classifies
        as ``TIMEOUT``, so the timeout retry budget applies inside
        ``RetryExecutor.aexecute_with_timeout``. Note that
        ``RetryPolicy.timeout_escalation`` only extends this guard on
        TIMEOUT retries. The underlying SDK client's own timeout stays
        pinned at ``request_timeout`` from construction, so the SDK may
        cut the request before an escalated guard fires.

        Each attempt holds one GlobalLLMLimiter permit for the wire call
        only (uniform per-attempt policy), so retry backoff sleeps never
        hold a permit. The permit wait itself is NOT bounded by the
        attempt timeout: a saturated limiter delays the attempt rather
        than timing it out, matching the legacy semaphore semantics.

        Args:
            model: The LangChain model (plain or structured) exposing
                ``ainvoke``.
            lc_messages: Provider-formatted messages to send.
            timeout: Optional per-call wall-clock timeout in seconds.
                When None, falls back to ``self._config.request_timeout``.
                When that is also None, the call is made without a wrapper.

        Returns:
            The raw model response object.

        Raises:
            asyncio.TimeoutError: If the call exceeds the effective timeout.
        """
        async with get_global_llm_limiter().borrow():
            if timeout is None:
                timeout = self._config.request_timeout
            if timeout is None:
                return await model.ainvoke(lc_messages)
            return await asyncio.wait_for(model.ainvoke(lc_messages), timeout=timeout)

    def _extract_from_text(self, response: Any, schema: type[T]) -> ParsePortResult[T]:
        """Extract structured data from a text response (fallback path).

        Args:
            response: The AIMessage response from the LLM.
            schema: The target Pydantic schema.

        Returns:
            ParsePortResult with the parsed model.

        Raises:
            ParseError: If JSON extraction or validation fails.
        """
        from langchain_core.messages import AIMessage

        content = ""
        if isinstance(response, AIMessage):
            content = response.content if isinstance(response.content, str) else str(response.content)
        else:
            content = str(response)

        usage = self._extract_usage_from_response(response)

        # Try to extract JSON from the text
        try:
            json_str = extract_json_from_response(content)
            data = json.loads(json_str)
            parsed = schema.model_validate(data)
            return ParsePortResult(parsed=parsed, usage=usage)
        except (json.JSONDecodeError, Exception) as e:
            raise ParseError(f"Failed to extract structured data from text response: {e}") from e

    def _extract_usage_from_response(self, response: Any) -> UsageMetadata:
        """Extract usage metadata from a LangChain response.

        Args:
            response: The response object (AIMessage or other).

        Returns:
            UsageMetadata with token counts if available.
        """
        from langchain_core.messages import AIMessage

        if isinstance(response, AIMessage):
            usage_meta = getattr(response, "usage_metadata", None)
            if usage_meta and isinstance(usage_meta, dict):
                return UsageMetadata(
                    input_tokens=usage_meta.get("input_tokens", 0),
                    output_tokens=usage_meta.get("output_tokens", 0),
                    total_tokens=usage_meta.get("input_tokens", 0) + usage_meta.get("output_tokens", 0),
                    model=self._config.model_name,
                )
        return UsageMetadata(model=self._config.model_name)

    def parse_to_pydantic(
        self,
        messages: list[Any],
        schema: type[T],
    ) -> ParsePortResult[T]:
        """Parse pre-assembled prompt messages (sync wrapper).

        Args:
            messages: Pre-assembled prompt messages.
            schema: A Pydantic model class defining the expected structure.

        Returns:
            ParsePortResult containing the parsed model and usage metadata.
        """
        from karenina.benchmark.verification.executor import get_async_portal

        portal = get_async_portal()
        if portal is not None:
            return portal.call(self.aparse_to_pydantic, messages, schema)

        try:
            asyncio.get_running_loop()
            # Fresh thread with the caller's context propagated, so
            # track_retries telemetry survives the dispatch.
            thread_timeout = compute_sync_wrapper_timeout(
                self._config.request_timeout,
                floor=DEEP_AGENTS_SYNC_WRAPPER_FLOOR,
                retry_policy=self._config.retry_policy,
                internal_call_sequences=PARSE_INTERNAL_CALL_SEQUENCES,
            )
            return run_coro_in_thread(self.aparse_to_pydantic, messages, schema, timeout=thread_timeout)

        except RuntimeError:
            return asyncio.run(self.aparse_to_pydantic(messages, schema))

    async def aclose(self) -> None:
        """Close underlying resources (no-op for this adapter)."""

Attributes

capabilities property

capabilities: PortCapabilities

Declare adapter capabilities.

Returns:

Type	Description
PortCapabilities	PortCapabilities with system_prompt=True and structured_output=True.

Functions

__init__

__init__(model_config: ModelConfig) -> None

Parameters:

Name	Type	Description	Default
model_config	ModelConfig	Configuration specifying model, provider, and interface.	required

Source code in src/karenina/adapters/langchain_deep_agents/parser.py

def __init__(self, model_config: ModelConfig) -> None:
    """Initialize the Deep Agents Parser adapter.

    Args:
        model_config: Configuration specifying model, provider, and interface.
    """
    self._config = model_config
    self._converter = DeepAgentsMessageConverter()

    retry_policy = model_config.retry_policy or RetryPolicy()
    self._retry_executor = RetryExecutor(retry_policy, ErrorRegistry())

aclose async

aclose() -> None

Close underlying resources (no-op for this adapter).

Source code in src/karenina/adapters/langchain_deep_agents/parser.py

async def aclose(self) -> None:
    """Close underlying resources (no-op for this adapter)."""

aparse_to_pydantic async

aparse_to_pydantic(
    messages: list[Any], schema: type[T]
) -> ParsePortResult[T]

Parse pre-assembled prompt messages into a Pydantic model.

Uses LangChain's with_structured_output() to constrain the LLM. Falls back to JSON extraction if structured output fails.

Parameters:

Name	Type	Description	Default
messages	list[Any]	Pre-assembled prompt messages (system + user).	required
schema	type[T]	A Pydantic model class defining the expected structure.	required

Returns:

Type	Description
ParsePortResult[T]	ParsePortResult containing the parsed model and usage metadata.

Raises:

Type	Description
ParseError	If the LLM fails to produce valid structured data.

Source code in src/karenina/adapters/langchain_deep_agents/parser.py

async def aparse_to_pydantic(
    self,
    messages: list[Any],
    schema: type[T],
) -> ParsePortResult[T]:
    """Parse pre-assembled prompt messages into a Pydantic model.

    Uses LangChain's with_structured_output() to constrain the LLM.
    Falls back to JSON extraction if structured output fails.

    Args:
        messages: Pre-assembled prompt messages (system + user).
        schema: A Pydantic model class defining the expected structure.

    Returns:
        ParsePortResult containing the parsed model and usage metadata.

    Raises:
        ParseError: If the LLM fails to produce valid structured data.
    """
    from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

    # Build LangChain message list
    lc_messages: list[Any] = []
    for msg in messages:
        text = msg.text or ""
        if msg.role.value == "system":
            lc_messages.append(SystemMessage(content=text))
        elif msg.role.value == "user":
            lc_messages.append(HumanMessage(content=text))
        elif msg.role.value == "assistant":
            lc_messages.append(AIMessage(content=text))

    # Create the model once, outside the retried callable, so that
    # retries repeat the API call rather than model construction.
    chat_model = create_chat_model(self._config)

    try:
        # Use include_raw=True to get the AIMessage alongside the parsed output.
        # This is critical for usage tracking: without it, with_structured_output
        # returns only the parsed dict/model, losing the AIMessage.response_metadata
        # where token counts live.
        structured_model = chat_model.with_structured_output(schema, include_raw=True)
        # This parser calls the raw LangChain model directly (it does
        # not delegate to an LLM adapter ainvoke), so it borrows its
        # own GlobalLLMLimiter permit per attempt inside
        # _ainvoke_with_timeout.
        raw_response = await self._retry_executor.aexecute_with_timeout(
            self._ainvoke_with_timeout,
            structured_model,
            lc_messages,
            timeout=self._config.request_timeout,
        )
    except Exception as e:
        if ErrorRegistry().classify(e).is_retryable():
            # A transient error here means the structured attempt
            # already exhausted its retry budget inside RetryExecutor.
            # Re-raise instead of paying a second budget on the text
            # fallback: a transient outage is not a missing
            # structured-output capability.
            raise
        logger.warning("Structured output failed, falling back to text extraction: %s", e)
        # Some OpenAI-compatible transports retain loop-affine connection
        # state after a structured-output failure. Use a fresh model for
        # the text fallback so it cannot inherit a stale async pool.
        fallback_model = create_chat_model(self._config)
        response = await self._retry_executor.aexecute_with_timeout(
            self._ainvoke_with_timeout,
            fallback_model,
            lc_messages,
            timeout=self._config.request_timeout,
        )
        return self._extract_from_text(response, schema)

    # include_raw=True returns {"raw": AIMessage, "parsed": dict/model, "parsing_error": ...}
    parsed_output = raw_response.get("parsed") if isinstance(raw_response, dict) else raw_response
    raw_msg = raw_response.get("raw") if isinstance(raw_response, dict) else None
    usage = self._extract_usage_from_response(raw_msg) if raw_msg else UsageMetadata(model=self._config.model_name)

    if isinstance(parsed_output, schema):
        return ParsePortResult(parsed=parsed_output, usage=usage)

    if isinstance(parsed_output, dict):
        try:
            parsed = schema.model_validate(parsed_output)
            return ParsePortResult(parsed=parsed, usage=usage)
        except Exception as e:
            raise ParseError(f"Failed to validate structured output: {e}") from e

    if isinstance(parsed_output, BaseModel):
        try:
            parsed = schema.model_validate(parsed_output.model_dump())
            return ParsePortResult(parsed=parsed, usage=usage)
        except Exception as e:
            raise ParseError(f"Failed to convert structured output to target schema: {e}") from e

    raise ParseError(f"Unexpected response type from structured output: {type(parsed_output).__name__}")

parse_to_pydantic

parse_to_pydantic(
    messages: list[Any], schema: type[T]
) -> ParsePortResult[T]

Parse pre-assembled prompt messages (sync wrapper).

Parameters:

Name	Type	Description	Default
messages	list[Any]	Pre-assembled prompt messages.	required
schema	type[T]	A Pydantic model class defining the expected structure.	required

Returns:

Type	Description
ParsePortResult[T]	ParsePortResult containing the parsed model and usage metadata.

Source code in src/karenina/adapters/langchain_deep_agents/parser.py

def parse_to_pydantic(
    self,
    messages: list[Any],
    schema: type[T],
) -> ParsePortResult[T]:
    """Parse pre-assembled prompt messages (sync wrapper).

    Args:
        messages: Pre-assembled prompt messages.
        schema: A Pydantic model class defining the expected structure.

    Returns:
        ParsePortResult containing the parsed model and usage metadata.
    """
    from karenina.benchmark.verification.executor import get_async_portal

    portal = get_async_portal()
    if portal is not None:
        return portal.call(self.aparse_to_pydantic, messages, schema)

    try:
        asyncio.get_running_loop()
        # Fresh thread with the caller's context propagated, so
        # track_retries telemetry survives the dispatch.
        thread_timeout = compute_sync_wrapper_timeout(
            self._config.request_timeout,
            floor=DEEP_AGENTS_SYNC_WRAPPER_FLOOR,
            retry_policy=self._config.retry_policy,
            internal_call_sequences=PARSE_INTERNAL_CALL_SEQUENCES,
        )
        return run_coro_in_thread(self.aparse_to_pydantic, messages, schema, timeout=thread_timeout)

    except RuntimeError:
        return asyncio.run(self.aparse_to_pydantic(messages, schema))

Functions

check_deep_agents_available

check_deep_agents_available() -> AdapterAvailability

Check if the deepagents package is installed.

Returns:

Type	Description
AdapterAvailability	AdapterAvailability with status and installation instructions.

Source code in src/karenina/adapters/langchain_deep_agents/availability.py

def check_deep_agents_available() -> AdapterAvailability:
    """Check if the deepagents package is installed.

    Returns:
        AdapterAvailability with status and installation instructions.
    """
    try:
        import deepagents  # noqa: F401

        return AdapterAvailability(
            available=True,
            reason="deepagents package is installed",
        )
    except ImportError:
        return AdapterAvailability(
            available=False,
            reason=("deepagents package not installed. Install with: pip install deepagents or: uv add deepagents"),
            fallback_interface=None,
        )

convert_mcp_to_tools async

convert_mcp_to_tools(
    mcp_servers: dict[str, Any] | None,
    exit_stack: AsyncExitStack,
) -> list[Any]

Convert MCP server configs to LangChain tools via langchain-mcp-adapters.

Creates a MultiServerMCPClient registered with the provided exit_stack so that MCP sessions remain open for the lifetime of the caller's exit_stack context.

Parameters:

Name	Type	Description	Default
mcp_servers	dict[str, Any] | None	Dict mapping server names to MCPServerConfig.	required
exit_stack	AsyncExitStack	AsyncExitStack managing session lifecycles. Sessions remain open until the exit stack closes.	required

Returns:

Type	Description
list[Any]	List of LangChain BaseTool instances from all MCP servers.

Source code in src/karenina/adapters/langchain_deep_agents/mcp.py

async def convert_mcp_to_tools(
    mcp_servers: dict[str, Any] | None,
    exit_stack: AsyncExitStack,
) -> list[Any]:
    """Convert MCP server configs to LangChain tools via langchain-mcp-adapters.

    Creates a MultiServerMCPClient registered with the provided exit_stack
    so that MCP sessions remain open for the lifetime of the caller's
    exit_stack context.

    Args:
        mcp_servers: Dict mapping server names to MCPServerConfig.
        exit_stack: AsyncExitStack managing session lifecycles. Sessions
            remain open until the exit stack closes.

    Returns:
        List of LangChain BaseTool instances from all MCP servers.
    """
    server_params = build_mcp_server_params(mcp_servers)
    if not server_params:
        return []

    from langchain_mcp_adapters.client import MultiServerMCPClient

    client = MultiServerMCPClient(cast(Any, server_params))
    await exit_stack.enter_async_context(cast(Any, client))
    tools: list[Any] = await client.get_tools()
    return tools

deep_agents_messages_to_raw_trace

deep_agents_messages_to_raw_trace(
    messages: list[Any], include_user_messages: bool = False
) -> str

Convert LangGraph messages to raw trace string format.

Produces the canonical karenina trace format shared by the langchain, claude_agent_sdk, and manual adapters: one --- AI Message --- section per turn carrying the text followed by an inline Tool Calls: block, and --- Tool Message (call_id: <id>) --- for each tool result. Thinking content surfaces as a separate --- Thinking --- section before the AI Message for that turn.

Parameters:

Name	Type	Description	Default
messages	list[Any]	List of LangGraph BaseMessage objects.	required
include_user_messages	bool	If True, include HumanMessage in trace.	False

Returns:

Type	Description
str	Formatted trace string with --- delimiters compatible with
str	regex highlighting, database storage, and the trace-evaluation
str	infrastructure shared with the other adapters.

Source code in src/karenina/adapters/langchain_deep_agents/trace.py

def deep_agents_messages_to_raw_trace(
    messages: list[Any],
    include_user_messages: bool = False,
) -> str:
    """Convert LangGraph messages to raw trace string format.

    Produces the canonical karenina trace format shared by the langchain,
    claude_agent_sdk, and manual adapters: one ``--- AI Message ---``
    section per turn carrying the text followed by an inline ``Tool Calls:``
    block, and ``--- Tool Message (call_id: <id>) ---`` for each tool
    result. Thinking content surfaces as a separate ``--- Thinking ---``
    section before the AI Message for that turn.

    Args:
        messages: List of LangGraph BaseMessage objects.
        include_user_messages: If True, include HumanMessage in trace.

    Returns:
        Formatted trace string with ``---`` delimiters compatible with
        regex highlighting, database storage, and the trace-evaluation
        infrastructure shared with the other adapters.
    """
    from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage

    if not messages:
        return ""

    parts: list[str] = []

    for msg in messages:
        if isinstance(msg, SystemMessage):
            continue

        if isinstance(msg, HumanMessage):
            if include_user_messages:
                parts.append(f"--- Human Message ---\n{msg.content}")
            continue

        if isinstance(msg, AIMessage):
            thinking = _extract_thinking_content(msg)
            if thinking:
                parts.append(f"--- Thinking ---\n{thinking}")

            text = _extract_ai_text(msg)
            tool_calls = getattr(msg, "tool_calls", None) or []

            if text or tool_calls:
                content_parts: list[str] = []
                if text:
                    content_parts.append(text)
                if tool_calls:
                    content_parts.append("\nTool Calls:")
                    for tc in tool_calls:
                        tool_name = tc.get("name", "unknown")
                        tool_id = tc.get("id", "unknown")
                        tool_args = tc.get("args", {})
                        content_parts.append(f"  {tool_name} (call_{tool_id})")
                        content_parts.append(f"   Call ID: {tool_id}")
                        if tool_args:
                            content_parts.append(f"   Args: {tool_args}")
                parts.append("--- AI Message ---\n" + "\n".join(content_parts))

        elif isinstance(msg, ToolMessage):
            content = msg.content if isinstance(msg.content, str) else str(msg.content)
            tool_call_id = getattr(msg, "tool_call_id", "unknown") or "unknown"
            parts.append(f"--- Tool Message (call_id: {tool_call_id}) ---\n{content}")

    return "\n\n".join(parts)

deep_agents_messages_to_trace_messages

deep_agents_messages_to_trace_messages(
    messages: list[Any], include_user_messages: bool = False
) -> list[dict[str, Any]]

Convert LangGraph messages to structured TraceMessage list.

Produces the structured list[dict] format consumed by the frontend TraceMessage component. Each assistant entry preserves per-call usage_metadata when the underlying AIMessage reports it (so post-hoc audits of token usage / cache hits per turn remain possible).

Parameters:

Name	Type	Description	Default
messages	list[Any]	List of LangGraph BaseMessage objects.	required
include_user_messages	bool	If True, include HumanMessage entries. Defaults to False to match the existing raw_trace behavior.	False

Returns:

Type	Description
list[dict[str, Any]]	List of TraceMessage dicts with role, content, block_index, and
list[dict[str, Any]]	optional tool_calls, tool_result, and usage_metadata fields.

Source code in src/karenina/adapters/langchain_deep_agents/trace.py

def deep_agents_messages_to_trace_messages(
    messages: list[Any],
    include_user_messages: bool = False,
) -> list[dict[str, Any]]:
    """Convert LangGraph messages to structured TraceMessage list.

    Produces the structured list[dict] format consumed by the frontend
    TraceMessage component. Each assistant entry preserves per-call
    usage_metadata when the underlying AIMessage reports it (so
    post-hoc audits of token usage / cache hits per turn remain possible).

    Args:
        messages: List of LangGraph BaseMessage objects.
        include_user_messages: If True, include HumanMessage entries. Defaults
            to False to match the existing raw_trace behavior.

    Returns:
        List of TraceMessage dicts with role, content, block_index, and
        optional tool_calls, tool_result, and usage_metadata fields.
    """
    from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage

    if not messages:
        return []

    result: list[dict[str, Any]] = []
    block_index = 0

    for msg in messages:
        if isinstance(msg, SystemMessage):
            continue

        if isinstance(msg, HumanMessage):
            if include_user_messages:
                content = msg.content if isinstance(msg.content, str) else str(msg.content)
                if content:
                    result.append(
                        {
                            "role": "user",
                            "content": content,
                            "block_index": block_index,
                        }
                    )
                    block_index += 1
            continue

        if isinstance(msg, AIMessage):
            text = _extract_ai_text(msg)
            thinking = _extract_thinking_content(msg)
            tool_calls: list[dict[str, Any]] = []
            if hasattr(msg, "tool_calls") and msg.tool_calls:
                for tc in msg.tool_calls:
                    tool_calls.append(
                        {
                            "id": tc.get("id", ""),
                            "name": tc.get("name", "unknown"),
                            "input": tc.get("args", {}) if isinstance(tc.get("args"), dict) else {},
                        }
                    )

            if text or tool_calls or thinking:
                trace_msg: dict[str, Any] = {
                    "role": "assistant",
                    "content": text,
                    "block_index": block_index,
                }

                if tool_calls:
                    trace_msg["tool_calls"] = tool_calls

                if thinking:
                    # Mirror the CSDK structured-trace shape so downstream
                    # consumers (frontend trace component, evaluators) can
                    # treat reasoning identically across adapters.
                    trace_msg["thinking"] = {"thinking": thinking}

                usage_metadata = _extract_ai_usage_metadata(msg)
                if usage_metadata is not None:
                    trace_msg["usage_metadata"] = usage_metadata

                result.append(trace_msg)
                block_index += 1
            continue

        if isinstance(msg, ToolMessage):
            content = msg.content if isinstance(msg.content, str) else str(msg.content)
            tool_use_id = getattr(msg, "tool_call_id", "") or ""
            result.append(
                {
                    "role": "tool",
                    "content": content,
                    "block_index": block_index,
                    "tool_result": {
                        "tool_use_id": tool_use_id,
                        "is_error": False,
                    },
                }
            )
            block_index += 1

    return result

extract_deep_agents_usage

extract_deep_agents_usage(
    messages: list[Any], model: str | None = None
) -> UsageMetadata

Extract aggregated usage metadata from LangGraph messages.

Sums token counts across all AIMessage instances in the conversation. Token counts come from AIMessage.usage_metadata (preferred) or AIMessage.response_metadata.token_usage (fallback).

Parameters:

Name	Type	Description	Default
messages	list[Any]	List of LangGraph BaseMessage objects.	required
model	str | None	Model name to include in usage metadata.	None

Returns:

Type	Description
UsageMetadata	Aggregated UsageMetadata for the entire agent run.

Source code in src/karenina/adapters/langchain_deep_agents/usage.py

def extract_deep_agents_usage(
    messages: list[Any],
    model: str | None = None,
) -> UsageMetadata:
    """Extract aggregated usage metadata from LangGraph messages.

    Sums token counts across all AIMessage instances in the conversation.
    Token counts come from AIMessage.usage_metadata (preferred) or
    AIMessage.response_metadata.token_usage (fallback).

    Args:
        messages: List of LangGraph BaseMessage objects.
        model: Model name to include in usage metadata.

    Returns:
        Aggregated UsageMetadata for the entire agent run.
    """
    from langchain_core.messages import AIMessage

    total_input = 0
    total_output = 0

    for msg in messages:
        if not isinstance(msg, AIMessage):
            continue

        usage_meta = getattr(msg, "usage_metadata", None)
        if usage_meta and isinstance(usage_meta, dict):
            total_input += usage_meta.get("input_tokens", 0)
            total_output += usage_meta.get("output_tokens", 0)
            continue

        resp_meta = getattr(msg, "response_metadata", None)
        if resp_meta and isinstance(resp_meta, dict):
            token_usage = resp_meta.get("token_usage", {})
            if isinstance(token_usage, dict):
                total_input += token_usage.get("prompt_tokens", 0)
                total_output += token_usage.get("completion_tokens", 0)

    return UsageMetadata(
        input_tokens=total_input,
        output_tokens=total_output,
        total_tokens=total_input + total_output,
        model=model,
    )

wrap_deep_agents_error

wrap_deep_agents_error(
    error: Exception,
) -> tuple[Exception, bool]

Map a Deep Agents / LangGraph exception to a karenina exception.

Parameters:

Name	Type	Description	Default
error	Exception	The original exception from Deep Agents or LangGraph.	required

Returns:

Type	Description
Exception	Tuple of (mapped_exception, limit_reached). The limit_reached flag
bool	is True when the error indicates the agent hit a recursion or turn limit.

Source code in src/karenina/adapters/langchain_deep_agents/errors.py

def wrap_deep_agents_error(error: Exception) -> tuple[Exception, bool]:
    """Map a Deep Agents / LangGraph exception to a karenina exception.

    Args:
        error: The original exception from Deep Agents or LangGraph.

    Returns:
        Tuple of (mapped_exception, limit_reached). The limit_reached flag
        is True when the error indicates the agent hit a recursion or turn limit.
    """
    error_str = str(error).lower()

    # Check for recursion / turn limit errors
    if "recursion" in error_str or "limit" in error_str or "max_turns" in error_str:
        return (
            AgentExecutionError(f"Agent hit turn limit: {error}"),
            True,
        )

    # Timeout errors
    if isinstance(error, TimeoutError | asyncio.TimeoutError):
        return AgentTimeoutError(f"Agent execution timed out: {error}"), False

    # Output parsing errors
    if ("parse" in error_str and "output" in error_str) or ("output" in error_str and "format" in error_str):
        return AgentResponseError(f"Failed to parse agent response: {error}"), False

    # GraphRecursionError from LangGraph
    try:
        from langgraph.errors import GraphRecursionError

        if isinstance(error, GraphRecursionError):
            return (
                AgentExecutionError(f"Agent hit recursion limit: {error}"),
                True,
            )
    except ImportError:
        pass

    # Default: general execution error
    return AgentExecutionError(f"Agent execution failed: {error}"), False