feat(adapters/langgraph): make agent structured output not hang or silently vanish

docs/pyagentspec/source/changelog.rst

@@ -17,6 +17,14 @@ Improvements
 Bug fixes
 ^^^^^^^^^
 
+* **LangGraph structured-output guard**
+
+  LangGraph agents with declared structured outputs now fail promptly with a clear error when
+  their model repeatedly returns prose or finishes without producing the declared output, rather
+  than looping until the recursion limit or silently returning empty outputs.
+
+  We thank @spichen for the contribution!
+
 Breaking Changes
 ^^^^^^^^^^^^^^^^
 

pyagentspec/src/pyagentspec/adapters/_utils.py

@@ -184,6 +184,16 @@ def _build_type_from_schema(
     return mapping.get(t, Any)
 
 
+def is_single_string_output(expected_outputs: List[AgentSpecProperty]) -> bool:
+    """Whether the declared outputs are a single string property.
+
+    Such an output is the model's free text, so an adapter can read it from the final
+    message instead of asking for structured generation. Lives here rather than in one
+    adapter because it is a property of the declared outputs.
+    """
+    return len(expected_outputs) == 1 and expected_outputs[0].type == "string"
+
+
 def create_pydantic_model_from_properties(
     model_name: str, properties: List[AgentSpecProperty]
 ) -> type[BaseModel]:

pyagentspec/src/pyagentspec/adapters/langgraph/_agent_output_guard.py

@@ -0,0 +1,129 @@
+# Copyright © 2025, 2026 Oracle and/or its affiliates.
+#
+# This software is under the Apache License 2.0
+# (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
+# (UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
+
+"""Makes an agent fail loudly when its model will not produce structured output.
+
+``create_agent(response_format=...)`` ties the agent's control flow to getting a
+parseable structured response. A model that never produces one fails in one of two
+ways, depending on whether the agent has real tools:
+
+* no tools: the agent loops. ``create_agent`` sets ``recursion_limit=9999``, so that is
+  thousands of model calls before ``GraphRecursionError``.
+* with tools: routing treats a turn with no tool calls as "done", so the agent exits
+  with no ``structured_response`` and the declared outputs come back empty, with no
+  error at all (upstream langchain issue #36349).
+
+:class:`StructuredOutputGuard` covers both. An agent whose only declared output is a
+string avoids the problem entirely, since ``response_format`` is skipped for it (see
+:func:`pyagentspec.adapters._utils.is_single_string_output`).
+"""
+
+from functools import lru_cache
+from typing import Annotated, Any, Dict, List, NoReturn, Optional
+
+from typing_extensions import NotRequired
+
+from pyagentspec.adapters.langgraph._types import (
+    AgentMiddleware,
+    AgentState,
+    langchain_middleware_types,
+)
+
+# A compliant run spends at most one turn without a structured response, so a small
+# bound separates "model is working on it" from "model will never comply".
+DEFAULT_MAX_STRUCTURED_OUTPUT_ATTEMPTS = 3
+
+# Must match the private field on _StructuredOutputAttemptState.
+_ATTEMPTS_STATE_KEY = "_pyagentspec_structured_output_attempts"
+
+
+class StructuredOutputNotProducedError(RuntimeError):
+    """Raised when a model will not produce the structured response an agent declares.
+
+    Subclasses ``RuntimeError`` so existing handlers around agent execution still catch it.
+    """
+
+
+class _StructuredOutputAttemptState(AgentState):
+    # On the state rather than the instance, so the count is per run and not shared
+    # between concurrent ones. LangChain omits PrivateStateAttr fields from user input
+    # and output schemas.
+    _pyagentspec_structured_output_attempts: NotRequired[int]
+
+
+@lru_cache
+def _structured_output_attempt_state_schema() -> type:
+    """Add LangChain's private-state marker when the guard is used."""
+
+    _StructuredOutputAttemptState.__annotations__[_ATTEMPTS_STATE_KEY] = NotRequired[
+        Annotated[int, langchain_middleware_types.PrivateStateAttr]
+    ]
+
+    return _StructuredOutputAttemptState
+
+
+class StructuredOutputGuard(AgentMiddleware):
+    """Fails an agent that cannot produce the structured output it declares.
+
+    Install only when ``response_format`` is set. Without it, a turn with no tool calls
+    just means the agent is done and an empty ``structured_response`` is fine.
+    """
+
+    def __init__(
+        self,
+        *,
+        agent_name: str,
+        output_titles: List[str],
+        model_id: str,
+        max_attempts: int = DEFAULT_MAX_STRUCTURED_OUTPUT_ATTEMPTS,
+    ) -> None:
+        super().__init__()
+        self.state_schema = _structured_output_attempt_state_schema()
+        self.agent_name = agent_name
+        self.output_titles = output_titles
+        self.model_id = model_id
+        self.max_attempts = max_attempts
+
+    def _fail(self, detail: str) -> NoReturn:
+        raise StructuredOutputNotProducedError(
+            f"Agent {self.agent_name!r} did not produce a structured response matching its "
+            f"declared outputs ({', '.join(self.output_titles)}). {detail} Model "
+            f"{self.model_id!r} may not support structured output. Declare a single string "
+            f"output to get the model's free text instead, or use a model that supports "
+            f"structured output."
+        )
+
+    # No async variants: LangGraph routes async runs to the sync hooks, and both are pure.
+    def after_model(self, state: Any, runtime: Any) -> Optional[Dict[str, int]]:
+        # Runs after every model turn. The last message decides between three cases:
+        #
+        # * an AIMessage with tool calls: progress, counter resets;
+        # * an AIMessage without tool calls: prose instead of the structured response,
+        #   counter increments and the run fails at the bound;
+        # * anything else: in a plain run the model's AIMessage is always last here, so
+        #   this means another middleware rewrote the history and the turn cannot be
+        #   classified. The counter is left untouched (None updates nothing) rather than
+        #   reset: unclassifiable is not evidence of progress, and resetting would let
+        #   such a middleware clear real evidence of a stuck model every turn.
+        messages = state.get("messages") or []
+        last = messages[-1] if messages else None
+        if last is None or getattr(last, "type", None) != "ai":
+            return None
+        if getattr(last, "tool_calls", None):
+            # A tool call is progress, so the bound applies to consecutive failures only.
+            # Under ToolStrategy the structured response is itself a tool call.
+            return {_ATTEMPTS_STATE_KEY: 0}
+        attempts = (state.get(_ATTEMPTS_STATE_KEY) or 0) + 1
+        if attempts >= self.max_attempts:
+            self._fail(f"It answered in prose {attempts} times instead.")
+        return {_ATTEMPTS_STATE_KEY: attempts}
+
+    def after_agent(self, state: Any, runtime: Any) -> None:
+        # after_model never sees the silent case: an agent with tools exits on its first
+        # prose turn, well before the bound.
+        if "structured_response" in state:
+            return None
+        self._fail("It ended its run without one.")

pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py

@@ -35,7 +35,9 @@
     SchemaRegistry,
     _build_type_from_schema,
     create_pydantic_model_from_properties,
+    is_single_string_output,
 )
+from pyagentspec.adapters.langgraph._agent_output_guard import StructuredOutputGuard
 from pyagentspec.adapters.langgraph._execution_span import patch_with_execution_span
 from pyagentspec.adapters.langgraph._managerworkers import (
     _MANAGER_NODE_KEY,
@@ -70,10 +72,7 @@
     langgraph_graph,
     langgraph_swarm,
 )
-from pyagentspec.adapters.langgraph.mcp_utils import (
-    _HttpxClientFactory,
-    run_async_in_sync,
-)
+from pyagentspec.adapters.langgraph.mcp_utils import _HttpxClientFactory, run_async_in_sync
 from pyagentspec.adapters.langgraph.tracing import (
     AgentSpecLlmCallbackHandler,
     AgentSpecToolCallbackHandler,
@@ -107,10 +106,7 @@
 )
 from pyagentspec.llms.ocigenaiconfig import OciGenAiConfig
 from pyagentspec.llms.ollamaconfig import OllamaConfig
-from pyagentspec.llms.openaicompatibleconfig import (
-    OpenAIAPIType,
-    OpenAiCompatibleConfig,
-)
+from pyagentspec.llms.openaicompatibleconfig import OpenAIAPIType, OpenAiCompatibleConfig
 from pyagentspec.llms.openaiconfig import OpenAiConfig
 from pyagentspec.llms.vllmconfig import VllmConfig
 from pyagentspec.managerworkers import ManagerWorkers as AgentSpecManagerWorkers
@@ -591,19 +587,15 @@ def _input_message_node_convert_to_langgraph(
         self,
         node: AgentSpecInputMessageNode,
     ) -> "NodeExecutor":
-        from pyagentspec.adapters.langgraph._node_execution import (
-            InputMessageNodeExecutor,
-        )
+        from pyagentspec.adapters.langgraph._node_execution import InputMessageNodeExecutor
 
         return InputMessageNodeExecutor(node)
 
     def _output_message_node_convert_to_langgraph(
         self,
         node: AgentSpecOutputMessageNode,
     ) -> "NodeExecutor":
-        from pyagentspec.adapters.langgraph._node_execution import (
-            OutputMessageNodeExecutor,
-        )
+        from pyagentspec.adapters.langgraph._node_execution import OutputMessageNodeExecutor
 
         return OutputMessageNodeExecutor(node)
 
@@ -668,9 +660,7 @@ def _catch_exception_node_convert_to_langgraph(
         config: RunnableConfig,
         middleware: List[Any],
     ) -> "NodeExecutor":
-        from pyagentspec.adapters.langgraph._node_execution import (
-            CatchExceptionNodeExecutor,
-        )
+        from pyagentspec.adapters.langgraph._node_execution import CatchExceptionNodeExecutor
 
         subflow = self.convert(
             catch_node.subflow,
@@ -1235,8 +1225,9 @@ def _create_react_agent_with_given_info(
         state_schema: Optional[Any] = None
         response_format: Any = None
 
-        # Build response (output) model (used for response_format)
-        if outputs:
+        # A single string output is read from the final message instead, so it needs no
+        # response_format. Same rule as LlmNodeExecutor.
+        if outputs and not is_single_string_output(outputs):
             output_model = create_pydantic_model_from_properties("AgentOutputModel", outputs)
             # Explicitly use ToolStrategy instead of letting LangChain select a provider
             # strategy. OpenAI-compatible models do not necessarily support provider-native
@@ -1266,6 +1257,17 @@ def _create_react_agent_with_given_info(
             response_format=response_format,
             state_schema=state_schema,
         )
+        if output_model is not None:
+            # Rebuild rather than append: `middleware` is shared across every agent of a
+            # swarm / manager-workers graph.
+            middleware = [
+                *middleware,
+                StructuredOutputGuard(
+                    agent_name=name,
+                    output_titles=[output.title for output in outputs],
+                    model_id=llm_config.model_id,
+                ),
+            ]
         if middleware:
             create_agent_kwargs["middleware"] = middleware
         compiled_graph: CompiledStateGraph[Any, Any, Any] = langchain_agents.create_agent(

pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py

@@ -16,7 +16,11 @@
     maybe_warn_about_unrestricted_templated_url,
     validate_url_against_allow_list,
 )
-from pyagentspec.adapters._utils import render_nested_object_template, render_template
+from pyagentspec.adapters._utils import (
+    is_single_string_output,
+    render_nested_object_template,
+    render_template,
+)
 from pyagentspec.adapters.langgraph._types import (
     BaseChatModel,
     BaseMessage,
@@ -606,20 +610,13 @@ def __init__(self, node: AgentSpecLlmNode, llm: BaseChatModel) -> None:
         super().__init__(node)
         if not isinstance(self.node, AgentSpecLlmNode):
             raise TypeError("LlmNodeExecutor can only be initialized with LlmNode")
-        outputs = self.node.outputs
-        if outputs is not None and len(outputs) == 1 and outputs[0].type == "string":
-            self.requires_structured_generation = False
-        else:
-            self.requires_structured_generation = True
         if not isinstance(llm, BaseChatModel):
             raise TypeError("Llm can only be initialized with a BaseChatModel")
 
         self.llm: BaseChatModel = llm
 
         node_outputs = self.node.outputs or []
-        self.requires_structured_generation = not (
-            len(node_outputs) == 1 and node_outputs[0].type == "string"
-        )
+        self.requires_structured_generation = not is_single_string_output(node_outputs)
 
         self.structured_llm: Any = None
 
@@ -933,7 +930,7 @@ def extract_outputs_from_invoke_result(
     # Extracts the outputs from the return value of an invoke call made on an agent
     # The outputs are typically exposed as part of the `structured_response`, or as entries in the result directly.
     # We give priority to the latter.
-    return {
+    outputs = {
         # Defaults if available
         **{
             output.title: output.default
@@ -949,3 +946,14 @@ def extract_outputs_from_invoke_result(
             if output.title in result
         },
     }
+    # No response_format is requested for a single string output, so read it from the
+    # final message.
+    if is_single_string_output(expected_outputs):
+        title = expected_outputs[0].title
+        if title not in outputs:
+            messages = result.get("messages")
+            if messages:
+                content = getattr(messages[-1], "content", None)
+                if content is not None:
+                    outputs[title] = content
+    return outputs

pyagentspec/src/pyagentspec/adapters/langgraph/_types.py

@@ -15,11 +15,12 @@
     # Otherwise, importing the module when they are not installed would lead to an import error.
 
     import langchain.agents as langchain_agents
+    import langchain.agents.middleware.types as langchain_middleware_types
     import langchain_ollama
     import langchain_openai
     import langgraph.graph as langgraph_graph
     import langgraph_swarm
-    from langchain.agents.middleware.types import AgentState
+    from langchain.agents.middleware.types import AgentMiddleware, AgentState
     from langchain_core.callbacks import BaseCallbackHandler
     from langchain_core.language_models import BaseChatModel
     from langchain_core.messages import BaseMessage, SystemMessage, ToolMessage
@@ -39,6 +40,7 @@
     langchain_openai = LazyLoader("langchain_openai")
     langgraph_graph = LazyLoader("langgraph.graph")
     langchain_agents = LazyLoader("langchain.agents")
+    langchain_middleware_types = LazyLoader("langchain.agents.middleware.types")
     BaseTool = LazyType("langchain_core.tools", "BaseTool")
     StructuredTool = LazyType("langchain_core.tools", "StructuredTool")
     Checkpointer = LazyType("langgraph.types", "Checkpointer")
@@ -59,6 +61,7 @@
     GenerationChunk = LazyType("langchain_core.outputs", "GenerationChunk")
     LLMResult = LazyType("langchain_core.outputs", "LLMResult")
     AgentState = LazyType("langchain.agents.middleware.types", "AgentState")
+    AgentMiddleware = LazyType("langchain.agents.middleware.types", "AgentMiddleware")
 
 
 LangGraphTool: TypeAlias = Union[BaseTool, Callable[..., Any]]
@@ -108,6 +111,7 @@ class FlowOutputSchema(TypedDict):
 __all__ = [
     "langgraph_graph",
     "langchain_agents",
+    "langchain_middleware_types",
     "langchain_ollama",
     "langchain_openai",
     "LangGraphTool",
@@ -136,6 +140,7 @@ class FlowOutputSchema(TypedDict):
     "ToolMessage",
     "BaseChatModel",
     "AgentState",
+    "AgentMiddleware",
     "Checkpointer",
     "interrupt",
     "RunnableConfig",