From c0ce28659d74014af2e31014619e2132986e3059 Mon Sep 17 00:00:00 2001 From: yzlin Date: Fri, 15 Sep 2023 22:40:22 +0800 Subject: [PATCH] basic examples --- examples/build_customized_agent.py | 136 ++++++++++++++++++++++++++ examples/debate.py | 145 ++++++++++++++++++++++++++++ examples/use_off_the_shelf_agent.py | 13 +++ 3 files changed, 294 insertions(+) create mode 100644 examples/build_customized_agent.py create mode 100644 examples/debate.py create mode 100644 examples/use_off_the_shelf_agent.py diff --git a/examples/build_customized_agent.py b/examples/build_customized_agent.py new file mode 100644 index 000000000..7122a9858 --- /dev/null +++ b/examples/build_customized_agent.py @@ -0,0 +1,136 @@ +import re +import subprocess + +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) + # logger.info(prompt) + + 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 + # exec(code_text) + 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] + + 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 + code_text = await SimpleWriteCode().run(instruction) + msg = Message(content=code_text, role=self.profile, cause_by=todo) + + elif isinstance(todo, SimpleRunCode): + code_text = msg.content + rsp = await SimpleRunCode().run(code_text) + msg = Message(content=rsp, 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) + +if __name__ == "__main__": + import asyncio + + async def main(): + msg = "write a function that calculates the sum of a list" + # role = SimpleCoder() + role = RunnableCoder() + logger.info(msg) + result = await role.run(msg) + logger.info(result) + + asyncio.run(main()) \ No newline at end of file diff --git a/examples/debate.py b/examples/debate.py new file mode 100644 index 000000000..c98705fcc --- /dev/null +++ b/examples/debate.py @@ -0,0 +1,145 @@ +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 Shout(Action): + + 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="Shout", 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 = "Trump", + **kwargs, + ): + super().__init__(name, profile, **kwargs) + self._init_actions([Shout]) + self._watch([Shout]) + self.name = "Trump" + self.opponent_name = "Biden" + + async def _observe(self) -> int: + await super()._observe() + self._rc.news = [ + msg for msg in self._rc.news if msg.send_to == self.name + ] # only relevant msgs count as observed news + 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([Shout]) + context = [] + for m in msg_history: + context.append(str(m)) + context = "\n".join(context) + + rsp = await Shout().run(context=context, name=self.name, opponent_name=self.opponent_name) + + msg = Message( + content=rsp, + role=self.profile, + cause_by=Shout, + sent_from=self.name, + send_to=self.opponent_name, + ) + self._publish_message(msg) + + return msg + +class Biden(Role): + def __init__( + self, + name: str = "Biden", + profile: str = "Biden", + **kwargs, + ): + super().__init__(name, profile, **kwargs) + self._init_actions([Shout]) + self._watch([BossRequirement, Shout]) + self.name = "Biden" + self.opponent_name = "Trump" + + async def _observe(self) -> int: + await super()._observe() + self._rc.news = [ + msg for msg in self._rc.news if msg.send_to == self.name or msg.cause_by == BossRequirement + ] # only relevant msgs count as observed news + 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, Shout]) + context = [] + for m in msg_history: + context.append(str(m)) + context = "\n".join(context) + + rsp = await Shout().run(context=context, name=self.name, opponent_name=self.opponent_name) + + msg = Message( + content=rsp, + role=self.profile, + cause_by=Shout, + sent_from=self.name, + send_to=self.opponent_name, + ) + self._publish_message(msg) + + return msg + +async def startup(idea: str, investment: float = 3.0, n_round: int = 5, + code_review: bool = False, run_tests: bool = False): + """Run a startup of presidents. 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, code_review: bool = False, run_tests: bool = False): + """ + We are a software startup comprised of AI. By investing in us, you are empowering a future filled with limitless possibilities. + :param idea: Your innovative idea, such as "Creating a snake game." + :param investment: As an investor, you have the opportunity to contribute a certain dollar amount to this AI company. + :param n_round: + :param code_review: Whether to use code review. + :return: + """ + if platform.system() == "Windows": + asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) + asyncio.run(startup(idea, investment, n_round, code_review, run_tests)) + + +if __name__ == '__main__': + fire.Fire(main) diff --git a/examples/use_off_the_shelf_agent.py b/examples/use_off_the_shelf_agent.py new file mode 100644 index 000000000..48fb2b19e --- /dev/null +++ b/examples/use_off_the_shelf_agent.py @@ -0,0 +1,13 @@ +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())