从原理到代码拆解AgentScope框架开发实践

（一）引言

前面我们分析了AutoGen框架"以对话驱动协作"的核心理念，本文将聚焦于另一个备受关注的多智能体开发框架——AgentScope。与AutoGen不同，AgentScope更像一个专为多智能体应用量身打造的全功能开发平台，它拥有一套极具吸引力的设计哲学和工程实践。

AgentScope的核心卖点在于提供了一套高度友好的编程接口。开发者无需绕很多弯子，就能轻松定义智能体、搭建通信网络，并高效管理整个应用的生命周期。其内置的消息传递机制与对分布式部署的原生支持，使它特别适合那些需要长期稳定运行的生产环境应用。简单来说，如果你正在规划复杂、大规模的多智能体系统，AgentScope会是一个非常顺手且高效的选择。

（二）AgentScope的核心机制

AgentScope的设计哲学非常清晰：通过消息驱动架构结合工业级工程实践，为开发者提供从开发、测试到部署的全生命周期支持。与许多框架喜欢采用继承式设计不同，AgentScope选择了组合式架构和消息驱动模式，这构成了它与众不同的关键差异。

1. 分层架构体系

从底层的基础组件到上层的应用编排，AgentScope通过分层模块化设计，构建了一个完整的智能体开发生态。你可以将其比作搭积木的过程——每一层各司其职，组合起来就能灵活应对各种场景。

（1）基础组件层

这一层是整个架构的根基，提供了以下几个核心模块：

• Message组件：定义了统一的消息格式，从简单的文本到复杂的多模态内容都能轻松搞定。
• Memory组件：管理短期与长期记忆，让智能体能够"记得住"上下文，保持对话连贯性。
• Model API层：抽象了对不同大语言模型的调用，更换模型就像更换插件一样简单方便。
• Tool组件：封装了智能体与外部世界交互的能力，例如调用API、查询数据库等。

（2）智能体基础设施层

这层不仅提供了各种预构建的智能体（Built-in Agents），还实现了经典的ReAct范式。最关键的是，它原生支持异步执行和实时控制。相比其他框架，这一点是AgentScope的一大看点——开发者无需额外折腾就能轻松处理并发任务。

（3）多智能体协作层

这是AgentScope真正展现亮点的部分，也是其核心创新所在。

• MsgHub：消息中心，负责智能体之间的消息路由和状态管理。
• Pipeline系统：提供灵活的工作流编排能力，支持顺序、并发等多种执行模式。

（4）开发与部署层

这一层体现了AgentScope对工程化的高度重视，也是它区别于众多学术框架的关键所在。

• AgentScope Runtime：提供生产级的运行时环境，保障系统稳定运行。
• AgentScope Studio：为开发者提供完整的可视化开发工具链，调试监控都非常便捷。

2. 消息驱动机制

AgentScope的另一项核心创新在于其消息驱动架构。在此架构中，所有智能体的交互都被抽象为消息的发送和接收，而非传统的函数调用。

from agentscope.message import Msg
# 消息的标准格式
message = Msg(
    name="Alice",
    content="Hello",
    role="user",
    metadata={
        "timestamp": "2026-05-26T22:21:00Z",
        "message_type": "text",
        "priority": "normal"
    }
)

将消息作为交互的基础单元，带来了几个显而易见的好处：

• 异步解耦：发送方和接收方在时间上完全解耦，无需互相等待，天然适合高并发场景。
• 位置透明：一个智能体无需关心另一个智能体是在本地还是远程服务器上，消息系统会自动处理路由。
• 可观测性：每一条消息都可以被记录、追踪和分析，调试复杂系统时能节省大量精力。
• 可靠性：消息可以持久化存储并支持重试机制，即使系统出现故障，也能保证交互的最终一致性。

3. 智能体生命周期管理

在AgentScope中，每个智能体都有明确的生命周期——初始化、运行、暂停、销毁。这一切都基于统一的基类AgentBase来实现。开发者通常只需要关注其中的reply方法，大大减少了心智负担。

from agentscope.agents import AgentBase

class CustomAgent(AgentBase):

    def __init__(self, name: str, **kwargs):
        super().__init__(name=name, **kwargs)
        # 智能体的初始化逻辑

    def reply(self, x: Msg) -> Msg:
        # 智能体的核心响应逻辑
        response = self.model(x.content)
        return Msg(name=self.name, content=response, role="assitant")

    def observe(self, x: Msg) -> None:
        # 智能体的观察逻辑
        self.memory.add(x)

这种设计巧妙地将智能体的内部逻辑与外部通信分离开来。你只需要在reply方法里告诉智能体"怎么想、怎么回应"，其余的事情框架会自动帮你管好。

4. 消息传递机制

AgentScope内置的消息中心（MsgHub）是整个消息驱动架构的中枢神经系统。它不仅仅负责路由和分发消息那么简单，还集成了持久化和分布式通信等高级功能，特点非常明确：

• 灵活的消息路由：支持点对点、广播、组播等多种通信方式，能够搭建各种复杂的交互网络。
• 消息持久化：所有消息自动保存到数据库，长期运行的任务状态可以通过这种方式轻松恢复。
• 原生分布式支持：智能体可以部署在不同进程或服务器上，MsgHub通过RPC自动处理跨节点通信，对开发者完全透明。

（三）AgentScope的工程实现

理论说得再多，不如看一个实际例子来得直观。我们尝试用AgentScope搭建一个融合中国古典文化的"三国狼人杀"游戏。这个案例既能展示AgentScope在处理复杂多智能体交互方面的优势，更重要的是，它演示了在需要实时协作、角色扮演和策略博弈的场景中，消息驱动架构的强大威力。

与传统狼人杀不同，我们把刘备、关羽、诸葛亮这些三国经典角色引入游戏。每个智能体不仅要完成狼人杀的基本任务——狼人击杀、预言家查验、村民推理，还要体现出对应三国人物的性格特点和行为模式。这种设计能够让我们观察到AgentScope在处理多层次角色建模时的表现。

游戏角色：狼人、预言家、女巫、猎人、村民。
三国人物：刘备、关羽、张飞、诸葛亮、曹操、司马懿。

1. 需求分析

我们遵循分层解耦的原则，将游戏逻辑划分成三个独立的层次，每个层次都映射了AgentScope的核心组件：

• 游戏控制层（ThreeKingdomsWerewolfGame）：负责游戏状态管理、流程控制和胜负判定。
• 智能体交互层（MsgHub）：负责消息路由、并发处理和状态同步。
• 角色建模层（DialogAgent）：通过角色提示词、结构化输出和行为约束来实现双重身份建模。

每个玩家都是一个基于DialogAgent的实例，通过系统提示词注入"游戏角色"和"三国人格"的双重身份。所有通信——无论是狼人之间的秘密协商，还是白天的公开辩论——都由MsgHub来路由和转发。游戏主控制器负责维护全局状态、推进流程和裁定胜负。

2. AgentScope框架编码实现

（1）架构设计与核心组件

架构的层次结构可以概括为：

游戏控制层 (ThreeKingdomsWerewolfGame)
├── 游戏状态管理
├── 流程控制
└── 胜负判定

智能体交互层 (MsgHub)
├── 消息路由
├── 并发处理
└── 状态同步

角色建模层 (DialogAgent)
├── 角色提示词
├── 结构化输出
└── 行为约束

（2）消息驱动的游戏流程

这个案例最核心的设计亮点，是用消息驱动代替状态机来管理游戏流程。传统实现中，游戏阶段的转换通常由一个中心化的状态机控制。但在AgentScope的范式下，游戏流程被自然地建模成一系列定义好的消息交互模式。

举个例子，狼人阶段的实现不是简单的函数调用，而是通过MsgHub动态创建一个临时的私密通信频道，只对狼人玩家开放：

# 狼人阶段：展示消息驱动的协作模式
async def werewolf_phase(self round_num: int):
    # 安全检查
    if not self.werewolves:
        return None

    # 1. 通过消息中心建立狼人专属通信频道
    async with MsgHub(
        self.werewolves,
        enable_auto_broadcast=True,
        announcement=await self.moderator.announce(
            f"狼人们，请讨论今晚的击杀目标。存活玩家：{format_player_list(self.alive_player)}"
        ),
    ) as werewolves_hub:
        # 2. 讨论阶段
        for _ in range(MAX_DISCUSSION_ROUND):
            for wolf in self.werewolves:
                await wolf(structured_model=DiscussionModelCN)

        # 3. 投票阶段
        werewolves_hub.set_auto_broadcast(False)
        kill_votes = await fanout_pipline(
            self.werewolves,
            msg=await self.moderator.announce("请选择击杀目标"),
            structured_model=WerewolfKillModelCN,
            enable_gather=False,
        )

这种设计的好处在于，游戏逻辑被清晰地表达为"在特定上下文中，以何种模式进行消息交互"，而不是一连串僵硬的状态转换。白天讨论、预言家查验等阶段也都遵循同样的范式。

（3）用结构化输出约束游戏规则

狼人杀游戏的一个关键挑战，是如何确保智能体的行为符合游戏规则。AgentScope通过结构化输出机制来提供解决方案。我们为不同的游戏行为定义了严格的数据模型：

class DiscussionModelCN(BaseModel):
    reach_agreement: bool = Field(description="是否已达成一致意见", default=False)
    confidence_level: int = Field(description="对当前推理的信心程度(1-10)", ge=1, le=10)
    key_evidence: Optional[str] = Field(description="支持你观点的关键证据", default=None)

class WitchActionModelCN(BaseModel):
    use_antidote: bool = Field(description="是否使用解药")
    use_poison: bool = Field(description="是否使用毒药")
    target_name: Optional[str] = Field(description="毒药目标玩家姓名")

通过统一输出格式，智能体的输出被严格约束。女巫不能同时对同一目标使用解药和毒药，预言家每晚只能查验一名玩家——这些约束都是通过数据模型的字段定义和验证逻辑自动执行的。

（4）角色建模的双重挑战

这个案例中最有趣的技术挑战，是如何让智能体同时扮演好游戏功能角色（狼人、预言家等）和文化人格角色（刘备、关羽等）。我们通过提示词工程来解决：

def get_role_prompt(role: str, character: str) -> str:
    base_prompt = f"""你是{character}，在这场三国狼人杀游戏中扮演{role}。重要规则:
1. 你只能通过对话和推理参与游戏
2. 不要尝试调用任何外部工具或函数
3. 严格按照需要的JSON格式回复
角色特点："""
    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}的性格参与讨论"""

这种设计会带来一个有趣的现象：不同的三国人物在扮演相同游戏角色时，会表现出完全不同的策略和说话风格。扮演狼人的曹操可能更狡猾、更善于伪装，而扮演狼人的张飞则可能更冲动、更直接。

（5）并发处理机制

AgentScope的异步框架在这个多智能体游戏中发挥了关键作用。游戏中经常出现需要同时收集多个智能体决策的场景，比如投票阶段：

vote_msgs = await fanout_pipeline(
    self.alive_players,
    await self.moderator.announce("请投票选择要淘汰的玩家"),
    structured_model=get_vote_model_cn(self.alive_players),
    enable_gather=False,
)

fanout_pipeline允许我们并行地向所有智能体发送相同的信息，并异步收集响应。这不仅提高了游戏效率，更重要的是模拟了真实狼人杀游戏中的"同时投票"场景。

（6）AgentScope框架的完整实现

这里以Ubuntu 22.04系统、Python 3.10版本为例，走一遍完整流程。

agentscope_demo/
├── main_cn.py          - 主要游戏逻辑和控制逻辑
├── prompt_cn.py        - 中文提示词管理
├── game_roles.py       - 游戏角色定义与配置
├── structured_output_cn.py - 结构化输出
├── utils_cn.py         - 游戏工具函数和辅助方法
├── .env
└── README.md

游戏流程设置：

夜晚阶段

1. 狼人讨论：狼人通过MsgHub协商击杀目标。
2. 预言家查验：预言家选择查验对象。
3. 女巫行动：女巫决定是否使用解药/毒药。

白天阶段

1. 死亡公布：公布夜晚死亡玩家。
2. 自由讨论：所有存活玩家参与讨论。
3. 投票淘汰：投票选择淘汰对象。
4. 猎人技能：被淘汰的猎人可开枪。

（i）安装依赖与配置APIKey

创建虚拟环境并激活：

python -m venv agentscope_venv
source agentscope_venv/bin/activate

安装依赖（使用清华镜像加速）：

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple -r requirements.txt

其中requirements.txt内容为：

agentscope
dashscope
pydantic

配置阿里云DashScope API Key到.env文件：

DASHSCOPE_API_KEY=sk-91c2*********a698

prompt_cn.py

# -*- coding: utf-8 -*-
"""三国狼人杀中文提示词"""

class ChinesePrompts:
    """中文提示词管理类"""

    @staticmethod
    def get_role_prompt(role: str, character: str) -> str:
        """获取角色提示词，role是游戏中的角色，character是三国中的角色"""
        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}的性格参与讨论"""

game_roles.py

# -*- coding: utf-8 -*-
"""三国狼人杀角色定义"""
from typing import Dict, List

class GameRoles:
    """游戏角色管理类"""
    ROLES = {
        "狼人": {"description": "狼人", "ability": "夜晚可以击杀一名玩家", "team": "狼人阵营"},
        "预言家": {"description": "预言家", "ability": "每晚可以查验一名玩家的身份", "team": "好人阵营"},
        "女巫": {"description": "女巫", "ability": "拥有解药和毒药各一瓶，可以救人或杀人", "team": "好人阵营"},
        "猎人": {"description": "猎人", "ability": "被投票出局时可以开枪带走一名玩家", "team": "好人阵营"},
        "村民": {"description": "村民", "ability": "无特殊技能，依靠推理和投票", "team": "好人阵营"},
    }
    CHARACTER_TRAITS = {
        "刘备": "仁德宽厚，善于团结众人",
        "关羽": "忠义刚烈，言辞直接",
        "张飞": "性格豪爽，容易冲动",
        "诸葛亮": "智慧超群，分析透彻",
        "曹操": "雄才大略，善于权谋",
        "司马懿": "深谋远虑，城府极深",
    }

utils_cn.py

这段代码包含了游戏常量、玩家名称列表、格式化函数、投票统计、胜负判定、发言分析以及游戏主持人（GameModerator）等核心工具。

structured_output_cn.py

这里定义了讨论模型、投票模型、女巫行动模型、预言家查验模型、猎人开枪模型、狼人击杀模型和游戏分析模型——每个模型都通过Pydantic的Field约束实现了严格的输出格式验证。

main.py

主函数负责加载环境变量、创建ThreeKingdomsWerewolfGame实例并运行游戏。游戏主循环包含夜晚阶段（狼人击杀、预言家查验、女巫行动）和白天阶段（死亡公布、自由讨论、投票淘汰、猎人开枪），每轮结束后检查胜负条件。

3. 测试效果

运行python main.py后，可以在终端看到完整的游戏输出，包括每轮的夜间行动、白天讨论、投票结果和最终的胜负判定。

（四）AgentScope的优势与局限性

1. 优势

• 消息驱动架构：将复杂的游戏流程优雅地映射为一系列开放、异步的消息传递事件，避免了状态机的僵硬与复杂。
• 结构化输出：将游戏规则直接转化为代码层面的约束，极大提升了系统的可预测性和可调试性。
• 容错处理：即使单个智能体出现异常，整体流程也能稳健运行。

2. 局限性

• 对开发者的技术要求较高，需要掌握异步编程、分布式通信等概念。
• 对于简单的多智能体对话场景，这种架构可能显得过于复杂，存在"过度工程化"的风险。
• 作为相对较新的智能体框架，其生态系统和社区资源还有待进一步完善。

总的来说，AgentScope在处理复杂、多层次的多智能体系统方面有着明显的优势，尤其是在需要长期稳定运行的生产环境中。但它也不是万能的银弹，选型时需要结合具体的应用场景和团队的技术储备来做判断。