当模型和供应商逐渐增多控制面是否需要跟随每个插件一起修改会直接影响发布频率和故障范围。if task_type codex-chat写进控制面鉴权、限流、计费和路由缠在一起。出图要上 ComfyUI、分割要上 SAM2你就得发一版网关。更烦的是插件一崩CUDA 泄漏、SDK 卡死把整个 FastAPI 进程带走聊天接口一起 502。本文以插件注册、按类型消费和能力心跳为参考实现讨论新模型如何接入执行器以及线程、进程、容器三种隔离粒度如何选择。这里的网关裁决和一类型一进程是工程选项不是可插拔 Worker 的唯一形态。适用场景已经具备异步任务网关和按类型拆分的队列准备将推理从控制面剥离并允许业务开发者自行接入插件。先说结论一种低耦合实现是网关不加载插件执行器尽量不直接依赖业务库而是通过内部回调或受控接口提交结果。Worker 直接条件更新任务表也是常见做法链路更短但需要下发数据库权限并统一状态机实现。插件接入通常可以收敛为一个register和重启对应执行器实例若团队使用静态配置或服务发现也可以维护显式路由表换取更强的变更审计。注册表只存可调用对象不强制继承基类。约定签名是(model_spec, payload, inference_params)。线程数、超时、凭证模式都写在装饰器上随能力发现信息同步到控制面。一类型一进程只是常见工程选项之一不是必须。它适合依赖冲突、CUDA 上下文或崩溃隔离要求较高的任务轻量 HTTP 插件可共享进程降低内存开销高风险 GPU 插件还可以提升到一类型一容器代价是部署和观测对象增多。控制面不应依赖扫描本地磁盘来发现能力能力发现可以来自心跳、注册中心或静态配置。若业务要求提交即要求当前可执行可以在创建时校验 live 能力并拒绝入队若允许离线排队则应接受任务并等待能力恢复。为什么网关里写 if-else 会越写越死规模较小时创建任务处直接分支是一种简单做法iftask_typeopenai-chat:resultawaitopenai_chat(payload)eliftask_typesam2:resultawaitsam2(payload)随着类型和资源差异增大它会出现三类扩展成本加供应商要改控制面、发版、回归鉴权计费。GPU 任务和 HTTP 任务抢同一个进程GIL 和 CUDA 互相伤害。网关要知道「现在有谁能跑」你却把能力写死在代码里扩容、摘节点都要改配置文件。采用控制面与执行面拆分后职责边界可以这样划分控制面网关 鉴权、限流、计费 任务接收与路由决策 运行态与终态裁决 能力可用性判断 管理端 执行面Worker 插件加载与能力注册 按类型消费消息 推理执行 结果上报内部回调或受控写库在这套边界中网关进程通常不需要插件目录执行器也不一定要直接连接任务状态使用的 Redis / PostgreSQL更常见的做法是通过内部回调提交结果。Worker 直接条件更新任务表同样成立链路更短但需要下发数据库权限并统一状态机实现。插件如何被发现文件、包还是显式注册一个插件可以对应一个.py文件也可以对应一个包关键是启动时由加载器发现并注册能力而不是由网关硬编码。参考加载器可能在执行器启动时扫描插件目录逐个加载.py并执行注册逻辑文件名因此可以不是常规模块名。若采用包结构还需要处理相对导入、命名冲突和入口模块选择并不总是“扫目录就能 import”那么简单。某个文件语法错误时一种策略是只记录 warning让其他插件继续注册代价是错误类型会缺席能力列表。另一种选择是启动时遇错即失败适合要求插件集合完整一致的环境。修改插件后可以检查启动日志里的Registered task types。少了一个常见原因是文件名以下划线开头被跳过或加载时发生异常。如果插件加载器使用runpy.run_path(..., run_name__main__)这类执行方式模块级if __name__ __main__:会在加载时被触发若采用标准import module则不会。无论采用哪种加载方式插件文件都不应在模块级发起真实外部调用否则启动阶段就可能去撞供应商。生产环境可以把插件挂到外部目录减少为每个模型重建镜像的频率代价是镜像内容不再完全决定运行版本需要额外校验插件签名、来源和版本。强调可复现部署时把插件随镜像发布更合适。最小冒烟插件长这样fromcore.plugin.registryimportregisterfromcore.shared.credential_modeimportCredentialModeregister(hello,worker_threads1,credential_modeCredentialMode.NONE,metadata_version1,timeout_seconds60,description无凭证 echo用来冒烟。,)definfer(model_spec,payload,inference_params):return{message:Hello world,echo:payloador{},model:model_spec.get(name)orhello,}创建任务curl-sS-XPOST http://127.0.0.1:80/api/v1/tasks\-HContent-Type: application/json\-HX-API-Key:$API_KEY\-d{ task_type: hello, payload: {message: plugin-ok} }在本文信封协议中credential_modenone表示创建路径会把凭证字段清空后再入队插件不应去账户池拿 key。这不是所有系统的通用规则而是能力元数据与入队信封之间的约定。很多网关会把 OpenAI 兼容接口做成「看起来同步」内部仍是这条hello任务类型包了一层皮。协议适配是皮插件执行是骨。register只往两张表里塞东西插件注册既可以采用基类继承与inspect扫描也可以采用装饰器。基类便于静态约束和 IDE 提示但基类签名变化会影响全部插件装饰器对文件名和函数实现限制较少代价是函数签名主要靠运行时校验。下面使用装饰器方案。全局就两张表函数 元数据。PROVIDERS:dict[str,tuple[Callable,int]]{}PROVIDER_METADATA:dict[str,TaskTypeMetadata]{}defregister(task_type:str,worker_threads:int1,*,credential_mode:strcaller_or_pool,metadata_version:int0,timeout_seconds:int|NoneNone,description:str,payload_schema:dict|NoneNone,):defwrapper(func):PROVIDERS[task_type](func,worker_threads)PROVIDER_METADATA[task_type]TaskTypeMetadata(task_typetask_type,credential_modecredential_mode,metadata_versionmetadata_version,timeout_secondstimeout_seconds,descriptiondescription,payload_schemapayload_schema,)returnfuncreturnwrapper注册表不约束函数签名。约定是执行器这边定的(model_spec, payload, inference_params)。以后若有别的宿主可以自己换调用方式不必改装饰器。调度器和能力心跳只通过查询接口拿函数、线程数、超时、元数据不 import 任何一个插件文件。这很重要主进程启动路径不要把 ComfyUI、torch 拖进来那些东西留给子进程。生产插件长什么样聊天类是高并发 HTTP不是 GPUregister(openai-chat,worker_threads10,credential_modecaller_or_pool,metadata_version1,timeout_seconds600,descriptionOpenAI Chat支持文本、图片和文件片段。,)definfer(model_spec,payload,inference_params):clientOpenAI(api_keyresolve_api_key(model_spec))timeoutget_plugin_timeout_seconds(600)respclient.chat.completions.create(modelmodel_spec.get(name)orgpt-4o,messagespayload.get(messages)or[],timeouttimeout,**(inference_paramsor{}),)return{text:resp.choices[0].message.content}在本文示例实现中worker_threads10同时驱动两件事该类型独立进程里的 10 线程池以及主进程对该类型的 prefetch / 槽位上限。这是为了避免消费者预取过量导致槽位错配不是线程池大小与 MQ prefetch 的自然等价关系。很多系统里 prefetch 还会单独根据消息大小、ack 时机和吞吐目标调整。credential_modecaller_or_pool的协议含义是调用方带 key 时使用调用方凭证未提供时允许入队并在执行期向账户池申请。因此在该模式下仅因请求未带 key 而返回 400 会与元数据语义冲突。若某类任务要求调用方自带凭证应使用独立的caller_required模式而不是改变caller_or_pool的含义。分割、出图这类 GPU 任务把worker_threads留 1避免两路推理抢同一块显存register(sam2,worker_threads1,credential_modenone,metadata_version1,timeout_seconds300,)definfer(model_spec,payload,inference_params):endpointmodel_spec[endpoint]timeoutget_plugin_timeout_seconds(30)# 下载图片别写死imagedownload(payload[image_url],timeouttimeout)returncall_segment(endpoint,image,payload.get(box))同文件可以放多个register每个装饰器向注册表写一条调度器会按task_type创建进程和队列。若多个类型需要共享大模型权重把它们放进同一进程可节省显存若更重视故障隔离则按类型拆分。文件组织本身不应被误当成隔离边界。启动顺序与失败条件执行器的启动顺序需要与对外就绪语义一致下面是一种安排连队列检查 broker 真的支持重试和死信打日志注册了哪些类型每个类型一个 Process示例中的隔离粒度也可改为共享进程或独立容器每个类型一条结果监听线程每个类型一个 asyncio consumer开始对网关心跳对应代码可以记这个骨架asyncdefstartup(registry,queue_client):awaitqueue_client.connect()ifnotqueue_client.supports_retry_queue:raiseRuntimeError(broker 不支持重试拓扑拒绝启动)task_typesregistry.get_registered_task_types()logger.info(Registered task types: %s,task_types)schedulerInferenceScheduler()scheduler.initialize_workers(registry)awaitscheduler.start_result_monitor()fortask_typeintask_types:asyncio.create_task(consume_for_type(task_type),namefconsumer-{task_type})capability_client.start()在以instance_id作为 Redis 主键的实现中每个存活实例需要使用不同 ID例如forwarder-001否则后写会覆盖前写。这是当前存储模型的协议约束。若改用数据库复合键或服务发现平台生成实例身份则可由基础设施保证唯一性。关闭顺序应先停止接收新任务并摘除能力再等待在途任务到宽限期最后停止子进程。示例可以先停心跳、取消 consumer、发送停止哨兵并在 join 超时后 terminate5 秒并非通用值应覆盖可接受的任务收尾时间。对不可中断的 GPU 推理还需要 checkpoint、任务重投或更长的 termination grace period。Worker 隔离线程、进程还是容器当推理会阻塞事件循环或需要独立故障域时可以放到子进程。调度器初始化示例如下frommultiprocessingimportProcess,Queue,Pipefortask_typeintask_types:task_queueQueue()result_queueQueue()recv_conn,send_connPipe(duplexFalse)threadsregistry.get_worker_threads(task_type)timeoutregistry.get_timeout_seconds(task_type)processProcess(targetrun_inference_worker,args(task_type,threads,timeout,task_queue,result_queue,send_conn),daemonTrue,)process.start()workers[task_type]{process:process,task_queue:task_queue,result_queue:result_queue,notify_conn:recv_conn,worker_threads:threads,active_tasks:0,}主进程只负责 MQ pull、槽位记账、进程间投递、收结果和回调。这样 CUDA 泄漏、SDK 卡死、GIL 占满通常局限在对应类型的子进程。不过进程仍共享宿主机、文件系统和 GPU 驱动无法提供容器级资源与安全隔离。三种粒度可以这样选同进程线程池适合可信、以 I/O 为主且依赖兼容的插件收益是通信开销小代价是崩溃和内存泄漏可能影响其他类型。一类型一进程适合需要 GIL/CUDA 上下文隔离、但仍希望单机统一调度的场景收益是故障面较小代价是模型内存可能重复加载IPC 和进程监管更复杂。一类型一容器或独立服务适合依赖冲突、资源限额和安全边界明确的场景收益是独立扩缩与资源配额代价是部署、网络和观测成本更高。子进程入口再分两种模式defrun_inference_worker(task_type,worker_threads,timeout,task_q,result_q,conn):ifworker_threads1:run_single_thread(task_type,timeout,task_q,result_q,conn)else:run_thread_pool(task_type,worker_threads,timeout,task_q,result_q,conn)单线程循环task_queue.get()阻塞 → 调infer()→result_queue.put→ pipe 写 1 字节叫醒主进程。适合 GPU也适合「虽然是 HTTP 但不要在同进程里并发把连接打爆」的场景。多线程主线程只分发ThreadPoolExecutor跑任务。适合 OpenCV / NumPy / 会释放 GIL 的 C 扩展。纯 Python GIL 的插件把 threads 开到 10收益有限还是会互相挡。别看配置好看就开满。子进程里只 import 真正跑推理的那一层不要把 FastAPI、OpenTelemetry 再初始化一遍。你会在日志里看到两套 tracer排障时对不上。子进程里真正调插件的地方definfer(task_id,task_type,model_spec,payload,inference_params,timeout_secondsNone,client_roleNone,mq_attempt1):funcregistry.get_inference_function(task_type)iffuncisNone:return{task_id:task_id,status:FAILED,error:fNo inference function for{task_type},}try:withaccount_pool_context(client_roleclient_role,mq_attemptmq_attempt),\ plugin_execution_timeout(timeout_seconds):outputfunc(model_spec,payload,inference_params)return{task_id:task_id,status:SUCCESS,result:{output:output}}exceptExceptionase:return{task_id:task_id,status:FAILED,error_code:E4000001,error:fInference failed:{e},}几点实际约定插件返回值可以是普通 dict也可以带账户池计费元数据。执行器会拆开网关回调时记账。子进程中的异常不会自动进入主进程的 ASGI/FastAPI 异常处理中间件。它只能通过返回值协议、IPC、返回码、状态文件、进程监控或重试队列上报。在本文的返回值协议中子进程应返回 FAILED dict主进程才能回调网关并按mq_action处理消息。若异常未捕获而进程退出主进程还需要存活检测、重启或摘除能力消息则保持 unacked 等待重投。超时可以用 ContextVar 传播语义插件内部get_plugin_timeout_seconds()能读到这次任务的超时配置。ContextVar 只负责传参不负责强制终止线程超时在 CPython 里不可靠真正的硬超时仍依赖外层 deadline、客户端 timeout、任务重试预算和网关task_deadline_at。插件里的 HTTP client 应读取这个值而不是写死 600。账户池上下文只在需要池凭证时有意义。hello这类none模式不会申请 Redis 租约需要凭证池的插件则应遵守分配、续租和释放协议。信封是网关组的。插件不要自己去 MQ 里再读一遍也不要假设能 import 网关的任务服务。字段就这几个model_spectask_data.get(model_spec)or{}payloadtask_data.get(payload)or{}inference_paramstask_data.get(inference_params)or{}mq_attempttask_data.get(mq_attempt,1)client_roletask_data.get(client_role)能力发现心跳、注册中心或静态配置网关不应依赖扫描本地磁盘来发现能力。执行器可以通过 HTTP 心跳让控制面知道codex-chat存在服务注册中心、Kubernetes API、管理员手工标记或静态配置也能承担发现职责。心跳实现直接但需要 TTL 治理注册中心能力更完整但增加基础设施依赖静态配置则适合节点集合稳定的小规模部署。每轮 POST 一份body{instance_id:instance_id,task_types:[hello,openai-chat,sam2],task_type_metadata:{hello:{credential_mode:none,timeout_seconds:60,metadata_version:1,description:无凭证 echo用来冒烟。,},...},started_at:started_at,updated_at:now.isoformat(),heartbeat_interval_seconds:10,}awaithttp.post(heartbeat_url,jsonbody,headers{X-Internal-Key:secret})网关把实例写进 Redisforwarder:capability:instance:{instance_id} # JSONTTL 覆盖心跳间隔与允许的抖动 forwarder:capability:task_type:{task_type} # SET of instance_id forwarder:capability:task_type_metadata_map # 对外可见的类型元数据同一task_type被多个实例注册时示例按metadata_version选择较新元数据。修改description或credential_mode时需要递增版本否则滚动发布期间旧实例心跳可能覆盖新字段。替代方案是按部署 revision 分组、拒绝同一能力的混合版本或让控制面保存静态元数据这些方案一致性更强但会降低异构实例并存的灵活性。实例消失可由 TTL 到期判断正常关闭时再 DELETE。若采用租约式心跳TTL 应大于心跳间隔并覆盖网络抖动与调度延迟具体倍数由可接受的摘除延迟决定。TTL 太短会误摘太长则会把已失效能力保留更久。创建任务时网关先问这个类型现在有没有 live 能力capabilityawaitget_public_task_type(client_task_type)ifcapabilityisNone:raiseInvalidTaskType(fUnsupported task type:{client_task_type})credential_modecapability[credential_mode]示例在没有 live 能力时拒绝创建以避免无消费者消息长期积压。这适合“提交即要求当前可执行”的接口但不是唯一语义允许离线排队的系统可以先接受任务等待能力恢复使用 RabbitMQ mandatory publish 和 alternate exchange 可以把无法路由的消息退回或停放静态白名单还可以区分“合法但暂时无实例”和“协议中不存在的类型”。只有在消费者确认永久不支持且没有转交路径时死信才是合适终点。公开列表给管理端/调用方探活。内部还能查某个类型有没有实例。别指望心跳一停就把队列删掉——RabbitMQ 里空队列留着比误删有积压的队列更安全。从创建到进进程整条链路长什么样可以记这张图调用方 task_typeopenai-chat ▼ 网关 Redis 里该类型 availabletrue读到 caller_or_pool │ 有凭证就规范化后入信封没有仍入队 ▼ 消息进 task.queue.openai-chat ▼ 只有加载了 openai-chat 插件的执行器在消费 ▼ 有槽位 → 放进该类型子进程的 Queue → infer() ▼ 结果回调网关mq_action 决定 ack / retry / dead在这套按类型声明队列和进程的实现中「聊天任务进了出图进程」由 routing key 与消费者绑定共同避免。新模型对应新文件、新task_type和新队列拓扑共享队列方案则需要消费者在接单前校验能力并处理误路由。两台执行器的插件集合可以不同。A 有sam2B 没有时B 通常不订阅sam2队列误消费时按协议拒绝或转交。按能力给节点分组能独立扩缩统一加载全部插件则配置更简单但资源占用更高。凭证模式属于能力元数据不是网关 if-else 表一种能力元数据可以定义三类模式模式典型任务创建时网关做什么nonehello、sam2清空 credential不进池caller_or_poolopenai-chat、codex-chat有 key 用调用方没有仍入队执行期租池其它以解析函数为准未知值不要 silently 当 none把模式写在插件上的好处换一个 OAuth 插件不必改网关代码。心跳把 mode 带到 Redis创建路径就能读到。timeout_seconds同样来自插件。网关还有请求体超时和全局默认实际 deadline 是创建路径里算的。插件超时用于执行侧约束 HTTP/推理网关超时用于任务截止。两套都要看只配一个会出现「插件还在跑、任务已被 deadline 判失败」。新增任务类型的接入流程丢一个my-model.py文件名随意。register(my-model, worker_threads..., credential_mode..., timeout_seconds..., metadata_version1)装饰一个def infer(model_spec, payload, inference_params)。模块级不要连供应商连供应商放进函数里方便加载失败时至少进程能起来。重启该执行器实例启动时 load 一次不是热加载。日志出现Registered task types含my-model。等一次心跳。能力列表或 Redis 里能看到。POST /tasksbody 里task_type: my-model。若采用「一类型三队列」拓扑RabbitMQ 可能出现主队列、重试队列和死信队列具体数量取决于 broker 拓扑设计。用 hello 的联调清单走一遍 ack再故意raise验证失败是否按既定队列协议进入 retry 或 dead。在动态能力发现方案中无需修改网关路由表。参考加载器不支持热更新因此改插件后要重启执行器若元数据按版本择新修改元数据时还需递增metadata_version。支持热加载或由注册中心管理版本时流程会不同。参考实现让执行器通过内部回调提交结果因此不需要 PostgreSQL 权限也不直接改终态表。另一种方案是执行器使用最小权限账号通过条件更新写终态收益是少一次 HTTP 跳转和更短的确认链路代价是状态机逻辑、数据库连接和凭证轮换进入每个执行器。无论选哪种worker_threads表示单实例并发而非集群总量。对确定性的非法 prompt若协议判定重试不会改变结果应标为不可重试若错误可能由模型版本或策略更新消除则可以进入有上限的延迟重试。怎么验证这套是通的启动执行器健康检查应 200。日志类似Registered task types: [hello, openai-chat, ...]以及Started N workers每个 worker 带 pid 和线程数。能力是否到网关。Redis 里应有实例 key以及task_type:hello这类 set。网关创建未知类型curl-sS-XPOST http://127.0.0.1:80/api/v1/tasks\-HContent-Type: application/json-HX-API-Key:$API_KEY\-d{task_type:does-not-exist,payload:{}}应返回类型不支持MQ 里不应出现对应队列拓扑是 publish 时才 declare失败路径不应 publish。hello 进程是否真的在跑。提交后日志应是收到 → 接下 → worker 打印 Processing → 内部回调。查结果output.echo应等于你的 payload。隔离。连续打 hello 的同时打一个故意time.sleep的本地插件临时 threads1。hello 的延迟不应跟 sleep 插件一样长。它们是不同进程、不同队列。如果一样长说明你把两个类型注册到了同一个函数或只起了一个 consumer。错误插件。在infer里raise RuntimeError(boom)。应回调 FAILEDerror 含Inference failed:。主进程还在健康检查仍 200。MQ 按网关mq_action走 retry 或 dead而不是把 FastAPI 打 500。停心跳。把执行器停掉等超过 TTL再 POST 原来的类型。应变成不支持。TTL 内仍可能创建成功——消息会堆在队列里实例回来继续消费。这是滚动发布窗口不是 bug窗口长度就是 Redis TTL。排障顺序Registered task types→ Redis 实例 key → 网关能否读到该类型 → 队列有没有 consumer。别一上来就怀疑模型 SDK。常见故障与排查线索插件里写了if __name__ __main__。若加载器采用run_name__main__的执行方式模块级脚本入口会在加载时被触发若采用标准 import则不会。无论哪种方式都不应在模块级发起真实外部调用。metadata_version 没加。在“按版本取新”的合并策略下滚动发布时旧心跳可能盖掉新字段因此修改凭证模式、超时或描述时需要递增版本。若注册中心以部署版本为准则遵循其冲突策略。实例 id 重复。两台 forwarder-001能力集合来回抖网关间歇性拒绝创建。线程超时当硬超时。CPython 杀不了卡住的线程。HTTP client 自己设 timeout任务截止靠网关 deadline。子进程未捕获异常。进程没了消息 unacked看起来像「队列堵住」。其实是 worker 死了。主进程要对子进程做存活检查挂了要拉起来或把实例摘掉。GPU 并发与显存预算不匹配。单模型占用接近整卡时可以从单线程起步支持动态批处理、MIG、模型并行或单次占用较低时多并发反而能提高利用率。线程数应由显存峰值、算子并发能力和延迟目标压测决定。凭证模式与创建校验不一致。caller_or_pool允许空凭证入队caller_required才要求请求携带凭证。创建校验应读取能力元数据避免把池凭证路径意外关闭。写在最后可插拔执行器中模型可以抽象成函数关键取舍是控制面和执行面拆到什么程度是否接受显式路由配置线程、进程、容器三种隔离的资源成本与失败边界心跳、注册中心或静态配置如何发布能力把这三块的契约写清楚后凭证租赁、队列确认和扩缩容才能在不依赖具体插件实现的前提下协作。最小实现可以从单进程、单线程和静态能力配置起步随后按故障隔离、吞吐和凭证治理需求增加进程、心跳与租约。若一开始就存在不可信插件、依赖冲突或严格 GPU 配额也可以直接采用容器隔离它增加部署成本却能提前建立清晰的资源边界。演进顺序应由主要风险决定。
当模型和供应商逐渐增多控制面是否需要跟随每个插件一起修改会直接影响发布频率和故障范围。if task_type codex-chat写进控制面鉴权、限流、计费和路由缠在一起。出图要上 ComfyUI、分割要上 SAM2你就得发一版网关。更烦的是插件一崩CUDA 泄漏、SDK 卡死把整个 FastAPI 进程带走聊天接口一起 502。本文以插件注册、按类型消费和能力心跳为参考实现讨论新模型如何接入执行器以及线程、进程、容器三种隔离粒度如何选择。这里的网关裁决和一类型一进程是工程选项不是可插拔 Worker 的唯一形态。适用场景已经具备异步任务网关和按类型拆分的队列准备将推理从控制面剥离并允许业务开发者自行接入插件。先说结论一种低耦合实现是网关不加载插件执行器尽量不直接依赖业务库而是通过内部回调或受控接口提交结果。Worker 直接条件更新任务表也是常见做法链路更短但需要下发数据库权限并统一状态机实现。插件接入通常可以收敛为一个register和重启对应执行器实例若团队使用静态配置或服务发现也可以维护显式路由表换取更强的变更审计。注册表只存可调用对象不强制继承基类。约定签名是(model_spec, payload, inference_params)。线程数、超时、凭证模式都写在装饰器上随能力发现信息同步到控制面。一类型一进程只是常见工程选项之一不是必须。它适合依赖冲突、CUDA 上下文或崩溃隔离要求较高的任务轻量 HTTP 插件可共享进程降低内存开销高风险 GPU 插件还可以提升到一类型一容器代价是部署和观测对象增多。控制面不应依赖扫描本地磁盘来发现能力能力发现可以来自心跳、注册中心或静态配置。若业务要求提交即要求当前可执行可以在创建时校验 live 能力并拒绝入队若允许离线排队则应接受任务并等待能力恢复。为什么网关里写 if-else 会越写越死规模较小时创建任务处直接分支是一种简单做法iftask_typeopenai-chat:resultawaitopenai_chat(payload)eliftask_typesam2:resultawaitsam2(payload)随着类型和资源差异增大它会出现三类扩展成本加供应商要改控制面、发版、回归鉴权计费。GPU 任务和 HTTP 任务抢同一个进程GIL 和 CUDA 互相伤害。网关要知道「现在有谁能跑」你却把能力写死在代码里扩容、摘节点都要改配置文件。采用控制面与执行面拆分后职责边界可以这样划分控制面网关 鉴权、限流、计费 任务接收与路由决策 运行态与终态裁决 能力可用性判断 管理端 执行面Worker 插件加载与能力注册 按类型消费消息 推理执行 结果上报内部回调或受控写库在这套边界中网关进程通常不需要插件目录执行器也不一定要直接连接任务状态使用的 Redis / PostgreSQL更常见的做法是通过内部回调提交结果。Worker 直接条件更新任务表同样成立链路更短但需要下发数据库权限并统一状态机实现。插件如何被发现文件、包还是显式注册一个插件可以对应一个.py文件也可以对应一个包关键是启动时由加载器发现并注册能力而不是由网关硬编码。参考加载器可能在执行器启动时扫描插件目录逐个加载.py并执行注册逻辑文件名因此可以不是常规模块名。若采用包结构还需要处理相对导入、命名冲突和入口模块选择并不总是“扫目录就能 import”那么简单。某个文件语法错误时一种策略是只记录 warning让其他插件继续注册代价是错误类型会缺席能力列表。另一种选择是启动时遇错即失败适合要求插件集合完整一致的环境。修改插件后可以检查启动日志里的Registered task types。少了一个常见原因是文件名以下划线开头被跳过或加载时发生异常。如果插件加载器使用runpy.run_path(..., run_name__main__)这类执行方式模块级if __name__ __main__:会在加载时被触发若采用标准import module则不会。无论采用哪种加载方式插件文件都不应在模块级发起真实外部调用否则启动阶段就可能去撞供应商。生产环境可以把插件挂到外部目录减少为每个模型重建镜像的频率代价是镜像内容不再完全决定运行版本需要额外校验插件签名、来源和版本。强调可复现部署时把插件随镜像发布更合适。最小冒烟插件长这样fromcore.plugin.registryimportregisterfromcore.shared.credential_modeimportCredentialModeregister(hello,worker_threads1,credential_modeCredentialMode.NONE,metadata_version1,timeout_seconds60,description无凭证 echo用来冒烟。,)definfer(model_spec,payload,inference_params):return{message:Hello world,echo:payloador{},model:model_spec.get(name)orhello,}创建任务curl-sS-XPOST http://127.0.0.1:80/api/v1/tasks\-HContent-Type: application/json\-HX-API-Key:$API_KEY\-d{ task_type: hello, payload: {message: plugin-ok} }在本文信封协议中credential_modenone表示创建路径会把凭证字段清空后再入队插件不应去账户池拿 key。这不是所有系统的通用规则而是能力元数据与入队信封之间的约定。很多网关会把 OpenAI 兼容接口做成「看起来同步」内部仍是这条hello任务类型包了一层皮。协议适配是皮插件执行是骨。register只往两张表里塞东西插件注册既可以采用基类继承与inspect扫描也可以采用装饰器。基类便于静态约束和 IDE 提示但基类签名变化会影响全部插件装饰器对文件名和函数实现限制较少代价是函数签名主要靠运行时校验。下面使用装饰器方案。全局就两张表函数 元数据。PROVIDERS:dict[str,tuple[Callable,int]]{}PROVIDER_METADATA:dict[str,TaskTypeMetadata]{}defregister(task_type:str,worker_threads:int1,*,credential_mode:strcaller_or_pool,metadata_version:int0,timeout_seconds:int|NoneNone,description:str,payload_schema:dict|NoneNone,):defwrapper(func):PROVIDERS[task_type](func,worker_threads)PROVIDER_METADATA[task_type]TaskTypeMetadata(task_typetask_type,credential_modecredential_mode,metadata_versionmetadata_version,timeout_secondstimeout_seconds,descriptiondescription,payload_schemapayload_schema,)returnfuncreturnwrapper注册表不约束函数签名。约定是执行器这边定的(model_spec, payload, inference_params)。以后若有别的宿主可以自己换调用方式不必改装饰器。调度器和能力心跳只通过查询接口拿函数、线程数、超时、元数据不 import 任何一个插件文件。这很重要主进程启动路径不要把 ComfyUI、torch 拖进来那些东西留给子进程。生产插件长什么样聊天类是高并发 HTTP不是 GPUregister(openai-chat,worker_threads10,credential_modecaller_or_pool,metadata_version1,timeout_seconds600,descriptionOpenAI Chat支持文本、图片和文件片段。,)definfer(model_spec,payload,inference_params):clientOpenAI(api_keyresolve_api_key(model_spec))timeoutget_plugin_timeout_seconds(600)respclient.chat.completions.create(modelmodel_spec.get(name)orgpt-4o,messagespayload.get(messages)or[],timeouttimeout,**(inference_paramsor{}),)return{text:resp.choices[0].message.content}在本文示例实现中worker_threads10同时驱动两件事该类型独立进程里的 10 线程池以及主进程对该类型的 prefetch / 槽位上限。这是为了避免消费者预取过量导致槽位错配不是线程池大小与 MQ prefetch 的自然等价关系。很多系统里 prefetch 还会单独根据消息大小、ack 时机和吞吐目标调整。credential_modecaller_or_pool的协议含义是调用方带 key 时使用调用方凭证未提供时允许入队并在执行期向账户池申请。因此在该模式下仅因请求未带 key 而返回 400 会与元数据语义冲突。若某类任务要求调用方自带凭证应使用独立的caller_required模式而不是改变caller_or_pool的含义。分割、出图这类 GPU 任务把worker_threads留 1避免两路推理抢同一块显存register(sam2,worker_threads1,credential_modenone,metadata_version1,timeout_seconds300,)definfer(model_spec,payload,inference_params):endpointmodel_spec[endpoint]timeoutget_plugin_timeout_seconds(30)# 下载图片别写死imagedownload(payload[image_url],timeouttimeout)returncall_segment(endpoint,image,payload.get(box))同文件可以放多个register每个装饰器向注册表写一条调度器会按task_type创建进程和队列。若多个类型需要共享大模型权重把它们放进同一进程可节省显存若更重视故障隔离则按类型拆分。文件组织本身不应被误当成隔离边界。启动顺序与失败条件执行器的启动顺序需要与对外就绪语义一致下面是一种安排连队列检查 broker 真的支持重试和死信打日志注册了哪些类型每个类型一个 Process示例中的隔离粒度也可改为共享进程或独立容器每个类型一条结果监听线程每个类型一个 asyncio consumer开始对网关心跳对应代码可以记这个骨架asyncdefstartup(registry,queue_client):awaitqueue_client.connect()ifnotqueue_client.supports_retry_queue:raiseRuntimeError(broker 不支持重试拓扑拒绝启动)task_typesregistry.get_registered_task_types()logger.info(Registered task types: %s,task_types)schedulerInferenceScheduler()scheduler.initialize_workers(registry)awaitscheduler.start_result_monitor()fortask_typeintask_types:asyncio.create_task(consume_for_type(task_type),namefconsumer-{task_type})capability_client.start()在以instance_id作为 Redis 主键的实现中每个存活实例需要使用不同 ID例如forwarder-001否则后写会覆盖前写。这是当前存储模型的协议约束。若改用数据库复合键或服务发现平台生成实例身份则可由基础设施保证唯一性。关闭顺序应先停止接收新任务并摘除能力再等待在途任务到宽限期最后停止子进程。示例可以先停心跳、取消 consumer、发送停止哨兵并在 join 超时后 terminate5 秒并非通用值应覆盖可接受的任务收尾时间。对不可中断的 GPU 推理还需要 checkpoint、任务重投或更长的 termination grace period。Worker 隔离线程、进程还是容器当推理会阻塞事件循环或需要独立故障域时可以放到子进程。调度器初始化示例如下frommultiprocessingimportProcess,Queue,Pipefortask_typeintask_types:task_queueQueue()result_queueQueue()recv_conn,send_connPipe(duplexFalse)threadsregistry.get_worker_threads(task_type)timeoutregistry.get_timeout_seconds(task_type)processProcess(targetrun_inference_worker,args(task_type,threads,timeout,task_queue,result_queue,send_conn),daemonTrue,)process.start()workers[task_type]{process:process,task_queue:task_queue,result_queue:result_queue,notify_conn:recv_conn,worker_threads:threads,active_tasks:0,}主进程只负责 MQ pull、槽位记账、进程间投递、收结果和回调。这样 CUDA 泄漏、SDK 卡死、GIL 占满通常局限在对应类型的子进程。不过进程仍共享宿主机、文件系统和 GPU 驱动无法提供容器级资源与安全隔离。三种粒度可以这样选同进程线程池适合可信、以 I/O 为主且依赖兼容的插件收益是通信开销小代价是崩溃和内存泄漏可能影响其他类型。一类型一进程适合需要 GIL/CUDA 上下文隔离、但仍希望单机统一调度的场景收益是故障面较小代价是模型内存可能重复加载IPC 和进程监管更复杂。一类型一容器或独立服务适合依赖冲突、资源限额和安全边界明确的场景收益是独立扩缩与资源配额代价是部署、网络和观测成本更高。子进程入口再分两种模式defrun_inference_worker(task_type,worker_threads,timeout,task_q,result_q,conn):ifworker_threads1:run_single_thread(task_type,timeout,task_q,result_q,conn)else:run_thread_pool(task_type,worker_threads,timeout,task_q,result_q,conn)单线程循环task_queue.get()阻塞 → 调infer()→result_queue.put→ pipe 写 1 字节叫醒主进程。适合 GPU也适合「虽然是 HTTP 但不要在同进程里并发把连接打爆」的场景。多线程主线程只分发ThreadPoolExecutor跑任务。适合 OpenCV / NumPy / 会释放 GIL 的 C 扩展。纯 Python GIL 的插件把 threads 开到 10收益有限还是会互相挡。别看配置好看就开满。子进程里只 import 真正跑推理的那一层不要把 FastAPI、OpenTelemetry 再初始化一遍。你会在日志里看到两套 tracer排障时对不上。子进程里真正调插件的地方definfer(task_id,task_type,model_spec,payload,inference_params,timeout_secondsNone,client_roleNone,mq_attempt1):funcregistry.get_inference_function(task_type)iffuncisNone:return{task_id:task_id,status:FAILED,error:fNo inference function for{task_type},}try:withaccount_pool_context(client_roleclient_role,mq_attemptmq_attempt),\ plugin_execution_timeout(timeout_seconds):outputfunc(model_spec,payload,inference_params)return{task_id:task_id,status:SUCCESS,result:{output:output}}exceptExceptionase:return{task_id:task_id,status:FAILED,error_code:E4000001,error:fInference failed:{e},}几点实际约定插件返回值可以是普通 dict也可以带账户池计费元数据。执行器会拆开网关回调时记账。子进程中的异常不会自动进入主进程的 ASGI/FastAPI 异常处理中间件。它只能通过返回值协议、IPC、返回码、状态文件、进程监控或重试队列上报。在本文的返回值协议中子进程应返回 FAILED dict主进程才能回调网关并按mq_action处理消息。若异常未捕获而进程退出主进程还需要存活检测、重启或摘除能力消息则保持 unacked 等待重投。超时可以用 ContextVar 传播语义插件内部get_plugin_timeout_seconds()能读到这次任务的超时配置。ContextVar 只负责传参不负责强制终止线程超时在 CPython 里不可靠真正的硬超时仍依赖外层 deadline、客户端 timeout、任务重试预算和网关task_deadline_at。插件里的 HTTP client 应读取这个值而不是写死 600。账户池上下文只在需要池凭证时有意义。hello这类none模式不会申请 Redis 租约需要凭证池的插件则应遵守分配、续租和释放协议。信封是网关组的。插件不要自己去 MQ 里再读一遍也不要假设能 import 网关的任务服务。字段就这几个model_spectask_data.get(model_spec)or{}payloadtask_data.get(payload)or{}inference_paramstask_data.get(inference_params)or{}mq_attempttask_data.get(mq_attempt,1)client_roletask_data.get(client_role)能力发现心跳、注册中心或静态配置网关不应依赖扫描本地磁盘来发现能力。执行器可以通过 HTTP 心跳让控制面知道codex-chat存在服务注册中心、Kubernetes API、管理员手工标记或静态配置也能承担发现职责。心跳实现直接但需要 TTL 治理注册中心能力更完整但增加基础设施依赖静态配置则适合节点集合稳定的小规模部署。每轮 POST 一份body{instance_id:instance_id,task_types:[hello,openai-chat,sam2],task_type_metadata:{hello:{credential_mode:none,timeout_seconds:60,metadata_version:1,description:无凭证 echo用来冒烟。,},...},started_at:started_at,updated_at:now.isoformat(),heartbeat_interval_seconds:10,}awaithttp.post(heartbeat_url,jsonbody,headers{X-Internal-Key:secret})网关把实例写进 Redisforwarder:capability:instance:{instance_id} # JSONTTL 覆盖心跳间隔与允许的抖动 forwarder:capability:task_type:{task_type} # SET of instance_id forwarder:capability:task_type_metadata_map # 对外可见的类型元数据同一task_type被多个实例注册时示例按metadata_version选择较新元数据。修改description或credential_mode时需要递增版本否则滚动发布期间旧实例心跳可能覆盖新字段。替代方案是按部署 revision 分组、拒绝同一能力的混合版本或让控制面保存静态元数据这些方案一致性更强但会降低异构实例并存的灵活性。实例消失可由 TTL 到期判断正常关闭时再 DELETE。若采用租约式心跳TTL 应大于心跳间隔并覆盖网络抖动与调度延迟具体倍数由可接受的摘除延迟决定。TTL 太短会误摘太长则会把已失效能力保留更久。创建任务时网关先问这个类型现在有没有 live 能力capabilityawaitget_public_task_type(client_task_type)ifcapabilityisNone:raiseInvalidTaskType(fUnsupported task type:{client_task_type})credential_modecapability[credential_mode]示例在没有 live 能力时拒绝创建以避免无消费者消息长期积压。这适合“提交即要求当前可执行”的接口但不是唯一语义允许离线排队的系统可以先接受任务等待能力恢复使用 RabbitMQ mandatory publish 和 alternate exchange 可以把无法路由的消息退回或停放静态白名单还可以区分“合法但暂时无实例”和“协议中不存在的类型”。只有在消费者确认永久不支持且没有转交路径时死信才是合适终点。公开列表给管理端/调用方探活。内部还能查某个类型有没有实例。别指望心跳一停就把队列删掉——RabbitMQ 里空队列留着比误删有积压的队列更安全。从创建到进进程整条链路长什么样可以记这张图调用方 task_typeopenai-chat ▼ 网关 Redis 里该类型 availabletrue读到 caller_or_pool │ 有凭证就规范化后入信封没有仍入队 ▼ 消息进 task.queue.openai-chat ▼ 只有加载了 openai-chat 插件的执行器在消费 ▼ 有槽位 → 放进该类型子进程的 Queue → infer() ▼ 结果回调网关mq_action 决定 ack / retry / dead在这套按类型声明队列和进程的实现中「聊天任务进了出图进程」由 routing key 与消费者绑定共同避免。新模型对应新文件、新task_type和新队列拓扑共享队列方案则需要消费者在接单前校验能力并处理误路由。两台执行器的插件集合可以不同。A 有sam2B 没有时B 通常不订阅sam2队列误消费时按协议拒绝或转交。按能力给节点分组能独立扩缩统一加载全部插件则配置更简单但资源占用更高。凭证模式属于能力元数据不是网关 if-else 表一种能力元数据可以定义三类模式模式典型任务创建时网关做什么nonehello、sam2清空 credential不进池caller_or_poolopenai-chat、codex-chat有 key 用调用方没有仍入队执行期租池其它以解析函数为准未知值不要 silently 当 none把模式写在插件上的好处换一个 OAuth 插件不必改网关代码。心跳把 mode 带到 Redis创建路径就能读到。timeout_seconds同样来自插件。网关还有请求体超时和全局默认实际 deadline 是创建路径里算的。插件超时用于执行侧约束 HTTP/推理网关超时用于任务截止。两套都要看只配一个会出现「插件还在跑、任务已被 deadline 判失败」。新增任务类型的接入流程丢一个my-model.py文件名随意。register(my-model, worker_threads..., credential_mode..., timeout_seconds..., metadata_version1)装饰一个def infer(model_spec, payload, inference_params)。模块级不要连供应商连供应商放进函数里方便加载失败时至少进程能起来。重启该执行器实例启动时 load 一次不是热加载。日志出现Registered task types含my-model。等一次心跳。能力列表或 Redis 里能看到。POST /tasksbody 里task_type: my-model。若采用「一类型三队列」拓扑RabbitMQ 可能出现主队列、重试队列和死信队列具体数量取决于 broker 拓扑设计。用 hello 的联调清单走一遍 ack再故意raise验证失败是否按既定队列协议进入 retry 或 dead。在动态能力发现方案中无需修改网关路由表。参考加载器不支持热更新因此改插件后要重启执行器若元数据按版本择新修改元数据时还需递增metadata_version。支持热加载或由注册中心管理版本时流程会不同。参考实现让执行器通过内部回调提交结果因此不需要 PostgreSQL 权限也不直接改终态表。另一种方案是执行器使用最小权限账号通过条件更新写终态收益是少一次 HTTP 跳转和更短的确认链路代价是状态机逻辑、数据库连接和凭证轮换进入每个执行器。无论选哪种worker_threads表示单实例并发而非集群总量。对确定性的非法 prompt若协议判定重试不会改变结果应标为不可重试若错误可能由模型版本或策略更新消除则可以进入有上限的延迟重试。怎么验证这套是通的启动执行器健康检查应 200。日志类似Registered task types: [hello, openai-chat, ...]以及Started N workers每个 worker 带 pid 和线程数。能力是否到网关。Redis 里应有实例 key以及task_type:hello这类 set。网关创建未知类型curl-sS-XPOST http://127.0.0.1:80/api/v1/tasks\-HContent-Type: application/json-HX-API-Key:$API_KEY\-d{task_type:does-not-exist,payload:{}}应返回类型不支持MQ 里不应出现对应队列拓扑是 publish 时才 declare失败路径不应 publish。hello 进程是否真的在跑。提交后日志应是收到 → 接下 → worker 打印 Processing → 内部回调。查结果output.echo应等于你的 payload。隔离。连续打 hello 的同时打一个故意time.sleep的本地插件临时 threads1。hello 的延迟不应跟 sleep 插件一样长。它们是不同进程、不同队列。如果一样长说明你把两个类型注册到了同一个函数或只起了一个 consumer。错误插件。在infer里raise RuntimeError(boom)。应回调 FAILEDerror 含Inference failed:。主进程还在健康检查仍 200。MQ 按网关mq_action走 retry 或 dead而不是把 FastAPI 打 500。停心跳。把执行器停掉等超过 TTL再 POST 原来的类型。应变成不支持。TTL 内仍可能创建成功——消息会堆在队列里实例回来继续消费。这是滚动发布窗口不是 bug窗口长度就是 Redis TTL。排障顺序Registered task types→ Redis 实例 key → 网关能否读到该类型 → 队列有没有 consumer。别一上来就怀疑模型 SDK。常见故障与排查线索插件里写了if __name__ __main__。若加载器采用run_name__main__的执行方式模块级脚本入口会在加载时被触发若采用标准 import则不会。无论哪种方式都不应在模块级发起真实外部调用。metadata_version 没加。在“按版本取新”的合并策略下滚动发布时旧心跳可能盖掉新字段因此修改凭证模式、超时或描述时需要递增版本。若注册中心以部署版本为准则遵循其冲突策略。实例 id 重复。两台 forwarder-001能力集合来回抖网关间歇性拒绝创建。线程超时当硬超时。CPython 杀不了卡住的线程。HTTP client 自己设 timeout任务截止靠网关 deadline。子进程未捕获异常。进程没了消息 unacked看起来像「队列堵住」。其实是 worker 死了。主进程要对子进程做存活检查挂了要拉起来或把实例摘掉。GPU 并发与显存预算不匹配。单模型占用接近整卡时可以从单线程起步支持动态批处理、MIG、模型并行或单次占用较低时多并发反而能提高利用率。线程数应由显存峰值、算子并发能力和延迟目标压测决定。凭证模式与创建校验不一致。caller_or_pool允许空凭证入队caller_required才要求请求携带凭证。创建校验应读取能力元数据避免把池凭证路径意外关闭。写在最后可插拔执行器中模型可以抽象成函数关键取舍是控制面和执行面拆到什么程度是否接受显式路由配置线程、进程、容器三种隔离的资源成本与失败边界心跳、注册中心或静态配置如何发布能力把这三块的契约写清楚后凭证租赁、队列确认和扩缩容才能在不依赖具体插件实现的前提下协作。最小实现可以从单进程、单线程和静态能力配置起步随后按故障隔离、吞吐和凭证治理需求增加进程、心跳与租约。若一开始就存在不可信插件、依赖冲突或严格 GPU 配额也可以直接采用容器隔离它增加部署成本却能提前建立清晰的资源边界。演进顺序应由主要风险决定。
