Culprit_Multi-agent_Interactive

[LaVineLeo]

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)```

[Maplemx]

Thanks for sharing!