1. 从“被动镜像”到“主动代理”一个范式转变的契机在工业物联网和智能系统的圈子里“数字孪生”这个词已经火了好几年。我们大多数人最初接触和实践的数字孪生本质上是一个被动镜像。什么意思呢就是我们在物理世界有一个设备比如一台机床、一条生产线或者一个风力发电机。然后我们在数字世界里用三维模型、传感器数据流和预设的仿真模型给它造一个“数字双胞胎”。这个双胞胎能实时反映物理实体的状态温度、压力、转速、位置。它能做可视化监控能做历史数据回溯甚至能基于预设规则做一些简单的预警。但它的核心是“反映”和“响应”它的行为逻辑是预先编程好的是对物理世界变化的被动映射。物理世界动一下数字世界跟着变一下就像一面镜子。然而当我们把视野从单台设备扩展到由无数异构设备、系统、甚至人组成的复杂网络时这种被动镜像的局限性就暴露无遗。网络是动态的、不确定的充满了突发流量、局部故障和不可预测的交互。一个被动的、只负责“看”的数字孪生在面对网络拥塞导致的控制指令延迟、某个节点异常引发的级联效应时往往无能为力。它只能事后告诉你“哪里出了问题”却无法在问题发生前主动调整、协同或补偿。这正是我们标题中“From Passive Mirrors to Active Agents”所指向的核心痛点我们需要数字孪生从被动的观察者转变为能够在网络中自主决策、主动协作的智能代理。这个转变的驱动力正是Physical AI over Networks网络化物理人工智能的兴起。Physical AI 不是指在云端跑一个大模型而是指AI能力下沉、嵌入到物理实体机器人、传感器、执行器和网络边缘让这些实体具备感知、学习和实时决策的能力。当这些智能体通过网络连接起来它们就形成了一个分布式的、协同的智能系统。这时传统的、中心化的、被动的数字孪生架构就成为了瓶颈。我们需要一种新的数字孪生范式它本身就是一个主动的、分布式的、具有认知能力的代理能够代表其物理对应物在网络中自主行动、与其他孪生体协商、并朝着整体系统目标优化。这就是Holonic Digital Twins合子数字孪生概念登场的时候。“Holonic”合子这个词源自“holon”由Arthur Koestler提出描述了一种“整体-部分”双重性系统。一个合子既是一个自包含的整体有自身的目标和功能同时又是更大整体的一部分。这完美契合了分布式网络系统的特征每个设备/节点是一个自主的智能体整体同时又是整个工厂、城市或交通网络更大整体的一个组成部分。合子数字孪生就是将这种哲学思想工程化每个物理实体或逻辑实体的孪生体都是一个具有自主性和协作能力的“代理”。它们通过网络相互连接形成层次化的、可递归的协作结构。底层的传感器孪生、设备孪生可以自主处理本地数据、做出快速反应它们又能向上聚合形成产线孪生、车间孪生在更高层级上进行更复杂的协同优化。这种结构天然适合处理网络化Physical AI场景中的复杂性、可扩展性和鲁棒性问题。2. 合子数字孪生的核心架构自主性与协作性的统一理解了“为什么需要转变”我们接下来拆解“如何构建”。一个合格的合子数字孪生其架构必须同时支撑两个看似矛盾的特性个体自主性和系统协作性。这不仅仅是软件分层更是一种设计范式的根本转变。2.1 自主层基于主动推理的“大脑”一个被动的镜像只需要数据接收器和规则引擎。而一个主动的代理需要一个持续与世界交互、更新内部信念并选择行动的“大脑”。这里Active Inference主动推理框架提供了一个极具潜力的数学模型。它源于神经科学和贝叶斯理论核心思想是智能体通过最小化一个叫“自由能”的量来生存。自由能可以粗略理解为“预测误差”和“意外性”的综合度量。在合子数字孪生的自主层我们可以这样实现每个孪生体内部维护一个生成模型。这个模型不仅编码了关于自身物理对应物状态的信念如“电机温度很可能在70°C左右”还编码了对环境和其他孪生体行为的预期如“如果我提高转速上游供料系统的孪生体可能会发出压力预警”。它通过传感器或来自物理实体的数据流持续接收观察然后用这些观察来更新自己的内部信念贝叶斯更新。关键的一步在于它不仅仅被动更新还会计算为了维持其期望状态如“设备健康运行”、“能耗最优”需要采取的行动。这个过程就是“主动”的由来它主动规划并执行动作通过向物理实体发送控制指令或向其他孪生体发送协商请求以使其未来的观察更符合其期望从而最小化长期的自由能。举个例子一个泵机的合子数字孪生其生成模型可能包含“流量-转速-能耗”的关系以及“管网压力”的预期。当它观察到压力低于预期时它不会简单地等待中心控制系统的指令而是会主动推理是应该增加自身转速还是应该向相邻泵机的孪生体发送协作请求询问它们是否可以分担负荷它会基于模型预测不同行动路径下的结果能耗、设备磨损、系统压力稳定性并选择那个预期自由能最小的行动。这就赋予了孪生体真正的、基于模型的自主决策能力。2.2 协作层网络即协调媒介单个孪生体的自主决策如果不加协调可能导致系统层面的混乱。因此合子数字孪生的第二个核心是网络化协作。这里的“网络”不仅指TCP/IP通信网络更指孪生体之间动态形成的协作关系网络。这种协作不是通过一个中心调度器来命令而是通过一套共享的协议和接口让孪生体之间能够相互发现、协商和达成共识。这通常需要定义一个清晰的合子接口。这个接口规定了孪生体对外暴露的能力我能做什么、目标我追求什么、和状态我当前情况。例如一个AGV自动导引车的孪生体可能暴露“运输能力负载、速度”、“当前位置”、“任务队列”和“能耗状态”。当整个物流系统需要完成一批紧急订单时不是一个中心算法去指派任务而是由代表“生产订单”的孪生体也是一个合子向网络“广播”或“定向发布”运输需求。各个AGV孪生体根据自身的状态和目标如最小化空驶距离、平衡电池消耗通过接口进行“投标”或协商最终自主形成任务分配方案。这种基于网络的协作天然具有弹性。任何一个孪生体离线或故障其任务可以由网络中其他具备类似能力的孪生体通过重新协商来接管。它也具有良好的可扩展性新增一个设备只需将其孪生体接入网络并发布其接口它就能自动融入现有的协作生态。这直接呼应了我们在热词中看到的像docker-compose.yml里定义networks那样的编排思想只不过这里编排的不是容器而是具有智能的代理。2.3 实现模式从理论到工程在工程实现上一个合子数字孪生通常可以看作一个微服务或一组协同微服务。每个服务封装了前述的自主推理引擎和协作接口。它们通过轻量级消息总线如MQTT、DDS或服务网格进行通信。数据模型和状态同步是关键挑战通常采用事件溯源和CQRS模式来保证每个孪生体内部状态的一致性以及对外状态发布的可观测性。一个常见的误区是试图用一个“超级孪生”来模拟整个复杂系统。合子架构告诉我们正确的方式是递归分解。先为最底层的物理实体传感器、执行器建立具有基本自主能力的孪生体。然后这些孪生体可以聚合成一个代表复合实体如一台机器的“高阶合子”。这个高阶合子同样具有自主性和协作接口它协调其内部子合子的行为并对外作为一个整体参与更上层的协作。如此层层向上形成一棵“合子树”。这种结构使得系统既能在低层级快速响应也能在高层级进行全局优化。3. 网络层的关键挑战与热词背后的实践启示当我们谈论“over Networks”时网络不再仅仅是数据传输的管道而是成了合子数字孪生“生存”的环境。网络的质量直接决定了这些主动代理的协作效能。这里我们可以结合最新的网络热词深入探讨几个关键挑战。3.1 网络发现与编配从“no networks found”到动态自组织热词unable to update cni config: no networks found in /etc/cni/net.d是Kubernetes容器网络接口的典型报错它指向一个根本问题实体如何发现并接入其赖以通信的网络。对于合子数字孪生而言这个问题同样存在且更为复杂。一个新上线的设备孪生体如何自动发现它应该加入的“协作网络”是车间级网络还是跨厂区的物流网络解决方案在于服务发现与动态编配机制。我们可以借鉴云原生生态的理念。每个合子数字孪生在启动时可以向一个轻量级的注册中心如Consul、Etcd或一个专门的合子注册服务注册自己的元数据包括其合子类型、能力描述、通信端点以及所属的“合子群”标签。同时它也可以从注册中心订阅它所关心的其他合子类型或事件。这样网络就从静态配置变成了动态自组织。更进一步我们可以引入像“addition networks插件”这样的概念。这里的“插件”可以理解为一种网络策略或协作协议插件。例如一个专注于“能效优化”的合子群可能需要所有成员孪生体都支持一种特定的能耗数据上报和协商协议。这个协议就可以作为一个“插件”在孪生体注册时声明其支持的能力或者由编配系统动态注入。这避免了为所有孪生体预装所有可能的协议实现了关注点分离和灵活扩展。3.2 通信语义与一致性超越字节流传统物联网通信关注的是数据包的到达QoS。对于主动代理我们更关注通信的语义和交互的一致性。一个孪生体向另一个发送“请求提高供水压力10%”的消息这不仅仅是一个数据这是一个行动意图。接收方需要理解这个意图根据自己的状态判断是否接受、协商或拒绝并给出有意义的回复。这就需要定义丰富的通信原语。不仅仅是Pub/Sub还需要支持Request/Reply、Negotiation、Auction拍卖、Commit等交互模式。消息的负载也需要结构化的语义描述例如使用类似JSON-LD的格式包含动作类型、参数、约束条件、有效期等。这保证了网络中的信息交换是机器可理解、可处理的而不仅仅是人类可读的。一致性挑战则体现在分布式决策上。当多个孪生体通过协商决定共同执行一个动作序列时如多台AGV协同搬运一个大部件如何确保所有参与者对最终计划达成一致并在出现网络分区或节点失效时优雅降级这需要引入分布式共识算法如Raft在轻量级场景下的应用或最终一致性模型并设计相应的超时、重试和补偿事务机制。3.3 网络安全与信任边界主动代理意味着每个孪生体都有执行动作的潜力。这极大地扩大了攻击面。一个被入侵的泵机孪生体可能会恶意提高转速损坏设备或发送虚假协商信息扰乱整个供水网络。因此零信任安全模型在合子网络中至关重要。每个孪生体都必须有明确的身份标识基于数字证书每次交互都需要进行双向认证和授权。授权策略需要细粒度到动作级别例如“AGV-01的孪生体有权向‘区域A充电站’合子请求充电但无权命令其他AGV停止”。策略的执行可以委托给一个策略决策点也可以基于属性如孪生体的角色、当前任务、安全等级进行动态判断。网络通信必须全程加密并且要有完整的审计日志记录下每个孪生体的每个决策和交互以便事后追溯和分析。4. 范畴论为合子系统提供形式化“语法”前面我们讨论了很多工程架构和网络实践但合子数字孪生和Physical AI系统本质上是高度复杂、由多种异质组件软件代理、物理动力学、网络协议耦合的系统。如何严谨地描述这些组件之间的关系、组合方式以及系统的整体行为如何确保我们设计的架构在数学上是自洽的、可组合的、可推理的这就是Category Theory范畴论可以大显身手的地方。别被这个抽象的名字吓到我们可以把它理解为给复杂系统设计提供一套形式化的“语法”或“设计模式语言”。4.1 作为“粘合剂”的范畴论思想范畴论不关心对象内部的具体细节它只关心对象之间的关系称为“态射”以及这些关系如何组合。这恰恰是合子系统的核心我们有很多孪生体对象它们之间通过各种各样的交互进行协作态射。范畴论提供了一套工具来刻画这种结构。例如我们可以把每个合子数字孪生看作一个范畴中的对象。这个孪生体对外提供的协作接口比如“提供数据”、“接受指令”、“参与协商”可以看作从这个对象出发的态射。而两个孪生体能够成功协作比如AGV孪生体成功从仓库孪生体领取任务可以看作这两个态射能够以某种方式“组合”成一个新的态射代表这个协作流程。范畴论中的“交换图”可以用来可视化并验证复杂的协作流程是否畅通无阻从起点到终点无论沿着哪条路径即不同的协作顺序组合态射最终结果都应该是一致的。这帮助我们检查系统设计是否存在死锁或逻辑矛盾。更强大的是我们可以用范畴论来定义什么是“合子的合子”。还记得合子的递归性吗一个高阶合子由低阶合子组合而成。在范畴论中这对应着函子的概念——一个函子可以把一个范畴低阶合子及其关系映射到另一个范畴作为高阶合子内部的一个子结构。这为我们形式化地定义和操作这种层次化组合提供了数学基础。4.2 指导模型集成与数据流Physical AI系统通常混合了多种模型物理仿真模型、数据驱动的机器学习模型、基于规则的专家系统、以及我们讨论的主动推理生成模型。这些模型输入输出格式各异运行在不同位置边缘、云端。如何将它们无缝地“粘合”到一个合子孪生体中范畴论特别是其分支如范畴化机器学习或开放系统理论提供了思路。我们可以将每种模型类型视为一个范畴模型的输入输出端口视为这个范畴的对象。那么连接两个模型比如将物理仿真的输出作为机器学习模型的输入就相当于在两个范畴之间寻找一个合适的函子或自然变换来转换数据类型和语义。这促使我们设计清晰、规范的模型接口使得模型之间的组合像乐高积木一样可靠。在工程上这可以推动我们采用标准化的模型包装器如PMML、ONNX格式的扩展并定义模型元数据来描述其输入输出范畴。4.3 实践中的启发从抽象到具体你可能会问这听起来很理论对一线工程师有什么用它的价值在于提升设计质量和降低沟通成本。设计模式化范畴论中的一些通用构造如单子、余极限已经被计算机科学吸收形成了诸如函数式编程中的Monad用于处理副作用、异步流程等模式。在合子系统设计中我们可以借鉴这些模式来处理不确定性、并发操作和资源管理。例如一个合子处理请求的完整生命周期接收、推理、行动、反馈可以被设计为一个“合子单子”从而以统一、可组合的方式处理错误、日志和状态传递。规范接口迫使我们去思考并明确定义每个孪生体、每个模型的“边界”和“连接点”。这直接导致更清晰、更稳定的API和通信协议设计减少了系统集成时的“猜谜游戏”。验证与测试虽然完全的形式化验证可能很重但范畴论的思维可以帮助我们设计更全面的集成测试用例。通过绘制系统关键交互的交换图我们可以系统地检查所有可能的交互路径确保逻辑一致性。对于大多数团队不需要深入研究范畴论的数学细节但吸收其核心思想——关注组件间的交互与组合而非孤立组件的内部——并将其应用于系统架构设计就能带来巨大的益处。它让我们的合子数字孪生系统不仅仅是代码的堆砌而是一个结构良好、易于理解和演进的有机整体。5. 迈向实战构建你的第一个合子数字孪生原型理论探讨之后让我们脚踏实地看看如何动手构建一个简单的合子数字孪生原型。我们将设计一个高度简化的场景一个智能温控系统。包含一个“温度传感器实体”、“一个加热器实体”和一个作为协调者的“房间温控合子”。我们将使用Python和一些轻量级工具来演示核心概念。5.1 环境准备与核心概念代码化首先我们需要一个模拟环境和一个消息总线。我们将使用paho-mqtt作为通信层模拟网络。每个合子将是一个独立的Python进程或线程通过MQTT主题进行通信。1. 定义合子基类与消息格式我们首先定义一个基础的Holon类它封装了身份、通信和基本的生命周期管理。同时我们需要定义合子间交换的消息格式这里使用JSON。import json import paho.mqtt.client as mqtt import uuid from typing import Any, Dict, Callable from dataclasses import dataclass, asdict from enum import Enum class MessageType(Enum): STATUS status # 发布状态 CAPABILITY capability # 宣告能力 REQUEST request # 请求行动 RESPONSE response # 请求回应 NEGOTIATE negotiate # 发起协商 dataclass class HolonMessage: msg_id: str sender_id: str msg_type: MessageType topic: str # 发送/订阅的主题 payload: Dict[str, Any] # 消息内容 timestamp: float def to_json(self): return json.dumps({ **asdict(self), msg_type: self.msg_type.value }) class Holon: def __init__(self, holon_id: str, brokerlocalhost, port1883): self.id holon_id self.broker broker self.port port self.client mqtt.Client(client_idholon_id) self.client.on_connect self._on_connect self.client.on_message self._on_message self._message_handlers {} def connect(self): self.client.connect(self.broker, self.port, 60) self.client.loop_start() def _on_connect(self, client, userdata, flags, rc): print(f[{self.id}] Connected to broker.) # 订阅自身相关的命令主题 self.client.subscribe(fholon/{self.id}/cmd/#) def _on_message(self, client, userdata, msg): try: payload json.loads(msg.payload.decode()) incoming_msg HolonMessage( msg_idpayload[msg_id], sender_idpayload[sender_id], msg_typeMessageType(payload[msg_type]), topicmsg.topic, payloadpayload[payload], timestamppayload[timestamp] ) self._route_message(incoming_msg) except Exception as e: print(f[{self.id}] Error processing message: {e}) def _route_message(self, msg: HolonMessage): 根据消息类型路由到对应的处理函数 handler self._message_handlers.get(msg.msg_type) if handler: handler(msg) else: print(f[{self.id}] No handler for message type: {msg.msg_type}) def register_handler(self, msg_type: MessageType, handler: Callable): self._message_handlers[msg_type] handler def publish_message(self, topic: str, msg_type: MessageType, payload: Dict): msg HolonMessage( msg_idstr(uuid.uuid4()), sender_idself.id, msg_typemsg_type, topictopic, payloadpayload, timestamptime.time() ) self.client.publish(topic, msg.to_json()) print(f[{self.id}] Published {msg_type.value} to {topic}) def announce_capability(self, capability: Dict): 宣告自身能力 self.publish_message( topicholon/capabilities, msg_typeMessageType.CAPABILITY, payload{holon_id: self.id, capability: capability} )5.2 实现具体合子传感器与执行器现在我们实现一个简单的温度传感器合子。它模拟读取温度并定期发布状态。它也具有基本的“主动”性当温度超过某个阈值时它会主动发出一个“协助请求”。import time import random class TemperatureSensorHolon(Holon): def __init__(self, sensor_id, locationroom_1, **kwargs): super().__init__(ftemp_sensor_{sensor_id}, **kwargs) self.location location self.current_temp 20.0 # 初始温度 self.threshold_high 25.0 # 注册处理函数它可以响应校准请求 self.register_handler(MessageType.REQUEST, self._handle_request) def run(self): self.connect() # 宣告能力提供温度数据 self.announce_capability({ type: sensor, quantity: temperature, unit: C, location: self.location, accuracy: /-0.5C }) while True: # 模拟温度变化 self.current_temp random.uniform(-0.5, 0.8) # 发布状态 self.publish_message( topicfholon/sensor/{self.location}/temperature, msg_typeMessageType.STATUS, payload{value: round(self.current_temp, 2), unit: C} ) # 主动推理逻辑如果温度过高主动发出协助请求 if self.current_temp self.threshold_high: print(f[{self.id}] Temperature ({self.current_temp:.1f}C) exceeds threshold. Requesting cooling action.) self.publish_message( topicholon/requests/cooling, msg_typeMessageType.REQUEST, payload{ type: cooling_request, requester: self.id, location: self.location, current_temp: self.current_temp, target_temp: 22.0 } ) time.sleep(5) # 每5秒更新一次 def _handle_request(self, msg: HolonMessage): 处理外部请求例如校准请求 if msg.payload.get(action) calibrate: # 模拟校准过程 print(f[{self.id}] Received calibration request from {msg.sender_id}.) # ... 执行校准逻辑 # 发送响应 self.publish_message( topicfholon/{msg.sender_id}/resp, msg_typeMessageType.RESPONSE, payload{request_id: msg.payload.get(request_id), status: calibrated} )接下来实现一个加热器合子。它能接收控制指令也能根据请求“投标”自己的加热服务。class HeaterHolon(Holon): def __init__(self, heater_id, max_power2000, locationroom_1, **kwargs): super().__init__(fheater_{heater_id}, **kwargs) self.max_power max_power # 瓦特 self.current_power 0 self.location location self.efficiency 0.9 # 能效 # 订阅加热请求和命令 self.client.subscribe(holon/requests/heating) self.client.subscribe(fholon/{self.id}/cmd/power) self.register_handler(MessageType.REQUEST, self._handle_heating_request) def run(self): self.connect() self.announce_capability({ type: actuator, action: heating, max_power_w: self.max_power, location: self.location, efficiency: self.efficiency }) # 模拟运行主要靠消息驱动 while True: time.sleep(1) def _handle_heating_request(self, msg: HolonMessage): if msg.topic holon/requests/heating: req msg.payload # 简单的“投标”逻辑如果我在请求的位置并且有能力就响应 if req.get(location) self.location: required_power req.get(required_power, 500) if required_power self.max_power: bid { bidder_id: self.id, offered_power: required_power, cost_estimate: required_power * 0.01, # 简单成本模型 estimated_time: 60 # 秒 } self.publish_message( topicfholon/{msg.sender_id}/bid, msg_typeMessageType.RESPONSE, payload{request_id: req.get(request_id), bid: bid} ) print(f[{self.id}] Submitted bid for heating request.)5.3 实现协调者合子房间温控代理现在我们创建一个高阶合子——RoomThermostatHolon。它不直接连接物理设备而是协调传感器和加热器实现房间温度的主动调节。它体现了“主动代理”和“合子”的特性它有自己的目标维持设定温度并通过与其他合子协作来实现。class RoomThermostatHolon(Holon): def __init__(self, room_id, target_temp22.0, **kwargs): super().__init__(fthermostat_{room_id}, **kwargs) self.room_id room_id self.target_temp target_temp self.current_temp None self.heater_assignments {} # 记录分配的任务 # 订阅相关主题 self.client.subscribe(fholon/sensor/{room_id}/temperature) self.client.subscribe(fholon/{self.id}/bid) # 接收投标 self.register_handler(MessageType.STATUS, self._handle_temp_update) self.register_handler(MessageType.RESPONSE, self._handle_bid_response) def run(self): self.connect() self.announce_capability({ type: coordinator, service: temperature_regulation, scope: self.room_id, target_temp: self.target_temp }) print(f[{self.id}] Thermostat started for {self.room_id}, target: {self.target_temp}C) while True: # 主循环可以执行周期性的协调逻辑例如检查任务完成情况 self._check_assignments() time.sleep(10) def _handle_temp_update(self, msg: HolonMessage): 处理温度传感器发来的状态更新 if fsensor/{self.room_id}/temperature in msg.topic: new_temp msg.payload.get(value) self.current_temp new_temp print(f[{self.id}] Current temperature updated: {new_temp}C) # 主动推理基于当前温度和目标温度决定行动 self._decide_action() def _decide_action(self): if self.current_temp is None: return temp_diff self.target_temp - self.current_temp deadband 0.5 # 死区避免频繁动作 if temp_diff deadband: # 太冷需要加热 required_power min(1500, int(abs(temp_diff) * 200)) # 简单的线性计算 print(f[{self.id}] Too cold ({self.current_temp}C). Requesting heating ({required_power}W).) # 发布加热请求到网络进行“招标” request_id str(uuid.uuid4()) self.heater_assignments[request_id] {status: pending, required_power: required_power} self.publish_message( topicholon/requests/heating, msg_typeMessageType.REQUEST, payload{ request_id: request_id, type: heating, location: self.room_id, required_power: required_power, requester: self.id } ) elif temp_diff -deadband: # 太热需要冷却本例中冷却请求由传感器直接发出 # 在我们的简单原型中冷却由传感器主动请求这里可以记录或触发其他动作 print(f[{self.id}] Too warm ({self.current_temp}C). Cooling may be requested by sensor.) # 可以在这里发布关闭加热器的命令等 for req_id, assignment in list(self.heater_assignments.items()): if assignment.get(status) active: print(f[{self.id}] Cancelling heating assignment {req_id} due to high temp.) # 通知加热器停止 self.publish_message( topicfholon/{assignment[heater_id]}/cmd/power, msg_typeMessageType.REQUEST, payload{action: set_power, value: 0} ) assignment[status] cancelled def _handle_bid_response(self, msg: HolonMessage): 处理加热器对请求的投标 if bid in msg.payload: bid msg.payload[bid] request_id msg.payload.get(request_id) if request_id in self.heater_assignments and self.heater_assignments[request_id][status] pending: # 简单的胜标逻辑选择第一个响应的实际中可根据成本、效率等选择 print(f[{self.id}] Accepting bid from {bid[bidder_id]} for request {request_id}.) self.heater_assignments[request_id].update({ status: active, heater_id: bid[bidder_id], bid: bid }) # 向中标加热器发送执行命令 self.publish_message( topicfholon/{bid[bidder_id]}/cmd/power, msg_typeMessageType.REQUEST, payload{action: set_power, value: self.heater_assignments[request_id][required_power]} ) # 通知其他投标者请求已关闭可选 # self.publish_message(topicholon/requests/heating/closed, ...) def _check_assignments(self): 周期性检查任务状态模拟任务完成后的清理 # 这里可以添加更复杂的逻辑比如根据温度反馈判断加热是否完成 for req_id, assignment in list(self.heater_assignments.items()): if assignment.get(status) active: # 假设加热任务持续一段时间后自动完成 if assignment.get(start_time) is None: assignment[start_time] time.time() elif time.time() - assignment[start_time] 30: # 30秒后停止 print(f[{self.id}] Heating assignment {req_id} completed.) self.publish_message( topicfholon/{assignment[heater_id]}/cmd/power, msg_typeMessageType.REQUEST, payload{action: set_power, value: 0} ) assignment[status] completed5.4 运行与观察要运行这个原型你需要启动一个MQTT代理如Mosquitto然后分别运行传感器、加热器和温控器的代码。你会看到在控制台中温度传感器定期报告温度并在温度过高时主动发出冷却请求。温控器订阅温度当温度过低时它会发布一个“加热请求”到网络。加热器收到请求后会“投标”自己的服务。温控器接受投标并向中标的加热器发送具体的功率设置命令。整个过程中没有中心控制器硬编码逻辑每个合子都是自主的协作是通过网络中的消息交换动态形成的。这个原型极其简化省略了错误处理、安全性、复杂的主动推理模型、正式的协商协议等。但它清晰地演示了从“被动镜像”传感器只发数据到“主动代理”传感器能发请求温控器能协调的转变以及合子之间通过网络进行自主协作的基本模式。你可以在此基础上引入更复杂的主动推理库如pymdp实现真正的贝叶斯信念更新和自由能最小化或者使用更健壮的服务发现如Consul甚至将每个合子容器化用docker-compose.yml定义它们的网络模拟更真实的分布式部署。通过这个实战演练我希望你能感受到构建合子数字孪生并非遥不可及。它始于对现有“物模型”或“数字孪生模型”的思维转变从单纯的数据容器转变为具有目标、内部模型和通信能力的主动实体。一旦迈出这一步一个更灵活、更智能、更鲁棒的Physical AI网络就在眼前。