第 10 章 端到端案例：机器人工具链¶
前九章的机制在机器人场景里会同时受三种约束：接口是流式的、动作有物理后果、失败可能不可逆。本章把 ROS 2 与 MuJoCo 一侧的能力封装成工具，走一遍"自然语言任务 → 工具调用 → 仿真验证 → 真机执行 → 轨迹复盘"的完整链路，并给出每一步的安全判据。接口语义以 ROS2-note 第 6–7 章为准，仿真建模以 mujoco-textbook 为准，本章只讲暴露成工具的那一层。
1. 工具面：从接口到 schema¶
ROS 2 的三类接口与工具的对应关系并不对称。话题（topic）是持续的数据流，没有请求与响应，必须包装成"取最近一帧并检查时效"的查询；服务（service）是请求与响应，可一对一映射；动作（action）是长任务，有目标、反馈、结果与取消，需要包装成带句柄的工具。
映射时三项要钉死：消息类型（一个话题只对应一个 schema）、坐标系（frame_id 必须在工具内部转换到任务坐标系）、单位（米与弧度进，不把角度与弧度混在参数里）。这三项一旦交给模型填写，同一个任务在不同轮里会落到不同坐标下，轨迹无法复现。
判据：不要给模型一个"发布任意话题"的工具。话题名、消息类型与取值范围都应固定在工具内部，模型只填有界参数。把发布权直接交给模型，等于绕过第 5 章的全部权限设计。
| ROS 2 接口 | 工具形态 | 关键参数 | 副作用 |
|---|---|---|---|
| 话题 | 查询最近一帧 | 超时、时效阈值 | 读 |
| 服务 | 同步调用 | 有界请求字段 | 视服务而定 |
| 动作 | 提交句柄 + 轮询或取消 | 目标、速度上限、超时 | 写或不可逆 |
2. 查询类工具¶
查询工具读状态：关节角、末端位姿、夹爪开合、相机帧、力传感器读数。返回结构应包含数值、时间戳与时效标志；超过时效阈值的数据标为过期，而不是静默当作当前值。
时效阈值由控制频率反推：100 Hz 的回路能容忍十毫秒量级的旧数据，人工确认场景可以放到百毫秒。图像与点云按体积单独设阈值，大帧的传输时延本身就超过控制周期。
判据：每个读数都带时间戳，每个使用点都检查时效。机器人状态变化快，拿十秒前的关节角做碰撞判断，比没有数据更危险，因为它看起来是有效的。
查询工具默认允许，但要限范围：能读哪些话题、单次返回多少、是否允许读图像。范围在工具内部固定，不通过参数放开。
3. 技能类工具¶
技能工具产生动作：移动到指定位姿、开合夹爪、抓取、放置。参数必须有界：目标位姿限制在工作空间内，速度与加速度有上限，超时可设，是否允许接触可声明。
速度与加速度上限不从模型参数里取，而是从安全层读：模型只能请求不高于上限的值，超出的请求在派发前被钳制到限值并回传调整后的值，而不是默默执行或直接报错——前者让模型以为按原值执行了，后者会让它反复重试同一个越界值。
长动作返回句柄，而不是阻塞到结束：句柄带目标标识、当前阶段、反馈与结果。轮询与取消都通过句柄进行。判据：每个技能工具都能被取消，且取消后的状态可查询；不能取消的动作不适合交给模型自主调用。
句柄至少包含五个字段：goal_id、阶段（已接受/执行中/已到位/已取消/已失败）、最新反馈、结果、取消凭证。取消是请求而不是结果，发出后仍要轮询到“已停在安全点”才算完成。
4. 验证类工具¶
验证工具只读，但决定分支：碰撞检查、工作空间检查、抓取稳定性判断、力阈值检查。它们的结果要可解释——不只返回布尔值，还要返回依据（哪对几何体相交、哪个关节超限、力峰值多少）。
结果取三值而不是两值：通过、不通过、未知。传感器数据过期、场景模型缺失、超时未算完都归入未知；未知不能当成通过，也不能直接当失败，它应当阻断自动执行并转人工或重算。
判据：验证工具的判据必须独立于技能工具的实现。用"技能说成功了"作为验证，是自证；用几何与力阈值复核，才是外部判据（第 8 章）。
5. 仿真先行¶
同一套工具 schema，背后换一个执行后端：仿真后端在 MuJoCo 里执行并可重置，真机后端走 ROS 2。模型看到的工具不变，变的是适配器。这样"先在仿真里跑一遍"不需要改提示、改 schema 或改任务描述。
适配器要连语义一起对齐：速度参数的含意（关节速度还是末端速度）、超时的计时起点、到位的判定条件（位置误差还是速度阈值）。仿真侧的状态回读要带“是否已重置/已加速”的标记，否则轨迹复盘时会把加速时间当成真实时长。
class MotionBackend:
def move_to_pose(self, pose, speed, timeout) -> Handle: ...
def state(self) -> dict: ... # 含时间戳与时效标志
class SimBackend(MotionBackend): # MuJoCo：可重置、可加速
...
class Ros2Backend(MotionBackend): # 真机：动作接口 + 反馈
...
判据：仿真与真机共用 schema，只换适配器。如果两边的工具参数不同，模型在仿真里学到的调用方式到真机就失效，"先仿真"只剩下心理安慰。
仿真的价值是可回滚：失败后重置场景再试，不消耗物料、不损伤设备。它不能替代真机验证的部分（接触摩擦、线缆干涉、传感器噪声）应在任务判据里显式标为未覆盖。
6. 安全硬门与看门狗¶
安全硬门必须位于模型之外，且模型无法关闭：速度与力矩限幅、工作空间边界、关节限位、碰撞停止、通信超时停止。硬门在控制器或安全层实现，harness 只能请求，不能改写阈值。
硬门的调节是单向的：运行期间只能收紧不能放宽，放宽需要停机改配置并留记录。阈值来源要可追溯（机型参数表、任务风险等级），不写在提示里，也不从工具参数传入。
看门狗负责"没有消息就是坏消息"：心跳中断、反馈停止、命令超时，都触发停止。判据：硬门失败时默认停止，而不是默认继续；任何"为了完成任务而临时放宽限幅"的路径都不应存在。
7. 真机确认与急停¶
真机执行前展示影响面：目标位姿、速度、预计时长、经过的工作空间、是否可能接触。确认门之后才下发；急停独立于软件，任何时刻可物理触发。
在途命令要等停稳再判定：已下发、执行中、已到位、已取消是四个状态，不能用"没返回"代替。超时后先查询实际状态，再决定重发还是中止；盲目重发会让同一动作执行两次。
8. 失败恢复与降级¶
失败按可恢复性分级：参数越界、目标不可达可在修正后重试；碰撞、力超限应停止并回退；通信中断先恢复连接再查状态；真机异常优先停机，不在现场自动重试。
降级顺序预先写死（仿真复现 → 换技能 → 缩小子任务 → 转人工），每一级带进入条件与最大次数。降级不是“越试越弱”的随意选择，而是按可逆性排序：先选不产生物理后果的路径，再选可撤销的，最后才考虑需要人工在场的。
判据：不可逆动作不做自动重试；可重试的只有读操作与幂等的写操作，且要有次数与间隔上限。
9. 完整链路示例¶
9.1 链路骨架¶
以"把红色方块放进盒子"为例，把各章的判据串成一条链路：
用户任务
→ [第 2 章] 模型调用，返回查询工具的调用
→ [第 4 章] 校验参数，返回场景摘要（含时间戳）
→ [第 3 章] 摘要与历史进入上下文，原始图像走引用
→ [第 4 章] 模型选择抓取技能，参数含目标位姿与速度上限
→ [第 5 章] 权限门：写操作，需确认；预演给出工作空间与时长
→ [第 10 章] 仿真后端执行，验证工具复核抓取与放置
→ [第 5 章] 真机确认门，展示影响面
→ [第 10 章] 真机执行，看门狗与硬门在线
→ [第 8 章] 结果回读与判据验收，轨迹落盘
每一环都有可失败点：查询过期、参数越界、上下文缺图、技能不可达、权限被拒、仿真与真机不一致、执行超时、验收不通过。排障时按轨迹找第一个异常节点（第 8 章），而不是从最后的报错往回猜。
工具粒度按"一次可撤销的物理动作"切，而不是按话题与服务切：切得过细，一次任务要几十次调用，预算与延迟都被拆碎；切得过粗，取消与回滚找不到可定位的边界。
9.2 落地形态：一次运行涉及的库¶
前九章的主线 harness 只用标准库，是为了让机制可见；真实交付会把每一层换成成熟库。常见五层与代表实现（只作对照，机制仍按前九章）：
| 层 | 代表库 | 职责 | 对应章 |
|---|---|---|---|
| 模型网关 | litellm、openai SDK |
多模型统一入口、本地成本表、降级 | 第 2 章 |
| 编排 | langgraph、langchain |
图状态机、检查点、组件库 | 第 1、6 章 |
| 工具协议 | mcp（Python SDK） |
MCP 服务端/客户端 | 第 7 章 |
| 服务与总线 | fastapi + uvicorn + sse_starlette、websockets、grpcio、nats-py；可视化服务 flask / dash + plotly；校验与 ORM 用 pydantic / sqlalchemy |
流式 HTTP、会话推送、事件总线、看板服务 | 第 9 章 |
| 存储与观测 | PostgreSQL（checkpoint）、Qdrant（向量）、Neo4j（图谱）、MinIO（对象）、OpenTelemetry → Jaeger、Prometheus + Grafana | 记忆与追踪 | 第 3、8 章 |
四种最小形态：
# 模型网关：model 前缀决定后端；成本表与重试在库内
from litellm import completion
resp = completion(model="openai/qwen-plus", messages=[{"role": "user", "content": "hi"}])
# 编排：图 = 节点 + 边；第 6 章的委派与终止约束在这里落成边条件
from langgraph.graph import END, StateGraph
g = StateGraph(dict)
g.add_node("plan", plan); g.add_node("act", act)
g.set_entry_point("plan")
g.add_edge("plan", "act"); g.add_edge("act", END)
app = g.compile() # checkpointer 接 PostgreSQL 即得断点续跑
# 流式服务：SSE 把 token 增量推给前端；与第 9 章的进度呈现对接
from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
app = FastAPI()
@app.get("/stream")
async def stream():
async def gen():
yield {"event": "token", "data": "..."}
return EventSourceResponse(gen())
# 事件总线：机器人与 agent 解耦的 pub/sub；topic 即第 9 章的会话边界
import asyncio, nats
async def main():
nc = await nats.connect("nats://127.0.0.1:4222")
await nc.publish("robot.event", b'{"kind":"skill_done"}')
asyncio.run(main())
MCP 服务端一行起手：from mcp.server.fastmcp import FastMCP; mcp = FastMCP("robot")，再用 @mcp.tool() 挂函数（协议细节见第 7 章）。
判据：换栈时按上表逐层替换，不要两层混用（自写循环与 LangGraph 同时记账，成本与终止预算会算两遍）；版本要钉（litellm、langgraph 的接口在次版本间会变），钉法同第 2 章的模型接口判据。
10. 要点¶
- 话题包装成带时效的查询，服务一对一映射，动作包装成可取消的句柄。
- 不给模型发布任意话题的工具；参数范围固定在工具内部。
- 查询读数必带时间戳，使用前检查时效。
- 技能工具参数有界、可取消、状态可查。
- 验证判据独立于技能实现，避免自证。
- 仿真与真机共用 schema，只换适配器；仿真覆盖不到的部分显式标注。
- 安全硬门在模型之外且不可关闭；看门狗在无消息时停止。
- 不可逆动作不自动重试；在途命令等停稳再判定。
- 排障看轨迹里的第一个异常节点，不从最后一条报错倒推。