117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
|
import asyncio
|
|||
|
|
import re
|
|||
|
|
from metagpt.actions.write_tutorial import WriteDirectory, WriteContent
|
|||
|
|
from metagpt.logs import logger
|
|||
|
|
from metagpt.roles.role import Role
|
|||
|
|
from metagpt.schema import Message
|
|||
|
|
from metagpt.actions import Action
|
|||
|
|
from metagpt.prompts.tutorial_assistant import DIRECTORY_PROMPT, CONTENT_PROMPT
|
|||
|
|
from metagpt.utils.common import OutputParser
|
|||
|
|
from datetime import datetime
|
|||
|
|
from pathlib import Path
|
|||
|
|
import os
|
|||
|
|
import time
|
|||
|
|
import yaml
|
|||
|
|
|
|||
|
|
# 创建对话
|
|||
|
|
def CreateDir():
|
|||
|
|
path = 'Interlocution'
|
|||
|
|
if os.path.exists(path) == False:
|
|||
|
|
os.makedirs(path)
|
|||
|
|
files = os.listdir(path)
|
|||
|
|
file_num = str(len(files))
|
|||
|
|
# 需提前创建一个Interlocution文件夹
|
|||
|
|
path = Path('Interlocution').joinpath(file_num + '.txt')
|
|||
|
|
with open(path, 'a') as f:
|
|||
|
|
f.write(f'创建时间{datetime.fromtimestamp(int(time.time()))}\n')
|
|||
|
|
return path
|
|||
|
|
|
|||
|
|
# 记录对话
|
|||
|
|
def Recording(question : str, answer : str, path : str):
|
|||
|
|
with open(path, 'a') as f:
|
|||
|
|
question = '病人：{}\n'.format(question)
|
|||
|
|
answer = '医生：{}\n'.format(answer)
|
|||
|
|
f.write(question)
|
|||
|
|
f.write(answer)
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
class EmoLLM(Action):
|
|||
|
|
|
|||
|
|
def __init__(self, question: str, choice: str):
|
|||
|
|
super().__init__()
|
|||
|
|
with open('config.yml', 'r', encoding='utf-8') as f:
|
|||
|
|
configs = yaml.load(f.read(), Loader=yaml.FullLoader)
|
|||
|
|
self.question = question
|
|||
|
|
self.choice = choice
|
|||
|
|
self.PROMPT_TEMPLATE = configs['PROMPT'][choice]
|
|||
|
|
self.PROMPT = configs['Emoji_PROMPT']
|
|||
|
|
self.name = choice
|
|||
|
|
|
|||
|
|
async def run(self, question):
|
|||
|
|
prompt = self.PROMPT_TEMPLATE.format(question=question)
|
|||
|
|
rsp = await self._aask(prompt)
|
|||
|
|
# 将回答进行专业化处理 -- 暂未完成
|
|||
|
|
# process_rsp =
|
|||
|
|
# 将回答添加表情
|
|||
|
|
prompt = self.PROMPT.format(answer=rsp)
|
|||
|
|
process_rsp = await self._aask(prompt)
|
|||
|
|
text = EmoLLM.parse_code(process_rsp)
|
|||
|
|
return process_rsp
|
|||
|
|
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def parse_code(rsp):
|
|||
|
|
pattern = r'```处理之后的回答(.*?)```'
|
|||
|
|
match = re.search(pattern, rsp, re.DOTALL)
|
|||
|
|
text = match.group(1) if match else rsp
|
|||
|
|
return text
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 设计人设
|
|||
|
|
class ch_aiwei(Role):
|
|||
|
|
"""
|
|||
|
|
角色类，继承自Role基类
|
|||
|
|
"""
|
|||
|
|
def __init__(self, question: str, choice: str):
|
|||
|
|
"""
|
|||
|
|
初始化aiwei角色
|
|||
|
|
"""
|
|||
|
|
super().__init__() # 调用基类构造函数
|
|||
|
|
self.question = question
|
|||
|
|
self.choice = choice
|
|||
|
|
self.set_actions([EmoLLM(question=self.question, choice=self.choice)]) # 目前只有一个动作
|
|||
|
|
self._set_react_mode(react_mode='by_order') # 顺序执行
|
|||
|
|
|
|||
|
|
async def _act(self) -> Message:
|
|||
|
|
"""
|
|||
|
|
定义角色行动逻辑
|
|||
|
|
"""
|
|||
|
|
logger.info(f"{self._setting}: 准备 {self.rc.todo}") # 记录日志
|
|||
|
|
todo = self.rc.todo # 按照排列顺序获取执行的动作
|
|||
|
|
|
|||
|
|
msg = self.get_memories(k=1)[0]
|
|||
|
|
|
|||
|
|
# 回答风格化
|
|||
|
|
result = await todo.run(msg.content)
|
|||
|
|
|
|||
|
|
# 构造 Message 对象
|
|||
|
|
msg = Message(content=result, role=self.profile, cause_by=type(todo))
|
|||
|
|
self.rc.memory.add(msg) # 将运行结果添加到记忆
|
|||
|
|
|
|||
|
|
return msg # 返回最终的 Message
|
|||
|
|
|
|||
|
|
async def main():
|
|||
|
|
with open('config.yml', 'r', encoding='utf-8') as f:
|
|||
|
|
configs = yaml.load(f.read(), Loader=yaml.FullLoader)
|
|||
|
|
path = CreateDir()
|
|||
|
|
question = input('你好，请问您需要什么帮助吗？')
|
|||
|
|
role = ch_aiwei(question, '爹系男友')
|
|||
|
|
logger.info(question)
|
|||
|
|
while question != 'exit':
|
|||
|
|
result = await role.run(question)
|
|||
|
|
logger.info(result)
|
|||
|
|
Recording(question, result, path)
|
|||
|
|
question = input()
|
|||
|
|
|
|||
|
|
asyncio.run(main())
|