Spaces:
Runtime error
Runtime error
| #!/usr/bin/env python | |
| # -*- coding: utf-8 -*- | |
| """ | |
| @Time : 2023/8/7 | |
| @Author : mashenquan | |
| @File : assistant.py | |
| @Desc : I am attempting to incorporate certain symbol concepts from UML into MetaGPT, enabling it to have the | |
| ability to freely construct flows through symbol concatenation. Simultaneously, I am also striving to | |
| make these symbols configurable and standardized, making the process of building flows more convenient. | |
| For more about `fork` node in activity diagrams, see: `https://www.uml-diagrams.org/activity-diagrams.html` | |
| This file defines a `fork` style meta role capable of generating arbitrary roles at runtime based on a | |
| configuration file. | |
| @Modified By: mashenquan, 2023/8/22. A definition has been provided for the return value of _think: returning false | |
| indicates that further reasoning cannot continue. | |
| """ | |
| import asyncio | |
| from pathlib import Path | |
| from metagpt.actions import ActionOutput | |
| from metagpt.actions.skill_action import ArgumentsParingAction, SkillAction | |
| from metagpt.actions.talk_action import TalkAction | |
| from metagpt.config import CONFIG | |
| from metagpt.learn.skill_loader import SkillLoader | |
| from metagpt.logs import logger | |
| from metagpt.memory.brain_memory import BrainMemory, MessageType | |
| from metagpt.roles import Role | |
| from metagpt.schema import Message | |
| class Assistant(Role): | |
| """Assistant for solving common issues.""" | |
| def __init__( | |
| self, | |
| name="Lily", | |
| profile="An assistant", | |
| goal="Help to solve problem", | |
| constraints="Talk in {language}", | |
| desc="", | |
| *args, | |
| **kwargs, | |
| ): | |
| super(Assistant, self).__init__( | |
| name=name, profile=profile, goal=goal, constraints=constraints, desc=desc, *args, **kwargs | |
| ) | |
| brain_memory = CONFIG.BRAIN_MEMORY | |
| self.memory = BrainMemory(**brain_memory) if brain_memory else BrainMemory() | |
| skill_path = Path(CONFIG.SKILL_PATH) if CONFIG.SKILL_PATH else None | |
| self.skills = SkillLoader(skill_yaml_file_name=skill_path) | |
| async def think(self) -> bool: | |
| """Everything will be done part by part.""" | |
| last_talk = await self.refine_memory() | |
| if not last_talk: | |
| return False | |
| prompt = f"Refer to this sentence:\n {last_talk}\n" | |
| skills = self.skills.get_skill_list() | |
| for desc, name in skills.items(): | |
| prompt += ( | |
| f"If want you to do {desc}, return `[SKILL]: {name}` brief and clear. For instance: [SKILL]: {name}\n" | |
| ) | |
| prompt += "If the preceding text presents a complete question and solution, rewrite and return `[SOLUTION]: {problem}` brief and clear. For instance: [SOLUTION]: Solution for distributing watermelon\n" | |
| prompt += "If the preceding text presents an unresolved issue and its corresponding discussion, rewrite and return `[PROBLEM]: {problem}` brief and clear. For instance: [PROBLEM]: How to distribute watermelon?\n" | |
| prompt += "Otherwise, rewrite and return `[TALK]: {talk}` brief and clear. For instance: [TALK]: distribute watermelon" | |
| logger.info(prompt) | |
| rsp = await self._llm.aask(prompt, []) | |
| logger.info(rsp) | |
| return await self._plan(rsp, last_talk=last_talk) | |
| async def act(self) -> ActionOutput: | |
| result = await self._rc.todo.run(**CONFIG.options) | |
| if not result: | |
| return None | |
| if isinstance(result, str): | |
| msg = Message(content=result) | |
| output = ActionOutput(content=result) | |
| else: | |
| msg = Message( | |
| content=result.content, instruct_content=result.instruct_content, cause_by=type(self._rc.todo) | |
| ) | |
| output = result | |
| self.memory.add_answer(msg) | |
| return output | |
| async def talk(self, text): | |
| self.memory.add_talk(Message(content=text)) | |
| async def _plan(self, rsp: str, **kwargs) -> bool: | |
| skill, text = Assistant.extract_info(input_string=rsp) | |
| handlers = { | |
| MessageType.Talk.value: self.talk_handler, | |
| MessageType.Problem.value: self.talk_handler, | |
| MessageType.Skill.value: self.skill_handler, | |
| } | |
| handler = handlers.get(skill, self.talk_handler) | |
| return await handler(text, **kwargs) | |
| async def talk_handler(self, text, **kwargs) -> bool: | |
| history = self.memory.history_text | |
| action = TalkAction( | |
| talk=text, knowledge=self.memory.get_knowledge(), history_summary=history, llm=self._llm, **kwargs | |
| ) | |
| self.add_to_do(action) | |
| return True | |
| async def skill_handler(self, text, **kwargs) -> bool: | |
| last_talk = kwargs.get("last_talk") | |
| skill = self.skills.get_skill(text) | |
| if not skill: | |
| logger.info(f"skill not found: {text}") | |
| return await self.talk_handler(text=last_talk, **kwargs) | |
| action = ArgumentsParingAction(skill=skill, llm=self._llm, **kwargs) | |
| await action.run(**kwargs) | |
| if action.args is None: | |
| return await self.talk_handler(text=last_talk, **kwargs) | |
| action = SkillAction(skill=skill, args=action.args, llm=self._llm, name=skill.name, desc=skill.description) | |
| self.add_to_do(action) | |
| return True | |
| async def refine_memory(self) -> str: | |
| history_text = self.memory.history_text | |
| last_talk = self.memory.last_talk | |
| if last_talk is None: # No user feedback, unsure if past conversation is finished. | |
| return None | |
| if history_text == "": | |
| return last_talk | |
| history_summary = await self._llm.get_summary(history_text, max_words=500) | |
| if last_talk and await self._llm.is_related(last_talk, history_summary): # Merge relevant content. | |
| last_talk = await self._llm.rewrite(sentence=last_talk, context=history_text) | |
| return last_talk | |
| self.memory.move_to_solution(history_summary) # Promptly clear memory after the issue is resolved. | |
| return last_talk | |
| def extract_info(input_string): | |
| from metagpt.provider.openai_api import OpenAIGPTAPI | |
| return OpenAIGPTAPI.extract_info(input_string) | |
| def get_memory(self) -> str: | |
| return self.memory.json() | |
| def load_memory(self, jsn): | |
| try: | |
| self.memory = BrainMemory(**jsn) | |
| except Exception as e: | |
| logger.exception(f"load error:{e}, data:{jsn}") | |
| async def main(): | |
| topic = "what's apple" | |
| role = Assistant(language="Chinese") | |
| await role.talk(topic) | |
| while True: | |
| has_action = await role.think() | |
| if not has_action: | |
| break | |
| msg = await role.act() | |
| logger.info(msg) | |
| # Retrieve user terminal input. | |
| logger.info("Enter prompt") | |
| talk = input("You: ") | |
| await role.talk(talk) | |
| if __name__ == "__main__": | |
| CONFIG.language = "Chinese" | |
| asyncio.run(main()) | |