医疗 AI 多智能体资源调度：用 Python 构建高性能 MCU 资源池
最近在重构我们的医疗 AI 服务平台时，遇到了一个典型的多智能体资源争用问题。想象一下这样的场景：
- 急诊风险预警智能体检测到患者可能发生脓毒症，需要立即调用 GPU 进行推理
- 同时，影像分析智能体正在处理一批 CT 扫描，也需要 GPU 资源
- 质控智能体要分析医嘱合规性，需要调用大语言模型接口
- 病历总结智能体正在为出院患者生成报告
所有智能体都在'抢'有限的 GPU 卡、模型并发槽位、API 调用额度。如果让每个智能体自己管理资源抢占，结果就是：
- 资源利用不均：有的 GPU 卡空闲，有的被排队挤爆
- 优先级混乱：急诊任务可能被常规任务阻塞
- 无法审计：谁占用了什么资源？为什么失败？说不清楚
这就是我们需要一个中央调度器的原因。在多方会议系统中，这类组件被称为MCU（多点控制单元），而在我们的架构里，它负责将计算资源抽象为可分配的对象。
架构设计思路
解决这个问题的核心在于解耦'业务逻辑'与'资源获取'。我们不需要在每个 Agent 内部硬编码 if gpu_available，而是统一交给调度层处理。
1. 资源抽象化
首先，把物理硬件包装成逻辑对象。无论是 NVIDIA A100 还是 LLM API Key，对外都表现为一个 ResourceToken。
from dataclasses import dataclass
from enum import Enum
class ResourceType(Enum):
GPU = "gpu"
LLM_API = "llm_api"
CPU = "cpu"
@dataclass
class ResourceToken:
id: str
type: ResourceType
priority_level: int # 1-10, 越高越紧急
owner_id: str # 请求来源 Agent ID
这里的关键是 priority_level。医疗场景下，急诊任务的优先级必须高于常规体检报告，这直接决定了调度器的行为。
2. 异步调度核心
Python 的 asyncio 非常适合这种高并发 IO 密集型场景。我们实现一个简单的优先级队列来管理请求。
import asyncio
import time
from typing import List, Dict
class ResourcePool:
def __init__(self, max_concurrent: int):
self.max_concurrent = max_concurrent
self.available = asyncio.Semaphore(max_concurrent)
self.queue = asyncio.PriorityQueue()
self.active_tasks: Dict[str, float] = {}
async def acquire(self, token: ResourceToken) -> bool:
# 尝试获取信号量，超时则进入队列
try:
await asyncio.wait_for(self.available.acquire(), timeout=5.0)
self.active_tasks[token.id] = time.time()
return True
except asyncio.TimeoutError:
# 放入优先级队列等待
await self.queue.put((token.priority_level, token))
return False
async def release(self, token_id: str):
if token_id in self.active_tasks:
del self.active_tasks[token_id]
self.available.release()
注意这里的 Semaphore 和 PriorityQueue 配合使用。普通任务会先尝试获取信号量，拿不到就排队；高优先级任务可以直接插队（实际生产中可能需要更复杂的抢占机制）。
实际运行中的坑
写代码容易，跑起来稳才难。在实际部署中，有几个点特别需要注意：
- 死锁预防：确保释放资源的逻辑一定在
finally块中执行，防止异常导致资源永久占用。 - 超时熔断：如果某个 Agent 长时间不释放资源，调度器需要有自动回收机制，避免单点故障拖垮整个池子。
- 监控埋点：记录每个 Token 的持有时长。如果发现某类任务平均耗时突增，可能是模型服务出现了性能瓶颈。
结语
通过引入中央调度器，我们将原本分散的资源竞争收敛到了单一入口。这不仅解决了优先级问题，还让后续的审计日志变得简单——只需要查调度器的中间件日志即可。
对于医疗 AI 这种对时效性和准确性要求极高的领域，这套基于 Python 异步原语构建的资源池方案，既保证了开发效率，又满足了生产环境的稳定性需求。后续还可以在此基础上扩展负载均衡策略，比如根据 GPU 显存剩余量动态调整并发数。
