route→service→driver 分层Java 架构思维、分层逻辑进行分析三层职责routeFlask 路由 只负责 HTTP解析请求、校验 API Key、返回 JSON、挂 request_id│service业务层 只负责动作分发/ARM_BUSY 互斥/错误收敛不 import Flask│driverArmPort MockArm / FakeArm / PantheraArmAdapter只实现 status/move_to_pose/gripper/estopRoute层Controller控制器 ↓ ArmServiceService业务层 ↓ ArmPortDAO接口契约 ↓ MockArm / RealArmDAO实现模拟器 或者 真实机械臂硬件驱动1.driver层MockArmfrom __future__ import annotations from dataclasses import dataclass, field from typing import Any, Protocol class ArmPort(Protocol): 上层 Agent 依赖的最小机械臂契约。 def status(self) - dict[str, Any]: ... def move_to_pose(self, x: float, y: float, z: float, speed: float 0.3) - bool: ... def gripper(self, action: str) - bool: ... def estop(self) - None: ... dataclass class MockArm: 内存 Mockidle/moving/error/estopped 状态 可注入故障。 state: str idle position: tuple[float, float, float] (0.0, 0.0, 0.0) gripper_open: bool True fault_step: str | None None # 注入在该步骤失败一次 _fault_used: bool field(defaultFalse, initFalse) def status(self) - dict[str, Any]: return { state: self.state, position: list(self.position), gripper_open: self.gripper_open, } def move_to_pose(self, x: float, y: float, z: float, speed: float 0.3) - bool: if self.state estopped: return False if self.fault_step move and not self._fault_used: self._fault_used True self.state error return False self.state moving self.position (x, y, z) self.state idle return True def gripper(self, action: str) - bool: if self.state estopped: return False if action close and self.fault_step gripper and not self._fault_used: self._fault_used True self.state error return False if action open: self.gripper_open True elif action close: self.gripper_open False else: return False return True def estop(self) - None: self.state estopped整体架构1.ArmPort接口 / 契约层对应 Java 的interface定义机械臂必须具备哪些功能查询状态移动坐标控制夹爪紧急停止作用上层 Agent、Service 层只认接口不认具体设备2.MockArm实现层 / 虚拟驱动对应 Java 实现类implements ArmPort纯内存模拟一台假机械臂包含机械臂状态空闲 / 运动 / 报错 / 急停当前 XYZ 坐标夹爪开合状态故障注入开关核心亮点代码链路分析1. 接口 ArmPort标准规范规定所有机械臂真的、模拟的必须有这 4 个方法status ()我是谁、我在哪、我状态咋样move_to_pose ()移动到指定坐标gripper ()开爪 / 关爪estop ()急停锁死2. MockArm 全局状态模拟器数据state设备状态 idle 空闲 /moving 运动 /error 故障 /estopped 急停position当前机械臂末端 XYZ 坐标gripper_open夹爪是否打开fault_step故障注入点位可以设置让「移动失败」或「夹爪失败」_fault_used故障只触发一次单次故障不卡死3. move_to_pose 移动逻辑最核心流程如果已经急停 → 直接拒绝移动如果设置了【move 故障】且还没触发过标记故障已触发设备进入 error 报错状态返回执行失败正常情况状态切为 moving 运动中更新当前坐标运动结束切回 idle 空闲返回成功通俗理解可以人为让机械臂移动时报错专门测试上层 Agent 会不会处理故障、重试、兜底。4. gripper 夹爪逻辑急停状态 → 禁止操作如果设置【gripper 故障】 关爪动作 → 触发单次故障、进入报错正常 open/close 修改夹爪状态5. estop 急停逻辑一键锁死设备进入 estopped 状态后续所有动作全部失效。2.service重点不启动Flask先把业务逻辑从app.py抽成service.py机械臂服务层route → service → driver 的中间层。 只依赖 ArmPort 协议不 import Flask负责动作分发、ARM_BUSY 互斥与错误收敛。 from __future__ import annotations from typing import Any from ..backend.mock_arm import ArmPort TASK_BUSY TASK_BUSY class ArmService: 把 ArmPort 动作封装成业务方法并维护忙状态ARM_BUSY。 def __init__(self, arm: ArmPort) - None: self.arm arm self._busy False def dispatch(self, action: str, params: dict[str, Any] | None None) - dict[str, Any]: params params or {} if self._busy: return {success: False, code: TASK_BUSY, message: 机械臂忙} self._busy True try: ok self._execute(action, params) return {success: ok, message: ok if ok else step failed} finally: self._busy False def status(self) - dict[str, Any]: return self.arm.status() def _execute(self, action: str, params: dict[str, Any]) - bool: if action gripper_open: return self.arm.gripper(open) if action gripper_close: return self.arm.gripper(close) if action move_to_pose: return self.arm.move_to_pose( float(params.get(x, 0.0)), float(params.get(y, 0.0)), float(params.get(z, 0.0)), float(params.get(speed, 0.3)), ) if action estop: self.arm.estop() return True return False整体项目三层架构对标 Java 经典Controller → Service → DAORoute层Controller控制器 ↓ ArmServiceService业务层就是这份代码 ↓ ArmPortDAO接口契约 ↓ MockArm / RealArmDAO实现模拟器 或者 真实机械臂硬件驱动代码链路分析1. 成员变量def __init__(self, arm: ArmPort) - None: self.arm arm self._busy Falseself.arm依赖注入只认接口ArmPort。你传 MockArm 就是跑仿真以后传 RealArm直接对接真实机械臂Service 代码一行不动。Java 理解Service 不 new 实现类外部把实例传进来面向接口编程方便单元测试、切换实现。self._busy False业务软锁。大白话机械臂同一时间只能干一件活。False空闲True正在干活。防止同时下发移动、夹爪多条指令把设备搞乱。 ⚠只是内存标记不是操作系统锁。单进程完全够用Flask 多进程模式这个锁会失效。常量TASK_BUSY忙状态的错误码返回给上层 Route/Agent 识别。2. dispatch () 对外总入口最重要方法def dispatch(self, action: str, params: dict[str, Any] | None None) - dict[str, Any]: params params or {} if self._busy: return {success: False, code: TASK_BUSY, message: 机械臂忙} self._busy True try: ok self._execute(action, params) return {success: ok, message: ok if ok else step failed} finally: self._busy False大白话执行流程上层传过来两件东西action动作名字字符串params动作需要的参数比如移动的 x/y/z 坐标如果_busyTrue机械臂正在干活直接拒绝任务返回告诉上层设备忙请稍后再发。把_busy置为 True标记我开始干活了新来的任务别进来。try执行真正动作_execute()✨finally是保命逻辑不管底层执行成功、执行失败、甚至代码抛出异常崩溃finally 一定会跑强制把_busy改回 False。 避免出现硬件报错卡死_busy永远 True整个机械臂永久拒绝所有新任务。返回统一格式字典{success, code, message}上层 Route 或者 LLM Agent 不用猜返回格式直接读取success判断任务成没成。Java 类比Controller 调用 Service 的dispatch方法Service 做前置校验、执行业务、统一封装返回 VO 对象。3. status () 状态查询def status(self) - dict[str, Any]: return self.arm.status()大白话查询机械臂当前位置、夹爪、设备状态。 查询不抢占_busy 锁机器干活的时候也允许读状态不能查询也被挡住。直接透传给底层驱动。4. _execute () 私有动作分发器内部使用外部不能直接调用def _execute(self, action: str, params: dict[str, Any]) - bool: if action gripper_open: return self.arm.gripper(open) if action gripper_close: return self.arm.gripper(close) if action move_to_pose: return self.arm.move_to_pose( float(params.get(x, 0.0)), float(params.get(y, 0.0)), float(params.get(z, 0.0)), float(params.get(speed, 0.3)), ) if action estop: self.arm.estop() return True return False大白话翻译层。 上层给字符串move_to_pose这个函数把字符串翻译成真正的方法调用。params.get(x,0.0)参数如果没传 x就给默认 0.0不会直接报错崩溃做参数容错。estop急停调用底层急停直接返回 True急停优先这里不校验返回传过来不认识的 action返回 False代表动作不支持。Java 类比类似于策略模式根据 action 字符串调用不同的 DAO 方法。3.route层Flask test_client四场景)Flask Mock 机械臂后端把 MockArm(ArmPort) 包装成 HTTP 服务。 from __future__ import annotations import os import uuid from flask import Flask, Response, jsonify, request from ..backend.mock_arm import ArmPort, MockArm API_KEY os.environ.get(ARM_API_KEY, demo-key) def _err(code: str, message: str, http_status: int) - tuple[Response, int]: return jsonify( {code: code, message: message, retryable: False, details: {}} ), http_status def create_app(arm: ArmPort | None None) - Flask: app Flask(__name__) arm arm if arm is not None else MockArm() app.before_request def require_api_key() - tuple[Response, int] | None: if request.headers.get(X-API-Key) ! API_KEY: return _err(UNAUTHORIZED, 无效或缺失 API Key, 401) return None app.after_request def add_request_id(resp: Response) - Response: resp.headers[X-Request-ID] request.headers.get(X-Request-ID) or uuid.uuid4().hex return resp app.errorhandler(Exception) def handle_unexpected(_exc: Exception) - tuple[Response, int]: # 未知异常统一 INTERNAL_ERROR不泄露堆栈 return _err(INTERNAL_ERROR, 系统内部错误, 500) app.post(/v1/arm/actions) def actions() - tuple[Response, int]: data request.get_json(silentTrue) if not isinstance(data, dict): return _err(INVALID_TASK, 请求体必须是 JSON 对象, 400) action data.get(action) if action gripper_open: ok arm.gripper(open) elif action gripper_close: ok arm.gripper(close) elif action move_to_pose: ok arm.move_to_pose( float(data.get(x, 0.0)), float(data.get(y, 0.0)), float(data.get(z, 0.0)), float(data.get(speed, 0.3)), ) elif action estop: arm.estop() ok True else: return _err(INVALID_TASK, f未知动作: {action}, 400) return jsonify( {success: ok, message: ok if ok else step failed, action: action} ), 200 app.get(/v1/arm/status) def status() - tuple[Response, int]: return jsonify(arm.status()), 200 return appJava 类比对照本文件 SpringBoot 的Controller 控制器只负责 HTTP 网络相关鉴权、接收 JSON 请求、HTTP 响应、全局异常捕获。create_app()相当于写一个 Bean 工厂生产 Flask 实例支持外部注入arm实例MockArm / 真实驱动注意这份代码目前没有调用 ArmService现在它直接调用底层 MockArm 驱动属于半成品正常业务应该在这里调用ArmService.dispatch()而不是直接arm.xxx()。代码链路分析1. 全局常量与工具函数API_KEY os.environ.get(ARM_API_KEY, demo-key) def _err(code: str, message: str, http_status: int) - tuple[Response, int]: return jsonify( {code: code, message: message, retryable: False, details: {}} ), http_statusAPI_KEY接口密钥。环境变量优先没配置就默认demo‑key请求头必须携带这个 key不然拒绝访问。_err()统一错误返回工具所有错误输出固定 JSON 结构统一错误码、提示信息、HTTP 状态码避免各个接口返回格式乱七八糟。Java 类比全局工具封装统一返回 ResultVO。2. create_app (arm: ArmPort | None None) 工厂函数def create_app(arm: ArmPort | None None) - Flask: app Flask(__name__) arm arm if arm is not None else MockArm() ... return app工厂模式创建 Flask 应用对象。可以外部传入已经实例化好的MockArm不传就内部自动新建一个MockArm()模拟器。好处单元测试的时候可以自己构造带故障注入的 MockArm 传给 app非常方便测试故障场景。 Java 理解相当于 Bean对外提供实例支持外部传入依赖。3. before_request 请求拦截器鉴权app.before_request def require_api_key() - tuple[Response, int] | None: if request.headers.get(X-API-Key) ! API_KEY: return _err(UNAUTHORIZED, 无效或缺失 API Key, 401) return NoneJava 类比Filter 拦截器每个 HTTP 请求进来先走这里。 大白话所有接口必须在请求头带X‑API‑Key密钥密钥不对直接返回 401 未授权拒绝访问。4. after_request 后置处理器追加 Request‑IDapp.after_request def add_request_id(resp: Response) - Response: resp.headers[X-Request-ID] request.headers.get(X-Request-ID) or uuid.uuid4().hex return resp给每一次 HTTP 响应打上X‑Request‑ID请求唯一编号。客户端可以自己传不传服务端自动生成 uuid。作用日志排查链路追踪一次请求从头到尾用同一个 ID 串起来对应你明日计划的日志中间件。5. app.errorhandler (Exception) 全局异常捕获app.errorhandler(Exception) def handle_unexpected(_exc: Exception) - tuple[Response, int]: return _err(INTERNAL_ERROR, 系统内部错误, 500)Java 类比RestControllerAdvice全局异常处理器。 大白话代码哪里抛出未捕获的异常不会直接把堆栈抛给 http 客户端统一返回 500只提示 “系统内部错误”保护内部代码信息不泄露。6. POST/v1/arm/actions动作执行接口核心接口app.post(/v1/arm/actions) def actions() - tuple[Response, int]: data request.get_json(silentTrue) if not isinstance(data, dict): return _err(INVALID_TASK, 请求体必须是 JSON 对象, 400) action data.get(action) if action gripper_open: ok arm.gripper(open) elif action gripper_close: ok arm.gripper(close) elif action move_to_pose: ok arm.move_to_pose( float(data.get(x, 0.0)), float(data.get(y, 0.0)), float(data.get(z, 0.0)), float(data.get(speed, 0.3)), ) elif action estop: arm.estop() ok True else: return _err(INVALID_TASK, f未知动作: {action}, 400) return jsonify( {success: ok, message: ok if ok else step failed, action: action} ), 200接收 JSON拿到action动作字符串直接调用底层arm驱动执行动作。现存重大问题当前直接调用 MockArm完全跳过 ArmService现在没有_busy忙互斥保护多个 HTTP 请求并发打过来可以同时下发移动、夹爪指令。 正确写法应该python运行ok_dict arm_service.dispatch(action, data) return jsonify(ok_dict),2007. GET/v1/arm/status查询状态接口app.get(/v1/arm/status) def status() - tuple[Response, int]: return jsonify(arm.status()), 200HTTP 接口查询机械臂当前位置、状态、夹爪状态。request_id INTERNAL_ERROR不泄露堆栈在 Controller‑ArmService‑ArmPort 三层架构里的作用先回顾三层分工ControllerFlask网络入口接收 HTTP对外暴露接口ArmService业务层互斥锁、动作分发、业务逻辑ArmPort/MockArm驱动层硬件 / 模拟器最底层容易出异常Java 类比 request_id ≈ TraceId链路追踪 ID INTERNAL_ERROR 不抛堆栈 ≈ RestControllerAdvice 全局异常屏蔽堆栈安全防护1、X‑Request‑ID 请求 ID在哪生效Controller 的after_request每个 HTTP 响应头带上X‑Request‑ID客户端可传入不传服务端自动生成 UUID。在三层每一层的作用1.ControllerFlask请求进来生成 / 拿到 Request‑ID放在 HTTP 响应头返回给调用方Agent / 测试脚本。Agent 调用接口出错时Agent 拿到这个 ID就可以说“本次报错是 X‑Request‑IDxxx去日志查这条记录”。2.ArmService业务层Service 本身不知道 HTTP所以要把 request_id 作为参数往下透传。打日志时带上这个 ID打印动作名、入参、忙锁状态、返回码。如果发生故障注入、TASK_BUSY忙、执行失败日志全部带上同一个 request_id。3.ArmPort / MockArm驱动层模拟器或者真实硬件抛出异常、触发故障注入日志同样打印这个 request_id。完整链路追踪效果Agent发起HTTP请求 X‑Request‑IDabc123 ↓ Controller记录日志[abc123]进入接口 ↓ ArmService.dispatch(..., request_idabc123)记录忙状态、动作日志[abc123] ↓ MockArm移动/故障注入打印日志[abc123] ↓ 异常向上抛出 → Controller捕获返回报文Response头带回 X‑Request‑IDabc123问题现象Agent 收到报错step failed排查直接拿abc123搜全部日志一次性看到HTTP 入参 → service 忙锁判断 → 底层驱动故障整条链路。如果没有 request_id 会发生什么多请求并发混打日志全部搅在一起。Agent 报一个错误你分不清是哪一次调用导致的很难复现 bug。2、INTERNAL_ERROR不对外泄露堆栈全局异常捕获app.errorhandler(Exception) def handle_unexpected(_exc: Exception): return _err(INTERNAL_ERROR, 系统内部错误,500)作用分层解读1.Controller 层职责对外边界、安全隔离底层Service、MockArm不管炸出什么异常参数错误、空值、除零、硬件驱动抛出异常。Controller 全部接住不把 Python traceback 堆栈返回给 HTTP 客户端Agent。❌ 不好直接把 Python 报错堆栈返回 HTTP。堆栈会泄露文件路径、变量名、代码结构如果对外服务属于安全漏洞而且 LLM Agent 拿到一堆 Python 栈信息完全不知道怎么处理。✅ 好统一返回{code:INTERNAL_ERROR,message:系统内部错误}给 Agent 看只知道 “内部出错本次失败”不会拿到源码细节。服务器本地日志依然完整保存堆栈只是不给 http 返回。2.ArmService 层Service 内部发生异常比如参数转换异常不需要自己做 HTTP 返回异常直接向上抛给 Controller。 Service 只管业务不关心 HTTP不处理 http 错误码符合分层原则业务层和网络层解耦。Service 只返回业务字典{success,code,message}真正 500/400 HTTP 错误码交给 Controller。3.ArmPort / MockArm 驱动层驱动只管模拟硬件行为可以随便抛异常。它完全不知道 HTTP 存在。 驱动只管要么返回 True/False要么抛出异常异常向上冒泡由最外层 Controller 兜底捕获。两种错误要分清非常关键1.业务预期内错误机械臂忙TASK_BUSY、动作执行失败step failed、未知 action。属于业务逻辑在 ArmService 内部消化返回success:falseHTTP 200 正常返回。这是业务失败不是系统崩溃。2.非预期系统异常代码 bug、参数类型错误、驱动抛出异常。属于真正系统故障冒泡到 Controller 全局异常处理器返回INTERNAL_ERROR 500。通俗一句话机械臂干活失败故障注入、忙 业务失败http 200我们代码本身炸了 INTERNAL_ERRORhttp500不把堆栈往外吐。如果没有全局异常捕获会发生什么MockArm 或者 Service 抛异常Flask 默认会把完整 Python 堆栈返回 HTTP。安全问题泄露代码信息Agent 拿到一大段报错栈无法解析不知道该重试还是放弃任务。结合三层架构总结表格组件X‑Request‑ID 角色INTERNAL_ERROR (屏蔽堆栈) 角色Controller(Flask)生成 / 接收 request‑id放在响应头把 id 向下传给 Service对外统一输出全局异常捕获边界唯一处理 HTTP 错误码的地方屏蔽堆栈对外返回统一错误 JSONArmService接收 request‑id写业务日志忙锁、动作分发结果不碰 HTTP不捕获未知异常业务异常包装成返回字典未知异常向上抛给 Controller业务层不处理 HTTPArmPort / MockArm接收 request‑id打印驱动、故障注入日志完全不知道 http 存在只管硬件逻辑正常返回 bool 或者抛出异常异常向上冒泡自己不做错误响应demo_layer_test.py 换 driver 不启 Flask demo_busy_arm.py ARM_BUSYdemo_fault_injection.py 测试验证demo_layer_test.py换 driver 不启 FlaskMockArm 与 FakeArm 共用同一 ArmServiceD2换 driver 不启 Flask —— 同一 ArmService 逻辑跑两个 driver。 演示MockArm 与 FakeArm 实现同一 ArmPortArmService 完全复用 证明 route/service 不依赖具体设备驱动为真机/仿真切换打基础。 from __future__ import annotations import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent / src)) from arm_agent.backend.mock_arm import MockArm from arm_agent.service.arm_service import ArmService class FakeArm: 另一个 driver 实现仅演示分层不涉及 Flask。 def __init__(self) - None: self.gripper_open True def status(self) - dict: return {state: idle, position: [0.0, 0.0, 0.0], gripper_open: self.gripper_open} def move_to_pose(self, x: float, y: float, z: float, speed: float 0.3) - bool: return True def gripper(self, action: str) - bool: if action open: self.gripper_open True elif action close: self.gripper_open False else: return False return True def estop(self) - None: pass def main() - int: seq [ (gripper_close, {}), (move_to_pose, {x: 0.1, y: 0.0, z: 0.03}), (gripper_open, {}), ] services [(MockArm, ArmService(MockArm())), (FakeArm, ArmService(FakeArm()))] print(f{driver:8s} | {action:14s} | result) for name, svc in services: for action, payload in seq: r svc.dispatch(action, payload) print(f{name:8s} | {action:14s} | success{r[success]} message{r[message]}) print(\nPASS: 同一 ArmService换 driver 不改代码未启动 Flask) return 0 if __name__ __main__: raise SystemExit(main())验证结果demo_busy_arm.pyARM_BUSY 可复现测试D2ARM_BUSY —— 机械臂执行期间重复下发返回忙可重试。 演示begin_busy(2s) → 立即下发 move_to_pose → ARM_BUSY2.5s 后再下发 → 成功。 from __future__ import annotations import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent / src)) from arm_agent.backend.mock_arm import MockArm from arm_agent.service.arm_service import ArmService def main() - int: svc ArmService(MockArm()) svc.begin_busy(2.0) # 模拟机械臂正在执行 r1 svc.dispatch(move_to_pose, {x: 0.1, y: 0.0, z: 0.03}) print(忙期间下发:, r1) time.sleep(2.5) r2 svc.dispatch(move_to_pose, {x: 0.1, y: 0.0, z: 0.03}) print(空闲后下发:, r2) assert r1.get(code) ARM_BUSY and not r1.get(success) assert r2.get(success) is True print(PASS: ARM_BUSY 可复现空闲后自动恢复) return 0 if __name__ __main__: raise SystemExit(main())验证结果demo_fault_injection.py失败注入演示 第二次失败-- 恢复D2可重复失败注入 —— 让第 N 次失败可复现失败后自动复位。 演示注入 gripper_close 第 2 次失败 → 第1次成功、第2次 STEP_FAILED、第3次恢复。 from __future__ import annotations import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent / src)) from arm_agent.backend.mock_arm import MockArm from arm_agent.service.arm_service import ArmService def main() - int: svc ArmService(MockArm()) svc.inject(gripper_close, 2) for i in (1, 2, 3): r svc.dispatch(gripper_close, {}) print(f第{i}次 gripper_close:, r) assert r.get(success) is True print(PASS: 失败注入可复现第2次失败且第3次自动恢复) return 0 if __name__ __main__: raise SystemExit(main())验证结果