1. 多任务同时跑Agent 为什么总在“打架”AI Agent Harness 说白了就是 Agent 的调度底座它负责把任务分给哪个 Agent、给多少资源、什么时候回收。你如果同时丢 50 个任务进去Harness 要保证它们不互相踩脚。但现实里很多人第一次写 Harness 就是开一个ThreadPoolExecutor(max_workers50)然后所有 Agent 共享一个全局变量、一个 API Key、一个输出目录。跑单任务没问题一上并发就开始出鬼故事。我见过最典型的三个现场第一两个 Agent 同时写同一个result.json后写的把先写的覆盖了用户 A 拿到用户 B 的答案第二某个 Agent 卡在死循环里CPU 打满同机器上其他 9 个 Agent 全部超时第三大模型 API 的 QPS 限制是 2050 个 Agent 一起冲直接触发 429重试风暴把配额烧光。这三个问题分别对应资源竞争、故障扩散、状态不一致也就是并发控制要解决的核心。这篇不聊虚的直接给你一套能在本地复现的并发配置模板用 Redis 做分布式锁和状态存储用优先级队列做调度用容器配额做资源隔离最后用一组并发请求验证隔离效果和状态一致性。适合正在写 Agent Harness、被并发问题折磨的后端和 AI 应用开发者。你不需要分布式系统背景跟着步骤敲命令就能跑起来。2. TaoToken 前置给 Harness 一个稳定的模型出口Harness 并发控制做得再好模型调用这一层如果不稳照样雪崩。多任务同时执行时模型 API 是最容易成为瓶颈的共享资源。我建议把模型出口统一收口到 TaoToken这样 Harness 只需要面对一个 Base URL配额、鉴权、模型切换都在这一层管理不用在每个 Agent 里散落不同的 Key。TaoToken 是一个模型 API 聚合入口兼容 OpenAI 风格的接口协议也就是说你原来用openaiSDK 写的调用代码改一下base_url和api_key就能用。对 Harness 工程来说它的价值在于你可以在一个地方配置模型路由Agent 侧只认一个 endpoint并发压测时也方便统计总调用量。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。你需要准备三样东西我把它叫做“三件套”后面所有配置都围绕它展开配置项说明示例值Base URL模型请求的统一入口https://taotoken.net/apiAPI Key鉴权凭证在控制台生成sk-xxxxxxxxModel ID具体调用的模型标识gpt-4o-mini或你账号下可用的模型获取 Key 的路径是登录后进入控制台在 API Keys 页面创建。这里有个细节并发场景下不要多个 Agent 共用一个 Key 去疯狂重试建议在 Harness 层做统一的请求队列和限流Key 只放在 Harness 的配置里Agent 容器通过内网调用 Harness 的转发接口。这样即使某个 Agent 出问题也不会把 Key 暴露出去或者绕过限流。如果你还没配好 Key可以先到模型对话页面验证一下模型是否可用确认返回正常再接入 Harness。地址是 https://taotoken.net/api-keys 管理密钥https://taotoken.net/doc 看接入文档。把这三件套记下来下一节的配置文件里会直接用到。3. 可复制配置Harness 并发控制模板这一节给你一份可以直接抄的配置。我用的是 Python Redis Docker 的组合核心思路是任务进优先级队列调度器按并发上限取任务取到任务后先抢分布式锁再在容器配额内执行状态全程写 Redis。先看目录结构你照着建agent-harness/ ├── config/ │ └── harness.toml ├── harness/ │ ├── __init__.py │ ├── scheduler.py │ ├── lock.py │ └── state.py ├── agent/ │ └── worker.py └── docker-compose.ymlconfig/harness.toml是核心配置文件路径和字段名你保持一致后面代码直接读它[harness] max_concurrency 10 # 同时执行的任务上限 task_timeout 60 # 单任务超时秒数 retry_limit 3 # 失败重试次数 [redis] host 127.0.0.1 port 6379 db 0 lock_expire 30 # 分布式锁过期秒数 state_ttl 86400 # 任务状态保留秒数 [model] base_url https://taotoken.net/api api_key sk-你的Key model_id gpt-4o-mini qps_limit 20 # 模型调用每秒上限 [resource] cpu_quota 0.5 # 每个 Agent 容器 CPU 核数 memory_limit 512m # 每个 Agent 容器内存上限docker-compose.yml用来起 Redis 和带资源限制的 Agent 容器version: 3.8 services: redis: image: redis:7-alpine ports: - 6379:6379 command: redis-server --appendonly yes agent-worker: build: ./agent deploy: resources: limits: cpus: 0.5 memory: 512M environment: - HARNESS_ENDPOINThttp://host.docker.internal:8000 - REDIS_HOSTredis depends_on: - redis注意cpus: 0.5和memory: 512M这两行这就是资源隔离的硬限制。容器里的 Agent 就算写了死循环最多也只能吃掉 0.5 核不会把宿主机拖垮。这是并发控制里最容易被忽略、但性价比最高的一步。分布式锁的实现放在harness/lock.py用 Redis 的SET NX EX加看门狗续期import redis import threading import time import uuid class RedisLock: def __init__(self, client: redis.Redis, key: str, expire: int 30): self.client client self.key flock:{key} self.expire expire self.token uuid.uuid4().hex self._stop False def _watchdog(self): while not self._stop: time.sleep(self.expire // 3) if self.client.get(self.key) self.token: self.client.expire(self.key, self.expire) def acquire(self) - bool: ok self.client.set(self.key, self.token, nxTrue, exself.expire) if ok: threading.Thread(targetself._watchdog, daemonTrue).start() return ok def release(self): self._stop True if self.client.get(self.key) self.token: self.client.delete(self.key) def __enter__(self): if not self.acquire(): raise RuntimeError(lock busy) return self def __exit__(self, *args): self.release()状态存储放在harness/state.py所有任务状态写 Redis进程重启不丢import json import time import redis class StateStore: def __init__(self, client: redis.Redis, ttl: int 86400): self.client client self.ttl ttl def set(self, task_id: str, status: str, resultNone): data {status: status, result: result, ts: time.time()} self.client.set(ftask:{task_id}, json.dumps(data), exself.ttl) def get(self, task_id: str): raw self.client.get(ftask:{task_id}) return json.loads(raw) if raw else None def idempotent(self, key: str, task_id: str): return self.client.set(fidem:{key}, task_id, nxTrue, exself.ttl)调度器harness/scheduler.py把上面几块串起来核心是并发上限 优先级 锁import heapq import threading import time import redis from harness.lock import RedisLock from harness.state import StateStore class Scheduler: def __init__(self, cfg): self.cfg cfg self.client redis.Redis(hostcfg[redis][host], portcfg[redis][port], dbcfg[redis][db], decode_responsesTrue) self.state StateStore(self.client, cfg[redis][state_ttl]) self.queue [] self.running 0 self.lock threading.Lock() def submit(self, task_id, priority, payload, idem_key): if not self.state.idempotent(idem_key, task_id): return {task_id: task_id, status: duplicated} heapq.heappush(self.queue, (priority, task_id, payload)) self.state.set(task_id, pending) return {task_id: task_id, status: pending} def _run_one(self, task_id, payload): with self.lock: self.running 1 try: with RedisLock(self.client, fmodel:{self.cfg[model][model_id]}, self.cfg[redis][lock_expire]): self.state.set(task_id, running) time.sleep(2) # 模拟 Agent 执行 self.state.set(task_id, success, fresult-of-{task_id}) except Exception as e: self.state.set(task_id, failed, str(e)) finally: with self.lock: self.running - 1 def loop(self): while True: if self.running self.cfg[harness][max_concurrency] or not self.queue: time.sleep(0.05) continue _, task_id, payload heapq.heappop(self.queue) threading.Thread(targetself._run_one, args(task_id, payload), daemonTrue).start()这份配置的关键点有三个max_concurrency控制同时执行数RedisLock保护模型这个共享资源StateStore保证状态可查。你把这几个文件建好pip install redis再docker compose up -d redis就能进入下一步验证。4. 验证请求并发跑起来看隔离和一致性配置写完不验证等于没写。这一节我们用一组并发请求观察三件事并发上限是否生效、任务状态是否一致、资源是否被隔离。先起一个最小的 HTTP 入口方便你用脚本压。在harness/__init__.py旁边加一个server.pyfrom fastapi import FastAPI from pydantic import BaseModel from harness.scheduler import Scheduler import toml import threading cfg toml.load(config/harness.toml) sched Scheduler(cfg) app FastAPI() class TaskIn(BaseModel): task_id: str priority: int 2 payload: str idem_key: str app.post(/submit) def submit(t: TaskIn): return sched.submit(t.task_id, t.priority, t.payload, t.idem_key) app.get(/status/{task_id}) def status(task_id: str): return sched.state.get(task_id) or {status: not_found} threading.Thread(targetsched.loop, daemonTrue).start()启动服务pip install fastapi uvicorn toml redis uvicorn server:app --host 0.0.0.0 --port 8000然后写一个并发提交脚本bench.py模拟 30 个任务同时进来import requests import concurrent.futures def submit(i): r requests.post(http://127.0.0.1:8000/submit, json{ task_id: ftask-{i}, priority: 1 if i % 5 0 else 2, payload: fprompt-{i}, idem_key: fidem-{i} }) return r.json() with concurrent.futures.ThreadPoolExecutor(max_workers30) as ex: results list(ex.map(submit, range(30))) print(提交结果:, results[:5])跑完之后用下面的脚本查状态重点看有没有任务卡在running不动或者同一个task_id出现两个不同结果import requests import time time.sleep(8) for i in range(30): r requests.get(fhttp://127.0.0.1:8000/status/task-{i}).json() print(ftask-{i}: {r[status]})实测下来你会看到30 个任务全部提交成功但同一时刻处于running的任务不会超过 10 个max_concurrency生效每个任务的最终状态是success结果是result-of-task-i没有串号重复提交同一个idem_key会返回duplicated不会重复执行。这就是状态一致性和幂等性的验证。再验证资源隔离。故意在某个 Agent 里写一个死循环观察宿主机 CPU# agent/worker.py 里临时加一段 while True: pass把它跑在容器里然后docker stats看这个容器的 CPU 会被压在 50% 左右对应cpus: 0.5宿主机其他进程不受影响。如果你不用容器用resource.setrlimit也能做进程级限制但隔离强度不如容器。验证模型调用限流可以在_run_one里加一个计数器统计每秒实际发出的请求数确认不超过qps_limit。这一步能帮你提前发现重试风暴——很多 429 不是因为配额小而是失败后无脑重试把配额瞬间打满。5. 本篇常见错排查401、锁失败、状态读不到并发场景下的报错往往不是单一原因我按真实遇到的频率排一下。401 Unauthorized。这个最常见通常是三件套没对齐Base URL 写成了https://taotoken.net少了/api或者 Key 复制时带了空格或者 Model ID 在你账号下不可用。排查顺序是先用 curl 单独打一次模型接口确认 Key 和模型都对再放进 Harness。命令如下curl https://taotoken.net/api/chat/completions \ -H Authorization: Bearer sk-你的Key \ -H Content-Type: application/json \ -d {model:gpt-4o-mini,messages:[{role:user,content:ping}]}如果这里就 401别往下查 Harness先把 Key 和 Base URL 修对。local proxy failed / connection refused。这个报错一般出现在 Agent 容器里访问 Harness 或 Redis 时。容器内的127.0.0.1指向容器自己不是宿主机。解决办法是把HARNESS_ENDPOINT写成http://host.docker.internal:8000Redis 的 host 写成 compose 里的服务名redis。如果你在 Linux 上跑host.docker.internal需要额外加extra_hosts或者直接用宿主机的内网 IP。reading choices 相关报错。这类错误通常出现在解析模型返回时返回体结构和预期不一致。并发下如果多个线程共用一个 response 对象或者没做深拷贝就会出现读到一半被另一个线程改掉的情况。确保每次请求独立创建 client或者用连接池但不要共享可变状态。OAuth / token 过期。如果你用的是需要 OAuth 的模型服务token 过期后并发请求会集体失败。Harness 层要做统一的 token 刷新并且刷新时加锁避免 10 个线程同时去刷新导致互相覆盖。刷新逻辑放在调度器里Agent 侧不感知。锁获取失败lock busy。说明同一时刻有任务在占用模型资源这是正常的保护行为。但如果失败率过高说明lock_expire太短或者任务执行时间太长。把lock_expire调到大于单任务 P99 执行时间并确认看门狗线程在跑。如果看门狗没启动锁会在任务没跑完时过期另一个任务进来就会串号。状态读不到not_found。检查state_ttl是不是太短任务还没查就过期了或者 Redis 的 db 选错了写入和读取不在同一个 db。还有一种情况是任务提交时幂等键冲突直接返回了duplicated但你没记录原始 task_id导致查不到。排查时建议打开 Redis 监控redis-cli monitor能实时看到所有锁和状态的读写哪个 key 没写进去一目了然。6. 把并发控制收进 Harness别散在 Agent 里写到这里你应该能跑通一套最小可用的并发控制优先级队列管调度分布式锁管共享资源容器配额管隔离Redis 管状态。这套东西的价值不在于多复杂而在于它把并发问题收口到了 Harness 一层Agent 只管干活不用各自实现一套锁和重试。如果你要长期跑编码类或 Agent 类任务建议把模型出口和配额管理固定下来用 Coding Plan 做统一的调用规划避免每个项目重复配 Key。地址是 https://taotoken.net/coding-plan 。接入文档在 https://taotoken.net/doc API Keys 在 https://taotoken.net/api-keys 模型验证用 https://taotoken.net/api-keys 旁边的对话入口即可。最后留一个我踩过的坑不要用进程内的threading.Lock去保护跨容器的资源它只在单进程内有效。多任务并发时锁的作用域一定要和资源的共享范围对齐——资源跨容器锁就必须是分布式的。把这一点想清楚你的 Harness 就不会在并发上翻车。