具有多个动作的智能体

如果一个智能体只能执行一个动作，那就不需要智能体这个概念，直接通过运行动作本身即可。

智能体的特点在于动作的组合，以及思考、记忆等操作。通过协同多个动作，可以构建一个工作流程，使智能体可以能够完成更复杂的任务。

假设现在需要输入一句自然语言，希望MetaGPT生成代码并执行，下面实现一下：

在MetaGPT代码里，Role类就是智能体的逻辑抽象，Action是动作的抽象。
那这个需求我们需要一个智能体叫RunnableCoder继承自Role
需要两个动作：SimpleWriteCode和SimpleRunCode，一个写代码的动作和一个运行代码的动作，均继承Action

定义动作

SimpleWriteCode

SimpleWriteCode是一个围绕prompt和LLM调用的动作，在这个动作里完成prompt构造和LLM的调用。

from metagpt.actions import Action

class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction} and provide two runnnable test cases.
    Return ```python your_code_here ``` with NO other texts,
    your code:
    """
    name: str = "SimpleWriteCode"

    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

SimpleRunCode

SimpleRunCode不需要LLM，只需要启动一个子进行来运行代码并获取结果。

class SimpleRunCode(Action):
    name: str = "SimpleRunCode"

    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

定义角色

Role可以执行特定的Action，可以拥有记忆、思考并采用各种策略。定义一个RunnableCoder角色步骤如下：

为其指定一个名称（name）和配置文件（profile）
在__init__中使用self.set_action为其初始化动作（Action）
指定每次 Role 会选择哪个 Action。我们将 react_mode 设置为 “by_order”，这意味着 Role 将按照 self.set_actions 中指定的顺序执行其能够执行的 Action。在这种情况下，当 Role 执行 _act 时，self.rc.todo 将首先是 SimpleWriteCode，然后是 SimpleRunCode。
重写 _act 函数。Role 从上一轮的人类输入或动作输出中检索消息，用适当的 Message 内容提供当前的 Action (self.rc.todo)，最后返回由当前 Action 输出组成的 Message。

class RunnableCoder(Role):
    name: str = "Alice"
    profile: str = "RunnableCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteCode, SimpleRunCode])
        self._set_react_mode(react_mode="by_order")

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        # react_mode="by_order"
        # todo 将会首先是 SimpleWriteCode() 然后再是 SimpleRunCode()
        todo = self.rc.todo

        msg = self.get_memories(k=1)[0]  # find the most k recent messages
        result = await todo.run(msg.content)

        msg = Message(content=result, role=self.profile, cause_by=type(todo))
        self.rc.memory.add(msg)
        return msg

运行

import asyncio

from metagpt.context import Context

async def main():
    msg = "write a function that calculates the sum of a list"
    context = Context()
    role = RunnableCoder(context=context)
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)

asyncio.run(main)

多个智能体组成的团队

现在想要使用MetaGPT构建一个软件开发团队，包含三个角色：

Coder：具有WriteCode动作，接受用户指令编写代码
Tester：具有WriteTest动作，获取Coder编写好的代码并为其提供测试套件
Reviewer：具有WriteReview动作，审查Tester输出的测试用力，并检查其测试范围和质量

定义动作

class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction}.
    Return ```python your_code_here ``` with NO other texts,
    your code:
    """
    name: str = "SimpleWriteCode"

    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
        rsp = await self._aask(prompt)
        code_text = parse_code(rsp)
        return code_text


class SimpleWriteTest(Action):
    PROMPT_TEMPLATE: str = """
    Context: {context}
    Write {k} unit tests using pytest for the given function, assuming you have imported it.
    Return ```python your_code_here ``` with NO other texts,
    your code:
    """

    name: str = "SimpleWriteTest"

    async def run(self, context: str, k: int = 3):
        prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)
        rsp = await self._aask(prompt)
        code_text = parse_code(rsp)
        return code_text


class SimpleWriteReview(Action):
    PROMPT_TEMPLATE: str = """
    Context: {context}
    Review the test cases and provide one critical comments:
    """

    name: str = "SimpleWriteReview"

    async def run(self, context: str):
        prompt = self.PROMPT_TEMPLATE.format(context=context)
        rsp = await self._aask(prompt)
        return rsp

定义角色

Coder

使用 set_actions 为Role配备适当的 Action，这与设置单智能体相同
多智能体操作逻辑：我们使Role _watch 来自用户或其他智能体的重要上游消息。回想我们的SOP，SimpleCoder接收用户指令，这是由MetaGPT中的UserRequirement引起的Message。因此，我们添加了 self._watch([UserRequirement])。

class SimpleCoder(Role):
    name: str = "Alice"
    profile: str = "SimpleCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._watch([UserRequirement])
        self.set_actions([SimpleWriteCode])

Tester

使用 set_actions 为SimpleTester配备 SimpleWriteTest 动作
使Role _watch 来自其他智能体的重要上游消息。SimpleTester从 SimpleCoder 中获取主代码，这是由 SimpleWriteCode 引起的 Message。因此，我们添加了 self._watch([SimpleWriteCode])。
重写 _act 函数，在这里，我们希望SimpleTester将所有记忆用作编写测试用例的上下文，并希望有5个测试用例。

class SimpleTester(Role):
    name: str = "Bob"
    profile: str = "SimpleTester"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteTest])
        self._watch([SimpleWriteCode])
        # self._watch([SimpleWriteCode, SimpleWriteReview])  # feel free to try this too

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo

        # context = self.get_memories(k=1)[0].content # use the most recent memory as context
        context = self.get_memories()  # use all memories as context

        code_text = await todo.run(context, k=5)  # specify arguments
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg

Reviwer

同样设置Reviwer的Action，设置watch观察Test的输出

class SimpleReviewer(Role):
    name: str = "Charlie"
    profile: str = "SimpleReviewer"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteReview])
        self._watch([SimpleWriteTest])

创建团队并添加角色

现在已经定义了三个Role，现在初始化这些Role，设置一个Team，hire这些角色，使其一起协同工作。

import asyncio
import typer
from metagpt.logs import logger
from metagpt.team import Team
app = typer.Typer()

@app.command()
def main(
    idea: str = typer.Argument(..., help="write a function that calculates the product of a list"),
    investment: float = typer.Option(default=3.0, help="Dollar amount to invest in the AI company."),
    n_round: int = typer.Option(default=5, help="Number of rounds for the simulation."),
):
    logger.info(idea)

    team = Team()
    team.hire(
        [
            SimpleCoder(),
            SimpleTester(),
            SimpleReviewer(),
        ]
    )

    team.invest(investment=investment)
    team.run_project(idea)
    await team.run(n_round=n_round)

if __name__ == '__main__':
    app()

以下是整个软件开发团队的运作流程：

MetaGPT学习 - 智能体实践

具有多个动作的智能体

定义动作

定义角色

运行

多个智能体组成的团队

定义动作

定义角色

Coder

Tester

Reviwer

创建团队并添加角色