【AI Agent系列】【MetaGPT多智能体学习】2. 重温单智能体开发 - 深入源码，理解单智能体运行框架

阅读本书更多章节>>>>

本系列文章跟随《MetaGPT多智能体课程》（https://github.com/datawhalechina/hugging-multi-agent），深入理解并实践多智能体系统的开发。

本文参考该课程的第三章 - 单智能体开发。在第三章的基础上，顺带着看下相关源码，深入理解单智能体的运行框架。

文章目录

0. 单智能体概念及运行框架0.1 概念 - Role0.2 单智能体运行周期0.3 RoleContext - 与环境上下文交互0.4 单智能体的动作定义 - Action 1. 实现一个单智能体1.1 实现步骤1.2 完整代码1.3 升级0.7版本后，教程中原代码的坑1.3.1 non-attribute was detected1.3.2 ‘SimpleCoder’ object has no attribute '_init_actions' 1.4 再加一个Action1.5 多Action之后增加的知识点总结 2. 技术文档助手实践 / OSS订阅智能体实践2.1 技术文档助手相关2.2 OSS订阅智能体相关2.3 智能体开发作业相关 3. 参考资料

0. 单智能体概念及运行框架

这部分为理论介绍和部分源码理解，想直接实战的同学可以直接跳过该部分。

0.1 概念 - Role

在MetaGPT中定义的agent运行周期如下：

一个agent在启动后他会观察自己能获取到的信息，加入自己的记忆中下一步进行思考，决定下一步的行动，也就是从Action1，Action2，Action3中选择执行的Action决定行动后，紧接着就执行对应行动，得到这个环节的结果

如下图，

所谓的单智能体，在MetaGPT中，其实就是一个Role实例。一个 Role 能执行特定的 Action，拥有记忆、思考并采用各种策略行动。

0.2 单智能体运行周期

我之前的文章中讨论过MetaGPT单智能体的运行周期，详细分析了Role类中run函数的运行方式和相关函数（如_observe、react函数）的作用，可以去看下：

_observe函数用来观察环境中的信息，获取关心的上下文react函数用来思考_think + 动作_act然后最后将最终的结果再放到环境中 publish_message

这个流程中，Role有个关键参数：RoleContext，这个类型组织了Role与上下文的交互内容。

0.3 RoleContext - 与环境上下文交互

Role 在与环境上下文进行交互时，是通过内部的 RoleContext 对象来实现的。下面是RoleContext的源码（0.7版本）及一些变量的含义：

class RoleContext(BaseModel):    """Role Runtime Context"""    model_config = ConfigDict(arbitrary_types_allowed=True)    # Environment 对象，当在 Environment 添加 Role 时会同时设置 Role 对 Environment 的引用。    env: "Environment" = Field(default=None, exclude=True)      # 一个 MessageQueue 对象，该对象是对 asyncio 的 Queue 进行简单封装，主要是提供了非阻塞的 pop / push 方法。Role 通过该对象与环境中的其他 Role 进行信息交互。    msg_buffer: MessageQueue = Field(        default_factory=MessageQueue, exclude=True    )      # 记忆对象。当 Role 执行 _act 时，会将执行得到的响应转换为 Message 对象放入 memory 中。另外当 Role 执行 _observe 时，会把 msg_buffer 的所有消息转移到 memory 中。    memory: Memory = Field(default_factory=Memory)    # long_term_memory: LongTermMemory = Field(default_factory=LongTermMemory)    working_memory: Memory = Field(default_factory=Memory)    # 记录 Role 的执行状态。初始状态值为 -1，当全部 Action 执行完成之后也会被重置为 -1。    state: int = Field(default=-1)      # 下一个待执行的 Action。当 state >= 0 时会指向最后一个 Action。    todo: Action = Field(default=None, exclude=True)    # 用 str 表示的当前 Role 观察的 Action 列表，目前用在 _observe 获取 news 时进行消息过滤。    watch: set[str] = Field(default_factory=set)    # 存储那些在本次执行  _observe 时读取到的与当前 Role 上下游相关的消息。    news: list[Type[Message]] = Field(default=[], exclude=True)  # TODO not used    # ReAct 循环的模式，目前支持 REACT、BY_ORDER、PLAN_AND_ACT 3种模式，默认使用 REACT 模式。    react_mode: RoleReactMode = (        RoleReactMode.REACT    )     # 在 react_mode 为 REACT 模式时生效，用于设置最大的思考-循环次数，超过后会停止 _react 执行。    max_react_loop: int = 1    @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()

从上面代码中，可以看到几个重要的变量：

环境，定义智能体所处的环境信息，此智能体的结果和其它属于此环境的智能体的结果会放到里面memory：记忆todo： Action：该智能体能执行的动作列表watch：关心的动作列表

下面，我们来看下赋予智能体行动能力的Action类。

0.4 单智能体的动作定义 - Action

Action定义Role对象可以进行的动作。看下它的源码参数：

class Action(SerializationMixin, ContextMixin, BaseModel):    model_config = ConfigDict(arbitrary_types_allowed=True)    name: str = ""    i_context: Union[        dict, CodingContext, CodeSummarizeContext, TestingContext, RunCodeContext, CodePlanAndChangeContext, str, None    ] = ""    prefix: str = ""  # aask*时会加上prefix，作为system_message    desc: str = ""  # for skill manager    node: ActionNode = Field(default=None, exclude=True)

其中，prefix参数会在调用大模型时作为Prompt的一部分带入。

1. 实现一个单智能体

1.1 实现步骤

要实现一个Agent，其实就是定义一个Role。该Role应该包含自己的Action。

（1）定义Actions，重写run函数，这里面决定了我们对传入的内容到底要做什么样的处理，例如调用大模型得到结果
（2）定义Role，在Role的初始化中初始化Actions
（3）Role重写_act函数或_react函数，Role run的时候会调用该函数。_react函数重写，一般是先思考_think下一步用哪个action，然后再_act

1.2 完整代码

直接上教程中的完整代码（有细微修改，因为升级0.7版本之后有坑）：

import reimport asynciofrom metagpt.actions import Actionfrom metagpt.roles import Rolefrom metagpt.schema import Messagefrom metagpt.logs import loggerclass SimpleWriteCode(Action):    PROMPT_TEMPLATE: str = """    Write a python function that can {instruction} and provide two runnable 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_textclass SimpleCoder(Role):    def __init__(self, **kwargs):        super().__init__(**kwargs)        self.set_actions([SimpleWriteCode])    async def _act(self) -> Message:        logger.info(f"{self._setting}: ready to {self.rc.todo}")        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 msgasync def main():    msg = "write a function that calculates the sum of a list"    role = SimpleCoder()    logger.info(msg)    result = await role.run(msg)    logger.info(result)asyncio.run(main())

运行结果：

1.3 升级0.7版本后，教程中原代码的坑

1.3.1 non-attribute was detected

解决办法：下面代码 PROMPT_TEMPLATE = 换成 PROMPT_TEMPLATE: str =

1.3.2 ‘SimpleCoder’ object has no attribute ‘_init_actions’

解决办法：将’_init_actions’改为’set_actions’

1.4 再加一个Action

定义另一个action：

class SimpleRunCode(Action):    name: str = "SimpleRunCode"    async def run(self, code_text: str):        # 在Windows环境下，result可能无法正确返回生成结果，在windows中在终端中输入python3可能会导致打开微软商店        result = subprocess.run(["python3", "-c", code_text], capture_output=True, text=True)        # 采用下面的可选代码来替换上面的代码        # result = subprocess.run(["python", "-c", code_text], capture_output=True, text=True)        # import sys        # result = subprocess.run([sys.executable, "-c", code_text], capture_output=True, text=True)        code_result = result.stdout        logger.info(f"{code_result=}")        return code_result

修改Role的初始化，将这个Action也加进去：

class SimpleCoder(Role):    def __init__(self, **kwargs):        super().__init__(**kwargs)        self.set_actions([SimpleWriteCode, SimpleRunCode])        self._set_react_mode(react_mode="by_order") # 多个Action之后，要指定一下动作的执行顺序，by_order表示按照顺序执行，by_name表示按照名称执行

运行遇到的坑：NameError: name ‘subprocess’ is not defined
解决办法：import subprocess

1.5 多Action之后增加的知识点总结

（1）self.set_actions([SimpleWriteCode, SimpleRunCode]) 之后要设置动作执行顺序，self._set_react_mode(react_mode="by_order")

（2）注意Role的_act函数里面的这几行代码：

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

从环境中获取了最近的记忆，在本例中也就是上一个Action执行完的结果（生成的代码）。
执行完后，将该Action的返回结果封装成Message对象，放到环境中。

至此，单智能体的基本搭建过程就完成了。

2. 技术文档助手实践 / OSS订阅智能体实践

教程中后面还有两个复杂的单智能体实战案例，看了下，与之前MetaGPT入门课程中的教程基本没变化，这里就不再重复了。

这两部分的笔记可见：

2.1 技术文档助手相关

【AI的未来 - AI Agent系列】【MetaGPT】2. 实现自己的第一个Agent

2.2 OSS订阅智能体相关

【AI的未来 - AI Agent系列】【MetaGPT】3. 实现一个订阅智能体，订阅消息并打通微信和邮件实战：Python+ 阅读本书更多章节>>>>

本文链接：https://www.kjpai.cn/gushi/2024-04-12/157618.html，文章来源：网络cs，作者：峨乐，版权归作者所有，如需转载请注明来源和作者，否则将追究法律责任！