From 3eaf65d7c4441fee7ba9d1348bddbbb57413292e Mon Sep 17 00:00:00 2001 From: tikhandesanket Date: Thu, 8 Oct 2026 18:53:32 +0530 Subject: [PATCH 1/2] feat(agents): add multi-agent supervisor primitives --- src/dynavec/agents/__init__.py | 7 ++ src/dynavec/agents/handoff.py | 13 +++ src/dynavec/agents/react.py | 21 ++-- src/dynavec/agents/state.py | 13 +++ src/dynavec/agents/supervisor.py | 32 ++++++ tests/test_agent_state.py | 20 ++++ tests/test_agents.py | 24 ++++- tests/test_handoff.py | 18 ++++ tests/test_supervisor.py | 164 +++++++++++++++++++++++++++++++ 9 files changed, 295 insertions(+), 17 deletions(-) create mode 100644 src/dynavec/agents/handoff.py create mode 100644 src/dynavec/agents/state.py create mode 100644 src/dynavec/agents/supervisor.py create mode 100644 tests/test_agent_state.py create mode 100644 tests/test_handoff.py create mode 100644 tests/test_supervisor.py diff --git a/src/dynavec/agents/__init__.py b/src/dynavec/agents/__init__.py index 262cfa9..be21a40 100644 --- a/src/dynavec/agents/__init__.py +++ b/src/dynavec/agents/__init__.py @@ -22,12 +22,19 @@ mcp_tools_from_session, mcp_tools_from_session_async, ) +from .handoff import Handoff from .planner import Planner from .react import ReActAgent from .registry import ToolRegistry, default_registry +from .state import AgentState +from .supervisor import AgentLike, Supervisor __all__ = [ + "AgentState", + "Handoff", + "Supervisor", "AgentResult", + "AgentLike", "AgentStep", "AgentTool", "Plan", diff --git a/src/dynavec/agents/handoff.py b/src/dynavec/agents/handoff.py new file mode 100644 index 0000000..09e3d66 --- /dev/null +++ b/src/dynavec/agents/handoff.py @@ -0,0 +1,13 @@ +"""Handoff primitives for multi-agent workflows.""" + +from __future__ import annotations + +from .state import AgentState + + +class Handoff: + """Pass shared agent state from one workflow step to another.""" + + def __call__(self, state: AgentState) -> AgentState: + """Return the same state so downstream agents share workflow context.""" + return state diff --git a/src/dynavec/agents/react.py b/src/dynavec/agents/react.py index 3bd9546..5904a16 100644 --- a/src/dynavec/agents/react.py +++ b/src/dynavec/agents/react.py @@ -8,6 +8,7 @@ from ..chat.base import ChatModel, Message, Tool, ToolCall from .base import AgentResult, AgentStep, AgentTool from .registry import ToolRegistry +from .state import AgentState if TYPE_CHECKING: from ..checkpoint import BaseCheckpointer @@ -93,9 +94,7 @@ def __init__( self._chat_tools.append(t.to_chat_tool()) elif tools: for raw_tool in tools: - agent_tool = ( - raw_tool if isinstance(raw_tool, AgentTool) else AgentTool(raw_tool) - ) + agent_tool = raw_tool if isinstance(raw_tool, AgentTool) else AgentTool(raw_tool) self._tools_map[agent_tool.name] = agent_tool self._chat_tools.append(agent_tool.to_chat_tool()) @@ -118,9 +117,7 @@ async def _adispatch_tool_call(self, tc: ToolCall) -> str: return await tool_instance.aexecute(tc.arguments) return f"Error: Tool {tc.name!r} is not registered in available tools." - def _process_tool_calls( - self, tool_calls: list[ToolCall], messages: list[Message] - ) -> list[str]: + def _process_tool_calls(self, tool_calls: list[ToolCall], messages: list[Message]) -> list[str]: step_observations: list[str] = [] for tc in tool_calls: obs = self._dispatch_tool_call(tc) @@ -150,12 +147,9 @@ async def _aprocess_tool_calls( ) return step_observations - def _finalize_result( - self, steps: list[AgentStep], total_tool_calls: int - ) -> AgentResult: + def _finalize_result(self, steps: list[AgentStep], total_tool_calls: int) -> AgentResult: last_output = ( - steps[-1].thought - or (steps[-1].observations[-1] if steps[-1].observations else "") + steps[-1].thought or (steps[-1].observations[-1] if steps[-1].observations else "") if steps else "" ) @@ -174,6 +168,7 @@ def run( *, thread_id: str | None = None, resume_from: str | None = None, + state: AgentState | None = None, **kwargs: Any, ) -> AgentResult: """Execute the ReAct loop synchronously until goal completion or max_steps.""" @@ -371,9 +366,7 @@ async def arun( ) total_tool_calls += len(msg.tool_calls) - step_observations = await self._aprocess_tool_calls( - list(msg.tool_calls), messages - ) + step_observations = await self._aprocess_tool_calls(list(msg.tool_calls), messages) step = AgentStep( step_number=step_idx, diff --git a/src/dynavec/agents/state.py b/src/dynavec/agents/state.py new file mode 100644 index 0000000..54ae44d --- /dev/null +++ b/src/dynavec/agents/state.py @@ -0,0 +1,13 @@ +"""Shared state primitives for multi-agent workflows.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class AgentState: + """Shared state passed between agents in a workflow.""" + + data: dict[str, Any] = field(default_factory=dict) diff --git a/src/dynavec/agents/supervisor.py b/src/dynavec/agents/supervisor.py new file mode 100644 index 0000000..99ba533 --- /dev/null +++ b/src/dynavec/agents/supervisor.py @@ -0,0 +1,32 @@ +"""Supervisor primitives for multi-agent workflows.""" + +from __future__ import annotations + +from typing import Protocol + +from .base import AgentResult +from .state import AgentState + + +class AgentLike(Protocol): + """Interface required by a supervisor-managed agent.""" + + def run( + self, + goal: str, + *, + state: AgentState, + ) -> AgentResult: + """Execute an agent using shared workflow state.""" + ... + + +class Supervisor: + """Delegate a goal to multiple agents using shared workflow state.""" + + def __init__(self, agents: list[AgentLike]) -> None: + self.agents = agents + + def run(self, goal: str, *, state: AgentState) -> list[AgentResult]: + """Run each specialist with the same shared state.""" + return [agent.run(goal, state=state) for agent in self.agents] diff --git a/tests/test_agent_state.py b/tests/test_agent_state.py new file mode 100644 index 0000000..78db07a --- /dev/null +++ b/tests/test_agent_state.py @@ -0,0 +1,20 @@ +from dynavec.agents import AgentState + + +def test_agent_state_stores_shared_data(): + state = AgentState() + + state.data["customer_id"] = "C123" + state.data["decline_code"] = "51" + + assert state.data["customer_id"] == "C123" + assert state.data["decline_code"] == "51" + + +def test_agent_state_has_independent_data(): + first = AgentState() + second = AgentState() + + first.data["customer_id"] = "C123" + + assert second.data == {} diff --git a/tests/test_agents.py b/tests/test_agents.py index 4ee91cb..328ed0f 100644 --- a/tests/test_agents.py +++ b/tests/test_agents.py @@ -9,6 +9,7 @@ import pytest from dynavec.agents import ( + AgentState, AgentTool, Plan, Planner, @@ -43,9 +44,7 @@ def invoke( if self.call_count >= len(self.responses): # Default response if script exhausted return ChatResult( - message=Message( - role="assistant", content="Scripted responses exhausted." - ), + message=Message(role="assistant", content="Scripted responses exhausted."), finish_reason="stop", ) @@ -343,3 +342,22 @@ async def test_planner_async_aplan() -> None: assert len(plan.steps) == 2 assert plan.steps[0].description == "Step A" assert plan.steps[1].description == "Step B" + + +def test_react_agent_accepts_shared_state(): + state = AgentState() + + agent = ReActAgent( + model=ScriptedChatModel( + [ + "Transaction analysis completed.", + ] + ) + ) + + result = agent.run( + "Analyze transaction TX123.", + state=state, + ) + + assert result.output == "Transaction analysis completed." diff --git a/tests/test_handoff.py b/tests/test_handoff.py new file mode 100644 index 0000000..c882cab --- /dev/null +++ b/tests/test_handoff.py @@ -0,0 +1,18 @@ +from dynavec.agents import AgentState, Handoff + + +def test_handoff_preserves_shared_state(): + state = AgentState() + + state.data["transaction_id"] = "TX123" + state.data["decline_code"] = "51" + + handoff = Handoff() + + next_state = handoff(state) + + assert next_state is state + assert next_state.data == { + "transaction_id": "TX123", + "decline_code": "51", + } diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py new file mode 100644 index 0000000..055da05 --- /dev/null +++ b/tests/test_supervisor.py @@ -0,0 +1,164 @@ +from typing import Any + +from dynavec.agents import ( + AgentResult, + AgentState, + ReActAgent, + Supervisor, +) +from dynavec.chat.base import ChatChunk, ChatModel, ChatResult, Message, Tool + + +class ScriptedChatModel(ChatModel): + """Deterministic chat model for supervisor tests.""" + + def __init__(self, responses: list[str]) -> None: + self.responses = responses + self.call_count = 0 + + def invoke( + self, + messages: list[Message], + tools: list[Tool] | None = None, + **kwargs: Any, + ) -> ChatResult: + response = self.responses[self.call_count] + self.call_count += 1 + + return ChatResult( + message=Message( + role="assistant", + content=response, + ), + finish_reason="stop", + ) + + def stream( + self, + messages: list[Message], + tools: list[Tool] | None = None, + **kwargs: Any, + ): + result = self.invoke(messages, tools, **kwargs) + + yield ChatChunk( + content=result.message.content, + tool_calls=result.message.tool_calls, + ) + + +class TransactionAgent: + def run(self, goal: str, *, state: AgentState) -> AgentResult: + state.data["transaction_id"] = "TX123" + state.data["decline_code"] = "51" + + return AgentResult( + output="Transaction declined with code 51.", + finished=True, + termination_reason="completed", + total_steps=0, + tool_calls_count=0, + ) + + +class PolicyAgent: + def run(self, goal: str, *, state: AgentState) -> AgentResult: + decline_code = state.data["decline_code"] + + return AgentResult( + output=f"Policy analysis for decline code {decline_code}.", + finished=True, + termination_reason="completed", + total_steps=0, + tool_calls_count=0, + ) + + +def test_supervisor_delegates_to_multiple_agents_with_shared_state(): + state = AgentState() + + supervisor = Supervisor( + agents=[ + TransactionAgent(), + PolicyAgent(), + ], + ) + + results = supervisor.run( + "Analyze the declined transaction.", + state=state, + ) + + assert len(results) == 2 + + assert results[0].output == "Transaction declined with code 51." + assert results[1].output == "Policy analysis for decline code 51." + + assert state.data["transaction_id"] == "TX123" + assert state.data["decline_code"] == "51" + + +def test_supervisor_runs_react_agents(): + state = AgentState() + + first_agent = ReActAgent(model=ScriptedChatModel(["Transaction analysis completed."])) + + second_agent = ReActAgent(model=ScriptedChatModel(["Policy analysis completed."])) + + supervisor = Supervisor( + agents=[first_agent, second_agent], + ) + + results = supervisor.run( + "Analyze transaction TX123.", + state=state, + ) + + assert [result.output for result in results] == [ + "Transaction analysis completed.", + "Policy analysis completed.", + ] + + +def test_supervisor_preserves_state_between_specialists(): + state = AgentState() + + class FirstAgent: + def run(self, goal: str, *, state: AgentState) -> AgentResult: + state.data["transaction_id"] = "TX123" + state.data["decline_code"] = "51" + + return AgentResult( + output="Transaction analyzed.", + finished=True, + termination_reason="completed", + total_steps=0, + tool_calls_count=0, + ) + + class SecondAgent: + def run(self, goal: str, *, state: AgentState) -> AgentResult: + assert state.data["transaction_id"] == "TX123" + assert state.data["decline_code"] == "51" + + return AgentResult( + output="Policy analyzed.", + finished=True, + termination_reason="completed", + total_steps=0, + tool_calls_count=0, + ) + + supervisor = Supervisor( + agents=[FirstAgent(), SecondAgent()], + ) + + results = supervisor.run( + "Analyze the declined transaction.", + state=state, + ) + + assert [result.output for result in results] == [ + "Transaction analyzed.", + "Policy analyzed.", + ] From 2b423aaed3c8a4c7546b2a86a6e7d9a5fd1716df Mon Sep 17 00:00:00 2001 From: tikhandesanket Date: Thu, 8 Oct 2026 19:25:48 +0530 Subject: [PATCH 2/2] feat: add multi-agent supervisor pattern --- src/dynavec/agents/react.py | 20 ++++++++++++++------ tests/test_supervisor.py | 2 ++ 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/src/dynavec/agents/react.py b/src/dynavec/agents/react.py index 5904a16..0ad3d8d 100644 --- a/src/dynavec/agents/react.py +++ b/src/dynavec/agents/react.py @@ -179,13 +179,21 @@ def run( if self.checkpointer is not None and thread_id is not None: existing_cp = self.checkpointer.get(thread_id, resume_from) + if existing_cp is not None: - state = existing_cp.state - if "messages" in state: - messages = [_dict_to_message(m) for m in state["messages"]] - if "steps" in state: - steps = [_dict_to_step(s) for s in state["steps"]] - total_tool_calls = int(state.get("total_tool_calls", 0)) + checkpoint_state = existing_cp.state + + if "messages" in checkpoint_state: + messages = [ + _dict_to_message(m) for m in checkpoint_state["messages"] + ] + + if "steps" in checkpoint_state: + steps = [ + _dict_to_step(s) for s in checkpoint_state["steps"] + ] + + total_tool_calls = int(checkpoint_state.get("total_tool_calls", 0)) start_step = existing_cp.step + 1 if not messages: diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py index 055da05..82f3895 100644 --- a/tests/test_supervisor.py +++ b/tests/test_supervisor.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import Any from dynavec.agents import (