AI Agent控制硬件设备：用Plumbing Spec构建标准化管道层

在实验室里让 AI 直接操作设备，最让人头疼的往往不是模型选型，而是设备之间五花八门的通信协议。你在 Agent 侧设计得很漂亮，结果到了设备端，有的走 Modbus，有的走 HTTP，有的只能通过串口读数据，还有的压根没有标准 API。近期 Anthropic 提出用一套 “plumbing spec” 来连接 AI Agent 与实验室器材、机器人，这个思路本质上就是在模型和设备之间补一条标准化的“管道层”。本文会从工程落地角度拆解这条管道层该怎么做，并给出一套最小可运行示例，帮助你把 Agent 从“对话玩具”变成真正能控制硬件的系统。

1. 为什么 Agent 需要一个 “plumbing spec”

1.1 从“会聊天”到“会动手”

大语言模型本身只会生成文本，它并没有直接操作设备的能力。要让 Agent 去控制一个温控器、读取一台光谱仪的数据、或者让机械臂做一个动作，必须先解决“模型怎么调用工具”的问题。这个调用过程和传统 API 对接不太一样：模型是动态决定调哪个函数、传什么参数的，而不是由开发者在代码里写死调用顺序。

很多人在第一步就会踩坑：把设备的 HTTP API 直接暴露给模型，然后让模型自己去拼 URL、拼 JSON。短期内看起来很灵活，但实际运行时会遇到大量问题。模型可能猜错参数名，可能把摄氏温度写成华氏温度，可能在设备未就绪时不断重试，甚至可能在一次任务里调用了一个危险操作。这些问题的根源不是模型不够聪明，而是我们缺少一层“能让模型和设备互相理解”的翻译层。

1.2 “plumbing spec” 到底解决什么问题

Anthropic 提出的 “plumbing spec”，字面意思是“管道规范”。这里的管道，指的不是水管，而是 AI Agent 在调用外部工具、设备、机器人时的连接链路。一条完整的管道至少要做三件事：描述设备能力、定义工具接口、处理命令和状态反馈。

第一件事是设备能力描述。设备不需要关心“模型是什么”，它只需要对外声明自己有哪些可操作项、每个参数的范围和单位是什么。第二件事是工具接口定义。Agent 侧需要把设备能力转化为模型可识别的工具函数，包括函数名、描述、参数名、参数类型、必填项等。第三件事是状态反馈闭环。设备执行命令后，要把结果、状态码、错误信息返回给 Agent，让模型能够根据反馈决定下一步动作。

如果这三件事没有统一规范，每接一台新设备就要为模型重新写一份工具定义，项目会越来越难维护。plumbing spec 的意义在于把“模型如何调用设备”定义为一种标准做法，让实验室设备和机器人可以用同样的方式被 Agent 调度。

1.3 适用场景：实验室自动化与机器人控制

这类规范最适合的场景是实验室自动化和机器人控制。实验室里通常有成百上千台设备：恒温箱、离心机、天平、移液工作站、显微镜。这些设备来自不同厂商，通信协议完全不同。如果能让 Agent 统一调用它们，就可以实现“帮我配一组 96 孔板，先加热到 37 度，再分装样本”这类复杂操作，而不需要人为编写一段固定流程脚本。

机器人控制场景也很典型。一个机器人有多个关节、多种传感器、不同运动模式。如果没有标准管道层，Agent 发出的指令和机器人控制器能识别的指令之间需要大量胶水代码。有了统一描述规范，Agent 可以把“把机械臂移动到位置 A，然后夹取零件”翻译成机器人控制器可执行的动作序列。

2. Agent 与硬件设备连接的整体架构

2.1 一条链路：模型 → 工具 → 协议 → 设备

要理解 plumbing spec 的作用，可以先从整体链路看起。下面是一个典型的 Agent 控制硬件设备流程，使用序号表达，不需要画图也能看明白：

1. 用户向 Agent 发出自然语言指令，例如“读取当前室内温度并记录到日志”。
2. Agent 根据指令决定调用哪个工具函数，并生成参数，例如调用read_temperature(sensor_id="sensor_01")。
3. 工具层把函数调用转换为标准化的设备命令，例如{"device_id": "sensor_01", "command": "read_temperature"}。
4. 协议桥接层通过 MQTT、HTTP、WebSocket 或串口把命令发给设备。
5. 设备执行命令，返回结果，例如{"value": 25.6, "unit": "celsius"}。
6. 工具层把结果转换为模型可读的文本或结构化对象，返回给 Agent。
7. Agent 根据结果生成自然语言回复。

这条链路的难点不在第 1 步和第 7 步，也就是模型理解自然语言的部分，其实现在已经做得比较成熟了。真正容易出问题的是中间那几步：工具定义是否和现实设备能力一致、命令能否被设备正确解析、结果是否能稳定回传。

2.2 管道层（plumbing）的职责

管道层是模型和设备之间的中间层，它的职责可以从功能上切分为五块。

第一块是设备接入，负责把不同厂商、不同协议的设备暴露为统一接口。第二块是能力注册，设备需要声明自己支持哪些操作。第三块是命令转换，把 Agent 发出的工具调用转换成设备能执行的具体指令格式。第四块是状态同步，包括设备状态、命令执行状态、错误状态。第五块是安全控制，包括权限校验、操作白名单、参数范围校验。

这五块功能合在一起，形成了一个稳定的“翻译层”。模型不需要知道仪器是哪个厂商的，也不需要知道通信协议是 TCP 还是 UDP，它只需要知道“有一台设备叫 temperature_controller，支持 set_temperature 操作，参数 temperature 的取值范围是 0 到 100”，然后按这个接口去调用就可以了。

2.3 统一设备描述与工具注册

管道层最核心的抽象是设备描述。一台设备可以描述为：

• 设备 ID 和名称。
• 设备类型，例如温控器、机械臂、光谱仪。
• 支持的操作列表，每个操作对应一个工具。
• 每个操作的参数定义，包含参数名、类型、单位、范围、是否必填。
• 返回结果的结构定义。

设备描述相当于设备和 Agent 之间的“合同”，两者的通信都围绕这份合同进行。Agent 侧工具注册表把设备操作“翻译”成模型能识别的工具，这样模型在生成调用时就有了明确的参数约束，不需要靠猜。

3. 环境准备与项目结构

3.1 技术选型

为了演示 plumbing spec 的落地思路，本文示例选择一组常见、易扩展的技术栈：

• Python 3.10+，用于 Agent 工具层和设备模拟服务。
• FastAPI，用于设备服务 HTTP API 和 Agent 工具调用回调。
• paho-mqtt，用于 MQTT 协议桥接。
• pydantic，用于设备描述和参数校验。
• openai 或 anthropic SDK 都可以接入模型，但本文为了聚焦管道层，先用一个模拟 Agent 调用的方式演示，再说明如何接入真实模型。

这里需要特别说明：Anthropic 公开提出的具体规范文本还在演进中，不同版本的实现可能不同。本文不会照抄某个版本，而是从工程角度实现一套“类似 plumbing spec 思路”的最小方案。你在实际项目中，可以按照本文的设备描述模型去对接任何 Agent 框架。

3.2 版本说明

下面的示例基于以下常见环境，但版本不强制锁定：

• Python 3.10 或更高版本。
• FastAPI 0.100 左右版本。
• paho-mqtt 1.6 或 2.x 版本。
• pydantic 2.x 版本。

如果你的环境版本不同，代码中的导入方式可能有细微差异。建议先创建虚拟环境，再安装依赖，避免污染全局环境。

3.3 项目目录

先创建一个名为agent-lab-bridge的项目目录，结构如下：

agent-lab-bridge/ ├── app.py # FastAPI 设备模拟服务 ├── device_schema.py # 设备描述与命令模型 ├── mqtt_bridge.py # MQTT 桥接层 ├── agent_tools.py # Agent 工具注册与调用 ├── requirements.txt # 依赖列表 └── README.md # 使用说明

这个结构足够简单，又能把不同职责分离开。后面每个文件都会给出完整代码。

4. 核心设计：设备描述与工具调用接口

4.1 设备描述 Schema

先来定义设备描述模型，也就是设备和 Agent 之间的合同。使用 Pydantic 可以有效校验参数类型和范围，避免脏数据进入设备控制链路。

# 文件路径：device_schema.py from enum import Enum from typing import Any, Dict, List, Literal, Optional from pydantic import BaseModel, Field class DeviceType(str, Enum): TEMPERATURE_CONTROLLER = "temperature_controller" ROBOT_ARM = "robot_arm" SPECTROMETER = "spectrometer" LABEL_PRINTER = "label_printer" class ParameterSchema(BaseModel): """描述设备操作的一个参数""" name: str = Field(..., description="参数名") type: Literal["integer", "number", "string", "boolean"] = Field(...) unit: Optional[str] = Field(None, description="单位，例如 celsius, cm") description: str = Field(..., description="参数含义") required: bool = Field(True, description="是否必填") minimum: Optional[float] = Field(None, description="最小值") maximum: Optional[float] = Field(None, description="最大值") enum: Optional[List[str]] = Field(None, description="允许的枚举值") class OperationSchema(BaseModel): """描述设备支持的其中一个操作""" operation_id: str = Field(..., description="操作 ID，例如 set_temperature") description: str = Field(..., description="操作含义描述") parameters: List[ParameterSchema] = Field(default_factory=list) class DeviceSchema(BaseModel): """设备描述，相当于设备和 Agent 之间的合同""" device_id: str device_name: str device_type: DeviceType operations: List[OperationSchema] class DeviceCommand(BaseModel): """设备命令，Agent 工具层发给设备的统一格式""" device_id: str operation_id: str arguments: Dict[str, Any] = Field(default_factory=dict) class DeviceResult(BaseModel): """设备执行结果""" device_id: str operation_id: str success: bool output: Optional[Any] = Field(default=None, description="输出数据") error: Optional[str] = Field(default=None, description="错误信息")

这段代码定义了三层结构：参数级ParameterSchema、操作级OperationSchema、设备级DeviceSchema，以及命令和结果模型。这样设计的好处是，设备只需对外提供一份DeviceSchema，Agent 工具层就能知道如何调用它。

4.2 命令消息模型

DeviceCommand是工具层和设备之间的统一命令格式。任何设备，不管底层协议如何，都接收这样的命令：

{ "device_id": "temperature_controller_01", "operation_id": "set_temperature", "arguments": { "temperature": 37.0, "hold_time": 120 } }

设备端收到这个命令后，需要把它转换成自己的具体动作。例如温控器会调用自己的控制 API，机械臂会解析出目标坐标和动作。统一命令格式的价值在于：Agent 侧只需要学会构造DeviceCommand，不需要处理每种设备的私有协议。

4.3 工具注册表

工具注册表负责把设备操作转换为模型工具定义。很多 Agent 框架支持tools参数，每个工具包含name、description、parameters三个要素。下面的代码演示如何从DeviceSchema自动生成工具定义。

# 文件路径：agent_tools.py from typing import Any, Dict, List from device_schema import DeviceSchema def build_openai_tools(devices: List[DeviceSchema]) -> List[Dict[str, Any]]: """把设备描述转换为 OpenAI 风格的 tools 定义""" tools = [] for device in devices: for op in device.operations: properties = {} required = [] for param in op.parameters: param_schema: Dict[str, Any] = { "type": param.type, "description": param.description, } if param.enum: param_schema["enum"] = param.enum if param.minimum is not None: param_schema["minimum"] = param.minimum if param.maximum is not None: param_schema["maximum"] = param.maximum properties[param.name] = param_schema if param.required: required.append(param.name) # 在参数里带上 device_id，让模型知道调用哪台设备 final_properties = { "device_id": {"type": "string", "description": "目标设备 ID，例如 temperature_controller_01"}, **properties, } required = ["device_id"] + required tools.append({ "type": "function", "function": { "name": op.operation_id, "description": f"[{device.device_id}] {op.description}", "parameters": { "type": "object", "properties": final_properties, "required": required, }, }, }) return tools

这个函数是管道层的核心，它把设备描述“翻译”成模型可以理解的工具定义。模型看到set_temperature这个工具时，会知道需要传temperature，并且范围是 0 到 100 摄氏度。这样生成的调用才安全可靠。

5. 完整实战：让 Agent 控制模拟实验室设备

5.1 模拟设备服务

我们先实现一个模拟实验室温控器服务。这个服务用 FastAPI 暴露 HTTP API，同时监听 MQTT 命令，模拟设备的真实行为。

# 文件路径：app.py import asyncio import json from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException from pydantic import BaseModel from device_schema import DeviceCommand, DeviceResult, DeviceSchema, DeviceType, OperationSchema, ParameterSchema # 模拟设备内部状态 device_state = { "current_temperature": 25.0, "target_temperature": None, "running": False, } def build_temperature_controller_schema() -> DeviceSchema: return DeviceSchema( device_id="temperature_controller_01", device_name="实验室温控器", device_type=DeviceType.TEMPERATURE_CONTROLLER, operations=[ OperationSchema( operation_id="set_temperature", description="设置目标温度，单位摄氏度", parameters=[ ParameterSchema( name="temperature", type="number", unit="celsius", description="目标温度", required=True, minimum=0, maximum=100, ), ParameterSchema( name="hold_time", type="integer", unit="seconds", description="到达目标温度后的保持时间", required=False, minimum=0, maximum=3600, ), ], ), OperationSchema( operation_id="get_temperature", description="读取当前温度", parameters=[], ), ], ) def execute_command(command: DeviceCommand) -> DeviceResult: """执行设备命令的模拟实现""" device_id = command.device_id op_id = command.operation_id if op_id == "set_temperature": temp = command.arguments.get("temperature") hold_time = command.arguments.get("hold_time", 0) if temp is None: return DeviceResult(device_id=device_id, operation_id=op_id, success=False, error="缺少 temperature 参数") # 模拟加热过程 device_state["target_temperature"] = temp device_state["running"] = True # 这里为了演示，直接异步缓慢逼近目标值 # 实际设备会通过传感器持续更新 current_temperature asyncio.create_task(simulate_heating(temp)) return DeviceResult( device_id=device_id, operation_id=op_id, success=True, output={ "target_temperature": temp, "hold_time": hold_time, "status": "heating_started", }, ) if op_id == "get_temperature": return DeviceResult( device_id=device_id, operation_id=op_id, success=True, output={"current_temperature": device_state["current_temperature"]}, ) return DeviceResult(device_id=device_id, operation_id=op_id, success=False, error=f"不支持的操作: {op_id}") async def simulate_heating(target: float): """模拟温度逐渐接近目标值""" while device_state["running"]: current = device_state["current_temperature"] diff = target - current if abs(diff) < 0.1: device_state["running"] = False return device_state["current_temperature"] = current + diff * 0.3 await asyncio.sleep(0.5) class MqttPublishStub(BaseModel): """用于 REST API 演示的请求体""" command: DeviceCommand @asynccontextmanager async def lifespan(app: FastAPI): print("设备服务已启动") yield app = FastAPI(title="Agent Lab Bridge 设备模拟服务", lifespan=lifespan) @app.get("/device/schema/{device_id}") async def get_device_schema(device_id: str): if device_id == "temperature_controller_01": return build_temperature_controller_schema().model_dump() raise HTTPException(status_code=404, detail="设备不存在") @app.post("/device/command") async def execute_device_command(payload: MqttPublishStub): command = payload.command result = execute_command(command) return result.model_dump()

这个模拟设备服务做了几件事：提供设备描述查询接口、提供命令执行接口、用异步任务模拟温度变化。你可以通过浏览器访问/device/schema/temperature_controller_01查看设备描述，也可以向/device/command发送命令。

需要注意，asyncio.create_task在 FastAPI 事件循环里使用是可行的，但真实设备接入时，这些逻辑应该放到设备驱动层，而不是和 HTTP API 混在一起。这里为了演示方便，做了一个极简模拟。

5.2 MQTT 桥接层

很多实验室设备和机器人并不暴露 HTTP API，而是通过 MQTT 消息进行通信。我们在 HTTP 服务之外增加一个 MQTT 监听器，模拟通过 MQTT 接收命令并把结果回传到指定主题。

# 文件路径：mqtt_bridge.py import json import paho.mqtt.client as mqtt from device_schema import DeviceCommand, DeviceResult, DeviceSchema from app import execute_command, build_temperature_controller_schema MQTT_BROKER_HOST = "localhost" MQTT_BROKER_PORT = 1883 MQTT_COMMAND_TOPIC = "lab/device/command" MQTT_RESULT_TOPIC = "lab/device/result" def on_connect(client, userdata, flags, reason_code, properties): print(f"MQTT 连接成功，reason_code={reason_code}") client.subscribe(MQTT_COMMAND_TOPIC) def on_message(client, userdata, msg): print(f"收到命令: {msg.payload.decode()} on topic {msg.topic}") try: payload = json.loads(msg.payload.decode()) command = DeviceCommand(**payload) result = execute_command(command) client.publish(MQTT_RESULT_TOPIC, result.model_dump_json()) except Exception as e: error_result = DeviceResult( device_id="unknown", operation_id="unknown", success=False, error=f"命令解析失败: {e}", ) client.publish(MQTT_RESULT_TOPIC, error_result.model_dump_json()) def start_mqtt_bridge(): client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2) client.on_connect = on_connect client.on_message = on_message client.connect(MQTT_BROKER_HOST, MQTT_BROKER_PORT, keepalive=60) client.loop_forever() if __name__ == "__main__": start_mqtt_bridge()

这段代码订阅了lab/device/command主题，收到命令后交给execute_command执行，再把结果发布到lab/device/result。这样 Agent 侧就可以通过 MQTT 发送DeviceCommand，再订阅结果主题获取返回值，整个过程中 HTTP 服务也不是必需的。

5.3 Agent 侧工具调用接入

现在用一个简化示例演示 Agent 侧如何调用设备。为了不依赖具体模型 SDK，我们先用一个“模拟模型决策”的脚本，把 plumbing 的完整链路跑通，再说明如何接入真实模型。

# 文件路径：agent_tools.py 追加 import json import requests from device_schema import DeviceCommand TOOLS = None def init_tools_from_device_service(device_service_url: str): """从设备服务拉取设备描述，构建工具列表，并缓存""" global TOOLS resp = requests.get(f"{device_service_url}/device/schema/temperature_controller_01") resp.raise_for_status() device_schema = DeviceSchema(**resp.json()) TOOLS = build_openai_tools([device_schema]) return TOOLS def call_device_tool(device_service_url: str, tool_name: str, arguments: dict): """执行 Agent 选择的工具，转换成 DeviceCommand 发给设备服务""" device_id = arguments.pop("device_id") command = DeviceCommand( device_id=device_id, operation_id=tool_name, arguments=arguments, ) resp = requests.post( f"{device_service_url}/device/command", json={"command": command.model_dump()}, ) resp.raise_for_status() return resp.json() def simulate_agent_decision(prompt: str): """ 真实项目中，会用 LLM 根据 prompt 和 TOOLS 输出 tool_calls。 这里用规则模拟一次 set_temperature 调用。 """ if "温度" in prompt and "设置" in prompt: return "set_temperature", {"device_id": "temperature_controller_01", "temperature": 37.0, "hold_time": 60} return None, None

这个示例把 Agent 决策部分简化成了规则匹配。真实场景中，你会把TOOLS传给模型，然后拿到模型返回的tool_calls，再把对应的工具调用转成DeviceCommand。plumbing 层的价值就在这里：模型返回什么工具、什么参数，都可能被管道层验证和转换，最后由管道层决定如何发往设备。

5.4 运行与验证

在运行之前，先安装依赖：

pip install fastapi uvicorn paho-mqtt requests pydantic

先启动设备模拟服务：

uvicorn app:app --host 0.0.0.0 --port 8000

再启动 MQTT 桥接层（如果本机没有 MQTT Broker，可先安装 Mosquitto 并启动）：

python mqtt_bridge.py

然后写一个简单测试脚本：

# 文件路径：test_demo.py import json from agent_tools import init_tools_from_device_service, call_device_tool, simulate_agent_decision SERVICE_URL = "http://localhost:8000" tools = init_tools_from_device_service(SERVICE_URL) print("工具列表:") print(json.dumps(tools, ensure_ascii=False, indent=2)) tool_name, arguments = simulate_agent_decision("请把温度设置到 37 度，并保持 60 秒") if tool_name: result = call_device_tool(SERVICE_URL, tool_name, arguments) print("设备执行结果:") print(json.dumps(result, ensure_ascii=False, indent=2)) else: print("Agent 未匹配到工具调用")

运行：

python test_demo.py

预期输出会打印两个部分。第一部分是工具列表，里面能看到set_temperature和get_temperature两个工具对模型的描述。第二部分是设备执行结果，模拟温控器返回了target_temperature、hold_time、status字段。之后你可以再查询一次设备状态，会看到当前温度在慢慢接近 37 度。

这样就完成了一个从设备描述、工具注册、命令执行到结果返回的闭环。接下来只要把simulate_agent_decision替换成真实的 LLM 调用，就能实现自然语言控制设备的效果。

6. 常见问题与排查思路

6.1 Agent 无法连接设备服务

现象：Agent 调用工具时，请求一直超时，或者收到Connection refused、Failed to connect等错误。

这种问题的原因通常集中在网络策略、服务状态和地址配置三个方面。先确认设备服务进程是否还在运行，curl http://localhost:8000/device/schema/temperature_controller_01能否正常返回。如果服务正常，再检查 Agent 运行环境是否能访问该地址。比如容器内运行 Agent，localhost指向容器自己，需要使用host.docker.internal或宿主机 IP 才能访问宿主机服务。

另一个容易被忽略的是 API 网关或防火墙配置。Agent 外部服务调用失败时，不一定是代码问题，可能是出网策略没有放行目标地址。排查时可以按顺序检查：服务进程、端口监听、网络连通性、防火墙白名单、API 地址是否拼错。

问题现象	常见原因	解决思路
连接超时	服务未启动或端口错误	检查进程和监听端口
Connection refused	服务启动失败或被防火墙拦截	查看服务日志，检查防火墙规则
缺少 API Key 或鉴权失败	需要配置认证信息	确认密钥、是否过期、权限范围
SSL 证书错误	使用 HTTPS 但证书不受信任	配置正确证书，或使用受信任的网关

6.2 工具参数和设备期望不一致

现象：模型成功调用了工具，但设备返回参数错误，或者设备执行了错误的值。

根本原因是设备描述写得不准确。比如参数范围写的是 0 到 100，但设备实际限制是 10 到 50；单位写成了celsius，模型却传了华氏度的值；枚举值没有写全，模型传入了未定义的状态。

排查这类问题，建议从三方面入手：查看设备原始手册或 API 文档，确认每个参数的真实约束；把设备描述的每个字段都填完整，特别是minimum、maximum、enum、unit；在管道层增加参数校验，如果模型传入的参数不在合法范围内，直接在工具层报错，而不是把异常参数发给设备。

这里要特别提醒：模型并不真正“理解”设备，它只是根据工具描述猜测参数。描述越精确，模型猜错的概率越低。实践中最常见的问题是把描述写得太模糊，比如只写“设置温度”，却没有写温度单位。建议每个参数描述都带上单位和使用示例。

6.3 设备状态反馈不及时

现象：Agent 发出了控制指令，但后续查询状态时返回的还是旧值，或者状态长时间不更新。

在本文的模拟设备中，温度是异步变化的，所以查询get_temperature时返回值会逐渐变化。如果真实设备也采用异步执行模式，管道层必须设计状态查询机制。一个做法是设备在完成命令后主动上报状态，另一个做法是 Agent 在需要结果时主动轮询设备。

设计时需要区分“命令已收到”和“命令已执行完成”两个状态。管道层应该把这两种状态都返回给 Agent，否则 Agent 可能会误以为命令执行成功。建议在DeviceResult中增加status字段，例如pending、running、success、failed，并把最终结果和命令状态分开。

6.4 安全与权限问题

Agent 控制设备的安全问题比普通 API 调用更敏感，因为设备操作可能涉及物理动作、加热、启停等危险操作。常见错误是直接把所有设备命令接口暴露出来，没有做身份认证和操作白名单。

在真实项目中，管道层一定要做三层控制：认证层确认“谁在调用”，权限层确认“这个调用者是否允许操作这个设备”，校验层确认“参数是否在安全范围内”。尤其是机器人控制，某些异常参数可能导致设备碰撞或损坏。最安全的做法是在管道层强制限制所有参数，即使是模型生成的调用也要经过校验，不能直接透传。

7. 最佳实践与工程建议

7.1 管道层不要塞业务逻辑

管道层的定位是“翻译和连接”，不应该承担业务流程编排。比如“先加热，再保温，然后记录结果”这种流程，应该由 Agent 或上层任务系统控制，而不是写在设备接入层。设备接入层一旦加入了业务逻辑，就很难做到通用，也不利于后续扩展新设备。保持管道层纯净，所有设备都只暴露原子操作，让 Agent 去组合这些操作。

7.2 重视反馈闭环

一个完整的 Agent 操作系统，不只是“发命令”和“收结果”，还需要关注上下文反馈。设备返回结果后，Agent 需要知道这次操作是否成功、当前处于什么状态、是否需要继续操作。反馈闭环如果做得不好，Agent 很容易进入死循环。建议在管道层记录每次请求的request_id，并关联设备和操作，方便追踪整条命令的执行过程。

7.3 最小权限与白名单

给模型开放设备操作权限时，不要一次性把所有操作都暴露出来。建议按任务场景动态注册工具：如果当前任务是环境监测，就只注册温度读取和报警相关工具；如果当前任务是温度控制，再注册设置温度相关工具。这个动态注册机制能显著降低模型误操作的风险。同时，所有危险操作都应该增加二次确认，或者要求由更高权限的调用者审批。

7.4 可观测性设计

AI Agent 控制设备时，问题定位的成本比普通软件更高。因为链路更长：自然语言 → 模型 → 工具调用 → 设备协议 → 物理动作。每个环节都可能出错，所以可观测性必须从第一天就设计。建议为每次工具调用生成唯一trace_id，记录模型生成的原始参数、校验后的参数、设备返回结果、耗时、异常堆栈。这样一旦出现问题，可以在日志中完整还原调用链。

7.5 向上兼容与版本化

设备固件升级、Agent 框架升级、模型版本升级，都可能影响管道层的稳定性。设备描述 schema 一定要做版本管理。比如DeviceSchema增加一个version字段，当不兼容的设备描述更新时，让 Agent 侧重新拉取工具列表。这样旧版本 Agent 和新版本设备之间不会因为参数变更而突然失效。

8. 总结与进一步学习

本文围绕 Anthropic 提出的 plumbing spec 思路，把 AI Agent 连接实验室设备和机器人的关键问题拆成了设备描述、工具注册、命令转换、状态反馈四个模块，并用 FastAPI、MQTT、Pydantic 实现了最小闭环示例。你可以把这个示例作为起点，替换成真实设备和真实模型，逐步构建自己的 Agent 硬件控制管道。

接下来可以继续关注几个方向：一是 Anthropic 官方以及主流 Agent 框架对工具调用的最新支持；二是实验室自动化和机器人领域现有的设备通信标准，例如 SiLA2、OPC UA，这些标准可以作为 plumbing spec 的底层协议来源；三是如何在管道层加入更完善的安全策略和审批流程，确保 AI Agent 在物理世界中的行为可控。

如果你正在做 Agent 与硬件设备对接，建议先跑通一个模拟设备，再逐步接入真实设备。你会在接入过程中发现，真正需要花时间的往往不是模型调用，而是设备描述的准确性和命令反馈的稳定性。希望这篇文章能帮你少踩一些坑。