feat(adapters/langgraph): support ManagerWorkers and Swarm orchestration

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

@@ -49,8 +49,10 @@
 from pyagentspec.flows.nodes import OutputMessageNode as AgentSpecOutputMessageNode
 from pyagentspec.flows.nodes import StartNode as AgentSpecStartNode
 from pyagentspec.flows.nodes import ToolNode as AgentSpecToolNode
+from pyagentspec.managerworkers import ManagerWorkers as AgentSpecManagerWorkers
 from pyagentspec.property import Property as AgentSpecProperty
 from pyagentspec.property import _empty_default as pyagentspec_empty_default
+from pyagentspec.swarm import Swarm as AgentSpecSwarm
 from pyagentspec.tracing.events import NodeExecutionEnd as AgentSpecNodeExecutionEnd
 from pyagentspec.tracing.events import NodeExecutionStart as AgentSpecNodeExecutionStart
 from pyagentspec.tracing.events.exception import ExceptionRaised
@@ -529,16 +531,95 @@ def _create_react_agent_with_given_input_values(
             )
         return self._agents_cache[system_prompt]
 
+    def _create_composite_graph_with_given_input_values(
+        self, inputs: Dict[str, Any]
+    ) -> CompiledStateGraph[Any, Any]:
+        """Compile the node's ``ManagerWorkers`` / ``Swarm`` into a runnable graph for
+        these inputs, cached by the rendered entry-agent prompt.
+
+        Such a graph runs over ``MessagesState`` and can't carry structured inputs to its
+        inner agents, so the node inputs are baked into the entry agent's ``system_prompt``
+        (the ``group_manager`` for a ManagerWorkers, the ``first_agent`` for a Swarm) and
+        the now-satisfied input ports dropped, so declared == inferred for the downstream
+        span re-validation. A non-Agent entry is passed through unchanged so the converter
+        raises its own clear error.
+        """
+        from pyagentspec.adapters.langgraph._langgraphconverter import (
+            AgentSpecToLangGraphConverter,
+        )
+
+        converter = AgentSpecToLangGraphConverter()
+        component = self.node.agent
+        if isinstance(component, AgentSpecManagerWorkers):
+            entry_agent = component.group_manager
+            convert = converter._manager_workers_convert_to_langgraph
+
+            def rebuild(rendered_entry: Any) -> Any:
+                return component.model_copy(update={"group_manager": rendered_entry, "inputs": []})
+
+        elif isinstance(component, AgentSpecSwarm):
+            entry_agent = component.first_agent
+            convert = converter._swarm_convert_to_langgraph
+
+            def rebuild(rendered_entry: Any) -> Any:
+                # The entry agent appears as first_agent and inside the relationship
+                # tuples, so swap it (matched by id) in both.
+                def _swap(agent: Any) -> Any:
+                    return rendered_entry if agent.id == entry_agent.id else agent
+
+                return component.model_copy(
+                    update={
+                        "first_agent": rendered_entry,
+                        "relationships": [
+                            (_swap(caller), _swap(recipient))
+                            for caller, recipient in component.relationships
+                        ],
+                        "inputs": [],
+                    }
+                )
+
+        else:
+            raise TypeError(
+                "_create_composite_graph_with_given_input_values requires a "
+                "ManagerWorkers or Swarm"
+            )
+
+        is_agent_entry = isinstance(entry_agent, AgentSpecAgent)
+        cache_key = (
+            render_template(entry_agent.system_prompt, inputs) if is_agent_entry else component.id
+        )
+        if cache_key not in self._agents_cache:
+            rendered = (
+                rebuild(entry_agent.model_copy(update={"system_prompt": cache_key, "inputs": []}))
+                if is_agent_entry
+                else component
+            )
+            self._agents_cache[cache_key] = convert(
+                rendered,
+                tool_registry=self.tool_registry,
+                converted_components=self.converted_components,
+                checkpointer=self.checkpointer,
+                config=self.config,
+                middleware=self._middleware,
+            )
+        return self._agents_cache[cache_key]
+
     def _prepare_agent_and_inputs(
         self, inputs: Dict[str, Any], messages: Messages
     ) -> Tuple[CompiledStateGraph[Any, Any], Dict[str, Any]]:
-        agent = self._create_react_agent_with_given_input_values(inputs)
         # LangGraph's agent expects at least one user message to drive execution.
         # When an AgentNode is used with a templated system prompt and no messages are provided
         # by the flow, the agent can crash. To avoid this, we artificially insert an empty
         # user message when the message list is empty.
         if not messages:
             messages = cast(Messages, [{"role": "user", "content": ""}])
+        if isinstance(self.node.agent, (AgentSpecManagerWorkers, AgentSpecSwarm)):
+            # A ManagerWorkers / Swarm flow step runs as a multi-agent graph over
+            # MessagesState: node inputs were baked into the entry agent's prompt, so the
+            # graph is driven by messages alone (not the agent's remaining_steps state).
+            graph = self._create_composite_graph_with_given_input_values(inputs)
+            return graph, {"messages": messages}
+        agent = self._create_react_agent_with_given_input_values(inputs)
         inputs |= {
             "remaining_steps": 20,  # Get the right number of steps left
             "messages": messages,
@@ -919,13 +1000,25 @@ def _accumulate_outputs(
                 outputs[collected_output_name].append(output_value)
 
 
+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, not a structured field, so the
+    adapter takes it directly from the agent's final message rather than forcing
+    structured generation. Mirrors ``LlmNodeExecutor``'s single-string handling
+    and lets a string output work on models without structured-output support.
+    """
+    outputs = expected_outputs or []
+    return len(outputs) == 1 and outputs[0].type == "string"
+
+
 def extract_outputs_from_invoke_result(
     result: Dict[str, Any], expected_outputs: List[AgentSpecProperty]
 ) -> Dict[str, Any]:
     # 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
@@ -941,3 +1034,17 @@ def extract_outputs_from_invoke_result(
             if output.title in result
         },
     }
+    # A single string output is the agent's free-text answer, not a structured
+    # field. When structured generation didn't populate it — because the model
+    # lacks structured-output support, or because none was requested (see
+    # ``_create_react_agent_with_given_info``) — fall back to the final message
+    # content so the output still carries the agent's response.
+    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/managerworkers.py

@@ -13,6 +13,7 @@
 from typing_extensions import Self
 
 from pyagentspec.agenticcomponent import AgenticComponent
+from pyagentspec.property import Property
 from pyagentspec.validation_helpers import model_validator_with_error_accumulation
 from pyagentspec.versioning import AgentSpecVersionEnum
 
@@ -65,6 +66,24 @@ class ManagerWorkers(AgenticComponent):
         default=AgentSpecVersionEnum.v25_4_2, init=False, exclude=True
     )
 
+    def _get_inferred_inputs(self) -> List[Property]:
+        """A ``ManagerWorkers`` exposes the inputs of its group manager.
+
+        The group manager is the component that drives the conversation and whose prompt
+        the run-time renders, so the manager-workers component accepts exactly the inputs
+        the group manager accepts (e.g. the ``{{placeholder}}`` inputs of an ``Agent``
+        group manager). Without this, the base default infers no inputs, so a
+        ``ManagerWorkers`` used as a flow ``AgentNode`` would expose no input ports and a
+        data-flow edge into it could not resolve.
+        """
+        group_manager = getattr(self, "group_manager", None)
+        return list(getattr(group_manager, "inputs", None) or [])
+
+    def _get_inferred_outputs(self) -> List[Property]:
+        """Outputs of the group manager; see :meth:`_get_inferred_inputs`."""
+        group_manager = getattr(self, "group_manager", None)
+        return list(getattr(group_manager, "outputs", None) or [])
+
     @model_validator_with_error_accumulation
     def _validate_one_or_more_workers(self) -> Self:
         if len(self.workers) == 0:

pyagentspec/src/pyagentspec/swarm.py

@@ -15,6 +15,7 @@
 
 from pyagentspec.agenticcomponent import AgenticComponent
 from pyagentspec.component import SerializeAsEnum
+from pyagentspec.property import Property
 from pyagentspec.validation_helpers import model_validator_with_error_accumulation
 from pyagentspec.versioning import AgentSpecVersionEnum
 
@@ -127,6 +128,24 @@ class Swarm(AgenticComponent):
         default=AgentSpecVersionEnum.v25_4_2, init=False, exclude=True
     )
 
+    def _get_inferred_inputs(self) -> List[Property]:
+        """A ``Swarm`` exposes the inputs of its entry agent (``first_agent``).
+
+        Symmetric with :meth:`ManagerWorkers._get_inferred_inputs`. The ``first_agent``
+        is the swarm's entry point — it interacts with the user before any handoff — so
+        the swarm component accepts exactly the inputs that agent accepts (e.g. the
+        ``{{placeholder}}`` inputs of an ``Agent`` entry's prompt). Without this, the base
+        default infers no inputs, so a flow ``AgentNode`` wrapping a swarm declares no
+        input ports and a ``DataFlowEdge`` into it fails to resolve at load.
+        """
+        first_agent = getattr(self, "first_agent", None)
+        return list(getattr(first_agent, "inputs", None) or [])
+
+    def _get_inferred_outputs(self) -> List[Property]:
+        """Outputs of the entry agent (``first_agent``); see :meth:`_get_inferred_inputs`."""
+        first_agent = getattr(self, "first_agent", None)
+        return list(getattr(first_agent, "outputs", None) or [])
+
     @model_validator(mode="before")
     def _raise_warning_if_handoff_is_bool(cls: Self, values: Any) -> Any:
         import warnings

pyagentspec/tests/adapters/langgraph/flows/test_agentnode.py

@@ -96,6 +96,97 @@ def test_agentnode_can_be_imported_and_executed(agent_flow: Flow) -> None:
     assert "car" in outputs
 
 
+def test_is_single_string_output() -> None:
+    """A lone string output is treated as free text, not a structured field."""
+    from pyagentspec.adapters.langgraph._node_execution import is_single_string_output
+    from pyagentspec.property import IntegerProperty, StringProperty
+
+    assert is_single_string_output([StringProperty(title="x")]) is True
+    assert is_single_string_output([]) is False
+    assert is_single_string_output([IntegerProperty(title="n")]) is False
+    assert is_single_string_output([StringProperty(title="a"), StringProperty(title="b")]) is False
+
+
+def test_single_string_output_taken_from_final_message_without_structured_generation() -> None:
+    """An AgentNode whose agent declares a single string output should resolve
+    that output from the agent's final message — no structured generation, so
+    it works on models without structured-output support.
+
+    The model is stubbed with a ``FakeMessagesListChatModel`` (no structured
+    output); if the converter still attached a ``response_format`` the output
+    would come back empty. Asserting it equals the message content proves the
+    single-string path takes the final message instead.
+    """
+    from unittest.mock import patch
+
+    from langchain_core.language_models.fake_chat_models import (
+        FakeMessagesListChatModel,
+    )
+    from langchain_core.messages import AIMessage
+    from langchain_openai import ChatOpenAI
+    from langgraph.checkpoint.memory import MemorySaver
+
+    from pyagentspec.adapters.langgraph import AgentSpecLoader
+    from pyagentspec.adapters.langgraph._langgraphconverter import (
+        AgentSpecToLangGraphConverter,
+    )
+    from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig
+
+    class _FakeModel(FakeMessagesListChatModel, ChatOpenAI):
+        pass
+
+    fake_llm = _FakeModel(responses=[AIMessage(content="42")])
+
+    answer = StringProperty(title="answer")
+    agent = Agent(
+        name="agent",
+        llm_config=OpenAiCompatibleConfig(name="agent_llm", model_id="fake", url="null"),
+        system_prompt="Answer the question.",
+        outputs=[answer],
+    )
+    agent_node = AgentNode(name="agent_node", agent=agent)
+    start_node = StartNode(name="start")
+    end_node = EndNode(name="end", outputs=[answer])
+    flow = Flow(
+        name="flow",
+        start_node=start_node,
+        nodes=[start_node, agent_node, end_node],
+        control_flow_connections=[
+            ControlFlowEdge(name="start_to_node", from_node=start_node, to_node=agent_node),
+            ControlFlowEdge(name="node_to_end", from_node=agent_node, to_node=end_node),
+        ],
+        data_flow_connections=[
+            DataFlowEdge(
+                name="answer_edge",
+                source_node=agent_node,
+                source_output=answer.title,
+                destination_node=end_node,
+                destination_input=answer.title,
+            ),
+        ],
+        outputs=[answer],
+    )
+
+    loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver())
+    with patch.object(
+        AgentSpecToLangGraphConverter,
+        "_llm_convert_to_langgraph",
+        autospec=True,
+        side_effect=lambda self_obj, llm_config, *a, **k: fake_llm,
+    ), patch.object(
+        FakeMessagesListChatModel,
+        "bind_tools",
+        new=lambda self_obj, *a, **k: self_obj,
+    ):
+        compiled = loader.load_component(flow)
+        result = compiled.invoke(
+            {"inputs": {}, "messages": [{"role": "user", "content": "What is 6*7?"}]},
+            {"configurable": {"thread_id": "agentnode-single-string"}},
+        )
+
+    assert result["outputs"]["answer"] == "42"
+
+
 @pytest.mark.anyio
 @retry_test(max_attempts=3, wait_between_tries=2)
 async def test_agentnode_can_be_executed_async(agent_flow: Flow) -> None: