一、前期环境准备

相关参考文档：

• MetaGPT Doc

• geekan/MetaGPT GitHub

• Hugging Muti Agent（二月学习）

• 智谱AI 接口文档

Python3.9 及以上环境

$ python --version
Python 3.10.13

安装 MetaGPT

$ pip install -i https://pypi.tuna.tsinghua.edu.cn/simple metagpt
$ pip list | grep metagpt
metagpt                      0.7.3

配置 MetaGPT

$ metagpt --init-config
$ vim ~/.metagpt/config2.yaml

llm:
  api_type: 'zhipuai'
  api_key: '从https://open.bigmodel.cn/usercenter/apikeys获取的api-key'
  model: 'glm-4'

测试示例

# main.py
import asyncio

from metagpt.actions import Action
from metagpt.environment import Environment
from metagpt.roles import Role
from metagpt.team import Team

action1 = Action(name="AlexSay", instruction="Express your opinion with emotion and don't repeat it")
action2 = Action(name="BobSay", instruction="Express your opinion with emotion and don't repeat it")
alex = Role(name="Alex", profile="Democratic candidate", goal="Win the election", actions=[action1], watch=[action2])
bob = Role(name="Bob", profile="Republican candidate", goal="Win the election", actions=[action2], watch=[action1])
env = Environment(desc="US election live broadcast")
team = Team(investment=10.0, env=env, roles=[alex, bob])

asyncio.run(team.run(idea="Topic: climate change. Under 80 words per message.", send_to="Alex", n_round=5))

二、智能体综述及多智能体框架

2.1 AI Agent 介绍

个人理解，AI Agent 是一种通过各种途径输入信息，并把这些信息进一步分解和规划，最后将结果依次馈入 LLM 大模型。或者说是具有多种不同模式的 LLM 聚合体，

https://e2b.dev/ai-agents

2.2 多智能体框架介绍

MetaGPT 是一种先进的元编程技术，它利用了大规模语言模型（LLM）和多智能体系统来模拟软件工程团队中的协作过程。通过编码标准操作程序（SOP）为提示，MetaGPT能够复制一个软件研发团队的结构和协作方式，以自动化和优化软件开发流程。旨在通过整合人类的程序知识来提高软件的鲁棒性，减少错误，并为复杂任务设计高效的软件解决方案。MetaGPT能够扮演不同的角色，如产品经理、架构师、项目经理和工程师，从而覆盖软件开发过程中的多个方面。只需要一个简单的需求输入，MetaGPT 就能够生成用户故事、竞品分析、需求文档、数据结构、API设计以及其他相关文档。它通过模拟不同角色的职责，高效地组织工作流程，并能够在给定任务的环境下主动观察和检索相关信息。

三、单智能体开发

3.1

v0.7.3 https://github.com/geekan/MetaGPT/blob/v0.7.3/metagpt/roles/role.py#L88

class RoleContext(BaseModel):
    """Role Runtime Context"""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    # # env exclude=True to avoid `RecursionError: maximum recursion depth exceeded in comparison`
    env: "Environment" = Field(default=None, exclude=True)  # # avoid circular import
    # TODO judge if ser&deser
    msg_buffer: MessageQueue = Field(
        default_factory=MessageQueue, exclude=True
    )  # Message Buffer with Asynchronous Updates
    memory: Memory = Field(default_factory=Memory)
    # long_term_memory: LongTermMemory = Field(default_factory=LongTermMemory)
    working_memory: Memory = Field(default_factory=Memory)
    state: int = Field(default=-1)  # -1 indicates initial or termination state where todo is None
    todo: Action = Field(default=None, exclude=True)
    watch: set[str] = Field(default_factory=set)
    news: list[Type[Message]] = Field(default=[], exclude=True)  # TODO not used
    react_mode: RoleReactMode = (
        RoleReactMode.REACT
    )  # see `Role._set_react_mode` for definitions of the following two attributes
    max_react_loop: int = 1

    def check(self, role_id: str):
        # if hasattr(CONFIG, "enable_longterm_memory") and CONFIG.enable_longterm_memory:
        #     self.long_term_memory.recover_memory(role_id, self)
        #     self.memory = self.long_term_memory  # use memory to act as long_term_memory for unify operation
        pass

    @property
    def important_memory(self) -> list[Message]:
        """Retrieve information corresponding to the attention action."""
        return self.memory.get_by_actions(self.watch)

    @property
    def history(self) -> list[Message]:
        return self.memory.get()

    @classmethod
    def model_rebuild(cls, **kwargs):
        from metagpt.environment.base_env import Environment  # noqa: F401

        super().model_rebuild(**kwargs)

env：Environment 对象，当在 Environment 添加 Role 时会同时设置 Role 对 Environment 的引用。
msg_buffer：一个 MessageQueue 对象，该对象是对 asyncio 的 Queue 进行简单封装，主要是提供了非阻塞的 pop / push 方法。Role 通过该对象与环境中的其他 Role 进行信息交互。
memory：记忆对象。当 Role 执行 _act 时，会将执行得到的响应转换为 Message 对象放入 memory 中。另外当 Role 执行 _observe 时，会把 msg_buffer 的所有消息转移到 memory 中。
state：记录 Role 的执行状态。初始状态值为 -1，当全部 Action 执行完成之后也会被重置为 -1。
todo：下一个待执行的 Action。当 state >= 0 时会指向最后一个 Action。
watch：用 str 表示的当前 Role 观察的 Action 列表，目前用在 _observe 获取 news 时进行消息过滤。
news：存储那些在本次执行 _observe 时读取到的与当前 Role 上下游相关的消息。
react_mode：ReAct 循环的模式，目前支持 REACT、BY_ORDER、PLAN_AND_ACT 3种模式，默认使用 REACT 模式。在 _set_react_mode 方法中有相关说明。简单来说，BY_ORDER 模式按照指定的 Action 顺序执行。PLAN_AND_ACT 则为一次思考后执行多个动作，即 _think -> _act -> act -> ...，而 REACT 模式按照 ReAct 论文中的思考——行动循环来执行，即 _think -> _act -> _think -> _act -> ...。
max_react_loop：在 react_mode 为 REACT 模式时生效，用于设置最大的思考-循环次数，超过后会停止 _react 执行。

class Role(SerializationMixin, ContextMixin, BaseModel):
    """Role/Agent"""

    model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")

    name: str = ""
    profile: str = ""
    goal: str = ""
    constraints: str = ""
    desc: str = ""
    is_human: bool = False

    role_id: str = ""
    states: list[str] = []

    # scenarios to set action system_prompt:
    #   1. `__init__` while using Role(actions=[...])
    #   2. add action to role while using `role.set_action(action)`
    #   3. set_todo while using `role.set_todo(action)`
    #   4. when role.system_prompt is being updated (e.g. by `role.system_prompt = "..."`)
    # Additional, if llm is not set, we will use role's llm
    actions: list[SerializeAsAny[Action]] = Field(default=[], validate_default=True)
    rc: RoleContext = Field(default_factory=RoleContext)
    addresses: set[str] = set()
    planner: Planner = Field(default_factory=Planner)

    # builtin variables
    recovered: bool = False  # to tag if a recovered role
    latest_observed_msg: Optional[Message] = None  # record the latest observed message when interrupted

    __hash__ = object.__hash__  # support Role as hashable type in `Environment.members`
    
    # ......

3.2 最简单的Agent

要自己实现一个最简单的Role，只需要重写Role基类的 _init_ 与 _act 方法。在 _init_ 方法中，我们需要声明 Agent 的name（名称）profile（类型）
我们使用 self._init_action 函数为其配备期望的动作 SimpleWriteCode 这个Action 应该能根据我们的需求生成我们期望的代码在 _act方法中，我们需要编写智能体具体的行动逻辑，智能体将从最新的记忆中获取人类指令，运行配备的动作，MetaGPT将其作为待办事项 (self.rc.todo) 在幕后处理，最后返回一个完整的消息。

实现 SimpleCoder

import asyncio
import re
import subprocess
import fire
from metagpt.actions import Action
from metagpt.logs import logger
from metagpt.roles.role import Role, RoleReactMode
from metagpt.schema import Message


# 编写SimpleWriteCode动作
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


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


# 设计SimpleCoder角色
class SimpleCoder(Role):
    name: str = "Alice"
    profile: str = "SimpleCoder"

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

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

        msg = self.get_memories(k=1)[0]  # find the most recent messages
        code_text = await todo.run(msg.content)
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg


async def main():
    msg="write a function that calculates the product of a list and run it"
    role = SimpleCoder()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)


if __name__ == "__main__":
    asyncio.run(main())

3.3 实现一个多动作Agent

import asyncio
import re
import subprocess
from metagpt.actions import Action
from metagpt.logs import logger
from metagpt.roles.role import Role, RoleReactMode
from metagpt.schema import Message


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


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


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

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

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

        msg = self.get_memories(k=1)[0]  # find the most recent messages
        code_text = await todo.run(msg.content)
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg


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=RoleReactMode.BY_ORDER.value)

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        # By choosing the Action by order under the hood
        # todo will be first SimpleWriteCode() then 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


# 运行SimpleCoder角色
async def main():
    msg = "write a function that calculates the product of a list and run it"
    role = RunnableCoder()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)


if __name__ == "__main__":
    asyncio.run(main())