aspcms三合一网站源码,手机版网页,福建设备公司网站,西安有哪些做网站建设的公司异步编程学习链接 智能体 LLM观察思考行动记忆 多智能体 智能体环境SOP评审路由订阅经济
教程地址
多动作的agent的本质是react#xff0c;这包括了think#xff08;考虑接下来该采取啥动作#xff09;act#xff08;采取行动#xff09;
在MetaGPT的examples/write_…异步编程学习链接 智能体 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 MySQLrole TutorialAssistant(languageChinese)await role.run(topic)if __name__ __main__:asyncio.run(main())这个函数是调用TutorialAssistant类进行run TutorialAssistant类继承了role类run也是用role类里的 role_raise_decoratorasync def run(self, with_messageNone) - Message | None:Observe, and think and act based on the results of the observationif with_message:msg Noneif isinstance(with_message, str):msg Message(contentwith_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 rsprun函数主要的功能为
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 Messageif 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(fUnsupported 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 dynamicallyactions_taken 0rsp Message(contentNo actions taken yet, cause_byAction) # 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(historyself.rc.history,states\n.join(self.states),n_stateslen(self.states) - 1,previous_stateself.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(fInvalid answer of state, {next_state}, will be set to -1)next_state -1else:next_state int(next_state)if next_state -1:logger.info(fEnd 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(k1)[0]self.topic msg.contentresp await todo.run(topicself.topic)logger.info(resp)return await self._handle_directory(resp)resp await todo.run(topicself.topic)logger.info(resp)if self.total_content ! :self.total_content \n\n\nself.total_content respreturn Message(contentresp, roleself.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 WriteDirectorylanguage: str Chineseasync 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(topictopic, languageself.language)resp await self._aask(promptprompt)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}\nself.total_content f# {self.main_title}actions list(self.actions)for first_dir in titles.get(directory):actions.append(WriteContent(languageself.language, directoryfirst_dir))key list(first_dir.keys())[0]directory f- {key}\nfor second_dir in first_dir[key]:directory f - {second_dir}\nself.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(contentresponse.content,instruct_contentresponse.instruct_content,roleself._setting,cause_byself.rc.todo,sent_fromself,)elif isinstance(response, Message):msg responseelse:msg Message(contentresponse or , roleself.profile, cause_byself.rc.todo, sent_fromself)self.rc.memory.add(msg)return msg二、RoleReactMode.BY_ORDER
如果是按顺序的话think会依次设置动作为下一个。对于TutorialAssistant类默认为react_modeRoleReactMode.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 dont 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反应
