Spaces:
Running
Running
File size: 7,106 Bytes
6b5fe86 052b76d 6b5fe86 539b420 6b5fe86 e5f93a6 6b5fe86 98c3c3c b362bc7 6b5fe86 b362bc7 6b5fe86 b362bc7 6b5fe86 b362bc7 6b5fe86 9aee639 052b76d 6af224d 6b5fe86 46e192e 6b5fe86 9491114 6b5fe86 09665c9 6b5fe86 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
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_SKILL
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
from wolf.prompt import DESC_PROMPT, VOTE_PROMPT, KILL_PROMPT, WOLF_SPEECH_PROMPT,GAME_RULE_PROMPT,CLEAN_USER_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("主持人：你好，你分配到的角色是[狼人]")
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:
# 其他玩家发言
# 可以使用模型来过滤掉玩家的注入消息，也可以换一个小模型，实际使用需要考虑对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)
else:
# 主持人发言
self.memory.append_history('主持人: 现在进入第{}天。'.format(str(req.round)))
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: # 投票环节
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_RESULT:
self.memory.append_history(req.message)
else:
raise NotImplementedError
def interact(self, req=AgentReq) -> AgentResp:
logger.info("wolf interact: {}".format(req))
if req.status == STATUS_DISCUSS:
if req.message:
self.memory.append_history(req.message)
teammates = self.memory.load_variable("teammates")
prompt = format_prompt(DESC_PROMPT,
{"name": self.memory.load_variable("name"),
"teammates": teammates,
"history": "\n".join(self.memory.load_history())
})
logger.info("prompt:" + prompt)
result = self.llm_caller(prompt)
logger.info("wolf interact result: {}".format(result))
return AgentResp(success=True, result=result, errMsg=None)
elif req.status == STATUS_VOTE:
self.memory.append_history('主持人: 到了投票的时候了。每个人，请指向你认为可能是狼人的人。')
teammates = self.memory.load_variable("teammates")
choices = [name for name in req.message.split(",")
if name != self.memory.load_variable("name") and name not in teammates] # 排除自己和队友
self.memory.set_variable("choices", choices)
prompt = format_prompt(VOTE_PROMPT, {"name": self.memory.load_variable("name"),
"teammates": teammates,
"choices": choices,
"history": "\n".join(self.memory.load_history())
})
logger.info("prompt:" + prompt)
result = self.llm_caller(prompt)
logger.info("wolf interact result: {}".format(result))
return AgentResp(success=True, result=result, errMsg=None)
elif req.status == STATUS_WOLF_SPEECH:
teammates = self.memory.load_variable("teammates")
prompt = format_prompt(WOLF_SPEECH_PROMPT, {
"name": self.memory.load_variable("name"),
"teammates": teammates,
"history": "\n".join(self.memory.load_history())
})
logger.info("prompt:" + prompt)
result = self.llm_caller(prompt)
logger.info("wolf speech result: {}".format(result))
return AgentResp(success=True, result=result, errMsg=None)
elif req.status == STATUS_SKILL:
teammates = self.memory.load_variable("teammates")
choices = [name for name in req.message.split(",")
if name != self.memory.load_variable("name") and name not in teammates] # 排除自己和队友
self.memory.set_variable("choices", choices)
prompt = format_prompt(KILL_PROMPT, {
"name": self.memory.load_variable("name"),
"choices": choices,
"history": "\n".join(self.memory.load_history())
})
logger.info("prompt:" + prompt)
result = self.llm_caller(prompt)
logger.info("wolf kill result: {}".format(result))
return AgentResp(success=True, result=result, skillTargetPlayer=result, errMsg=None)
else:
raise NotImplementedError |