diff --git a/src/crewai/crew.py b/src/crewai/crew.py index 9cecfed3a..8f24d7ac0 100644 --- a/src/crewai/crew.py +++ b/src/crewai/crew.py @@ -1236,6 +1236,86 @@ class Crew(BaseModel): def __repr__(self): return f"Crew(id={self.id}, process={self.process}, number_of_agents={len(self.agents)}, number_of_tasks={len(self.tasks)})" + def to_structured_dict(self): + """Return a structured dictionary representation of the Crew object. + + This method provides a frontend-friendly representation of the Crew's + structure and relationships, suitable for visualization and rendering. + + Returns: + Dict: A structured dictionary containing the Crew's configuration, + agents, tasks, and their relationships. + """ + # Basic crew information + result = { + "id": str(self.id), + "name": self.name, + "process": str(self.process), + "verbose": self.verbose, + "memory": self.memory, + "agents": [], + "tasks": [], + "task_relationships": [] + } + + # Add agent information + for agent in self.agents: + agent_info = { + "id": str(agent.id), + "role": agent.role, + "goal": agent.goal, + "backstory": agent.backstory, + "allow_delegation": agent.allow_delegation, + "verbose": agent.verbose + } + result["agents"].append(agent_info) + + # Add task information - safely handle potential errors + for task in self.tasks: + try: + task_info = { + "id": str(task.id), + "description": task.description, + "expected_output": task.expected_output + } + + # Safely add agent information if available + try: + if hasattr(task, 'agent') and task.agent: + task_info["agent"] = task.agent.role + except Exception: + # Skip adding agent info if there's an error + pass + + # Safely add async_execution if available + if hasattr(task, 'async_execution'): + task_info["async_execution"] = task.async_execution + + result["tasks"].append(task_info) + + # Add task relationships if context exists + if hasattr(task, 'context') and task.context: + relationship = { + "task_id": str(task.id), + "depends_on": [str(context_task.id) for context_task in task.context] + } + result["task_relationships"].append(relationship) + except Exception as e: + print(f"Error processing task: {e}") + + # Add manager agent if exists + if hasattr(self, 'manager_agent') and self.manager_agent: + result["manager_agent"] = { + "id": str(self.manager_agent.id), + "role": self.manager_agent.role, + "goal": self.manager_agent.goal, + "backstory": self.manager_agent.backstory, + "allow_delegation": self.manager_agent.allow_delegation, + "verbose": self.manager_agent.verbose + } + + return result + def reset_memories(self, command_type: str) -> None: """Reset specific or all memories for the crew. diff --git a/tests/test_crew_structured_dict.py b/tests/test_crew_structured_dict.py new file mode 100644 index 000000000..82f5d8833 --- /dev/null +++ b/tests/test_crew_structured_dict.py @@ -0,0 +1,175 @@ +import unittest +from unittest.mock import patch, MagicMock, PropertyMock +import json +import uuid + +from crewai.agent import Agent +from crewai.crew import Crew +from crewai.task import Task +from crewai.process import Process + +class TestCrewStructuredDict(unittest.TestCase): + def setUp(self): + # Create test agents + self.researcher = Agent( + role="Researcher", + goal="Research and gather information", + backstory="You are an expert researcher with a keen eye for detail." + ) + + self.writer = Agent( + role="Writer", + goal="Write compelling content", + backstory="You are a skilled writer who can create engaging content." + ) + + # Create test tasks + self.research_task = Task( + description="Research the latest AI developments", + expected_output="A summary of the latest AI developments", + agent=self.researcher + ) + + self.writing_task = Task( + description="Write an article about AI developments", + expected_output="A well-written article about AI developments", + agent=self.writer, + context=[self.research_task] + ) + + # Create a crew with tasks + self.crew = Crew( + agents=[self.researcher, self.writer], + tasks=[self.research_task, self.writing_task], + process=Process.sequential + ) + + # Create a crew with manager + self.manager = Agent( + role="Manager", + goal="Manage the team", + backstory="You are an experienced manager who oversees projects." + ) + + self.hierarchical_crew = Crew( + agents=[self.researcher, self.writer], + tasks=[self.research_task], + process=Process.hierarchical, + manager_agent=self.manager + ) + + def test_to_structured_dict_basic_properties(self): + """Test that to_structured_dict includes basic Crew properties.""" + result = self.crew.to_structured_dict() + + # Check basic properties + self.assertEqual(str(self.crew.id), result["id"]) + self.assertEqual(self.crew.name, result["name"]) + self.assertEqual(str(self.crew.process), result["process"]) + self.assertEqual(self.crew.verbose, result["verbose"]) + self.assertEqual(self.crew.memory, result["memory"]) + + def test_to_structured_dict_agents(self): + """Test that to_structured_dict includes agent information.""" + result = self.crew.to_structured_dict() + + # Check agents + self.assertEqual(len(self.crew.agents), len(result["agents"])) + + # Check first agent properties + agent_result = result["agents"][0] + agent = self.crew.agents[0] + self.assertEqual(str(agent.id), agent_result["id"]) + self.assertEqual(agent.role, agent_result["role"]) + self.assertEqual(agent.goal, agent_result["goal"]) + self.assertEqual(agent.backstory, agent_result["backstory"]) + + def test_to_structured_dict_tasks(self): + """Test that to_structured_dict includes task information.""" + result = self.crew.to_structured_dict() + + # Check tasks + self.assertEqual(len(self.crew.tasks), len(result["tasks"])) + + # Check first task properties + task_result = result["tasks"][0] + task = self.crew.tasks[0] + self.assertEqual(str(task.id), task_result["id"]) + self.assertEqual(task.description, task_result["description"]) + self.assertEqual(task.expected_output, task_result["expected_output"]) + self.assertEqual(task.agent.role, task_result["agent"]) + + def test_to_structured_dict_task_relationships(self): + """Test that to_structured_dict includes task relationships.""" + result = self.crew.to_structured_dict() + + # Check task relationships + self.assertTrue("task_relationships" in result) + self.assertEqual(1, len(result["task_relationships"])) + + # Check relationship properties + relationship = result["task_relationships"][0] + self.assertEqual(str(self.writing_task.id), relationship["task_id"]) + self.assertEqual(1, len(relationship["depends_on"])) + self.assertEqual(str(self.research_task.id), relationship["depends_on"][0]) + + def test_to_structured_dict_manager_agent(self): + """Test that to_structured_dict includes manager agent information.""" + result = self.hierarchical_crew.to_structured_dict() + + # Check manager agent + self.assertTrue("manager_agent" in result) + manager_result = result["manager_agent"] + self.assertEqual(str(self.manager.id), manager_result["id"]) + self.assertEqual(self.manager.role, manager_result["role"]) + self.assertEqual(self.manager.goal, manager_result["goal"]) + self.assertEqual(self.manager.backstory, manager_result["backstory"]) + + def test_to_structured_dict_empty_tasks(self): + """Test that to_structured_dict handles empty tasks list.""" + crew = Crew( + agents=[self.researcher], + tasks=[], + process=Process.sequential + ) + + result = crew.to_structured_dict() + + # Check empty tasks + self.assertEqual(0, len(result["tasks"])) + self.assertEqual(0, len(result["task_relationships"])) + + def test_to_structured_dict_error_handling(self): + """Test that to_structured_dict handles errors gracefully.""" + # Create a task with a valid agent + task = Task( + description="Test description", + expected_output="Test output", + agent=self.researcher + ) + + # Create a crew with the task + crew = Crew( + agents=[self.researcher], + tasks=[task], + process=Process.sequential + ) + + # Patch the task's context property to raise an exception + # We'll do this by patching the specific method that accesses context + with patch('crewai.crew.Crew._get_context', side_effect=Exception("Test exception")): + # The method should not raise an exception + result = crew.to_structured_dict() + + # Basic verification that the result still contains the task + self.assertEqual(1, len(result["tasks"])) + task_result = result["tasks"][0] + self.assertEqual(str(task.id), task_result["id"]) + self.assertEqual(task.description, task_result["description"]) + self.assertEqual(task.expected_output, task_result["expected_output"]) + + # Verify no task relationships were added due to the exception + self.assertEqual(0, len(result["task_relationships"])) + +if __name__ == "__main__": + unittest.main()