前面学习了AutoGen今天来看另一款多智能体框架——AgentScope。如果说AutoGen 更强调对话驱动协作那么 AgentScope 更强调工程化能力它不仅支持多智能体协作还提供了消息驱动、异步执行、结构化输出等能力更适合构建生产级 Agent 应用。一、什么是 AgentScopeAgentScope 是阿里达摩院开源的一款多智能体开发框架。相比 AutoGen它最大的特点就是一切围绕消息Message展开。Agent 之间不是互相调用函数而是不断发送和接收消息。整个流程可以理解成Agent A │ Message │ ▼ MsgHub │ Message │ ▼ Agent B因此 AgentScope 也被称为消息驱动Message Driven多智能体框架。二、AgentScope 的核心设计1、消息驱动Message Driven这是 AgentScope 最重要的设计。官方定义消息from agentscope.message import Msg message Msg( nameAlice, contentHello, roleuser )每条消息都包含name发送者content消息内容role角色所有 Agent 都围绕 Msg 进行通信。相比直接调用函数这种方式有几个优点Agent 之间更加解耦支持异步执行更容易记录日志更方便扩展到分布式环境所以 AgentScope 更适合大型项目。2、Agent 生命周期AgentScope 中所有 Agent 都继承自AgentBase。例如class CustomAgent(AgentBase): def reply(self, x: Msg): response self.model(x.content) return Msg( nameself.name, contentresponse, roleassistant )开发者主要关注reply()收到消息↓思考↓返回消息整个 Agent 就完成了一次交互。3、MsgHub如果说 Msg 是消息那么 MsgHub 就像消息中转站。所有 Agent 的通信都会经过 MsgHub。例如Agent1 │ Agent2 ── MsgHub ── Agent3 │ Agent4它负责消息广播消息路由消息同步也是 AgentScope 最核心的组件之一。三、案例三国狼人杀官方给出的案例非常有意思。使用 AgentScope 实现一个三国狼人杀。不仅要完成狼人杀规则还要求 AI 扮演刘备曹操孙权张飞周瑜等不同角色。所以每个 Agent 都拥有两层身份游戏身份狼人女巫村民预言家人物身份曹操刘备孙权……因此每个人说话风格都会不同。四、狼人讨论为什么适合 AgentScope狼人阶段正好体现了消息驱动。官方代码async with MsgHub( self.werewolves, enable_auto_broadcastTrue, ):这一步会创建狼人聊天室。只有狼人才能收到消息。之后狼人不断讨论狼人A ↓ 狼人B ↓ 狼人A ↓ 狼人B最后统一投票。整个过程完全由 MsgHub 管理。不用自己维护各种状态。五、结构化输出狼人杀还有一个问题AI 万一乱回答怎么办例如女巫回答今晚我觉得天气不错。游戏就没法继续了。所以 AgentScope 使用了Pydantic。例如class WitchActionModelCN(BaseModel): use_antidote: bool use_poison: bool target_name: str这样模型只能返回{ use_antidote: true, use_poison: false, target_name:曹操 }游戏规则也更容易控制。六、并发处理狼人杀里很多操作需要所有玩家同时行动。例如投票。官方使用fanout_pipeline(...)它会主持人 ↓ 所有玩家 ↓ 同时回复 ↓ 统一收集结果整个过程都是异步执行。效率比一个一个等待回复更高。七、容错处理如果某个 Agent 出现异常怎么办官方也加入了异常处理。try: response await wolf(...) except Exception: default_response ...即使一个 Agent 出错游戏仍然能够继续运行。这也是 AgentScope 偏工程化的重要体现。八、AgentScope 的优缺点优点消息驱动架构Agent 解耦程度高。原生支持异步并发更适合大型项目。结构化输出方便约束 AI 行为。工程化能力强适合生产环境。不足学习成本高需要理解异步编程和消息机制。对简单项目来说有些重量级。社区生态相比 LangGraph、AutoGen 仍在发展中。九、完整代码game_roles.py# -*- coding: utf-8 -*- 三国狼人杀角色定义 from typing import Dict, List class GameRoles: 游戏角色管理类 ROLES { 狼人: { description: 狼人, ability: 夜晚可以击杀一名玩家, win_condition: 消灭所有好人或与好人数量相等, team: 狼人阵营 }, 预言家: { description: 预言家, ability: 每晚可以查验一名玩家的身份, win_condition: 消灭所有狼人, team: 好人阵营 }, 女巫: { description: 女巫, ability: 拥有解药和毒药各一瓶可以救人或杀人, win_condition: 消灭所有狼人, team: 好人阵营 }, 猎人: { description: 猎人, ability: 被投票出局时可以开枪带走一名玩家, win_condition: 消灭所有狼人, team: 好人阵营 }, 村民: { description: 村民, ability: 无特殊技能依靠推理和投票, win_condition: 消灭所有狼人, team: 好人阵营 }, 守护者: { description: 守护者, ability: 每晚可以守护一名玩家免受狼人攻击, win_condition: 消灭所有狼人, team: 好人阵营 } } CHARACTER_TRAITS { 刘备: 仁德宽厚善于团结众人说话温和有礼, 关羽: 忠义刚烈言辞直接重情重义, 张飞: 性格豪爽说话大声直接容易冲动, 诸葛亮: 智慧超群分析透彻言辞谨慎, 赵云: 忠勇双全话语简洁有力, 曹操: 雄才大略善于权谋话语犀利, 司马懿: 深谋远虑城府极深言辞含蓄, 周瑜: 才华横溢略显傲气分析精准, 孙权: 年轻有为善于决断话语果决 } classmethod def get_role_desc(cls, role: str) - str: 获取角色描述 return cls.ROLES.get(role, {}).get(description, 未知角色) classmethod def get_role_ability(cls, role: str) - str: 获取角色技能 return cls.ROLES.get(role, {}).get(ability, 无特殊技能) classmethod def get_character_trait(cls, character: str) - str: 获取角色性格特点 return cls.CHARACTER_TRAITS.get(character, 性格温和说话得体) classmethod def is_werewolf(cls, role: str) - bool: 判断是否为狼人 return role 狼人 classmethod def is_villager_team(cls, role: str) - bool: 判断是否为好人阵营 return cls.ROLES.get(role, {}).get(team) 好人阵营 classmethod def get_standard_setup(cls, player_count: int) - List[str]: 获取标准角色配置 if player_count 6: return [狼人, 狼人, 预言家, 女巫, 村民, 村民] elif player_count 8: return [狼人, 狼人, 狼人, 预言家, 女巫, 猎人, 村民, 村民] elif player_count 9: return [狼人, 狼人, 狼人, 预言家, 女巫, 猎人, 守护者, 村民, 村民] else: # 默认配置约1/3狼人 werewolf_count max(1, player_count // 3) roles [狼人] * werewolf_count # 添加神职 remaining player_count - werewolf_count if remaining 1: roles.append(预言家) remaining - 1 if remaining 1: roles.append(女巫) remaining - 1 if remaining 1: roles.append(猎人) remaining - 1 # 剩余为村民 roles.extend([村民] * remaining) return rolesmain_cn.py# -*- coding: utf-8 -*- 三国狼人杀 - 基于AgentScope的中文版狼人杀游戏 融合三国演义角色和传统狼人杀玩法 import asyncio import os import random from typing import List, Dict, Optional from agentscope.agent import ReActAgent from agentscope.model import DashScopeChatModel from agentscope.pipeline import MsgHub, sequential_pipeline, fanout_pipeline from agentscope.formatter import DashScopeMultiAgentFormatter from prompt_cn import ChinesePrompts from game_roles import GameRoles from structured_output_cn import ( DiscussionModelCN, get_vote_model_cn, WitchActionModelCN, get_seer_model_cn, get_hunter_model_cn, WerewolfKillModelCN ) from utils_cn import ( check_winning_cn, majority_vote_cn, get_chinese_name, format_player_list, GameModerator, MAX_GAME_ROUND, MAX_DISCUSSION_ROUND, ) class ThreeKingdomsWerewolfGame: 三国狼人杀游戏主类 def __init__(self): self.players: Dict[str, ReActAgent] {} self.roles: Dict[str, str] {} self.moderator GameModerator() self.alive_players: List[ReActAgent] [] self.werewolves: List[ReActAgent] [] self.villagers: List[ReActAgent] [] self.seer: List[ReActAgent] [] self.witch: List[ReActAgent] [] self.hunter: List[ReActAgent] [] # 女巫道具状态 self.witch_has_antidote True self.witch_has_poison True async def create_player(self, role: str, character: str) - ReActAgent: 创建具有三国背景的玩家 name get_chinese_name(character) self.roles[name] role agent ReActAgent( namename, sys_promptChinesePrompts.get_role_prompt(role, character), modelDashScopeChatModel( model_nameqwen-max, api_keyos.environ[DASHSCOPE_API_KEY], enable_thinkingTrue, ), formatterDashScopeMultiAgentFormatter(), ) # 角色身份确认 await agent.observe( await self.moderator.announce( f【{name}】你在这场三国狼人杀中扮演{GameRoles.get_role_desc(role)} f你的角色是{character}。{GameRoles.get_role_ability(role)} ) ) self.players[name] agent return agent async def setup_game(self, player_count: int 6): 设置游戏 print( 开始设置三国狼人杀游戏...) # 获取角色配置 roles GameRoles.get_standard_setup(player_count) characters random.sample([ 刘备, 关羽, 张飞, 诸葛亮, 赵云, 曹操, 司马懿, 周瑜, 孙权 ], player_count) # 创建玩家 for i, (role, character) in enumerate(zip(roles, characters)): agent await self.create_player(role, character) self.alive_players.append(agent) # 分配到对应阵营 if role 狼人: self.werewolves.append(agent) elif role 预言家: self.seer.append(agent) elif role 女巫: self.witch.append(agent) elif role 猎人: self.hunter.append(agent) else: self.villagers.append(agent) # 游戏开始公告 await self.moderator.announce( f三国狼人杀游戏开始参与者{format_player_list(self.alive_players)} ) print(f✅ 游戏设置完成共{len(self.alive_players)}名玩家) async def werewolf_phase(self, round_num: int): 狼人阶段 if not self.werewolves: return None await self.moderator.announce(f 狼人请睁眼选择今晚要击杀的目标...) # 狼人讨论 async with MsgHub( self.werewolves, enable_auto_broadcastTrue, announcementawait self.moderator.announce( f狼人们请讨论今晚的击杀目标。存活玩家{format_player_list(self.alive_players)} ), ) as werewolves_hub: # 讨论阶段 for _ in range(MAX_DISCUSSION_ROUND): for wolf in self.werewolves: await wolf(structured_modelDiscussionModelCN) # 投票击杀 werewolves_hub.set_auto_broadcast(False) kill_votes await fanout_pipeline( self.werewolves, msgawait self.moderator.announce(请选择击杀目标), structured_modelWerewolfKillModelCN, enable_gatherFalse, ) # 统计投票 votes {} for i, vote_msg in enumerate(kill_votes): # 检查vote_msg是否为None或metadata是否存在 if vote_msg is not None and hasattr(vote_msg, metadata) and vote_msg.metadata is not None: votes[self.werewolves[i].name] vote_msg.metadata.get(target) else: # 如果返回无效,随机选择一个目标 print(f⚠️ {self.werewolves[i].name} 的击杀投票无效,随机选择目标) import random valid_targets [p.name for p in self.alive_players if p.name not in [w.name for w in self.werewolves]] votes[self.werewolves[i].name] random.choice(valid_targets) if valid_targets else None killed_player, _ majority_vote_cn(votes) return killed_player async def seer_phase(self): 预言家阶段 if not self.seer: return seer_agent self.seer[0] await self.moderator.announce( 预言家请睁眼选择要查验的玩家...) check_result await seer_agent( structured_modelget_seer_model_cn(self.alive_players) ) # 检查返回结果是否有效 if check_result is None or not hasattr(check_result, metadata) or check_result.metadata is None: print(f⚠️ 预言家查验失败,跳过此阶段) return target_name check_result.metadata.get(target) if not target_name: print(f⚠️ 预言家未选择查验目标,跳过此阶段) return target_role self.roles.get(target_name, 村民) # 告知预言家结果 result_msg f查验结果{target_name}是{狼人 if target_role 狼人 else 好人} await seer_agent.observe(await self.moderator.announce(result_msg)) async def witch_phase(self, killed_player: str): 女巫阶段 if not self.witch: return killed_player, None witch_agent self.witch[0] await self.moderator.announce(‍♀️ 女巫请睁眼...) # 告知女巫死亡信息 death_info f今晚{killed_player}被狼人击杀 if killed_player else 今晚平安无事 await witch_agent.observe(await self.moderator.announce(death_info)) # 女巫行动 witch_action await witch_agent(structured_modelWitchActionModelCN) saved_player None poisoned_player None # 检查返回结果是否有效 if witch_action is None or not hasattr(witch_action, metadata) or witch_action.metadata is None: print(f⚠️ 女巫行动失败,视为不使用技能) else: if witch_action.metadata.get(use_antidote) and self.witch_has_antidote: if killed_player: saved_player killed_player self.witch_has_antidote False await witch_agent.observe(await self.moderator.announce(f你使用解药救了{killed_player})) if witch_action.metadata.get(use_poison) and self.witch_has_poison: poisoned_player witch_action.metadata.get(target_name) if poisoned_player: self.witch_has_poison False await witch_agent.observe(await self.moderator.announce(f你使用毒药毒杀了{poisoned_player})) # 确定最终死亡玩家 final_killed killed_player if not saved_player else None return final_killed, poisoned_player async def hunter_phase(self, shot_by_hunter: str): 猎人阶段 if not self.hunter: return None hunter_agent self.hunter[0] if hunter_agent.name shot_by_hunter: await self.moderator.announce( 猎人发动技能可以带走一名玩家...) hunter_action await hunter_agent( structured_modelget_hunter_model_cn(self.alive_players) ) # 检查返回结果是否有效 if hunter_action is None or not hasattr(hunter_action, metadata) or hunter_action.metadata is None: print(f⚠️ 猎人技能使用失败,视为放弃开枪) return None if hunter_action.metadata.get(shoot): target hunter_action.metadata.get(target) if target: await self.moderator.announce(f猎人{hunter_agent.name}开枪带走了{target}) return target else: print(f⚠️ 猎人选择开枪但未指定目标,视为放弃) return None return None def update_alive_players(self, dead_players: List[str]): 更新存活玩家列表 for dead_name in dead_players: if dead_name: # 从存活列表移除 self.alive_players [p for p in self.alive_players if p.name ! dead_name] # 从各阵营移除 self.werewolves [p for p in self.werewolves if p.name ! dead_name] self.villagers [p for p in self.villagers if p.name ! dead_name] self.seer [p for p in self.seer if p.name ! dead_name] self.witch [p for p in self.witch if p.name ! dead_name] self.hunter [p for p in self.hunter if p.name ! dead_name] async def day_phase(self, round_num: int): 白天阶段 await self.moderator.day_announcement(round_num) # 讨论阶段 async with MsgHub( self.alive_players, enable_auto_broadcastTrue, announcementawait self.moderator.announce( f现在开始自由讨论。存活玩家{format_player_list(self.alive_players)} ), ) as all_hub: # 每人发言一轮 await sequential_pipeline(self.alive_players) # 投票阶段 all_hub.set_auto_broadcast(False) vote_msgs await fanout_pipeline( self.alive_players, await self.moderator.announce(请投票选择要淘汰的玩家), structured_modelget_vote_model_cn(self.alive_players), enable_gatherFalse, ) # 统计投票 votes {} for i, vote_msg in enumerate(vote_msgs): # 检查vote_msg是否为None或metadata是否存在 if vote_msg is not None and hasattr(vote_msg, metadata) and vote_msg.metadata is not None: votes[self.alive_players[i].name] vote_msg.metadata.get(vote) else: # 如果返回无效,默认弃票 print(f⚠️ {self.alive_players[i].name} 的投票无效,视为弃票) votes[self.alive_players[i].name] None voted_out, vote_count majority_vote_cn(votes) await self.moderator.vote_result_announcement(voted_out, vote_count) return voted_out async def run_game(self): 运行游戏主循环 try: await self.setup_game() for round_num in range(1, MAX_GAME_ROUND 1): print(f\n 第{round_num}轮游戏开始 ) # 夜晚阶段 await self.moderator.night_announcement(round_num) # 狼人击杀 killed_player await self.werewolf_phase(round_num) # 预言家查验 await self.seer_phase() # 女巫行动 final_killed, poisoned_player await self.witch_phase(killed_player) # 更新死亡玩家 night_deaths [p for p in [final_killed, poisoned_player] if p] self.update_alive_players(night_deaths) # 死亡公告 await self.moderator.death_announcement(night_deaths) # 检查胜利条件 winner check_winning_cn(self.alive_players, self.roles) if winner: await self.moderator.game_over_announcement(winner) return # 白天阶段 voted_out await self.day_phase(round_num) # 猎人技能 hunter_shot await self.hunter_phase(voted_out) # 更新死亡玩家 day_deaths [p for p in [voted_out, hunter_shot] if p] self.update_alive_players(day_deaths) # 检查胜利条件 winner check_winning_cn(self.alive_players, self.roles) if winner: await self.moderator.game_over_announcement(winner) return print(f第{round_num}轮结束存活玩家{format_player_list(self.alive_players)}) except Exception as e: print(f❌ 游戏运行出错{e}) import traceback traceback.print_exc() async def main(): 主函数 # 检查环境变量 if DASHSCOPE_API_KEY not in os.environ: print(❌ 请设置环境变量 DASHSCOPE_API_KEY) return print( 欢迎来到三国狼人杀) # 创建并运行游戏 game ThreeKingdomsWerewolfGame() await game.run_game() if __name__ __main__: asyncio.run(main())prompt_cn.py# -*- coding: utf-8 -*- 三国狼人杀中文提示词 class ChinesePrompts: 中文提示词管理类 staticmethod def get_role_prompt(role: str, character: str) - str: 获取角色提示词 base_prompt f你是{character}在这场三国狼人杀游戏中扮演{role}。 请严格按照以下JSON格式回复不要添加任何其他文字 {{ reach_agreement: true/false, confidence_level: 1-10的数字, key_evidence: 你的证据或观点 }} 角色特点 if role 狼人: return base_prompt f - 你是狼人阵营目标是消灭所有好人 - 夜晚可以与其他狼人协商击杀目标 - 白天要隐藏身份误导好人 - 以{character}的性格说话和行动 elif role 预言家: return base_prompt f - 你是好人阵营的预言家目标是找出所有狼人 - 每晚可以查验一名玩家的真实身份 - 要合理公布查验结果引导好人投票 - 以{character}的智慧和洞察力分析局势 elif role 女巫: return base_prompt f - 你是好人阵营的女巫拥有解药和毒药各一瓶 - 解药可以救活被狼人击杀的玩家 - 毒药可以毒杀一名玩家 - 要谨慎使用道具在关键时刻发挥作用 elif role 猎人: return base_prompt f - 你是好人阵营的猎人 - 被投票出局时可以开枪带走一名玩家 - 要在关键时刻使用技能带走狼人 - 以{character}的勇猛和决断力行动 else: # 村民 return base_prompt f - 你是好人阵营的村民 - 没有特殊技能只能通过推理和投票 - 要仔细观察找出狼人的破绽 - 以{character}的性格参与讨论 requirements.txtagentscope1.0.2structured_output_cn.py# -*- coding: utf-8 -*- 三国狼人杀游戏的结构化输出模型 from typing import Literal, Optional, List from pydantic import BaseModel, Field from agentscope.agent import AgentBase class DiscussionModelCN(BaseModel): 中文版讨论输出格式 reach_agreement: bool Field( description是否已达成一致意见, ) confidence_level: int Field( description对当前推理的信心程度(1-10), ge1, le10 ) key_evidence: Optional[str] Field( description支持你观点的关键证据, defaultNone ) def get_vote_model_cn(agents: list[AgentBase]) - type[BaseModel]: 获取中文版投票模型 class VoteModelCN(BaseModel): 中文版投票输出格式 vote: Literal[tuple(_.name for _ in agents)] Field( description你要投票淘汰的玩家姓名, ) reason: str Field( description投票理由简要说明为什么选择此人, ) suspicion_level: int Field( description对被投票者的怀疑程度(1-10), ge1, le10 ) return VoteModelCN class WitchActionModelCN(BaseModel): 中文版女巫行动模型 use_antidote: bool Field( description是否使用解药救人, defaultFalse ) use_poison: bool Field( description是否使用毒药杀人, defaultFalse ) target_name: Optional[str] Field( description目标玩家姓名救人或毒杀的对象, defaultNone ) action_reason: Optional[str] Field( description行动理由, defaultNone ) def get_seer_model_cn(agents: list[AgentBase]) - type[BaseModel]: 获取中文版预言家模型 class SeerModelCN(BaseModel): 中文版预言家查验格式 target: Literal[tuple(_.name for _ in agents)] Field( description要查验的玩家姓名, ) check_reason: str Field( description查验此人的原因, ) priority_level: int Field( description查验优先级(1-10), ge1, le10 ) return SeerModelCN def get_hunter_model_cn(agents: list[AgentBase]) - type[BaseModel]: 获取中文版猎人模型 class HunterModelCN(BaseModel): 中文版猎人开枪格式 shoot: bool Field( description是否使用开枪技能, ) target: Optional[Literal[tuple(_.name for _ in agents)]] Field( description开枪目标玩家姓名, defaultNone ) shoot_reason: Optional[str] Field( description开枪理由, defaultNone ) return HunterModelCN class WerewolfKillModelCN(BaseModel): 中文版狼人击杀模型 target: str Field( description要击杀的玩家姓名, ) kill_strategy: str Field( description击杀策略说明, ) team_coordination: Optional[str] Field( description与狼队友的配合计划, defaultNone ) class GameAnalysisModelCN(BaseModel): 中文版游戏分析模型 suspected_werewolves: List[str] Field( description怀疑的狼人名单, default_factorylist ) trusted_players: List[str] Field( description信任的玩家名单, default_factorylist ) key_clues: List[str] Field( description关键线索列表, default_factorylist ) next_strategy: str Field( description下一步策略, )utils_cn.py# -*- coding: utf-8 -*- 三国狼人杀游戏工具函数 import asyncio import random from typing import List, Dict, Optional, Any from collections import Counter from agentscope.agent import AgentBase from agentscope.message import Msg # 游戏常量 MAX_GAME_ROUND 10 MAX_DISCUSSION_ROUND 3 CHINESE_NAMES [ 刘备, 关羽, 张飞, 诸葛亮, 赵云, 曹操, 司马懿, 典韦, 许褚, 夏侯惇, 孙权, 周瑜, 陆逊, 甘宁, 太史慈, 吕布, 貂蝉, 董卓, 袁绍, 袁术 ] def get_chinese_name(character: str None) - str: 获取中文角色名 if character and character in CHINESE_NAMES: return character return random.choice(CHINESE_NAMES) def format_player_list(players: List[AgentBase], show_roles: bool False) - str: 格式化玩家列表为中文显示 if not players: return 无玩家 if show_roles: return 、.join([f{p.name}({getattr(p, role, 未知)}) for p in players]) else: return 、.join([p.name for p in players]) def majority_vote_cn(votes: Dict[str, str]) - tuple[str, int]: 中文版多数投票统计 if not votes: return 无人, 0 vote_counts Counter(votes.values()) most_voted vote_counts.most_common(1)[0] return most_voted[0], most_voted[1] def check_winning_cn(alive_players: List[AgentBase], roles: Dict[str, str]) - Optional[str]: 检查中文版游戏胜利条件 alive_roles [roles.get(p.name, 村民) for p in alive_players] werewolf_count alive_roles.count(狼人) villager_count len(alive_roles) - werewolf_count if werewolf_count 0: return 好人阵营胜利所有狼人已被淘汰 elif werewolf_count villager_count: return 狼人阵营胜利狼人数量已达到或超过好人 return None def analyze_speech_pattern(speech: str) - Dict[str, Any]: 分析发言模式中文优化 analysis { word_count: len(speech), confidence_keywords: 0, doubt_keywords: 0, emotion_score: 0 } # 中文关键词分析 confidence_words [确定, 肯定, 一定, 绝对, 必须, 显然] doubt_words [可能, 也许, 或许, 怀疑, 不确定, 感觉] for word in confidence_words: analysis[confidence_keywords] speech.count(word) for word in doubt_words: analysis[doubt_keywords] speech.count(word) # 简单情感分析 positive_words [好, 棒, 赞, 支持, 同意] negative_words [坏, 差, 反对, 不行, 错误] for word in positive_words: analysis[emotion_score] speech.count(word) for word in negative_words: analysis[emotion_score] - speech.count(word) return analysis class GameModerator(AgentBase): 中文版游戏主持人 def __init__(self) - None: super().__init__() self.name 游戏主持人 self.game_log: List[str] [] async def announce(self, content: str) - Msg: 发布游戏公告 msg Msg( nameself.name, contentf {content}, rolesystem ) self.game_log.append(content) await self.print(msg) return msg async def night_announcement(self, round_num: int) - Msg: 夜晚阶段公告 content f 第{round_num}夜降临天黑请闭眼... return await self.announce(content) async def day_announcement(self, round_num: int) - Msg: 白天阶段公告 content f☀️ 第{round_num}天天亮了请大家睁眼... return await self.announce(content) async def death_announcement(self, dead_players: List[str]) - Msg: 死亡公告 if not dead_players: content 昨夜平安无事无人死亡。 else: content f昨夜{format_player_list_str(dead_players)}不幸遇害。 return await self.announce(content) async def vote_result_announcement(self, voted_out: str, vote_count: int) - Msg: 投票结果公告 content f投票结果{voted_out}以{vote_count}票被淘汰出局。 return await self.announce(content) async def game_over_announcement(self, winner: str) - Msg: 游戏结束公告 content f 游戏结束{winner} return await self.announce(content) def format_player_list_str(players: List[str]) - str: 格式化玩家姓名列表 if not players: return 无人 return 、.join(players) def calculate_suspicion_score(player_name: str, game_history: List[Dict]) - float: 计算玩家可疑度分数 score 0.0 for event in game_history: if event.get(type) vote and event.get(target) player_name: score 0.3 elif event.get(type) accusation and event.get(target) player_name: score 0.2 elif event.get(type) defense and event.get(player) player_name: score - 0.1 return min(max(score, 0.0), 1.0) async def handle_interrupt(*args: Any, **kwargs: Any) - Msg: 处理游戏中断 return Msg( name系统, content游戏被中断, rolesystem )总结AgentScope 与 AutoGen 最大的区别在于它采用了消息驱动Message Driven的设计。所有 Agent 都围绕Msg和MsgHub进行通信再结合结构化输出、异步执行等能力使其更适合构建大型、多智能体的生产级应用。如果说AutoGen更像一个多 Agent 对话框架CAMEL更强调角色扮演协作那么AgentScope更像一个面向企业级应用的多智能体开发平台在复杂协作、并发处理和工程化方面具有明显优势。