“养虾记”这个名字，第一眼看上去挺有迷惑性，不知道的人以为是个农业养殖项目。实际上，它是我手头一个内部系统的代号——一个跑在基础自动化组件和智能辅助工具之间的轻量中间件。我之所以把它叫“养虾”，是因为调试这个系统的时候，感觉就像在水池边养一缸虾：水质、温度、饵料、溶氧，哪个环节没伺候好，整缸虾就给你颜色看。同理，这个中间件只要有一个环节配置不到位，整个任务链路就会静默失败，你还找不到原因。
先把这个项目说清楚：底层那边，我们叫它 openclaw，是一个自建的开源系自动化执行组件，负责真实环境里的脚本执行、服务编排、插件调度；上层那边，挂着 copilot 这一类的智能辅助体，负责理解用户的自然语言请求、拆解成可执行的任务计划。而我写的中间件，就是悬在两者之间的那一层：接收上层来的任务意图，把意图翻译成下层能识别的指令，再把下层的执行结果回传、归档、通知上层。它不生产命令，也不理解语义，它只负责一件事——让两边高效协作且互不拖累。
你要是现在也在做 Agent 化改造、运维自动化或者智能工单系统，多半会遇到同样的尴尬：上层大模型很聪明，但要真让它直接操作生产环境，风险谁也兜不住；下层工具很稳定，但接口又臭又硬，和对话式交互完全不在一个频道。中间加一层翻译加调度加保险丝的中间件，是目前最务实的解法。这篇文章就把我这个“养虾记”项目从架构思路到实际落地的过程完整拆开来讲，包括我踩过的坑。
1. 整体设计与思路拆解：为什么要有一层中间件
1.1 三个角色各司其职
很多人一听到“中间件”就容易往消息队列、API 网关上联想。我这个中间件定位不太一样，它做的事情更像是一个“任务调度中枢 + 协议转换器 + 审计闸门”。
先定义好三个角色的边界：
- openclaw（底层执行端）：它不需要理解业务，只需要忠实地执行指令。比如跑一段脚本、拉起一个容器、调用某个内部服务的接口。它的特点是指令明确、结果确定、但接口比较原始。
- copilot（上层智能端）：它负责把人类模糊的语言变成清晰的任务列表。比如用户说“帮我查一下线上服务最近有没有异常日志”，它要能拆出：查哪个服务、过滤什么关键字、时间窗口多大、结果怎么展示。
- 中间件（我写的这层）：它负责承接 copilot 形成的任务计划，做合法性校验、去重、拆分、排队、调度，然后按 openclaw 能接受的格式下发；执行过程中的状态变化、日志、结果，也统一由它收集和存储。
界面划分清楚之后，一个很直接的收益就是：两边都可以独立迭代。openclaw 升级接口不用通知 copilot 改提示词；copilot 换模型供应商也不影响底层任务执行。中间件成了唯一需要关心两边协议细节的地方。
1.2 为什么不能直接集成
一开始有人建议：既然 openclaw 和 copilot 都支持 HTTP 调用，为什么不直接在 copilot 的工具调用配置里写 openclaw 的地址，省掉中间层？
我试过，而且很快就放弃了。直接集成有几个很现实的问题：
第一，会话上下文不在同一条链路上。copilot 一次对话可能包含多轮意图，每个意图又可能对应多个底层操作。如果直接调用，每个底层操作都是无状态的一次性请求，高层拿不到全局的任务关联关系。出了问题你想追溯“用户这句话导致了哪几个底层操作”，完全没有线索。
第二，权限边界模糊。copilot 如果直连 openclaw，意味着智能体天生拥有底层执行组的全部能力。它生成的指令到底合不合规、有没有越权，完全没人把关。这在生产环境是不可接受的。
第三，失败补偿逻辑无处安放。底层执行挂掉了，是重试还是放弃？是升级给人工处理还是自动换个方案？这些策略如果写在智能体的提示词里，稳定性和可调试性都差得离谱。放在中间件里，就是一段普通代码，想怎么改怎么改。
中间的隔离层就像给智能体装了一个需要“安检过关”的通道：意图随便提，但真正能落到执行层的指令，必须经过规则校验、资源配额检查和格式转换。这不是效率上的浪费，是风险控制上的必要冗余。
1.3 技术选型考虑
这个中间件的选型约束很简单：不能重，但必须稳定。我最终确定了几个选型原则：
- 通信方式：采用“HTTP + 事件回调”双通道。健康检查、指令下发走 HTTP；任务完成的状态变更走事件回调，避免长连接占用太多资源。
- 状态存储：用关系型数据库保存任务主数据，用 Redis 做分布式锁和实时状态缓存。为什么不只用 Redis？因为任务记录要长期查询、做审计，关系型数据库更可靠。
- 任务队列：初期直接用了数据库表加轮询，没上专门的消息队列。原因是任务量远没到秒级万级的程度，一个状态字段加索引足够支撑。后面如果量大了再抽象队列层也来得及。
- 部署形态：独立进程，容器化部署，配置全部走环境变量。不选框架，就是一个标准的后端服务，方便任何熟悉 Web 开发的同事接手。
有一次我和同事聊到这个设计，他说你这就是个带审批功能的数据库壳子。我说对，中间件很多时候不是技术难点，而是治理难点，把边界搞清楚，比用什么牛逼框架更重要。
2. 核心细节解析与实操要点
2.1 任务描述与状态机设计
整个中间件的核心数据模型就是“任务”。我设计了完整的任务状态机，每个状态转换都有明确触发条件。
任务生命周期的关键状态：
- PENDING（已接收）：copilot 提交的计划已落库，但还没有通过前置校验。
- VALIDATING（校验中）：中间件在检查参数格式、资源配额、权限白名单。
- QUEUED（排队中）：校验通过，等待调度器分配执行窗口。
- DISPATCHED（已下发）：指令已成功发给 openclaw，等待回执。
- RUNNING（执行中）：openclaw 回报开始执行。
- INTERRUPTED（需人工干预）：执行中遇到预期外错误，需要处理。
- COMPLETED（已完成）：结果正常返回。
- FAILED（失败）：经过重试仍然失败，任务终结。
这里最容易踩坑的是超时处理。执行中的任务可能因为 openclaw 那边的脚本卡死而长时间不返回。我在中间件里给每个任务预设了超时阈值，超过阈值且没有心跳更新，调度器就自动把任务置为 INTERRUPTED，同时通知 copilot 侧“任务卡死，需要人工确认”。这个机制救过我很多次，没有超时强杀，下层的僵尸任务会越积越多，最后把整个执行队列堵死。
2.2 上下文持久化：中间件最重要的能力
中间件存在的最大价值，不是转发消息，而是把散落的交互记录攒成一条完整的链路。我用一张上下文表来保存每次交互的父子关系和执行轨迹。
任务上下文的关键字段：
| 字段 | 含义 |
|---|---|
| trace_id | 一次用户对话的全局唯一标识 |
| task_id | 单个任务的唯一标识 |
| parent_task_id | 父任务标识，用于任务拆分后的挂载 |
| source | 来源标识，比如 copilot 还是人工后台 |
| user_context | 用户原始请求和附加信息 |
| payload | 中间件标准化后的执行指令 |
| status | 当前任务状态 |
| result_summary | 结果摘要，写给 copilot 侧看的短文本 |
| raw_result | 底层返回的完整原始数据 |
| timestamps | 各阶段时间戳，用于耗时分析 |
为什么必须做父子任务的层级关系？因为一个用户请求往往不是一次底层操作就能解决的。比如用户说“检查所有网关节点的磁盘空间，如果超过 80% 就重启对应服务”，这句话会拆成“检查所有节点”和“重启超阈值节点”两个阶段。每个节点又是一次独立执行。没有父子层级，你就无法回答“这个重启动作是哪个用户请求触发的”这种审计问题。
多轮对话的场景则要更细：用户先创建了一个任务，再追问“刚才那个任务结果怎么样”，这时候需要按 trace_id 去关联历史上下文，把上一轮的任务结果摘要拼进新的回复。没有上下文持久化，这种多轮交互就做不了。
2.3 配置管理与参数传递
中间件的配置管理我坚持一套原则：所有可变参数全部外置，代码里不写死任何环境相关的常量。配置文件用 YAML 做基础结构，敏感信息通过环境变量注入。
核心配置模块划分：
- executor 配置：openclaw 的地址、认证方式、最大并发数、超时时间。
- adapter 配置：适配器启停开关，比如哪些类型的任务允许下发，哪些只做记录不执行。
- scheduler 配置：队列策略、重试策略、死信策略。
- observer 配置：日志级别、指标采集开关、审计策略。
参数传递层面，我特别注意幂等性。每个任务下发时都会带一个由中间件生成的唯一幂等键（幂等键的生成规则是 trace_id + task_id + 动作序号）， openclaw 收到相同幂等键的请求时会直接返回上一次的执行结果，而不是重复执行。这个设计在做失败重试的时候至关重要：没有幂等键，一次网络超时后的重试就可能让底层脚本跑两遍，破坏性极大。
提示：如果你也在做类似系统，幂等键务必由中间件生成并持久化，千万不要依赖上层传的 request_id。上层可能生成重复值，也可能在重试的时候改变值，这样底层就无法去重了。
3. 实操过程与核心环节实现
3.1 环境准备与工程结构
技术栈用的是 Python + FastAPI + SQLAlchemy + Redis，选 Python 是因为团队里后端同事都能上手，FastAPI 做对外接口和异步任务都很顺手。
工程结构大致如下：
middleware/ ├── api/ # HTTP 接口层 │ ├── routes_executor.py # 对接底层执行端的回调 │ ├── routes_copilot.py # 对接上层智能体的提交入口 │ └── routes_admin.py # 人工管理后台接口 ├── core/ │ ├── dispatcher.py # 调度器：任务校验、排队、下发 │ ├── adapter.py # 适配器：不同任务类型转不同指令格式 │ ├── state_machine.py # 状态机：状态流转与超时强拆 │ └── validator.py # 校验器：参数白名单、资源配额检查 ├── repositories/ # 数据库访问层 ├── schemas/ # Pydantic 数据模型 ├── config/ # 配置文件与环境变量加载 └── main.py # 启动入口环境依赖很简单：Python 3.10+、PostgreSQL、Redis。开发环境我直接用 Docker Compose 起一套，生产环境拆开独立部署。
我当时的启动顺序是：先起数据库和 Redis，再起中间件主服务，最后配置 openclaw 的事件回调地址指向中间件的/callback/executor路径。如果顺序反了，openclaw 那边回调会有一堆连接拒绝的错误。
3.2 实现任务提交与校验
主要入口是/api/copilot/task，接收上层智能体提交的任务计划。请求体包含 trace_id、任务描述、任务类型、参数列表和期望的优先级。
校验器会做几件事：
- 检查 trace_id 是否合法且未过期。
- 检查任务类型是否在白名单内。
- 检查参数列表是否包含必填字段，以及字段值是否符合枚举和范围要求。
- 检查该来源的配额是否耗尽。
校验不通过的任务直接返回不可执行的错误信息，并且带上了原因说明，方便上层直接转给用户看。这块的提示语设计也踩过坑：错误信息如果太技术化，用户会看得莫名其妙；所以我统一维护了一套错误码表，每个错误码对应一句面向用户的话语，同时详细的技术原因单独存日志。
@app.post("/api/copilot/task") async def create_task(req: TaskCreateRequest): # 校验来源和配额 source = get_source_config(req.source) if not source.enabled: raise BizException(code="SOURCE_DISABLED", detail="该来源暂不可用") if not quota_check(req.source, req.estimated_cost): raise BizException(code="QUOTA_EXCEEDED", detail="配额不足") # 校验参数 validator = get_validator(req.task_type) issues = validator.validate(req.params) if issues: raise BizException(code="PARAM_INVALID", detail="; ".join(issues)) # 生成任务记录 task = create_task_record(req) dispatch_or_queue(task) return {"task_id": task.id, "status": task.status}这里有个细节：新增任务提交入口时，不要把任务执行的整个过程都放在请求等待里。就是让接口快速返回 task_id，后续通过状态查询或回调的方式把结果通知给上层。如果同步等执行结果，用户那边一个简单的查询请求可能要挂几十秒，体验很糟糕。我们在开发中把接口响应时间控制在 50ms 以内，耗时动作全部异步化。
3.3 与底层执行端对接：回调协议与指令下发的细节
openclaw 侧的执行指令有两种：一种是同步指令，比如查询类的操作；一种是异步指令，比如批量重启类的操作。中间件在适配器层对这两类做了不同处理。
同步指令的实现比较简单：中间件发起 HTTP 请求，openclaw 同一个响应里返回结果。但必须设置客户端超时，我们默认是 10 秒，超过就按失败处理。超时时间不要一刀切，要根据具体操作类型分别配置。比如简单的文件查询 5 秒足够，但一次全网节点巡检可能就要 60 秒以上。我把超时配置做进了任务类型定义里，每个任务类型带自己的超时阈值。
异步指令则依赖回调。openclaw 在任务完成时向中间件注册的回调地址发一个事件。
回调请求体包含：
- 幂等键，用于关联中间件下发的原始指令。
- 执行状态，是成功还是失败，还是部分成功。
- 结果数据，可以是文本摘要、结构化 JSON 或两者兼有。
- 耗时信息，方便统计执行效率。
中间件收到回调后，先根据幂等键找到对应的任务记录，然后更新状态，存储结果，并触发后续流程。如果找不到对应任务，说明回调异常或者记录被清理过，这时候会记录一条告警日志，而不是直接丢弃。
@app.post("/callback/executor") async def executor_callback(cb: ExecutorCallback): task = task_repo.get_by_idempotent_key(cb.idempotent_key) if task is None: logger.warning("orphan callback received: %s", cb.idempotent_key) return {"ok": False, "reason": "task_not_found"} if task.status == "RUNNING": task.status = "COMPLETED" if cb.success else "FAILED" task.raw_result = cb.payload task.result_summary = summarize_result(cb.payload) task_repo.save(task) notify_upstream(task) else: logger.error("unexpected callback for task %s in status %s", task.id, task.status) return {"ok": True}回调接口必须是幂等的。openclaw 可能因为网络问题重复回调，如果中间件不处理重复回调，就会造成任务状态被重复推进甚至状态错乱。我在状态推进前先判断当前状态，只有允许的状态转移才会执行更新。
给 openclaw 下发异步任务时，还有一个细节：下发成功不代表执行成功。中间件把指令发给 openclaw 后，openclaw 返回“已接收”，但可能因为环境问题，脚本执行到一半挂了。所以光有下发成功还不够，我会在任务上挂一个“执行确认”的机制：异步任务下发后，如果超过设定时间没有收到执行开始的信号，中间件会主动去 openclaw 查询任务状态。这是防止回调丢失的兜底方案。
3.4 与上层智能体对接：状态查询与结果回传
copilot 侧和中间件的交互，基本就是两件事：提交任务、查询结果。
结果查询接口/api/copilot/task/{task_id}返回的内容必须是面向智能体友好处理的摘要。我单独做了一个摘要器，把底层返回的碎片化 JSON 整理成简洁的文本描述，比如：
- 底层返回：
{"nodes": [{"ip": "10.0.1.2", "disk_usage": 0.84}, ...]} - 摘要器输出：
共检查 12 个节点，其中 3 个节点磁盘使用率超过 80%，建议关注：10.0.1.2、10.0.1.7、10.0.1.11。
上层智能体拿到摘要，就可以直接组织语言回复给用户，不需要再解析原始 JSON。这也算是中间件帮智能体做了一层“认知减负”，让它不用处理底层数据结构的细节。
多轮追问的场景，我额外提供了一个明细查询接口：如果用户想深挖某个异常节点的具体数据，智能体可以带上节点的标识调用一手数据接口，拿到完整原始结果。这样既有简洁的概览，又有深入的路径。
4. 常见问题与排查技巧实录
这个系统上线大半年，我积累了一份问题速查表。凡是做类似架构的人，很大概率也会撞上其中几个问题。
| 现象 | 可能原因 | 排查步骤 | 解决方案 |
|---|---|---|---|
| 任务一直停留在 QUEUED | 调度循环阻塞 | 检查 Redis 连接是否异常、队列消费线程是否退出 | 给队列入队增加超时告警，调度循环做健康自检 |
| openclaw 收到重复执行 | 中间件重试时幂等键生成不一致 | 检查重试请求里的幂等键是否和首次一致 | 幂等键必须在创建任务记录时就生成并落库，重试时从库里取，而不是重新生成 |
| 回调找不到对应任务 | 回调里带的幂等键格式变了 | 对比回调侧和中间件侧的幂等键日志 | 全链路统一幂等键模板，输出结构化日志方便 grep |
| 执行成功但上层没收到通知 | notify 环节异常被吞 | 查看 notify 通道是否有异常抛出、是否有重试机制 | 通知机制必须带重试和失败告警，不允许静默吞异常 |
| 数据库连接池耗尽 | 回调峰值时大量异步写入 | 查看数据库连接池监控，调整池大小 | 给数据库操作加统一超时，设置连接池上限和排队等待上限 |
| 底层执行报权限错误 | 中间件使用的服务账号权限配置不对 | 检查中间件调用 openclaw 时使用的身份信息 | 给中间件单独建最小权限账号，别复用管理员账号 |
第一个值得展开的问题：队列阻塞。我遇到过几次调度器线程假死的情况，现象就是任务全部停在 QUEUED 状态，新的请求还能进来，但没有任何任务被执行。排查到最后发现是 Redis 客户端长连接被服务端断开，旧连接没有自动重连导致队列消费彻底卡死。解决方式是加了一把看门狗：调度器每 10 秒向健康检查表写入一条心跳，外部探活系统发现心跳停止超过 30 秒就自动重启调度实例。这个机制保证了即使代码出现未捕获异常，系统也能自愈。
第二个值得展开的问题：回调风暴。测试期间 openclaw 的某个批量任务一次性向中间件发回了上千个回调，中间件的数据库连接池瞬间被打满，很多真正常规任务的查询反而被阻塞。后来我给回调接口单独规划了一个更小的连接池，并且做了请求排队，回调峰值超过阈值时直接丢弃最老的重复回调（因为回调是幂等的，丢弃旧的重发影响不大）。
日志排查的技巧：所有任务的日志必须包含 trace_id 和 task_id 两个字段。这个习惯强烈推荐。没有这两个字段，排查问题时只能靠时间猜，效率极低。我们项目从一开始就给日志模板统一加了这两个字段，每个动作打点都带上，后面任何链路追踪都很顺畅。
还有一个容易忽略的点：时区问题。任务的时间戳如果混用了 UTC 和本地时间，展示的时候会差 8 个小时，排查问题时极容易误判“任务是不是执行了很久”。我的做法是数据库统一存 UTC，接口输出统一转北京时间，任何中间展示层都不允许直接用服务器本地时间。
5. 实测心得与下一步玩法
这层中间件上线到现在，我手里跑过的任务量大约是几万次级别，整体成功率稳定在 99.6% 以上，剩余失败主要集中在底层依赖服务不稳定。中间件的优势在这里体现得最明显：底层执行挂掉时，copilot 侧的用户体验基本不受影响，因为中间件会拦截错误、重试、或者转人工，而不是直接把异常堆栈抛给用户。
我实操过程中印象最深的收获，是中间件无意中成为了全系统的“混沌实验场”。由于所有指令都要过中间件，我可以在中间件层面做拦截、修改、模拟故障，而不用真的去动底层环境。比如测试某个新任务类型时，我直接把 openclaw 的响应 mock 掉，中间件完全感知不到底层是假的，整个上层联调流程可以并行跑，不需要等底层环境就绪。
下一步有几个明确想做的方向：
- 多租户支持：目前是单团队使用，后续想让其他团队接入，必须在中间件层做资源隔离和配额隔离。
- 审批流接入：部分高风险操作需要人工审批后才能下发。准备把审批状态作为一个前置节点插进状态机，审批通过后自动继续流转。
- 可视化面板：给任务状态机和链路追踪做成实时面板，能直观看到任务卡在哪个环节。
- 插件化适配器：现在每接一个新的底层执行器类型都要改代码，后面计划做成配置化插件，让新类型接入不用发版。
最后分享一个小技巧：我给中间件加了一个“干跑模式”。开启后，所有指令只做校验和记录，不真正下发到 openclaw。这个模式在联调和演示的时候救了我好几次。有一次演示前有个同事改了一个底层参数的默认值，如果直接下发，会把生产配置覆盖掉。正是因为开了干跑模式，任务在最后一步被拦下，才避免了一次事故。所以强烈建议任何做这类中间层的同行都保留这个开关，平时开发默认全开干跑，需要真实执行时再显式关闭。
养虾养的是水和菌，养系统养的是边界和细节。中间件表面上是技术问题，实际上是把混沌的交互收敛成可控状态机的过程。这套架构不复杂，但把边界划清楚之后，上层智能体和底层执行组都可以各自安心迭代，不会再互相拉垮。希望这篇记录对同样在搭中间层的你有用。
