multi-agent langgraph orchestration rate-limiting production

通过三层契约机制应对多智能体系统中的流量激增问题

在流量骤增十倍的情况下保持多智能体系统正常运行：队列管控机制、针对各工具的流量限制以及同步策略检查，再加上相关指标与流量激增演示。

发布者：REACTAPP.TOP加入本站 2026年10月28日阅读约 8 分钟1869 词

来源: Multi-Agent Orchestration in Production: Handling Traffic Spikes at Scale

代理图在每秒10次请求时仍能正常运行，但一旦达到100次请求就会崩溃。请求不断堆积，策略检查被跳过，各种工具调用相互冲突，原本可靠的系统反而变成了负担。问题往往不出在代理本身，而是缺少队列、编排层与约束机制之间的明确约定。大多数团队都是在实际流量已经受到影响时才意识到这一缺陷。

为何流量峰值会比预期更早出现

代理系统已从演示阶段进入实际生产环境，因此流量峰值在所难免：限时促销、某个功能迅速走红、定时批处理任务，或是简单地在周二上午所有用户同时访问API。随着负载增加，两件事会同时发生。

首先，路由需要实时处理。决定由哪个代理处理请求，或是判断是否立即将其放入队列，都属于需要量化的工作。

其次，系统在压力下往往会出现异常开放的情况。这通常并非是惊慌失措的工程师绕过了检查机制，而是由于错误的错误处理方式——在超时时默认允许操作，或是采用会悄悄跳过验证中间件的备用路由策略。当负载增加导致策略执行失效时，这类架构缺陷往往是根本原因，而非人为干预。

解决之道在于让三个组成部分之间达成明确且可执行的协议：队列（处理什么请求）、编排图（请求去往何处）以及策略引擎（允许执行什么操作）。尽早确定这些协议，就能将突发流量问题转化为可解决的常规问题，而非紧急状况。

三层架构协议

第一层：作为守门人的队列

队列是执行规则的第一道关卡。在任何请求到达处理节点之前，都会先经过一项涵盖三方面的协议检查：

优先级。不同的请求并非同等重要。交互式用户查询的优先级可能高于批量任务，付费客户的优先级也会高于免费用户。队列系统应负责处理这种排序，这样客服人员无需了解相关细节。
速率限制与配额。在消耗任何计算资源之前直接拒绝请求是最经济的方式。达到每日API调用限制的用户应得到明确的拒绝答复，而非进入排队队列，因为这比让客服处理要便宜得多。
架构遵循性。系统无法处理的输入数据应在入口处就被拒绝，而不应在客服处理过程中才发现。

这一层的作用是确保队列中的每个请求都经过验证、排序并获得使用资源的许可。

第二层：编排系统决定路由与处理速度

请求被放入队列后，调度系统会决定其处理目标以及该请求可执行的操作。并发控制和工具访问权限均在这一环节进行管理。

路由契约需要回答几个问题：例如，简单查询应交由快速处理的代理处理，而复杂推理则由较慢的代理处理，那么该请求应由哪组代理负责？每个代理同时能处理多少个请求？当所有代理都处于忙碌状态时，应如何处理——排队等待、拒绝请求还是采取备用方案？

工具限流契约考虑到不同工具的特性差异。每秒调用一百次外部API与运行本地嵌入模型是完全不同的情况。每种工具都有各自的并发限制：基于数据库的工具可能允许同时进行十次调用，而需要访问第三方API的工具则可能仅允许五次。当某工具达到其并发上限时，相关请求会在该工具的专用队列中等待，而不会进入主队列。

为保持这些工具队列的稳定并防止故障级联，生产级基础组件可提供帮助：

电路断路器与令牌桶在工具层即可快速触发故障，从而避免在依赖项饱和时出现重试风暴。
死信队列用于存放格式错误或损坏的工具请求数据，这样单个异常请求就不会导致整个代理循环停止。
为每个工具单独设置的泄漏桶机制可应用该工具自身的速率限制，从而避免并行运行的代理受到外部限制的影响。

有了这些机制，工具层的瓶颈会被控制在可控范围内且清晰可见，不会蔓延为整个系统的不稳定。该层的承诺是：请求会被发送到有空闲容量的代理，每个工具都会遵守其并发上限，而如果某个工具已饱和，请求将等待该工具而非代理。

第三层：永不妥协的策略

难点在于这些策略必须在每次执行时都运行，包括在高负载情况下。不存在“稍后再检查”或“这个看起来没问题”的情况。

策略合约包含四个条款：

在任何工具调用之前，都会同步且阻塞地执行策略检查。
该检查的成本很低，最多只需几毫秒。如果耗时过长，说明设计有误。
一旦检查失败，请求就会失败，代理没有机会重试或协商。
所有失败情况都会被记录并监控，策略失败次数的上升本身也是一种信号。

典型的策略如下：代理只能使用为该用户标记为只读的工具；某个工具只能由已通过身份验证的调用者使用；代理每次请求最多可调用五种工具。该层保证每次工具调用之前都会进行检查，并且这些策略具有快速、确定且绝对的特性。

处理流程的结构

在最简的伪图中，请求首先进入队列守门人，然后传递给路由器，最终到达代理；而每次工具调用在执行之前都要经过策略验证以及针对该工具的调用限制检查。

[ Incoming Request ]
         │
         ▼
[ Queue Gatekeeper ]
  • Validate Schema
  • Enforce Rate Limits
  • Assign Priority
         │ (Pass)
         ▼
[ Orchestration Router ]
  • Check Agent Capacity
  • Route to Target Node
         │
         ▼
[ Tool-Level Gate ]
  • Sync Policy Check
  • Per-Tool Concurrency Limits
  • Circuit Breaker / DLQ
         │ (Pass)
         ▼
[ Tool Execution ]

关键在于策略的适用范围。它们在工具层面生效，而非代理层面，这样就能防止代理通过以意外方式使用工具而绕过这些策略。

有合同与无合同的流量激增场景

想象一下，一名客户支持代理能够查询订单、处理退款并发送邮件，而此时流量突然增加了十倍。

没有合同的情况下，故障会分步出现：

涌入的请求耗尽了可用的代理资源。
代理开始排队调用工具，send_email工具积累了500个待处理请求。
邮件服务开始出现超时现象。
代理重新尝试，此时共有1,000个请求在等待处理。
故障开放路径被触发。为了保持系统可用，超时或过载的备用路由会绕过策略限制继续工作。

客服人员将退款邮件发送给了错误的客户。

在有合同约束的情况下，同样的情况会呈现不同的处理方式：

队列管理器会根据优先级和配额来控制任务处理顺序。
路由器检测到邮件发送工具的并发处理上限已达到10次，因此会将后续的邮件请求排队到该工具而非客服人员处。
每次调用工具时都会进行策略检查。如果客服人员试图向错误客户发送退款，检查会失败，请求将被拒绝。
系统依然保持稳定。虽然延迟会增加，但可靠性不会下降。

综合分析

从LangGraph开始

已经在 LangGraph 上的团队掌握了大部分基础组件。智能体和工具是节点，边代表路由选择，而条件边则根据状态进行路由。需要添加的是一个前置队列，可以是 Redis、RabbitMQ 或简单的内存队列，此外还需要为工具调用封装策略中间件。

该策略封装层本身可以是一个小型装饰器，在工具函数执行之前先进行检查，若发现违规则抛出错误。

from functools import wraps
from typing import Callable, Any

class PolicyViolation(Exception):
    """Raised when policy check fails."""
    pass

def enforce_tool_policy(
    policy_fn: Callable[[str, dict], bool]
):
    """Sync tool execution guardrail."""
    def decorator(func: Callable):
        @wraps(func)
        def wrapper(*args, **kwargs):
            tool_name = func.__name__
            context = kwargs.get(
                "request_context", {}
            )

            # Must fail closed
            if not policy_fn(tool_name, context):
                raise PolicyViolation(
                    f"Denied: {tool_name}"
                )

            return func(*args, **kwargs)
        return wrapper
    return decorator

# Example Usage
def strict_policy_check(tool_name: str, context: dict) -> bool:
    if not context.get("is_authenticated"):
        return False

    if (context.get("role") != "admin" and tool_name.startswith("execute_")):
        return False

    return True

@enforce_tool_policy(strict_policy_check)
def execute_refund(
    order_id: str,
    amount: float,
    request_context: dict = None
):
    # Downstream API call
    return f"Refund ${amount} sent: {order_id}"

进行测量

仅仅实施并寄予希望是不够的。需要通过工具进行监控：

各层的队列深度。
各类工具达到处理极限的频率及具体情况。
策略是否在拒绝请求，以及原因是什么。
针对不同请求类型分别统计的p50、p95和p99延迟值。
负载下的每请求成本。如果该成本上升，说明存在效率问题。

怀疑的观点

可以说，对于一个尚未出现的问题而言，这似乎是过度的基础设施投入。一个每秒只能处理十次请求且不会增长的系统并不需要这些设施。而对于可能迅速走红的功能，或流量难以预测的企业客户，则情况不同，这类设计应在产品发布前就完成。

提前构建的成本远低于在系统因流量激增而崩溃后再进行改造。同时也不应过度设计。一个简单的队列、一个策略检查函数以及针对不同工具的并发限制就能解决大多数流量激增问题。既不需要分布式共识算法，也不需要机器学习负载预测器。重要的是清晰性和约束条件。

要点

流量激增会暴露设计缺陷。当代理系统在负载过高时出现故障，并非因为代理本身的处理速度过慢，而是由于没有为队列、路由和策略之间的协作制定明确的规则。
队列同时兼具执行控制的功能。它能够验证请求、确定优先级并拒绝不必要的请求，从而避免计算资源浪费。
策略是绝对且不可更改的。它们会在工具层面同步运行，每次都严格执行，没有任何例外。

应在智能体之前对工具进行限流处理。瓶颈通常出现在为智能体提供上下文的外部 API 或数据库中，而非 LLM 提供商端。

需对所有指标进行测量：队列深度、工具并发数、策略失败情况、延迟百分位数以及每次请求的成本。无法测量的内容就无法得到改进。

从何处开始

明确队列规则：哪些请求会被接受，哪些会被拒绝。
明确路由规则：哪些智能体负责处理哪些任务，以及并发限制是多少。
明确策略规则：每次调用工具之前需要执行哪些检查。
使用 LangGraph 等工具构建系统，加入队列层，并在工具调用前加入策略检查。
进行压力测试。模拟流量增加十倍的情况，观察系统会出现哪些问题。

所有这些工作最好在正式上线之前完成，那时还有时间从容处理。