最近在整理 AI Agent 接入物理设备的思路时看到一条标题为 “Anthropic proposes plumbing spec to link AI agents to lab kit and robots” 的技术讨论。很多朋友看到 plumbing 第一反应是水管但在 Agent 工程里plumbing 指的是模型与工具、工具与设备、设备与状态反馈之间那条“看不见但绕不开”的连接管道。没有这套管道单纯把大模型接上机械臂或实验仪器往往只能停留在 Demo 阶段。本文将结合 plumbing spec 的设计主张拆解一套从设备服务到 Agent 工具循环的最小落地实现并给出排错和工程化建议。1. 背景与核心概念1.1 AI Agent 接设备时到底难在哪当一个 AI Agent 需要控制实验室设备或机器人时我们通常会觉得最难的是“让模型听懂人话”比如把“把加热台升温到 60 度”拆解成一条控制指令。但在真实项目中真正消耗时间的往往是模型之外的连接问题。设备厂商不同通信协议就不同。有些设备走串口有些走 USB有些走 HTTP有些走私有 TCP 二进制协议。设备能力描述也不统一同一类“加热台”可能叫set_temperature也可能叫temp_ctrl或heat_to。状态上报更是五花八门有的设备返回温度值有的返回一个字符串有的需要主动轮询有的会推送事件。AI Agent 要同时对接多种设备时这些差异会被无限放大。如果没有统一规范和适配层Agent 代码里就会塞满各种 if-else 分支。今天接一个加热台明天接一个机械臂后天接一个液滴分配器每接一个设备都要改一遍业务逻辑。这种状态在工程上很难维护也不可扩展。plumbing spec 要解决的核心问题正是这种“设备侧碎片化”带来的接入成本。1.2 什么是 plumbing specplumbing spec 不是一个具体的软件也不是某个 SDK而是一组关于“智能体如何连接外部设备”的约定和设计思路。它借鉴了水管工程的思想水管本身不显眼但整套房子能不能住人很大程度上取决于水管是否顺畅、接口是否统一、阀门是否可靠。放到 AI Agent 场景里这套“管道规范”通常包括几部分设备通信协议Agent 和设备之间用什么协议传输命令比如 HTTP、MQTT、WebSocket。设备能力描述如何告诉 Agent 一台设备有哪些可调用能力比如通过 JSON Schema 定义工具参数。命令格式一次设备调用应该包含哪些字段比如设备 ID、命令 ID、命令名、参数。状态格式设备返回给 Agent 的结果应该如何组织比如状态码、执行结果、时间戳、当前设备状态。错误处理与权限控制设备调用失败时如何反馈敏感操作如何授权。当这些约定统一后Agent 侧只需要理解一种“通用语言”不同设备通过适配器转换成各自的私有协议。这样既降低了模型侧的复杂度也让后续接入新设备变成增量工作而不是推倒重来。1.3 典型应用场景这套思路在以下几个场景里非常有价值。实验室自动化是典型场景。一个自动化实验平台可能同时包含加热台、离心机、移液器、酶标仪等多种仪器。科研人员希望用自然语言描述实验步骤Agent 根据步骤调用不同仪器的接口并汇报每一步的执行结果。机器人控制也是重点方向。桌面机械臂、移动底盘、无人机等设备通常有实时控制要求命令频率高安全性要求也高。Agent 需要把任务拆成一系列控制动作并持续读取设备状态确认动作是否真正完成。还有一些边缘场景比如自动化测试设备、数据采集终端、智能硬件网关。这些设备往往网络环境复杂接口差异更大更需要一层稳定的管道规范。2. 理解 Agent 工具调用与设备连接规范2.1 Agent 的感知-决策-执行链路AI Agent 接入物理设备时通常遵循一条“感知-决策-执行”的闭环链路。感知阶段Agent 需要知道当前有哪些设备可用、每台设备处于什么状态。比如查询设备列表读取加热台的当前温度读取机械臂的当前坐标。决策阶段Agent 根据用户意图和感知到的信息从工具列表里选择一个最合适的工具并提取必要参数。比如用户说“把加热台升到 60 度”模型就应该选择thermostat_set_temperature这个工具参数是{value: 60}。执行阶段Agent 将工具调用转发给设备适配层或设备服务由服务完成真实硬件控制并把执行结果回传给模型。模型根据返回结果生成最终回复比如“加热台已开始升温”。这个链路看起来简单但每一层都可能出问题。plumbing spec 的价值在于它把每一层之间的接口都标准化了让 Agent 不需要关心某台设备底层是串口还是 HTTP只需要关心“这个工具接受什么参数、返回什么结果”。2.2 工具调用的最小数据闭环下面是一个最小可执行的工具调用闭环。第一步Agent 拿到用户指令例如“请把加热台设置到 60 摄氏度”。第二步Agent 在本地维护一份工具清单每个工具包含名称、描述、参数 schema。模型根据用户指令和工具描述判断应该调用哪个工具并生成一条结构化的 tool call 消息。第三步Agent 运行时拦截这条 tool call解析出工具名和参数然后调用设备适配器中对应的方法。第四步适配器把参数转换成设备服务要求的 JSON 报文通过 HTTP 发送给设备服务。设备服务校验参数后把命令下发给真实硬件。第五步设备服务返回执行结果适配器把结果原样返回给 Agent 运行时。运行时再将结果作为一条 tool result 消息追加到对话上下文中并继续调用模型让模型生成最终回复。整个闭环和普通函数调用非常相似区别只是多了一个“模型来决定调用哪个函数”的环节。所以我们在设计设备接口时应该把它当成一个“可被 AI 调用的函数”来设计而不是传统的内部接口。2.3 三个关键组件要把 plumbing 规范落地通常需要三个关键组件。第一个是统一协议层。它定义了 Agent 与设备服务之间通信的报文格式。最简单的方式是用 JSON over HTTP优点是调试方便、生态成熟几乎任何语言都能解析。后续需要实时性时再考虑 WebSocket 或 MQTT。第二个是设备适配层。每个设备对应一个适配器适配器负责把通用协议转换成设备私有协议。设备私有协议可能是厂商 SDK也可能是串口指令。适配器隔离了差异让上层 Agent 保持稳定。第三个是 Agent 工具描述层。它把适配器暴露的方法转换成模型能理解的工具 schema。描述越清晰模型选对工具的概率就越高。比如工具名要带领域前缀描述要写清楚单位、取值范围、副作用参数要用 JSON Schema 描述。3. 环境准备与项目结构3.1 运行环境说明下面用一个最小示例来演示整套流程。示例环境如下操作系统Windows / macOS / Linux 均可。Python 版本3.10 或更高因为代码中使用了dict[str, Any]类型注解。依赖库FastAPI、Uvicorn、Pydantic、Requests。开发工具任意文本编辑器或 IDE推荐 VSCode 或 PyCharm。如果你本机还没有安装 Python可以先从官网下载 Python 3.10 以上版本安装时勾选“Add Python to PATH”。版本不需要严格固定只要保证 Python 3.10 以上即可。FastAPI 和 Pydantic 的版本建议安装当前稳定版不要使用太老的版本。3.2 安装依赖建议先创建虚拟环境避免污染全局 Python 环境。在项目根目录执行以下命令。python -m venv venv source venv/bin/activateWindows 下激活命令稍有不同venv\Scripts\activate然后安装依赖。pip install fastapi uvicorn pydantic requests安装完成后可以用下面的命令检查版本。python -c import fastapi, uvicorn, pydantic, requests; print(deps ok)如果出现依赖版本冲突建议使用一个新的虚拟环境重新安装不要强行升级全局依赖。3.3 项目结构说明为了演示清晰我们把项目拆成三个核心文件。plumbing-spec-demo/ ├── device_server.py # 设备服务端模拟实验室设备和机器人 ├── device_adapters.py # 设备适配器层把 Agent 工具调用转换成设备请求 ├── agent_loop.py # Agent 工具循环负责调用模型和执行工具 └── requirements.txt # 可选依赖清单device_server.py构建一个 HTTP 服务模拟加热台和机械臂。device_adapters.py提供两个适配器类分别是加热台适配器和机械臂适配器。agent_loop.py实现一个最简 Agent 运行器它维护工具列表接收用户指令调用模型执行工具再生成最终结果。4. 完整实战从设备服务到 Agent 调用4.1 定义统一命令报文在设计设备接入接口时第一步不是写代码而是约定报文。我们定义一个统一的命令请求 JSON 格式。{ protocol_version: 1.0, device_id: thermostat-001, command: { id: cmd-1234, name: set_temperature, params: { value: 60 } } }字段含义如下。protocol_version协议版本号。后续协议进化时通过版本号兼容旧设备。device_id目标设备 ID必须全局唯一。command.id命令 ID由调用方生成。建议使用 UUID用于日志追踪和幂等去重。command.name命令名对应设备能力。command.params命令参数具体字段由各设备能力决定。设备服务端返回结果也需要统一格式。{ protocol_version: 1.0, status: accepted, command_id: cmd-1234, result: { current_temperature_c: 22.0, target_temperature_c: 60.0 }, ts: 2025-01-01T12:00:00Z }status表示命令是否被接受result是设备执行后的状态快照ts是服务端时间戳。统一格式的好处是Agent 无论控制什么设备都只需要解析同一种结构。4.2 编写设备服务端设备服务端用 FastAPI 实现模拟一台加热台和一台桌面机械臂。实际项目中这一层会通过厂商 SDK 与真实硬件通信这里用内存状态代替。# 文件路径plumbing-spec-demo/device_server.py from datetime import datetime, timezone from typing import Any from uuid import uuid4 from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field app FastAPI(titleLab Device Plumbing Server) # 模拟设备状态真实项目中这些数据来自硬件或设备网关 devices { thermostat-001: { name: 加热台, type: thermostat, status: idle, current_temperature_c: 22.0, target_temperature_c: None, }, robot-arm-001: { name: 桌面机械臂, type: robot_arm, status: idle, pose: {x: 0, y: 0, z: 0}, }, } class CommandDetail(BaseModel): id: str Field(default_factorylambda: str(uuid4()), description命令ID) name: str Field(..., description命令名) params: dict[str, Any] Field(default_factorydict, description命令参数) class CommandRequest(BaseModel): protocol_version: str Field(default1.0, description协议版本) device_id: str Field(..., description目标设备ID) command: CommandDetail Field(..., description命令内容) app.get(/devices) def list_devices(): return {protocol_version: 1.0, devices: list(devices.keys())} app.get(/devices/{device_id}/status) def get_device_status(device_id: str): if device_id not in devices: raise HTTPException(status_code404, detaildevice not found) return { protocol_version: 1.0, device_id: device_id, state: devices[device_id], ts: datetime.now(timezone.utc).isoformat(), } app.post(/devices/{device_id}/commands) def execute_device_command(device_id: str, req: CommandRequest): if req.device_id ! device_id: raise HTTPException(status_code400, detaildevice_id mismatch) if device_id not in devices: raise HTTPException(status_code404, detaildevice not found) device devices[device_id] command_name req.command.name params req.command.params command_id req.command.id # 真实项目中这里应该通过厂商 SDK 或串口指令控制硬件 if device[type] thermostat: if command_name set_temperature: target float(params.get(value, 0)) device[target_temperature_c] target device[status] running return { protocol_version: 1.0, status: accepted, command_id: command_id, result: { current_temperature_c: device[current_temperature_c], target_temperature_c: target, }, ts: datetime.now(timezone.utc).isoformat(), } if device[type] robot_arm: if command_name move_to: x float(params.get(x, 0)) y float(params.get(y, 0)) z float(params.get(z, 0)) device[pose] {x: x, y: y, z: z} return { protocol_version: 1.0, status: accepted, command_id: command_id, result: {pose: device[pose]}, ts: datetime.now(timezone.utc).isoformat(), } raise HTTPException(status_code400, detailfunsupported command: {command_name})这段代码做了几件事。首先定义了请求模型用于校验外部传入的 JSON 是否符合统一报文规范。然后提供了两个查询接口分别用于获取设备列表和设备当前状态。最后实现命令执行接口根据设备类型和命令名分发到不同处理逻辑。注意我在返回结果时把command_id原样返回了。这是为了给上层 Agent 一个追踪依据后续查询日志或排查问题时非常有用。4.3 编写设备适配层设备适配层的作用是屏蔽设备服务细节让 Agent 侧只需要面向 Python 方法调用。每个适配器都继承同一个基类并实现to_tool_schema方法便于统一注册成工具。# 文件路径plumbing-spec-demo/device_adapters.py import uuid from abc import ABC, abstractmethod import requests class BaseDeviceConnector(ABC): def __init__(self, base_url: str, device_id: str): self.base_url base_url.rstrip(/) self.device_id device_id abstractmethod def to_tool_schema(self) - dict: 返回符合模型工具调用规范的 schema pass def send_command(self, name: str, params: dict) - dict: 把统一命令报文发送给设备服务端 payload { protocol_version: 1.0, device_id: self.device_id, command: { id: str(uuid.uuid4()), name: name, params: params, }, } response requests.post( f{self.base_url}/devices/{self.device_id}/commands, jsonpayload, timeout10, ) response.raise_for_status() return response.json() class ThermostatConnector(BaseDeviceConnector): def to_tool_schema(self) - dict: return { name: thermostat_set_temperature, description: 设置加热台的目标温度单位摄氏度。, input_schema: { type: object, properties: { value: { type: number, description: 目标温度例如 60 表示 60 摄氏度, } }, required: [value], }, } def set_temperature(self, value: float) - dict: return self.send_command(set_temperature, {value: value}) class RobotArmConnector(BaseDeviceConnector): def to_tool_schema(self) - dict: return { name: robot_arm_move_to, description: 移动机械臂末端到指定的空间坐标单位为毫米。, input_schema: { type: object, properties: { x: {type: number, description: x 坐标}, y: {type: number, description: y 坐标}, z: {type: number, description: z 坐标}, }, required: [x, y, z], }, } def move_to(self, x: float, y: float, z: float) - dict: return self.send_command(move_to, {x: x, y: y, z: z})BaseDeviceConnector中的send_command是通用部分它负责把命令名和参数打包成统一 JSON再通过 HTTP 发送。to_tool_schema是每个设备必须实现的方法返回的是模型工具调用规范需要的工具描述。从工程角度看适配器层是整套 plumbing 规范里最值得投资的部分。设备厂商 SDK 更新时只需要改适配器内部逻辑Agent 上层完全不用动。4.4 编写 Agent 工具循环Agent 工具循环是连接大模型和工具调用的桥梁。为了不绑定具体厂商我这里用一个mock_llm_chat模拟模型返回。实际项目中