创新实训项目开发记录（二）：从需求拆解到多智能体骨架-程序员充电站

一、写在前面

目前项目还处在初步开发阶段，很多功能还没有完全打磨完，但我负责的两部分工作已经逐渐形成了比较清晰的骨架：

项目整体规划、需求拆解、进度管控
多智能体架构设计、Agent 协作流程、状态机设计

回头看，前期最重要的事情其实不是“先把页面做漂亮”或者“先堆很多功能”，而是先回答两个问题：

这个项目到底要拆成哪些稳定的模块？
AI 角色和游戏流程，应该靠什么机制稳定地跑起来？

我们最后给出的答案是：先做状态机，再做多智能体协作骨架。只有这两层先稳定下来，后面的搜证、讨论、私聊、投票、复盘这些功能才不会越做越乱。

二、项目初期的需求拆解

这个项目表面上看是一个“多人在线剧本杀 + AI 角色参与”的系统，但如果从开发视角拆开，它至少包含四条主线：

房间与玩家管理
游戏流程推进
AI Agent 协作
前后端实时同步

所以在项目初期，我先做的不是细节实现，而是先把这几个问题变成可执行的模块边界。

我当时对项目的理解大致是这样的：

1. 房间系统负责“载体”

也就是创建房间、加入房间、分配角色、保存房间状态。

2. 状态机负责“节奏”

它决定一局游戏当前是在准备阶段、自我介绍阶段、搜证阶段，还是讨论、投票、复盘阶段。

3. 多智能体负责“角色扮演和协作”

Lobby Agent 负责选本和开局引导，DM Agent 负责主持流程，NPC Agent 负责扮演具体角色。

4. SSE 负责“把后端变化实时推给前端”

也就是说，后端不是等前端一直轮询，而是主动把 DM 发言、NPC 发言、阶段切换、投票结果推送到界面上。

三、为什么先做状态机

我认为这种项目最核心的问题，不是“功能多不多”，而是“流程稳不稳”。

因为剧本杀天然是一个强流程型系统。玩家不是随时随地都能做任意动作的，不同阶段允许的行为完全不一样：

准备阶段要先确认角色卡
自我介绍阶段要按顺序发言
搜证阶段要按行动次数获取线索
讨论阶段要控制发言顺序和提问逻辑
投票阶段要统计投票并处理平票重投
复盘阶段要还原真相与证据链

所以我在项目初期就把这个流程抽象成了一个统一的阶段枚举。

class GamePhase(str, Enum): WAITING = "waiting" PREPARATION = "preparation" INTRO = "intro" SELF_INTRO = "self_intro" INVESTIGATION = "investigation" DISCUSSION = "discussion" PRIVATE_WINDOW = "private_window" FINAL_STATEMENT = "final_statement" VOTING = "voting" REVEAL = "reveal" COMPLETE = "complete"

这意味着：后续所有接口、前端页面、Agent 行为，都要围绕统一阶段运转。

也正因为有了这个阶段定义，我们后面才能继续把房间对象扩展成一个真正的“游戏状态容器”。例如引擎中的GameRoom 已经开始承担这些职责：

@dataclass class GameRoom: id: str name: str scenario_id: str scenario_title: str phase: GamePhase = GamePhase.WAITING players: list[Player] = field(default_factory=list) clues: list[Clue] = field(default_factory=list) votes: list[Vote] = field(default_factory=list) questions: list[Question] = field(default_factory=list) messages: list[dict] = field(default_factory=list) investigation_round: int = 0 discussion_round: int = 0 private_threads: list[PrivateThread] = field(default_factory=list) alliances: list[Alliance] = field(default_factory=list) final_statements: dict[str, str] = field(default_factory=dict)

从这里就能看出来，我们不是在做一个“聊天室”，而是在做一个带有明确阶段推进、角色信息、线索系统、提问系统、私聊系统、联盟系统、投票系统的完整对局状态机。

四、Agent 协作流程

为了让多智能体不只是概念，我们在代码里先跑通了一条最关键的链路：

玩家选本 -> Lobby 开局 -> DM 接收剧本 -> NPC 接收角色包

在 Lobby Agent 的 _execute_game_boot() 中，它做了几件很清晰的事：

1. 读取剧本 YAML

先从 data/scenarios/ 中加载剧本内容，拿到标题、背景、角色信息。

2. 通知 DM

把 scenario_id、标题、背景等信息打包发给 dm。

dm_payload = { "type": "SCENARIO_SELECTED", "scenario_id": scenario_id, "title": title, "channel": channel, "background": background, } await self.send_direct( to="dm", content={"text": "【SCENARIO_SELECTED】\n" + json.dumps(dm_payload, ensure_ascii=False, indent=2)}, )

3. 向 NPC 下发 ROLE PACK

也就是把每个 AI 角色真正“激活”起来。

pack = { "role_id": role_id, "display": role.get("display"), "identity": role.get("identity"), "common_knowledge": role.get("common_knowledge", []), "disclosable_info": role.get("disclosable_info", []), "hidden_info": role.get("hidden_info", []), "faction": role.get("faction", "无辜"), } await self.send_direct( to=agent_id, content={"text": "【ROLE PACK】\n" + json.dumps(pack, ensure_ascii=False, indent=2)}, )

这一步其实就是我理解的“多智能体协作真正落地”的分界线。因为只有到了这里，NPC 才不再是一个空壳模型，而是拿到了自己的身份、可公开信息、隐藏信息和行为指导。

同时，NPC agent会在收到 ROLE PACK 后激活角色，并向 DM 回传确认信息。这就形成了一个最小闭环：

Lobby 负责下发任务
DM 负责主持全局
NPC 负责扮演局部角色
各 Agent 之间靠消息协作，而不是彼此硬编码耦合

从架构设计的角度看，这比“把所有逻辑写在一个类里”更有扩展性，也更符合我前期对系统边界的规划。

五、状态机和多智能体之间的衔接

如果只有阶段，没有 Agent，那么系统只是“流程定义”；
如果只有 Agent，没有阶段，那么系统就会“谁都能随便说话”。

所以真正关键的不是二选一，而是把两者接起来。

我专门做了阶段切换消息的发布逻辑：

async def _publish_phase_change(room_id, old_phase, new_phase, *, description="", metadata=None): payload = {"old_phase": old_phase.value, "new_phase": new_phase.value} await message_bus.publish( GameMessage( type=MessageType.PHASE_CHANGE, room_id=room_id, sender_id="system", sender_name="系统", content=content, metadata=payload, ) )

而房间的阶段变化也会真正落到状态对象上：

def change_phase(self, room_id: str, new_phase: GamePhase) -> bool: room = self.get_room(room_id) if room is None: return False old_phase = room.phase room.phase = new_phase ... return True

这说明现在项目里已经形成了一个比较明确的控制逻辑：

GameEngine 持有状态
GamePhase 约束阶段
AgentRunner 调度 DM 和 NPC
games.py 负责 API 与流程推进
MessageBus 负责向前端广播变化

六、前后端联调为什么要尽早做实时链路

在项目初步开发时，我有一个很明确的判断：前后端一定要尽早把实时通信打通。

因为这个项目不是静态页面，而是一个过程不断变化的互动系统。DM 发言、NPC 回答、阶段切换、投票结果、线索发现，这些东西如果前端不能及时看到，整个体验都会断掉。

所以我们在后端设计了一个统一的消息总线 message_bus.py：

class MessageType(str, Enum): DM_MESSAGE = "dm_message" NPC_MESSAGE = "npc_message" PLAYER_MESSAGE = "player_message" SYSTEM_NOTICE = "system_notice" PHASE_CHANGE = "phase_change" CLUE_DISCOVERED = "clue_discovered" VOTE_RESULT = "vote_result"

然后通过 stream.py 提供 SSE 接口：

@router.get("/{room_id}/stream") async def game_stream(room_id: str, request: Request, since: float = Query(default=0.0)):

前端再通过 useGameStream.ts 订阅这些事件：

const eventNames: StreamMessage['type'][] = [ 'dm_message', 'npc_message', 'player_message', 'phase_change', 'system_notice', 'clue_discovered', 'vote_result', ]

这一步的意义在于，前端已经不只是“请求一次接口然后渲染”，而是开始具备了实时感知游戏进度的能力。