We read every piece of feedback, and take your input very seriously.
To see all available qualifiers, see our documentation.
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Applying the GLM model to multi-agent game interaction, the scenario is based on identifying the culprit.
import Agently import ENV import json # 加载 JSON 配置文件 with open('glm.json', 'r', encoding='utf-8') as f: glm = json.load(f) api_key = glm['api_key'] agent_factory = ( Agently.AgentFactory() .set_settings("current_model", "ZhipuAI") .set_settings("model.ZhipuAI.auth", { "api_key": api_key }) # glm-4 model is used by default, if you want to use glm-3-turbo, use settings down below #.set_settings("model.ZhipuAI.options", { "model": "glm-3-turbo" }) ) agent = agent_factory.create_agent() #创建一个协通作用的沙盒环境 class CooperationSandbox(object): def __init__(self): # 沙盒中管理的关键信息: # - agent共享信息 self.public_notice = { "main_quest": "", "plan": {}, "results": [], "agent_can_see": {} } # - 主控agent self.controller_agent = None # - 协作agent:名单(供主控agent分配任务用)+agent池(实际调取agent执行任务) self.worker_agent_list = [] self.worker_agent_dict = {} # - 协作agent编号器:在没有给agent命名时自动命名 self.worker_number_counter = 0 # 添加主控agent def add_controller_agent(self, agent: object): self.controller_agent = agent return self # 添加协作agent def add_worker_agent(self, agent: object, *, desc: str, name: str=None): if name is None: name = f"worker_{ str(worker_number_counter) }" worker_number_counter += 1 self.worker_agent_list.append({ "agent_name": name, "agent_desc": desc }) self.worker_agent_dict.update({ name: agent }) return self # 拆解任务 def _divide_quest(self, quest: str, current_quest: dict): plan = self.controller_agent\ .input(quest)\ .info("worker_agent_list", self.worker_agent_list)\ .instruct("""按以下步骤进行思考: 1. 根据{worker_agent_list}提供的成员信息,其中的一个执行者是否可以处理{input}提出的问题? 2. 如果可以,根据{worker_agent_list.name}请给出执行者的名字 3. 如果不可以,请根据{worker_agent_list}提供的信息,进一步拆解{input}提出的问题,使其能够被其中的一个执行者处理""" )\ .output({ "single_agent_can_handle": ( "Boolean", """{worker_agent_list}中的一个执行者可以处理{input}提出的问题吗? 如果此项为true,需要输出接下来{executor}的内容; 如果此项为false,需要输出接下来{divide_quest}的内容"""), "executor": ("String from {worker_agent_list.name} | Null", "给出执行者的{name}"), "divided_quests": [{ "quest_desc": ("String", "基于{input}拆解任务的详细描述,需要能够让一个{worker_agent_list}的执行者处理"), "target_agent": ("String from {worker_agent_list.name}", "给出拆解任务选择的执行者的{name}") }], })\ .start() if plan["single_agent_can_handle"]: self._active_handle_process(quest, plan["executor"]) else: for sub_quest in plan["divided_quests"]: self._active_handle_process(sub_quest["quest_desc"], sub_quest["target_agent"]) # 执行任务 def _active_handle_process(self, quest: str, executor_name: str): # 获取自己可见的信息 if executor_name in self.public_notice["agent_can_see"]: finished_work = self.public_notice["agent_can_see"][executor_name] else: finished_work = "暂无" # 开始执行工作 result = self.worker_agent_dict[executor_name]\ .input({ "任务": quest, "协作者名单": self.worker_agent_list, "已经完成的任务": finished_work, })\ .output({ "result": ("String", "你对{任务}的处理结果"), "to_coworkers": [("String from {协作者名单.name}", "你认为{result}应该同步给所有人,没有可以输出[我们需要更加详细的说明昨天的经过]")] })\ .start() # 提交工作结果 self.public_notice["results"].append({ "quest": quest, "result": result, }) # 共享工作结果给应该看到的协作者 if result and "to_coworkers" in result and result["to_coworkers"]: for coworker_name in result["to_coworkers"]: if coworker_name not in self.public_notice["agent_can_see"]: self.public_notice["agent_can_see"].update({ coworker_name: [] }) self.public_notice["agent_can_see"][coworker_name].append({ "quest": quest, "result": result, }) # 根据执行过程获取结果 def _get_result(self): result = self.controller_agent\ .info("执行信息", self.public_notice)\ .output({ "进行推理": ("String", "根据{执行信息.main_quest}的要求,进行案件的推理"), "事件判断": ("String", "若{执行信息.main_quest}未识别正确的凶手,则继续进行推理"), })\ .start() return result # 发布任务,启动执行 def start(self, quest: str): self.public_notice.update({ "main_quest": quest }) self._divide_quest(quest, self.public_notice["plan"]) return self._get_result() # 从字典中获取值并赋给变量 desc1 = '作家,你是平民,请协助找出凶手' desc2 = '厨师,你是平民,请协助找出凶手' desc3 = '服务员,你是狼人,请极力隐藏自己,避免暴露' # 准备agent实例 controller_agent = agent_factory.create_agent() Agent_1 = agent_factory.create_agent()\ .toggle_component("Search", True)\ .set_role("角色1", desc1) Agent_2 = agent_factory.create_agent()\ .set_role("角色2", desc2) Agent_3 = agent_factory.create_agent()\ .set_role("角色3", desc3) # 准备sandbox sandbox = CooperationSandbox() sandbox\ .add_controller_agent(controller_agent)\ .add_worker_agent( Agent_1, desc = desc1, name = "Agent_1" )\ .add_worker_agent( Agent_2, desc = desc2, name = "Agent_2" )\ .add_worker_agent( Agent_3, desc = desc3, name = "Agent_3" ) target = data['target'] # 开始测试 result = sandbox.start(target) print("完成: \n", result) print("===========\n沙盒信息: \n", sandbox.public_notice)```
The text was updated successfully, but these errors were encountered:
Thanks for sharing!
Sorry, something went wrong.
data['target']没有定义呀~
No branches or pull requests
Applying the GLM model to multi-agent game interaction, the scenario is based on identifying the culprit.
The text was updated successfully, but these errors were encountered: