diff --git a/evoagentx/evaluators/evaluator.py b/evoagentx/evaluators/evaluator.py index a6790196..913b9f44 100644 --- a/evoagentx/evaluators/evaluator.py +++ b/evoagentx/evaluators/evaluator.py @@ -14,7 +14,20 @@ from ..workflow.workflow import WorkFlow from ..workflow.action_graph import ActionGraph from ..workflow.workflow_graph import WorkFlowGraph +from ..agents.agent import Agent from ..agents.agent_manager import AgentManager +from ..memory.memory import ShortTermMemory + + +def _agents_with_fresh_short_term_memory(agents: List[Agent]) -> List[Agent]: + """Clone each agent with its own ShortTermMemory, sharing everything else. + + ``AgentManager`` copies (``Evaluator._create_new_agent_manager``, the async + per-task manager below) keep the same ``Agent`` instances across concurrent + examples, so ``Agent.short_term_memory``, mutated on every execution, + is shared with no lock and one example's messages leak into another's. + """ + return [agent.model_copy(update={"short_term_memory": ShortTermMemory()}) for agent in agents] class Evaluator: @@ -241,8 +254,12 @@ def _create_new_agent_manager(self) -> AgentManager: """Create a new agent manager with the same configuration but new locks""" if self.agent_manager is None: return None - # Create a new agent manager - new_manager = AgentManager(agents=self.agent_manager.agents, storage_handler=self.agent_manager.storage_handler) + # Create a new agent manager, with its own agent copies so concurrent + # examples don't share short_term_memory (see _agents_with_fresh_short_term_memory) + new_manager = AgentManager( + agents=_agents_with_fresh_short_term_memory(self.agent_manager.agents), + storage_handler=self.agent_manager.storage_handler + ) return new_manager def _get_thread_agent_manager(self) -> AgentManager: @@ -502,9 +519,10 @@ async def _async_execute_workflow_graph(self, graph: WorkFlowGraph, inputs: dict graph_copy = WorkFlowGraph(goal=graph.goal, graph=graph) graph_copy.reset_graph() # reset the status of all nodes to pending - # Make a local copy of agent_manager for thread-safety in async context + # Make a local copy of agent_manager for thread-safety in async context, + # with fresh per-agent short_term_memory (_agents_with_fresh_short_term_memory) local_agent_manager = AgentManager( - agents=self.agent_manager.agents, + agents=_agents_with_fresh_short_term_memory(self.agent_manager.agents), storage_handler=self.agent_manager.storage_handler ) diff --git a/tests/src/evaluator/test_evaluator.py b/tests/src/evaluator/test_evaluator.py index 2c62eb73..e200d98c 100644 --- a/tests/src/evaluator/test_evaluator.py +++ b/tests/src/evaluator/test_evaluator.py @@ -1,10 +1,13 @@ +import threading import unittest from unittest.mock import Mock, patch -from evoagentx.evaluators.evaluator import Evaluator +from evoagentx.evaluators.evaluator import Evaluator, _agents_with_fresh_short_term_memory from evoagentx.benchmark.benchmark import Benchmark from evoagentx.workflow.workflow_graph import WorkFlowGraph from evoagentx.workflow.action_graph import ActionGraph +from evoagentx.agents.agent import Agent from evoagentx.agents.agent_manager import AgentManager +from evoagentx.core.message import Message, MessageType from evoagentx.models.base_model import BaseLLM class TestEvaluator(unittest.TestCase): @@ -188,5 +191,57 @@ def test_empty_data_evaluation(self): self.assertEqual(results, {}) self.assertEqual(len(self.evaluator.get_all_evaluation_records()), 0) +class TestEvaluatorConcurrentAgentMemoryIsolation(unittest.TestCase): + + def test_agents_with_fresh_short_term_memory_clones_memory_only(self): + # Unit-tests the helper shared by _create_new_agent_manager (thread + # path) and _async_execute_workflow_graph (async path): each clone + # must get its own ShortTermMemory while everything else is shared. + agent = Agent(name="worker", description="test agent", is_human=True) + + clone_one, clone_two = _agents_with_fresh_short_term_memory([agent, agent]) + + self.assertIsNot(clone_one.short_term_memory, agent.short_term_memory) + self.assertIsNot(clone_two.short_term_memory, clone_one.short_term_memory) + self.assertEqual(clone_one.name, agent.name) + self.assertIs(clone_one.llm, agent.llm) + + def test_create_new_agent_manager_isolates_short_term_memory_across_threads(self): + # Reproduces the reported bug: two "independent" per-thread + # AgentManagers created by _create_new_agent_manager must not let one + # example's messages leak into another's short_term_memory. + shared_agent = Agent(name="worker", description="test agent", is_human=True) + agent_manager = AgentManager(agents=[shared_agent]) + evaluator = Evaluator(llm=Mock(spec=BaseLLM), num_workers=2, agent_manager=agent_manager) + + manager_a = evaluator._create_new_agent_manager() + manager_b = evaluator._create_new_agent_manager() + self.assertIsNot(manager_a.agents[0], manager_b.agents[0]) + + barrier = threading.Barrier(2) + results = {} + + def run_example(manager, example_id): + agent = manager.agents[0] + barrier.wait() + for i in range(5): + agent.short_term_memory.add_message( + Message(content=f"example-{example_id}-msg-{i}", msg_type=MessageType.INPUT) + ) + results[example_id] = [m.content for m in agent.short_term_memory.get(n=5)] + + threads = [ + threading.Thread(target=run_example, args=(manager_a, "A")), + threading.Thread(target=run_example, args=(manager_b, "B")), + ] + for t in threads: + t.start() + for t in threads: + t.join() + + self.assertTrue(all(c.startswith("example-A-") for c in results["A"])) + self.assertTrue(all(c.startswith("example-B-") for c in results["B"])) + + if __name__ == '__main__': unittest.main() \ No newline at end of file