File size: 6,968 Bytes
360d784 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 |
#!/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
@staticmethod
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())
|