把模型推理直接写进控制循环，最先出问题的不是性能，而是可维护性：换一版动作输出格式，控制代码要跟着改；机器人动作不对，也说不清是观测没送到、推理输出异常，还是执行层自己在抖。比较稳的接入顺序是——先把推理封成独立进程，用本机回环把一条「观测进、动作出」的完整链路跑通，再去接本体。这样接硬件时剩下的变量只有硬件本身，而不是模型、通信、控制三件事缠在一起。

只要推理进程和控制进程能各自独立起停、消息格式固定下来，模型迭代通常就不需要动控制代码。适用场景是推理与控制同机或近距离部署的机器人工程。操作动作是先跑回环链路，再人为制造一次单侧故障验证问题归属。验证方式是看两侧日志里的序号进度与超时记录。边界要说清：拆进程不解决实时性问题，推理耗时与控制周期的匹配仍要单独测。

划清两侧职责：谁采观测，谁发动作

职责边界必须先定，否则后面谈不上「独立替换」。一个常用且省事的划分是：控制侧贴近硬件，推理侧贴近模型。

每侧允许执行的操作要收敛到上表范围。控制侧允许做的：采集、时间对齐、消息打包、边界校验、限幅、写硬件。推理侧允许做的：加载权重、前向计算、把结果编码成约定的动作字段。

明确禁止越界的部分同样重要：控制侧不读模型权重文件、不 import 推理框架、不因为「顺手」去改动作数值；推理侧不打开串口、不直接写 PWM 或力控指令、不在推理进程里做积分或滤波。这些约束看起来刻板，但它们决定了后面单侧故障时，报错能不能落在正确的一侧。

定义进程间消息的最小字段

让两侧通过数据结构耦合，而不是通过函数调用或共享内存里的对象耦合。字段名可以按自己工程改，但下面这几类信息建议一个都不要少，尤其是时间戳与序号，它们后面是排查问题的唯一依据。

// 观测消息（控制侧 -> 推理侧）
{
  'ts': 1730000000.123,        // 采集时刻，单调时钟
  'seq': 10432,                // 控制侧自增，用于配对
  'obs': {
    'joint_pos': [ ... ],      // 长度与顺序固定
    'joint_vel': [ ... ],
    'ee_pose': [ ... ],        // 可选
    'images': [
      {'name': 'cam_front', 'encoding': 'jpeg', 'data': '...'}
    ]
  },
  'meta': {'control_period_ms': 20, 'unit': 'rad'}
}

// 动作消息（推理侧 -> 控制侧）
{
  'ts': 1730000000.140,        // 推理完成时刻
  'seq': 10432,                // 原样回带，便于配对
  'infer_ms': 17,              // 仅用于观察，不参与控制
  'action': {
    'joint_target': [ ... ],
    'gripper': 0.0,
    'unit': 'rad',
    'frame': 'base_link'
  },
  'status': 'ok'               // ok / degraded / error
}

几个容易被忽略的点：数组长度和关节顺序必须在两侧配置里写死，不要靠推理侧推断；时间戳用单调时钟，避免系统时间被调整后误判超时；seq 由请求带过来、响应原样回带，这样即使中间丢了一条也能立刻看出来。status 字段建议保留，推理侧在输入异常时返回 degraded，控制侧就不会把异常输出当成正常动作执行。

用本地回环先跑通一条完整消息链路

没有硬件的时候，本机 127.0.0.1 上的一条 TCP 连接就够验证通路了。建议用换行符做消息定界，天然规避「一次 recv 收到半条消息」的问题；如果消息体较大，也可以在头部加长度前缀，两种方式选一种并在两侧统一。

# 推理侧：接收观测、返回动作
import socket, json, time

srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(('127.0.0.1', 5055))
srv.listen(1)
print('infer: listening', flush=True)
conn, _ = srv.accept()
print('infer: connected', flush=True)

buf = b''
while True:
    chunk = conn.recv(65536)
    if not chunk:
        print('infer: peer closed', flush=True)
        break
    buf += chunk
    while b'\n' in buf:
        line, buf = buf.split(b'\n', 1)
        try:
            msg = json.loads(line)
        except ValueError as e:
            print('infer: bad json %s' % e, flush=True)
            continue
        print('infer: recv seq=%s bytes=%d' % (msg['seq'], len(line)), flush=True)
        reply = {'ts': time.monotonic(), 'seq': msg['seq'],
                 'action': {'joint_target': [0.0] * 6, 'unit': 'rad'},
                 'status': 'ok'}
        conn.sendall((json.dumps(reply) + '\n').encode('utf-8'))
        print('infer: send seq=%s' % reply['seq'], flush=True)

控制侧对称地发一条、收一条，每条都打印 seq 和字节数。判断消息是否被完整接收，看三件事：一是控制侧发出的 seq 序列在推理侧日志里是否连续、有没有跳号；二是推理侧打印的字节数与控制侧发送时的字节数是否一致；三是 json 解析有没有抛异常。三者同时成立，基本可以认为收发通路可用，此时再录一段假观测数据（比如固定长度的关节角数组）反复发送，确认长时间运行不会因为缓冲区处理不当而粘包。

制造一次单侧故障，看报错落在哪一侧

解耦有没有真正起作用，要看故障发生时报错的位置。建议在回环阶段就主动做两次实验，不要等接了本体再试。

• 停掉推理进程。控制侧的现象通常是：连接被关闭或后续发送阻塞，收到的动作停止更新。日志表现是发送侧打印了 seq 之后，接收循环返回空或触发超时，错误出现在通信封装这一层。判断方法：如果控制侧报的是「发送/接收失败」而不是「串口写失败」，说明问题归属清楚，控制代码本身没被污染。
• 停掉控制进程。推理侧的现象是 recv 返回空、退出接收循环，回到 accept 等待。日志表现是没有新的 recv 行，seq 停在最后一个值。判断方法：硬件侧没有任何异常记录，说明推理进程没有越界去碰硬件。

还可以做第三种：让推理侧返回一条字段缺失或长度不对的动作消息。预期现象是控制侧在边界校验处报错并丢弃，而不是把错误数据下发出去。如果这一步报错出现在执行层内部，说明校验位置放错了，需要往前挪到进程边界。

把限幅与超时保护放在进程边界上

边界是拦截异常最自然的位置。控制侧收到动作消息后，先校验再限幅，最后才写硬件；推理侧输出前也可以做一次自检，目的是在调试阶段更早暴露异常，但不要以此替代控制侧的兜底。

# 控制侧边界处理（示例，字段名按自己工程改）
action_timeout_ms: 100      # 超过则采用 default_action
drop_if_ts_lag_ms: 50       # 时间戳落后超过阈值则丢弃
seq_must_increase: true     # 序号回退或重复则丢弃
joint_limit_rad: [[-1.57, 1.57], [-1.57, 1.57]]  # 与关节顺序一一对应
default_action: hold_last   # 也可选 zero_velocity 或 hold_position
drop_log_every_n: 10        # 丢弃计数按此频率打日志，避免刷屏

超时后的默认动作需要提前定死：保持上一帧目标适合短时抖动，零速或保持位适合安全优先的场景，具体选哪个要结合机械结构和任务确认。不建议在超时后无限期重发上一条动作，那会让故障被掩盖。

消息丢弃规则建议至少覆盖三类：时间戳过期、序号回退或重复、字段缺失或长度不匹配。每一类都单独计数并打日志，不要把三种情况混成一个计数器。

限幅发生的位置要能通过日志验证：手动构造一条明显越界的 joint_target 发过去，看控制侧是否打印限幅记录、并且写出去的数值落在配置范围内；再构造一条时间戳过期的消息，看是否被计入丢弃。这两个动作做完，边界保护的生效与否就不需要靠推测了。