端边云统一推理运行时：Physical AI模型实时调度的工程解耦之路

实际运行一个物理世界的智能体时问题往往不在单个模型能不能跑通而在于端侧能不能扛住低时延推理、边缘节点能不能承接算力、云端大模型能不能在对的时刻介入同一个决策闭环。PhyAI 这个由北邮、北大、清华、明体科技等机构联合提出的端边云统一推理运行时瞄准的正是这一层工程问题把“算法应该部署在哪里”从应用代码中解耦出来让 Physical AI 任务可以按设备状态、时延预算、模型能力实时选择端、边、云中的任意一个位置完成推理。本文不会把它包装成一套官方功能清单而是从工程实现角度拆解它背后的运行机制并给出一个可以独立搭建的最小参考骨架用于理解运行时抽象、模型描述、调度策略和故障排查。需要先说明边界。高校联合项目常见的产出形态是论文、原型系统和开放接口对外可复现材料可能并不完整。因此下面的方案是围绕“PhyAI 想解决的问题”设计的一套通用实现思路不承诺与官方源码逐行一致。真正动手前应把本文当成一个学习支架再结合目标设备、模型格式和推理框架补齐细节。1. 先理解 Physical AI 为什么需要统一推理运行时1.1 Physical AI 推理和大模型文本推理是两个问题Physical AI 指的是能够感知三维物理环境、理解任务目标并生成动作指令的智能系统常见形态包括机械臂、移动机器人、自动驾驶、无人机和工业质检设备。它和 ChatBot 类型的推理有本质差异任务输入不是纯文本而是相机图像、点云、深度图、IMU、关节角速度、力传感器等多模态数据任务输出也不只是 token而是姿态、轨迹、夹爪开合量、避障路径等连续或离散动作。这里出现一个很容易被忽略的点同一个任务往往需要多个模型接力完成。例如机械臂完成一次“从桌上抓取螺丝刀并放入左侧盒子”的任务至少经历三层推理端侧目标检测判断螺丝刀在图像中的位置和角度。边缘深度估算与抓取点规划根据深度图预测抓取姿态。云端语义校验与异常重规划当螺丝刀被遮挡或处于混乱场景时借助更大的世界模型重新理解场景并调整动作序列。传统的模型服务框架通常把推理当成一次 HTTP 调用请求进去、结果出来很少考虑延迟分层和链路接力。Physical AI 场景则要求端、边、云模型在同一个任务里协作并且协作方式随环境动态变化。这不是单一模型的问题也不是单一设备的问题而是“把多设备组成一个推理整体”的运行时问题。1.2 端、边、云三层的算力边界必须用统一视图描述端侧通常指机器人或设备上的算力单元例如 Jetson、RK3588、STM32 加 NPU、工业电脑。边侧指靠近设备部署的小型服务器或边缘网关通常有独立 GPU 或较高性能 CPU。云侧则负责大规模训练模型、复杂语义理解或超长时域规划。三个层级的差异可以汇总到下面的表格维度端侧边缘节点云侧典型算力低功耗 CPU、NPU、边缘 GPU中高端 GPU、多核 CPU大规模 GPU 集群时延目标仅任务闭环范围内小网络模型可控制在 10ms 到 30ms 级别依赖网络通常 30ms 到 200ms数百毫秒到数秒级网络稳定性不依赖外网无线或有线局域网稳定性中依赖公网或专线不稳定因素更多能完成的任务单帧检测、低精度姿态、设备本地安全逻辑复杂感知、路径规划、多路视频流处理场景理解、多模态推理、全局任务调度不适合的任务大模型前端部署、全局建图、超大规模历史分析超大规模模型训练、长期任务分布式记忆毫秒级实时控制、无网络环境下的兜底如果每个层级都使用不同的推理 API、模型包装方式和日志格式上层应用就要写大量分支代码来判断“当前请求该找谁”。更麻烦的是一次网络抖动后从端侧切换到云侧若没有统一输出格式下游控制器可能因为拿到不同结构的数据直接崩溃。PhyAI 这类端边云统一推理运行时的核心思路是把“端侧算力、边缘节点、云服务”都注册为一组抽象的推理执行器向上层暴露同一个接口。上层只描述任务不直接指定某一个固定地址。这样端、边、云的角色就变成运行时的可调度资源而不是被写死在业务代码里的常量。1.3 统一运行时要回答四个关键问题模型装在哪里同一个模型能否压缩成端侧版本是否要同时保留边侧和云侧版本。算力给到谁在时间片内端侧模型、边缘任务、云交互应该怎样抢占资源或排队。数据去哪里不同模态的输入数据要先经过端侧过滤还是直接上传到边缘。如何保证切换后可用模型版本和设备地址如何注册、发现、过期。这四个问题听起来像是部署平台的问题但它们直接影响运行时接口。模型注册表不能只存版本和地址还要存输入输出格式、设备能力、校验方式。调度器不能只看响应时间还要看链路质量、电量、算力负载和任务优先级。只有把这些策略统一起来端边云协同才有可能做到可预测。2. 拆解 PhyAI 类系统的运行时骨架四层抽象虽然不同项目的内部实现不同端边云统一推理运行时大体可以拆成四层设备与硬件抽象层、模型注册层、推理调度层、链路同步与反馈层。下面按这四层解释核心职责和常见设计。2.1 设备与硬件抽象层让上层不知道 GPU、NPU、CPU 的差异Physical AI 设备端的算力组件差异非常大。同样是执行一个 YOLO 检测Jetson Orin 可以调用 CUDARK3588 要调用 RKNN而某些工控机只能靠 CPU 运行 OpenVINO。运行时的第一层工作就是把这些差异封装成统一的硬件能力描述。一个边缘节点注册到运行时中时至少要上报以下字段node_id: edge-gateway-01 region: workshop-a enabled: true capabilities: compute: gpu: available: true type: nvidia memory_mb: 8192 npu: available: false io: input_modalities: [rgb, depth, imu] output_modalities: [pose, trajectory] priorities: max_concurrent: 2 latency_budget_ms: 100这段描述的含义是该节点当前拥有可用 GPU 显存 8GB可以处理 RGB、深度和 IMU 三类输入最多同时执行两个推理任务并希望在 100ms 内完成。注册只是第一步运行时还要能主动探测心跳防止节点掉线后调度器仍把任务派给不可用节点。从工程角度看这层最需要统一的是“设备能力枚举”。很多项目后期出问题都是因为端侧模型文件能跑但缺少对硬件显存和并发度的管理任务一多就触发 OOM。运行时若能在任务调度前检查剩余显存和模型所需内存整个系统就会稳定得多。2.2 模型注册层一个模型要同时描述推理语义和部署元信息普通模型服务通常只关心“模型输入什么、输出什么”。端边云协同运行时还要关心同一逻辑模型在不同位置的实现。在实际项目中可以用逻辑名和一个模型清单解决版本混乱。例如“grasp_pose_estimator”在云端是完整大模型在边缘是剪枝后的中等模型在端侧是量化后的轻量模型。它们结构不同但对外承担同一个逻辑功能。模型注册项大致如下model: logical_name: grasp_pose_estimator version: 2025.06.01 input: - name: color type: tensor shape: [1, 3, 480, 640] - name: depth type: tensor shape: [1, 1, 480, 640] output: - name: poses type: list[pose] required_confidence: 0.8 placements: - node_type: edge model_file: ./models/grasp_pose_pose_edgetpu.onnx runtime: onnx estimated_latency_ms: 35 - node_type: cloud model_file: ./models/grasp_pose_pose_full.onnx runtime: trt estimated_latency_ms: 120这里的关键不是写字段而是表达一个理念同一个逻辑模型可以有多个物理实现运行时在调度时不依赖模型文件名而依赖逻辑语义。端侧模型是否被选择可以通过实际延迟、内存占用和置信度来打分而不是靠配置固定写死。2.3 推理调度层从固定调用变成带约束的前端控制器调度层负责回答“这一帧推理到底跑在哪一层”。最简单的策略是按模型名直接路由但 Physical AI 场景会面临更频繁的变化网络良好时可复用边缘在线模型网络断开后必须切到端侧模型边缘节点负载过高时高优先级任务可以抢占低优先级任务。调度器可以理解为一段决策代码。它评估任务、节点、网络、历史结果后返回一个执行计划class InferenceDispatch: def select(self, task, node_profiles, network_probe): # 1. 排除不可用节点 alive [n for n in node_profiles if n.enabled and n.health_ok] # 2. 优先找满足时延预算的节点 for n in sorted(alive, keylambda x: x.est_latency): # 3. 如果任务允许端侧执行且端侧模型置信度足够 if task.endpoint_preference edge and n.can_afford(task): return n # 4. 边缘或云端兜底 return self.cloud_fallback(task)单纯看代码并不复杂复杂的是决策条件的优先级。不同任务会有不同要求抓取姿态估计优先级极高切换云端会导致整个控制周期失效而场景语义理解的优先级相对低可以容忍更多延迟更适合丢给云端。调度层应该遵循一个简单原则运行时只负责找到满足约束的执行器不负责猜测任务内容。因此每个任务要显式声明自己的端边云偏好、置信度阈值、时延预算和回退策略才能在运行时形成可评估的策略。2.4 链路同步与反馈层端边云模型不能各跑各的端边云统一不只是请求转发还需要把推理结果以统一格式回传。上层控制器收到结果后需要知道结果是来自端侧轻模型还是云端大模型因为两者的置信度含义不同后续控制策略也不同。运行时可以在返回消息中增加一个统一消息头避免把调度细节散落到业务代码{ request_id: 8f3c2a9e-88c1-4b52-8f32-1f805bbcc1a7, inference_id: grasp_pose_estimator, executed_on: edge, model_hash: a1b2c3d4, latency_ms: 42, confidence: 0.91, output: { grasp_pose: { x: 0.32, y: -0.15, z: 0.2 } } }上面的 JSON 包含 request_id 和 model_hash可以为链路追踪提供基础。开发者排查问题时不用反复猜测“响应到底来自哪个模型”。在真实机器人系统中这一点尤其重要。若端侧模型输出了低置信度结果而运行时没有把节点来源标记出来控制器很可能会把错误的抓取点发送给机械臂。3. 搭建一个最小可运行的端边云参考运行时如果只停留在概念层面很难真正理解 PhyAI 的价值。下面用一个最小项目演示端边云统一推理运行时的调度过程。这个项目不依赖特定硬件用 Python 模拟端侧模型、边缘模型和云端模型三个执行器。3.1 环境准备与依赖推荐使用 Python 3.10 及以上版本。依赖尽量精简便于在普通开发机上运行python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn httpx pyyaml这里使用 FastAPI 和 Uvicorn 是为了快速起服务。如果你的实际设备不支持 Python可以用 C 或 Go 重新实现同样的接口核心逻辑不变。演示目录结构如下phyai-minimal/ ├── runtime/ │ ├── registry.py # 模型注册表 │ ├── dispatcher.py # 调度器 │ ├── worker_cloud.py # 云端模拟执行器 │ ├── worker_edge.py # 边缘模拟执行器 │ └── worker_device.py # 端侧模拟执行器 ├── config/ │ └── nodes.yaml # 节点和模型配置 ├── schemas/ │ └── task.py # 推理任务请求 └── main.py # FastAPI 入口这一节的主要目的是让后续的推理请求可以按照“端侧优先、失败转边缘、兜底转云端”的顺序被调度。3.2 定义推理任务的数据结构统一推理运行时最重要的数据结构是任务描述。任务描述必须包含输入、输出、允许的执行位置和回退策略。写一份 Task dataclassfrom dataclasses import dataclass, field from enum import Enum from typing import Any class NodeType(str, Enum): DEVICE device EDGE edge CLOUD cloud class TaskPriority(str, Enum): REALTIME REALTIME NORMAL NORMAL BEST_EFFORT BEST_EFFORT dataclass class InferenceTask: task_id: str model_name: str input: dict[str, Any] allowed_nodes: list[NodeType] field( default_factorylambda: [NodeType.DEVICE, NodeType.EDGE, NodeType.CLOUD] ) max_latency_ms: int 100 min_confidence: float 0.7 priority: TaskPriority TaskPriority.NORMAL这类结构的价值在于业务调用方不需要知道哪个服务在什么 IP 上只需要说明“我想做一次抓取姿态估计最多允许 100ms 延迟低于 0.7 置信度叫失败”。调度器拿到这些信息后自行决定把任务发给谁。3.3 模型注册表将节点信息写入配置在真实场景里设备节点是在运行时动态注册的。为了这个最小示例足够简单直接用 YAML 配置两个节点nodes: - id: device-wh01 type: device endpoint: http://127.0.0.1:9101/predict models: [grasp_pose_device_v1] enabled: true - id: edge-wh01 type: edge endpoint: http://127.0.0.1:9102/predict models: [grasp_pose_edge_v1, scene_understand_v1] enabled: true - id: cloud-01 type: cloud endpoint: http://127.0.0.1:9103/predict models: [grasp_pose_cloud_v1, scene_understand_v1] enabled: true配置里的三个节点都提供/predict接口。模型名从特定前缀到泛化名称表示端侧、边缘、云端都有同一个逻辑模型的实现。调度器只需要判断当前目标节点是否能处理该模型。3.4 实现最小调度器调度器核心逻辑是根据模型名找到能处理该模型的节点。按节点类型排序端侧优先。检查节点是否在线且允许处理。发起推理请求。若推理失败或置信度低于阈值回退到下一个节点。以下是一个可运行的调度器示例import httpx class Dispatcher: def __init__(self, nodes): self.nodes nodes def available_nodes_for_model(self, model_name): result [] for node in self.nodes: if not node[enabled]: continue if model_name in node[models]: result.append(node) return result def score_node(self, node): order {device: 0, edge: 1, cloud: 2} return order.get(node[type], 99) async def dispatch(self, task): candidates self.available_nodes_for_model(task.model_name) candidates.sort(keylambda n: self.score_node(n)) for node in candidates: try: async with httpx.AsyncClient() as client: response await client.post( node[endpoint], json{ model_name: task.model_name, input: task.input, }, timeouttask.max_latency_ms / 1000, ) if response.status_code ! 200: continue payload response.json() if payload.get(confidence, 0) task.min_confidence: continue return { request_id: task.task_id, node: node[id], node_type: node[type], result: payload, } except httpx.TimeoutException: continue return { request_id: task.task_id, node: None, error: all_nodes_failed, }端侧优先并不等于永远优先。这里的排序只解决默认情况如果任务里写明了max_latency_ms20但端侧模型最近一次平均延迟已经到 50ms调度器就应主动把端侧排除。真正生产级的调度器还需要把历史延迟、模型当前排队数、节点剩余显存都纳入评分。4. 让端、边、云模型在演示环境里跑通上一节只写了框架代码这一步把它变成三个可调用的 worker并验证调度结果是按端、边、云的优先级返回的。4.1 写三个模拟执行器三个执行器可以通过同一个基类实现只是节点类型和延迟不同。下面是一个用 FastAPI 写的边缘 worker 示例from fastapi import FastAPI from pydantic import BaseModel import random, time app FastAPI() class PredictRequest(BaseModel): model_name: str input: dict app.post(/predict) def predict(req: PredictRequest): time.sleep(0.03) # 模拟实际推理耗时 return { model_name: req.model_name, confidence: 0.93, executed_on: edge, latency_ms: 30, output: {pose: [1, 2, 3]} }端侧 worker 可以用更短的延迟但更低的置信度云端 worker 用更高延迟但较高置信度。这样演示是可行的因为它抽象的只是调度流程而不是模型本身。在真实项目中各节点要读取自己的模型文件并调用 PyTorch、TensorRT、OpenVINO 或 RKNN 推理引擎。这个最小示例的作用是隔离出“调度链路”。4.2 发起一次推理请求并观察路由跳转用一个 Python 脚本作为客户端import asyncio from runtime.registry import load_nodes from runtime.dispatcher import Dispatcher from schemas.task import InferenceTask, NodeType async def main(): nodes load_nodes(./config/nodes.yaml) dispatcher Dispatcher(nodes) task InferenceTask( task_idreal-request-001, model_namegrasp_pose_edge_v1, input{rgb: camera_01.png, depth: depth_01.png}, allowed_nodes[NodeType.DEVICE, NodeType.EDGE, NodeType.CLOUD], max_latency_ms100, ) result await dispatcher.dispatch(task) print(result) asyncio.run(main())正常情况下结果中会打印node_type: device因为端侧节点存在且能够处理grasp_pose_edge_v1模型。若把端侧节点在 YAML 里改成enabled: false再次运行调度器会回退到边缘节点。第三次运行时再把边缘节点关闭就能看到回退到云端。这个过程就直观体现了“端边云统一推理运行时”的一个价值上层代码完全不需要改只调节点配置推理执行位置就会变化。4.3 端边云模型版本如何避免冲突当端侧、边缘、云端同时存在同一个逻辑模型时模型版本不一致是常见问题。例如边缘节点还在跑 6 月版本云端已经切到 7 月版本如果调度结果没有带上模型哈希客户端很难意识到自己收到的结果来自旧模型。在演示场景中可以给每个节点返回model_hash字段。实际生产不要认为模型文件上传后自动生效建议在推理结果中同时返回模型加载时间和版本号由运行时统一存到日志系统。跟踪问题时先按 request_id 查询调度结果再把同一请求的模型哈希对比一下很快能找到是哪个节点返回了过期结果。4.4 验证结果时不要只看延迟统一运行时是否好用要同时验证三个维度正确性结果数据结构和预期一致。可控性调度器能按参数自动切换执行节点。可观测性日志能还原完整链路至少包含节点 ID、模型哈希、延迟和置信度。只看延迟会误导判断。例如端侧模型虽然快但物体遮挡严重时可能输出错误抓取点。调度器必须依赖置信度阈值判断端侧结果是否可接受。演示代码中min_confidence就是这一层约束的体现。5. 端边云协同推理最常见的故障与排查链路统一推理运行时的优势要到生产环境才显现但生产环境也会暴露一系列问题。下面按现象、原因、检查方式、处理建议整理。问题现象常见原因检查方式处理建议端侧一直不执行任务全跑到云端节点配置被禁用或端侧模型名不匹配检查nodes.yaml中 enabled 字段检查模型是否存在打开节点 enable 开关统一逻辑模型名边缘模型执行后返回低置信度结果输入图像尺寸、通道顺序不匹配查看端侧前置处理是否做 resize / 归一化在输入侧增加固定预处理层云端请求超时云端模型排队过长或网络抖动查看 worker 日志和平均排队时间增加超时时间或在端侧先做一次低精度兜底调度结果不稳定时快时慢节点列表包含无效节点检查/health接口清理不健康节点调度器定期探测健康状态回退后下游控制器崩溃不同节点返回的 pose 结构不一致对比不同节点的输出 JSON统一输出 schema增加版本字段5.1 从一次“云端过载”排查链路节点不健康问题这类问题更具体的排查步骤可以归纳为输入是否正确确认测试请求的model_name与节点注册模型完全一致。文件路径和命名确认模型名不存在空格、大小写或前缀不一致。依赖版本确认端侧推理框架版本与导出模型时一致。配置是否生效配置修改后需要重启 worker 或动态刷新注册表。权限、端口、网络检查每个节点是否能被主控端访问。日志是否出现明确异常优先看节点的最新日志而不是只看调度器日志。工具或框架限制确认端侧模型是否支持 INT8是否受硬件限制。实际项目里我会先加日志埋点再跑一个最小请求链路中任何一个环节缺失都会在日志里留下缺口。没有日志就往回追调度器 HTTP 状态码这是最直接的排查方式。5.2 端侧模型占用内存过高导致相机掉帧很多 Physical AI 项目会把模型不断叠加到端侧设备上目标检测一个模型、分割一个模型、姿态估计一个模型。最终结果不是单模型性能差而是设备总内存不足导致图像采集线程被系统杀掉。处理方式不是删模型而是引入运行时资源感知。任务调度前先检查节点内存剩余async def can_afford(node, model_memory_mb): return node[available_memory_mb] model_memory_mb 256建议端侧节点只保存当前主任务需要的少数模型边缘节点做大部分模型推理。端侧更适合做前置过滤避免把大量无效帧上传。5.3 边缘重启导致正在执行的推理任务失败边缘节点可能因升级或崩溃重启统一运行时如果只记录节点健康状态不停掉正在执行中的任务就会丢请求。应对方式是故障转移加任务序列号。节点重启后请求可以回退到云端同时把未完成的控制指令重新生成。遇到这类问题不能只加一个重试还要考虑重试后是否会造成机械臂重复执行同一动作。必要时用请求幂等 ID 做去重。6. 从演示走向生产端边云统一运行时的落地建议6.1 学习演示环境和生产环境的差距上面的最小示例能在普通电脑上运行但它离生产环境还有不小距离。把两者的差异对照起来会更容易看清后续该做什么环境角色配置方式需要考虑的关键因素开发环境单机进程模拟快速迭代不需要高可用测试环境容器化部署固定测试版本模拟低网速生产环境真实设备 边缘集群 云服务动态注册、监控告警、安全与权限、模型灰度关键运行要求状态可恢复节点重启后自动重新拉取模型并上线服务生产环境至少还要增加四块日志收集、健康检查、模型灰度、服务降级。没有监控的端边云协同系统几乎无法定位端侧掉线或模型切换引发的隐性错误。6.2 构建你自己的端边云统一运行时路线图如果是第一次在项目中落地这套思路不要一开始就追求完整生产平台。可以按以下阶段逐步推进先做逻辑模型分层把同一个任务在端、边、云的模型都命名成同一个逻辑名。做统一 API让所有推理节点都提供/predict和/health接口。写最小调度器先支持固定策略例如端侧优先、失败转边缘。加置信度控制不满足阈值时自动转下一个节点。加日志链路把 request_id、model_hash、node_id 全部串起来。再考虑编排系统用 Kubernetes、K3s 或自定义管理服务承载边缘节点。这套路线把抽象问题拆成了可验证的工程步骤。先在单机模拟里跑通第 1 到第 4 步再接入真实设备时才不会把硬件问题和调度问题混在一起。6.3 优化方向和更值得投入的实践点后续可以继续深挖的方向包括多感知模态融合后的数据路由、端侧模型压缩与量化、边缘节点多租户隔离、云端长时任务状态缓存、端侧离线模式下的安全动作兜底、模型发布的 A/B 切换。这篇文章的核心判断是PhyAI 提出的方向并不是把模型搬到端侧或云端而是让一套“逻辑任务”透明地运行在端边云组成的真实连续环境中。对于希望跟进的开发者最有价值的练习不是复现模型训练而是先写一个只有 3 个节点、几十行代码的最小任务调度器亲手观察端侧失败后任务如何回退到边缘。理解了这个过程再看到 Physical AI 项目中的复杂问题时就会知道问题通常不是单个模型不准而是多个模型和多个硬件节点没被同一个运行时视角管理起来。