MetaGPT-Flow

单个智能体可以执行一个或多个动作，那是不是一直使用一个智能体即可呢？存在以下两个问题：

• Prompt 依赖：动作越多，要求 Prompt 越精细，才能准确控制其动作，尤其是有相似动作时
• 动作选择上限：Openai 建议一个 agent 绑定不超过 20 个动作，动作太多，agent 选择工具能力下降 Function calling - OpenAI API
• SOP (标准作业流程) 问题：在现实工作中，常常存在一个标准流程，如果每次节点都依赖 llm 去选择后续的逻辑分支，软件变得不可控

基于以上问题，提出多 agent，不同的 agent 有自己的 prompt、动作，多个 agent，降低了多 prompt 的依赖，并且按照不同工作的 SOP 限定 agent 的先后顺序

import nest_asyncio
nest_asyncio.apply()
import re
from metagpt.actions import Action
from metagpt.logs import logger

# 设置模型花费，避免出现警告
from metagpt.utils.cost_manager import TOKEN_COSTS
TOKEN_COSTS.update({'qwen2.5:latest':{'prompt': 0.0015, 'completion': 0.002}})

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
        
class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction}.
    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 = parse_code(rsp)
        return code_text
        
class SimpleWriteTest(Action):
    PROMPT_TEMPLATE: str = """
    Context: {context}
    Write {k} unit tests using pytest for the given function, assuming you have imported it.
    Return ```python your_code_here ``` with NO other texts,
    your code:
    """
    name: str = "SimpleWriteTest"
    async def run(self, context: str, k: int = 3):
        prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)
        rsp = await self._aask(prompt)
        code_text = parse_code(rsp)
        return code_text
        
class SimpleWriteReview(Action):
    PROMPT_TEMPLATE: str = """
    Context: {context}
    Review the test cases and provide one critical comments:
    """
    name: str = "SimpleWriteReview"
    async def run(self, context: str):
        prompt = self.PROMPT_TEMPLATE.format(context=context)
        rsp = await self._aask(prompt)
        return rsp

2024-12-13 10:50:43.508 | INFO | metagpt.const:get_metagpt_package_root:29 - Package root set to c:\CodeRepos\python\MyCode\Learnning_MetaGPT

from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.actions import UserRequirement
class SimpleCoder(Role):
    name: str = "Alice"
    profile: str = "SimpleCoder"
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._watch([UserRequirement])
        self.set_actions([SimpleWriteCode])
class SimpleTester(Role):
    name: str = "Bob"
    profile: str = "SimpleTester"
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteTest])
        self._watch([SimpleWriteCode])
        # self._watch([SimpleWriteCode, SimpleWriteReview])  # feel free to try this too
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo
        # context = self.get_memories(k=1)[0].content # use the most recent memory as context
        context = self.get_memories()  # use all memories as context
        code_text = await todo.run(context, k=5)  # specify arguments
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
        return msg
class SimpleReviewer(Role):
    name: str = "Charlie"
    profile: str = "SimpleReviewer"
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteReview])
        self._watch([SimpleWriteTest])    

import fire
from metagpt.logs import logger
from metagpt.team import Team
async def main(
    idea: str = "write a function that calculates the product of a list",
    investment: float = 3.0,
    n_round: int = 5,
):
    logger.info(idea)
    team = Team()
    team.hire(
        [
            SimpleCoder(),
            SimpleTester(),
            SimpleReviewer(),
            # SimpleReviewer(is_human=True),
        ]
    )
    team.invest(investment=investment)
    team.run_project(idea)
    await team.run(n_round=n_round)
if __name__ == "__main__":
    fire.Fire(main)

2024-12-13 11:18:07.884 | INFO     | metagpt.const:get_metagpt_package_root:29 - Package root set to C:\CodeRepos\python\MyCode\Learnning_MetaGPT
2024-12-13 11:18:11.354 | INFO     | __main__:main:127 - 写一个函数，能计算列表所有元素的乘积
2024-12-13 11:18:11.367 | INFO     | metagpt.team:invest:90 - Investment: $3.0.
2024-12-13 11:18:11.375 | INFO     | metagpt.roles.role:_act:391 - Alice(SimpleCoder): to do SimpleWriteCode(SimpleWriteCode)
2024-12-13 11:19:01.147 | WARNING  | metagpt. Utils. Cost_manager:update_cost: 49 - Model qwen 2.5:14b not found in TOKEN_COSTS. 
2024-12-13 11:19:01.156 | INFO     | __main__:_act: 96 - Bob (SimpleTester): to do SimpleWriteTest (SimpleWriteTest)
2024-12-13 11:19:03.656 | WARNING  | metagpt. Utils. Cost_manager:update_cost: 49 - Model qwen 2.5:14b not found in TOKEN_COSTS.
2024-12-13 11:19:03.662 | INFO     | metagpt. Roles. Role:_act: 391 - Charlie (SimpleReviewer): to do SimpleWriteReview (SimpleWriteReview)
测试用例覆盖了多种情况，包括正数、负数、包含零的列表以及空列表和单元素列表。这有助于确保函数在各种输入下的正确性。
一个关键性的评论是：测试用例应该考虑到列表中可能存在的非整数值（如浮点数或复数），以验证函数是否能够处理这些情况并返回正确的
结果。例如，可以添加如下测试案例：
这将确保函数不仅限于整数列表，还能处理更广泛的数据类型。
2024-12-13 11:19:06.484 | WARNING  | metagpt. Utils. Cost_manager:update_cost: 49 - Model qwen 2.5:14b not found in TOKEN_COSTS. 

在多 Agent 中，Meta 定义多个角色，并通过_watch 监视某个动作是否出现执行结果，这就类似构建一个流程，自身的执行依赖被监视的结果出现，和 llamaIndex 构建 Flow 时，使用的 @listen () 一个道理
在人工构建_watch 逻辑过程中，相当于构建了多 Agent 的执行流程，也就是构建了处理任务的 SOP，类似下图：用户需求 ->SimpleCodeer->SimpleTester->SimpleReviewer，但是虽然有流程，但是 llm 会根据输出，存在回环的问题，比如 SimpleReviewer->SimpleTester。