1. 项目概述：为什么要在macOS上构建一个“撞不死”的AI代理？

如果你在macOS上跑过一些需要长期运行的后台脚本或服务，比如数据抓取、模型微调、或者像我最近在折腾的AI智能体（Agent），那你大概率经历过这种糟心事：电脑一合盖休眠，脚本就停了；系统一更新重启，服务就挂了；甚至有时候就是网络波动一下，你的Agent就“失联”了，之前积累的状态全丢，一切得从头再来。这感觉就像养了只电子宠物，你稍微不留神，它就“饿死”了，非常影响效率和心情。

所以，这个项目的核心目标很明确： 在macOS上，打造一个具备高容错能力、能抵抗各种意外崩溃、并能实现自愈的AI Agent服务。 这里的“撞不死”（Crash-Tolerant）不是指它代码里没bug，而是指当外部环境发生意外（如系统重启、进程被误杀、网络中断）时，我们的服务能最大限度地保持状态，并在条件恢复后自动拉起，继续之前的任务，仿佛什么都没发生过。

为什么选择 launchd ？在macOS生态里， launchd 是系统级的服务管理守护进程，相当于Linux系统中的 systemd 。它由苹果官方维护，深度集成于系统，具备以下无可替代的优势：

1. 自启动与守护 ：可以配置服务在系统启动时、用户登录时、或定时自动启动。更重要的是，如果服务进程意外退出， launchd 能自动将其重新启动。
2. 资源与依赖管理 ：可以精细控制服务运行所需的环境变量、工作目录、文件描述符限制等。还能定义服务之间的依赖关系（虽然 launchd 的依赖管理相对简单）。
3. 按需启动 ：支持 socket 或 path 激活。例如，你的AI Agent是一个HTTP服务，可以配置成只有当有网络请求到达特定端口时， launchd 才启动你的服务进程，平时不占用资源，非常节能。
4. 与系统生命周期绑定 ：可以配置服务在特定系统事件（如网络状态变化、文件系统挂载）时启动或停止。

基于 launchd ，我们就能为AI Agent构建一个坚固的“托管外壳”。这个项目不仅适用于AI Agent，任何需要7x24小时稳定运行的Python脚本、Go服务、乃至Shell脚本，都可以套用这个框架。接下来，我会从设计思路、配置文件详解、到状态保持与自愈策略，一步步拆解如何实现这个“撞不死”的AI Agent。

2. 核心设计思路：将AI Agent包装成系统服务

构建一个健壮的服务，关键在于 解耦 与 状态管理 。我们不能让业务逻辑（AI的推理、决策）和生存逻辑（进程守护、崩溃恢复）纠缠在一起。

2.1 分层架构设计

我的设计通常分为三层：

1. Agent核心层 ：这是你的AI大脑，包含具体的任务逻辑，比如调用大语言模型API、处理工具调用、维护对话记忆或任务链。这一层只关心“做什么”。
2. 服务包装层 ：一个轻量的包装脚本（通常是Python）。它的职责是：
   • 以 daemon （守护进程）模式运行Agent核心。
   • 捕获并处理信号（如 SIGTERM , SIGINT ），实现优雅退出。
   • 将Agent的核心状态（如当前的对话ID、任务进度）定期持久化到磁盘（例如，每处理完一个任务单元就保存一次）。
   • 在启动时，尝试加载之前保存的状态，实现“断点续传”。
3. launchd 托管层 ：通过 .plist 配置文件，告诉系统如何管理我们的服务包装层。这是实现“撞不死”特性的基石。

注意 ：永远不要尝试直接将一个会交互式输出大量日志、或依赖特定终端环境的脚本丢给 launchd 。 launchd 运行的环境是“非交互式登录shell”，很多你终端里好用的配置（如 conda 环境、 PATH 变量）在这里都不存在。必须通过包装层明确地设置好一切。

2.2 状态持久化策略

“撞不死”的前提是“记忆不死”。Agent在运行中产生的状态必须落地。根据Agent的复杂程度，可以选择不同策略：

• 简单场景（单次任务独立） ：如果每次任务都是独立的，没有上下文关联，那么状态就是“当前任务是否完成”。可以在任务开始时写入一个锁文件（ lockfile ），任务完成后删除。服务重启后检查锁文件，如果存在，则意味着上次任务被中断，可以选择重新执行或清理。
• 中等场景（维护会话/任务链） ：这是AI Agent的典型场景。需要将对话历史、工具调用结果、任务步骤等序列化保存。推荐使用轻量级的本地数据库，如 SQLite 。它的单文件特性非常适合这种场景，无需额外服务。每次有状态更新，就原子化地写入数据库。
• 复杂场景（分布式/高频率） ：如果状态变更极快或需要跨节点同步，可以考虑 Redis （内存快照+持久化）或更专业的消息队列、状态服务器。但对于绝大多数跑在单台Mac上的个人或小型项目Agent，SQLite绰绰有余。

我个人的选择是SQLite，因为它零依赖，与Python集成极好，并且可靠性经过时间检验。我们将状态保存逻辑嵌入到服务包装层中。

3. 实战：创建launchd的.plist配置文件

这是整个项目的控制中心。我们将创建一个XML格式的 .plist 文件，通常放在 ~/Library/LaunchAgents/ （用户级服务）或 /Library/LaunchDaemons/ （系统级服务，需要 sudo ）。为了便于管理，我们使用用户级。

假设我们的项目名为 CrashTolerantAIAgent ，在用户目录下有一个项目文件夹。

3.1 配置文件详解

创建一个文件： ~/Library/LaunchAgents/com.yourname.crashtolerantaiagent.plist

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <!-- 1. 服务唯一标识 -->
    <key>Label</key>
    <string>com.yourname.crashtolerantaiagent</string>

    <!-- 2. 指定要运行的程序和参数 -->
    <key>ProgramArguments</key>
    <array>
        <!-- 关键点：必须使用绝对路径。这里假设使用python3 -->
        <string>/usr/local/bin/python3</string>
        <!-- 我们的服务包装脚本路径 -->
        <string>/Users/yourusername/Projects/CrashTolerantAIAgent/agent_wrapper.py</string>
        <!-- 可以传递参数给脚本，例如指定配置文件 -->
        <string>--config</string>
        <string>/Users/yourusername/Projects/CrashTolerantAIAgent/config.yaml</string>
    </array>

    <!-- 3. 工作目录：脚本执行时的当前目录 -->
    <key>WorkingDirectory</key>
    <string>/Users/yourusername/Projects/CrashTolerantAIAgent</string>

    <!-- 4. 环境变量：这是最容易出错的地方！ -->
    <key>EnvironmentVariables</key>
    <dict>
        <!-- 确保PATH包含常用路径，特别是你的python环境路径 -->
        <key>PATH</key>
        <string>/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin</string>
        <!-- 如果你的Agent依赖虚拟环境，必须手动激活或指定PYTHONPATH -->
        <key>PYTHONPATH</key>
        <string>/Users/yourusername/Projects/CrashTolerantAIAgent</string>
        <!-- 其他自定义环境变量 -->
        <key>MY_AGENT_LOG_LEVEL</key>
        <string>INFO</string>
    </dict>

    <!-- 5. 标准输出和错误输出重定向：方便调试和查看日志 -->
    <key>StandardOutPath</key>
    <string>/Users/yourusername/Projects/CrashTolerantAIAgent/logs/agent.stdout.log</string>
    <key>StandardErrorPath</key>
    <string>/Users/yourusername/Projects/CrashTolerantAIAgent/logs/agent.stderr.log</string>

    <!-- 6. 运行方式：以当前用户身份运行 -->
    <key>RunAtLoad</key>
    <true/>
    <!-- 设置为true，表示加载本配置后立即启动服务。我们设为true，让Agent开机自启。 -->

    <!-- 7. 崩溃重启策略：“撞不死”的核心 -->
    <key>KeepAlive</key>
    <dict>
        <!-- SuccessfulExit: 如果进程正常退出（exit code为0）则不重启。我们设为false，因为Agent可能设计为常驻。 -->
        <key>SuccessfulExit</key>
        <false/>
        <!-- 也可以使用Crashed状态，但SuccessfulExit更直观 -->
    </dict>
    <!-- 限制重启频率，防止崩溃循环刷屏 -->
    <key>ThrottleInterval</key>
    <integer>10</integer>
    <!-- 至少间隔10秒才重启一次 -->

    <!-- 8. 进程退出信号处理：优雅关闭 -->
    <key>ExitTimeOut</key>
    <integer>30</integer>
    <!-- 给进程30秒时间处理SIGTERM信号，进行优雅退出（如保存状态）。超时则发SIGKILL -->

</dict>
</plist>

实操心得 ： EnvironmentVariables 和 ProgramArguments 的绝对路径是两大“坑王”。务必使用 which python3 命令确认你的python解释器路径。对于项目依赖，更稳妥的做法是在包装脚本内部，使用虚拟环境的绝对路径来执行，例如在脚本开头写 #!/Users/yourusername/.virtualenvs/ai-agent/bin/python ，然后在 .plist 的 ProgramArguments 中直接指向这个包装脚本本身（第一项），并赋予其可执行权限。这样可以绕过 launchd 环境变量设置的复杂性。

3.2 服务包装脚本（agent_wrapper.py）骨架

这个脚本是Agent核心的“保姆”。

#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
AI Agent 服务包装层。
负责进程守护、信号处理和状态持久化。
"""
import signal
import sys
import time
import logging
import sqlite3
from pathlib import Path
import your_agent_core_module  # 导入你的AI Agent核心模块

# 配置日志，方便在launchd的日志文件中查看
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        # 同时输出到文件和控制台（如果launchd允许）
        logging.FileHandler(Path(__file__).parent / 'logs' / 'agent_wrapper.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class AIAgentDaemon:
    def __init__(self, state_db_path='agent_state.db'):
        self.should_stop = False
        self.state_db_path = Path(state_db_path)
        self.agent = None
        self._init_state_db()
        # 注册信号处理器，用于优雅退出
        signal.signal(signal.SIGTERM, self._handle_signal)
        signal.signal(signal.SIGINT, self._handle_signal)

    def _init_state_db(self):
        """初始化状态数据库。"""
        conn = sqlite3.connect(self.state_db_path)
        c = conn.cursor()
        # 创建一个简单的表来存储状态。根据你的Agent状态复杂度设计表结构。
        c.execute('''
            CREATE TABLE IF NOT EXISTS agent_state (
                key TEXT PRIMARY KEY,
                value TEXT,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        conn.commit()
        conn.close()
        logger.info(f"状态数据库已初始化: {self.state_db_path}")

    def _handle_signal(self, signum, frame):
        """处理退出信号，触发优雅关闭流程。"""
        logger.info(f"接收到信号 {signum}，开始优雅关闭...")
        self.should_stop = True

    def save_state(self, key, value):
        """保存状态到数据库。"""
        try:
            conn = sqlite3.connect(self.state_db_path)
            c = conn.cursor()
            c.execute(
                "REPLACE INTO agent_state (key, value) VALUES (?, ?)",
                (key, value)
            )
            conn.commit()
            conn.close()
            logger.debug(f"状态已保存: {key}")
        except Exception as e:
            logger.error(f"保存状态失败: {e}")

    def load_state(self, key, default=None):
        """从数据库加载状态。"""
        try:
            conn = sqlite3.connect(self.state_db_path)
            c = conn.cursor()
            c.execute("SELECT value FROM agent_state WHERE key = ?", (key,))
            row = c.fetchone()
            conn.close()
            return row[0] if row else default
        except Exception as e:
            logger.error(f"加载状态失败: {e}")
            return default

    def run(self):
        """主运行循环。"""
        logger.info("AI Agent 守护进程启动。")
        # 1. 尝试加载上次的运行状态（例如，未完成的任务ID）
        last_task_id = self.load_state('last_task_id')
        if last_task_id:
            logger.info(f"检测到未完成的任务，恢复任务ID: {last_task_id}")
            # 这里可以调用你的Agent核心，告知其从last_task_id恢复
            recovery_context = {'task_id': last_task_id}
        else:
            recovery_context = None
            logger.info("无历史状态，启动新任务。")

        # 2. 初始化你的AI Agent核心，并传入恢复上下文
        # 假设你的核心模块有一个主类 `AIAgent`
        self.agent = your_agent_core_module.AIAgent(recovery_context=recovery_context)

        # 3. 主循环
        try:
            while not self.should_stop:
                # 这里是你Agent核心的工作循环
                # 例如：从队列取任务、处理、然后保存进度
                task_result = self.agent.do_work()  # 假设的Agent工作方法

                if task_result and task_result.get('task_id'):
                    # 假设每个任务单元完成后，保存当前任务ID作为进度
                    self.save_state('last_task_id', task_result['task_id'])
                    logger.info(f"任务进度已保存: {task_result['task_id']}")

                # 添加一个短暂休眠，避免空转CPU
                time.sleep(1)

        except KeyboardInterrupt:
            logger.info("收到键盘中断。")
        except Exception as e:
            logger.exception(f"主循环发生未预期异常: {e}")
            # 发生未捕获异常，进程会退出，launchd会根据KeepAlive策略重启它
            raise
        finally:
            # 4. 清理资源
            self._cleanup()
            logger.info("AI Agent 守护进程已停止。")

    def _cleanup(self):
        """清理资源，如断开AI API连接等。"""
        if self.agent:
            try:
                self.agent.close()
            except Exception as e:
                logger.error(f"清理Agent资源时出错: {e}")
        # 可以选择在完全退出前保存一个最终状态
        logger.info("资源清理完成。")

if __name__ == '__main__':
    daemon = AIAgentDaemon()
    daemon.run()

这个包装脚本实现了信号处理、状态持久化/加载、以及一个安全的主循环。当 launchd 发送 SIGTERM 信号要求停止服务时， _handle_signal 方法会被触发，设置 should_stop 标志，主循环会在完成当前工作单元后退出，并在 finally 块中进行清理。如果进程因为代码bug崩溃（未捕获异常）， launchd 会在 ThrottleInterval 后重启它。

4. 服务的加载、管理与调试

配置文件和服务脚本准备好后，真正的操作才开始。

4.1 加载与启动服务

1. 将.plist文件放到正确位置 ：

   cp ~/Desktop/com.yourname.crashtolerantaiagent.plist ~/Library/LaunchAgents/
   

2. 加载服务 ：使用 launchctl load 命令告诉 launchd 管理这个服务。

   launchctl load ~/Library/LaunchAgents/com.yourname.crashtolerantaiagent.plist
   

   由于我们设置了 <key>RunAtLoad</key><true/> ，加载后服务会立即启动。
3. 手动启动/停止/重启 （在服务已加载的前提下）：

   # 启动
   launchctl start com.yourname.crashtolerantaiagent
   # 停止 (会发送SIGTERM，触发我们的优雅退出逻辑)
   launchctl stop com.yourname.crashtolerantaiagent
   # 重启 (先stop再start)
   launchctl kickstart -k com.yourname.crashtolerantaiagent
   # 注意：`unload`是移除服务，不是停止。移除前应先stop。
   

4.2 查看状态与日志

1. 检查服务状态 ：

   launchctl list | grep com.yourname.crashtolerantaiagent
   

   输出中，第二列是进程的PID。如果是数字，表示正在运行；如果是 - ，表示已停止或出错。
2. 查看日志 ：这是最重要的调试手段。日志有两部分：
   • launchd 自身日志 ：使用 syslog 查看。

     log stream --predicate 'subsystem == "com.apple.xpc.launchd"' --info
     

     可以过滤出与你服务相关的事件，如启动、退出、重启。
   • 服务输出的日志 ：就是我们配置文件中 StandardOutPath 和 StandardErrorPath 指定的文件。直接 tail 查看：

     tail -f ~/Projects/CrashTolerantAIAgent/logs/agent.stdout.log
     tail -f ~/Projects/CrashTolerantAIAgent/logs/agent.stderr.log
     

     你的包装脚本中的 logger.info/error 输出会在这里看到。

4.3 常见问题与排查技巧实录

即使设计得再周全，实际部署时也难免踩坑。下面是我遇到过的典型问题及解决方法：

问题1：服务显示已加载（ launchctl list 能看到），但PID是 - ，没有启动。

• 排查 ：首先检查错误日志文件（ agent.stderr.log ）。最常见的原因是：
  • 路径错误 ： .plist 中的 ProgramArguments 、 WorkingDirectory 或脚本内部的路径使用了相对路径或 ~ 。 launchd 不解析 ~ ，必须使用绝对路径。
  • 权限问题 ：脚本文件没有执行权限。执行 chmod +x /path/to/your/agent_wrapper.py 。
  • 环境问题 ：在包装脚本开头打印 sys.path 和 os.environ ，重定向到日志文件，检查Python能否找到你的模块，环境变量是否正确。
• 解决 ：在 .plist 的 ProgramArguments 中，临时将脚本改为一个简单的测试脚本，如 <string>/bin/echo</string><string>Hello from launchd</string> ，并重定向输出到文件，确认 launchd 能正常调用。然后逐步恢复，定位问题点。

问题2：服务不断重启，形成崩溃循环。

• 排查 ：查看 stderr.log ，通常里面有Python的异常堆栈。可能是：
  • 代码Bug ：Agent核心代码有未处理的异常，导致进程一启动就崩溃。
  • 依赖缺失 ：在 launchd 环境下找不到第三方库。确保在包装脚本内使用绝对路径的Python解释器，并且该解释器的 site-packages 包含了所有依赖。使用虚拟环境是很好的实践。
  • 资源争用 ：例如，状态数据库文件被锁，或者端口冲突。
• 解决 ：
  1. 仔细阅读 stderr.log 。
  2. 在包装脚本的 run 方法最外层加一个大的 try-except ，将异常详情记录到日志，并让进程 sleep 一段时间再退出，避免 launchd 的 ThrottleInterval 还没到就立刻重启，方便你看日志。
  3. 检查 ThrottleInterval 设置是否太短。

问题3：执行 launchctl stop 后，进程没有优雅退出，而是被 SIGKILL 了。

• 排查 ：检查 .plist 中的 ExitTimeOut 值（默认10秒）。如果你的Agent清理工作（如保存大模型、断开多个连接）超过这个时间， launchd 就会强制杀死进程。
• 解决 ：增加 ExitTimeOut 的值，比如设为60。同时，优化你的 _cleanup 方法，让它更快。更关键的是，确保信号处理函数 _handle_signal 被正确设置，并且主循环能及时检测到 should_stop 标志。

问题4：合盖休眠后，网络任务中断，且恢复后Agent状态不对。

• 排查 ：这是“撞不死”要解决的核心场景之一。合盖休眠会触发网络中断，可能导致你的Agent正在进行的HTTP请求超时或失败。
• 解决 ：
  • 状态持久化 ：确保每个原子操作后都保存状态。例如，不是等一个长达10分钟的API调用完成才保存，而是将任务拆分为“准备请求”、“发送请求”、“处理结果”多个步骤，每步之后都保存进度。
  • 重试与幂等 ：在包装层或Agent核心，对网络请求实现带退避的重试机制。并且设计任务逻辑是幂等的，即同一个任务（用唯一ID标识）被重复执行多次，结果应该是一致的，不会产生副作用。
  • 监听系统事件（进阶） ： launchd 可以配置 WatchPaths 或 QueueDirectories ，但更复杂。一个更实用的方法是，在你的Agent主循环中，定期检查网络连通性，如果发现从断网恢复，则主动进行一次状态检查和任务恢复。

问题5：如何更新服务（代码或配置）？

• 标准流程 ：
  1. launchctl stop com.yourname.crashtolerantaiagent
  2. 更新你的代码或配置文件。
  3. launchctl unload ~/Library/LaunchAgents/com.yourname.crashtolerantaiagent.plist (如果需要修改 .plist 文件)。
  4. 修改 .plist 文件（如果需要）。
  5. launchctl load ~/Library/LaunchAgents/com.yourname.crashtolerantaiagent.plist
• 热更新技巧 ：对于只更新Agent核心逻辑（如 .py 文件），而包装层和 .plist 不变的情况，可以更优雅。在包装脚本的 run 循环中，可以检查一个特定的“重启信号文件”的时间戳。当你想更新时，只需 touch 这个文件，包装脚本检测到后，可以调用 sys.exit(0) 优雅退出。由于 KeepAlive 配置， launchd 会立即重启服务，加载新的代码。这避免了手动执行 stop 和 start 。

5. 进阶：实现更智能的自愈与监控

基础的“崩溃重启”只是容错的第一步。一个真正健壮的Agent，还需要更智能。

5.1 健康检查与自我修复

可以在包装脚本的主循环中，定期执行健康检查：

def _health_check(self):
    """检查Agent核心是否健康。"""
    if not self.agent:
        return False
    # 示例检查：1. 心跳API是否可达 2. 内部队列是否堆积 3. 数据库连接是否正常
    try:
        # 假设Agent有一个ping方法
        return self.agent.ping(timeout=5)
    except Exception:
        return False

# 在主循环中
while not self.should_stop:
    if time.time() - last_health_check > 60:  # 每分钟检查一次
        if not self._health_check():
            logger.error("健康检查失败，尝试重启Agent核心...")
            self._restart_agent_core()  # 重新初始化self.agent，但保持包装进程不退出
            last_health_check = time.time()
    # ... 正常工作任务 ...

这样，即使Agent核心逻辑“僵死”（进程还在，但不工作了），也能被检测并修复，而无需等待整个进程崩溃。

5.2 资源限制与防护

在 .plist 中，你可以添加资源限制，防止Agent发疯吃掉所有内存或CPU：

<key>HardResourceLimits</key>
<dict>
    <key>Core</key>
    <integer>0</integer> <!-- 核心文件大小限制 -->
    <key>CPU</key>
    <integer>300</integer> <!-- 最大CPU时间（秒），超过可能被限制 -->
    <key>Data</key>
    <integer>67108864</integer> <!-- 数据段大小限制，64MB -->
    <key>FileSize</key>
    <integer>unlimited</integer>
    <key>MemoryLock</key>
    <integer>unlimited</integer>
    <key>NumberOfFiles</key>
    <integer>1024</integer>
    <key>NumberOfProcesses</key>
    <integer>50</integer> <!-- 最大子进程数 -->
    <key>ResidentSetSize</key>
    <integer>268435456</integer> <!-- 最大物理内存，256MB -->
    <key>Stack</key>
    <integer>8388608</integer> <!-- 栈大小，8MB -->
</dict>
<key>SoftResourceLimits</key>
<dict>
    <!-- 软限制类似，略 -->
</dict>

这对于使用大语言模型的Agent尤其重要，可以防止单次请求消耗过多内存导致系统卡顿。

5.3 与通知系统集成

当服务多次重启失败，或健康检查持续不通过时，除了写日志，还可以发送通知。可以在包装脚本中集成邮件、Slack或macOS原生的 osascript 命令发送桌面通知。

import subprocess
def send_mac_notification(title, message):
    script = f'display notification "{message}" with title "{title}"'
    subprocess.run(['osascript', '-e', script])

在捕获到关键错误或重启次数过多时调用它，让你能及时介入。

经过以上步骤，你的AI Agent就已经从一个脆弱的脚本，进化成了一个由 launchd 托管、具备状态持久化、优雅退出、崩溃自启、甚至初步健康检查能力的“撞不死”的可靠服务。这套方案的核心思想—— 通过系统服务管理器进行进程守护，业务层实现状态持久化与优雅恢复 ——是跨平台的。在Linux上，你可以将 launchd 的 .plist 配置几乎一对一地翻译成 systemd 的 .service 文件，整体架构完全通用。