diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_execution_span.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_execution_span.py index 21817f46..cb22b199 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_execution_span.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_execution_span.py @@ -7,13 +7,11 @@ """Wrap a compiled graph's ``stream``/``astream`` in an Agent Spec execution span. Agent, Flow and ManagerWorkers graphs all need the same wrapper: open a span, emit a -Start event carrying the invocation inputs, replay the chunks the underlying stream -yields while remembering the last state chunk, then emit an End event built from that -final state. Only the span class and the two event payloads differ, so they come in -as factories. - -``invoke``/``ainvoke`` need no patch of their own; they go through ``stream``/ -``astream`` internally. +Start event carrying the invocation inputs, yield the chunks the underlying stream +produces while remembering the last state chunk, then emit an End event built from +that final state. Only the span class and the two event payloads differ, so they come +in as factories. ``invoke``/``ainvoke`` need no patch; they use ``stream``/``astream`` +internally. """ from typing import Any, AsyncGenerator, Callable, Dict, Generator diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py index da908d9b..cc461558 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py @@ -1092,21 +1092,15 @@ def _manager_workers_convert_to_langgraph( └─ no tool_call ─→ END The manager is a react-agent holding one synthetic ``delegate_to_`` - tool per worker. The parent graph's conditional edge inspects its last - AIMessage to pick the next node, and the worker node runs in an isolated - message context and answers with a ``ToolMessage`` matched to the pending - delegation id. - - Workers are converted recursively and wired in as subgraph nodes, so - ``astream_events`` still exposes the parent/child boundary - (``subgraph=True``) for tracing and SSE streaming. Workers that are - themselves ``ManagerWorkers`` compose through ``self.convert(...)``. + 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 + wired in as subgraph nodes, so ``astream_events`` still exposes the + parent/child boundary (``subgraph=True``) for tracing and streaming. """ if not isinstance(mw.group_manager, AgentSpecAgent): - # The manager has to decide which worker to delegate to, so it needs a - # chat-LLM that emits tool_calls. Only Agent (and its SpecializedAgent - # subclass) has that shape; a Flow, Swarm or nested ManagerWorkers gives - # us nothing to route on. + # Delegation is routed off the manager's tool_calls, so the manager needs + # a chat-LLM; a Flow, Swarm or nested ManagerWorkers gives nothing to route on. raise NotImplementedError( f"ManagerWorkers.group_manager must be an Agent for LangGraph " f"conversion; got {type(mw.group_manager).__name__}." @@ -1137,9 +1131,8 @@ def _manager_workers_convert_to_langgraph( [(node_name, worker.description or "") for node_name, worker in named_workers], ) - # The delegation tools do execute inside the react loop: their body returns a - # Command(graph=PARENT), which is how the call escapes the react subgraph so - # the conditional edge below can route on it. + # The delegation tools execute inside the react loop: their Command(graph=PARENT) + # is how the call escapes the subgraph so the conditional edge below can route on it. manager_graph = self._create_react_agent_with_given_info( name=manager_agent.name, system_prompt=rendered_prompt, @@ -1258,7 +1251,9 @@ def _create_react_agent_with_given_info( make_span=lambda: AgentSpecAgentExecutionSpan( name=f"AgentExecution[{agent.name}]", agent=agent ), - make_start_event=lambda inputs: AgentSpecAgentExecutionStart(agent=agent, inputs=inputs), + make_start_event=lambda inputs: AgentSpecAgentExecutionStart( + agent=agent, inputs=inputs + ), make_end_event=lambda result: AgentSpecAgentExecutionEnd( agent=agent, outputs=extract_outputs_from_invoke_result(result, agent.outputs or []), diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py index 971fb2ae..c331722e 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py @@ -4,19 +4,15 @@ # (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. -"""Helpers for compiling a ``ManagerWorkers`` into LangGraph. +"""Helpers for compiling a ``ManagerWorkers`` into LangGraph, orchestrated by +``AgentSpecToLangGraphConverter._manager_workers_convert_to_langgraph``. -``AgentSpecToLangGraphConverter._manager_workers_convert_to_langgraph`` orchestrates -these; they live here to keep the converter module from growing further. - -Nothing hides the routing protocol. ``delegate_to_`` calls stream like any -other tool call, because which worker got which task is usually the most useful thing -a run reports. Consumers that would rather not render it can filter on +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 functools import lru_cache from typing import Any, Dict, List, Tuple from pyagentspec.adapters.langgraph._execution_span import patch_with_execution_span @@ -35,13 +31,13 @@ # Cannot collide with a normalized worker node name, which is always [a-z0-9_]. _MANAGER_NODE_KEY = "__manager__" -#: Prefix the manager's LLM uses to address a delegation tool, suffixed with the -#: normalized worker node name. Public so consumers can recognize the protocol. +#: 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_" -# Carried on the per-delegation ``Send`` payload so a worker run knows its task and -# which ``tool_call_id`` its reply must answer. Routing per delegation instead of off -# shared state lets one manager turn delegate to several workers at once. +# 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) +# lets one manager turn delegate to several workers at once. _DELEGATE_TASK_KEY = "__delegate_task__" _DELEGATE_CALL_ID_KEY = "__delegate_tool_call_id__" @@ -61,21 +57,13 @@ 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 sees ``delegate_to_`` as a tool name and has to emit it - reliably, so node names stay ASCII identifiers. Falls back to the component id, - normalized the same way, when the name slugifies to nothing. + 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. """ return _normalize_identifier(name) or _normalize_identifier(fallback_id) or "worker" -def _tc_get(tool_call: Any, key: str) -> Any: - """Read ``key`` off a tool call, which langchain emits as a dict or an object - depending on the message source.""" - if isinstance(tool_call, dict): - return tool_call.get(key) - return getattr(tool_call, key, None) - - def _messages_of(state: Any) -> List[Any]: """Read ``messages`` off a state, which langgraph injects as a dict or an object.""" if isinstance(state, dict): @@ -83,23 +71,7 @@ def _messages_of(state: Any) -> List[Any]: return list(getattr(state, "messages", []) or []) -def _surface_to_parent_command(state: Any) -> Any: - """Break out of the manager's react loop, projecting the subgraph's messages onto - the parent state (including the AIMessage carrying the triggering tool call). - - Carries no ``goto``: routing is the parent graph's job. The ``add_messages`` - reducer dedupes by id, so re-surfacing existing messages is a no-op. Modelled on - ``langgraph_swarm.create_handoff_tool``. - """ - from langgraph.types import Command - - return Command(graph=Command.PARENT, update={"messages": _messages_of(state)}) - - -def _append_workers_roster( - system_prompt: str, - entries: List[Tuple[str, str]], -) -> str: +def _append_workers_roster(system_prompt: str, entries: List[Tuple[str, str]]) -> str: """Append an ``Available workers:`` block listing ``- : ``. Descriptions are flattened to one line each, since the LLM routes off the block's @@ -114,19 +86,15 @@ def _append_workers_roster( return f"{system_prompt}\n\n{roster}" if system_prompt else roster -@lru_cache(maxsize=256) 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. - The body carries no ``goto``. Routing fans out one ``Send`` per delegation in - :func:`_route_manager_to_worker_or_end`; a ``goto`` here would collapse several - same-turn delegations into one parent Command and leave the other - ``tool_call_id``s unanswered. - - Memoized because the tool depends only on the node name and holds no per-graph - state. Without it every compile re-runs the ``@tool`` decorator, which costs a - ``get_type_hints`` pass and a pydantic args-schema build. + 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 + would collapse several same-turn delegations into one parent Command and leave the + other ``tool_call_id``s unanswered. """ from typing import Annotated @@ -136,9 +104,8 @@ def _make_worker_delegation_tool(worker_node_name: str) -> Any: tool_name = f"{DELEGATE_TOOL_PREFIX}{worker_node_name}" description = ( - f"Delegate a task to the {worker_node_name} worker and receive " - f"its response. Use this when the task fits the worker's " - f"described capability." + f"Delegate a task to the {worker_node_name} worker and receive its response. " + f"Use this when the task fits the worker's described capability." ) @tool(tool_name, description=description) @@ -146,12 +113,12 @@ def _delegate( task: str, state: Annotated[Any, InjectedState], tool_call_id: Annotated[str, InjectedToolCallId], - ) -> Command: - # Declared for the LLM-facing schema but unused here: executing is only how the - # call escapes the react subgraph, and the routing edge recovers both off the - # surfaced AIMessage's tool_calls. + ) -> Any: + # task and tool_call_id are declared for the LLM-facing schema; the routing + # edge recovers both off the surfaced AIMessage's tool_calls. The + # add_messages reducer dedupes by id, so re-surfacing messages is a no-op. del task, tool_call_id - return _surface_to_parent_command(state) + return Command(graph=Command.PARENT, update={"messages": _messages_of(state)}) return _delegate @@ -160,28 +127,25 @@ 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. - Every delegation gets its own ``Send`` carrying the task and ``tool_call_id``, so - each 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. + 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. """ from langgraph.types import Send messages = state.get("messages") or [] - if not messages: - return langgraph_graph.END - last = messages[-1] - tool_calls = getattr(last, "tool_calls", None) or [] + last = messages[-1] if messages else None sends = [] - for tc in tool_calls: - name = _tc_get(tc, "name") + for tool_call in getattr(last, "tool_calls", None) or []: + name = tool_call.get("name") if is_delegation_tool_name(name): - args = _tc_get(tc, "args") or {} + args = tool_call.get("args") or {} sends.append( Send( name[len(DELEGATE_TOOL_PREFIX) :], { _DELEGATE_TASK_KEY: args.get("task") or "", - _DELEGATE_CALL_ID_KEY: _tc_get(tc, "id") or "", + _DELEGATE_CALL_ID_KEY: tool_call.get("id") or "", }, ) ) @@ -189,13 +153,7 @@ def _route_manager_to_worker_or_end(state: Dict[str, Any]) -> Any: def _worker_input(state: Dict[str, Any]) -> Dict[str, Any]: - """The single-message context a worker run starts from. - - Passes no explicit config, so the worker inherits this node's ambient run config. - Its ``checkpoint_ns`` (``:``) streams the worker's token - events under the worker node, and the per-superstep namespace keeps repeated - delegations isolated without a fresh thread_id. - """ + """The single-message context a worker run starts from.""" from langchain_core.messages import HumanMessage return {"messages": [HumanMessage(content=state.get(_DELEGATE_TASK_KEY) or "")]} @@ -214,51 +172,34 @@ def _worker_reply(state: Dict[str, Any], result: Any) -> Dict[str, Any]: } -class _WorkerSubgraphNode: - """Runs one worker subgraph as a node of the ManagerWorkers parent graph. - - Hierarchical rather than shared-state like a Swarm: workers never see each other's - messages, and each run is handed only the manager's chosen task. The worker's last - message comes back as the ToolMessage content, so the manager's react loop sees a - well-formed tool response on its next turn. - - A class rather than a closure, so the long-lived node holds only the graph instead - of keeping a whole factory frame alive. - """ - - __slots__ = ("_graph",) - - def __init__(self, worker_graph: CompiledStateGraph[Any, Any, Any]) -> None: - self._graph = worker_graph - - def run(self, state: Dict[str, Any]) -> Dict[str, Any]: - return _worker_reply(state, self._graph.invoke(_worker_input(state))) - - async def arun(self, state: Dict[str, Any]) -> Dict[str, Any]: - return _worker_reply(state, await self._graph.ainvoke(_worker_input(state))) - - def _wrap_worker_for_subgraph( worker_graph: CompiledStateGraph[Any, Any, Any], worker_node_name: str, ) -> Any: - """Wrap a worker subgraph as a node exposing sync and async entrypoints. + """Wrap a worker subgraph as a node of the ManagerWorkers parent graph. - LangGraph picks between them depending on whether the parent graph was invoked - via ``invoke`` or ``ainvoke``. + Hierarchical rather than shared-state like a Swarm: each run is handed only the + manager's chosen task, and the worker's answer comes back as a ToolMessage so the + manager's react loop sees a well-formed tool response on its next turn. The worker + is invoked with no explicit config and inherits this node's ambient run config, + which streams its token events under the worker node's checkpoint namespace. """ from pyagentspec.adapters.langgraph._types import RunnableLambda - node = _WorkerSubgraphNode(worker_graph) - return RunnableLambda(func=node.run, afunc=node.arun, name=f"worker:{worker_node_name}") + def run(state: Dict[str, Any]) -> Dict[str, Any]: + return _worker_reply(state, worker_graph.invoke(_worker_input(state))) + + async def arun(state: Dict[str, Any]) -> Dict[str, Any]: + return _worker_reply(state, await worker_graph.ainvoke(_worker_input(state))) + + return RunnableLambda(func=run, afunc=arun, name=f"worker:{worker_node_name}") def _patch_with_manager_workers_execution_span( compiled_graph: CompiledStateGraph[Any, Any, Any], mw: AgentSpecManagerWorkers, ) -> None: - """Wrap ``stream``/``astream`` so each run emits a ``ManagerWorkersExecutionSpan``, - using the same patcher as the Agent and Flow graphs.""" + """Wrap ``stream``/``astream`` so each run emits a ``ManagerWorkersExecutionSpan``.""" patch_with_execution_span( compiled_graph, make_span=lambda: AgentSpecManagerWorkersExecutionSpan( diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py index f8fb156e..f3df8873 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py @@ -543,20 +543,15 @@ def _create_manager_workers_with_given_input_values( The graph runs over ``MessagesState``, which can't carry structured inputs inward to the group manager, so the node inputs are rendered into its - ``system_prompt`` and the satisfied ports dropped from both the manager and - the component. That keeps declared and inferred ports equal for the - downstream span re-validation, and the graph runs on messages alone. - - Cached by rendered prompt, the same key - :meth:`_create_react_agent_with_given_input_values` uses. + ``system_prompt`` and the satisfied ports dropped. Cached by rendered prompt, + the same key :meth:`_create_react_agent_with_given_input_values` uses. """ from pyagentspec.adapters.langgraph._langgraphconverter import AgentSpecToLangGraphConverter converter = AgentSpecToLangGraphConverter() entry_agent = component.group_manager if not isinstance(entry_agent, AgentSpecAgent): - # Not routable. Nothing to render or cache, and the converter owns the - # error message for this case. + # Nothing to render or cache; the converter owns the error for this case. return converter._manager_workers_convert_to_langgraph( component, **self._conversion_kwargs() ) @@ -609,6 +604,17 @@ def _format_agent_result(self, result: Dict[str, Any]) -> ExecuteOutput: ] return {}, NodeExecutionDetails(generated_messages=generated_messages) + 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() + 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 8b4c3b32..f89853db 100644 --- a/pyagentspec/src/pyagentspec/managerworkers.py +++ b/pyagentspec/src/pyagentspec/managerworkers.py @@ -67,22 +67,14 @@ class ManagerWorkers(AgenticComponent): ) def _get_inferred_inputs(self) -> List[Property]: - """A ``ManagerWorkers`` exposes the inputs of its group manager. - - 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 - (for an ``Agent`` manager, its ``{{placeholder}}`` inputs). The base default - infers none, which would leave a ``ManagerWorkers`` used as a flow ``AgentNode`` - with no input ports for a data-flow edge to resolve against. - - The ``hasattr`` guard matches :meth:`Flow._get_inferred_inputs` and - :meth:`AgentNode._get_inferred_inputs`: error-accumulating validators can run - this against a partially-constructed model with no ``group_manager`` assigned. - """ + # 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. return (self.group_manager.inputs or []) if hasattr(self, "group_manager") else [] def _get_inferred_outputs(self) -> List[Property]: - """Outputs of the group manager; see :meth:`_get_inferred_inputs`.""" + # Symmetric with the inferred inputs: the group manager's outputs. return (self.group_manager.outputs or []) if hasattr(self, "group_manager") else [] @model_validator_with_error_accumulation diff --git a/pyagentspec/tests/adapters/conftest.py b/pyagentspec/tests/adapters/conftest.py index b999abc7..8b5714ba 100644 --- a/pyagentspec/tests/adapters/conftest.py +++ b/pyagentspec/tests/adapters/conftest.py @@ -127,17 +127,10 @@ def _resolve(dotted: str) -> Any: @pytest.fixture def allow_llm_config_construction(): - """Opt out of the blanket ``SKIP_LLM_TESTS=1`` construction guard. - - That guard skips a test the moment it constructs an LLM config. Right for tests - that go on to call a model, wrong for tests that only need a config object and - stub the conversion: those should run offline, and instead they skip silently in - CI, leaving the code path they cover unverified. - - Restores the real constructors for one test, overriding the guards in both this - conftest and ``tests/conftest.py``. - - Only request this from a test that provably never reaches a model endpoint. + """ + 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. diff --git a/pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py b/pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py index 35ea5e5a..db551602 100644 --- a/pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py +++ b/pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py @@ -4,37 +4,28 @@ # (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. -"""A ManagerWorkers used as a flow step (AgentNode). - -Regression coverage for two coupled behaviours: - * ``ManagerWorkers._get_inferred_inputs`` exposes the group manager's inputs, so a - flow ``AgentNode`` wrapping a manager declares input ports and a ``DataFlowEdge`` - into it resolves at load (previously: "node does not have any input property..."). - * ``AgentNodeExecutor`` runs a ManagerWorkers node (previously: TypeError "can only - be used with AgentSpecAgent agents"), rendering the node inputs into the group - manager's prompt and returning its result. -""" +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") -@pytest.fixture(autouse=True) -def _offline(allow_llm_config_construction: None) -> None: - """These tests only need an LLM *config* object: the two inference tests never - convert at all, and the flow-step test stubs the chat model. Without this the - SKIP_LLM_TESTS guard skips all three and the flow-step path goes unverified.""" +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.""" - llm = {"name": "m", "model_id": "fake", "url": "null"} - from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig - cfg = OpenAiCompatibleConfig(**llm) +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, @@ -47,18 +38,13 @@ def test_managerworkers_infers_inputs_from_group_manager_prompt() -> None: def test_managerworkers_infers_outputs_from_group_manager() -> None: - """Symmetric with inputs: a ManagerWorkers exposes the group manager's outputs, - so a flow AgentNode wrapping it can wire its result downstream (or surface it as a - leaf).""" - from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig - - cfg = OpenAiCompatibleConfig(name="m", model_id="fake", url="null") - answer = StringProperty(title="answer") + """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=[answer], + outputs=[StringProperty(title="answer")], ) worker = Agent(name="worker", llm_config=cfg, system_prompt="You help.") mw = ManagerWorkers(name="mw", group_manager=manager, workers=[worker]) @@ -67,36 +53,27 @@ def test_managerworkers_infers_outputs_from_group_manager() -> None: 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. - - The model is stubbed, so there is no delegation: the manager produces a final - message and routes straight to END. Loading proves the manager node exposes the - ``joke`` input the data edge targets; running proves the manager's answer comes - back as the node's single string output. - """ - from unittest.mock import patch - + """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.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 - from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig class _FakeModel(FakeMessagesListChatModel, ChatOpenAI): pass - # Final message has no tool_calls → the manager routes to END without delegating. + # The final message has no tool_calls → the manager routes to END without delegating. fake_llm = _FakeModel(responses=[AIMessage(content="لماذا...")]) - cfg = OpenAiCompatibleConfig(name="agent_llm", model_id="fake", url="null") + cfg = _llm_config() joke = StringProperty(title="joke") translated = StringProperty(title="translated") @@ -108,8 +85,6 @@ class _FakeModel(FakeMessagesListChatModel, ChatOpenAI): ) worker = Agent(name="worker", llm_config=cfg, system_prompt="You translate.") mw = ManagerWorkers(name="translator", group_manager=manager, workers=[worker]) - # The manager node exposes the group manager's `joke` input, and the single - # `translated` output (inherited from the group manager) for the leaf edge. assert [p.title for p in (mw.inputs or [])] == ["joke"] manager_node = AgentNode(name="manager_node", agent=mw) @@ -162,5 +137,4 @@ class _FakeModel(FakeMessagesListChatModel, ChatOpenAI): {"configurable": {"thread_id": "managerworkers-node"}}, ) - assert "outputs" in result assert result["outputs"]["translated"] == "لماذا..." diff --git a/pyagentspec/tests/adapters/langgraph/test_managerworkers.py b/pyagentspec/tests/adapters/langgraph/test_managerworkers.py index 0f9f28d2..da90d28c 100644 --- a/pyagentspec/tests/adapters/langgraph/test_managerworkers.py +++ b/pyagentspec/tests/adapters/langgraph/test_managerworkers.py @@ -1,42 +1,35 @@ -# Copyright © 2025 Oracle and/or its affiliates. +# 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. -"""Offline tests for the LangGraph ``ManagerWorkers`` converter. - -These cover the hierarchical topology, roster prompt rendering, the -worker-isolation invariant (each worker sees only its delegated task), -and the recursive nesting case. The LLM is stubbed with -``FakeMessagesListChatModel`` so the tests run without network or model -endpoints. -""" - from typing import Any from unittest.mock import patch import pytest -# ─── Shared helpers ────────────────────────────────────────────────────────── - +from pyagentspec.agent import Agent +from pyagentspec.llms import OpenAiCompatibleConfig +from pyagentspec.managerworkers import ManagerWorkers -@pytest.fixture(autouse=True) -def _offline(allow_llm_config_construction: None) -> None: - """Every test in this module stubs the chat model (``_fake_manager`` or an - explicitly patched ``_llm_convert_to_langgraph``) and never reaches an - endpoint, so the SKIP_LLM_TESTS construction guard would only hide them.""" +# 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 _llm_cfg(name: str) -> Any: - from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig - - return OpenAiCompatibleConfig(name=name, model_id="fake", url="null") +def _agent(name: str, llm_name: str, description: str = "", system_prompt: str = ".") -> Agent: + return Agent( + name=name, + description=description, + system_prompt=system_prompt, + llm_config=OpenAiCompatibleConfig(name=llm_name, model_id="fake", url="null"), + ) -def _fake_manager(*ai_responses: Any) -> Any: - """A FakeMessagesListChatModel subclassed under ChatOpenAI so the - manager's react-agent treats it as an OpenAI-style chat model.""" +def _fake_llm(*ai_responses: Any) -> Any: + """A FakeMessagesListChatModel subclassed under ChatOpenAI so the react-agent + treats it as an OpenAI-style chat model.""" from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel from langchain_openai import ChatOpenAI @@ -47,13 +40,11 @@ class _FakeModel(FakeMessagesListChatModel, ChatOpenAI): def _load_with_fake_llms(mw: Any, default: Any = None, **fakes_by_llm_name: Any) -> Any: - """Compile ``mw`` offline, answering each LLM config with a queued fake. + """Compile ``mw`` offline, answering each LLM config (keyed by ``llm_config.name``) + with a queued fake; ``default`` answers any config not named. - Keys are ``llm_config.name``; ``default`` answers any config not named. - - ``create_agent`` calls ``model.bind_tools(...)``, and ``FakeMessagesListChatModel`` - inherits ``bind_tools`` from the real ``ChatOpenAI``, which calls out to OpenAI. - Binding is stubbed to return the same fake, preserving its response queue. + ``bind_tools`` is stubbed to return the same fake because + ``FakeMessagesListChatModel`` inherits it from ``ChatOpenAI``, which calls OpenAI. """ from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel from langgraph.checkpoint.memory import MemorySaver @@ -79,33 +70,18 @@ def _dispatch(_self: Any, llm_config: Any, *args: Any, **kwargs: Any) -> Any: return loader.load_component(mw) -# ─── Pure-helper unit tests (no LLM) ──────────────────────────────────────── - - -def test_safe_node_name_lowercases_and_collapses_punctuation() -> None: - from pyagentspec.adapters.langgraph._managerworkers import ( - _safe_node_name, - ) +def test_safe_node_name_normalizes_and_falls_back() -> None: + from pyagentspec.adapters.langgraph._managerworkers import _safe_node_name assert _safe_node_name("Research Helper", "id-1") == "research_helper" assert _safe_node_name("My-Worker!! v2", "id-1") == "my_worker_v2" - - -def test_safe_node_name_falls_back_to_normalized_id() -> None: - from pyagentspec.adapters.langgraph._managerworkers import ( - _safe_node_name, - ) - - # Name slugifies to empty → id used (and also normalized). + # Name slugifies to empty → normalized id; both empty → constant fallback. assert _safe_node_name("!!!", "sub-1") == "sub_1" - # Both empty → constant fallback. assert _safe_node_name("", "") == "worker" -def test_append_workers_roster_appends_block_after_existing_prompt() -> None: - from pyagentspec.adapters.langgraph._managerworkers import ( - _append_workers_roster, - ) +def test_append_workers_roster_renders_one_line_per_worker() -> None: + from pyagentspec.adapters.langgraph._managerworkers import _append_workers_roster out = _append_workers_roster( "Coordinate the team.", @@ -117,58 +93,23 @@ def test_append_workers_roster_appends_block_after_existing_prompt() -> None: "- research_helper: Handles research\n" "- drafter: Drafts text" ) - - -def test_append_workers_roster_flattens_multiline_descriptions() -> None: - from pyagentspec.adapters.langgraph._managerworkers import ( - _append_workers_roster, - ) - - out = _append_workers_roster( - "", - [("helper", "First line\nsecond line\n third line ")], - ) - # Whitespace flattened so the one-line-per-worker shape survives. + # Multiline descriptions are flattened so the one-line-per-worker shape survives. + out = _append_workers_roster("", [("helper", "First line\nsecond line\n third line ")]) assert out == "Available workers:\n- helper: First line second line third line" -def test_route_manager_to_worker_or_end_sends_to_pending_delegation() -> None: - from langchain_core.messages import AIMessage - from langgraph.types import Send - - from pyagentspec.adapters.langgraph._managerworkers import ( - _DELEGATE_CALL_ID_KEY, - _DELEGATE_TASK_KEY, - _route_manager_to_worker_or_end, - ) - - delegating = AIMessage( - content="", - tool_calls=[{"name": "delegate_to_research_helper", "args": {"task": "hi"}, "id": "c1"}], - ) - sends = _route_manager_to_worker_or_end({"messages": [delegating]}) - # One delegation → a single Send to the worker node carrying the task - # and the tool_call_id its reply must answer. - assert isinstance(sends, list) and len(sends) == 1 - assert isinstance(sends[0], Send) - assert sends[0].node == "research_helper" - assert sends[0].arg == {_DELEGATE_TASK_KEY: "hi", _DELEGATE_CALL_ID_KEY: "c1"} - - 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 _route_manager_to_worker_or_end 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 -def test_route_manager_to_worker_or_end_fans_out_every_delegation() -> None: +def test_route_manager_to_worker_or_end_fans_out_one_send_per_delegation() -> None: from langchain_core.messages import AIMessage from langgraph.types import Send @@ -187,53 +128,31 @@ def test_route_manager_to_worker_or_end_fans_out_every_delegation() -> None: ], ) sends = _route_manager_to_worker_or_end({"messages": [msg]}) - # Every delegation gets its own Send so each tool_call_id is answered. - # The non-delegation tool call was already executed inside the manager's + # 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. assert all(isinstance(s, Send) for s in sends) assert [s.node for s in sends] == ["drafter", "research_helper"] - assert [s.arg[_DELEGATE_CALL_ID_KEY] for s in sends] == ["c1", "c2"] assert [s.arg[_DELEGATE_TASK_KEY] for s in sends] == ["x", "y"] - - -# ─── Topology test (no LLM execution; checks compiled graph shape) ────────── + assert [s.arg[_DELEGATE_CALL_ID_KEY] for s in sends] == ["c1", "c2"] def test_manager_workers_compiles_to_hierarchical_graph_topology() -> None: from langchain_core.messages import AIMessage from langgraph.graph import START - from pyagentspec.adapters.langgraph._managerworkers import ( - _MANAGER_NODE_KEY, - ) - from pyagentspec.agent import Agent - from pyagentspec.managerworkers import ManagerWorkers - - manager_agent = Agent( - name="Coordinator", - description="Coordinates", - system_prompt="Coordinate the team.", - llm_config=_llm_cfg("manager_llm"), - ) - worker_a = Agent( - name="Research Helper", - description="Handles research", - system_prompt="Research.", - llm_config=_llm_cfg("worker_a_llm"), - ) - worker_b = Agent( - name="Drafter", - description="Drafts text", - system_prompt="Draft.", - llm_config=_llm_cfg("worker_b_llm"), - ) + from pyagentspec.adapters.langgraph._managerworkers import _MANAGER_NODE_KEY + mw = ManagerWorkers( name="ResearchTeam", - group_manager=manager_agent, - workers=[worker_a, worker_b], + group_manager=_agent("Coordinator", "manager_llm", system_prompt="Coordinate the team."), + workers=[ + _agent("Research Helper", "worker_a_llm", description="Handles research"), + _agent("Drafter", "worker_b_llm", description="Drafts text"), + ], ) - compiled = _load_with_fake_llms(mw, default=_fake_manager(AIMessage(content="Done."))) + compiled = _load_with_fake_llms(mw, default=_fake_llm(AIMessage(content="Done."))) builder = compiled.builder assert _MANAGER_NODE_KEY in builder.nodes @@ -246,88 +165,43 @@ def test_manager_workers_compiles_to_hierarchical_graph_topology() -> None: assert ("research_helper", _MANAGER_NODE_KEY) in edge_pairs assert ("drafter", _MANAGER_NODE_KEY) in edge_pairs - # Manager → worker is a conditional edge, and branches live separately from - # plain edges on the builder. - branches = builder.branches.get(_MANAGER_NODE_KEY) or {} - assert branches, "expected a conditional branch from the manager node" + # Manager → worker is a conditional edge. + assert builder.branches.get(_MANAGER_NODE_KEY) def test_manager_workers_registers_a_delegation_tool_per_worker() -> None: - """Each worker gets a ``delegate_to_`` tool on the manager, matching the - ``Available workers:`` roster the converter renders into its system prompt (the - roster text itself is covered by the ``_append_workers_roster`` unit tests). - """ from langchain_core.messages import AIMessage - from pyagentspec.adapters.langgraph._managerworkers import ( - _MANAGER_NODE_KEY, - ) - from pyagentspec.agent import Agent - from pyagentspec.managerworkers import ManagerWorkers - - manager_agent = Agent( - name="Coordinator", - description="Coordinates", - system_prompt="Coordinate the team.", - llm_config=_llm_cfg("manager_llm"), - ) - worker = Agent( - name="Research Helper", - description="Handles research tasks", - system_prompt="Research.", - llm_config=_llm_cfg("worker_llm"), - ) + from pyagentspec.adapters.langgraph._managerworkers import _MANAGER_NODE_KEY + mw = ManagerWorkers( name="Team", - group_manager=manager_agent, - workers=[worker], + group_manager=_agent("Coordinator", "manager_llm"), + workers=[_agent("Research Helper", "worker_llm", description="Handles research tasks")], ) - compiled = _load_with_fake_llms(mw, default=_fake_manager(AIMessage(content="Done."))) + compiled = _load_with_fake_llms(mw, default=_fake_llm(AIMessage(content="Done."))) - # The manager react-agent is itself a subgraph; the delegation tool the roster - # advertises is registered on its tools node, so the LLM has the matching contract. + # The delegation tool the roster advertises is registered on the manager + # 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 -# ─── End-to-end execution test (offline, fake LLM emitting delegation) ────── - - def test_manager_workers_delegates_and_routes_back_with_tool_message() -> None: - """End to end, the path that proves subgraph composition works. - - The manager emits a delegate_to_ call, the parent graph routes to the - worker subgraph in an isolated message context, the worker's answer comes back - as a ToolMessage matched to the pending tool_call_id, and the manager's next - turn terminates the graph. - """ + """The manager delegates, the worker runs in an isolated message context and its + answer comes back as a ToolMessage matched to the pending tool_call_id, and the + manager's next turn terminates the graph.""" from langchain_core.messages import AIMessage, HumanMessage - from pyagentspec.agent import Agent - from pyagentspec.managerworkers import ManagerWorkers - - manager_agent = Agent( - name="Coordinator", - description="Coordinates", - system_prompt="You coordinate.", - llm_config=_llm_cfg("manager_llm"), - ) - worker = Agent( - name="Research Helper", - description="Handles research", - system_prompt="You research.", - llm_config=_llm_cfg("worker_llm"), - ) mw = ManagerWorkers( name="Team", - group_manager=manager_agent, - workers=[worker], + group_manager=_agent("Coordinator", "manager_llm", system_prompt="You coordinate."), + workers=[_agent("Research Helper", "worker_llm", description="Handles research")], ) - # Manager turn 1: delegate to research_helper. - # Manager turn 2: produce final answer (no tool call → END). + # Manager turn 1: delegate. Manager turn 2: final answer (no tool call → END). manager_responses = [ AIMessage( content="", @@ -341,68 +215,43 @@ def test_manager_workers_delegates_and_routes_back_with_tool_message() -> None: ), AIMessage(content="The worker reports: Saturn has rings."), ] - # Worker turn 1: produce its own final answer. worker_responses = [AIMessage(content="Saturn has rings.")] compiled = _load_with_fake_llms( mw, - manager_llm=_fake_manager(*manager_responses), - worker_llm=_fake_manager(*worker_responses), + manager_llm=_fake_llm(*manager_responses), + worker_llm=_fake_llm(*worker_responses), ) - # Sync invocation only. FakeMessagesListChatModel overrides ``_generate`` but not - # ``_agenerate``, so the async path would resolve up the MRO to the real - # ``ChatOpenAI._agenerate`` and call OpenAI. + # Sync invocation only: FakeMessagesListChatModel overrides ``_generate`` but not + # ``_agenerate``, so the async path would resolve to ``ChatOpenAI._agenerate`` + # and call OpenAI. result = compiled.invoke( {"messages": [HumanMessage(content="Tell me about Saturn.")]}, {"configurable": {"thread_id": "mw-1"}}, ) messages = result["messages"] - msg_types = [type(m).__name__ for m in messages] - assert "HumanMessage" in msg_types - assert "ToolMessage" in msg_types assert isinstance(messages[-1], AIMessage) assert "Saturn has rings" in messages[-1].content - - # Matching the pending delegation id proves the isolation wrapper threaded the - # call id through. tool_msgs = [m for m in messages if type(m).__name__ == "ToolMessage"] assert tool_msgs and tool_msgs[0].tool_call_id == "call_1" assert "Saturn has rings" in tool_msgs[0].content def test_manager_workers_answers_every_delegation_in_a_single_turn() -> None: - """Regression: when the manager emits SEVERAL ``delegate_to_`` - tool calls in one turn (e.g. "spin up 5 sub-agents"), every delegation - must run and be answered by its own ToolMessage matched to the - originating tool_call_id. - - Before the fix the parent graph routed only the first delegation and left the - other tool_call_ids unanswered. That is an invalid tool-call/tool-result - sequence, and the manager hallucinated the missing replies. - """ + """When one manager turn emits several delegations, each must be answered by its + own ToolMessage matched to the originating tool_call_id; an unanswered one is an + invalid tool-call/result sequence the manager would hallucinate around.""" from langchain_core.messages import AIMessage, HumanMessage - from pyagentspec.agent import Agent - from pyagentspec.managerworkers import ManagerWorkers - - manager_agent = Agent( - name="Coordinator", - description="Coordinates", - system_prompt="You coordinate.", - llm_config=_llm_cfg("manager_llm"), - ) - worker = Agent( - name="Sub Agent", - description="Writes poems", - system_prompt="You write poems.", - llm_config=_llm_cfg("worker_llm"), + mw = ManagerWorkers( + name="Team", + group_manager=_agent("Coordinator", "manager_llm", system_prompt="You coordinate."), + workers=[_agent("Sub Agent", "worker_llm", description="Writes poems")], ) - mw = ManagerWorkers(name="Team", group_manager=manager_agent, workers=[worker]) - # Turn 1: three delegations to the SAME worker in one AIMessage. - # Turn 2: terminate (no tool call). + # Turn 1: three delegations to the same worker in one AIMessage. Turn 2: terminate. manager_responses = [ AIMessage( content="", @@ -414,13 +263,12 @@ def test_manager_workers_answers_every_delegation_in_a_single_turn() -> None: ), AIMessage(content="Here are your three poems."), ] - # Each worker invocation pops one reply; provide enough for the fan-out. worker_responses = [AIMessage(content=f"poem #{i}") for i in range(1, 6)] compiled = _load_with_fake_llms( mw, - manager_llm=_fake_manager(*manager_responses), - worker_llm=_fake_manager(*worker_responses), + manager_llm=_fake_llm(*manager_responses), + worker_llm=_fake_llm(*worker_responses), ) result = compiled.invoke( @@ -429,138 +277,68 @@ def test_manager_workers_answers_every_delegation_in_a_single_turn() -> None: ) messages = result["messages"] - # Every delegation tool_call_id must be answered by exactly one ToolMessage. - requested = { - tc["id"] - for m in messages - if isinstance(m, AIMessage) - for tc in (m.tool_calls or []) - if tc["name"].startswith("delegate_to_") - } - answered = [m.tool_call_id for m in messages if type(m).__name__ == "ToolMessage"] - assert requested == {"call_1", "call_2", "call_3"} - assert sorted(answered) == [ - "call_1", - "call_2", - "call_3", - ], f"unanswered delegations: {requested - set(answered)}" - # No duplicate replies, and each carries a worker poem. - assert len(answered) == 3 tool_msgs = [m for m in messages if type(m).__name__ == "ToolMessage"] + answered = sorted(m.tool_call_id for m in tool_msgs) + assert answered == ["call_1", "call_2", "call_3"] assert all(m.content.startswith("poem #") for m in tool_msgs) -# ─── Recursive nesting ────────────────────────────────────────────────────── - - def test_nested_manager_workers_compiles_recursively() -> None: - """A worker that is itself a ManagerWorkers compiles through the same dispatch, - becoming a CompiledStateGraph the outer parent graph wires in as a subgraph node.""" + """A worker that is itself a ManagerWorkers compiles through the same dispatch and + is wired in as a subgraph node of the outer parent graph.""" from langchain_core.messages import AIMessage - from pyagentspec.agent import Agent - from pyagentspec.managerworkers import ManagerWorkers - - leaf = Agent( - name="Leaf", - description="Leaf task", - system_prompt="Leaf.", - llm_config=_llm_cfg("leaf_llm"), - ) - inner_manager = Agent( - name="InnerManager", - description="Inner", - system_prompt="Manage leaves.", - llm_config=_llm_cfg("inner_llm"), - ) inner_mw = ManagerWorkers( name="Inner", - group_manager=inner_manager, - workers=[leaf], - ) - outer_manager = Agent( - name="OuterManager", - description="Outer", - system_prompt="Manage subteams.", - llm_config=_llm_cfg("outer_llm"), + group_manager=_agent("InnerManager", "inner_llm", system_prompt="Manage leaves."), + workers=[_agent("Leaf", "leaf_llm", description="Leaf task")], ) outer_mw = ManagerWorkers( name="Outer", - group_manager=outer_manager, + group_manager=_agent("OuterManager", "outer_llm", system_prompt="Manage subteams."), workers=[inner_mw], ) - compiled = _load_with_fake_llms(outer_mw, default=_fake_manager(AIMessage(content="Done."))) + compiled = _load_with_fake_llms(outer_mw, default=_fake_llm(AIMessage(content="Done."))) - # Outer parent graph has a node for the inner ManagerWorkers worker. assert "inner" in compiled.builder.nodes def test_rejects_non_agent_group_manager() -> None: - """group_manager must be an Agent. Pyagentspec accepts any AgenticComponent, but - the adapter needs a chat-LLM emitting tool_calls to decide where to delegate.""" + """A nested ManagerWorkers as group_manager is valid per the pyagentspec + validators, but the adapter needs a chat-LLM emitting tool_calls to route on.""" from langgraph.checkpoint.memory import MemorySaver from pyagentspec.adapters.langgraph import AgentSpecLoader - from pyagentspec.agent import Agent - from pyagentspec.managerworkers import ManagerWorkers - - # A nested ManagerWorkers as group_manager: valid per the pyagentspec - # validators, unsupported here. - leaf = Agent( - name="Leaf", - description="L", - system_prompt="L.", - llm_config=_llm_cfg("l"), - ) - inner_manager = Agent( - name="Inner", - description="I", - system_prompt="I.", - llm_config=_llm_cfg("i"), - ) + inner_mw = ManagerWorkers( name="Inner", - group_manager=inner_manager, - workers=[leaf], + group_manager=_agent("Inner", "i"), + workers=[_agent("Leaf", "l")], ) outer_mw = ManagerWorkers( name="Outer", group_manager=inner_mw, - workers=[ - Agent(name="Other", description="O", system_prompt="O.", llm_config=_llm_cfg("o")), - ], + workers=[_agent("Other", "o")], ) + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) with pytest.raises(NotImplementedError, match="group_manager must be an Agent"): loader.load_component(outer_mw) -# ─── Worker name collision ────────────────────────────────────────────────── - - def test_workers_with_name_slug_collision_are_rejected() -> None: - """Two workers whose names normalize to the same node identifier - would silently overwrite each other in the parent graph; raise at - load time instead.""" + """Two workers whose names normalize to the same node identifier would silently + overwrite each other in the parent graph; raise at load time instead.""" from langgraph.checkpoint.memory import MemorySaver from pyagentspec.adapters.langgraph import AgentSpecLoader - from pyagentspec.agent import Agent - from pyagentspec.managerworkers import ManagerWorkers - a = Agent(name="Helper A", description="x", system_prompt=".", llm_config=_llm_cfg("a")) - b = Agent(name="helper-a", description="x", system_prompt=".", llm_config=_llm_cfg("b")) - # Both normalize to "helper_a". + # Both worker names normalize to "helper_a". mw = ManagerWorkers( name="T", - group_manager=Agent( - name="M", - description="m", - system_prompt=".", - llm_config=_llm_cfg("m"), - ), - workers=[a, b], + group_manager=_agent("M", "m"), + workers=[_agent("Helper A", "a"), _agent("helper-a", "b")], ) loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) @@ -569,24 +347,19 @@ def test_workers_with_name_slug_collision_are_rejected() -> None: def test_worker_events_stream_natively_namespaced_under_worker_node() -> None: - """Regression: a worker's token events must stream under the worker - node's checkpoint namespace so a consumer can attribute them to the - sub-agent. The wrapper must inherit the ambient run config (no fresh - thread_id); a fresh thread_id detaches the worker into a top-level - ``agent:`` run with no worker prefix, which is unattributable.""" + """Regression: a worker's token events must stream under the worker node's + checkpoint namespace so a consumer can attribute them to the sub-agent. The + wrapper must inherit the ambient run config; a fresh thread_id would detach the + worker into an unattributable top-level ``agent:`` run.""" import asyncio - from langchain_core.language_models.fake_chat_models import ( - GenericFakeChatModel, - ) + from langchain_core.language_models.fake_chat_models import GenericFakeChatModel from langchain_core.messages import AIMessage, HumanMessage from langgraph.graph import END, START, MessagesState, StateGraph - from pyagentspec.adapters.langgraph._managerworkers import ( - _wrap_worker_for_subgraph, - ) + from pyagentspec.adapters.langgraph._managerworkers import _wrap_worker_for_subgraph - # A minimal worker compiled graph that streams some content. + # A minimal worker graph that streams some content. wmodel = GenericFakeChatModel(messages=iter([AIMessage(content="Saturn has rings")] * 9)) wb = StateGraph(MessagesState) @@ -598,8 +371,8 @@ async def _wagent(state: Any) -> Any: wb.add_edge("agent", END) worker_graph = wb.compile() - # Parent: a plain manager node emits the delegate tool call, then routes - # to the wrapped worker node named "research_helper". + # Parent: a plain manager node emits the delegate tool call, then routes to the + # wrapped worker node. pb = StateGraph(MessagesState) def _manager(state: Any) -> Any: @@ -639,115 +412,9 @@ async def _collect() -> Any: namespaces = asyncio.run(_collect()) assert namespaces, "expected the worker to emit token-stream events" - # Every worker token event is namespaced under the worker node, so a - # consumer can attribute the stream to the sub-agent. assert all(ns.startswith("research_helper:") for ns in namespaces), namespaces -# ─── Shared low-level helper unit tests (no LLM) ───────────────────────────── - - -def test_normalize_identifier_lowercases_collapses_and_strips() -> None: - """The single normalization used for both worker node names and - ``transfer_to_`` tool names.""" - from pyagentspec.adapters.langgraph._managerworkers import _normalize_identifier - - assert _normalize_identifier("Research Helper") == "research_helper" - assert _normalize_identifier("My-Worker!! v2") == "my_worker_v2" - # Punctuation-only / empty slugify to the empty string (callers add a fallback). - assert _normalize_identifier("!!!") == "" - assert _normalize_identifier("") == "" - - -def test_messages_of_reads_dict_and_object_state() -> None: - """The delegation tool receives state as a dict or an attribute-bearing - object depending on the langgraph injection path.""" - from langchain_core.messages import AIMessage - - from pyagentspec.adapters.langgraph._managerworkers import _messages_of - - msg = AIMessage(content="hi") - assert _messages_of({"messages": [msg]}) == [msg] - assert _messages_of({"messages": None}) == [] - assert _messages_of({}) == [] - - class _State: - messages = [msg] - - assert _messages_of(_State()) == [msg] - - class _Empty: - pass - - assert _messages_of(_Empty()) == [] - - -def test_surface_to_parent_command_projects_messages_with_no_goto() -> None: - """The placeholder tool's body: break to the parent graph, project the - subgraph messages, carry no ``goto`` (routing is the parent's job).""" - from langchain_core.messages import AIMessage - from langgraph.types import Command - - from pyagentspec.adapters.langgraph._managerworkers import _surface_to_parent_command - - m1, m2 = AIMessage(content="a"), AIMessage(content="b") - cmd = _surface_to_parent_command({"messages": [m1, m2]}) - - assert isinstance(cmd, Command) - assert cmd.graph == Command.PARENT - assert cmd.goto == () # no goto; the parent graph decides where to go - assert cmd.update == {"messages": [m1, m2]} - - -def test_delegation_tool_exposes_expected_name_and_description() -> None: - """The placeholder tool the manager's LLM addresses by name.""" - from pyagentspec.adapters.langgraph._managerworkers import _make_worker_delegation_tool - - delegate = _make_worker_delegation_tool("research_helper") - assert delegate.name == "delegate_to_research_helper" - assert "research_helper" in delegate.description - - -# ─── _wrap_worker_for_subgraph: pending-delegation extraction (no LLM) ──────── - - -def _echo_worker_graph(reply: str = "WORKER REPLY") -> Any: - """A worker CompiledStateGraph whose only node returns a fixed AIMessage, enough - to exercise the wrapper without an LLM.""" - from langchain_core.messages import AIMessage - from langgraph.graph import END, START, MessagesState, StateGraph - - wb = StateGraph(MessagesState) - wb.add_node("agent", lambda state: {"messages": [AIMessage(content=reply)]}) - wb.add_edge(START, "agent") - wb.add_edge("agent", END) - return wb.compile() - - -def test_wrap_worker_uses_send_payload_task_and_call_id() -> None: - """Fan-out path: the routing edge's ``Send`` payload carries the task and - the originating tool_call_id directly, so the worker reply ToolMessage is - matched to that call.""" - from langchain_core.messages import ToolMessage - - from pyagentspec.adapters.langgraph._managerworkers import ( - _DELEGATE_CALL_ID_KEY, - _DELEGATE_TASK_KEY, - _wrap_worker_for_subgraph, - ) - - node = _wrap_worker_for_subgraph(_echo_worker_graph("DONE"), "research_helper") - out = node.invoke({_DELEGATE_TASK_KEY: "do it", _DELEGATE_CALL_ID_KEY: "call_9"}) - - (reply,) = out["messages"] - assert isinstance(reply, ToolMessage) - assert reply.content == "DONE" - assert reply.tool_call_id == "call_9" - - -# ─── Delegation visibility: the public consumer-side filter ────────────────── - - def test_is_delegation_tool_name_matches_only_the_synthetic_prefix() -> None: from pyagentspec.adapters.langgraph._managerworkers import ( DELEGATE_TOOL_PREFIX, @@ -764,34 +431,16 @@ def test_is_delegation_tool_name_matches_only_the_synthetic_prefix() -> None: def test_manager_workers_leaves_astream_events_unwrapped() -> None: - """The delegation protocol is deliberately visible: nothing wraps - ``astream_events`` to scrub it. Only stream/astream are patched, for the - ManagerWorkersExecutionSpan.""" - - from pyagentspec.agent import Agent - from pyagentspec.managerworkers import ManagerWorkers - + """The delegation protocol is deliberately visible: only stream/astream are + patched (for the ManagerWorkersExecutionSpan); nothing wraps ``astream_events`` + to scrub the delegation tool calls.""" mw = ManagerWorkers( name="Team", - group_manager=Agent( - name="Coordinator", - description="c", - system_prompt=".", - llm_config=_llm_cfg("manager_llm"), - ), - workers=[ - Agent( - name="Research Helper", - description="r", - system_prompt=".", - llm_config=_llm_cfg("worker_llm"), - ), - ], + group_manager=_agent("Coordinator", "manager_llm"), + workers=[_agent("Research Helper", "worker_llm")], ) - compiled = _load_with_fake_llms(mw, default=_fake_manager()) + compiled = _load_with_fake_llms(mw, default=_fake_llm()) - assert getattr(compiled.astream_events, "__name__", "") != "patched_astream_events" - # The execution-span patches are still applied. - assert getattr(compiled.stream, "__name__", "") == "patched_stream" - assert getattr(compiled.astream, "__name__", "") == "patched_astream" + assert "stream" in compiled.__dict__ and "astream" in compiled.__dict__ + assert "astream_events" not in compiled.__dict__