[Bug] Two routers share one Conversation across threads, so every caller gets every task's messages (#2054)

swarms/structs/hybrid_hiearchical_peer_swarm.py

@@ -97,7 +97,6 @@ class HybridHierarchicalClusterSwarm:
         swarms (List[SwarmRouter]): A list of available swarm routers.
         max_loops (int): The maximum number of loops for task processing.
         output_type (str): The format of the output (e.g., list).
-        conversation (Conversation): An instance of the Conversation class to manage interactions.
         router_agent (Agent): An instance of the Agent class responsible for routing tasks.
     """
 
@@ -118,8 +117,6 @@ def __init__(
         self.max_loops = max_loops
         self.output_type = output_type
 
-        self.conversation = Conversation()
-
         self.router_agent = Agent(
             agent_name="Router Agent",
             agent_description="A router agent that routes tasks to the appropriate swarms.",
@@ -146,7 +143,8 @@ def run(self, task: str, *args, **kwargs):
         if not task:
             raise ValueError("Task cannot be empty.")
 
-        self.conversation.add(role="User", content=task)
+        conversation = Conversation()
+        conversation.add(role="User", content=task)
 
         response = self.router_agent.run(task=task)
 
@@ -167,10 +165,10 @@ def run(self, task: str, *args, **kwargs):
                 f"Please check the response format from the model: {self.router_agent.model_name}."
             )
 
-        self.route_task(swarm_name, task_description)
+        self.route_task(swarm_name, task_description, conversation)
 
         return history_output_formatter(
-            self.conversation, self.output_type
+            conversation, self.output_type
         )
 
     def find_swarm_by_name(self, swarm_name: str):
@@ -188,13 +186,19 @@ def find_swarm_by_name(self, swarm_name: str):
                 return swarm
         return None
 
-    def route_task(self, swarm_name: str, task_description: str):
+    def route_task(
+        self,
+        swarm_name: str,
+        task_description: str,
+        conversation: Conversation,
+    ):
         """
         Routes the task to the specified swarm.
 
         Args:
             swarm_name (str): The name of the swarm to route the task to.
             task_description (str): The description of the task to be executed.
+            conversation (Conversation): The conversation to record the output in.
 
         Raises:
             ValueError: If the swarm is not found.
@@ -203,7 +207,7 @@ def route_task(self, swarm_name: str, task_description: str):
 
         if swarm:
             output = swarm.run(task_description)
-            self.conversation.add(role=swarm.name, content=output)
+            conversation.add(role=swarm.name, content=output)
         else:
             raise ValueError(f"Swarm '{swarm_name}' not found.")
 

swarms/structs/multi_agent_router.py

@@ -165,7 +165,6 @@ def __init__(
         self.system_prompt = system_prompt
         self.skip_null_tasks = skip_null_tasks
         self.agents = list(agents) if agents else []
-        self.conversation = Conversation()
 
         router_system_prompt = ""
 
@@ -232,15 +231,18 @@ def _create_boss_system_prompt(self) -> str:
         return agent_boss_router_prompt(agent_descriptions)
 
     def handle_single_handoff(
-        self, boss_response_str: dict, task: str
+        self,
+        boss_response_str: dict,
+        task: str,
+        conversation: Conversation,
     ) -> None:
         """
         Execute the single agent selected by the boss and record its response.
 
         Looks up the agent named in the first (and only) handoff, runs it on
         the modified task (falling back to the original task if the boss did
         not rewrite it), and appends the agent's response to
-        ``self.conversation``. When ``skip_null_tasks`` is True and the
+        ``conversation``. When ``skip_null_tasks`` is True and the
         resolved task is empty or ``None``, execution is skipped.
 
         Args:
@@ -284,12 +286,15 @@ def handle_single_handoff(
         # Use the agent's run method directly
         agent_response = selected_agent.run(final_task)
 
-        self.conversation.add(
+        conversation.add(
             role=selected_agent.agent_name, content=agent_response
         )
 
     def handle_multiple_handoffs(
-        self, boss_response_str: dict, task: str
+        self,
+        boss_response_str: dict,
+        task: str,
+        conversation: Conversation,
     ) -> None:
         """
         Execute every agent selected by the boss and record the first response.
@@ -299,7 +304,7 @@ def handle_multiple_handoffs(
         task (or the original ``task`` when the boss did not rewrite it). When
         ``skip_null_tasks`` is True, agents whose resolved task is empty or
         ``None`` are skipped. After execution, the first selected agent's
-        response is appended to ``self.conversation``.
+        response is appended to ``conversation``.
 
         Args:
             boss_response_str (dict): Parsed boss decision containing a
@@ -366,7 +371,7 @@ def handle_multiple_handoffs(
                     )
                 )
 
-            self.conversation.add(
+            conversation.add(
                 role=selected_agents[0].agent_name,
                 content=agent_responses[0],
             )
@@ -384,7 +389,8 @@ def route_task(self, task: str) -> dict:
             dict: A dictionary containing the routing result, including the selected agent, reasoning, and response.
         """
         try:
-            self.conversation.add(role="user", content=task)
+            conversation = Conversation()
+            conversation.add(role="user", content=task)
 
             # Get boss decision using function calling
             boss_response_str = self.function_caller.run(task)
@@ -398,12 +404,16 @@ def route_task(self, task: str) -> dict:
                 )
 
             if len(boss_response_str["handoffs"]) > 1:
-                self.handle_multiple_handoffs(boss_response_str, task)
+                self.handle_multiple_handoffs(
+                    boss_response_str, task, conversation
+                )
             else:
-                self.handle_single_handoff(boss_response_str, task)
+                self.handle_single_handoff(
+                    boss_response_str, task, conversation
+                )
 
             return history_output_formatter(
-                conversation=self.conversation, type=self.output_type
+                conversation=conversation, type=self.output_type
             )
 
         except Exception as e:
@@ -474,11 +484,6 @@ def concurrent_batch_run(self, tasks: List[str] = []):
         Tasks are dispatched to a ``ThreadPoolExecutor``. Failures are logged
         with the task that caused them and omitted from the result list.
 
-        Note:
-            All tasks share this router's ``Conversation``, so their histories
-            interleave. Prefer :meth:`batch_run` when the per-task history
-            matters.
-
         Args:
             tasks (List[str]): Tasks to route concurrently.
 

tests/structs/test_multi_agent_router.py

@@ -1,3 +1,4 @@
+import json
 import os
 
 import time
@@ -6,6 +7,7 @@
 
 from swarms.structs.agent import Agent
 from swarms.structs.multi_agent_router import MultiAgentRouter
+from swarms.structs.conversation import Conversation
 
 
 def _minimal_agents():
@@ -462,7 +464,9 @@ def test_skip_null_tasks_actually_skips_on_the_single_path():
     router, ran = _local_router()
 
     router.handle_single_handoff(
-        {"handoffs": [{"agent_name": "A1", "task": None}]}, ""
+        {"handoffs": [{"agent_name": "A1", "task": None}]},
+        "",
+        Conversation(),
     )
 
     assert ran == [], f"the agent ran despite a null task: {ran}"
@@ -473,7 +477,9 @@ def test_a_handoff_without_a_task_key_falls_back_to_the_original():
     router, ran = _local_router()
 
     router.handle_single_handoff(
-        {"handoffs": [{"agent_name": "A1"}]}, "the original task"
+        {"handoffs": [{"agent_name": "A1"}]},
+        "the original task",
+        Conversation(),
     )
 
     assert ran == [("A1", "the original task")]
@@ -485,6 +491,7 @@ def test_multiple_handoffs_without_a_task_key_fall_back():
     router.handle_multiple_handoffs(
         {"handoffs": [{"agent_name": "A1"}, {"agent_name": "A2"}]},
         "the original task",
+        Conversation(),
     )
 
     assert [task for _, task in ran] == [
@@ -503,7 +510,7 @@ def test_selected_agents_run_concurrently():
     }
 
     started = time.time()
-    router.handle_multiple_handoffs(handoffs, "x")
+    router.handle_multiple_handoffs(handoffs, "x", Conversation())
     elapsed = time.time() - started
 
     assert len(ran) == 3
@@ -524,6 +531,7 @@ def test_an_unknown_agent_raises_before_any_agent_runs():
                 ]
             },
             "x",
+            Conversation(),
         )
 
     assert ran == [], f"an agent ran before validation failed: {ran}"
@@ -548,3 +556,28 @@ def flaky(task):
 
 if __name__ == "__main__":
     pytest.main([__file__])
+
+
+def test_concurrent_batch_run_keeps_each_task_history_separate():
+    router, ran = _local_router(agents=["A1"], delay=0.05)
+
+    class Boss:
+        def run(self, task):
+            return json.dumps(
+                {"handoffs": [{"agent_name": "A1", "task": task}]}
+            )
+
+    router.function_caller = Boss()
+    router.print_on = False
+    router.output_type = "str"
+
+    tasks = ["alpha", "beta", "gamma"]
+    results = router.concurrent_batch_run(tasks)
+
+    assert len(results) == 3
+    for task, result in zip(tasks, results):
+        assert task in result
+        for other in set(tasks) - {task}:
+            assert (
+                other not in result
+            ), f"{task!r} history leaked {other!r}: {result!r}"