Merge pull request #332 from garylin2099/customized_agent_examples

basic examples

examples/agent_creator.py

@@ -0,0 +1,100 @@
+'''
+Filename: MetaGPT/examples/agent_creator.py
+Created Date: Tuesday, September 12th 2023, 3:28:37 pm
+Author: garylin2099
+'''
+import re
+
+from metagpt.const import PROJECT_ROOT, WORKSPACE_ROOT
+from metagpt.actions import Action
+from metagpt.roles import Role
+from metagpt.schema import Message
+from metagpt.logs import logger
+
+with open(PROJECT_ROOT / "examples/build_customized_agent.py", "r") as f:
+    # use official example script to guide AgentCreator
+    MULTI_ACTION_AGENT_CODE_EXAMPLE = f.read()
+
+class CreateAgent(Action):
+
+    PROMPT_TEMPLATE = """
+    ### BACKGROUND
+    You are using an agent framework called metagpt to write agents capable of different actions,
+    the usage of metagpt can be illustrated by the following example:
+    ### EXAMPLE STARTS AT THIS LINE
+    {example}
+    ### EXAMPLE ENDS AT THIS LINE
+    ### TASK
+    Now you should create an agent with appropriate actions based on the instruction, consider carefully about
+    the PROMPT_TEMPLATE of all actions and when to call self._aask()
+    ### INSTRUCTION
+    {instruction}
+    ### YOUR CODE
+    Return ```python your_code_here ``` with NO other texts, your code:
+    """
+
+    async def run(self, example: str, instruction: str):
+
+        prompt = self.PROMPT_TEMPLATE.format(example=example, instruction=instruction)
+        # logger.info(prompt)
+
+        rsp = await self._aask(prompt)
+
+        code_text = CreateAgent.parse_code(rsp)
+
+        return code_text
+
+    @staticmethod
+    def parse_code(rsp):
+        pattern = r'```python(.*)```'
+        match = re.search(pattern, rsp, re.DOTALL)
+        code_text = match.group(1) if match else ""
+        with open(WORKSPACE_ROOT / "agent_created_agent.py", "w") as f:
+            f.write(code_text)
+        return code_text
+
+class AgentCreator(Role):
+    def __init__(
+        self,
+        name: str = "Matrix",
+        profile: str = "AgentCreator",
+        agent_template: str = MULTI_ACTION_AGENT_CODE_EXAMPLE,
+        **kwargs,
+    ):
+        super().__init__(name, profile, **kwargs)
+        self._init_actions([CreateAgent])
+        self.agent_template = agent_template
+
+    async def _act(self) -> Message:
+        logger.info(f"{self._setting}: ready to {self._rc.todo}")
+        todo = self._rc.todo
+        msg = self._rc.memory.get()[-1]
+
+        instruction = msg.content
+        code_text = await CreateAgent().run(example=self.agent_template, instruction=instruction)
+        msg = Message(content=code_text, role=self.profile, cause_by=todo)
+
+        return msg
+
+if __name__ == "__main__":
+    import asyncio
+
+    async def main():
+
+        agent_template = MULTI_ACTION_AGENT_CODE_EXAMPLE
+
+        creator = AgentCreator(agent_template=agent_template)
+
+        # msg = """Write an agent called SimpleTester that will take any code snippet (str)
+        #     and return a testing code (str) for testing
+        #     the given code snippet. Use pytest as the testing framework."""
+
+        msg = """
+        Write an agent called SimpleTester that will take any code snippet (str) and do the following:
+        1. write a testing code (str) for testing the given code snippet, save the testing code as a .py file in the current working diretory;
+        2. run the testing code.
+        You can use pytest as the testing framework.
+        """
+        await creator.run(msg)
+
+    asyncio.run(main())

examples/build_customized_agent.py

@@ -0,0 +1,139 @@
+'''
+Filename: MetaGPT/examples/build_customized_agent.py
+Created Date: Tuesday, September 19th 2023, 6:52:25 pm
+Author: garylin2099
+'''
+import re
+import subprocess
+import asyncio
+
+import fire
+
+from metagpt.actions import Action
+from metagpt.roles import Role
+from metagpt.schema import Message
+from metagpt.logs import logger
+
+class SimpleWriteCode(Action):
+
+    PROMPT_TEMPLATE = """
+    Write a python function that can {instruction} and provide two runnnable test cases.
+    Return ```python your_code_here ``` with NO other texts,
+    example: 
+    ```python
+    # function
+    def add(a, b):
+        return a + b
+    # test cases
+    print(add(1, 2))
+    print(add(3, 4))
+    ```
+    your code:
+    """
+
+    def __init__(self, name="SimpleWriteCode", context=None, llm=None):
+        super().__init__(name, context, llm)
+
+    async def run(self, instruction: str):
+
+        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
+
+        rsp = await self._aask(prompt)
+
+        code_text = SimpleWriteCode.parse_code(rsp)
+
+        return code_text
+
+    @staticmethod
+    def parse_code(rsp):
+        pattern = r'```python(.*)```'
+        match = re.search(pattern, rsp, re.DOTALL)
+        code_text = match.group(1) if match else rsp
+        return code_text
+
+class SimpleRunCode(Action):
+    def __init__(self, name="SimpleRunCode", context=None, llm=None):
+        super().__init__(name, context, llm)
+
+    async def run(self, code_text: str):
+        result = subprocess.run(["python3", "-c", code_text], capture_output=True, text=True)
+        code_result = result.stdout
+        logger.info(f"{code_result=}")
+        return code_result
+
+class SimpleCoder(Role):
+    def __init__(
+        self,
+        name: str = "Alice",
+        profile: str = "SimpleCoder",
+        **kwargs,
+    ):
+        super().__init__(name, profile, **kwargs)
+        self._init_actions([SimpleWriteCode])
+
+    async def _act(self) -> Message:
+        logger.info(f"{self._setting}: ready to {self._rc.todo}")
+        todo = self._rc.todo
+
+        msg = self._rc.memory.get()[-1] # retrieve the latest memory
+        instruction = msg.content
+
+        code_text = await SimpleWriteCode().run(instruction)
+        msg = Message(content=code_text, role=self.profile, cause_by=todo)
+
+        return msg
+
+class RunnableCoder(Role):
+    def __init__(
+        self,
+        name: str = "Alice",
+        profile: str = "RunnableCoder",
+        **kwargs,
+    ):
+        super().__init__(name, profile, **kwargs)
+        self._init_actions([SimpleWriteCode, SimpleRunCode])
+
+    async def _think(self) -> None:
+        if self._rc.todo is None:
+            self._set_state(0)
+            return
+
+        if self._rc.state + 1 < len(self._states):
+            self._set_state(self._rc.state + 1)
+        else:
+            self._rc.todo = None
+
+    async def _act(self) -> Message:
+        logger.info(f"{self._setting}: ready to {self._rc.todo}")
+        todo = self._rc.todo
+        msg = self._rc.memory.get()[-1]
+
+        if isinstance(todo, SimpleWriteCode):
+            instruction = msg.content
+            result = await SimpleWriteCode().run(instruction)
+
+        elif isinstance(todo, SimpleRunCode):
+            code_text = msg.content
+            result = await SimpleRunCode().run(code_text)
+
+        msg = Message(content=result, role=self.profile, cause_by=todo)
+        self._rc.memory.add(msg)
+        return msg
+
+    async def _react(self) -> Message:
+        while True:
+            await self._think()
+            if self._rc.todo is None:
+                break
+            await self._act()
+        return Message(content="All job done", role=self.profile)
+
+def main(msg="write a function that calculates the sum of a list"):
+    # role = SimpleCoder()
+    role = RunnableCoder()
+    logger.info(msg)
+    result = asyncio.run(role.run(msg))
+    logger.info(result)
+
+if __name__ == '__main__':
+    fire.Fire(main)

examples/debate.py

@@ -0,0 +1,148 @@
+'''
+Filename: MetaGPT/examples/debate.py
+Created Date: Tuesday, September 19th 2023, 6:52:25 pm
+Author: garylin2099
+'''
+import asyncio
+import platform
+import fire
+
+from metagpt.software_company import SoftwareCompany
+from metagpt.actions import Action, BossRequirement
+from metagpt.roles import Role
+from metagpt.schema import Message
+from metagpt.logs import logger
+
+class ShoutOut(Action):
+    """Action: Shout out loudly in a debate (quarrel)"""
+
+    PROMPT_TEMPLATE = """
+    ## BACKGROUND
+    Suppose you are {name}, you are in a debate with {opponent_name}.
+    ## DEBATE HISTORY
+    Previous rounds:
+    {context}
+    ## YOUR TURN
+    Now it's your turn, you should closely respond to your opponent's latest argument, state your position, defend your arguments, and attack your opponent's arguments,
+    craft a strong and emotional response in 80 words, in {name}'s rhetoric and viewpoints, your will argue:
+    """
+
+    def __init__(self, name="ShoutOut", context=None, llm=None):
+        super().__init__(name, context, llm)
+
+    async def run(self, context: str, name: str, opponent_name: str):
+
+        prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name)
+        # logger.info(prompt)
+
+        rsp = await self._aask(prompt)
+
+        return rsp
+
+class Trump(Role):
+    def __init__(
+        self,
+        name: str = "Trump",
+        profile: str = "Republican",
+        **kwargs,
+    ):
+        super().__init__(name, profile, **kwargs)
+        self._init_actions([ShoutOut])
+        self._watch([ShoutOut])
+        self.name = "Trump"
+        self.opponent_name = "Biden"
+
+    async def _observe(self) -> int:
+        await super()._observe()
+        # accept messages sent (from opponent) to self, disregard own messages from the last round
+        self._rc.news = [msg for msg in self._rc.news if msg.send_to == self.name]  
+        return len(self._rc.news)
+
+    async def _act(self) -> Message:
+        logger.info(f"{self._setting}: ready to {self._rc.todo}")
+
+        msg_history = self._rc.memory.get_by_actions([ShoutOut])
+        context = []
+        for m in msg_history:
+            context.append(str(m))
+        context = "\n".join(context)
+
+        rsp = await ShoutOut().run(context=context, name=self.name, opponent_name=self.opponent_name)
+
+        msg = Message(
+            content=rsp,
+            role=self.profile,
+            cause_by=ShoutOut,
+            sent_from=self.name,
+            send_to=self.opponent_name,
+        )
+
+        return msg
+
+class Biden(Role):
+    def __init__(
+        self,
+        name: str = "Biden",
+        profile: str = "Democrat",
+        **kwargs,
+    ):
+        super().__init__(name, profile, **kwargs)
+        self._init_actions([ShoutOut])
+        self._watch([BossRequirement, ShoutOut])
+        self.name = "Biden"
+        self.opponent_name = "Trump"
+
+    async def _observe(self) -> int:
+        await super()._observe()
+        # accept the very first human instruction (the debate topic) or messages sent (from opponent) to self,
+        # disregard own messages from the last round
+        self._rc.news = [msg for msg in self._rc.news if msg.cause_by == BossRequirement or msg.send_to == self.name]
+        return len(self._rc.news)
+
+    async def _act(self) -> Message:
+        logger.info(f"{self._setting}: ready to {self._rc.todo}")
+
+        msg_history = self._rc.memory.get_by_actions([BossRequirement, ShoutOut])
+        context = []
+        for m in msg_history:
+            context.append(str(m))
+        context = "\n".join(context)
+
+        rsp = await ShoutOut().run(context=context, name=self.name, opponent_name=self.opponent_name)
+
+        msg = Message(
+            content=rsp,
+            role=self.profile,
+            cause_by=ShoutOut,
+            sent_from=self.name,
+            send_to=self.opponent_name,
+        )
+
+        return msg
+
+async def startup(idea: str, investment: float = 3.0, n_round: int = 5,
+                  code_review: bool = False, run_tests: bool = False):
+    """We reuse the startup paradigm for roles to interact with each other.
+    Now we run a startup of presidents and watch they quarrel. :) """
+    company = SoftwareCompany()
+    company.hire([Biden(), Trump()])
+    company.invest(investment)
+    company.start_project(idea)
+    await company.run(n_round=n_round)
+
+
+def main(idea: str, investment: float = 3.0, n_round: int = 10):
+    """
+    :param idea: Debate topic, such as "Topic: The U.S. should commit more in climate change fighting" 
+                 or "Trump: Climate change is a hoax"
+    :param investment: contribute a certain dollar amount to watch the debate
+    :param n_round: maximum rounds of the debate
+    :return:
+    """
+    if platform.system() == "Windows":
+        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+    asyncio.run(startup(idea, investment, n_round))
+
+
+if __name__ == '__main__':
+    fire.Fire(main)

examples/use_off_the_shelf_agent.py

@@ -0,0 +1,18 @@
+'''
+Filename: MetaGPT/examples/use_off_the_shelf_agent.py
+Created Date: Tuesday, September 19th 2023, 6:52:25 pm
+Author: garylin2099
+'''
+import asyncio
+
+from metagpt.roles.product_manager import ProductManager
+from metagpt.logs import logger
+
+async def main():
+    msg = "Write a PRD for a snake game"
+    role = ProductManager()
+    result = await role.run(msg)
+    logger.info(result.content[:100])
+
+if __name__ == '__main__':
+    asyncio.run(main())