仿真实验闭环工作流开发教程（2）：闭环技术栈全景——三层一总线与选型矩阵

kkchenjj

10人浏览 · 2026-09-06 11:12:00

kkchenjj · 2026-09-06 11:12:00 发布

仿真实验闭环工作流开发教程（2）：闭环技术栈全景——三层一总线与选型矩阵

版本声明块

工具/软件：ASE 3.29.0（2026-06-21，官网 ase-lib.org）｜QuAcc（ASE 官方生态页 workflows 分类）｜opentrons 9.1.2（apiLevel 2.27）｜pylabrobot 0.2.2｜sila2 0.14.0｜Benchling REST API v2 + benchling-sdk 1.25.0｜SENAITE LIMS 2.6.0 + SENAITE.JSONAPI｜LabArchives REST-like API｜isatools 0.14.3
语言/环境：Python 3.11+ / pydantic v2 / bash / YAML / SQLite
本文目标：给你一张能贴在项目墙上的技术栈地图，并交付第一版可运行的契约代码与目录骨架

一句话结论：闭环在工程上只有"三层一总线"——仿真层（ASE 3.29.0 的单点计算 + QuAcc 的高通量工作流）、执行层（Opentrons/Hamilton/Tecan 三家接口成熟度差异极大，一律经 PyLabRobot 0.2.2 的 LiquidHandler + STARBackend/OpentronsOT2Backend 或 SiLA 2 的 sila2 抽象）、记录层（Benchling /api/v2、SENAITE JSONAPI、LabArchives 签名式 XML）；"总线"不是 Kafka，而是 schema_version + loop_id + 主题命名 + 幂等键 四件事，本篇用 contracts/envelope.py 一个 pydantic 文件把它写死。

〇、本篇要解决的认知问题

Q1：为什么闭环技术栈要按"三层"切，而不是"一个仿真软件 + 一个 LIMS"两件套就完事？

Q2：仿真层里 ASE 和 QuAcc 是什么关系？我到底该在哪个上面写二次开发代码？

Q3：Hamilton、Tecan、Beckman 这些主流液体处理工作站的官方接口开放程度到底怎么样？为什么说"有 API"和"有能用的 API"是两回事？

Q4：Benchling、SENAITE、LabArchives 三种记录层的鉴权与数据形态差在哪？选型时该按什么维度排？

Q5：所谓"一总线"是不是就该上 Kafka/RabbitMQ？为什么本系列说它只能算"工程实践参考"？

一、机制解析

1.1 为什么非三层不可：时钟与故障模式不同频

把仿真、设备、记录塞进一个进程，是闭环项目最典型的第一版死法。根本原因是这三层时间尺度、事务边界、故障模式全都不同：

维度	仿真层	执行层	记录层
单次动作时间尺度	秒到天（一次 NEB 扫描可能排队几小时）	毫秒到分钟（一个移液动作、一次 home）	毫秒（一次 HTTP 写入）
事务语义	可重跑、可并行、结果幂等	不可回滚——枪头已经吸了，液已经 dispense 了	可追加、不可遮盖（Part 11 §11.10 审计追踪不得遮盖先前信息）
主要故障	算得不收敛、作业排队失败、输入结构文件损坏	撞针、试剂余量不足、孔位错位、温控超范围	401/403 权限、404 租户写错、速率限制耗尽、区域 host 选错
谁为它负责	计算化学/材料岗	自动化工程师 + 实验室值班	IT/QA + 数据管理员
正确解耦方式	队列 + 版本化输入	先仿真后实机（铁律 2）+ 人工闸门（铁律 10）	幂等写回 + 审计归属（铁律 3）

三层的"翻译官"就是数据契约。它之所以要独立成一层（本篇第二节、第 5 篇整篇），是因为它是三层唯一共享的东西：仿真层产出它、执行层引用它、记录层存储它。共享逻辑一旦散落在三层代码里，任何一层升级都会把另两层拖垮。

为什么这对你重要：你在三层里犯的错误代价完全不对称。仿真层写错=浪费机时，重来即可；执行层写错=毁掉一批不可再生的样品；记录层写错=审计时发现证据链断裂，前面所有的实验在合规意义上"没有发生过"。分层不是为了架构好看，是为了让错误的代价可控。

1.2 仿真层：ASE 管"怎么算一个体系"，QuAcc 管"怎么算一万次"

ASE（Atomic Simulation Environment） 是这一层的底座。当前基线 3.29.0（2026-06-21 发布），官网已迁至 ase-lib.org，文档在 docs.ase-lib.org，源码 gitlab.com/ase/ase。对二次开发最相关的几个事实面：

结构构造与计算器解耦：from ase.build import molecule + from ase.calculators.emt import EMT，能量取 atoms.get_potential_energy()；
几何优化在 ase.optimize（BFGS/LBFGS/FIRE/MDMin/sciopt）；
过渡态方法自 3.23 起在 ase.mep 命名空间下（from ase.mep import NEB, DyNEB）——旧教程里的 ase.neb 路径属于历史；
ASE 4.0 的原型在 ase._4 实验命名空间，ase.ga 已拆为独立项目 ase-ga；
⚠️ ASE 没有官方 calibration/calibrate 包（仓库逐文件核验为 404）。搜索时看到 ase.calibrate、ase.calibration 一律是臆造，本系列禁用；参数标定落在第 13 篇的 AMICI + pyPESTO。

QuAcc 不是 ASE 的替代品，而是 ASE 官方生态页收录的工作流平台，官方一句话定位是"A flexible platform for high-throughput, database-driven computational materials science and quantum chemistry workflows"（分类为 workflows）。同一个生态页还收录了 calorine、matgl、CHGNet、NequIP、matscipy、icet、CLEASE 等——这份名单本身就是" ASE 生态在哪"的最可靠路标。

两者关系可以这样记：

	ASE	QuAcc
抽象层次	单个原子体系的 API	批量工作流 + 数据库驱动
闭环里的角色	`simulate(params) -> prediction` 的实现体（第 9 篇）	当候选空间是"千级/万级"时的吞吐方案（第 12 篇筛选漏斗）
你写的代码	计算器、优化器、属性提取	工作流函数 + 结果库 schema

为什么这对你重要：闭环的学习层每轮只要几十个建议，绝大多数时候你不需要高通量平台。先按"一个函数 simulate()"设计契约（第 9 篇），量级真的上去了再换 QuAcc，契约不动——这就是"总线"的价值。

1.3 执行层：厂商面的参差，与抽象层为什么是必需品

这一层是三个厂商接口成熟度差异最大的地方，也是铁律 9（不臆造厂商 API）的主战场。真实情况按证据分层：

厂商/产品	官方开发者入口（可核实）	公开 Python SDK	建议接法
Opentrons Flex / OT-2	`docs.opentrons.com/v2`（Protocol API v2）；机器人本机 HTTP API `:31950`，OpenAPI 在设备本机 `/openapi.json`、文档 UI `/redoc`	有（`pip install opentrons`，9.1.2）	直接写 v2 协议；集成面走 HTTP（第 4 篇）
Hamilton（VENUS / Prep）	developer.hamiltoncompany.com 门户列 Prep API、VENUS API、Developer Forum；VENUS 页面能力含 method 的 export/import/validation、method loading 与 execution、设备列表；Prep 为 REST + WebSocket 实时事件 `ws://[IP_address]/NimbusLite/instinctevents`，OpenAPI 文档在设备本机	官方 SDK 素材未逐字登记，需向厂商索取	原生通道做 method 校验与执行；跨设备任务交 PyLabRobot `STARBackend`
Tecan Freedom EVOware	官方页原文两句：“Additional drivers can be created using the Freedom EVOware Developer Studio.” / “Freedom EVOware can be controlled by other software via its API…”	无公开 Python SDK / 无公开 REST 文档	申请官方接口 + 经 SiLA 2 或 PyLabRobot 抽象；绝不写 `import tecan`
Beckman Biomek i-Series（i5/i7）+ vWorks	产品页公开；vWorks Scripting Interface（VBScript/JavaScript）仅第三方记载	无公开官方 API 文档	需联系厂商获取；外部集成走抽象层

两个必须记住的纠错点（属于禁用清单，只能以否定语境出现）：Hamilton 的正确说法是 VENUS API / Prep API，“sierra API” 不存在（第三方薄封装 PyVenus 也非官方）；Beckman 的 “Biomek NX API” 未找到官方页。

于是抽象层不是"锦上添花"，而是唯一能让闭环跨设备活下去的路：

PyLabRobot（pip install pylabrobot，0.2.2；docs.pylabrobot.org）：纯 Python 硬件无关，核心对象 LiquidHandler，backend 有 STARBackend（Hamilton STAR）、OpentronsOT2Backend（Opentrons OT-2），还覆盖酶标仪/泵/天平/加热振荡器；布局用 Deck.load_from_json_file("hamilton-layout.json") 描述。同一份 API 换 backend 即切设备（官方 README 原文示例）。
SiLA 2（pip install sila2，0.14.0，PyPI 自述 “Python implementation of the SiLA 2 standard for lab automation”；标准站 sila-standard.com，仓库 gitlab.com/sila2/sila_python）：设备侧暴露标准化的服务/命令，把"厂商协议"关在设备那一侧。顺带一个信号：AnIML 官网首页放了"AnIML + SiLA"的组合文章——数据格式 + 设备通信这条官方组合叙事，正是本篇"一总线 + 执行层"的官方版背书。

为什么这对你重要：设备采购决策往往在研发之后。如果你今天把 Hamilton 的私有调用写进业务代码，明年换 Tecan 时闭环就得重写；写进 LiquidHandler 抽象，你只需要一个新 backend。这是二次开发者能给公司省下的最大一笔隐性成本。

一个真实的坑值得单列：早年间流传的一套"硬件无关 SDK"（PySciMe / PyLab_server / pylabserver.com 这一族名字）经核实全部查无此物（PyPI 404、域名 NXDOMAIN、Crossref 零命中）；ChemistryOS 域名如今是一个待售停放页。生态会洗牌，闭环的可移植性只能建立在你自己的 schema 上，不能建立在某个第三方桥接库的存活假设上。

1.4 记录层：三家接口范式完全不同

对比项	Benchling	SENAITE LIMS	LabArchives
形态	商业云 ELN+Registry+Inventory	开源 LIMS（GPL-2.0，Plone/Zope 生态，release 2.6.0/2025-04-04）	商业云 ELN（现属 Dotmatics）
接口	REST API v2，形如 `https://{tenant}.benchling.com/api/v2/...`	SENAITE.JSONAPI add-on：HTTP GET/POST 的 CRU（README 明示非完整 CRUD）	官方定位原文：“目标是让第三方通过 REST like API 访问大部分功能”，多数返回 XML
鉴权	个人 API key（Basic，key 作 username 密码留空）／Apps OAuth2 client_credentials→Bearer（`expires_in: 900`）／OIDC id token	Login/Logout 与 Basic Authentication	Access Key ID + Access Password 计算 `sig` + `expires`，访问用户数据另需 `uid`
主机	租户在域名里	自建（二开主路径：buildout `eggs` 装配 add-on 后 `bin/buildout`）	分区域：`api.`/`ukapi.`/`euapi.`/`caapi.`/`auapi.labarchives.com`
闭环可写的钩子	Apps 可托管（官方推荐 AWS Lambda/EC2 或本地）	PUSH endpoint（自定义 jobs）：写入即触发	无官方 webhook 规范（素材未登记）
Python 官方支持	`benchling-sdk` 1.25.0（`benchling-api` 不存在）	服务端即 Python（91.2%），二开用 add-on	社区薄封装 `labarchives-py`（非官方）

为什么这对你重要：选记录层时最该问的不是"字段够不够多"，而是**“闭环能不能在它上面留下可审计的动作来源”**。Benchling 的 App service account 让脚本动作可区分、可审计；SENAITE 的 PUSH 让"写入即触发下一轮"成为一等公民；LabArchives 的签名式接口则意味着你的重试逻辑要处理 expires 过期。这三点决定了第 7 篇写回通道怎么写。

1.5 “一总线”：契约 + 标识 + 主题命名 + 幂等键

先给结论：本系列的"总线"首先是一组约定，其次才可能是一个中间件进程。

   仿真层                     记录层                      执行层                    学习层
   ┌────────┐                ┌────────┐                 ┌────────┐               ┌────────┐
   │simulate│                │Benchling│                │protocol│               │Ax ask  │
   └───┬────┘                └───┬────┘                 └───┬────┘               └───┬────┘
       │                         │                          │                        │
       ▼                         ▼                          ▼                        ▼
  ┌──────────────────────────────────────────  总线：约定 + 传输  ───────────────────────────┐
  │ ① 契约 schema_version="1.0"  pydantic 校验（contracts/envelope.py，本节代码）            │
  │ ② 持久标识 loop_id / trial_index / sample_id   → FAIR R1.2，闭环台账主键                 │
  │ ③ 主题命名 loop.<domain>.<stage>.<version>     → 例：loop.chem.suggested.v1.0          │
  │ ④ 幂等键 idempotency_key = 消息内容哈希(2.3)   → 重试不产生第二条记录（第 7、15 篇）      │
  ├──────────────────────────────────────────  传输（择一，可换） ────────────────────────────┤
  │ 起步：outbox 表（SQLite）+ 轮询  ← 本系列默认，零运维、可审计、断点续跑友好              │
  │ 进阶：RabbitMQ / Kafka / Redis Streams  ← 工程实践参考，非任何 LIMS/仪器厂商官方规范      │
  │ 事件推送：SENAITE PUSH jobs、Benchling Apps 托管（AWS Lambda/EC2 或本地）                │
  └──────────────────────────────────────────────────────────────────────────────────────────┘

诚实边界（必须写进设计文档）：把 Kafka/RabbitMQ 说成"LIMS 官方推荐集成骨干"、或说"XX LIMS 原生支持 REST webhook"，都没有厂商官方一手规范支撑（Benchling/SENAITE/LabArchives 素材中均无此类规范）。所以本系列的口径是：消息中间件是你自己的架构选择，属于工程实践参考；而"契约、标识、命名、幂等"是跨层协议，无论底下用文件、SQLite 还是 Kafka 都必须先定下来。这也正是铁律 1 的完整表述。

对绝大多数从 0 到 1 的团队，起步推荐 outbox 表 + 轮询：一行 SQL 就能查"卡在哪一步"，闭环调试阶段的可观测性远好于消息队列的黑盒；等吞吐或延迟真的成为瓶颈，再把传输层换掉而契约不动——因为①②③④四件事跟中间件无关。

为什么这对你重要：一旦跳过①②③④直接选中间件，你会在两个月后得到"技术上很先进、但每轮实验都说不清是谁做的"的系统——那是闭环最难看的死法。

1.6 选型矩阵

打分口径：●=强／◐=可用有条件／○=弱或缺证据。"闭环适配度"只看一件事：能否把每轮迭代留下的信息变成可审计的结构化记录。

层	组件	许可/形态	接口成熟度	二次开发成本	闭环适配度	一句话理由
仿真	ASE 3.29.0	开源，官网 ase-lib.org	●	低（Python API）	●	结构/计算器/优化器全暴露，最容易包成 `simulate()`
仿真	QuAcc	开源，ASE 生态收录	◐	中	●	数据库驱动，天然适合批量候选与结果台账
仿真	GROMACS/Schrödinger 类引擎	商业或开源	◐	中高（作业层）	◐	本系列不重复讲，衔接前系列作业封装经验
执行	Opentrons 9.1.2	硬件+开源软件	●	低	●	协议即代码、本地可仿真、本机有 HTTP API 与 openapi.json
执行	PyLabRobot 0.2.2	开源	●	低	●	一份任务跨 backend，闭环可移植性的保险
执行	SiLA 2 0.14.0	开放标准	◐	中	●	标准侧解耦；需设备已实现 SiLA 服务
执行	Hamilton VENUS/Prep	厂商门户	◐	中，需向厂商索取	◐	method 校验/执行链完整，Python 面证据不足
执行	Tecan EVOware	闭源	○	高	◐	官方确认有 API 与 Developer Studio，但无公开 Python SDK
执行	Beckman Biomek/vWorks	闭源	○	高	○	vWorks Scripting 仅第三方记载，需联系厂商
记录	Benchling	商业云	●	低	●	三种鉴权 + 官方 SDK + App 服务账号可审计归属
记录	SENAITE 2.6.0	开源 GPL-2.0	●	中（buildout）	●	CRU + PUSH jobs，自建可控，成本最低
记录	LabArchives	商业云	◐	中	◐	REST-like + XML + 签名，解析与重试要多写代码
契约	ISA（isatools 0.14.3）	开源 CPAL	●	低	●	Investigation-Study-Assay 三段结构直接对应闭环（第 5 篇）
契约	AnIML	ASTM 标准	●	中（XML）	●	分析数据的 vendor-neutral 交换格式（第 5、8 篇）
学习	ax-platform 1.3.1	开源	●	低	●	`ax.api.client.Client` 天生 ask/tell
学习	BoTorch 0.18.1	开源	●	中	●	批量采集函数对应"一板 96 条件"
学习	GPyOpt / scikit-optimize	已归档	○	—	○	铁律 6：可讲思想，不进新闭环

二、完整代码与逐行剖析

2.1 最小闭环 demo 目录骨架（bash）

# 一次性建出"三层一总线"的物理结构。目录即架构，评审时不用画图也能对齐。
mkdir -p looplab/{contracts,contracts/tests} \
         looplab/simulation/{workflows,outputs} \
         looplab/execution/{protocols,layouts,httpapi} \
         looplab/record/{adapters,mappings} \
         looplab/learning \
         looplab/bus \
         looplab/docs
# 每层留一个包初始化文件，后续 from contracts import envelope 才可用
touch looplab/contracts/__init__.py looplab/simulation/__init__.py \
      looplab/execution/__init__.py looplab/record/__init__.py looplab/learning/__init__.py

cat > looplab/requirements-loop.txt <<'TXT'
ase==3.29.0
opentrons==9.1.2
pylabrobot==0.2.2
sila2==0.14.0
benchling-sdk==1.25.0
isatools==0.14.3
pydantic
requests
TXT
# 为什么单独一份 requirements-loop.txt：闭环项目的依赖必须与业务项目分开锁版（铁律 4）

python -c "import pathlib;print('\n'.join(sorted(str(p.relative_to('looplab')) for p in pathlib.Path('looplab').rglob('*'))))"

目录约定与理由：

目录	放什么	不许放什么
`contracts/`	pydantic 模型、`schema_version`、主题名常量	任何 HTTP 调用、任何厂商对象
`simulation/`	`simulate(params)->pred` 实现、QuAcc 风格工作流	硬编码的孔位/板型
`execution/protocols/`	Opentrons v2 协议文件	业务决策逻辑（该建议做什么不由协议决定）
`execution/layouts/`	`Deck.load_from_json_file` 用的 JSON 布局	—
`record/adapters/`	Benchling/SENAITE/LabArchives 三家适配器	直接给业务层用的函数（必须先转契约）
`bus/`	outbox 表与轮询器	消息中间件的厂商专有逻辑

2.2 pydantic 契约雏形 `looplab/contracts/envelope.py`

"""闭环消息信封与三类核心消息：总线上的第一个字节就是它。
注意：以下字段名是本系列的工程约定（不是任何厂商 API 的字段），
      厂商字段到这些字段的映射写在 record/mappings/ 里（第 7 篇）。
"""
from __future__ import annotations

from datetime import datetime, timezone          # 时间戳必须显式带时区：闭环横跨多台机器、可能跨区
from enum import Enum
from typing import Any, Literal

from pydantic import BaseModel, Field, field_validator   # pydantic v2 API

SCHEMA_VERSION = "1.0"        # 全局契约版本；改字段=升版本，禁止原地改语义


class Stage(str, Enum):
    """闭环的五个阶段常量。字符串枚举的价值：日志与数据库里可直接读。"""
    SUGGESTED = "suggested"    # 学习层给出建议（第 10/11 篇）
    SIMULATED = "simulated"    # 仿真层预测（第 9 篇）
    EXECUTED = "executed"      # 设备执行回执（第 4/6 篇）
    MEASURED = "measured"      # 仪器回传测量（第 8 篇）
    RECORDED = "recorded"      # 写回 LIMS 成功（第 7 篇）


class Envelope(BaseModel):
    """所有跨层消息的共同头。对应第 1.5 节的①②③④。"""
    schema_version: str = SCHEMA_VERSION          # ①契约版本：消费方先看它，再看 payload
    loop_id: str = Field(min_length=1)            # ②持久标识：一轮闭环一个，FAIR R1.2 的落点
    stage: Stage                                  # 消息属于哪个阶段
    trial_index: int | None = None                # 与 Ax 的 trial 对齐；铁律 5 要求可配对
    produced_at: datetime                         # 由生产者写入，不做服务端改写
    producer: str                                 # 例 "ax-client"/"opentrons-flex-D1"/"benchling-app"
    actor: str                                    # 人或服务账号身份；审计归属用（铁律 3）
    idempotency_key: str = ""                     # ④幂等键：重试不产生第二条记录，见 2.3

    @field_validator("produced_at")
    @classmethod
    def _must_be_tzaware(cls, v: datetime) -> datetime:
        # 为什么硬性要求：naive datetime 在跨机器比较时会产生"未来时间"，
        # 直接毁掉按时间排序的审计链（也是 Part 11 时间戳条款的常见踩点）
        if v.tzinfo is None:
            raise ValueError("produced_at 必须带时区，请用 datetime.now(timezone.utc)")
        return v

    @classmethod
    def now(cls, **kw: Any) -> "Envelope":
        """便捷构造：默认 UTC now，调用方忘了给时间也不会拿到 naive 值。"""
        kw.setdefault("produced_at", datetime.now(timezone.utc))
        return cls(**kw)


class Suggestion(Envelope):
    """学习层 → 执行层：一次实验建议。"""
    parameters: dict[str, float]                  # 参数名必须与仿真侧输入同名（否则标定错位）
    constraints_note: str = ""                    # 例："试剂 X 余量需 ≥ 2 mL"，交给闸门判断

    def to_ax_trial(self) -> tuple[int | None, dict[str, float]]:
        """交给 ax.api.client.Client 的 next_trials 结果消费方（第 10 篇）。"""
        return self.trial_index, dict(self.parameters)


class ExecutionAck(Envelope):
    """执行层 → 记录层：机器人跑完了，怎么跑的记录必须留下。"""
    protocol_name: str
    protocol_rev: str                             # 协议文件 Git rev：铁律 8 的"协议版本"
    robot: Literal["Flex", "OT-2"]                # 槽位体系不同，混写会让坐标解释错（第 4 篇）
    api_level: str = "2.27"                       # requirements 里的 apiLevel 显式落库
    dry_run: bool = True                          # 默认 True：铁律 2，未确认前不得记为实测


class Measurement(Envelope):
    """记录层 ←→ 仿真层：一次测量的最小诚实表达。"""
    quantity: str                                 # 量名，如 "absorbance_450nm"
    value: float | None = None                    # None 表示"未成功测得"，绝不允许填 0
    unit: str                                     # 单位必填：缺单位的消息在校验期就被拒
    uncertainty: float | None = None              # 不确定度；None 会在校验后被降权
    lod: float | None = None                      # 检出限；value<lod 时应置 quality_flag
    quality_flag: Literal["ok", "below_lod", "failed", "manual_review"] = "ok"
    instrument: str = ""
    raw_file: str = ""                            # 仪器原始文件位置（mzML/AnIML/CSV 的路径或链接）

    @field_validator("unit")
    @classmethod
    def _unit_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("unit 不得为空：这是闭环里最贵的一类静默错误")
        return v


def make_topic(stage: Stage, domain: str = "chem") -> str:
    """③主题命名：loop.<domain>.<stage>.<schema_version>。
    把版本放进主题名，是为了让新旧契约在过渡期可共存（铁律 4 的双轨策略）。"""
    return f"loop.{domain}.{stage.value}.v{SCHEMA_VERSION}"


if __name__ == "__main__":
    m = Measurement.now(
        loop_id="L-2026-09-05-01", stage=Stage.MEASURED, trial_index=3,
        producer="plate-reader-A", actor="svc-looplab-app",
        quantity="absorbance_450nm", value=0.031, unit="OD",
        uncertainty=0.004, lod=0.02, instrument="plate-reader-A", raw_file="runs/2026-09-05/plate01.csv",
    )
    m = m.model_copy(update={"quality_flag": "ok"})   # 0.031 > lod 0.02，标记合理
    print(make_topic(m.stage))                        # loop.chem.measured.v1.0
    print(m.model_dump_json(indent=2))                # 落库/入队都是这一串字节

2.3 幂等键与 outbox 表（bash + SQL）

python - <<'PY'
import sqlite3
con = sqlite3.connect("looplab/bus/outbox.sqlite")
con.executescript("""
CREATE TABLE IF NOT EXISTS outbox(
  idempotency_key TEXT PRIMARY KEY,     -- 主键即幂等：同 key 重投自动失败，不产生第二条
  loop_id TEXT NOT NULL,
  topic   TEXT NOT NULL,
  payload TEXT NOT NULL,                -- 存 Envelope.model_dump_json() 的原串，便于事后取证
  sent_at TEXT,
  tries   INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS ix_outbox_pending ON outbox(sent_at, topic) WHERE sent_at IS NULL;
""")
con.commit(); print("outbox ready")
PY

# 幂等键怎么生成：内容寻址，而不是时间戳。这样"同一份建议被重算两次"也能正确去重。
import hashlib, json
key = hashlib.sha256(json.dumps(m.model_dump(mode="json"), sort_keys=True).encode()).hexdigest()[:16]
# INSERT OR IGNORE INTO outbox(idempotency_key,loop_id,topic,payload) VALUES(?,?,?,?)

2.4 一张实验单：`looplab/docs/experiment.yaml`

# 实验单：把"业务想做什么"与"技术怎么做"彻底分开，调度器（第 15 篇）只读它
schema_version: "1.0"
loop_id: L-2026-09-05-01
domain: chem
objective: minimize
parameters:            # 参数名必须与 simulation.simulate() 的入参、contracts 里的键完全一致
  - {name: ph,       bounds: [5.0, 8.0],  unit: "pH"}
  - {name: conc_mM,  bounds: [0.1, 10.0], unit: "mM"}
budget: {max_trials: 20, reagent_reserve_pct: 15}   # 余量约束交给铁律 10 的人工闸门
stages: [suggested, simulated, executed, measured, recorded]   # 与 Stage 枚举同名
record_target: benchling        # 或 senaite / labarchives；适配器见 record/adapters/

三、常见报错与排查

1）现象：pydantic 报 ValidationError: produced_at 必须带时区。
根因：datetime.now() 是 naive 时间。解法：统一用 datetime.now(timezone.utc)，或直接用本文件的 Envelope.now() 便捷构造。不要为了让校验通过就把校验删掉——这条规则的存在理由在第 7 篇的审计时间戳。

2）现象：换了一台设备后所有孔位坐标错位。
根因：把 Flex 与 OT-2 的槽位体系混用（Flex 用 A1–A3/B1–B3/C1–C3/D1–D3 这类字母数字布局，OT-2 用数字槽位 “1”/“2”）。解法：让 ExecutionAck.robot 字段必须显式填写，并在适配层按 robot 分支；协议内部不要写死坐标字符串。

3）现象：from pylabrobot.liquid_handling.backends import STARBackend 成功，但 setup() 时报连接错误。
根因：backend 的 host/驱动未配置，或该设备并未真正暴露对应通道。解法：先用虚拟/仿真 backend 跑通任务逻辑（第 6 篇），实机连接信息需向厂商索取；Tecan、Beckman 无公开 Python SDK 属已知事实，不要试图找"另一个 pip 包"。

4）现象：pip install 时报找不到 benchling-api / pylabserver。
根因：包名不存在（前者不存在，后者是已消失的 PySciMe 一族）。解法：Benchling 官方 SDK 是 benchling-sdk；硬件无关层是 pylabrobot。写一份内部"包名白名单"到 docs/，能挡住大部分时间浪费。

5）现象：outbox 里同一条建议被消费两次，LIMS 出现重复记录。
根因：把幂等键算成了时间戳，或者消费者没用 INSERT OR IGNORE。解法：按 2.3 用内容寻址（对 payload 做哈希）；消费侧唯一约束必须在，别只靠"我记得不会重复"。

6）现象：同事把 Kafka 接进来后，回归测试发现同一轮 loop 的记录少了两条。
根因：只测了吞吐没测幂等与顺序。解法：传输层可以换，但②持久标识与④幂等键必须由你的消费者强制执行；把"缺回报的 trial 不许进下一轮拟合"（铁律 5）写成断言，而不是文档。

四、动手练习

练习 1｜让契约拒绝脏数据（客观判据）
分别构造三种消息并确认结果：unit=""、produced_at=datetime.now()、value=None 但 quality_flag="ok"。
判定标准：前两种必须抛 ValidationError；第三种（逻辑不自洽）当前不报错——请补一个 model_validator 让它在 value is None and quality_flag == "ok" 时失败，并给出你的单测通过截图/输出。

练习 2｜一份任务两个 backend（可观察输出）
用 PyLabRobot 的 LiquidHandler + Deck.load_from_json_file(...) 写一次吸吐（pick_up_tips / aspirate(vols=100) / dispense / return_tips），先在仿真 backend 上跑，再只改 backend 导入行（STARBackend ↔ OpentronsOT2Backend）复跑。
判定标准：两次运行任务代码零改动；把差异写进 docs/portability-notes.md，并列出至少一处仍需向厂商确认的能力边界。

练习 3｜选型矩阵本地化（结构化产出）
把 1.6 的矩阵按你们实验室真实设备与软件重排，新增一列"现有凭据/权限状态"。
判定标准：至少标出一个"闭环适配度=○"的组件并写出替代路径（例如 Tecan → PyLabRobot/SiLA 2）；矩阵里出现 schema_version 与 loop_id 两行说明。

五、小结与下一篇预告

本篇把闭环拆成三层一总线，并给出四条硬结论：一，分层的原因是时钟与故障代价不同频，不是为了好看；二，仿真层先按"一个 simulate() 函数"设计，需要吞吐时再换 QuAcc，契约不动；三，执行层的厂商接口成熟度差异极大（Opentrons 全开放、Hamilton 门户有但需索取、Tecan/Beckman 无公开 Python SDK），所以 PyLabRobot/SiLA 2 不是可选装饰而是必需品（铁律 9）；四，"总线"的本体是契约版本、持久标识、主题命名与幂等键，中间件只是可替换的传输（Kafka/RabbitMQ 属工程实践参考，无 LIMS 官方规范）。

第 1 篇给了动机与体检脚本，本篇给了地图和骨架。从下一篇开始动手：**第 3 篇《第一次握手：Benchling API 读实验数据》**会把本篇 record/adapters/ 的第一个文件写出来（https://{tenant}.benchling.com/api/v2、三种鉴权、benchling-sdk 与 rate_limit_remaining）；第 4 篇去 :31950 遥控 Opentrons；第 5 篇把本篇 Measurement 里那四个字段扩展成 ISA/AnIML/FAIR 对齐的完整 loop record；第 6 篇展开本篇只给结论的抽象层。

本篇认知问题回显（FAQ）

Q1：仿真实验闭环为什么必须切成仿真层、执行层、记录层三层？
A：三层的时间尺度（秒/天 vs 毫秒/分钟 vs 毫秒）、事务语义（可重跑幂等 vs 不可回滚的物理动作 vs 只追加不遮盖的审计记录）和故障模式（不收敛 vs 撞针/余量不足 vs 401/404/速率限制）完全不同。共享的只有数据契约，因此契约独立成"总线"层，用 schema_version、loop_id、主题命名、幂等键四件事粘合。

Q2：ASE 与 QuAcc 在闭环仿真层各自扮演什么角色？
A：ASE（3.29.0，官网 ase-lib.org）是单体系 API：ase.build.molecule + ase.calculators.emt.EMT + ase.optimize.BFGS，过渡态用 from ase.mep import NEB；QuAcc 是 ASE 官方生态页收录的高通量数据库驱动工作流平台（“high-throughput, database-driven computational materials science and quantum chemistry workflows”）。先按 simulate(params)->prediction 一个函数封装，量级上去了再换 QuAcc。注意 ASE 无官方 calibration 包。

Q3：Opentrons、Hamilton、Tecan、Beckman 的官方接口开放程度差别有多大，该怎么接？
A：Opentrons 最开放（Protocol API v2 + 本机 :31950 HTTP API，/openapi.json 为权威）；Hamilton 有 developer.hamiltoncompany.com 门户的 VENUS API / Prep API（method export/import/validation、loading/execution，REST + WebSocket ws://[IP]/NimbusLite/instinctevents，OpenAPI 在设备本机），Python SDK 素材未登记需向厂商索取；Tecan Freedom EVOware 官方确认"can be controlled by other software via its API"并有 Developer Studio，但无公开 Python SDK；Beckman vWorks Scripting Interface 仅第三方记载。结论：业务代码写进 PyLabRobot LiquidHandler（STARBackend/OpentronsOT2Backend）或 SiLA 2（sila2 0.14.0），厂商通道只做原生 method 校验执行。

Q4：Benchling、SENAITE、LabArchives 三种记录层的鉴权与数据形态差别是什么？
A：Benchling 用 REST API v2 形态 https://{tenant}.benchling.com/api/v2/...，三种鉴权（个人 API key 走 HTTP Basic、Apps OAuth2 client_credentials→Bearer 且 expires_in: 900、OIDC id token），官方 benchling-sdk 1.25.0；SENAITE LIMS 2.6.0（GPL-2.0，Plone/Zope，buildout eggs 装 senaite.jsonapi）JSONAPI 为 GET/POST 的 CRU，鉴权 Login/Logout 与 Basic Auth，并有 PUSH 自定义 jobs 端点；LabArchives 是 REST-like API 返回 XML（元素顺序不固定），用 Access Key ID + Password 计算 sig+expires，访问用户数据另需 uid，并按区域分 api./ukapi./euapi./caapi./auapi.labarchives.com。

Q5：闭环的"消息总线"应该直接上 Kafka 或 RabbitMQ 吗？
A：不必。Kafka/RabbitMQ/webhook 作为 LIMS 集成骨干没有任何厂商的官方一手规范，只能作为工程实践参考。本系列默认从 outbox 表（SQLite）+ 轮询起步，先落实四件事：schema_version 契约版本、loop_id 持久标识（FAIR R1.2）、loop.<domain>.<stage>.v<version> 主题命名、内容寻址的 idempotency_key；吞吐或延迟成瓶颈时再换传输层，契约保持不变。