参考:
- metagpt环境配置参考
- 模型智能体开发之metagpt-单智能体实践
需求分析
- 之前有过单智能体的测试case,但是现实生活场景是很复杂的,所以单智能体远远不能满足我们的诉求,所以仍然还需要了解多智能体的实现。通过多个role对动作的关联、组合来构建一个工作流程,从而使智能体能够完成更加复杂的任务
- 基于单智能体测试case的扩展,我们的诉求在简单的输出code的基础上新增一条就是生成code并且立刻运行code。那么这个时候我们就需要两个action,一个负责生成code,一个负责执行code
实现
-
定义一个负责生成code的action,参照单智能体的测试case
模型智能体开发之metagpt-单智能体实践 -
定义一个负责运行code的action
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
- 运行code不需要调用llm,所以不涉及到prompt模版的设计
- 这里通过python的标准库 subprocess来fork一个子进程,运行一个外部程序
- subprocess:包内定义了多个可以创建子进程的函数,这些函数分别以不同的方法来创建子进程,所以按需使用即可
- 在本次的case里面通过subprocess.run在fork一个子进程执行传入的代码,那么在fork之后,存在两个进程,一个是python程序本身的进程,另一个就是subprocess.run创建的子进程,两个进程是互不干预的
- 在父进程中通过result.stdout来获取子进程的执行结果
-
定义 RunnableCoder 角色
-
完整的代码
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
-
可以看到在重写init方法的时候,这里关联了两个actionSimpleWriteCode, SimpleRunCode
- 将
react_mode
设置为 “by_order”,这意味着Role
将按照self._init_actions
中指定的顺序执行其能够执行的Action
。在这种情况下,当Role
执行_act
时,self._rc.todo
将首先是SimpleWriteCode
,然后是SimpleRunCode
。
def __init__(self, **kwargs): super().__init__(**kwargs) self.set_actions([SimpleWriteCode, SimpleRunCode]) self._set_react_mode(react_mode=RoleReactMode.BY_ORDER.value)
- 将
-
重写act方法
- 覆盖
_act
函数。Role
从上一轮的人类输入或动作输出中检索消息,用适当的Message
内容提供当前的Action
(self._rc.todo
),最后返回由当前Action
输出组成的Message
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
- 覆盖
-
测试
-
代码
async def main(): msg = "write a function that calculates the sum of a list" role = RunnableCoder() logger.info(msg) result = await role.run(msg) logger.info(result) asyncio.run(main())
-
运行
-
-
demo如果想正常运行的话,需要调用llm的key,环境配置可以参照 metagpt环境配置参考