第5章 安全与可扩展

学习目标

• 掌握多 Agent 系统的监控、日志与全链路追踪设计
• 理解负载均衡、容错与水平扩展的实现方式
• 能够设计故障恢复机制，保障系统在异常下仍可用
• 识别多 Agent 系统的安全风险并实施隔离与防护
• 使用 Python 实现带监控与容错的多 Agent 系统

问题导入

某企业上线了一个 5 Agent 协作系统，运行一周后出现诡异问题：某 Agent 偶发超时导致整个流水线卡死；某次某 Agent 被注入恶意 Prompt 后疯狂调用付费工具，单日烧掉数万元；日志分散在各 Agent，故障时无法定位是哪个环节出错。这些问题表明：多 Agent 系统的"能跑"与"能上线"之间隔着监控、容错、安全三座大山。本章讲解如何让多 Agent 系统安全、稳定、可扩展地运行在生产环境。

核心概念

1. 监控与日志

多 Agent 系统的监控比单 Agent 复杂——需追踪 Agent 间的调用关系与消息流。

三层监控体系：

层级	监控内容	工具示例
业务层	任务成功率、产出质量、用户满意度	自研看板、Prometheus
Agent 层	每个 Agent 的调用量、延迟、token 消耗、错误率	LangSmith、Langfuse
基础设施层	CPU、内存、队列积压、API 限流	Grafana、Datadog

结构化日志规范：每条日志应含 trace_id（链路）、agent_id（Agent）、task_id（任务）、msg_id（消息）、timestamp、level、event、payload。这样能按 trace_id 串联一次完整的多 Agent 协作链路。

全链路追踪：借鉴分布式追踪（OpenTelemetry），为每个用户请求生成 trace_id，Agent 间传递时继承，形成调用树（span）。这是排查"哪个 Agent 慢/错"的关键。

2. 负载均衡与容错

负载均衡：当同类 Agent 有多副本时，需把请求分摊到各副本： - 轮询（Round Robin）：依次分配，简单公平。 - 最少连接（Least Connections）：分给当前负载最轻的，更智能。 - 一致性哈希：按请求 key 哈希分配，便于缓存命中。

容错策略： - 超时重试：调用超时后重试（带指数退避），避免雪崩。 - 熔断（Circuit Breaker）：错误率超阈值时暂停调用，防止级联故障。 - 降级（Fallback）：主 Agent 不可用时切换备用 Agent 或返回默认值。 - 冗余部署：关键 Agent 多副本，单点故障不影响整体。

3. 水平扩展

垂直扩展（加机器配置）有上限，水平扩展（加 Agent 副本）是多 Agent 系统的核心扩展方式。

水平扩展前提： - Agent 无状态化：状态存外部（Redis/数据库），Agent 进程可随时重启。 - 消息队列解耦：通过队列削峰填谷，Agent 按消费速度拉取。 - 服务发现：Agent 副本动态注册与发现（如 Consul、K8s Service）。 - 共享存储：Agent 共享上下文与记忆（如向量库、Redis）。

扩展模式： - 读写分离：查询类 Agent 多副本，写入类 Agent 单副本。 - 分片（Sharding）：按用户/任务 key 分片到不同 Agent 组。 - 弹性伸缩：根据队列长度自动增减 Agent 副本（K8s HPA）。

4. 故障恢复

故障类型	恢复策略
Agent 进程崩溃	K8s 自动重启 + 队列消息持久化重投
LLM API 限流/故障	切换备用模型（如 GPT→Claude）+ 降级提示
消息队列宕机	主备切换 + 持久化消息不丢
部分结果错误	检查点（Checkpoint）回滚到上一正确状态
死循环	最大轮次 + 超时强制终止 + 告警

检查点机制：在多步协作的关键节点保存中间状态（如 LangGraph 的 checkpoint）。故障后从最近的检查点恢复，而非从头开始，这对长流程尤其重要。

5. 安全风险与隔离

多 Agent 系统放大了单 Agent 的安全风险——一个 Agent 被攻破可能波及整个系统。

主要安全风险：

风险	描述	防护手段
Prompt 注入	恶意输入劫持 Agent 行为	输入过滤、Guardrails、隔离指令
工具滥用	Agent 被诱导调用付费/危险工具	权限分级、调用审批、配额限制
信息泄露	Agent 把敏感信息传给其他 Agent 或外部	数据脱敏、最小权限、出站过滤
Agent 间投毒	被攻破的 Agent 向其他 Agent 发恶意消息	消息校验、签名、零信任
资源耗尽	Agent 失控消耗算力/资金	配额、速率限制、预算熔断

安全隔离原则： - 最小权限：每个 Agent 只授予完成任务所需的最小工具与数据权限。 - 零信任：Agent 间消息不默认信任，校验来源与内容。 - 沙箱执行：Agent 执行代码/工具在隔离环境（容器、沙箱）中，限制文件与网络访问。 - 审计留痕：所有工具调用、Agent 间消息记录可审计。 - 人工兜底：高危操作（如删除、支付）需人工确认（human-in-the-loop）。

6. 可扩展架构设计原则

1. 无状态优先：Agent 进程无状态，状态外部化。
2. 队列解耦：Agent 间通过队列通信，便于扩展与容错。
3. 能力声明：Agent 声明能力，调度器动态匹配。
4. 分级缓存：缓存中间结果，减少重复 LLM 调用。
5. 渐进扩展：从单副本起步，按指标渐进扩容，避免过度设计。

原理图

flowchart TB
    User[用户请求] --> GW[API 网关\n限流/鉴权]
    GW --> LB[负载均衡器]

    LB --> A1[Agent 副本 1]
    LB --> A2[Agent 副本 2]
    LB --> A3[Agent 副本 3]

    A1 --> MQ[(消息队列\n持久化)]
    A2 --> MQ
    A3 --> MQ

    MQ --> B1[Worker Agent 1]
    MQ --> B2[Worker Agent 2]

    subgraph 监控与追踪
        A1 -.指标.-> Mon[监控系统\nPrometheus/Grafana]
        A2 -.日志.-> Log[日志中心\nELK]
        A3 -.追踪.-> Tr[链路追踪\nJaeger/LangSmith]
    end

    subgraph 容错与恢复
        CP[(检查点存储)] --> B1
        CP --> B2
        B1 -.熔断.-> CB[熔断器]
    end

    subgraph 安全隔离
        Sand[沙箱容器] --> B1
        Perm[权限策略] --> B1
        Audit[审计日志] --> B1
    end

    B1 --> Result[最终结果]
    B2 --> Result
    Result --> User

图：生产级多 Agent 系统的可扩展架构，含负载均衡、监控、容错与安全隔离

示例代码

实现一个带监控、容错与安全配额的多 Agent 调度基类。

import time
import logging
from dataclasses import dataclass, field
from collections import defaultdict
from functools import wraps

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("mas")


def traced(func):
    """装饰器：记录 Agent 调用的追踪日志。"""
    @wraps(func)
    def wrapper(self, trace_id, *args, **kwargs):
        start = time.time()
        try:
            result = func(self, *args, **kwargs)
            elapsed = (time.time() - start) * 1000
            logger.info(f"trace={trace_id} agent={self.agent_id} "
                        f"event=ok elapsed_ms={elapsed:.0f}")
            self.metrics["success"] += 1
            return result
        except Exception as e:
            elapsed = (time.time() - start) * 1000
            logger.error(f"trace={trace_id} agent={self.agent_id} "
                         f"event=error elapsed_ms={elapsed:.0f} err={e}")
            self.metrics["error"] += 1
            raise
        finally:
            self.metrics["latency_ms"].append(elapsed)
    return wrapper


@dataclass
class CircuitBreaker:
    """熔断器：错误率超阈值时暂停调用。"""
    threshold: float = 0.5
    window: int = 10
    failures: int = 0
    calls: int = 0
    open_until: float = 0

    def allow(self) -> bool:
        if time.time() < self.open_until:
            return False  # 熔断中
        return True

    def record(self, success: bool):
        self.calls += 1
        if not success:
            self.failures += 1
        if self.calls >= self.window:
            rate = self.failures / self.calls
            if rate > self.threshold:
                self.open_until = time.time() + 30  # 熔断 30 秒
                logger.warning(f"熔断器开启，错误率 {rate:.0%}")
            self.calls = 0
            self.failures = 0


@dataclass
class QuotaGuard:
    """配额守卫：限制 Agent 的工具调用次数与 token 消耗。"""
    max_calls: int = 100
    max_tokens: int = 100_000
    calls: int = 0
    tokens: int = 0

    def check(self, est_tokens: int = 0):
        if self.calls >= self.max_calls:
            raise RuntimeError(f"调用次数超限 {self.max_calls}")
        if self.tokens + est_tokens >= self.max_tokens:
            raise RuntimeError(f"token 预算超限 {self.max_tokens}")

    def consume(self, tokens: int = 0):
        self.calls += 1
        self.tokens += tokens


class MonitoredAgent:
    """带监控、熔断、配额的多 Agent 基类。"""

    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.metrics = defaultdict(lambda: 0, {"latency_ms": []})
        self.breaker = CircuitBreaker()
        self.quota = QuotaGuard()

    @traced
    def run(self, task: str, est_tokens: int = 500) -> str:
        """执行任务，含熔断与配额检查。"""
        if not self.breaker.allow():
            return f"[{self.agent_id}] 降级：熔断中"
        self.quota.check(est_tokens)
        try:
            # 模拟执行（实际调用 LLM）
            result = self._execute(task)
            self.quota.consume(est_tokens)
            self.breaker.record(success=True)
            return result
        except Exception as e:
            self.breaker.record(success=False)
            raise

    def _execute(self, task: str) -> str:
        """子类重写：实际业务逻辑。"""
        return f"[{self.agent_id}] 处理: {task}"


class FallbackOrchestrator:
    """带降级的主备 Agent 编排。"""

    def __init__(self, primary: MonitoredAgent, backup: MonitoredAgent,
                 max_retries: int = 2):
        self.primary = primary
        self.backup = backup
        self.max_retries = max_retries

    def execute(self, task: str, trace_id: str) -> str:
        for attempt in range(self.max_retries + 1):
            try:
                return self.primary.run(task)
            except Exception as e:
                logger.warning(f"trace={trace_id} 主 Agent 失败 "
                               f"attempt={attempt} err={e}")
                if attempt == self.max_retries:
                    logger.info(f"trace={trace_id} 降级到备用 Agent")
                    return self.backup.run(task)
                time.sleep(2 ** attempt)  # 指数退避
        return self.backup.run(task)


if __name__ == "__main__":
    primary = MonitoredAgent("primary-001")
    backup = MonitoredAgent("backup-001")
    orch = FallbackOrchestrator(primary, backup)

    # 正常执行
    print(orch.execute("分析销售数据", "trace-001"))

    # 模拟主 Agent 故障，降级到备用
    primary._execute = lambda t: (_ for _ in ()).throw(RuntimeError("LLM 限流"))
    print(orch.execute("生成周报", "trace-002"))

    # 查看监控指标
    print("主 Agent 指标:", dict(primary.metrics))
    print("备用 Agent 指标:", dict(backup.metrics))

企业案例

案例 1：客服系统的弹性伸缩

某客服多 Agent 系统部署在 K8s，根据 Redis 队列长度自动扩缩容：队列积压 >100 时自动增加 Agent 副本，<20 时缩容。大促期间 Agent 副本从 5 个自动扩到 30 个，平稳应对 6 倍流量峰值，成本仅增加 2 倍。

案例 2：投研系统的检查点恢复

某投研系统用 LangGraph 的 checkpoint 机制，在每个 Agent 完成后保存中间状态。一次 LLM API 故障导致第 4 个 Agent 失败，系统从第 3 个 Agent 的检查点恢复，无需重跑前 3 步，节省 70% 的重试成本与 token。

案例 3：Agent 工具滥用的安全防护

某团队发现某 Agent 被 Prompt 注入后疯狂调用付费搜索工具。事后加固：①工具调用加配额（每 Agent 每日 50 次）；②高危工具（支付/删除）需人工确认；③Agent 间消息加签名校验；④所有工具调用记审计日志。加固后再未发生滥用事件。

常见问题

Q1：多 Agent 系统必须用 K8s 吗？ A：不必。小规模可用单机多进程 + Redis 队列；中等规模可用 Docker Compose；大规模、需弹性伸缩才推荐 K8s。避免过度设计。

Q2：Agent 无状态化是不是很难？ A：取决于业务。把对话历史、中间结果存 Redis/数据库，Agent 进程只负责处理当前消息，即可实现无状态。LangGraph 的 checkpoint 已内置此能力。

Q3：如何防止一个 Agent 被攻破后波及全网？ A：①零信任——Agent 间消息校验签名；②最小权限——每个 Agent 只能用必要的工具；③网络隔离——Agent 运行在独立容器/网络命名空间；④出站过滤——限制 Agent 可访问的外部地址。

Q4：监控应该关注哪些核心指标？ A：①任务成功率与平均延迟；②每个 Agent 的 token 消耗与成本；③队列积压长度；④错误率与熔断状态；⑤工具调用次数与配额使用率；⑥端到端用户满意度。

面试题

1. 设计一个支持水平扩展与容错的生产级多 Agent 架构。 答题要点：Agent 无状态化 + 多副本；消息队列解耦 + 持久化；负载均衡（最少连接）；熔断器 + 降级 + 重试；检查点恢复；K8s 弹性伸缩；全链路追踪；最小权限 + 沙箱 + 审计。

2. 多 Agent 系统如何实现故障恢复？ 参考答案：①消息队列持久化，进程崩溃后重投；②检查点保存中间状态，从最近检查点恢复；③熔断器防止级联故障；④主备 Agent 降级切换；⑤LLM 故障时切换备用模型；⑥最大轮次与超时兜底死循环。

3. 多 Agent 系统有哪些安全风险？如何防护？ 参考答案：Prompt 注入（Guardrails+过滤）、工具滥用（配额+审批）、信息泄露（脱敏+最小权限）、Agent 间投毒（签名+零信任）、资源耗尽（速率限制+预算熔断）。核心原则：最小权限、零信任、沙箱执行、审计留痕、人工兜底。

练习题

1. 监控实操：为本章的 MonitoredAgent 增加一个 export_metrics() 方法，输出 Prometheus 格式的指标（agent_success_total、agent_latency_ms 等）。
2. 容错设计：实现一个"主→备→默认值"三级降级链：主 Agent 失败转备用，备用也失败则返回预设默认结果，保证用户总能拿到响应。
3. 安全加固：为一个"可执行用户上传代码"的 Agent 设计沙箱方案（提示：Docker 容器、CPU/内存限制、无网络），列出隔离措施清单。

小结

安全与可扩展是多 Agent 系统从"原型"到"生产"的关键跨越。监控（三层体系+全链路追踪）是可观测的基础；负载均衡与水平扩展应对流量增长；容错（熔断/降级/检查点）保障可用性；安全（最小权限/零信任/沙箱/审计）防护各类风险。核心原则：无状态优先、队列解耦、渐进扩展、零信任安全。生产级系统的设计不是"能跑"，而是"在异常下仍可控、可恢复、可审计"。本卷至此结束，建议结合第 7 卷 MCP 安全、第 8 卷 LangGraph、第 12 卷部署运维综合实践。延伸阅读：OpenTelemetry 文档、《Site Reliability Engineering》、《Building Secure Multi-Agent Systems》。