医疗 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 ,
:
():
.max_concurrent = max_concurrent
.available = asyncio.Semaphore(max_concurrent)
.queue = asyncio.PriorityQueue()
.active_tasks: [, ] = {}
() -> :
:
asyncio.wait_for(.available.acquire(), timeout=)
.active_tasks[token.] = time.time()
asyncio.TimeoutError:
.queue.put((token.priority_level, token))
():
token_id .active_tasks:
.active_tasks[token_id]
.available.release()
