CodeNodeX AI
注册写作
#AI Agent#DevOps

Agent 的并发与调度

单个 Agent 跑得通,不等于能服务多用户、多任务。本文讲清 Agent 服务的并发模型(每请求一 Agent vs Agent 池)、资源管理(模型 API 并发限/工具并发)、调度(优先级/队列/限流)、多用户状态隔离,把 Agent 从"单线程 demo"推到"能扛流量的服务"。

本文要点

  • 主题聚焦：AI Agent、DevOps
  • 适合人群：独立开发者、全栈工程师、AI 应用开发者
  • 核心价值：快速理解背景、实践路径、常见误区和工程建议
作者 @admin·发布于 2025年12月11日·更新于 2026年7月23日

把一个 Agent 跑通,在今天已经不难。随便一个 LangGraph、OpenAI Agents SDK 的 demo,本地 loop 一遍,工具调通了,看着就像"能上线"。

但 demo 和服务之间,隔着一座山:并发。

单个 Agent 是个长任务——动辄几十秒、几十步,占着模型连接、占着工具、占着内存里的 KV cache 和会话状态。一旦同时来了 100 个用户,模型 API 的 RPM/TPM 撞顶、数据库连接池打满、队列堵死、付费用户的请求排在一个写垃圾 prompt 的免费用户后面……这些问题 demo 永远遇不到,生产环境里每天都会要你的命。

这篇文章讲清楚:Agent 服务到底怎么扛流量。

一、先看清 Agent 并发的"难"在哪

普通 Web 请求是无状态的、毫秒级的、一次性的。Agent 完全相反:

  • 长生命周期:一次对话可能跑 30 秒到 5 分钟,中间多轮模型调用 + 多次工具调用。
  • 有状态:Agent 持有会话记忆、工具句柄、中间 plan,不能像 HTTP 一样"用完即弃"。
  • 重资源:占着一条到模型 API 的并发通道、占着工具连接(数据库、浏览器、Sandbox)、占着 GPU/KV cache(自托管模型时)。
  • 不可中断的中间态:工具调用到一半被 kill,可能留下脏数据、半完成的副作用。

所以"并发"这件事在 Agent 服务里,本质上是在多个维度上同时抢资源:模型 API 的并发额度、工具侧的并发额度、进程/内存。任何一维打满,服务就崩。

一个残酷的事实:你买的是模型 API 的 1000 RPM,不是 1000 个 Agent 并发。一个 Agent 一次任务可能打 8 次模型 API,所以你的真实并发 Agent 数大概是 RPM / 平均调用轮数 / 用户峰值倍数。算明白这一笔账,后面所有的设计才有的放矢。

二、并发模型:每请求一 Agent vs Agent 池

两种主流模型,各自有坑。

模型 A:每请求创建一个 Agent(Per-request Agent)

每次请求 new Agent(),跑完销毁。最简单、最不容易出错,状态天然隔离。

python
async def handle_turn(session_id: str, user_msg: str):
    agent = Agent(
        llm=build_llm(),
        tools=build_tools(),
        memory=await load_memory(session_id),
    )
    try:
        return await agent.run(user_msg)
    finally:
        await agent.close()  # 释放工具连接、刷盘记忆

适用:会话短、工具轻、无重预热成本的场景(客服、问答、单轮 RAG)。缺点是每次都要重建工具连接、重载系统提示,冷启动成本高。

模型 B:预热 Agent 池(Agent Pool)

预先 warm 一批 Agent 常驻,请求来了从池子里借一个用完还回去。适合工具重(浏览器、Sandbox、数据库长连接)、系统提示长(重建一次要几万 token)的场景。

typescript
class AgentPool {
  private idle: AsyncQueue<Agent> = new AsyncQueue();
  private inUse = new Set<Agent>();

  constructor(size: number) {
    for (let i = 0; i < size; i++) {
      this.idle.push(warmAgent()); // 预热:加载工具、建立长连接
    }
  }

  async acquire(userId: string, timeout = 30_000): Promise<Agent> {
    const agent = await this.idle.pop(timeout); // 满了就排队/拒绝
    if (!agent) throw new PoolExhaustedError();
    this.inUse.add(agent);
    return agent.bindContext(userId); // 关键:每次绑定用户上下文
  }

  release(agent: Agent) {
    agent.resetContext(); // 关键:用完必须清状态
    this.inUse.delete(agent);
    this.idle.push(agent);
  }
}

池化模型最大的坑,是状态串台。 一个 Agent 上一个用户刚问完"我的订单",下一个用户接手,如果 resetContext 没做干净,下一个用户会看到上一个用户的订单。这是 Agent 池唯一的、也是最致命的 bug 来源。我的建议:只要团队规模允许,先上 Per-request 模型,等真出现冷启动瓶颈(监控能看到)再切池化,而且池化一定要配上完备的 reset 测试。

池子大小怎么定?取 min(模型API并发额度, 工具并发额度, 内存预算) 再打个 0.8 的安全系数。不是拍脑袋。

三、资源管理:三个独立的水龙头

Agent 服务的资源要分成三个独立的水龙头去管,任何混用都会失控。

水龙头 1:模型 API 的并发与速率

模型厂商给你的限制是分层的:RPM(requests/min)、TPM(tokens/min)、并发连接数。三者都可能成为瓶颈,而且不同模型/不同 tier 额度不同。一个常见错误:你同时调了 GPT-4o 和 Claude,以为"加起来 2000 RPM",结果 Claude 那边单独打爆了。

工程上必须有一个全局的、按 provider 维度独立的令牌桶 + 并发 semaphore:

typescript
const providerLimits = {
  openai: { rpm: 1000, tpm: 2_000_000, maxConcurrent: 50 },
  anthropic: { rpm: 500, tpm: 800_000, maxConcurrent: 30 },
};

// 每 provider 一个 semaphore
const llmSem: Record<string, Semaphore> = Object.fromEntries(
  Object.entries(providerLimits).map(([k, v]) => [k, new Semaphore(v.maxConcurrent)])
);

水龙头 2:工具的并发

工具资源比模型更容易被忽略。你的 Agent 调数据库——连接池是 20;调浏览器自动化——Sandbox 实例上限是 10;调第三方 API——对方限 5 QPS。这些必须每个工具一个独立的并发限制,不能和模型共用 semaphore。

typescript
const toolLimits = {
  postgres: new Semaphore(20),
  browser_sandbox: new Semaphore(10),
  search_api: new RateLimiter({ qps: 5 }),
};

async function callTool(name: keyof typeof toolLimits, input: unknown) {
  const guard = name === 'search_api'
    ? await (toolLimits[name] as RateLimiter).acquire()
    : await (toolLimits[name] as Semaphore).acquire();
  try {
    return await tools[name].run(input);
  } finally {
    guard.release();
  }
}

水龙头 3:内存与 KV cache

自托管模型时,KV cache 是稀缺资源——一个长会话动辄吃几百 MB 显存。Agent 服务要么在 prompt 超长时主动 truncate/compress history,要么用 prefix cache 复用公共部分。托管 API 用户也别掉以轻心:会话越长,延迟和成本都线性涨,要做 context window 预算管理。

四、调度:优先级、队列、限流、回收

资源就这么多,谁来先用?这就是调度。

优先级:付费用户优先于免费、交互式优先于后台批处理。优先级不是嘴上说,是要在队列里用 priority queue 实现的:

typescript
const queue = new PriorityQueue<Job>((a, b) => a.priority - b.priority);
// 优先级:付费实时(0) > 免费实时(10) > 后台任务(100)

队列削峰:模型 API 的额度是刚性的,但用户请求是脉冲的。队列的职责是把脉冲摊平到时间上,而不是无脑堆积。队列长度必须有上限——超过上限直接拒绝(返回 429 + Retry-After),而不是让用户等 10 分钟最后超时。让用户等 = 让用户骂街;让用户立刻知道"现在忙,稍后再试" = 还能挽留。

限流(防滥用):按 user_id 维度的并发上限 + QPS 上限,防止一个用户(或一个爬虫)把整个池子吃光。

typescript
const perUserConcurrent = new Map<string, Semaphore>();
function getUserGate(userId: string, max = 2): Semaphore {
  if (!perUserConcurrent.has(userId)) {
    perUserConcurrent.set(userId, new Semaphore(max));
  }
  return perUserConcurrent.get(userId)!;
}

超时与回收:这是 Agent 服务稳定性的最后一道闸。每个 Agent 必须有硬超时(比如 5 分钟),超时后强制 cancel——不只是取消 LLM 调用,还要回滚/标记它正在执行的工具副作用。否则一个卡死的 Agent 会一直占着池子,直到把你榨干。

python
async def run_with_reclaim(agent, task, hard_timeout=300):
    try:
        return await asyncio.wait_for(agent.run(task), timeout=hard_timeout)
    except asyncio.TimeoutError:
        await agent.cancel()          # 中断 LLM 流
        await agent.mark_orphaned()   # 标记可能未完成的工具副作用
        raise ServiceUnavailable()

五、多用户状态隔离:不能串,一串就是事故

这是 Agent 服务最容易出 P0 事故的地方,值得单独一节。

记忆、会话历史、缓存,一切按 user_id + session_id 严格分区。落到数据库就是强制的复合主键;落到 Redis 就是 key 前缀;落到内存 Map 就是从不跨用户读取。

python
# 正确:所有记忆查询都强制带 user_id
async def load_memory(user_id: str, session_id: str) -> Memory:
    return await db.fetch_one(
        "SELECT * FROM memories WHERE user_id=$1 AND session_id=$2",
        [user_id, session_id],
    )

# 错误(会串):只按 session_id 查
#   如果两个用户巧合或攻击构造了相同的 session_id,就串了

几个实战铁律:

  1. Key 永远是 (user_id, session_id),不是 session_id 单独。session_id 可伪造、可碰撞,只信 user_id 是不行的,user_id + session_id 才是安全边界。
  2. 池化 Agent 换用户时,所有上下文变量、工具内的连接句柄、缓存,必须全部 reset。养成"reset 是一等方法"的代码习惯,而不是散落各处的赋值。
  3. 工具返回里禁止出现其他用户的数据——例如一个 list_orders 工具,query 必须强制注入 WHERE user_id = current_user,不能信任 LLM 传上来的 user_id。
  4. 日志和 trace 里也别打别人的 user_id——隔离意识要贯彻到可观测层。

把状态串台当 P0 来设计,而不是 P0 来修。

六、长任务:不要让 HTTP 连接陪你等到天荒地老

Agent 任务经常超过 30 秒,而 HTTP/网关层的超时往往就是 30 秒。让用户一直 hold 着连接等结果,是反模式——连接会被网关掐断、体验极差、也无法做进度反馈。

正确的架构是:请求即入队 + 后台 Agent 执行 + 状态查询 + 进度推送。

text
Client ──POST /jobs──> [优先级队列] ──> [Worker: Agent]
           │                                   │
           └──返回 job_id                      ├──写状态到 Redis/DB
                                               │
Client <───GET /jobs/{id} (轮询)               │
Client <───SSE/WebSocket (推送进度) <──────────┘
typescript
// 入队
app.post('/jobs', auth, async (req, res) => {
  const job = await queue.enqueue({
    userId: req.user.id,
    prompt: req.body.prompt,
    priority: req.user.tier === 'paid' ? 0 : 10,
  });
  res.status(202).json({ jobId: job.id, status: 'queued' });
});

// 状态查询
app.get('/jobs/:id', auth, async (req, res) => {
  const job = await jobs.get(req.params.id, req.user.id); // 强制带 user_id
  res.json({ status: job.status, progress: job.progress, result: job.result });
});

// 进度推送
app.get('/jobs/:id/stream', auth, (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream');
  jobs.subscribe(req.params.id, req.user.id, (evt) => {
    res.write(`data: ${JSON.stringify(evt)}\n\n`);
  });
});

后台 Worker 是长期运行的进程,从队列里取 job、起 Agent、把 agent.run() 里的中间事件(step / tool_call / token)写回 Redis,前端轮询或 SSE 拿。这个架构让你同时获得了削峰、可观测、可恢复(worker 挂了 job 还在队列里)。

七、可观测:并发数、排队时长、失败率

线上没有可观测的 Agent 服务,就是在裸奔。你至少要看清这几组指标:

  • 并发指标:当前活跃 Agent 数、每 provider 的 LLM 在途请求数、每工具的在途调用数、队列深度。这些是容量规划的依据。
  • 延迟拆解:排队时长(从入队到 Agent 开始执行)、Agent 执行时长、模型调用 P95、工具调用 P95。排队时长是用户体验的第一杀手——它单独成段,不要平均进总延迟里。
  • 失败与回收:Agent 失败率(按阶段分:LLM 失败 / 工具失败 / 超时)、硬超时触发次数、429 拒绝率、池耗尽次数。
  • 成本:每次任务的 token 消耗、调用轮数分布,按 user/tier 聚合。这个直接关系到账单能不能撑住。

每条 Agent 任务必须有一个 trace ID,把 LLM 调用、工具调用、队列等待串起来。出问题时能在一屏里看到一个 Agent 到底卡在哪一步、等了多久、调了什么工具。Langfuse、OpenTelemetry、Arize 这类工具能省你大量排障时间。

一个被低估的实践:给每个 Agent 任务打 tag(plan 轮数、工具调用数、最终是否成功),长期下来你会看到哪类 prompt 最烧钱、哪类最容易超时——这些是产品迭代的弹药。

小结

  1. Agent 是长任务、重资源、有状态——它不是 HTTP 请求,不能用 Web 那套"来一个处理一个"的思路硬扛,必须从模型 API、工具、内存三个维度独立管资源。
  2. 并发模型先选 Per-request,后选池化。池化的唯一致命问题是状态串台,只有在冷启动真的成为瓶颈时才上,且必须配齐 reset 测试。
  3. 调度四件套:优先级、队列、限流、回收。队列要有上限、Agent 要有硬超时、用户要有并发上限——这三道闸任何一道缺了都会出生产事故。
  4. 状态隔离按 (user_id, session_id) 强制分区,贯穿到数据库 key、工具 SQL、日志 tag。串台是 P0,设计阶段就要堵死。
  5. 长任务走"入队 + 后台 + 推送",别让 HTTP 连接陪 Agent 一起等。这套架构顺带送你削峰和可恢复。

金句:单个 Agent 跑通是 demo,一组 Agent 调度得起来才是服务。把资源、调度、隔离、可观测这四件事做扎实,你的 Agent 才有资格谈"扛流量"。

Related

相关推荐

文章

从 Copilot 到 Codex：AI 编程工具进入 Agent 时代

围绕 AI、AI Agent、Vibe Coding 的技术趋势、工程实践和独立开发视角。

文章

MCP 为什么重要？一文看懂 AI 工具协议

围绕 MCP、AI Agent 的技术趋势、工程实践和独立开发视角。

文章

AI Agent 的核心架构：规划、记忆、工具和执行

围绕 AI Agent、大模型 的技术趋势、工程实践和独立开发视角。

文章

RAG 和 Agent 应该怎么结合？

围绕 RAG、AI Agent 的技术趋势、工程实践和独立开发视角。