在机器人巡检项目里Grok 的任务是把“沿走廊走一圈并返回充电桩”拆成 12 个可执行步骤。第一次联调时前 4 步执行得很顺利第 5 步导航超时后执行器尝试重试但因为没有记录状态进程重启后整个计划又从第 1 步开始。第二次问题变成了对话上下文太长Grok 开始重复输出生成的 JSON 被截断计划卡在解析阶段。这类现象在真实项目中非常常见。Grok 负责规划机器人负责执行。要让“Grok 机器人计划”真正运行得更持久不能只依赖模型能力而是要在计划生成、状态保存、上下文控制、资源占用和故障恢复五个方向做工程化处理。下面 6 个技巧来自实际机器人项目调试经验适用于 ROS2、边缘设备、资源受限机器人以及使用 Grok 或 grok build 辅助构建的自动化任务。1. 先定位 Grok 机器人计划失效的三个层次在讨论解决方案之前先要清楚长时任务为什么会中断。Grok 机器人计划不是一段静态代码而是一条从“用户指令”到“机器人动作”的链路。每个环节都可能成为断点。1.1 一条完整的计划运行链路典型的运行链路如下用户输入高层指令例如“沿走廊巡检一圈并返回充电桩”。Grok 将指令拆成多个步骤输出包含导航点、动作类型、参数、失败策略的计划。执行器读取计划按顺序执行每个步骤。每一步执行后记录结果并更新状态。全部步骤完成后生成任务总结。在这个链路里Grok 只负责“生成计划”和“局部决策”真正的动作执行由导航模块、机械臂控制模块、传感器采集模块完成。这意味着计划一旦生成就不再应该依赖模型反复读取完整对话。1.2 失效点分布在模型层、执行层和资源层实际运行时故障点主要集中在三个层面失效层次典型现象对计划持久运行的影响模型层上下文变长后响应变慢、输出重复、JSON 被截断计划解析失败任务无法前进执行层单步超时、设备离线、步骤状态未持久化进程重启后从头执行大量工作浪费资源层内存不足导致进程被杀、日志占满磁盘、网络抖动执行器直接被终止计划中断很多人遇到问题后第一反应是“换一个更强模型”实际上模型层只是其中一个环节。如果状态没有持久化无论模型多强重启后任务还是会丢失。真正让计划运行持久不是保证不出错而是保证出错后能快速恢复并且恢复时不会重复执行已经成功的步骤。注意先接受“长任务一定会遇到异常”这个前提再围绕异常设计恢复机制比追求单次运行的稳定性更有效。2. 技巧一把计划从对话上下文中抽离持久化为结构化文件Grok 生成计划后最常见的错误是让执行器继续和模型对话每执行一步都要把历史记录重新发给模型。这样做的后果是上下文越来越长最终超出模型处理能力。正确的做法是让 Grok 只负责一次性生成计划然后把计划落成结构化文件执行器只读取文件不再依赖对话状态。2.1 计划文件设计目标、步骤、失败策略计划文件建议使用 JSON因为 JSON 结构清晰、易于校验也容易被 Python、C、ROS2 节点解析。一个最小可用的计划文件包含三部分plan_id计划唯一标识。goal任务目标用于恢复时确认上下文。steps步骤列表每个步骤包含类型、参数、超时、重试次数和失败策略。示例{ plan_id: plan_20250510_001, goal: 沿走廊巡检一圈并返回充电桩, created_at: 2025-05-10T10:00:00Z, max_failures: 2, on_failure: STOP, steps: [ { id: s1, type: navigation, params: { target: corridor_start }, timeout_sec: 30, retry: 1, on_failure: RETRY }, { id: s2, type: capture_image, params: { camera: front, save_path: /data/frames/s2.jpg }, timeout_sec: 10, retry: 0, on_failure: STOP } ] }这里的timeout_sec表示单步执行的最大时长retry表示瞬时错误的最大重试次数on_failure表示该步骤失败后的策略。每个字段都可以在执行器里被解释。2.2 用代码校验计划文件Grok 生成的 JSON 不一定规范可能包含 markdown 围栏、多余注释或字段缺失。加载计划时先做基础校验import json def load_plan(path): with open(path, r, encodingutf-8) as f: return json.load(f) def validate_plan(plan): required [plan_id, goal, steps] for key in required: if key not in plan: raise ValueError(fplan 缺少字段: {key}) step_ids set() for step in plan[steps]: if step[id] in step_ids: raise ValueError(fstep id 重复: {step[id]}) step_ids.add(step[id]) if timeout_sec not in step: step[timeout_sec] 30 return plan plan validate_plan(load_plan(plan.json)) print(plan[goal], len(plan[steps]))如果原始输出带有 markdown 围栏可以先移除再解析import re raw_text grok_output_text raw_text re.sub(r^(?:json)?|$, , raw_text.strip(), flagsre.MULTILINE) plan json.loads(raw_text)这在模型输出不稳定时尤其有用。不要直接把模型输出往json.loads里塞先做清理和校验能减少大量解析类异常。2.3 为什么执行器不应该每次调用模型当计划已经落盘执行器就可以在没有网络、没有模型服务的情况下继续运行。导航、机械臂控制、传感器采集都是确定性动作不需要模型参与。只有遇到“计划之外的局面”时才需要让 Grok 参与局部决策。这种“计划与执行分离”的设计还能带来一个额外好处计划文件本身可以追溯、审计、回放。哪怕模型版本变了旧计划的执行结果仍然可复现。3. 技巧二用 SQLite 状态机支持断点续跑计划文件只解决了“计划可保存”的问题还没有解决“执行到哪一步”的问题。如果执行器在第 7 步崩溃重启后必须知道第 7 步之前已经完成并直接从第 7 步或第 8 步继续。3.1 为每个步骤定义状态机建议使用以下状态PENDING等待执行。RUNNING正在执行。SUCCEEDED执行成功。FAILED执行失败且不再重试。SKIPPED由于依赖步骤失败而跳过。执行前先把所有步骤初始化为PENDING。执行时将当前步骤从PENDING改为RUNNING。执行完成后再改为SUCCEEDED或FAILED。这样进程重启后只需要查询第一个非终态的步骤即可。3.2 SQLite 状态表设计SQLite 非常适合机器人本地环境不用额外部署数据库服务单文件存储断电恢复能力也比较稳定。CREATE TABLE IF NOT EXISTS step_state ( id INTEGER PRIMARY KEY AUTOINCREMENT, plan_id TEXT NOT NULL, step_id TEXT NOT NULL, status TEXT NOT NULL DEFAULT PENDING, attempt INTEGER DEFAULT 0, error TEXT, output TEXT, updated_at TEXT, UNIQUE(plan_id, step_id) );核心字段字段含义plan_id计划唯一标识step_id步骤标识对应 JSON 中的 step.idstatus步骤状态attempt当前重试次数error最近一次失败的简要原因output步骤执行结果用于后续决策3.3 断点续跑的执行逻辑下面的代码演示了如何初始化状态、查找下一个待执行步骤、更新状态import sqlite3 from datetime import datetime def init_plan_state(conn, plan_id, steps): for step in steps: conn.execute( INSERT OR IGNORE INTO step_state (plan_id, step_id, status, updated_at) VALUES (?, ?, PENDING, ?), (plan_id, step[id], datetime.utcnow().isoformat()) ) conn.commit() def get_next_pending_step(conn, plan_id): row conn.execute( SELECT step_id FROM step_state WHERE plan_id? AND statusPENDING ORDER BY id LIMIT 1, (plan_id,) ).fetchone() return row[0] if row else None def mark_step(conn, plan_id, step_id, status, errorNone, outputNone): conn.execute( UPDATE step_state SET status?, error?, output?, updated_at? WHERE plan_id? AND step_id?, (status, error, output, datetime.utcnow().isoformat(), plan_id, step_id) ) conn.commit()主循环def run_plan(plan, conn): plan_id plan[plan_id] init_plan_state(conn, plan_id, plan[steps]) step_id get_next_pending_step(conn, plan_id) while step_id: step next(s for s in plan[steps] if s[id] step_id) mark_step(conn, plan_id, step_id, RUNNING) try: result execute_step(step) mark_step(conn, plan_id, step_id, SUCCEEDED, outputresult) except Exception as exc: mark_step(conn, plan_id, step_id, FAILED, errorstr(exc)) # 根据失败策略决定是否继续 step_id get_next_pending_step(conn, plan_id)这里的关键点是将“计划数据”和“执行状态”分开存储。JSON 是静态输入SQLite 是动态状态。只要两个文件都保留机器人可以在任意时间点恢复。注意不要把状态只保存在内存列表里。机器人进程可能被系统 OOM killer 杀掉也可能因为电池耗尽断电内存态在进程退出后全部丢失。4. 技巧三用摘要和局部重规划控制上下文爆炸当执行器需要 Grok 参与决策时比如某一步导航失败、需要重新规划路径必须控制发送给模型的上下文长度。4.1 上下文爆炸的典型现象把完整对话历史、传感器日志、错误堆栈全部发给模型时会出现以下现象响应速度明显变慢。模型开始重复已经输出的内容。输出 JSON 被截断关键字缺失。模型忽略后半段指令只回应开头内容。这些不是模型“变笨”而是上下文已经超过合理工作范围。解决思路是每次请求只发送“决策所需的最小信息集”。4.2 维护一份任务摘要不建议把整段历史对话传给模型。建议在执行器里动态生成一份任务摘要包含原始目标。当前成功的步骤列表。当前失败的步骤和错误原因。剩余步骤。本次需要模型解决的问题。示例结构任务目标: 沿走廊巡检一圈并返回充电桩 当前进度: s1 已到达廊道起点, s2 已采集第一段图像 最近错误: s3 导航超时 1 次 剩余步骤: s3, s4, s5 请基于以上信息只针对 s3 给出新的导航参数。这段文本通常只有几百字远小于完整对话历史。每次生成摘要后可以使用固定前缀提示模型只输出 JSON。4.3 失败时只做局部重规划某一步失败不要重新生成整个计划。局部重规划只针对失败步骤以及依赖它的后续步骤。比如s3导航失败模型需要重新决策的是s3的导航目标或路径不是让s1、s2重跑。局部重规划的好处减少上下文输入量。避免已成功步骤被模型误改。保留用户原始意图防止模型“发挥过度”。如果当前模型支持结构化输出建议在请求中要求返回固定的 JSON 字段例如{ step_id: s3, new_params: { target: corridor_mid, tolerance_m: 0.2 }, reason: 原目标被临时障碍物阻挡 }解析成功后执行器更新计划文件中的对应步骤并继续执行。如果解析失败就重试一次第二次仍未成功则进入人工告警。5. 技巧四超时、重试和幂等要按机器人执行特点设计机器人执行计划和普通后端服务有一个明显区别机器人动作有真实物理后果。导航指令一旦下发机器人可能正在移动重试前必须确认当前状态否则可能重复执行“前进”或“夹取”动作。5.1 区分瞬时错误和确定性错误重试策略不能对所有错误一视同仁。错误类型示例是否重试瞬时错误网络超时、服务端限流、设备短暂离线是确定性错误目标点不存在、参数非法、步骤类型不支持否状态冲突机器人正在执行运动无法接收新指令先等待再重试代码中可以通过异常类型区分class TransientError(Exception): pass class DeterministicError(Exception): pass只有TransientError才走重试逻辑。5.2 使用指数退避防止重试风暴重试不能写死间隔 1 秒。网络恢复瞬间大量步骤同时重试反而会把模型 API 或机器人控制接口打挂。推荐指数退避加抖动import random import time def sleep_with_backoff(attempt, base2.0, max_delay30.0): delay min(max_delay, base * (2 ** attempt)) time.sleep(delay random.uniform(0, 0.5))调用方式for attempt in range(max_retries): try: return execute_step(step) except TransientError: sleep_with_backoff(attempt)5.3 每个步骤都要有超时机器人执行步骤不能无限等待。建议在计划文件中为每个步骤配置timeout_sec执行器使用带超时的调用方式import asyncio async def execute_with_timeout(coro, timeout_sec): return await asyncio.wait_for(coro, timeouttimeout_sec)如果超时发生执行器将步骤标记为FAILED然后根据该步骤的on_failure策略决定是重试、跳过还是停止整个计划。不要设计成“没有超时”的步骤错误处理无法切入。5.4 执行操作要幂等导航、读取传感器天然可以重复执行但上传数据、发送告警、控制夹爪等操作不能因为重试而重复触发。常见的做法是给每个步骤分配一个request_id执行前写入数据库执行完更新结果。重试时先查这个request_id是否已经执行成功。def execute_step_with_idempotency(step, request_id, conn): row conn.execute( SELECT status FROM request_state WHERE request_id?, (request_id,) ).fetchone() if row and row[0] SUCCEEDED: return already_done result execute_step(step) conn.execute( INSERT OR REPLACE INTO request_state (request_id, status, output, updated_at) VALUES (?, SUCCEEDED, ?, ?), (request_id, result, datetime.utcnow().isoformat()) ) conn.commit() return result6. 技巧五资源受限机器人的降级和裁剪很多机器人平台算力有限比如树莓派、Jetson Nano甚至 MCU 加 Linux 的组合。长时任务对内存、CPU、磁盘的占用必须做预算。6.1 控制本地执行器的并发度Grok 模型调用放在远端本地执行器只负责调度和执行动作。即便如此也可能出现多个 asyncio 任务同时运行导致内存暴涨。建议机器人本地的步骤执行并发数控制在 1 到 2。每秒最多发起一次模型调用避免本地缓冲区堆积。禁止在每一步执行时都加载完整的历史日志到内存。6.2 日志按大小轮转避免磁盘写满磁盘写满会导致进程异常退出这在机器人上比后端服务器更容易被忽略。使用 Python 的RotatingFileHandler控制日志大小import logging from logging.handlers import RotatingFileHandler handler RotatingFileHandler( robot_executor.log, maxBytes2 * 1024 * 1024, backupCount3, encodingutf-8 ) logging.basicConfig( handlers[handler], levellogging.INFO, format%(asctime)s %(levelname)s %(message)s )这样单份日志超过 2MB 后自动轮转最多保留 3 个历史文件。机器人存储空间有限时这是最基本的保护。6.3 传感器数据分级降级长时巡检任务如果每秒保存一张完整图像很快会占满存储。可以采用分级采样正常状态1 帧/秒降低分辨率。检测到异常临时提高到 20 帧/秒保存完整分辨率。磁盘剩余空间低于阈值降低保存频率并告警。模型参与决策时也不要把原始图像直接发给 Grok而是先做降采样或提取关键帧降低传输和模型处理成本。6.4 远程规划与本地执行分离设计上建议让 Grok 完全跑在远端本地机器人不加载模型。网络断开时本地执行器继续跑已经下载到文件里的计划遇到需要模型决策的步骤时标记为“等待决策”网络恢复后再继续。这种分离能带来一个关键收益即使网络中断 5 分钟机器人也不会因为一次 API 调用失败而终止整个任务。7. 技巧六日志、监控和恢复机制要能兜底前五个技巧尽量降低失败概率但长时任务最终还是要靠监控和恢复兜底。7.1 每个关键节点都写结构化日志日志至少要包含event、plan_id、step_id、attempt和时间。这样查询日志时能快速定位执行到哪一步。logging.info({ event: step_started, plan_id: plan_id, step_id: step_id, attempt: attempt, })不要只写普通字符串日志因为无法通过字段快速过滤。建议统一使用 JSON 行日志格式方便后续接入日志平台。7.2 设置状态卡死告警执行器运行过程中状态表长时间没有变化可能是进程卡死或死锁。可以在独立线程里检查step_state的updated_atdef check_stuck(conn, plan_id, threshold_sec120): row conn.execute( SELECT step_id, updated_at FROM step_state WHERE plan_id? AND statusRUNNING, (plan_id,) ).fetchone() if row: updated_at datetime.fromisoformat(row[1]) if (datetime.utcnow() - updated_at).total_seconds() threshold_sec: send_alert(fstep {row[0]} stuck)告警方式可以是 MQTT、HTTP webhook 或机器人状态灯具体按项目条件选择。关键是告警要存在并且能唤起人工介入。7.3 恢复流程要简单直接恢复脚本越简单越好最好一条命令启动#!/bin/bash python executor.py --resume --plan plan.json --state state.dbexecutor.py接收--resume参数后先加载计划文件再连接 SQLite查询第一个非终态步骤从那里继续跑。如果步骤状态是RUNNING且没有新进程在执行恢复逻辑应该把这类“僵尸 RUNNING”状态重置为PENDING否则恢复后会一直卡在同一个步骤。def reset_stale_running(conn, plan_id): conn.execute( UPDATE step_state SET statusPENDING WHERE plan_id? AND statusRUNNING, (plan_id,) ) conn.commit()8. 常见问题排查整理一份问题排查表实际出问题时可以按表定位。问题现象常见原因检查方式处理建议计划跑到一半卡住执行器崩溃、步骤超时未生效、状态锁未释放查询step_state中RUNNING状态和updated_at将僵尸RUNNING重置为PENDING增加超时看门狗Grok 提示上下文超限把完整历史发送给模型检查模型请求中的 token 使用量改用摘要 剩余步骤不做全量重放重启后计划从第 1 步开始状态只保存在内存中检查是否创建了 SQLite 状态表初始化状态时使用INSERT OR IGNORE执行后立即落库重试风暴瞬时错误重试间隔固定且多步骤同时重试查看日志中同一秒发起的请求数使用指数退避和随机抖动重试前先确认目标状态模型输出 JSON 解析失败输出包含 markdown 围栏或字段缺失打印原始输出文本解析前清理围栏使用严格 JSON 模式失败后提示模型只输出 JSON日志文件占满磁盘没用日志轮转执行df -h查看磁盘占用使用RotatingFileHandler限制日志保留数量如果计划文件本身有问题也要先排查plan_id是否唯一。step.id是否重复。步骤类型是否被执行器支持。参数是否在合理范围内。timeout_sec是否配置过小。很多时候模型生成的计划本身没问题但执行器对未知步骤类型没有默认处理策略导致任务卡死或回退。建议在执行器里加一个unsupported_step分支遇到不认识的操作类型时明确记录并停止而不是静默跳过。9. 最佳实践一份可带走的工程清单把前面的技巧整理成一张落地清单适合在发布长时任务前逐项确认。9.1 机器人长时任务发布前检查清单计划文件生成后立刻落盘并经过 schema 校验。执行器启动时能从 SQLite 恢复未完成任务。每个步骤都有timeout_sec和明确的失败策略。重试只针对瞬时错误并使用指数退避。可能存在副作用的重试操作具备幂等设计。模型调用前生成上下文摘要不发送全量历史。单步失败时只做局部重规划不重生成整个计划。日志带plan_id和step_id且使用轮转文件。磁盘空间、内存占用、状态卡死都有告警。恢复脚本只依赖计划文件和状态库不需要重新连接模型。其中最容易遗漏的是“幂等设计”。因为大多数测试环境网络稳定不会触发重试但生产环境中网络抖动、进程重启几乎是必然发生。一旦重试出现在有副作用的动作上可能造成重复拍照、重复上传、重复发送控制指令。9.2 可以继续扩展的方向这套方案只是一个起点。更复杂的长时任务可以继续扩展用行为树替代简单 JSON 步骤列表表达更丰富的控制流。用 DAG 表达步骤依赖关系支持并行分支。将 SQLite 状态同步到远端调度系统实现多机协同。在边缘侧部署轻量模型做实时决策云端 Grok 只处理复杂规划。增加计划版本管理当模型升级后旧计划仍能按原有逻辑执行。真正让计划持久运行不是祈祷模型每次都能输出正确答案而是把计划变成可保存、可恢复、可观测、可降级的工程产物。先做状态持久化再做上下文控制最后补上资源兜底长时任务的稳定性就会明显提升。