Spaces:
Sleeping
Sleeping
File size: 13,157 Bytes
8eadaad | 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 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 | 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
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:
# 其他玩家发言
# 可以使用模型来过滤掉玩家的注入消息，也可以换一个小模型，实际使用需要考虑对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('主持人: 每个玩家描述自己的信息。')
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)
else:
raise NotImplementedError
def interact(self, req=AgentReq) -> AgentResp:
logger.info("wolf interact: {}".format(req))
try:
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)
elif req.status == STATUS_SHERIFF_ELECTION:
teammates = self.memory.load_variable("teammates")
prompt = format_prompt(SHERIFF_ELECTION_PROMPT, {
"name": self.memory.load_variable("name"),
"teammates": teammates,
"history": "\n".join(self.memory.load_history())
})
logger.info("wolf agent sheriff election prompt:" + prompt)
result = self.llm_caller(prompt)
return AgentResp(success=True, result=result, errMsg=None)
elif req.status == STATUS_SHERIFF_SPEECH:
teammates = self.memory.load_variable("teammates")
prompt = format_prompt(SHERIFF_SPEECH_PROMPT, {
"name": self.memory.load_variable("name"),
"teammates": teammates,
"history": "\n".join(self.memory.load_history())
})
logger.info("wolf agent sheriff speech prompt:" + prompt)
result = self.llm_caller(prompt)
return AgentResp(success=True, result=result, errMsg=None)
elif req.status == STATUS_SHERIFF_PK:
teammates = self.memory.load_variable("teammates")
prompt = format_prompt(SHERIFF_SPEECH_PROMPT, {
"name": self.memory.load_variable("name"),
"teammates": teammates,
"history": "\n".join(self.memory.load_history())
})
logger.info("wolf agent sheriff pk prompt:" + prompt)
result = self.llm_caller(prompt)
return AgentResp(success=True, result=result, errMsg=None)
elif req.status == STATUS_SHERIFF_VOTE:
teammates = self.memory.load_variable("teammates")
choices = [name for name in req.message.split(",")]
prompt = format_prompt(SHERIFF_VOTE_PROMPT, {
"name": self.memory.load_variable("name"),
"teammates": teammates,
"choices": choices,
"history": "\n".join(self.memory.load_history())
})
logger.info("wolf agent sheriff vote prompt:" + prompt)
result = self.llm_caller(prompt)
return AgentResp(success=True, result=result, errMsg=None)
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("wolf agent sheriff speech order prompt:" + prompt)
result = self.llm_caller(prompt)
return AgentResp(success=True, result=result, errMsg=None)
elif req.status == STATUS_SHERIFF:
# 警长转移警徽
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]
prompt = format_prompt(SHERIFF_TRANSFER_PROMPT, {
"name": self.memory.load_variable("name"),
"teammates": teammates,
"choices": choices,
"history": "\n".join(self.memory.load_history())
})
logger.info("wolf agent sheriff transfer prompt:" + prompt)
result = self.llm_caller(prompt)
return AgentResp(success=True, result=result, errMsg=None)
return AgentResp(success=True, result=None, errMsg=None)
except Exception as e:
logger.error("WolfAgent interact failed", exc_info=True)
return AgentResp(success=False, result=None, errMsg=str(e)) |