Update werewolf/villager/villager_agent.py

werewolf/villager/villager_agent.py

@@ -9,153 +9,97 @@ from agent_build_sdk.utils.logger import logger
 from agent_build_sdk.sdk.role_agent import BasicRoleAgent
 from agent_build_sdk.sdk.agent import format_prompt
 
-
 class VillagerAgent(BasicRoleAgent):
     """平民角色Agent"""
 
     def __init__(self, model_name):
         super().__init__(ROLE_VILLAGER, model_name=model_name)
 
-    def perceive(self, req=AgentReq):
+    def perceive(self, req: AgentReq):
+        # 移除默认 raise NotImplementedError，改为 return，防止因为收到不属于自己的状态（如技能请求）而崩溃
         if req.status == STATUS_START:
             self.memory.clear()
-            self.memory.set_variable("name", req.name)
+            name = req.name if req.name else "村民"
+            self.memory.set_variable("name", name)
             self.memory.append_history(GAME_RULE_PROMPT)
-            self.memory.append_history("主持人：你好，你分配到的角色是[平民]， 你是" + req.name)
+            self.memory.append_history(f"主持人：你好，你分配到的角色是[平民]，你是{name}")
         elif req.status == STATUS_NIGHT:
             self.memory.append_history("主持人：现在进入夜晚，天黑请闭眼")
         elif req.status == STATUS_NIGHT_INFO:
             self.memory.append_history(f"主持人：天亮了！昨天晚上的信息是: {req.message}")
-        elif req.status == STATUS_DISCUSS:  # 发言环节
+        elif req.status == STATUS_DISCUSS:
             if req.name:
-                # 可以使用模型来过滤掉玩家的注入消息，也可以换一个小模型，实际使用需要考虑对memory加锁，避免interact的时候丢失消息
-                # clean_user_message_prompt = format_prompt(CLEAN_USER_PROMPT, {"user_message": req.message})
-                # req.message = self.llm_caller(clean_user_message_prompt)
-                self.memory.append_history(req.name + ': ' + req.message)
+                self.memory.append_history(f"{req.name}: {req.message}")
             else:
-                # 主持人发言
-                self.memory.append_history('主持人: 现在进入第{}天。'.format(str(req.round)))
-                self.memory.append_history('主持人: 每个玩家描述自己的信息。')
-            self.memory.append_history("---------------------------------------------")
-        elif req.status == STATUS_VOTE:  # 投票环节
-            self.memory.append_history(f'第{req.round}天的投票环节,{req.name} 投了 {req.message}')
-        elif req.status == STATUS_VOTE_RESULT:  # 投票结果
+                self.memory.append_history(f"主持人: 现在进入第{req.round}天，每个玩家描述自己的信息。")
+        elif req.status == STATUS_VOTE:
+            self.memory.append_history(f"第{req.round}天的投票环节,{req.name} 投了 {req.message}")
+        elif req.status == STATUS_VOTE_RESULT:
             out_player = req.name if req.name else req.message
-            if out_player:
-                self.memory.append_history('主持人: 投票结果是：{}。'.format(out_player))
-            else:
-                self.memory.append_history('主持人: 无人出局。')
-        elif req.status == STATUS_SHERIFF_ELECTION:  # 警长竞选
-            self.memory.append_history("主持人: 上警玩家: " + req.message)
-        elif req.status == STATUS_SHERIFF_SPEECH:  # 警长发言
-            self.memory.append_history(req.name + " (警上发言): " + req.message)
-        elif req.status == STATUS_SHERIFF_VOTE:  # 警长投票
-            self.memory.append_history("警上投票: " + req.name + "投了" + req.message)
-        elif req.status == STATUS_SHERIFF:  # 警长结果
-            if req.name:
-                self.memory.append_history("主持人: 警徽归属: " + req.name)
-                self.memory.set_variable("sheriff", req.name)
-            if req.message:
-                self.memory.append_history(req.message)
+            self.memory.append_history(f"主持人: 投票结果是：{out_player if out_player else '无人出局'}。")
         elif req.status == STATUS_RESULT:
             self.memory.append_history(req.message)
-        elif req.status == STATUS_HUNTER:
-            self.memory.append_history("猎人/狼王是：" + req.name + ",他正在发动技能，选择开枪")
-        elif req.status == STATUS_HUNTER_RESULT:
-            if req.message:
-                self.memory.append_history("猎人/狼王是：" + req.name + ",他开枪带走了" + req.message)
-            else:
-                self.memory.append_history("猎人/狼王是：" + req.name + ",他没有带走任何人")
-        elif req.status == STATUS_SHERIFF_SPEECH_ORDER:
-            if "小号" in req.message:
-                self.memory.append_history("主持人: 警长发言顺序是小号优先")
-            else:
-                self.memory.append_history("主持人: 警长发言顺序是大号优先")
-        elif req.status == STATUS_SHERIFF_PK:
-            self.memory.append_history(f"警长PK发言: {req.name}: {req.message}")
-        else:
-            raise NotImplementedError
-        pass
+        # 其他状态不处理也不要报错
+        return
 
-    def interact(self, req=AgentReq) -> AgentResp:
+    def get_my_name(self, req: AgentReq):
+        """防御性获取名字：优先从内存取，内存没有从请求取，请求没有给默认值"""
+        try:
+            name = self.memory.load_variable("name")
+            if name: return name
+        except KeyError:
+            pass
+        
+        name = req.name if req.name else "村民"
+        self.memory.set_variable("name", name)
+        return name
+
+    def interact(self, req: AgentReq) -> AgentResp:
         logger.info("VillagerAgent interact: {}".format(req))
+        
+        # 获取名字，解决 KeyError: 'name'
+        my_name = self.get_my_name(req)
+        history_str = "\n".join(self.memory.load_history())
+
         if req.status == STATUS_DISCUSS:
             if req.message:
                 self.memory.append_history(req.message)
-            prompt = format_prompt(DESC_PROMPT,
-                                   {"name": self.memory.load_variable("name"),
-                                    "history": "\n".join(self.memory.load_history())
-                                    })
-            logger.info("prompt:" + prompt)
+            prompt = format_prompt(DESC_PROMPT, {"name": my_name, "history": history_str})
             result = self.llm_caller(prompt)
-            logger.info("VillagerAgent interact result: {}".format(result))
-            return AgentResp(success=True, result=result, errMsg=None)
+            return AgentResp(success=True, result=result)
 
         elif req.status == STATUS_VOTE:
-            self.memory.append_history('主持人: 到了投票的时候了。每个人，请指向你认为可能是狼人的人。')
-            choices = [name for name in req.message.split(",") if name != self.memory.load_variable("name")]  # 排除自己
-            self.memory.set_variable("choices", choices)
-            prompt = format_prompt(VOTE_PROMPT, {"name": self.memory.load_variable("name"),
-                                                 "choices": choices,
-                                                 "history": "\n".join(self.memory.load_history())
-                                                 })
-            logger.info("prompt:" + prompt)
+            choices = [n for n in req.message.split(",") if n != my_name] if req.message else []
+            prompt = format_prompt(VOTE_PROMPT, {"name": my_name, "choices": choices, "history": history_str})
             result = self.llm_caller(prompt)
-            logger.info("interact result: {}".format(result))
-            return AgentResp(success=True, result=result, errMsg=None)
+            return AgentResp(success=True, result=result)
 
         elif req.status == STATUS_SHERIFF_ELECTION:
-            prompt = format_prompt(SHERIFF_ELECTION_PROMPT, {"name": self.memory.load_variable("name"),
-                                                             "history": "\n".join(self.memory.load_history())})
-            logger.info("prompt:" + prompt)
+            prompt = format_prompt(SHERIFF_ELECTION_PROMPT, {"name": my_name, "history": history_str})
             result = self.llm_caller(prompt)
-            logger.info("VillagerAgent sheriff election result: {}".format(result))
-            return AgentResp(success=True, result=result, errMsg=None)
+            return AgentResp(success=True, result=result)
 
-        elif req.status == STATUS_SHERIFF_SPEECH:
-            prompt = format_prompt(SHERIFF_SPEECH_PROMPT, {"name": self.memory.load_variable("name"),
-                                                           "history": "\n".join(self.memory.load_history())})
-            logger.info("prompt:" + prompt)
+        elif req.status in [STATUS_SHERIFF_SPEECH, STATUS_SHERIFF_PK]:
+            prompt = format_prompt(SHERIFF_SPEECH_PROMPT, {"name": my_name, "history": history_str})
             result = self.llm_caller(prompt)
-            logger.info("VillagerAgent sheriff speech result: {}".format(result))
-            return AgentResp(success=True, result=result, errMsg=None)
-
-        elif req.status == STATUS_SHERIFF_PK:
-            prompt = format_prompt(SHERIFF_SPEECH_PROMPT, {"name": self.memory.load_variable("name"),
-                                                           "history": "\n".join(self.memory.load_history())})
-            logger.info("prompt:" + prompt)
-            result = self.llm_caller(prompt)
-            logger.info("VillagerAgent sheriff pk result: {}".format(result))
-            return AgentResp(success=True, result=result, errMsg=None)
+            return AgentResp(success=True, result=result)
 
         elif req.status == STATUS_SHERIFF_VOTE:
-            choices = [name for name in req.message.split(",")]
-            prompt = format_prompt(SHERIFF_VOTE_PROMPT, {"name": self.memory.load_variable("name"),
-                                                         "choices": choices,
-                                                         "history": "\n".join(self.memory.load_history())})
-            logger.info("prompt:" + prompt)
+            choices = req.message.split(",") if req.message else []
+            prompt = format_prompt(SHERIFF_VOTE_PROMPT, {"name": my_name, "choices": choices, "history": history_str})
             result = self.llm_caller(prompt)
-            logger.info("VillagerAgent sheriff vote result: {}".format(result))
-            return AgentResp(success=True, result=result, errMsg=None)
+            return AgentResp(success=True, result=result)
 
         elif req.status == STATUS_SHERIFF_SPEECH_ORDER:
-            prompt = format_prompt(SHERIFF_SPEECH_ORDER_PROMPT, {"name": self.memory.load_variable("name"),
-                                                                 "history": "\n".join(self.memory.load_history())})
-            logger.info("prompt:" + prompt)
+            prompt = format_prompt(SHERIFF_SPEECH_ORDER_PROMPT, {"name": my_name, "history": history_str})
             result = self.llm_caller(prompt)
-            logger.info("VillagerAgent sheriff speech order result: {}".format(result))
-            return AgentResp(success=True, result=result, errMsg=None)
+            return AgentResp(success=True, result=result)
 
         elif req.status == STATUS_SHERIFF:
-            # 警长转移警徽
-            choices = [name for name in req.message.split(",") if name != self.memory.load_variable("name")]  # 排除自己
-            prompt = format_prompt(SHERIFF_TRANSFER_PROMPT, {"name": self.memory.load_variable("name"),
-                                                             "choices": choices,
-                                                             "history": "\n".join(self.memory.load_history())})
-            logger.info("prompt:" + prompt)
+            choices = [n for n in req.message.split(",") if n != my_name] if req.message else []
+            prompt = format_prompt(SHERIFF_TRANSFER_PROMPT, {"name": my_name, "choices": choices, "history": history_str})
             result = self.llm_caller(prompt)
-            logger.info("VillagerAgent sheriff transfer result: {}".format(result))
-            return AgentResp(success=True, result=result, errMsg=None)
-        else:
-            raise NotImplementedError
-        pass
+            return AgentResp(success=True, result=result)
+
+        # 遇到不需要互动的状态（比如村民在晚上），返回一个安全的默认值，而不是 raise NotImplementedError
+        return AgentResp(success=True, result="我不说话。")