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, broker="localhost", port=1883): self.id = holon_id self.broker = broker self.port = port self.client = mqtt.Client(client_id=holon_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(f"holon/{self.id}/cmd/#") def _on_message(self, client, userdata, msg): try: payload = json.loads(msg.payload.decode()) incoming_msg = HolonMessage( msg_id=payload['msg_id'], sender_id=payload['sender_id'], msg_type=MessageType(payload['msg_type']), topic=msg.topic, payload=payload['payload'], timestamp=payload['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_id=str(uuid.uuid4()), sender_id=self.id, msg_type=msg_type, topic=topic, payload=payload, timestamp=time.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( topic="holon/capabilities", msg_type=MessageType.CAPABILITY, payload={"holon_id": self.id, "capability": capability} )5.2 实现具体合子：传感器与执行器
现在，我们实现一个简单的温度传感器合子。它模拟读取温度，并定期发布状态。它也具有基本的“主动”性：当温度超过某个阈值时，它会主动发出一个“协助请求”。
import time import random class TemperatureSensorHolon(Holon): def __init__(self, sensor_id, location="room_1", **kwargs): super().__init__(f"temp_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( topic=f"holon/sensor/{self.location}/temperature", msg_type=MessageType.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( topic="holon/requests/cooling", msg_type=MessageType.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( topic=f"holon/{msg.sender_id}/resp", msg_type=MessageType.RESPONSE, payload={"request_id": msg.payload.get("request_id"), "status": "calibrated"} )接下来，实现一个加热器合子。它能接收控制指令，也能根据请求“投标”自己的加热服务。
class HeaterHolon(Holon): def __init__(self, heater_id, max_power=2000, location="room_1", **kwargs): super().__init__(f"heater_{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(f"holon/{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( topic=f"holon/{msg.sender_id}/bid", msg_type=MessageType.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_temp=22.0, **kwargs): super().__init__(f"thermostat_{room_id}", **kwargs) self.room_id = room_id self.target_temp = target_temp self.current_temp = None self.heater_assignments = {} # 记录分配的任务 # 订阅相关主题 self.client.subscribe(f"holon/sensor/{room_id}/temperature") self.client.subscribe(f"holon/{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 f"sensor/{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( topic="holon/requests/heating", msg_type=MessageType.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( topic=f"holon/{assignment['heater_id']}/cmd/power", msg_type=MessageType.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( topic=f"holon/{bid['bidder_id']}/cmd/power", msg_type=MessageType.REQUEST, payload={"action": "set_power", "value": self.heater_assignments[request_id]["required_power"]} ) # 通知其他投标者请求已关闭（可选） # self.publish_message(topic="holon/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( topic=f"holon/{assignment['heater_id']}/cmd/power", msg_type=MessageType.REQUEST, payload={"action": "set_power", "value": 0} ) assignment["status"] = "completed"5.4 运行与观察
要运行这个原型，你需要启动一个MQTT代理（如Mosquitto），然后分别运行传感器、加热器和温控器的代码。你会看到在控制台中，温度传感器定期报告温度，并在温度过高时主动发出冷却请求。温控器订阅温度，当温度过低时，它会发布一个“加热请求”到网络。加热器收到请求后，会“投标”自己的服务。温控器接受投标，并向中标的加热器发送具体的功率设置命令。整个过程中，没有中心控制器硬编码逻辑，每个合子都是自主的，协作是通过网络中的消息交换动态形成的。
这个原型极其简化，省略了错误处理、安全性、复杂的主动推理模型、正式的协商协议等。但它清晰地演示了从“被动镜像”（传感器只发数据）到“主动代理”（传感器能发请求，温控器能协调）的转变，以及合子之间通过网络进行自主协作的基本模式。你可以在此基础上，引入更复杂的主动推理库（如pymdp），实现真正的贝叶斯信念更新和自由能最小化；或者使用更健壮的服务发现（如Consul）；甚至将每个合子容器化，用docker-compose.yml定义它们的网络，模拟更真实的分布式部署。
通过这个实战演练，我希望你能感受到，构建合子数字孪生并非遥不可及。它始于对现有“物模型”或“数字孪生模型”的思维转变：从单纯的数据容器，转变为具有目标、内部模型和通信能力的主动实体。一旦迈出这一步，一个更灵活、更智能、更鲁棒的Physical AI网络就在眼前。
