在上一篇中，我通过提示词工程把一个 LLM 调教成了合格的赛季教练。但当分析维度变多，单个 Agent 开始“顾此失彼”。这一篇，我将整个分析引擎重构成三个独立 Agent——经济、战力、站位——让它们通过消息总线实时协商，并发执行，像一支微型专家团队一样协同工作。

一、从单 Agent 到多 Agent：为什么需要拆分？

在测试过程中发现，当阵容较为复杂（如8人口、多羁绊、多装备组合）时，当前模型存在以下问题：

1. 装备推荐与站位建议存在逻辑冲突（例如建议将缺乏装备的核心输出置于前排）
2. 经济评估未能有效传导至战力分析（在游戏前期却采用后期标准来警示装备不足）

这些问题的主要成因是：所有推理过程共用一个上下文环境，导致模型难以对各子任务分配足够的专注力。为解决此问题，建议采用多智能体架构方案：

• 将每个子任务独立设置为专属Agent
• 每个Agent拥有独立的上下文和专用提示词
• 通过消息传递机制实现Agent间的结论交互

二、架构全景：消息驱动的专家团队

新架构由三部分组成：

AgentMessage（消息信封） → AgentBus（消息总线） → 3 个 Agent + 1 个编排器

• AgentMessage：标准化的消息格式，包含发送方、接收方、消息类型及载荷内容。

  AgentBus：线程安全的消息总线，提供发布、订阅和拉取功能。

  专用Agent：

• EconomyAgent：负责经济运营管理
• PowerAgent：专注于战力装备调配
• PositionAgent：处理站位布阵策略

AgentOrchestrator：通过线程池并发调度所有Agent，并整合最终输出结果。

各Agent仍沿用之前的提示词模板（角色设定+结构化输出），但每个Agent仅专注于单一功能维度，显著降低了单个prompt的复杂度。

三、让 Agent 互相“通气”的消息总线

Agent 之间不直接调用，而是通过总线异步通信。这是实现解耦的关键。

class AgentBus:
    def __init__(self):
        self._lock = threading.Lock()
        self._mailboxes = defaultdict(list)    # 邮箱
        self._handlers  = defaultdict(list)    # 回调注册

    def subscribe(self, agent_name, msg_type, callback):
        with self._lock:
            self._handlers[msg_type].append((agent_name, callback))

    def publish(self, msg: AgentMessage):
        with self._lock:
            # 投递到邮箱
            ...
            # 触发回调
            handlers = list(self._handlers.get(msg.msg_type, []))
        for agent_name, handler in handlers:
            if agent_name != msg.sender:
                handler(msg)

两种通信模式：

1. 邮箱拉取
   Agent 在 run() 方法中通过主动调用 get_messages() 来获取待处理消息。

2. 回调推送
   Agent 使用 subscribe() 方法注册回调函数，消息到达时系统会实时触发回调。

四、三个专家的实现（节选）

EconomyAgent：判断阶段并广播

class EconomyAgent(BaseAgent):
    def run(self, analysis: dict) -> str:
        # 根据 team_size 判断阶段
        if team_size <= 5:
            phase = "早期"
        # ...
        # 广播阶段结论
        self.emit("phase_result", {"phase": phase, "avg_star": avg_star})
        # 生成经济建议（内部调用 LLM + 提示词模板）
        return self._generate_advice(analysis, phase)

PowerAgent：订阅经济消息，动态调整装备阈值

class PowerAgent(BaseAgent):
    def __init__(self, bus, trait_dict):
        super().__init__("PowerAgent", bus)
        self._phase = "未知"
        bus.subscribe(self.name, "phase_result", self._on_phase)

    def _on_phase(self, msg: AgentMessage):
        self._phase = msg.payload.get("phase", "未知")

    def run(self, analysis: dict) -> str:
        # 根据 self._phase 调整装备检查阈值
        cost_thr = 3 if "后期" in self._phase else 4
        # 检查缺装备的高费英雄 ...
        # 广播战力结论
        self.emit("power_result", {"no_item_carries": no_item_carries})

PositionAgent：订阅战力消息，给出保护站位

class PositionAgent(BaseAgent):
    def __init__(self, bus):
        super().__init__("PositionAgent", bus)
        self._no_item_carries = []
        bus.subscribe(self.name, "power_result", self._on_power)

    def _on_power(self, msg: AgentMessage):
        self._no_item_carries = msg.payload.get("no_item_carries", [])

    def run(self, analysis: dict) -> str:
        # 如果主C缺装备，建议后排保护
        if self._no_item_carries:
            advice += f"💡 [{', '.join(self._no_item_carries)}] 缺装备，建议后排受保护位置"

五、编排器：并发调度与结论融合

class AgentOrchestrator:
    def run_parallel(self, analysis, timeout=10.0):
        with ThreadPoolExecutor(max_workers=3) as exe:
            futures = {exe.submit(ag.run, analysis): ag.name for ag in self.agents}
            for fut in as_completed(futures):
                name = futures[fut]
                self._results[name] = fut.result()
        return self._results

    def synthesize(self) -> str:
        order = ["EconomyAgent", "PowerAgent", "PositionAgent"]
        # 按经济→战力→站位拼装报告

六、与提示词工程的闭环联动

细心的读者不难发现：每个 Agent 的 run() 方法最终仍会调用大模型，并沿用了上篇的提示词模板。以 EconomyAgent 为例，其 prompt 头部依然包含角色设定和结构化输出要求，只是将讨论范围聚焦于经济领域。

这正是两篇博客的内在联系：提示词工程确保单个 Agent 的应答质量，而多智能体架构则实现 Agent 间的分工协作。二者相辅相成，共同构建出完整的 AI 阵容教练系统。
至此，TFT 系列的五篇博客全部结束。从一个 API 请求都不会写的新手，到搭建出多智能体分析引擎，我最大的感悟是：架构不是一开始就设计好的，而是在迭代中一步步“长出来”的。希望这个从 0 到 1 的过程，能给你带来一些启发。