模型智能体开发之metagpt-单智能体实践

需求分析

1. 根据诉求完成函数代码的编写，并实现测试case，输出代码

代码实现

定义写代码的action

1. action是动作的逻辑抽象，通过将预设的prompt传入llm，来获取输出，并对输出进行格式化

2. 具体的实现如下

   1. 定义prompt模版

      1. prompt是传给llm的入参，所以llm对prompt的需求解析越准确，那么llm的输出就会越符合我们的诉求
      2. 如何抽象出最合适的prompt模版

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

   2. 调用llm生成代码

      1. 通过传入的instruction参数来格式化llm入参，之后通过aask调用llm进行输出。因为llm的输出是并不一定会符合我们的诉求，所以需要按照需求对output进行格式化

      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  
      

   3. 对llm output进行格式化

      1. 正则表达式提取其中的code部分，llm在返回给我们代码时通常带有一些格式化标识，而这些格式化标识往往是我们所不需要的
      2. 格式方法：

       @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
      

3. 完整代码

   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
   
   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
   

创建一个role

1. 初始化上下文

   class SimpleCoder(Role):
       name: str = "Alice"
       profile: str = "SimpleCoder"
   
       def __init__(self, **kwargs):
           super().__init__(**kwargs)
           self.set_actions([SimpleWriteCode])
   

   1. 可以看到创建了一个名为SimpleCoder的类，继承了Role，标明当前类是一个role的定位
   2. 其中name指定了当前role的名称
   3. 其中name指定了当前role的类型
   4. 然后我们重写了__init__方法，
   5. 绑定要执行的action是SimpleWriteCode，这个Action 能根据我们的需求生成我们期望的代码，定义的行动SimpleWriteCode会被加入到代办self._rc.todo中，

2. 定义执行规则

   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 
   

   1. 重写_act，编写智能体具体的行动逻辑
   2. self.rc.todo：待办事项
   3. self.get_memories(k=1)[0]：获取最新的一条memory，即本次case里面的用户下达的指令
      1. 在本次的case里面，当用户输出instruction的时候，role需要把instruction传递给action，这里就涉及到了user如何传递消息给agent的部分，是通过memory来传递的
      2. memory作为agent的记忆合集，当role在进行初始化的时候，role就会初始化一个memory对象来作为self._rc.memory属性，在之后的_observe中存储每一个message，以便后续的检索，所以也可以理解role的memory就是一个含有message的list
      3. 当需要获取memory（llm的对话context）的时候，就可以使用get_memories(self, k=0) -> list[Message] 方法
   4. todo.run(msg.content)：使用待办事项来处理最新一条memory
   5. Message：作为metagpt里面统一的消息处理格式

3. 完整代码

       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
       ```
       
   

4. 测试demo

   1. 代码

      async 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())
      

   2. 运行

      1. 如下图，role alice 关联到了action，并且action调用了llm，获取到的llm输出是一条代码。注意，代码格式有python格式化标识，所以在代码实现层面我们通过parse_code方法去掉了python的格式化标识。
      2. llm输出分为两部分，一部分是方法，另外一部分是测试case
         

demo如果想正常运行的话，需要调用llm的key，环境配置可以参照 metagpt环境配置参考