大模型服务容灾实践：多模型集群混合调度+心跳探测异常识别自动故障切换.169

原创

未闻花名

发布于 2026-07-18 11:56:44

一、核心概念

1. 多模型混合调度

多模型混合调度，是在一套统一调度框架下，同时纳管本地私有化部署大模型、云端在线API大模型、以及多版本同类型模型的综合性资源管理方式。系统不会固定把所有推理请求绑定某一个单一模型，而是依据模型实时健康状态、响应延迟、业务场景优先级、成本预算限制等多重维度，智能分配每一次用户请求的推理落点。

从业务落地视角来看，单一模型部署存在天然短板：

本地模型受硬件显存、服务器负载、进程稳定性影响，容易出现卡顿、推理超时、进程崩溃等问题；
云端API则会面临网络波动、接口限流、调用额度耗尽、外网数据传输安全风险等隐患。
多模型混合调度的核心设计初衷，就是把两类模型的优势互补，规避各自单点缺陷，让整体大模型服务不再依赖某一个节点、某一个接口的稳定性。

它不只是简单做请求转发，还包含状态监控、阈值判定、流量编排、异常隔离、自动降级等一整套能力，是大模型从小量原型走向生产环境的必备中间件能力。

2. 故障切换

故障切换也常被称作故障转移、容灾切换，特指当主服务节点出现不可用状态时，调度系统无需人工登录服务器、无需手动修改配置，自动将业务流量迁移至备用服务节点的机制。

在大模型场景中，故障判定不只是简单的服务宕机，还包含多层异常场景：接口请求超时、返回报错码、推理结果为空、连续多次响应失败、延迟远超正常阈值、模型进程 OOM 卡死等，都属于需要触发切换的故障范畴。

无缝切换是故障切换的核心要求，切换过程中前端用户无感知、对话上下文不中断、请求不需要重新发起，调度引擎在底层完成重试与转发。同时支持两种模式：故障永久切换、故障恢复自动切回，适配不同企业的运维习惯与业务 SLA 要求。

3. 本地模型与云端API对比

3.1 本地私有化大模型

3.1.1 核心优势

数据安全可控：用户提问、业务私有数据全程在内网流转，不向外网传输，完全满足政企、金融、政务行业的数据合规要求。
推理低延迟：内网局域网调用无公网网络损耗，短文本问答、知识库检索、简单指令类推理响应速度远优于云端API。
无按量计费：一次性硬件部署后，无单次调用费用，高频内部办公、企业知识库场景长期成本更低。
自定义自由度高：可自由微调模型、定制Prompt模板、接入私有知识库，不受云端模型能力与版本限制。

3.1.2 常见短板

硬件门槛高：需要高性能GPU显卡、充足显存与服务器运维资源，初期投入成本高。
运维复杂度高：需要专人维护模型启动、显存清理、进程守护、版本更新，容错自愈能力弱。
并发能力有限：受单机硬件限制，高并发场景容易排队、超时，扩容需要新增服务器节点。

3.2 云端API大模型

3.2.1 核心优势

超高可用性：厂商提供集群化部署，自带容灾扩容，日常几乎不会出现整体服务瘫痪。
零运维成本：无需关心模型部署、显卡维护、版本迭代，直接调用接口即可使用。
模型能力更强：通用知识、多轮对话、长文本理解、复杂逻辑推理能力普遍优于普通本地开源模型。
弹性扩容无忧：面对突发流量自动扩容，无需企业自行规划硬件资源。

3.2.2 常见短板

按量调用收费：使用量越大成本越高，高频大规模业务长期开销不可忽视。
网络依赖严重：公网延迟、跨区域网络抖动会直接影响对话体验。
数据外泄风险：业务敏感数据需上传外网接口，无法满足高密级隐私场景合规要求。
存在限流与风控：接口有调用频率限制、IP风控、内容审核限制，部分业务场景受限。

4. 双活架构解析

双活架构区别于传统的主备冷备架构：

传统冷备模式下备用模型平时处于闲置状态，只有主模型彻底故障后才会启动接管；
而本地 + 云端双活架构下，两个模型始终保持实时在线、持续健康探测、随时具备流量接管能力。

双活架构的关键特征：

两个服务节点地位对等，互相作为对方的容灾备份，调度引擎实时同步两者的健康状态、平均延迟、错误率等指标。
正常业务时段，可以按流量比例分摊请求；任意一方出现异常，立刻自动隔离故障节点，全部流量平滑切换至正常节点。

这种架构彻底消除了大模型服务的单点故障隐患，把整体服务可用性提升到生产级99.9%以上标准，也是政企、企业办公、智能客服等核心业务的首选部署架构。

5. 健康探测核心作用

健康探测是多模型混合调度系统的感知中枢，相当于整个架构的体检仪与哨兵。它会按照固定时间间隔，向本地模型服务、云端 API 接口发送轻量化探测请求，持续采集每一个模型节点的连通性、响应状态、推理延迟、错误次数等关键指标。

探测请求不会使用复杂长文本，一般采用极简固定Prompt，既不占用模型推理资源，又能精准验证服务是否可用。系统会内置阈值规则：连续探测失败次数、单次最大响应超时、平均延迟上限等，一旦指标突破阈值，立刻将该模型标记为不健康状态，暂停分配新流量，进入故障隔离状态。

如果没有健康探测机制，调度系统只能被动等待用户请求报错后再做补救，无法提前预判故障、提前隔离风险，极易造成大批量用户请求失败，严重影响业务体验。

6. 自动切流机制说明

自动切流是调度引擎根据健康探测结果，动态调整流量分配比例与流向的核心能力。它不是固定配置死路由，而是动态自适应调整：

当本地主模型健康状态正常、延迟稳定时，默认将绝大多数业务流量分配给本地模型，兼顾低延迟与数据安全；
当主模型延迟飙升、连续探测失败，自动缩减直至切断主模型流量，全部切换到云端备用模型；
当双模型都处于健康状态时，可配置权重分流策略，比如70%流量走本地、30%流量走云端，实现负载均衡与压力分摊。

同时自动切流支持场景化路由配置：

简单日常问答、内部知识库检索优先走本地；
复杂逻辑推理、长文本创作、多模态需求自动路由到云端强能力模型，实现能力与资源的精准匹配。

二、混合调度基础原理

1. 基础原理

基于三层核心分层架构，多模型混合调度整套体系严格分为三层，每层职责边界清晰、解耦独立，便于后续迭代扩展和运维维护。

1.1 第一层：接入层

作为所有用户请求的统一入口，承接前端页面、客户端、业务系统、第三方集成的全部大模型推理请求。
核心职责包含：统一请求协议解析、请求参数校验、身份权限鉴权、请求日志记录、接口协议适配转换。
接入层屏蔽了后端多个模型的接口差异，前端只需要对接统一入口，无需关心背后是本地模型还是云端模型，实现前端与底层模型解耦。

1.2 第二层：调度引擎层

这是整个架构的核心大脑，也是多模型混合调度与故障切换的核心实现层。
核心包含五大模块：健康状态管理模块、故障规则判定模块、流量策略编排模块、主备切换执行模块、日志监控统计模块。
所有模型的状态采集、阈值判断、切流决策、故障隔离、恢复回迁，全部由调度引擎集中计算与执行，是整套高可用体系的控制中枢。

1.3 第三层：模型执行层

作为实际承担推理任务的底层服务集群，包含两类核心节点：

一是本地开源大模型集群，部署在企业内网服务器，如Qwen、Llama、ChatGLM、DeepSeek等私有化部署服务；
二是云端第三方API服务，包含各大厂商开放大模型接口。

执行层只负责接收调度引擎转发的请求、完成推理、返回结果，不参与任何调度决策，专注推理能力本身。

2. 高可用底层原理

彻底消除单点故障，大模型服务高可用的底层逻辑，本质就是冗余部署 + 实时监控 + 自动转移，通过三层机制彻底打破单点依赖,。

首先是节点冗余，不再依赖单一模型、单一服务器、单一外网接口，同时部署本地与云端两套独立推理服务，形成物理层面的双节点冗余。
其次是实时状态监控，通过定时健康探测，不间断监控每一个节点的运行状态，及时捕捉异常苗头，做到早发现、早隔离。
最后是流量自动转移，一旦某一节点判定故障，调度引擎立刻停止向该节点分发新请求，把后续所有流量自动转移至正常备用节点，全程无需人工介入。

这套原理可以规避各类常见故障：服务器宕机、GPU显存溢出、模型进程卡死、公网网络中断、云端接口限流、账号密钥过期等，任意单点故障都不会导致整体业务瘫痪。

3. 对大模型应用的意义

当下大模型落地从实验阶段转向产业落地，单一模型部署的弊端已经完全暴露：本地模型稳定性不足、云端 API 存在安全与成本隐患，都无法单独支撑企业核心业务长期运行。

多模型混合调度架构的出现，从根本上解决了三大核心痛点：

1. 保障业务连续性，实现全年高可用，避免因模型故障、网络波动导致智能客服、办公助手、知识库问答等业务停服；
2. 平衡数据安全与模型能力，敏感私有数据走本地内网推理，复杂高难度任务调用云端强能力模型，兼顾合规与效果；
3. 控制使用成本，优先消耗无费用的本地模型算力，仅在故障或高复杂场景调用云端API，大幅降低长期接口调用开销。

可以说，混合调度与故障切换，已经成为大模型应用落地不可或缺的基础设施。

4. 大模型应用中的价值

稳定性底座：为所有大模型上层应用提供底层容灾能力，屏蔽底层模型故障，让业务应用无需关心底层服务波动。
成本优化调节器：通过流量智能分配，最大化利用免费本地算力，最小化云端按量计费消耗，实现算力成本最优。
数据安全隔离墙：敏感业务数据限定在内网本地模型流转，非敏感通用任务使用云端模型，划分安全边界。
响应性能加速器：常规请求优先内网本地推理，规避公网延迟，大幅提升日常对话交互体验。
无人值守故障自愈：自动探测、自动切换、自动恢复回迁，减少人工运维介入，降低运维人力成本。

三、完整执行流程

1. 标准流程

步骤 1：用户发起对话/推理请求

包含对话 Prompt、上下文参数、业务场景标识等信息，统一提交至系统接入层。

步骤 2：接入层接收请求

接入层完成请求合法性校验、参数格式解析、用户权限校验，过滤非法请求与无效参数，同时记录请求日志与时间戳。

步骤 3：调度引擎查询模型实时健康状态

请求转发至调度引擎，引擎实时调取所有已接入模型的最新健康状态、平均响应延迟、当前错误率等监控数据。

步骤 4：按策略选择最优模型

调度引擎按照预设策略做路由决策：优先匹配业务场景路由、再判断主备模型健康状态、最后结合延迟与权重选定目标模型。

步骤 5：请求转发到对应模型

调度引擎将请求协议适配转换，转发至选定的本地模型或云端API服务节点。

步骤 6：模型返回结果

底层模型接收请求后执行推理计算，生成对话结果、token消耗、推理耗时等数据，原路返回至调度引擎。

步骤 7：结果返回用户

调度引擎统一封装返回结果格式，屏蔽不同模型的返回结构差异，标准化后回传给接入层。

步骤 8：记录日志、更新监控、更新健康状态

接入层将结果返回前端用户，同时完成全链路日志归档、性能指标统计、模型状态数据更新，完成一次完整请求闭环。

2. 故障切换触发流程

1. 健康探测连续失败多次
- 健康探测模块按固定间隔发送轻量化检测请求，连续多次收到超时、报错、无响应等异常结果。
2. 系统标记模型 = 不健康
- 系统累计失败次数达到预设阈值，触发故障判定规则，将当前主模型标记为不健康故障状态。
3. 触发切换策略
- 调度引擎立即执行故障隔离，停止向故障模型分配任何新的业务请求，避免更多用户请求失败。
4. 流量自动切向备用模型
- 自动触发流量切换逻辑，将全部待处理请求和新增请求，无缝转发至备用模型节点。
- 切换过程中对正在处理的请求做内部重试，保证用户会话不中断、无感知。
5. 持续探测主模型
- 健康探测持续后台轮询故障主模型，一旦连续探测多次恢复正常响应，判定服务自愈。
6. 主模型恢复，自动切回
- 可配置自动回迁策略，在业务低峰期逐步把流量切回主模型，恢复默认调度策略。

3. 健康探测执行流程

1. 固定时间发送探测请求
- 系统初始化时，为每一个本地模型、云端API配置独立探测参数：探测间隔、超时时间、失败阈值、恢复阈值。
2. 判断响应是否正常
- 后台定时任务按间隔自动发起探测，采用统一极简Prompt，不消耗大量推理资源。
3. 判断延迟是否超限
- 记录每一次探测的连接状态、HTTP 状态码、推理耗时、返回内容有效性。
4. 连续失败次数累计
- 单次探测异常仅做计数累加，不立即标记故障，避免网络瞬时抖动造成误切换。
5. 达到阈值，标记故障
- 连续失败次数达到设定阈值后，正式标记模型为不健康，纳入故障隔离列表。
6. 连续成功，标记恢复
- 对已故障模型保持低频持续探测，一旦连续多次探测成功，重置失败计数，恢复为健康可用状态。

四、应用实践

以下是基于本地+云端双活大模型调度系统示例，实现健康检查、自动故障切换和流量调度功能。当本地主模型异常时自动切换至云端备用模型，保障服务高可用性。我们以不可用在线大模型为主模型，在经过失败验证后自动切换到本地大模型实现结构输出；

应用前启动本地大模型接口http://192.168.3.6:8000/v1/chat/completions，保障接口正常可用：

1. 模型状态枚举

from enum import Enum
import time
import requests
from typing import Dict, Optional

class ModelStatus(Enum):
    HEALTHY = "healthy"
    UNHEALTHY = "unhealthy"
    PENDING = "pending"

2. 模型类：本地\云端统一封装

class Model:
    def __init__(self, name: str, model_type: str, endpoint: str, api_key: str = ""):
        self.name = name
        self.model_type = model_type  # local / cloud
        self.endpoint = endpoint
        self.api_key = api_key
        self.status = ModelStatus.PENDING
        self.fail_count = 0
        self.latency = 0.0
        self.max_fail_threshold = 3

    def invoke(self, prompt: str) -> dict:
        try:
            start = time.time()
            if self.model_type == "local":
                resp = requests.post(
                    self.endpoint,
                    json={"prompt": prompt, "stream": False, "temperature": 0.7},
                    timeout=100
                )
            else:
                resp = requests.post(
                    self.endpoint,
                    headers={"Authorization": f"Bearer {self.api_key}"},
                    json={"prompt": prompt},
                    timeout=10
                )
            self.latency = round(time.time() - start, 3)
            if resp.status_code == 200:
                try:
                    data = resp.json()
                    self._set_healthy()
                    return {"success": True, "data": data, "latency": self.latency}
                except Exception as e:
                    self._set_unhealthy()
                    return {"success": False, "error": f"JSON解析失败: {str(e)}", "raw": resp.text[:200]}
            else:
                self._set_unhealthy()
                return {"success": False, "error": f"HTTP状态码异常: {resp.status_code}", "raw": resp.text[:200]}
        except requests.exceptions.ConnectionError:
            self._set_unhealthy()
            return {"success": False, "error": "连接失败: 请检查模型服务是否已启动"}
        except requests.exceptions.Timeout:
            self._set_unhealthy()
            return {"success": False, "error": "请求超时"}
        except Exception as e:
            self._set_unhealthy()
            return {"success": False, "error": str(e)}

    def _set_healthy(self):
        self.status = ModelStatus.HEALTHY
        self.fail_count = 0

    def _set_unhealthy(self):
        self.fail_count += 1
        if self.fail_count >= self.max_fail_threshold:
            self.status = ModelStatus.UNHEALTHY

3. 核心调度引擎

class ModelScheduler:
    def __init__(self, primary: Model, backup: Model):
        self.primary = primary
        self.backup = backup

    def health_check(self):
        self.primary.invoke("你好")
        self.backup.invoke("你好")
        print(f"【主模型】{self.primary.name} 状态：{self.primary.status.value}")
        print(f"【备用模型】{self.backup.name} 状态：{self.backup.status.value}")

    def dispatch(self, prompt: str) -> dict:
        # 优先尝试主模型（即使状态不是HEALTHY，也尝试一次）
        if self.primary.status == ModelStatus.HEALTHY:
            print("流量正常调度：使用本地主模型")
        else:
            print("主模型状态异常，但仍尝试调用...")
        
        res = self.primary.invoke(prompt)
        if res["success"]:
            print(f"主模型调用成功，延迟: {res.get('latency', 'N/A')}s")
            return res
        else:
            print(f"主模型调用失败: {res.get('error', '未知错误')}")

        # 主模型失败，切换到备用模型
        print("检测到主模型异常，自动切换至云端备用模型")
        backup_res = self.backup.invoke(prompt)
        if backup_res["success"]:
            print(f"备用模型调用成功，延迟: {backup_res.get('latency', 'N/A')}s")
        else:
            print(f"备用模型调用失败: {backup_res.get('error', '未知错误')}")
        return backup_res

4. 启动示例

if __name__ == "__main__":
    # 初始化云端模型（备）
    local_model = Model(
        name="chatglm2本地模型",
        model_type="local",
        endpoint="http://192.168.3.6:8000/v1/chat/completions"
    )

    # 初始化本地模型（主）
    cloud_model = Model(
        name="云端API",
        model_type="cloud",
        endpoint="https://api.example.com/v1/invoke",
        api_key="YOUR_API_KEY"
    )

    # 调度器
    scheduler = ModelScheduler(cloud_model, local_model)

    # 健康检查
    scheduler.health_check()

    # 执行请求（自动切流）
    result = scheduler.dispatch("请简要介绍大模型混合调度的作用")
    print("最终推理结果：", result)

5. 输出结果

【主模型】云端API 状态：pending 【备用模型】chatglm2本地模型状态：healthy 主模型状态异常，但仍尝试调用... 主模型调用失败: 连接失败: 请检查模型服务是否已启动检测到主模型异常，自动切换至云端备用模型备用模型调用成功，延迟: 7.352s 最终推理结果： {'success': True, 'data': {'id': 'chatcmpl-local', 'object': 'text_completion', 'model': 'ZhipuAI/chatglm2-6b', 'choices': [{'text': '大模型混合调度（如Distributed Unified Model-Agnostic Training，DUMA）是一种混合训练方法，旨在提高模型的训练效率和泛化性能。它的核心思想是将模型的训练分配到多个设备上，每个设备使用自己设备上的数据和模型进行训练。这种方法的主要作用有以下几点：\n\n1. 提高训练效率：将模型分布式训练可以避免在单一设备上训练模型时，需要等待所有设备完成数据预处理和模型预训练的时间。大模型混合调度可以将预处理和预训练分散到多个设备上，从而缩短训练时间。\n\n2. 提升泛化性能：大模型混合调度可以通过在不同设备上进行模型训练来提高模型的泛化性能。当模型在多个设备上训练时，设备之间的差异可以帮助模型更好地泛化到新的数据环境中。\n\n3. 更好的资源利用率：大模型混合调度可以更好地利用整个计算资源，避免在训练过程中出现资源浪费。通过将模型分布式训练，每个设备都可以利用自己设备上的资源进行训练，从而提高模型的训练效率。\n\n4. 可扩展性：大模型混合调度具有很好的可扩展性，可以通过增加更多的设备来进一步提高模型的训练效率和泛化性能。这种方法可以在大规模预训练模型训练中发挥重要作用，如BERT、RoBERTa等大型预训练模型。', 'index': 0, 'finish_reason': 'stop'}]}, 'latency': 7.352}

五、实践经验细节说明

1. 健康探测核心配置

健康探测不能只做简单连通，生产环境需要精细化参数配置，规避误判和漏判问题。

探测超时建议设置5~10秒，适配本地模型推理启动延迟和云端公网网络波动；
探测间隔设置3~5秒，既能及时捕捉故障，又不会频繁发送请求占用资源。
失败阈值推荐设置连续3次探测失败标记故障，避免瞬时网络抖动、单次接口波动造成不必要的流量切换；
恢复阈值设置连续5次探测成功再标记恢复，保证服务真正稳定后再回迁流量。
探测固定使用极简短文本Prompt，不使用长文本、复杂逻辑提问，减少模型推理负载，只做服务可用性校验即可。

2. 主流故障切换策略详解

即时切换策略：主模型单次请求失败，立刻切换至备用模型，适合对可用性要求极高、允许少量资源消耗的业务。
连续失败切换策略：累计多次请求失败再触发切换，过滤瞬时异常，避免频繁切流造成服务抖动，是最常用生产策略。
延迟阈值切换策略：不只是看是否报错，当模型平均响应延迟超过预设阈值，自动切流，保障用户交互体验。
灰度权重切换策略：故障后不一次性全切流量，按比例逐步迁移，防止备用模型瞬间承载过大流量引发雪崩。

3. 无缝切换底层实现原理

无缝切换的核心在于三层封装：请求重试封装、模型结果统一格式化、会话上下文透传。

调度引擎内部对失败请求做自动重试转发，用户前端无需重新发送请求；
屏蔽本地与云端模型返回字段、格式差异，统一输出结构；
保留对话历史上下文，切换模型后延续会话逻辑，用户无感知断层。

同时底层做故障节点请求熔断，不再向故障节点分发新流量，保护故障服务不会被大量请求压垮，给后台自愈留出恢复时间。

4. 多模型流量切分高级策略

除基础主备切换外，还支持多种高级编排：

主备独占模式：正常全部走主模型，故障全切备机；
双活权重分流：按7:3、8:2比例分摊流量，均衡两台模型负载；
业务场景路由：知识库检索、内部问答走本地，长文本创作、多轮复杂对话走云端；
优先级路由：付费用户优先调度高可用云端模型，普通用户优先本地模型，实现差异化服务。

七、总结

林林总总，到今天我们已经深刻的了解到本地模型兼顾数据隐私和低延迟，云端API拥有高可用、强推理能力，把两者做双活混搭，再配上健康探测、自动切流和无缝故障切换，刚好补齐了单一模型的所有短板。这套架构的核心逻辑并不复杂，本质就是靠多节点冗余兜底，用定时健康巡检实时监控模型状态，再通过调度引擎做智能流量分配，一旦主模型出现超时、报错、进程异常等问题，就能无感切到备用模型，全程不用人工干预。

通常我们在早期接触学习大模型只专注于模型调用和Prompt调试，很容易忽略高可用、容灾容错这类工程化细节，但真正落地做项目，稳定性远比单纯的对话效果更重要。我们上手前先吃透核心概念和三层架构逻辑，了解透基础调度流程；接着可以自己拓展探测阈值、权重分流、自动回迁这些高级配置；最后试着把本地私有化模型和主流云端API接入实战，慢慢积累生产级调优经验，真正彻底的把这些应用到生产场景中。

附录：完整应用示例

from enum import Enum
import time
import requests
from typing import Dict, Optional

class ModelStatus(Enum):
    HEALTHY = "healthy"
    UNHEALTHY = "unhealthy"
    PENDING = "pending"

class Model:
    def __init__(self, name: str, model_type: str, endpoint: str, api_key: str = ""):
        self.name = name
        self.model_type = model_type  # local / cloud
        self.endpoint = endpoint
        self.api_key = api_key
        self.status = ModelStatus.PENDING
        self.fail_count = 0
        self.latency = 0.0
        self.max_fail_threshold = 3

    def invoke(self, prompt: str) -> dict:
        try:
            start = time.time()
            if self.model_type == "local":
                resp = requests.post(
                    self.endpoint,
                    json={"prompt": prompt, "stream": False, "temperature": 0.7},
                    timeout=100
                )
            else:
                resp = requests.post(
                    self.endpoint,
                    headers={"Authorization": f"Bearer {self.api_key}"},
                    json={"prompt": prompt},
                    timeout=10
                )
            self.latency = round(time.time() - start, 3)
            if resp.status_code == 200:
                try:
                    data = resp.json()
                    self._set_healthy()
                    return {"success": True, "data": data, "latency": self.latency}
                except Exception as e:
                    self._set_unhealthy()
                    return {"success": False, "error": f"JSON解析失败: {str(e)}", "raw": resp.text[:200]}
            else:
                self._set_unhealthy()
                return {"success": False, "error": f"HTTP状态码异常: {resp.status_code}", "raw": resp.text[:200]}
        except requests.exceptions.ConnectionError:
            self._set_unhealthy()
            return {"success": False, "error": "连接失败: 请检查模型服务是否已启动"}
        except requests.exceptions.Timeout:
            self._set_unhealthy()
            return {"success": False, "error": "请求超时"}
        except Exception as e:
            self._set_unhealthy()
            return {"success": False, "error": str(e)}

    def _set_healthy(self):
        self.status = ModelStatus.HEALTHY
        self.fail_count = 0

    def _set_unhealthy(self):
        self.fail_count += 1
        if self.fail_count >= self.max_fail_threshold:
            self.status = ModelStatus.UNHEALTHY

class ModelScheduler:
    def __init__(self, primary: Model, backup: Model):
        self.primary = primary
        self.backup = backup

    def health_check(self):
        self.primary.invoke("你好")
        self.backup.invoke("你好")
        print(f"【主模型】{self.primary.name} 状态：{self.primary.status.value}")
        print(f"【备用模型】{self.backup.name} 状态：{self.backup.status.value}")

    def dispatch(self, prompt: str) -> dict:
        # 优先尝试主模型（即使状态不是HEALTHY，也尝试一次）
        if self.primary.status == ModelStatus.HEALTHY:
            print("流量正常调度：使用本地主模型")
        else:
            print("主模型状态异常，但仍尝试调用...")
        
        res = self.primary.invoke(prompt)
        if res["success"]:
            print(f"主模型调用成功，延迟: {res.get('latency', 'N/A')}s")
            return res
        else:
            print(f"主模型调用失败: {res.get('error', '未知错误')}")

        # 主模型失败，切换到备用模型
        print("检测到主模型异常，自动切换至云端备用模型")
        backup_res = self.backup.invoke(prompt)
        if backup_res["success"]:
            print(f"备用模型调用成功，延迟: {backup_res.get('latency', 'N/A')}s")
        else:
            print(f"备用模型调用失败: {backup_res.get('error', '未知错误')}")
        return backup_res

if __name__ == "__main__":
    # 初始化云端模型（备）
    local_model = Model(
        name="chatglm2本地模型",
        model_type="local",
        endpoint="http://192.168.3.6:8000/v1/chat/completions"
    )

    # 初始化本地模型（主）
    cloud_model = Model(
        name="云端API",
        model_type="cloud",
        endpoint="https://api.example.com/v1/invoke",
        api_key="YOUR_API_KEY"
    )

    # 调度器
    scheduler = ModelScheduler(cloud_model, local_model)

    # 健康检查
    scheduler.health_check()

    # 执行请求（自动切流）
    result = scheduler.dispatch("请简要介绍大模型混合调度的作用")
    print("最终推理结果：", result)

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

第四期热点征文-大模型技术

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

第四期热点征文-大模型技术

#大模型应用

#多模型集群混合调度

登录后参与评论

0 条评论

热度