Spaces:
Sleeping
Sleeping
| # -*- coding: utf-8 -*- | |
| from agent_build_sdk.model.roles import ROLE_WOLF | |
| from agent_build_sdk.model.werewolf_model import ( | |
| AgentResp, AgentReq, STATUS_START, STATUS_WOLF_SPEECH, | |
| STATUS_VOTE_RESULT, STATUS_SKILL, STATUS_SKILL_RESULT, STATUS_NIGHT_INFO, | |
| STATUS_DAY, STATUS_DISCUSS, STATUS_VOTE, STATUS_RESULT, STATUS_NIGHT, | |
| STATUS_SHERIFF_ELECTION, STATUS_SHERIFF_SPEECH, STATUS_SHERIFF_PK, | |
| STATUS_SHERIFF_VOTE, STATUS_SHERIFF_SPEECH_ORDER, STATUS_SHERIFF, STATUS_HUNTER, STATUS_HUNTER_RESULT | |
| ) | |
| 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 | |
| # 建议使用相对导入 .prompt 确保在包内正确引用 | |
| from wolf.prompt import ( | |
| DESC_PROMPT, VOTE_PROMPT, KILL_PROMPT, WOLF_SPEECH_PROMPT, GAME_RULE_PROMPT, | |
| CLEAN_USER_PROMPT, SHERIFF_ELECTION_PROMPT, SHERIFF_SPEECH_PROMPT, | |
| SHERIFF_VOTE_PROMPT, SHERIFF_SPEECH_ORDER_PROMPT, SHERIFF_TRANSFER_PROMPT | |
| ) | |
| class WolfAgent(BasicRoleAgent): | |
| """狼人角色Agent""" | |
| def __init__(self, model_name): | |
| super().__init__(ROLE_WOLF, model_name=model_name) | |
| self.memory.set_variable("teammates", []) # 存储队友信息 | |
| def perceive(self, req: AgentReq): | |
| if req.status == STATUS_START: | |
| self.memory.clear() | |
| self.memory.set_variable("name", req.name) | |
| self.memory.set_variable("teammates", []) | |
| self.memory.append_history(GAME_RULE_PROMPT) | |
| self.memory.append_history(f"主持人：你好，你分配到的角色是[狼人]， 你是{req.name}") | |
| if req.message: | |
| teammates = req.message.split(",") | |
| self.memory.set_variable("teammates", teammates) | |
| self.memory.append_history(f"主持人：你的狼人队友是: {req.message}") | |
| elif req.status == STATUS_NIGHT: | |
| self.memory.append_history("主持人：现在进入夜晚，天黑请闭眼") | |
| elif req.status == STATUS_WOLF_SPEECH: | |
| if req.name: | |
| self.memory.append_history(f"狼人{req.name}说: {req.message}") | |
| else: | |
| self.memory.append_history("主持人：狼人请睁眼，狼人请互相确认身份，并选择要击杀的对象") | |
| elif req.status == STATUS_SKILL_RESULT: | |
| self.memory.append_history(f"主持人：狼人请今晚选择击杀的目标是:{req.name}") | |
| elif req.status == STATUS_NIGHT_INFO: | |
| self.memory.append_history(f"主持人：天亮了！昨天晚上的信息是: {req.message}") | |
| elif req.status == STATUS_DISCUSS: | |
| if req.name: | |
| self.memory.append_history(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: | |
| if req.name: | |
| self.memory.append_history(f'主持人: 投票结果是：{req.name}。') | |
| else: | |
| self.memory.append_history('主持人: 无人出局。') | |
| elif req.status == STATUS_SHERIFF_ELECTION: | |
| self.memory.append_history(f"主持人: 上警玩家: {req.message}") | |
| elif req.status == STATUS_SHERIFF_SPEECH: | |
| self.memory.append_history(f"{req.name} (警上发言): {req.message}") | |
| elif req.status == STATUS_SHERIFF_VOTE: | |
| self.memory.append_history(f"警上投票: {req.name}投了{req.message}") | |
| elif req.status == STATUS_SHERIFF: | |
| if req.name: | |
| self.memory.append_history(f"主持人: 警徽归属: {req.name}") | |
| self.memory.set_variable("sheriff", req.name) | |
| if req.message: | |
| 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}") | |
| elif req.status == STATUS_RESULT: | |
| self.memory.append_history(req.message) | |
| def interact(self, req: AgentReq) -> AgentResp: | |
| logger.info("wolf interact: {}".format(req)) | |
| try: | |
| name = self.memory.load_variable("name") | |
| history = "\n".join(self.memory.load_history()) | |
| teammates = self.memory.load_variable("teammates") | |
| if req.status == STATUS_DISCUSS: | |
| prompt = format_prompt(DESC_PROMPT, {"name": name, "teammates": teammates, "history": history}) | |
| result = self.llm_caller(prompt) | |
| return AgentResp(success=True, result=result) | |
| elif req.status == STATUS_VOTE: | |
| choices = [n for n in req.message.split(",") if n != name and n not in teammates] | |
| prompt = format_prompt(VOTE_PROMPT, {"name": name, "teammates": teammates, "choices": choices, "history": history}) | |
| result = self.llm_caller(prompt) | |
| return AgentResp(success=True, result=result) | |
| elif req.status == STATUS_WOLF_SPEECH: | |
| prompt = format_prompt(WOLF_SPEECH_PROMPT, {"name": name, "teammates": teammates, "history": history}) | |
| result = self.llm_caller(prompt) | |
| return AgentResp(success=True, result=result) | |
| elif req.status == STATUS_SKILL: | |
| choices = [n for n in req.message.split(",") if n != name and n not in teammates] | |
| prompt = format_prompt(KILL_PROMPT, {"name": name, "choices": choices, "history": history}) | |
| result = self.llm_caller(prompt) | |
| return AgentResp(success=True, result=result, skillTargetPlayer=result) | |
| elif req.status == STATUS_SHERIFF_ELECTION: | |
| prompt = format_prompt(SHERIFF_ELECTION_PROMPT, {"name": name, "teammates": teammates, "history": history}) | |
| result = self.llm_caller(prompt) | |
| return AgentResp(success=True, result=result) | |
| elif req.status in [STATUS_SHERIFF_SPEECH, STATUS_SHERIFF_PK]: | |
| prompt = format_prompt(SHERIFF_SPEECH_PROMPT, {"name": name, "teammates": teammates, "history": history}) | |
| result = self.llm_caller(prompt) | |
| return AgentResp(success=True, result=result) | |
| elif req.status == STATUS_SHERIFF_VOTE: | |
| choices = req.message.split(",") | |
| prompt = format_prompt(SHERIFF_VOTE_PROMPT, {"name": name, "teammates": teammates, "choices": choices, "history": history}) | |
| result = self.llm_caller(prompt) | |
| return AgentResp(success=True, result=result) | |
| elif req.status == STATUS_SHERIFF_SPEECH_ORDER: | |
| prompt = format_prompt(SHERIFF_SPEECH_ORDER_PROMPT, {"name": name, "history": history}) | |
| result = self.llm_caller(prompt) | |
| return AgentResp(success=True, result=result) | |
| elif req.status == STATUS_SHERIFF: | |
| choices = [n for n in req.message.split(",") if n != name] | |
| prompt = format_prompt(SHERIFF_TRANSFER_PROMPT, {"name": name, "teammates": teammates, "choices": choices, "history": history}) | |
| result = self.llm_caller(prompt) | |
| return AgentResp(success=True, result=result) | |
| return AgentResp(success=True, result=None) | |
| except Exception as e: | |
| logger.error("WolfAgent interact failed", exc_info=True) | |
| return AgentResp(success=False, errMsg=str(e)) | |