引言

Agent的很多操作是独立的——搜索三个不同的关键词、查询多个数据源、分析多个文件。如果串行执行，总时间是所有操作时间的总和；如果并发执行，总时间接近最慢操作的时间。

并发执行是Agent性能优化的重要手段。但并发带来了复杂性：资源竞争、状态同步、错误处理、死锁风险。2026年的Agent并发架构需要在性能和复杂性之间找到最佳平衡。

一、Agent并发场景

1.1 工具调用并发

一个请求需要调用多个独立工具时，可以并发执行：

串行: search_web(3s) → query_db(2s) → call_api(1s) = 6s
并发: search_web(3s) ─┐
      query_db(2s)   ─┼─ 并行 = 3s
      call_api(1s)   ─┘

1.2 多Agent并发

多个Agent同时处理不同子任务：

用户: "调研三家公司的市场表现"
Agent A: 调研公司甲 ─┐
Agent B: 调研公司乙 ─┼─ 并行
Agent C: 调研公司丙 ─┘

1.3 批量处理并发

对一批数据并行处理：

100个文档需要摘要
→ 分成10批，每批10个文档
→ 10个Worker并行处理
→ 总时间 ≈ 单个文档的10倍（而非100倍）

1.4 竞争执行

同一任务由多个Agent竞争执行，取最优结果：

Agent A → 方案A ─┐
Agent B → 方案B ─┼→ 评估器选择最优
Agent C → 方案C ─┘

二、并发模型

2.1 异步IO模型

Agent的绝大多数操作是IO密集型（LLM API调用、工具调用），适合异步IO模型：

import asyncio

async def execute_tasks_concurrently(tasks):
    """并发执行多个独立任务"""
    results = await asyncio.gather(
        *[task.execute() for task in tasks],
        return_exceptions=True
    )
    
    # 分离成功和失败结果
    successes = []
    failures = []
    for task, result in zip(tasks, results):
        if isinstance(result, Exception):
            failures.append((task, result))
        else:
            successes.append((task, result))
    
    return successes, failures

2.2 Actor模型

每个Agent是一个Actor，拥有独立的状态，通过消息传递通信：

class AgentActor:
    def __init__(self, agent_id):
        self.agent_id = agent_id
        self.mailbox = asyncio.Queue()
        self.state = {}
    
    async def run(self):
        while True:
            message = await self.mailbox.get()
            result = await self.process(message)
            if message.reply_to:
                await message.reply_to.put(result)
    
    async def process(self, message):
        # Agent处理逻辑
        pass

Actor模型天然支持并发，且避免了共享状态的问题。

2.3 Map-Reduce模型

适合批量数据处理：

async def map_reduce(data_items, map_fn, reduce_fn, parallelism=10):
    # Map阶段：并行处理
    semaphore = asyncio.Semaphore(parallelism)
    
    async def limited_map(item):
        async with semaphore:
            return await map_fn(item)
    
    mapped = await asyncio.gather(*[limited_map(item) for item in data_items])
    
    # Reduce阶段：聚合结果
    return reduce_fn(mapped)

2.4 流水线模型

将任务分解为阶段，各阶段并行执行：

阶段1: 数据收集 ──→ 阶段2: 数据清洗 ──→ 阶段3: 分析
     ↑ 持续进行           ↑ 持续进行         ↑ 持续进行

async def pipeline(stages, input_stream):
    """流水线执行"""
    current_stream = input_stream
    for stage in stages:
        current_stream = stage.process_stream(current_stream)
    return current_stream

三、并发控制

3.1 并发度控制

过高的并发度会压垮下游服务。需要限制并发数：

class ConcurrencyController:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.active = 0
        self.waiting = 0
    
    async def execute(self, task):
        self.waiting += 1
        async with self.semaphore:
            self.waiting -= 1
            self.active += 1
            try:
                return await task()
            finally:
                self.active -= 1
    
    def stats(self):
        return {"active": self.active, "waiting": self.waiting}

3.2 分级并发限制

不同资源设置不同的并发限制：

class TieredConcurrencyManager:
    def __init__(self):
        self.limits = {
            "llm_api": ConcurrencyController(max_concurrent=5),
            "web_search": ConcurrencyController(max_concurrent=3),
            "database": ConcurrencyController(max_concurrent=20),
            "file_io": ConcurrencyController(max_concurrent=50),
        }
    
    async def execute(self, resource_type, task):
        controller = self.limits[resource_type]
        return await controller.execute(task)

3.3 自适应并发

根据系统负载和成功率动态调整并发度：

class AdaptiveConcurrency:
    def __init__(self, initial=10, min_val=1, max_val=50):
        self.current = initial
        self.min_val = min_val
        self.max_val = max_val
        self.success_window = []
        self.window_size = 100
    
    def on_success(self):
        self.success_window.append(1)
        if len(self.success_window) > self.window_size:
            self.success_window.pop(0)
        self._adjust()
    
    def on_failure(self):
        self.success_window.append(0)
        if len(self.success_window) > self.window_size:
            self.success_window.pop(0)
        self._adjust()
    
    def _adjust(self):
        if not self.success_window:
            return
        success_rate = sum(self.success_window) / len(self.success_window)
        if success_rate > 0.95:
            self.current = min(self.current + 1, self.max_val)
        elif success_rate < 0.8:
            self.current = max(self.current - 2, self.min_val)

四、数据一致性

4.1 共享状态保护

并发Agent访问共享状态时需要同步：

class SharedState:
    def __init__(self):
        self._state = {}
        self._lock = asyncio.Lock()
    
    async def update(self, key, value):
        async with self._lock:
            self._state[key] = value
    
    async def get(self, key):
        async with self._lock:
            return self._state.get(key)
    
    async def update_with_fn(self, key, update_fn):
        """原子更新：读取-修改-写回"""
        async with self._lock:
            current = self._state.get(key)
            new_value = update_fn(current)
            self._state[key] = new_value
            return new_value

4.2 结果聚合

并发执行的结果需要安全聚合：

async def concurrent_aggregate(tasks, aggregator):
    """并发执行并聚合结果"""
    results = []
    lock = asyncio.Lock()
    
    async def task_wrapper(task):
        result = await task.execute()
        async with lock:
            results.append(result)
            # 实时聚合，不等待全部完成
            if len(results) % 10 == 0:
                await aggregator.partial_aggregate(results)
    
    await asyncio.gather(*[task_wrapper(t) for t in tasks])
    return await aggregator.final_aggregate(results)

4.3 顺序保证

某些场景需要保持结果顺序：

async def ordered_concurrent(tasks):
    """并发执行但保持结果顺序"""
    results = await asyncio.gather(*[t.execute() for t in tasks])
    # asyncio.gather 保证结果顺序与输入顺序一致
    return list(zip(tasks, results))

五、错误处理

5.1 部分失败

并发任务中部分失败是常见情况：

async def execute_with_partial_failure(tasks, min_success_ratio=0.7):
    results = await asyncio.gather(
        *[wrap_task(t) for t in tasks],
        return_exceptions=True
    )
    
    successes = []
    failures = []
    for task, result in zip(tasks, results):
        if isinstance(result, Exception):
            failures.append((task, result))
        else:
            successes.append((task, result))
    
    success_ratio = len(successes) / len(tasks)
    if success_ratio < min_success_ratio:
        raise PartialFailureError(
            f"Only {success_ratio:.0%} tasks succeeded (minimum: {min_success_ratio:.0%})",
            successes=successes,
            failures=failures
        )
    
    return successes, failures

5.2 快速失败

某些场景下，一个失败就应该取消所有：

async def execute_all_or_nothing(tasks):
    try:
        results = await asyncio.gather(
            *[t.execute() for t in tasks],
            return_exceptions=False  # 任何一个失败都会抛出
        )
        return results
    except Exception as e:
        # 取消所有未完成的任务
        raise AllOrNothingError(f"Task failed: {e}")

5.3 超时取消

为并发任务设置总超时：

async def execute_with_timeout(tasks, timeout=30):
    try:
        results = await asyncio.wait_for(
            asyncio.gather(*[t.execute() for t in tasks], return_exceptions=True),
            timeout=timeout
        )
        return results
    except asyncio.TimeoutError:
        # 超时后未完成的任务被自动取消
        raise ConcurrentTimeoutError(f"Tasks did not complete within {timeout}s")

六、性能优化

6.1 任务分组

将任务按优先级和资源需求分组：

async def grouped_execution(tasks):
    # 按优先级分组
    high_priority = [t for t in tasks if t.priority == "high"]
    medium_priority = [t for t in tasks if t.priority == "medium"]
    low_priority = [t for t in tasks if t.priority == "low"]
    
    # 高优先级立即执行
    high_results = await asyncio.gather(*[t.execute() for t in high_priority])
    
    # 中优先级并发执行
    medium_results = await asyncio.gather(*[t.execute() for t in medium_priority])
    
    # 低优先级串行执行（利用空闲时间）
    low_results = []
    for task in low_priority:
        low_results.append(await task.execute())
    
    return high_results + medium_results + low_results

6.2 结果流式返回

不必等所有并发任务完成，先完成的结果先返回：

async def streaming_concurrent(tasks):
    """并发执行，先完成的结果先yield"""
    queue = asyncio.Queue()
    
    async def task_wrapper(task):
        result = await task.execute()
        await queue.put(result)
    
    # 启动所有任务
    asyncio.create_task(asyncio.gather(*[task_wrapper(t) for t in tasks]))
    
    # 流式返回结果
    completed = 0
    total = len(tasks)
    while completed < total:
        result = await queue.get()
        completed += 1
        yield result

结语

并发执行是Agent性能提升的关键手段。从串行到并发，不仅是代码层面的改变，更是思维方式的转变——从"一步做一件事"到"同时做多件事"。

但并发也是复杂度的来源。2026年的最佳实践是：先用串行实现确保正确性，再在性能热点引入并发。并发度从小开始，逐步提高，持续监控。不要过早优化——如果一个任务串行执行只需要2秒，引入并发带来的复杂度可能不值得。

记住：并发不是目的，而是手段。用户体验的提升才是最终目标。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。