需求分析
- 根据诉求完成函数代码的编写,并实现测试case,输出代码
代码实现
定义写代码的action
-
action是动作的逻辑抽象,通过将预设的prompt传入llm,来获取输出,并对输出进行格式化
-
具体的实现如下
-
定义prompt模版
- prompt是传给llm的入参,所以llm对prompt的需求解析越准确,那么llm的输出就会越符合我们的诉求
- 如何抽象出最合适的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:"""
-
调用llm生成代码
- 通过传入的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
-
对llm output进行格式化
- 正则表达式提取其中的code部分,llm在返回给我们代码时通常带有一些格式化标识,而这些格式化标识往往是我们所不需要的
- 格式方法:
@staticmethoddef parse_code(rsp):pattern = r'```python(.*)```'match = re.search(pattern, rsp, re.DOTALL)code_text = match.group(1) if match else rspreturn code_text
-
-
完整代码
import asyncio import re import subprocessimport firefrom metagpt.actions import Action from metagpt.logs import logger from metagpt.roles.role import Role, RoleReactMode from metagpt.schema import Messageclass 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@staticmethoddef parse_code(rsp):pattern = r"```python(.*)```"match = re.search(pattern, rsp, re.DOTALL)code_text = match.group(1) if match else rspreturn code_text
创建一个role
-
初始化上下文
class SimpleCoder(Role):name: str = "Alice"profile: str = "SimpleCoder"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([SimpleWriteCode])
- 可以看到创建了一个名为SimpleCoder的类,继承了Role,标明当前类是一个role的定位
- 其中name指定了当前role的名称
- 其中name指定了当前role的类型
- 然后我们重写了__init__方法,
- 绑定要执行的action是SimpleWriteCode,这个Action 能根据我们的需求生成我们期望的代码,定义的行动
SimpleWriteCode
会被加入到代办self._rc.todo
中,
-
定义执行规则
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 messagescode_text = await todo.run(msg.content)msg = Message(content=code_text, role=self.profile, cause_by=type(todo))return msg
- 重写_act,编写智能体具体的行动逻辑
- self.rc.todo:待办事项
- self.get_memories(k=1)[0]:获取最新的一条memory,即本次case里面的用户下达的指令
- 在本次的case里面,当用户输出instruction的时候,role需要把instruction传递给action,这里就涉及到了user如何传递消息给agent的部分,是通过memory来传递的
- memory作为agent的记忆合集,当role在进行初始化的时候,role就会初始化一个memory对象来作为
self._rc.memory
属性,在之后的_observe
中存储每一个message,以便后续的检索,所以也可以理解role的memory就是一个含有message的list - 当需要获取memory(llm的对话context)的时候,就可以使用get_memories(self, k=0) -> list[Message] 方法
- todo.run(msg.content):使用待办事项来处理最新一条memory
- Message:作为metagpt里面统一的消息处理格式
-
完整代码
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 messagescode_text = await todo.run(msg.content)msg = Message(content=code_text, role=self.profile, cause_by=type(todo))return msg```
-
测试demo
-
代码
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())
-
运行
- 如下图,role alice 关联到了action,并且action调用了llm,获取到的llm输出是一条代码。注意,代码格式有python格式化标识,所以在代码实现层面我们通过parse_code方法去掉了python的格式化标识。
- llm输出分为两部分,一部分是方法,另外一部分是测试case
-
demo如果想正常运行的话,需要调用llm的key,环境配置可以参照 metagpt环境配置参考