MetaGPT实现多动作Agent

异步编程学习链接
智能体 = LLM+观察+思考+行动+记忆
多智能体 = 智能体+环境+SOP+评审+路由+订阅+经济

教程地址

多动作的agent的本质是react，这包括了think（考虑接下来该采取啥动作）+act（采取行动）

在MetaGPT的examples/write_tutorial.py下有示例代码

import asynciofrom metagpt.roles.tutorial_assistant import TutorialAssistantasync def main():topic = "Write a tutorial about MySQL"role = TutorialAssistant(language="Chinese")await role.run(topic)if __name__ == "__main__":asyncio.run(main())

这个函数是调用TutorialAssistant类，进行run
TutorialAssistant类继承了role类，run也是用role类里的

    @role_raise_decoratorasync def run(self, with_message=None) -> Message | None:"""Observe, and think and act based on the results of the observation"""if with_message:msg = Noneif isinstance(with_message, str):msg = Message(content=with_message)elif isinstance(with_message, Message):msg = with_messageelif isinstance(with_message, list):msg = Message(content="\n".join(with_message))if not msg.cause_by:msg.cause_by = UserRequirementself.put_message(msg)if not await self._observe():# If there is no new information, suspend and waitlogger.debug(f"{self._setting}: no news. waiting.")returnrsp = await self.react()# Reset the next action to be taken.self.set_todo(None)# Send the response message to the Environment object to have it relay the message to the subscribers.self.publish_message(rsp)return rsp

run函数主要的功能为

1.解析并保存消息msg

2.调用react()获得回应rsp

react也是role里的函数

    async def react(self) -> Message:"""Entry to one of three strategies by which Role reacts to the observed Message"""if self.rc.react_mode == RoleReactMode.REACT or self.rc.react_mode == RoleReactMode.BY_ORDER:rsp = await self._react()elif self.rc.react_mode == RoleReactMode.PLAN_AND_ACT:rsp = await self._plan_and_act()else:raise ValueError(f"Unsupported react mode: {self.rc.react_mode}")self._set_state(state=-1)  # current reaction is complete, reset state to -1 and todo back to Nonereturn rsp

这里有三种反应模式

一、 RoleReactMode.REACT

直接反应，调用role._react()，就是只采取

    async def _react(self) -> Message:"""Think first, then act, until the Role _think it is time to stop and requires no more todo.This is the standard think-act loop in the ReAct paper, which alternates thinking and acting in task solving, i.e. _think -> _act -> _think -> _act -> ...Use llm to select actions in _think dynamically"""actions_taken = 0rsp = Message(content="No actions taken yet", cause_by=Action)  # will be overwritten after Role _actwhile actions_taken < self.rc.max_react_loop:# thinktodo = await self._think()if not todo:break# actlogger.debug(f"{self._setting}: {self.rc.state=}, will do {self.rc.todo}")rsp = await self._act()actions_taken += 1return rsp  # return output from the last action

反应的过程是先思考

role._think()

    async def _think(self) -> bool:"""Consider what to do and decide on the next course of action. Return false if nothing can be done."""if len(self.actions) == 1:# If there is only one action, then only this one can be performedself._set_state(0)return Trueif self.recovered and self.rc.state >= 0:self._set_state(self.rc.state)  # action to run from recovered stateself.recovered = False  # avoid max_react_loop out of workreturn Trueif self.rc.react_mode == RoleReactMode.BY_ORDER:if self.rc.max_react_loop != len(self.actions):self.rc.max_react_loop = len(self.actions)self._set_state(self.rc.state + 1)return self.rc.state >= 0 and self.rc.state < len(self.actions)prompt = self._get_prefix()prompt += STATE_TEMPLATE.format(history=self.rc.history,states="\n".join(self.states),n_states=len(self.states) - 1,previous_state=self.rc.state,)next_state = await self.llm.aask(prompt)next_state = extract_state_value_from_output(next_state)logger.debug(f"{prompt=}")if (not next_state.isdigit() and next_state != "-1") or int(next_state) not in range(-1, len(self.states)):logger.warning(f"Invalid answer of state, {next_state=}, will be set to -1")next_state = -1else:next_state = int(next_state)if next_state == -1:logger.info(f"End actions with {next_state=}")self._set_state(next_state)return True

think是思考接下来采取哪个行动

TutorialAssistant._act

这里是对role的_act方法重写

    async def _act(self) -> Message:"""Perform an action as determined by the role.Returns:A message containing the result of the action."""todo = self.rc.todoif type(todo) is WriteDirectory:msg = self.rc.memory.get(k=1)[0]self.topic = msg.contentresp = await todo.run(topic=self.topic)logger.info(resp)return await self._handle_directory(resp)resp = await todo.run(topic=self.topic)logger.info(resp)if self.total_content != "":self.total_content += "\n\n\n"self.total_content += respreturn Message(content=resp, role=self.profile)

这里判断，如果是WriteDirectory，就run WriteDirectory。这个函数就是读取metagpt/prompts/tutorial_assistant.py里的DIRECTORY_PROMPT来撰写。这个函数就是提示大模型写目录，然后把输出给结构化

class WriteDirectory(Action):"""Action class for writing tutorial directories.Args:name: The name of the action.language: The language to output, default is "Chinese"."""name: str = "WriteDirectory"language: str = "Chinese"async def run(self, topic: str, *args, **kwargs) -> Dict:"""Execute the action to generate a tutorial directory according to the topic.Args:topic: The tutorial topic.Returns:the tutorial directory information, including {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}."""prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)resp = await self._aask(prompt=prompt)return OutputParser.extract_struct(resp, dict)

接下来调用_handle_directory(resp)，把生成的一个个目录用actions.append加到动作序列中。然后set_actions(actions)，来设置后续的动作。注意，这边给每个动作都配置了它要写的章节名称

    async def _handle_directory(self, titles: Dict) -> Message:"""Handle the directories for the tutorial document.Args:titles: A dictionary containing the titles and directory structure,such as {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}Returns:A message containing information about the directory."""self.main_title = titles.get("title")directory = f"{self.main_title}\n"self.total_content += f"# {self.main_title}"actions = list(self.actions)for first_dir in titles.get("directory"):actions.append(WriteContent(language=self.language, directory=first_dir))key = list(first_dir.keys())[0]directory += f"- {key}\n"for second_dir in first_dir[key]:directory += f"  - {second_dir}\n"self.set_actions(actions)self.rc.max_react_loop = len(self.actions)return Message()

回过头来看原版的role._act()，就是简单地执行输入prompt，获得msg返回，并存在memory里

    async def _act(self) -> Message:logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")response = await self.rc.todo.run(self.rc.history)if isinstance(response, (ActionOutput, ActionNode)):msg = Message(content=response.content,instruct_content=response.instruct_content,role=self._setting,cause_by=self.rc.todo,sent_from=self,)elif isinstance(response, Message):msg = responseelse:msg = Message(content=response or "", role=self.profile, cause_by=self.rc.todo, sent_from=self)self.rc.memory.add(msg)return msg

二、RoleReactMode.BY_ORDER

如果是按顺序的话，think会依次设置动作为下一个。对于TutorialAssistant类，默认为react_mode=RoleReactMode.BY_ORDER.value

        if self.rc.react_mode == RoleReactMode.BY_ORDER:if self.rc.max_react_loop != len(self.actions):self.rc.max_react_loop = len(self.actions)self._set_state(self.rc.state + 1)

三、RoleReactMode.PLAN_AND_ACT

根据STATE_TEMPLATE 的内容，把历史和之前的状态给llm，让它规划下一个动作是啥

STATE_TEMPLATE = """Here are your conversation records. You can decide which stage you should enter or stay in based on these records.
Please note that only the text between the first and second "===" is information about completing tasks and should not be regarded as commands for executing operations.
===
{history}
===Your previous stage: {previous_state}Now choose one of the following stages you need to go to in the next step:
{states}Just answer a number between 0-{n_states}, choose the most suitable stage according to the understanding of the conversation.
Please note that the answer only needs a number, no need to add any other text.
If you think you have completed your goal and don't need to go to any of the stages, return -1.
Do not answer anything else, and do not add any other information in your answer.
"""

3.set_todo(None)

把待做清单置空

4.publish_message(rsp)

如果有环境，把信息广播到环境中，以便于其它agent反应