diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py index cc461558..fc00e5fb 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py @@ -38,9 +38,9 @@ from pyagentspec.adapters.langgraph._managerworkers import ( _MANAGER_NODE_KEY, _append_workers_roster, + _make_manager_router, _make_worker_delegation_tool, _patch_with_manager_workers_execution_span, - _route_manager_to_worker_or_end, _safe_node_name, _wrap_worker_for_subgraph, ) @@ -1085,13 +1085,13 @@ def _manager_workers_convert_to_langgraph( Topology:: - ┌─ delegate_to_w1 ─→ worker_1 ─┐ - START → manager ┤ ├→ manager (loop) - └─ delegate_to_w2 ─→ worker_2 ─┘ + ┌─ __delegate_to__w1 ─→ worker_1 ─┐ + START → manager ┤ ├→ manager (loop) + └─ __delegate_to__w2 ─→ worker_2 ─┘ │ └─ no tool_call ─→ END - The manager is a react-agent holding one synthetic ``delegate_to_`` + The manager is a react-agent holding one synthetic ``__delegate_to__`` tool per worker. A conditional edge routes each delegation to its worker, which runs in an isolated message context and answers with a ``ToolMessage`` matched to the pending delegation id. Workers are converted recursively and @@ -1160,7 +1160,7 @@ def _manager_workers_convert_to_langgraph( builder.add_edge(langgraph_graph.START, _MANAGER_NODE_KEY) builder.add_conditional_edges( _MANAGER_NODE_KEY, - _route_manager_to_worker_or_end, + _make_manager_router(worker_node_names), # The path map covers every worker plus END, so langgraph can validate # the routing statically. {node_name: node_name for node_name in worker_node_names} diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py index c331722e..05077eec 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py @@ -7,13 +7,13 @@ """Helpers for compiling a ``ManagerWorkers`` into LangGraph, orchestrated by ``AgentSpecToLangGraphConverter._manager_workers_convert_to_langgraph``. -The delegation protocol is visible on purpose: ``delegate_to_`` calls stream -like any other tool call. Consumers that would rather not render them can filter on -:func:`is_delegation_tool_name`. +The delegation protocol is visible on purpose: ``__delegate_to__`` calls +stream like any other tool call. Consumers that would rather not render them can +filter on :func:`is_delegation_tool_name`. """ import re -from typing import Any, Dict, List, Tuple +from typing import Any, Dict, Iterable, List, Tuple from pyagentspec.adapters.langgraph._execution_span import patch_with_execution_span from pyagentspec.adapters.langgraph._types import CompiledStateGraph, langgraph_graph @@ -31,9 +31,11 @@ # Cannot collide with a normalized worker node name, which is always [a-z0-9_]. _MANAGER_NODE_KEY = "__manager__" -#: Prefix of the synthetic ``delegate_to_`` tool names the manager's LLM uses -#: to address a worker. Public so consumers can recognize the protocol. -DELEGATE_TOOL_PREFIX = "delegate_to_" +#: Prefix of the synthetic ``__delegate_to__`` tool names the manager's LLM +#: uses to address a worker. The dunder prefix, like the delegation keys below, keeps +#: it from colliding with a real tool named ``delegate_to_``. Public so +#: consumers can recognize the protocol. +DELEGATE_TOOL_PREFIX = "__delegate_to__" # Keys of the per-delegation ``Send`` payload: the task to run, and the tool_call_id # the worker's reply must answer. Routing per delegation (instead of off shared state) @@ -45,7 +47,7 @@ def is_delegation_tool_name(name: Any) -> bool: - """True for the synthetic ``delegate_to_`` tool names a manager emits.""" + """True for the synthetic ``__delegate_to__`` tool names a manager emits.""" return isinstance(name, str) and name.startswith(DELEGATE_TOOL_PREFIX) @@ -57,7 +59,7 @@ def _normalize_identifier(s: str) -> str: def _safe_node_name(name: str, fallback_id: str) -> str: """Normalize a worker name into a LangGraph node identifier. - The LLM has to emit ``delegate_to_`` reliably as a tool name, so node + The LLM has to emit ``__delegate_to__`` reliably as a tool name, so node names stay ASCII identifiers. Falls back to the normalized component id when the name slugifies to nothing. """ @@ -87,12 +89,12 @@ def _append_workers_roster(system_prompt: str, entries: List[Tuple[str, str]]) - def _make_worker_delegation_tool(worker_node_name: str) -> Any: - """Build the ``delegate_to_`` tool the manager's LLM emits to route to a - worker. + """Build the ``__delegate_to__`` tool the manager's LLM emits to route to + a worker. Executing the tool is only how the call escapes the react subgraph: its body surfaces the subgraph messages to the parent with ``Command(graph=PARENT)`` and no - ``goto``. Routing stays in :func:`_route_manager_to_worker_or_end`; a ``goto`` here + ``goto``. Routing stays in the edge built by :func:`_make_manager_router`; a ``goto`` here would collapse several same-turn delegations into one parent Command and leave the other ``tool_call_id``s unanswered. """ @@ -123,33 +125,45 @@ def _delegate( return _delegate -def _route_manager_to_worker_or_end(state: Dict[str, Any]) -> Any: - """Route the parent graph off the manager's last AIMessage: one ``Send`` per - ``delegate_to_`` tool call, or ``END`` when it emitted none. +def _make_manager_router(worker_node_names: Iterable[str]) -> Any: + """Build the conditional edge routing the parent graph off the manager's last + AIMessage: one ``Send`` per ``__delegate_to__`` tool call, or ``END`` + when it emitted none. Every delegation gets its own ``Send``, so each tool_call_id is answered independently; an unanswered one breaks the manager's next-turn tool-call/result - sequence. Plain tool calls already ran inside the react loop. + sequence. Plain tool calls already ran inside the react loop; that includes a + real tool whose name merely starts with the prefix, which is why a suffix that + is not a worker node is not routed. """ - from langgraph.types import Send - - messages = state.get("messages") or [] - last = messages[-1] if messages else None - sends = [] - for tool_call in getattr(last, "tool_calls", None) or []: - name = tool_call.get("name") - if is_delegation_tool_name(name): + known_workers = frozenset(worker_node_names) + + def _route_manager_to_worker_or_end(state: Dict[str, Any]) -> Any: + from langgraph.types import Send + + messages = state.get("messages") or [] + last = messages[-1] if messages else None + sends = [] + for tool_call in getattr(last, "tool_calls", None) or []: + name = tool_call.get("name") + if not is_delegation_tool_name(name): + continue + worker_node_name = name[len(DELEGATE_TOOL_PREFIX) :] + if worker_node_name not in known_workers: + continue args = tool_call.get("args") or {} sends.append( Send( - name[len(DELEGATE_TOOL_PREFIX) :], + worker_node_name, { _DELEGATE_TASK_KEY: args.get("task") or "", _DELEGATE_CALL_ID_KEY: tool_call.get("id") or "", }, ) ) - return sends or langgraph_graph.END + return sends or langgraph_graph.END + + return _route_manager_to_worker_or_end def _worker_input(state: Dict[str, Any]) -> Dict[str, Any]: diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py index f3df8873..915abc1c 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py @@ -493,6 +493,17 @@ def __init__( super().__init__(node) if not isinstance(self.node, AgentSpecAgentNode): raise TypeError("AgentNodeExecutor can only be initialized with AgentNode") + if isinstance(self.node.agent, AgentSpecManagerWorkers): + # The hierarchical graph runs over MessagesState, which cannot carry a + # structured_response outward: the manager's final message is the only + # result, so anything but a single string output cannot be honored. + # Raising here fails at conversion time rather than mid-run. + outputs = self.node.outputs or [] + if outputs and (len(outputs) != 1 or outputs[0].type != "string"): + raise NotImplementedError( + "A ManagerWorkers flow step supports a single string output; " + f"node `{self.node.name}` declares {[o.title for o in outputs]}." + ) self.tool_registry = tool_registry self.checkpointer = checkpointer self.converted_components = converted_components @@ -607,13 +618,9 @@ def _format_agent_result(self, result: Dict[str, Any]) -> ExecuteOutput: if isinstance(self.node.agent, AgentSpecManagerWorkers): # The hierarchical graph runs over MessagesState, which cannot carry a # structured_response outward: the manager's final message is the result. - outputs = self.node.outputs - if len(outputs) != 1 or outputs[0].type != "string": - raise NotImplementedError( - "A ManagerWorkers flow step supports a single string output; " - f"node `{self.node.name}` declares {[o.title for o in outputs]}." - ) - return {outputs[0].title: result["messages"][-1].content}, NodeExecutionDetails() + # __init__ already rejected any shape but a single string output. + node_outputs = self.node.outputs or [] + return {node_outputs[0].title: result["messages"][-1].content}, NodeExecutionDetails() outputs = extract_outputs_from_invoke_result(result, self.node.outputs or []) return outputs, NodeExecutionDetails() diff --git a/pyagentspec/src/pyagentspec/managerworkers.py b/pyagentspec/src/pyagentspec/managerworkers.py index f89853db..f38526a7 100644 --- a/pyagentspec/src/pyagentspec/managerworkers.py +++ b/pyagentspec/src/pyagentspec/managerworkers.py @@ -67,10 +67,11 @@ class ManagerWorkers(AgenticComponent): ) def _get_inferred_inputs(self) -> List[Property]: - # The group manager drives the conversation and is the component whose prompt - # the runtime renders, so the group accepts exactly the inputs the manager - # accepts. The hasattr guard matches Flow._get_inferred_inputs: validators can - # run this against a partially-constructed model with no group_manager yet. + # Per the language spec, the inputs of a ManagerWorkers are the inputs of its + # group manager (same name and type): the manager drives the conversation and + # is the component whose prompt the runtime renders. The hasattr guard matches + # Flow._get_inferred_inputs: validators can run this against a + # partially-constructed model with no group_manager yet. return (self.group_manager.inputs or []) if hasattr(self, "group_manager") else [] def _get_inferred_outputs(self) -> List[Property]: @@ -91,3 +92,26 @@ def _validate_group_manager_is_not_included_as_a_worker(self) -> Self: if any(self.group_manager is agent for agent in self.workers): raise ValueError("Group manager cannot be a worker.") return self + + @model_validator_with_error_accumulation + def _validate_ios_match_group_manager_ios(self) -> Self: + # Per the language spec, the I/Os of a ManagerWorkers must be the I/Os of its + # group manager, same name and type. The base ComponentWithIO validators + # already enforce matching titles; enforce matching types here. + if not hasattr(self, "group_manager"): + return self + for kind, own_properties, manager_properties in ( + ("input", self.inputs or [], self.group_manager.inputs or []), + ("output", self.outputs or [], self.group_manager.outputs or []), + ): + manager_type_by_title = {p.title: p.type for p in manager_properties} + for own_property in own_properties: + manager_type = manager_type_by_title.get(own_property.title) + if manager_type is not None and own_property.type != manager_type: + raise ValueError( + f"The {kind}s of a `ManagerWorkers` must match the {kind}s of its " + f"group manager (same name and type), but {kind} " + f"`{own_property.title}` has type `{own_property.type}` while the " + f"group manager declares `{manager_type}`." + ) + return self diff --git a/pyagentspec/tests/adapters/conftest.py b/pyagentspec/tests/adapters/conftest.py index 8b5714ba..d5732212 100644 --- a/pyagentspec/tests/adapters/conftest.py +++ b/pyagentspec/tests/adapters/conftest.py @@ -6,8 +6,6 @@ import os import ssl -from contextlib import ExitStack -from importlib import import_module from pathlib import Path from typing import Any from unittest.mock import patch @@ -114,34 +112,6 @@ def _skip(*_args, **_kwargs): p.stop() -def _resolve(dotted: str) -> Any: - module_path, _, attr = dotted.rpartition(".") - module_path, _, cls_name = module_path.rpartition(".") - return getattr(getattr(import_module(module_path), cls_name), attr) - - -# Captured at conftest import, before any session fixture starts patching, so these -# are the genuine constructors rather than a skip stub. -_REAL_LLM_INITS = {dotted: _resolve(dotted) for dotted in LLM_MOCKED_METHODS} - - -@pytest.fixture -def allow_llm_config_construction(): - """ - Opt out of the SKIP_LLM_TESTS=1 construction guard for one test, restoring the - real LLM config constructors. Only request this from a test that stubs the model - and never reaches an endpoint; such tests should run offline instead of skipping. - """ - if not should_skip_llm_test(): - # Nothing patched the constructors, so there is nothing to restore. - yield - return - with ExitStack() as stack: - for dotted, real in _REAL_LLM_INITS.items(): - stack.enter_context(patch(dotted, new=real)) - yield - - @pytest.fixture(scope="package") def json_server(json_server_port: int): api_server = Path(__file__).parent / "api_server.py" diff --git a/pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py b/pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py deleted file mode 100644 index db551602..00000000 --- a/pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py +++ /dev/null @@ -1,140 +0,0 @@ -# Copyright © 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. - -from unittest.mock import patch - -import pytest - -from pyagentspec.agent import Agent -from pyagentspec.llms import OpenAiCompatibleConfig -from pyagentspec.managerworkers import ManagerWorkers -from pyagentspec.property import StringProperty - -# These tests only need an LLM config object and stub the chat model, so they run -# offline even under SKIP_LLM_TESTS=1. -pytestmark = pytest.mark.usefixtures("allow_llm_config_construction") - - -def _llm_config() -> OpenAiCompatibleConfig: - return OpenAiCompatibleConfig(name="agent_llm", model_id="fake", url="null") - - -def test_managerworkers_infers_inputs_from_group_manager_prompt() -> None: - """A ManagerWorkers exposes the group manager's prompt placeholders as inputs, so - a flow AgentNode wrapping it declares input ports a DataFlowEdge can resolve.""" - cfg = _llm_config() - manager = Agent( - name="manager", - llm_config=cfg, - system_prompt="Translate the following to Arabic:\n\n{{joke}}\n\nMake {{count}} variants.", - ) - worker = Agent(name="worker", llm_config=cfg, system_prompt="You translate.") - mw = ManagerWorkers(name="mw", group_manager=manager, workers=[worker]) - - assert sorted(p.title for p in (mw.inputs or [])) == ["count", "joke"] - - -def test_managerworkers_infers_outputs_from_group_manager() -> None: - """Symmetric with inputs: a ManagerWorkers exposes the group manager's outputs.""" - cfg = _llm_config() - manager = Agent( - name="manager", - llm_config=cfg, - system_prompt="Answer the question.", - outputs=[StringProperty(title="answer")], - ) - worker = Agent(name="worker", llm_config=cfg, system_prompt="You help.") - mw = ManagerWorkers(name="mw", group_manager=manager, workers=[worker]) - - assert [p.title for p in (mw.outputs or [])] == ["answer"] - - -def test_managerworkers_runs_as_a_flow_step_with_data_edge_inputs() -> None: - """A ManagerWorkers flow step loads with its data edge resolved and executes: - loading proves the node exposes the ``joke`` input the edge targets, running - proves the manager's answer comes back as the node's single string output.""" - 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.flows.edges import ControlFlowEdge, DataFlowEdge - from pyagentspec.flows.flow import Flow - from pyagentspec.flows.nodes import AgentNode, EndNode, StartNode - - class _FakeModel(FakeMessagesListChatModel, ChatOpenAI): - pass - - # The final message has no tool_calls → the manager routes to END without delegating. - fake_llm = _FakeModel(responses=[AIMessage(content="لماذا...")]) - - cfg = _llm_config() - joke = StringProperty(title="joke") - translated = StringProperty(title="translated") - - manager = Agent( - name="manager", - llm_config=cfg, - system_prompt="Translate the following to Arabic:\n\n{{joke}}", - outputs=[translated], - ) - worker = Agent(name="worker", llm_config=cfg, system_prompt="You translate.") - mw = ManagerWorkers(name="translator", group_manager=manager, workers=[worker]) - assert [p.title for p in (mw.inputs or [])] == ["joke"] - - manager_node = AgentNode(name="manager_node", agent=mw) - start_node = StartNode(name="start", inputs=[joke]) - end_node = EndNode(name="end", outputs=[translated]) - flow = Flow( - name="flow", - start_node=start_node, - nodes=[start_node, manager_node, end_node], - control_flow_connections=[ - ControlFlowEdge(name="start_to_node", from_node=start_node, to_node=manager_node), - ControlFlowEdge(name="node_to_end", from_node=manager_node, to_node=end_node), - ], - data_flow_connections=[ - DataFlowEdge( - name="joke_edge", - source_node=start_node, - source_output=joke.title, - destination_node=manager_node, - destination_input=joke.title, - ), - DataFlowEdge( - name="translated_edge", - source_node=manager_node, - source_output=translated.title, - destination_node=end_node, - destination_input=translated.title, - ), - ], - outputs=[translated], - ) - - 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": {"joke": "Why did the car..."}, - "messages": [{"role": "user", "content": ""}], - }, - {"configurable": {"thread_id": "managerworkers-node"}}, - ) - - assert result["outputs"]["translated"] == "لماذا..." diff --git a/pyagentspec/tests/adapters/langgraph/test_managerworkers.py b/pyagentspec/tests/adapters/langgraph/test_managerworkers.py index da90d28c..96cdc656 100644 --- a/pyagentspec/tests/adapters/langgraph/test_managerworkers.py +++ b/pyagentspec/tests/adapters/langgraph/test_managerworkers.py @@ -4,7 +4,7 @@ # (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. -from typing import Any +from typing import Any, List, Optional from unittest.mock import patch import pytest @@ -12,18 +12,22 @@ from pyagentspec.agent import Agent from pyagentspec.llms import OpenAiCompatibleConfig from pyagentspec.managerworkers import ManagerWorkers +from pyagentspec.property import Property -# Every test stubs the chat model and never reaches an endpoint, so they run offline -# even under SKIP_LLM_TESTS=1. -pytestmark = pytest.mark.usefixtures("allow_llm_config_construction") - -def _agent(name: str, llm_name: str, description: str = "", system_prompt: str = ".") -> Agent: +def _agent( + name: str, + llm_name: str, + description: str = "", + system_prompt: str = ".", + outputs: Optional[List[Property]] = None, +) -> Agent: return Agent( name=name, description=description, system_prompt=system_prompt, llm_config=OpenAiCompatibleConfig(name=llm_name, model_id="fake", url="null"), + outputs=outputs, ) @@ -102,11 +106,12 @@ def test_route_manager_to_worker_or_end_returns_end_when_no_delegation() -> None from langchain_core.messages import AIMessage from langgraph.graph import END - from pyagentspec.adapters.langgraph._managerworkers import _route_manager_to_worker_or_end + from pyagentspec.adapters.langgraph._managerworkers import _make_manager_router + route = _make_manager_router(["drafter"]) not_delegating = AIMessage(content="Done.", tool_calls=[]) - assert _route_manager_to_worker_or_end({"messages": [not_delegating]}) == END - assert _route_manager_to_worker_or_end({"messages": []}) == END + assert route({"messages": [not_delegating]}) == END + assert route({"messages": []}) == END def test_route_manager_to_worker_or_end_fans_out_one_send_per_delegation() -> None: @@ -116,18 +121,19 @@ def test_route_manager_to_worker_or_end_fans_out_one_send_per_delegation() -> No from pyagentspec.adapters.langgraph._managerworkers import ( _DELEGATE_CALL_ID_KEY, _DELEGATE_TASK_KEY, - _route_manager_to_worker_or_end, + _make_manager_router, ) + route = _make_manager_router(["drafter", "research_helper"]) msg = AIMessage( content="", tool_calls=[ {"name": "some_other_tool", "args": {}, "id": "c0"}, - {"name": "delegate_to_drafter", "args": {"task": "x"}, "id": "c1"}, - {"name": "delegate_to_research_helper", "args": {"task": "y"}, "id": "c2"}, + {"name": "__delegate_to__drafter", "args": {"task": "x"}, "id": "c1"}, + {"name": "__delegate_to__research_helper", "args": {"task": "y"}, "id": "c2"}, ], ) - sends = _route_manager_to_worker_or_end({"messages": [msg]}) + sends = route({"messages": [msg]}) # Every delegation gets its own Send carrying the task and the tool_call_id its # reply must answer. The non-delegation tool call already ran inside the manager's # react loop and is ignored by routing. @@ -137,6 +143,23 @@ def test_route_manager_to_worker_or_end_fans_out_one_send_per_delegation() -> No assert [s.arg[_DELEGATE_CALL_ID_KEY] for s in sends] == ["c1", "c2"] +def test_route_manager_ignores_prefixed_tool_whose_suffix_is_not_a_worker() -> None: + """A tool call that merely looks like a delegation must not be routed: its suffix + is not a worker node, so a Send would target a non-existing node. It already ran + as a plain tool inside the react loop.""" + from langchain_core.messages import AIMessage + from langgraph.graph import END + + from pyagentspec.adapters.langgraph._managerworkers import _make_manager_router + + route = _make_manager_router(["drafter"]) + msg = AIMessage( + content="", + tool_calls=[{"name": "__delegate_to__nobody", "args": {"task": "x"}, "id": "c1"}], + ) + assert route({"messages": [msg]}) == END + + def test_manager_workers_compiles_to_hierarchical_graph_topology() -> None: from langchain_core.messages import AIMessage from langgraph.graph import START @@ -186,7 +209,7 @@ def test_manager_workers_registers_a_delegation_tool_per_worker() -> None: # react-agent's tools node, so the LLM has the matching contract. manager_subgraph = compiled.builder.nodes[_MANAGER_NODE_KEY].runnable tools_node = manager_subgraph.builder.nodes["tools"].runnable - assert "delegate_to_research_helper" in tools_node.tools_by_name + assert "__delegate_to__research_helper" in tools_node.tools_by_name def test_manager_workers_delegates_and_routes_back_with_tool_message() -> None: @@ -207,7 +230,7 @@ def test_manager_workers_delegates_and_routes_back_with_tool_message() -> None: content="", tool_calls=[ { - "name": "delegate_to_research_helper", + "name": "__delegate_to__research_helper", "args": {"task": "Look up Saturn"}, "id": "call_1", } @@ -256,9 +279,21 @@ def test_manager_workers_answers_every_delegation_in_a_single_turn() -> None: AIMessage( content="", tool_calls=[ - {"name": "delegate_to_sub_agent", "args": {"task": "Spanish poem"}, "id": "call_1"}, - {"name": "delegate_to_sub_agent", "args": {"task": "French poem"}, "id": "call_2"}, - {"name": "delegate_to_sub_agent", "args": {"task": "German poem"}, "id": "call_3"}, + { + "name": "__delegate_to__sub_agent", + "args": {"task": "Spanish poem"}, + "id": "call_1", + }, + { + "name": "__delegate_to__sub_agent", + "args": {"task": "French poem"}, + "id": "call_2", + }, + { + "name": "__delegate_to__sub_agent", + "args": {"task": "German poem"}, + "id": "call_3", + }, ], ), AIMessage(content="Here are your three poems."), @@ -382,7 +417,7 @@ def _manager(state: Any) -> Any: content="", tool_calls=[ { - "name": "delegate_to_research_helper", + "name": "__delegate_to__research_helper", "args": {"task": "Saturn"}, "id": "c1", } @@ -421,11 +456,13 @@ def test_is_delegation_tool_name_matches_only_the_synthetic_prefix() -> None: is_delegation_tool_name, ) - assert DELEGATE_TOOL_PREFIX == "delegate_to_" - assert is_delegation_tool_name("delegate_to_research_helper") + assert DELEGATE_TOOL_PREFIX == "__delegate_to__" + assert is_delegation_tool_name("__delegate_to__research_helper") assert not is_delegation_tool_name("get_weather") - # A real tool merely *containing* the prefix mid-name is not a delegation. - assert not is_delegation_tool_name("please_delegate_to_someone") + # A real tool plausibly named delegate_to_ is not a delegation. + assert not is_delegation_tool_name("delegate_to_someone") + # Nor is one merely *containing* the prefix mid-name. + assert not is_delegation_tool_name("please__delegate_to__someone") assert not is_delegation_tool_name(None) assert not is_delegation_tool_name(123) @@ -444,3 +481,143 @@ def test_manager_workers_leaves_astream_events_unwrapped() -> None: assert "stream" in compiled.__dict__ and "astream" in compiled.__dict__ assert "astream_events" not in compiled.__dict__ + + +def test_managerworkers_infers_inputs_from_group_manager_prompt() -> None: + """A ManagerWorkers exposes the group manager's prompt placeholders as inputs, so + a flow AgentNode wrapping it declares input ports a DataFlowEdge can resolve.""" + manager = _agent( + "manager", + "manager_llm", + system_prompt="Translate the following to Arabic:\n\n{{joke}}\n\nMake {{count}} variants.", + ) + worker = _agent("worker", "worker_llm", system_prompt="You translate.") + mw = ManagerWorkers(name="mw", group_manager=manager, workers=[worker]) + + assert sorted(p.title for p in (mw.inputs or [])) == ["count", "joke"] + + +def test_managerworkers_infers_outputs_from_group_manager() -> None: + """Symmetric with inputs: a ManagerWorkers exposes the group manager's outputs.""" + from pyagentspec.property import StringProperty + + manager = _agent( + "manager", + "manager_llm", + system_prompt="Answer the question.", + outputs=[StringProperty(title="answer")], + ) + worker = _agent("worker", "worker_llm", system_prompt="You help.") + mw = ManagerWorkers(name="mw", group_manager=manager, workers=[worker]) + + assert [p.title for p in (mw.outputs or [])] == ["answer"] + + +def _flow_with_manager_workers_step(outputs: List[Property]) -> Any: + """A start → AgentNode(ManagerWorkers) → end flow whose end node exposes + ``outputs``, with the data edges resolving the manager's ``joke`` input and + every output.""" + from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge + from pyagentspec.flows.flow import Flow + from pyagentspec.flows.nodes import AgentNode, EndNode, StartNode + from pyagentspec.property import StringProperty + + joke = StringProperty(title="joke") + manager = _agent( + "manager", + "manager_llm", + system_prompt="Translate the following to Arabic:\n\n{{joke}}", + outputs=outputs, + ) + worker = _agent("worker", "worker_llm", system_prompt="You translate.") + mw = ManagerWorkers(name="translator", group_manager=manager, workers=[worker]) + assert [p.title for p in (mw.inputs or [])] == ["joke"] + + manager_node = AgentNode(name="manager_node", agent=mw) + start_node = StartNode(name="start", inputs=[joke]) + end_node = EndNode(name="end", outputs=outputs) + return Flow( + name="flow", + start_node=start_node, + nodes=[start_node, manager_node, end_node], + control_flow_connections=[ + ControlFlowEdge(name="start_to_node", from_node=start_node, to_node=manager_node), + ControlFlowEdge(name="node_to_end", from_node=manager_node, to_node=end_node), + ], + data_flow_connections=[ + DataFlowEdge( + name="joke_edge", + source_node=start_node, + source_output=joke.title, + destination_node=manager_node, + destination_input=joke.title, + ), + ] + + [ + DataFlowEdge( + name=f"{output.title}_edge", + source_node=manager_node, + source_output=output.title, + destination_node=end_node, + destination_input=output.title, + ) + for output in outputs + ], + outputs=outputs, + ) + + +def test_managerworkers_runs_as_a_flow_step_with_data_edge_inputs() -> None: + """A ManagerWorkers flow step loads with its data edge resolved and executes: + loading proves the node exposes the ``joke`` input the edge targets, running + proves the manager's answer comes back as the node's single string output.""" + from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel + from langchain_core.messages import AIMessage + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import AgentSpecToLangGraphConverter + from pyagentspec.property import StringProperty + + # The final message has no tool_calls → the manager routes to END without delegating. + fake_llm = _fake_llm(AIMessage(content="لماذا...")) + flow = _flow_with_manager_workers_step(outputs=[StringProperty(title="translated")]) + + 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": {"joke": "Why did the car..."}, + "messages": [{"role": "user", "content": ""}], + }, + {"configurable": {"thread_id": "managerworkers-node"}}, + ) + + assert result["outputs"]["translated"] == "لماذا..." + + +def test_managerworkers_flow_step_with_unsupported_outputs_fails_at_conversion() -> None: + """A ManagerWorkers flow step supports a single string output only; any other + shape must be rejected when the flow is converted, not once the step runs.""" + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.property import StringProperty + + flow = _flow_with_manager_workers_step( + outputs=[StringProperty(title="translated"), StringProperty(title="notes")] + ) + + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with pytest.raises(NotImplementedError, match="single string output"): + loader.load_component(flow) diff --git a/pyagentspec/tests/validation/test_agentic_patterns_validation.py b/pyagentspec/tests/validation/test_agentic_patterns_validation.py index 81a13af8..f951ce3a 100644 --- a/pyagentspec/tests/validation/test_agentic_patterns_validation.py +++ b/pyagentspec/tests/validation/test_agentic_patterns_validation.py @@ -14,6 +14,7 @@ from pyagentspec.flows.nodes.startnode import StartNode from pyagentspec.llms import OpenAiConfig from pyagentspec.managerworkers import ManagerWorkers +from pyagentspec.property import FloatProperty, StringProperty from pyagentspec.swarm import Swarm @@ -79,6 +80,72 @@ def test_managerworkers_with_different_agentic_components_can_be_validated() -> ) +def test_managerworkers_with_ios_matching_the_group_manager_can_be_validated() -> None: + manager_agent = Agent( + name="manager_agent", + system_prompt="Answer about {{topic}}.", + llm_config=OpenAiConfig(name="default", model_id="test_model"), + outputs=[StringProperty(title="answer")], + ) + worker_agent = Agent( + name="worker_agent", + system_prompt="You help.", + llm_config=OpenAiConfig(name="default", model_id="test_model"), + ) + + # The I/Os of a ManagerWorkers must be the I/Os of its group manager (same name + # and type); redeclaring them explicitly is valid. + _ = ManagerWorkers( + name="managerworkers", + group_manager=manager_agent, + workers=[worker_agent], + inputs=[StringProperty(title="topic")], + outputs=[StringProperty(title="answer")], + ) + + +def test_managerworkers_with_ios_not_matching_the_group_manager_raises_errors() -> None: + manager_agent = Agent( + name="manager_agent", + system_prompt="Answer about {{topic}}.", + llm_config=OpenAiConfig(name="default", model_id="test_model"), + outputs=[StringProperty(title="answer")], + ) + worker_agent = Agent( + name="worker_agent", + system_prompt="You help.", + llm_config=OpenAiConfig(name="default", model_id="test_model"), + ) + + # Same title as a group manager input, but a different type. + with pytest.raises(ValueError, match="must match the inputs of its group manager"): + ManagerWorkers( + name="managerworkers", + group_manager=manager_agent, + workers=[worker_agent], + inputs=[FloatProperty(title="topic")], + ) + + # Same title as a group manager output, but a different type. + with pytest.raises(ValueError, match="must match the outputs of its group manager"): + ManagerWorkers( + name="managerworkers", + group_manager=manager_agent, + workers=[worker_agent], + outputs=[FloatProperty(title="answer")], + ) + + # A title the group manager does not declare is rejected by the base + # ComponentWithIO validation. + with pytest.raises(ValueError, match="expected only properties with the titles"): + ManagerWorkers( + name="managerworkers", + group_manager=manager_agent, + workers=[worker_agent], + outputs=[StringProperty(title="answer"), StringProperty(title="extra")], + ) + + def test_swarm_with_empty_relationships_raises_errors() -> None: first_agent = Agent( name="first_agent",