import sys import os current_directory = os.path.dirname(os.path.abspath(__file__)) sys.path.append(current_directory) import agentscope from agentscope.message import Msg import os import asyncio import threading import json from agent.flex_service_toolkit import * from AlgoriAgent.projects.algoriAgent.agent.algori_agent import * from agentscope.service import ( ServiceToolkit, ServiceResponse, ServiceExecStatus, ) import configparser config = configparser.ConfigParser() config.read('config.ini') openai_api_key = config['Global']['api_key'] OPENAI_CFG_DICT = { "config_name": "openai_cfg", # 此配置的名称，必须保证唯一 "model_type": "openai_chat", # 模型类型 "model_name": "gpt-4o-mini", # 模型名称 #"model_name": "gpt-4", # 模型名称 #"model_name": "llama3", "api_key": openai_api_key, # OpenAI API key. 如果没有设置，将使用环境变量中的 OPENAI_API_KEY "client_args": { "base_url": config['Global']['base_chat_url'] }, } import uuid from AlgoriAgent.projects.algoriAgent.tools.judge_tools import judge EMPTY_CHAPTER_CHAIN = [ Chapter(1, CHAPTER_FOCUS, "本章是未打开某个具体章节时的默认章节。", "处于本章节时不会有任何章节跳转。请作为一名经验丰富的算法教师，回答用户的问题。") ] def tool_name_to_tool(tool_name_list): tools = [] for tool_name in tool_name_list: if tool_name == "": continue if tool_name == "judge": tools.append(judge) return tools class AgentManager: def __init__(self): agentscope.init(model_configs=[OPENAI_CFG_DICT])#, studio_url="http://0.0.0.0:5000") self.agents = {} def new_agent(self, markdown:str, markdown_prompt:str, score_prompt:str, id = None): ''' markdown: 教案的markdown文件内容 markdown_prompt: 教案的prompt的markdown文件内容 score_prompt: 评分的prompt的markdown文件内容 根据3个教案的prompt，生成一个agent，返回agent的id与agent ''' markdown_list = markdown.split("\n") markdown_prompt_list = markdown_prompt.split("\n") score_prompt_list = score_prompt.split("\n") # 获取 H1 标题 title = "" for line in markdown_list: if line.startswith("# "): title = line[2:] break # 对于 H3 标题 构造每一个 Chapter # 首先找到所有的 H3 标题 chapter_dict = {} chapter_sequence = [] for line in markdown_list: if line.startswith("### "): chapter_name = line[4:] chapter_dict[chapter_name] = {} chapter_sequence.append(chapter_name) # 将 H3 标题 和 其对应的内容 构造成一个 Chapter h3_name = "" content = "" for i in range(len(markdown_list)): if markdown_list[i].startswith("### "): if(h3_name != ""): chapter_dict[h3_name]["markdown"] = content h3_name = markdown_list[i][4:] content = "" continue if h3_name != "": content += markdown_list[i]+"\n" if(h3_name != ""): chapter_dict[h3_name]["markdown"] = content h3_name = "" content = "" require_tools = [] for i in range(len(markdown_prompt_list)): if markdown_prompt_list[i].startswith("### "): if(h3_name != ""): chapter_dict[h3_name]["markdown_prompt"] = content chapter_dict[h3_name]["require_tools"] = tool_name_to_tool(require_tools) h3_name = markdown_prompt_list[i][4:] content = "" continue if h3_name != "": if markdown_prompt_list[i].startswith("_require_tools"): require_tools.append(markdown_prompt_list[i].split("=")[1].strip()) continue content += markdown_prompt_list[i]+"\n" if(h3_name != ""): chapter_dict[h3_name]["markdown_prompt"] = content chapter_dict[h3_name]["require_tools"] = tool_name_to_tool(require_tools) h3_name = "" content = "" for i in range(len(score_prompt_list)): if score_prompt_list[i].startswith("### "): if(h3_name != ""): chapter_dict[h3_name]["score_prompt"] = content h3_name = score_prompt_list[i][4:] content = "" continue if h3_name != "": content += score_prompt_list[i]+"\n" if(h3_name != ""): chapter_dict[h3_name]["score_prompt"] = content chapter_chain = [] No = 1 print (chapter_dict) for chapter_name in chapter_sequence: chapter_chain.append(Chapter(No, CHAPTER_LATTER, title, chapter_dict[chapter_name]["markdown"], chapter_dict[chapter_name]["markdown_prompt"], chapter_dict[chapter_name]["score_prompt"],chapter_dict[chapter_name]["require_tools"])) No+=1 chapter_chain[0].Focus() # Prepare the tools for the agent service_toolkit = FlexServiceToolkit() # for tool_function in unity_function_list: # service_toolkit.add(tool_function) agent = ChapterChainAgent( name="assistant", model_config_name="openai_cfg", verbose=True, service_toolkit=service_toolkit, max_iters=5, chapter_chain=chapter_chain ) self.agents[id] = agent return id, agent def new_agent_with_chain(self, chapter_chain = None, id = None): ''' ''' if chapter_chain is None: chapter_chain = EMPTY_CHAPTER_CHAIN # Prepare the tools for the agent service_toolkit = FlexServiceToolkit() # for tool_function in unity_function_list: # service_toolkit.add(tool_function) agent = ChapterChainAgent( name="assistant", model_config_name="openai_cfg", verbose=True, service_toolkit=service_toolkit, max_iters=5, chapter_chain=EMPTY_CHAPTER_CHAIN ) if id is None: id = uuid.uuid4() self.agents[id] = agent return id, agent def get_agent(self, id=None): ''' 如果在get agent之前没有new agent，并用id进行访问，就返回一个章节链为空的agent ''' if id: if id in self.agents: return self.agents[id] self.agents[id] = self.new_agent(id)[1] else: id = str(uuid.uuid4()) self.agents[id] = self.new_agent()[1] return id, self.agents[id] def invoke(self,id, query, user_backboard): msg = Msg("user", query, role="user") return self.agents[id](msg, user_backboard = user_backboard) import time if __name__ == '__main__': # tool_demo = ToolDemo("Lava") # response = tool_demo.invoke("I want such a Gurouce, when use it, it will be thrown and fly towards the mouse position for a short time, and stop. And will damage Enemy who step on it. I call it Lava Gurouce.") # print(response) # Start the WebSocket server in a separate thread manager = AgentManager() id, agent = manager.new_agent() # Main thread logic try: # Your code that might be interrupted while True: input_data = input("用户：") print(manager.invoke(id, input_data)) except KeyboardInterrupt: print("Process interrupted by user.") # You can add any cleanup code here if needed finally: print("Exiting program.") # Code to run before the program exits