4.3 最佳实践与设计模式

4.3 最佳实践与设计模式

在体系中的位置：前三章给了零件和用法，这一节把经验凝成可复用套路。设计模式的价值不是"显得专业"，而是让你避开别人已经踩过的坑——多智能体系统有几种反复出现的失败形态，模式就是对应的疫苗。

一个反问：你有没有写过多智能体，跑通了但三个月后不敢动它？代码能跑和代码可维护是两件事。模式的意义，就是让系统"能被后人读懂、能被安全地改"。

模式一：角色隔离（Role Isolation）

对应 1.3 的临界点——当任务里出现需要隔离的不同角色纪律，拆成独立智能体。招聘初筛（筛选员客观、沟通员礼貌）是典型。

# 角色隔离：客观与礼貌分两个智能体，互不污染上下文
screener = ReActAgent(name="筛选员", sys_prompt="只输出结构化评分，严格客观。",
                      model=model, toolkit=tk)
communicator = DialogAgent(name="沟通员",
                            sys_prompt="基于评分写礼貌邮件，不掺主观评价。",
                            model=model)
# 两者经 MsgHub 协作，筛选员产评分消息，沟通员订阅后写邮件
​

运行说明：筛选员和沟通员各管各的上下文，客观与礼貌不互相稀释。这是 2.2 边界隔离思想的直接应用，也是复用率最高的模式。

模式二：故障即消息（Failure-as-Message）

对应 3.6 心法——任何失败都表达成消息，让系统协商而非崩溃。工具返回 ERROR 文本、环境广播失败事件，都是这模式的落地。

# 故障即消息：工具失败返回文本而非抛异常
@tool
def safe_call(retry: int = 3) -> str:
    for i in range(retry):
        try:
            return _do()
        except Exception:
            if i == retry - 1:
                return "ERROR: 调用失败"   # 上游智能体据此协商
    return "ERROR"
​

模式三：单一事实来源（Single Source of Truth）

对应 2.3——共享状态和规则放进环境，不散落智能体。改规则不动业务。

# 规则进环境：价格上限校验集中在 MarketEnvironment，智能体只管策略
class MarketEnvironment(EnvironmentBase):
    def quote(self, agent, price):
        if price > self.state["price"] * 1.2:
            return "REJECT: 超现价 20%"
        self.state["price"] = price
        return "OK"
​

模式四：主备切换（Primary-Backup）

对应 3.6 故障隔离——关键路径放备用智能体，主失败由 MsgHub 广播触发切换。

#     await backup()
​

四种模式的关系图

案例：用四个模式重做客服系统

背景：1.2/3.6 提的客服系统，初版易崩、难维护。

操作：①角色隔离——接待/查单/退款拆三智能体；②故障即消息——查单工具返回 ERROR 文本；③单一事实来源——订单状态进环境而非某智能体记忆；④主备切换——查单挂时备用话术接管。

结果：系统可维护（规则在环境）、敢上生产（故障被协商）、易扩展（加角色不改旧代码）。

解读：四个模式不是孤立的，它们在同一个系统里各管一摊，合起来就是"生产级多智能体"的雏形。你新建系统，照这四格自检一遍，能避开绝大多数经典坑。

变式：小项目不必全用。若只有两个智能体、无共享状态，角色隔离和主备就够，单一事实来源可能过度设计。模式是武器，不是清单。

六之一、模式五：工厂模式（Factory）——集中生产智能体

当系统里智能体种类多、构造参数杂，散落在业务里 new 一堆 ReActAgent 既难读又难改。工厂模式把"造哪个、怎么造"收口到一个函数，业务只说"我要一个查单智能体"。

from agentscope.agents import ReActAgent, DialogAgent

def make_agent(kind: str, model, toolkit):
    """按类型造智能体，构造细节收口在工厂，业务不碰参数。"""
    if kind == "screener":
        return ReActAgent(name="筛选员",
                          sys_prompt="只输出结构化评分，严格客观。",
                          model=model, toolkit=toolkit)
    if kind == "communicator":
        return DialogAgent(name="沟通员",
                           sys_prompt="基于评分写礼貌邮件，不掺主观。",
                           model=model)
    if kind == "cashier":
        return ReActAgent(name="收银员",
                          sys_prompt="核对金额并完成结算，输出凭证。",
                          model=model, toolkit=toolkit)
    raise ValueError(f"未知智能体类型：{kind}")


# '收银员'
​

运行说明：加新角色时只改 make_agent 一个分支，业务层调用完全不动。这把"角色隔离"模式的造物过程也收口了——避免隔离出来的智能体在十处被不同姿势构造，后续统一升级提示词或模型时一地改全漏。工厂是角色隔离模式的天然搭档。

六之二、模式六：策略模式（Strategy）——可替换的行为算法

多智能体常遇到"同一种决策，不同场景用不同算法"。比如路由智能体决定把消息发给谁，高峰期走"负载均衡"，低峰走"就近"。策略模式把这些算法各自封装，运行时切换。

class RouteStrategy:
    """路由策略接口：输入候选智能体列表，输出被选中的那个。"""
    def select(self, candidates):
        raise NotImplementedError

class LoadBalanceStrategy(RouteStrategy):
    def __init__(self):
        self._idx = 0
    def select(self, candidates):
        pick = candidates[self._idx % len(candidates)]
        self._idx += 1
        return pick                      # 轮流，均衡压力

class AffinityStrategy(RouteStrategy):
    def select(self, candidates):
        # 简单演示：按名字稳定哈希到固定智能体，保证会话粘性
        return min(candidates, key=lambda a: hash(a.name) % 1000)

# ['A', 'B', 'C']          # 轮流均衡
​

运行说明：编排层持有 RouteStrategy 引用，高峰注入 LoadBalanceStrategy、低峰注入 AffinityStrategy，切换无需改路由逻辑本身。这把"换算法"从改 if-else 变成换对象，符合开闭原则——新增策略只加类，不动旧代码。

六之三、模式七：观察者模式（Observer）——事件驱动的兜底联动

3.6 的"故障即消息"在模式层就是观察者：失败事件被广播，多个订阅者各自响应（告警、降级、人工介入），发布者完全不认识它们。

class FailureHub:
    """极简观察者：失败事件广播，订阅者各自响应。"""
    def __init__(self):
        self._subs = []
    def subscribe(self, fn):
        self._subs.append(fn)
    def publish(self, event: str):
        for fn in self._subs:
            fn(event)                    # 每个订阅者独立处理

# 订阅者：告警、降级、人工，三者互不知情
hub = FailureHub()
hub.subscribe(lambda e: print(f"[告警] 收到失败：{e}"))
hub.subscribe(lambda e: print(f"[降级] 切缓存答案：{e}"))
hub.subscribe(lambda e: print(f"[人工] 建工单：{e}"))

# [人工] 建工单：查单接口超时
​

运行说明：发布 查单接口超时 后，三个订阅者各做各的，发布者 FailureHub 不需要知道有几个订阅者、它们干嘛。这正是主备切换和故障即消息能"解耦协作"的底层机制——加新订阅者（比如埋点统计）只 subscribe 一行，不碰现有逻辑。

六之四、七模式自检清单

把前面七个模式摆一起，它们解决的是不同维度的坑：

• 角色隔离 / 工厂：解决"智能体怎么拆、怎么造"（结构维度）。
• 故障即消息 / 观察者：解决"出了事系统怎么协商"（容错维度）。
• 单一事实来源：解决"共享状态放哪"（状态维度）。
• 主备切换：解决"关键路径断了怎么办"（可用性维度）。
• 策略模式：解决"同场景不同算法怎么换"（行为维度）。

新建系统按"结构—状态—行为—容错—可用性"五格过一遍，每格挑对应模式，基本能避开大多数经典坑。模式之间可叠加，但不必贪多——两智能体小项目，角色隔离加主备足矣。

本节要点回顾

• 角色隔离：不同角色纪律拆独立智能体，防上下文互相污染（复用率最高）。
• 故障即消息：失败表达成消息让系统协商，而非崩溃。
• 单一事实来源：共享状态与规则进环境，改规则不动业务。
• 主备切换：关键路径放备用，主失败广播触发接管。
• 四模式各管一摊，合起来是生产级雏形；小项目按需取用，别过度设计。

⚠️ 模式是武器不是清单，小项目别全上。两个智能体、无共享状态，主备和隔离就够，强行套单一事实来源是过度设计，反而增复杂度。

💡 新建系统照"四格自检"：有没有角色冲突？有没有单点故障？有没有散落规则？有没有关键路径无兜底？每格中招就上对应模式。