diff --git a/examples/di/atomization_capacity_plan.py b/examples/di/atomization_capacity_plan.py new file mode 100644 index 000000000..c84a2376b --- /dev/null +++ b/examples/di/atomization_capacity_plan.py @@ -0,0 +1,78 @@ +from metagpt.logs import logger +from metagpt.roles.di.team_leader import TeamLeader +import fire + + +async def main(): + tl = TeamLeader() + logger.info("\n=== Adding Initial Tasks ===") + tl.planner.plan.append_task( + task_id="T1", + dependent_task_ids=[], + instruction="Create Product Requirements Document (PRD)", + assignee="Alice" + ) + tl.planner.plan.append_task( + task_id="T2", + dependent_task_ids=["T1"], + instruction="Design System Architecture", + assignee="Bob" + ) + + # 2. Add Development Tasks + logger.info("\n=== Adding Development Tasks ===") + tl.planner.plan.append_task( + task_id="T3", + dependent_task_ids=["T2"], + instruction="Implement Core Function Modules", + assignee="Alex" + ) + + tl.planner.plan.append_task( + task_id="T4", + dependent_task_ids=["T2"], + instruction="Implement User Interface", + assignee="Alex" + ) + + # 3. Complete Some Tasks + logger.info("\n=== Execute and Complete Tasks ===") + logger.info(f"Current Task: {tl.planner.plan.current_task.instruction}") + tl.planner.plan.finish_current_task() # Complete T1 + + logger.info(f"Current Task: {tl.planner.plan.current_task.instruction}") + tl.planner.plan.finish_current_task() # Complete T2 + + # 4. Replace Tasks + logger.info("\n=== Replace Task ===") + tl.planner.plan.replace_task( + task_id="T3", + new_dependent_task_ids=["T2"], + new_instruction="Implement Core Function Modules (Add New Features)", + new_assignee="Senior_Developer" + ) + + # 5. Add Testing Tasks + logger.info("\n=== Add Testing Tasks ===") + tl.planner.plan.append_task( + task_id="T5", + dependent_task_ids=["T3", "T4"], + instruction="Execute Integration Tests", + assignee="Edward" + ) + + # 6. Reset Task Demonstration + logger.info("\n=== Reset Task ===") + logger.info("Reset Task T3 (This will also reset T5 which depends on it)") + tl.planner.plan.reset_task("T3") + + # Display Final Status + logger.info("\n=== Final Status ===") + logger.info(f"Completed Tasks: {len([t for t in tl.planner.plan.tasks if t.is_finished])}") + logger.info(f"Current Task: {tl.planner.plan.current_task.instruction}") + logger.info("All Tasks:") + for task in tl.planner.plan.tasks: + logger.info(f"- {task.task_id}: {task.instruction} (Completed: {task.is_finished})") + +if __name__ == "__main__": + fire.Fire(main) diff --git a/examples/di/automated_planning_of_tasks.py b/examples/di/automated_planning_of_tasks.py new file mode 100644 index 000000000..d3b04ec1f --- /dev/null +++ b/examples/di/automated_planning_of_tasks.py @@ -0,0 +1,22 @@ +import fire + +from metagpt.logs import logger +from metagpt.roles.di.team_leader import TeamLeader + + +async def main(): + # Create an instance of TeamLeader + tl = TeamLeader() + + # Update the plan with the goal to create a 2048 game + # This will auto generate tasks needed to accomplish the goal + await tl.planner.update_plan(goal="create a 2048 game.") + + # Iterate through all tasks in the plan + # Log each task's ID, instruction and completion status + for task in tl.planner.plan.tasks: + logger.info(f"- {task.task_id}: {task.instruction} (Completed: {task.is_finished})") + + +if __name__ == "__main__": + fire.Fire(main) diff --git a/examples/di/data_analyst_write_code.py b/examples/di/data_analyst_write_code.py new file mode 100644 index 000000000..30afa410c --- /dev/null +++ b/examples/di/data_analyst_write_code.py @@ -0,0 +1,40 @@ +import fire + +from metagpt.logs import logger +from metagpt.roles.di.data_analyst import DataAnalyst + + +async def main(): + # Create an instance of DataAnalyst role + analyst = DataAnalyst() + + # Set the main goal for the planner - constructing a 2D array + analyst.planner.plan.goal = "construct a two-dimensional array" + + # Add a specific task to the planner with detailed parameters: + # - task_id: Unique identifier for the task + # - dependent_task_ids: List of tasks that need to be completed before this one (empty in this case) + # - instruction: Description of what needs to be done + # - assignee: Who will execute the task (David) + # - task_type: Category of the task (DATA_ANALYSIS) + analyst.planner.plan.append_task( + task_id="1", + dependent_task_ids=[], + instruction="construct a two-dimensional array", + assignee="David", + task_type="DATA_ANALYSIS", + ) + + # Execute the code generation and execution for creating a 2D array + # The write_and_exec_code method will: + # 1. Generate the necessary code for creating a 2D array + # 2. Execute the generated code + # 3. Return the result + result = await analyst.write_and_exec_code("construct a two-dimensional array") + + # Log the result of the code execution + logger.info(result) + + +if __name__ == "__main__": + fire.Fire(main) diff --git a/examples/di/interacting_with_human.py b/examples/di/interacting_with_human.py new file mode 100644 index 000000000..c0ce02a40 --- /dev/null +++ b/examples/di/interacting_with_human.py @@ -0,0 +1,37 @@ +import fire +from metagpt.environment.mgx.mgx_env import MGXEnv +from metagpt.logs import logger +from metagpt.roles.di.team_leader import TeamLeader +from metagpt.schema import Message + + +async def main(): + # Initialize the MetaGPT environment + env = MGXEnv() + # Add a TeamLeader role to the environment + env.add_roles([TeamLeader()]) + + # Get input from human user about what they want to do + human_rsp = await env.ask_human("What do you want to do？") + + # Log the human response for tracking + logger.info(human_rsp) + # Create and publish a message with the human response in the environment + env.publish_message(Message(content=human_rsp, role="user")) + + # Get the TeamLeader role instance named 'Mike' + tl = env.get_role("Mike") + # Execute the TeamLeader's tasks + await tl.run() + + # Log information about each task in the TeamLeader's plan + for task in tl.planner.plan.tasks: + logger.info(f"- {task.task_id}: {task.instruction} (Completed: {task.is_finished})") + + # Send an empty response back to the human and log it + resp = await env.reply_to_human("") + logger.info(resp) + + +if __name__ == "__main__": + fire.Fire(main) diff --git a/examples/use_off_the_shelf_agent.py b/examples/use_off_the_shelf_agent.py index 4445a6c62..0d224a92b 100644 --- a/examples/use_off_the_shelf_agent.py +++ b/examples/use_off_the_shelf_agent.py @@ -7,11 +7,20 @@ import asyncio from metagpt.logs import logger from metagpt.roles.product_manager import ProductManager +from metagpt.environment.mgx.mgx_env import MGXEnv +from metagpt.schema import Message +from metagpt.roles.di.team_leader import TeamLeader async def main(): msg = "Write a PRD for a snake game" - role = ProductManager() + env = MGXEnv() + env.add_roles([TeamLeader(), ProductManager()]) + env.publish_message(Message(content=msg, role="user")) + tl = env.get_role("Mike") + await tl.run() + + role = env.get_role("Alice") result = await role.run(msg) logger.info(result.content[:100]) diff --git a/examples/write_design.py b/examples/write_design.py new file mode 100644 index 000000000..7eaa1a87b --- /dev/null +++ b/examples/write_design.py @@ -0,0 +1,24 @@ +import asyncio + +from metagpt.logs import logger +from metagpt.roles.architect import Architect +from metagpt.environment.mgx.mgx_env import MGXEnv +from metagpt.schema import Message +from metagpt.roles.di.team_leader import TeamLeader + + +async def main(): + msg = "Write a TRD for a snake game" + env = MGXEnv() + env.add_roles([TeamLeader(), Architect()]) + env.publish_message(Message(content=msg, role="user")) + tl = env.get_role("Mike") + await tl.run() + + role = env.get_role("Bob") + result = await role.run(msg) + logger.info(result) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/write_game_code.py b/examples/write_game_code.py new file mode 100644 index 000000000..f7b51b581 --- /dev/null +++ b/examples/write_game_code.py @@ -0,0 +1,24 @@ +import asyncio + +from metagpt.logs import logger +from metagpt.environment.mgx.mgx_env import MGXEnv +from metagpt.schema import Message +from metagpt.roles.di.team_leader import TeamLeader +from metagpt.roles.di.engineer2 import Engineer2 + + +async def main(): + msg = "Write code for a 2048 game" + env = MGXEnv() + env.add_roles([TeamLeader(), Engineer2()]) + env.publish_message(Message(content=msg, role="user")) + tl = env.get_role("Mike") + await tl.run() + + role = env.get_role("Alex") + result = await role.run(msg) + logger.info(result) + + +if __name__ == "__main__": + asyncio.run(main())