3. MetaGPT框架组件介绍

• agent运行过程：Role: _observe -> _think -> _act -> publish

• Class Role是agent的逻辑抽象；Role拥有记忆，思考，action策略，etc

3.1 实现一个单动作Agent SimpleCoder

• Override Role base class的__init__() 和 _act();
  • init(): 声明agent name， profile； 使用self._init_action()配备该Role类期望的动作； e.g. 调用SimpleCoder Role -> SimpleWriteCode Action
  • _act(): agent具体的行动逻辑，从记忆中获取人类指令，运行配备动作，e.g. self._rc.todo, 是在__init__中通过_init_action挂载的这个role执行的action； 返回Messeage
• Simplecoder的流程：输入用户需求->记忆需求-> 根据已知信息和需求编写代码
• 定义目标的Action的子类，SimpleWriteCode(Action); 通常override init(), run()

# Action类中包含和大模型通信的方法, self._aask, 是通过llm.aask和大模型的通信
async def_aask(self, prompt: str, system_msgs: Optional[list[str]] = None) -> str:
    """Append default prefix"""
    return await self.llm.aask(prompt. system_msgs)

# 定义Action的子类
class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction} and provide 2 runnable test cases.
    Return ```python your_code_here ``` with NO other texts,
    your code:
    """

    def __init__(self, name="SiimpleWriteCode", context=None, llm=None):
        super().__init__(name, context, llm)

    # @param instruction是由role调用action的run方法传入;
    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLAGE.format(instruction=instruction)
        rsp = await self._aask(prompt)
        code_text = SimpleWriteCode.parse_code(rsp);
        return code_text

# regex 从llm response解析出python代码；
@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

• Message类
  • Message properties: content, role, cause_by; 除content外，其他data membrer是可选的；
    • content: 信息；
    • role: 发出信息的role；
    • cause_by: 是哪个action产生的message

• Role， 解决的问题是如何调用action, 如何把用户信息传递给这个动作；如前面所述，覆盖__init__和_act 2个方法

from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger

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    # todo是在Role中设定的SimpleWriteCoder
    msg = self.get_memories(k=1)[0]    # 取出最近的记忆
    code_text = await todo.run(msg.content)    # 最近记忆任务的content as instruction
    msg = Messsage（content=code_text, role=self.profile, cause_by=type(todo))
    return msg

• 运行角色

import asyncio

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

• 总结一个单动作的Agent：
  • 实现一个Action子类；
  • 实现有一个Role子类
  • 启动role的run()

3.2 实现一个多动作Agent

• Agent的力量，在于Role可以动作的组合，以及记忆; 构建一个工作流程，可以完成复杂任务

• 一个写代码，又立即运行的Role，需要2个Action: SimpleWriteCode, SimpleRunCode
  • SimpleWriteCode 和前面 一致
  • SimpleRunCode， 启动一个子进程运行代码， subprocess fork子进程