从心跳脚本到AI CLI监督者：构建可靠AI系统的架构思维

1. 从心跳脚本到AI CLI工作者的监督者一个架构思维的转变很多关于“构建你自己的贾维斯”的故事往往从一个炫酷的演示开始。我的故事或者说我老板的故事起点却截然不同它始于一个心跳。最初的版本只是一个名为jarvis-heart.fsx的小脚本用F#写成。它的功能简单到近乎简陋检查贾维斯是否还活着如果挂了就启动它监视它是否卡住并在必要时轻轻推它一把。文件很小目标也很小但核心思想已经萌芽AI助手本身是需要被监督的对象。这个看似微小的决定最终被证明比选择哪个大模型重要得多。这背后反映的是一种架构思维的彻底转变。我们不是在构建一个无所不能的“超级大脑”而是在设计一个监督优先的系统。贾维斯不应该是一个包揽所有编码任务的英雄式工人它更应该是一个协调者。它醒来审视当前状态决定什么重要将任务委派给合适的“工人”检查这些工人是否健康并在事情偏离轨道时恢复连续性。这听起来可能不如“自主智能体”那么炫酷但它更接近现实世界的工作方式可靠性源于系统性的监督与恢复而非单个组件的绝对强大。如果你厌倦了那些一断线就丢失全部上下文的“玩具”或者对需要一个完美模型才能工作的“空中楼阁”感到失望那么这种监督优先的架构思路或许能为你构建真正可用的个人AI系统提供一条更坚实、更可信的路径。2. 核心架构从“智能体优先”到“监督者优先”2.1 为何“监督者优先”是更优解在AI工具开发的早期我们很容易陷入“智能体优先”的思维定式寻找或训练一个最强大的模型赋予它复杂的提示词和工具链期望它能理解并执行一切。这种模式的问题在于它将所有的智能、记忆和可靠性都押注在单个、通常是黑盒的进程上。一旦这个进程崩溃、失忆上下文窗口溢出、或者陷入逻辑循环整个系统就瘫痪了。你得到的不是一个助手而是一个需要你时刻照看的“巨婴”。“监督者优先”的架构则反其道而行之。它将系统的智能分为两层监督层Orchestrator负责系统的元认知。它不直接处理具体任务比如写一段代码而是管理任务的生命周期。它的核心职责包括状态感知系统现在在干嘛健康吗、任务调度接下来该谁干活、健康检查工人还活着吗、错误恢复工人卡住了怎么办以及连续性维护如何让工作在不同会话间无缝衔接。工作层Workers由一个个具体的、功能单一的“工人”组成。每个工人可以是一个AI驱动的CLI工具如基于Copilot Chat的命令行、Qwen-Coder、Gemini Code Assist也可以是一个传统的脚本或程序。它们职责明确在监督者的调度下执行具体任务。这种架构的优势是显而易见的。首先它极大地提升了系统的鲁棒性。一个工人的崩溃不会导致全盘皆输监督者可以重启它或换一个。其次它实现了能力的热插拔。你可以随时加入新的、更擅长某类任务的AI模型作为工人而无需重构整个系统。最后它让持续工作成为可能。监督者确保了即使系统休眠、重启或发生意外正在进行的工作和上下文状态也能被保存和恢复。2.2 监督系统的核心组件与职责一个完整的监督优先系统其核心组件远不止一个“心跳”。它应该是一个精密的控制平面包含以下几个关键部分心跳与健康检查Heartbeat Health Check这是系统的脉搏。它定期向所有注册的工人发送“ping”信号检查其进程是否存在、响应是否及时。如果某个工人超时或无响应监督者会将其标记为不健康并触发恢复流程。这里的技巧在于检查不仅要看进程是否存在还要看它是否在“有效工作”例如是否在合理时间内产生了新的输出而不是卡在某个循环里。会话与状态管理器Session State Manager这是系统的记忆中枢。它负责管理每个工作会话的完整生命周期和状态。一个“会话”代表一项正在进行的工作单元比如“修复项目X的登录BUG”。状态管理器需要将会话的完整上下文——包括对话历史、已执行的操作、生成的文件、当前目标等——以持久化的方式保存下来通常是写入到文件或数据库中。这样当会话因任何原因中断后监督者可以“重新水合”这个会话从断点处继续用户完全感知不到中断。任务队列与调度器Task Queue Scheduler监督者需要决定接下来做什么。它维护一个持久化的任务队列。当监督者被唤醒可以是定时也可以由事件触发它会检查队列评估当前系统状态和可用工人然后将最合适的任务分配给最合适的工人。调度逻辑可以很简单如FIFO也可以很复杂基于优先级、工人负载、任务类型匹配。技能与工人注册表Skill Worker Registry系统需要知道它拥有哪些能力。一个动态的注册表记录了所有可用的“技能”及其对应的“工人”实现。例如“代码生成”技能可能由claude-worker提供“代码审查”技能由gemini-worker提供。监督者通过查询这个注册表来了解它能做什么以及该让谁去做。这使得扩展系统变得异常简单只需编写一个新的工人脚本并将其注册到系统中即可。3. 实现细节将心跳扩展为健壮的系统3.1 持久化状态从内存到磁盘的飞跃早期原型最大的痛点就是“失忆”。所有任务列表、会话上下文都保存在内存里。关掉终端或者进程意外退出半天的工作就烟消云散。这种脆弱性对于旨在长期运行的助手系统是致命的。解决方案是将状态持久化到磁盘。这听起来平淡无奇但却是从“演示”走向“工具”的关键一步。具体实现上我们采用了文件夹备份的持久化状态。每个会话、每个任务目标、甚至每次重要的交互记录都被序列化为JSON或类似格式的文件存储在一个结构化的目录树中。work_state/ ├── sessions/ │ ├── session_20240415_fix_login_bug/ │ │ ├── meta.json # 会话元数据创建时间、最后活跃时间、关联目标ID │ │ ├── context.json # 完整的对话历史与上下文 │ │ └── artifacts/ # 会话中生成或修改的文件快照/链接 │ └── ... ├── goals/ │ ├── goal_001_implement_feature_x/ │ │ ├── description.md │ │ ├── status.json # 状态pending, in_progress, completed, failed │ │ └── related_sessions.json │ └── ... └── workers/ ├── claude_coder/ │ └── health_status.json └── ...这种设计带来了几个巨大优势可恢复性系统重启后监督者可以扫描sessions/和goals/目录重建中断前的工作现场。可审计性所有操作都有迹可循方便回溯和调试。状态共享不同的监督者实例例如在开发机和服务器上各运行一个可以通过共享存储如网络文件系统来同步状态实现简单的分布式协作。实操心得状态序列化的权衡在序列化状态时要特别注意对复杂对象如打开的文件句柄、网络连接的处理。我们通常只保存足以“重新水合”会话的元数据和引用而不是尝试保存整个运行时对象。例如保存当前正在编辑的文件路径和光标位置而不是文件内容在内存中的完整表示。恢复时再根据这些信息重新打开文件。3.2 会话管理将CLI视为可监督的工人这是整个系统中最精妙的一环。传统的CLI工具是“一锤子买卖”你运行它它输出结果然后退出。如何让一个CLI工具变成一个可以被监督、被交互、被持久化的“工人”关键在于一个名为oly的会话管理器来自Open Relay项目。oly的核心思想是将CLI会话如一个正在运行python解释器或nodeREPL的终端视为一等公民对其进行抽象和管理。它允许你启动一个附着到特定PTY伪终端的持久化CLI会话。向会话发送输入就像你在终端里打字一样。从会话读取输出包括标准输出和标准错误。检查会话是否存活。给会话打标签方便后续查找和关联。休眠和唤醒会话暂时挂起或恢复其执行。有了oly监督者与AI CLI工人的交互模式就发生了根本变化。监督者不再是通过一次性调用API来获取结果而是“雇佣”了一个长期的、有状态的工人。例如监督者可以为“代码生成”任务启动一个claude-worker会话。向该会话发送指令“在文件src/auth.py中实现一个OAuth2登录函数。”异步地、增量地读取工人的输出同时进行健康检查。如果工人卡住比如超过30秒没有新输出监督者可以发送一个“\n”或预设的唤醒词进行“轻推”。任务完成后监督者可以将会话标记为“空闲”或让其进入低功耗休眠状态以备下次使用同时保存完整的交互历史到会话状态中。这种模式使得Copilot、Qwen、Gemini等任何提供了命令行交互界面的AI工具都能无缝接入系统成为被监督的工人。贾维斯不需要它们内部实现一致它只需要一个统一的方式来“管理”它们。3.3 技能系统的插件化设计为了让系统易于扩展我们采用了插件化的技能系统。监督者本身并不硬编码任何具体的任务逻辑。相反它维护一个技能目录。每个技能插件通常包含技能描述一个元数据文件说明这个技能是什么如“代码重构”、需要什么输入、产生什么输出、推荐由哪个工人执行。工人适配器一段代码或配置告诉监督者如何与实现该技能的特定CLI工人进行交互例如发送给claude-worker的提示词模板。结果解析器可选用于从工人的原始输出中提取结构化信息。当用户提出一个请求或监督者需要推进一个目标时它会分析当前上下文和需求。查询技能注册表寻找匹配的技能。根据技能描述选择合适的工人考虑工人当前负载、对该技能的历史成功率等。通过对应的工人适配器将任务派发出去。收集结果并用解析器如果有处理然后更新任务状态和会话历史。例如当用户说“帮我优化这个函数的性能”监督者可能会匹配到“代码性能分析”和“代码重构”两个技能。它可能先调用“代码性能分析”技能由某个擅长分析的工人执行获取分析报告再将报告和原始代码作为输入调用“代码重构”技能由另一个擅长重写的工人执行。4. 构建你自己的监督系统实操指南4.1 技术栈选择与基础搭建你不需要从零开始。基于开源组件我们可以快速搭建一个监督系统的骨架。核心依赖会话管理Open Relay的oly。这是基石它提供了管理CLI会话的核心能力。你可以直接从其GitHub仓库获取并集成。监督框架可以选择用任何你熟悉的语言编写监督者逻辑。原文中使用F#因其函数式特性对状态管理和并发很友好。但Python利用asyncio、Go利用goroutine和channel或Node.js也都是绝佳选择社区生态更丰富。关键在于语言要能方便地处理异步I/O和进程管理。持久化对于起步使用本地文件系统如SQLite数据库或简单的JSON文件就足够了。后期可以考虑更强大的存储如Redis用于队列和缓存或PostgreSQL用于复杂状态关系。工人AI CLI工具选择你日常使用的AI编码工具确保它们有命令行接口或能通过API被脚本调用。例如GitHub Copilot CLI (gh copilot)通义千问Qwen的CodeQwen系列命令行工具Claude的Claude Desktop命令行工具任何支持--interactive或类似模式的LLM命令行工具。基础目录结构my_jarvis/ ├── supervisor/ # 监督者核心代码 │ ├── main.py # 主循环调度逻辑 │ ├── session_manager.py # 基于oly的会话管理封装 │ ├── state_manager.py # 状态持久化与加载 │ └── skill_registry.py # 技能插件的加载与管理 ├── workers/ # 各个AI工人适配器 │ ├── copilot_worker/ │ ├── qwen_worker/ │ └── ... ├── skills/ # 技能插件定义 │ ├── code_generation/ │ ├── code_review/ │ └── ... ├── data/ # 持久化数据 │ ├── sessions/ │ ├── goals/ │ └── queue.json └── config.yaml # 配置文件4.2 编写监督者主循环监督者的主循环是其大脑一个简化的Python示例如下import asyncio import time from session_manager import SessionManager from state_manager import StateManager from skill_registry import SkillRegistry class Supervisor: def __init__(self): self.session_mgr SessionManager() self.state_mgr StateManager() self.skill_registry SkillRegistry() self.heartbeat_interval 30 # 秒 self.wakeup_interval 60 # 秒 async def run(self): 主监督循环 print(Supervisor starting...) # 1. 恢复系统状态 await self.state_mgr.recover_state() # 2. 启动所有已注册工人的会话并检查健康状态 await self.session_mgr.restore_sessions() while True: cycle_start time.time() # 3. 执行心跳检查 await self._perform_heartbeat() # 4. 检查并处理任务队列 await self._process_task_queue() # 5. 检查并恢复停滞的会话 await self._recover_stalled_sessions() # 6. 持久化当前状态增量 await self.state_mgr.checkpoint() # 控制循环频率 elapsed time.time() - cycle_start sleep_time max(1, self.wakeup_interval - elapsed) await asyncio.sleep(sleep_time) async def _perform_heartbeat(self): 检查所有活跃工人的健康状态 for worker_id, session in self.session_mgr.active_sessions.items(): is_alive await session.is_alive() if not is_alive: print(fWorker {worker_id} is dead. Attempting restart...) await self.session_mgr.restart_worker(worker_id) else: # 进一步检查是否“假死”有进程但无输出 if await session.is_stalled(timeout120): print(fWorker {worker_id} seems stalled. Sending nudge...) await session.send_input(\n) # 发送一个换行符尝试唤醒 async def _process_task_queue(self): 从队列中取出任务匹配技能并委派 tasks self.state_mgr.get_pending_tasks(limit5) for task in tasks: # 为任务寻找匹配的技能 matched_skills self.skill_registry.match_skills(task.description) if not matched_skills: print(fNo skill matched for task: {task.id}) task.status failed continue # 选择最合适的技能和空闲工人 chosen_skill, chosen_worker await self._select_worker_for_skill(matched_skills) if chosen_worker: print(fDispatching task {task.id} to {chosen_worker} using skill {chosen_skill.name}) task.assigned_worker chosen_worker task.status in_progress # 实际委派执行异步 asyncio.create_task(self._execute_task(task, chosen_skill, chosen_worker)) else: print(fNo available worker for task: {task.id}) async def _recover_stalled_sessions(self): 恢复因各种原因中断的会话连续性 # 检查是否有“in_progress”状态的任务对应的会话已丢失 # 如果有尝试重新水合会话或重新派发任务 pass async def _execute_task(self, task, skill, worker_session): 实际执行任务并更新状态 try: # 使用技能适配器与工人会话交互 result await skill.adapter.execute(worker_session, task.input_context) task.result result task.status completed print(fTask {task.id} completed successfully.) except Exception as e: print(fTask {task.id} failed with error: {e}) task.status failed task.error str(e) finally: # 无论成功失败都更新状态 await self.state_mgr.update_task(task)这个循环实现了监督的核心职责监控、调度、恢复。它周期性地运行确保系统始终朝着既定目标前进并能从意外中恢复。4.3 集成第一个AI工人以Copilot CLI为例让我们以GitHub Copilot CLI为例展示如何将其集成为一个被监督的工人。首先你需要一个能与Copilot CLI交互的适配器。假设我们通过子进程调用它# workers/copilot_adapter.py import asyncio import json from session_manager import OlySession # 假设我们有一个oly会话封装 class CopilotWorkerAdapter: def __init__(self, session: OlySession): self.session session async def execute(self, prompt: str, context_files: list None) - dict: 向Copilot会话发送一个编码任务。 # 1. 构建给Copilot的完整指令 full_instruction self._construct_prompt(prompt, context_files) # 2. 清除会话之前的输出可选 await self.session.clear_buffer() # 3. 发送指令 await self.session.send_input(full_instruction \n) # 输入指令并回车 # 4. 等待并收集响应 # 这里需要实现一个逻辑来判定Copilot何时输出完毕。 # 一个简单的方法是等待直到输出中包含特定结束标记或一段时间没有新输出。 output timeout 120 # 最大等待120秒 start_time asyncio.get_event_loop().time() while (asyncio.get_event_loop().time() - start_time) timeout: new_output await self.session.read_output(timeout5) if new_output: output new_output # 简单判断如果输出包含常见的代码块结束或自然语言结束模式 if in new_output and output.count() % 2 0: # 成对的代码块 # 再等待一小会儿看是否还有后续解释 await asyncio.sleep(2) final_chunk await self.session.read_output(timeout3) if final_chunk: output final_chunk break # 或者判断是否长时间没有新输出假设Copilot已说完 last_activity asyncio.get_event_loop().time() else: # 如果超过10秒没有新输出认为完成 if (asyncio.get_event_loop().time() - last_activity) 10: break await asyncio.sleep(0.5) # 5. 解析输出提取代码和解释 parsed_result self._parse_copilot_output(output) return { raw_output: output, generated_code: parsed_result.get(code), explanation: parsed_result.get(explanation), success: bool(parsed_result.get(code)) } def _construct_prompt(self, prompt, context_files): # 这里可以构建一个更精确的提示词包含上下文 context_str if context_files: # 简单起见假设我们只引用文件名 context_str f\n\nReference files: {, .join(context_files)} return fPlease generate code for the following task:{context_str}\n\nTask: {prompt}\n\nProvide the code in a markdown code block. def _parse_copilot_output(self, output): # 简单解析提取 lang ... 之间的代码 import re code_blocks re.findall(r(?:\w)?\n(.*?), output, re.DOTALL) code code_blocks[0] if code_blocks else # 解释部分可以认为是代码块之前或之后的文本 explanation output.replace(f\n{code}\n, ).strip() if code else output return {code: code.strip(), explanation: explanation}然后在技能注册表中注册这个工人# skills/code_generation/skill.yaml name: code_generation description: Generates code based on a natural language description. preferred_workers: [copilot_worker, qwen_coder_worker] input_schema: type: object properties: prompt: type: string description: The description of the code to generate. context_files: type: array items: {type: string} description: List of relevant file paths for context.最后监督者在匹配到“代码生成”任务时就会尝试寻找一个可用的copilot_worker会话并通过上述适配器与之交互。5. 避坑指南与进阶思考5.1 常见问题与排查技巧在构建和运行这样一个系统时你会遇到许多意料之外的问题。以下是一些典型问题及其解决思路问题1工人会话“假死”现象进程存在但长时间数分钟没有任何输出任务卡住。排查首先检查会话是否还有标准输入/输出。可以通过oly或类似工具发送一个简单的换行符或唤醒词如“继续”。如果仍无响应可能是工人内部崩溃或陷入死循环。解决监督者应设置一个“无输出超时”如90秒。超时后首先尝试“轻推”发送\n或CtrlC模拟。如果轻推无效则温和地终止该会话进程发送SIGTERM然后根据持久化状态重新启动一个新的会话并尝试从上一个可用的检查点恢复任务。问题2状态持久化与恢复不一致现象系统重启后恢复的会话状态与中断前不一致导致任务逻辑错误。排查检查状态序列化和反序列化的逻辑。确保所有必要的上下文信息如打开的文件、编辑位置、对话历史中的关键决策点都被正确保存。特别注意循环引用或无法序列化的对象。解决采用“事件溯源”或“状态快照增量日志”的模式。每次重要的状态变更都记录一个事件或打一个快照。恢复时从最近的完整快照开始重放之后的事件日志以确保状态重建的准确性。对于文件内容保存文件路径和版本哈希比保存完整内容更可靠。问题3技能匹配不准确或工人选择不当现象监督者将一个“代码调试”任务错误地派给了擅长“代码生成”的工人结果不理想。排查检查技能描述和任务描述的匹配算法。简单的关键词匹配很容易出错。解决引入更智能的匹配。可以使用嵌入模型如Sentence-BERT将技能描述和任务描述转换为向量计算余弦相似度。为每个工人维护一个历史成功率矩阵记录其对不同类型任务的成功率优先选择成功率高的工人。实现一个简单的反馈循环任务完成后让用户或自动校验流程对结果评分用这个评分来调整该工人对此类任务的权重。问题4资源竞争与死锁现象多个任务同时竞争同一个文件或同一个外部服务导致死锁或数据损坏。排查系统中有没有共享的、可变的资源如同一个代码文件被多个工人同时修改解决在监督层引入资源锁机制。当一个任务需要修改某个文件时先向状态管理器申请该文件的锁。如果锁被占用任务进入等待队列。任务完成后释放锁。对于数据库或API调用也要考虑速率限制和并发控制。5.2 从监督到协作系统的未来演进当你的监督系统稳定运行后可以考虑以下几个进阶方向让它从“自动”走向“智能协作”多工人协作流水线将一个复杂任务分解为多个子任务由不同的工人接力完成。例如“实现一个登录功能”可以分解为工人A分析设计API接口工人B生成编写后端代码工人C生成编写前端组件工人D审查进行代码审查。监督者负责协调整个流水线传递中间产物。反思与元提示工程监督者不仅可以派发任务还可以对工人的输出进行“反思”。例如在工人生成代码后监督者可以启动另一个“审查工人”对代码进行静态分析或单元测试如果发现问题则生成新的提示词让原工人进行修正。这形成了一个自我改进的循环。人机交互界面当前的监督者主要是在后台运行。可以为其增加一个轻量级的聊天界面如Telegram Bot、Slack App或本地Web UI允许用户以自然语言直接向监督者下达高级指令如“本周帮我优化项目X的性能”监督者将其分解为任务并在完成后汇总报告。用户也可以中途干预提供反馈指导后续任务方向。学习与适应系统可以记录每一次任务派发和结果形成一个经验库。通过分析历史数据监督者可以学习到“对于‘修复内存泄漏’这类任务工人B的成功率比工人A高20%”或者“在下午时段API的响应速度较慢需要设置更长的超时”。这些经验可以动态调整未来的调度策略。构建这样一个系统最大的收获可能不是得到了一个多么强大的AI助手而是深刻地理解到在现实世界中让智能体可靠工作的关键往往不在于算法本身的尖端而在于围绕它构建的、能够容错、恢复和持续运作的“系统韧性”。那个最初的心跳脚本象征的正是这种韧性——它不负责创造只负责守护创造的进程持续跳动。从这个角度看最好的AI架构或许就是那个让你几乎忘记AI存在只觉得工作流程自然而然变得顺畅的“隐形守护者”。