1. 长任务多Agent协作的痛点与Manifest交接模式的由来

做过AI Agent编排的人大概都有过这种体验：一个稍微复杂点的任务，比如让Agent自动完成一份行业调研报告，从检索资料、整理数据、生成图表到最终排版输出，单靠一个Agent串行跑下来，中间只要有一环卡住或者上下文超限，整个任务就得推倒重来。更麻烦的是，当你想把任务拆给多个Agent并行处理时，文件怎么共享、状态怎么同步、谁改了哪个文件、改到哪一步了，这些问题会瞬间把复杂度拉满。

我最早接触这类需求是在一个自动化内容生产项目里，当时需要让三个Agent分别负责素材采集、内容撰写和格式校验，它们要共享同一批中间文件。最开始用的是最朴素的办法——所有Agent直接读写同一个目录，靠文件名约定来区分职责。结果跑了不到两天就出问题了：Agent A正在写文件的时候Agent B去读，读到了半截内容；Agent C误删了Agent A还没处理完的临时文件；更头疼的是任务中断后重启，根本不知道上次跑到哪了，只能全部重来。

后来我尝试过用数据库记录状态、用消息队列做同步，甚至考虑过引入完整的分布式文件系统，但要么太重，要么在Agent这种轻量级、高频次、短生命周期的场景下水土不服。直到我把思路从“共享一个文件系统”转向“共享一份交接清单”，整个问题才豁然开朗。这个交接清单，就是我今天要聊的Manifest交接模式。

简单来说，Manifest就是一份结构化的清单文件，它不存实际数据，只记录“有哪些文件、每个文件的状态是什么、由哪个Agent负责、校验值是多少、下一步该谁接手”。每个Agent在开始工作前先读Manifest，确认自己该处理哪些文件；处理完后更新Manifest，把状态标记为完成并写入校验值；下一个Agent再根据更新后的Manifest继续接力。整个过程就像工厂流水线上的交接单，工人不需要知道整条产线怎么运转，只需要看自己工位上的单子，做完签字传给下一站。

这套模式特别适合那些长周期、多阶段、需要多个Agent接力完成的任务场景，比如自动化报告生成、批量数据处理、多轮内容审核、代码生成与测试流水线等。不管你是刚接触Agent开发的新手，还是已经在做多Agent编排的老手，只要你的任务里出现了“多个Agent要共享文件”这个需求，Manifest交接模式都值得你花时间了解一下。接下来我会从设计思路、核心细节、实操实现和问题排查四个维度，把这套模式掰开揉碎讲清楚。

2. Manifest交接模式的核心设计与选型考量

2.1 为什么是Manifest而不是共享目录或数据库

在决定用Manifest之前，我认真对比过几种常见的多Agent文件共享方案，这里把当时的思考过程整理出来，方便你在自己的场景里做判断。

方案一：裸共享目录。所有Agent直接读写同一个文件夹，靠文件名和目录结构来约定职责。优点是实现简单，零额外依赖；缺点是没有任何状态管理能力，并发读写没有保护，任务中断后无法恢复现场。适合那种一次性的、Agent之间不需要严格顺序的简单场景，一旦任务变长、Agent变多，维护成本会指数级上升。

方案二：中心化数据库。用SQLite或PostgreSQL记录文件路径、状态、负责人等信息，Agent通过数据库事务来协调。优点是状态管理能力强，支持复杂查询和并发控制；缺点是引入了额外依赖，Agent需要维护数据库连接，对于轻量级任务来说太重了。而且数据库里的记录和实际文件系统之间容易出现不一致，需要额外的对账逻辑。

方案三：消息队列驱动。用Redis或RabbitMQ做Agent之间的消息传递，文件路径和状态通过消息来同步。优点是解耦彻底，扩展性好；缺点是调试困难，消息丢失或重复消费的处理逻辑复杂，对于文件这种“有状态”的资源来说，消息队列并不是最自然的载体。

方案四：Manifest交接模式。用一份结构化的清单文件（通常是JSON或YAML）作为唯一的协调中心，Agent通过读写这份清单来交接工作。它本质上是把“共享目录”和“状态管理”合二为一：清单本身就是一个普通文件，放在共享目录里，但它承载了所有协调所需的信息。优点是轻量、无额外依赖、状态可追溯、中断可恢复；缺点是对高频并发写入的支持有限，需要通过设计来规避冲突。

我最终选择Manifest，核心原因是它在轻量和可靠之间找到了一个很好的平衡点。对于大多数Agent协作场景来说，并发写入的频率并不高——一个Agent处理完一批文件才更新一次Manifest，而不是每写一个字节就更新。这种低频写入的特性，让文件锁或原子替换就能解决大部分并发问题，不需要引入复杂的分布式协调机制。

2.2 Manifest的数据结构设计

Manifest的结构设计直接决定了整套模式好不好用。我踩过几次坑之后，总结出一个比较通用的结构，你可以根据自己的场景调整。

{ "manifest_version": "1.0", "task_id": "report_gen_20250101_001", "created_at": "2025-01-01T10:00:00Z", "updated_at": "2025-01-01T10:35:00Z", "current_stage": "content_writing", "stages": [ { "name": "data_collection", "agent": "collector_agent", "status": "completed", "started_at": "2025-01-01T10:00:00Z", "finished_at": "2025-01-01T10:15:00Z" }, { "name": "content_writing", "agent": "writer_agent", "status": "in_progress", "started_at": "2025-01-01T10:15:00Z", "finished_at": null }, { "name": "format_check", "agent": "checker_agent", "status": "pending", "started_at": null, "finished_at": null } ], "files": [ { "path": "raw_data/source_001.json", "stage": "data_collection", "status": "completed", "sha256": "a1b2c3d4e5f6...", "size_bytes": 20480, "produced_by": "collector_agent", "consumed_by": ["writer_agent"] }, { "path": "draft/report_v1.md", "stage": "content_writing", "status": "in_progress", "sha256": null, "size_bytes": null, "produced_by": "writer_agent", "consumed_by": ["checker_agent"] } ] }

这个结构里有几个关键设计点值得展开说。

task_id和manifest_version是整套机制的基石。task_id用来区分不同的任务实例，避免多个任务共用一份Manifest时互相干扰；manifest_version用来做结构演进，将来字段有增减时，Agent可以根据版本号决定怎么解析。

stages数组定义了任务的阶段划分和每个阶段的负责人。每个阶段有明确的状态机：pending → in_progress → completed。Agent启动时先看current_stage，确认自己是不是当前该干活的角色；干完活把当前阶段标记为completed，把下一个阶段标记为in_progress，同时更新current_stage。这样即使任务中断，重启后Agent也能从Manifest里准确知道该从哪继续。

files数组是Manifest的核心，记录了每个文件的完整生命周期。path是相对路径，相对于共享目录的根；stage表示这个文件属于哪个阶段；status表示文件当前的状态；sha256是文件内容的校验值，用来确保下一个Agent读到的文件没有被篡改或损坏；produced_by和consumed_by记录了文件的来源和去向，方便追溯。

注意：sha256的计算时机很关键。我的做法是Agent写完文件后立即计算并写入Manifest，下一个Agent读取文件后先校验sha256，确认一致再开始处理。如果校验不通过，说明文件在传输或存储过程中出了问题，需要重新生成或从备份恢复。

2.3 为什么用SHA256做文件校验

在Manifest里记录文件的SHA256值，这个设计一开始我觉得有点多余——文件就在本地磁盘上，又不是网络传输，能出什么问题？但实际跑下来发现，校验值至少帮我排查了三类问题。

第一类是写入不完整。Agent写文件时如果进程被意外终止，可能只写了一半。下一个Agent读到这个半截文件，如果直接处理，轻则报错，重则产生错误结果还浑然不知。有了SHA256，读取时先校验，不匹配就拒绝处理，把问题暴露在早期。

第二类是文件被意外修改。多个Agent共享目录时，难免出现某个Agent误操作改了不属于自己的文件。SHA256校验能立刻发现这种越界修改，避免错误扩散。

第三类是任务恢复时的完整性确认。任务中断后重启，Agent需要确认之前已经完成的文件是否完好。逐个校验SHA256，比重新生成一遍要快得多，也可靠得多。

计算SHA256在Python里非常简单，几行代码就能搞定：

import hashlib def calculate_sha256(file_path): sha256_hash = hashlib.sha256() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): sha256_hash.update(chunk) return sha256_hash.hexdigest()

这里用分块读取而不是一次性读入内存，是为了处理大文件时不会把内存撑爆。8192字节的块大小是一个经验值，兼顾了读取效率和内存占用。实测下来，对于一个100MB的文件，分块读取和一次性读取的耗时差异在5%以内，但内存占用差了上百倍。

2.4 并发写入的规避策略

Manifest模式最大的软肋是并发写入。如果两个Agent同时更新Manifest，后写的会覆盖先写的，导致状态丢失。我的应对策略分三层。

第一层是流程设计上避免并发。Manifest交接模式天然适合串行流水线：Agent A完成 → 更新Manifest → Agent B开始。只要在流程设计上保证同一时刻只有一个Agent在写Manifest，并发问题就不存在。这也是为什么我在stages里用current_stage来标识当前阶段，而不是让多个Agent同时处理不同阶段。

第二层是文件锁。如果确实需要多个Agent同时运行，可以用文件锁来串行化Manifest的写入。Python的fcntl模块（Linux/macOS）或msvcrt模块（Windows）都支持文件锁。下面是一个跨平台的简单实现：

import os import time import json class ManifestLock: def __init__(self, manifest_path, timeout=30): self.lock_path = manifest_path + ".lock" self.timeout = timeout def acquire(self): start = time.time() while time.time() - start < self.timeout: try: fd = os.open(self.lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) os.write(fd, str(os.getpid()).encode()) os.close(fd) return True except FileExistsError: time.sleep(0.1) raise TimeoutError("无法获取Manifest锁") def release(self): try: os.remove(self.lock_path) except FileNotFoundError: pass

这个实现用“创建独占文件”来模拟锁，简单但有效。获取锁时如果文件已存在就等待重试，释放锁时删除文件。超时时间设为30秒，避免死锁时无限等待。

第三层是原子替换。更新Manifest时，先写入一个临时文件，再用os.replace原子替换原文件。这样即使写入过程中进程崩溃，原Manifest也不会被破坏。

def update_manifest(manifest_path, new_data): tmp_path = manifest_path + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(new_data, f, ensure_ascii=False, indent=2) os.replace(tmp_path, manifest_path)

os.replace在大多数操作系统上都是原子操作，能保证Manifest要么是旧版本，要么是新版本，不会出现中间状态。

3. 实操落地：从零搭建Manifest交接系统

3.1 环境准备与目录结构规划

在动手写代码之前，先把目录结构规划好，后面会省很多事。我的习惯是按“共享区+Agent工作区+日志区”来划分：

project_root/ ├── shared/ │ ├── manifest.json # 核心交接清单 │ ├── manifest.json.lock # 锁文件（运行时生成） │ └── files/ # 所有共享文件存放处 │ ├── raw_data/ │ ├── draft/ │ └── final/ ├── agents/ │ ├── collector_agent.py │ ├── writer_agent.py │ └── checker_agent.py ├── logs/ │ └── agent_activity.log └── config/ └── manifest_schema.json # Manifest结构定义

shared目录是所有Agent都能访问的公共区域，manifest.json放在这里。files子目录按阶段再分子目录，避免所有文件堆在一起。agents目录放各个Agent的实现代码，每个Agent独立一个文件，方便单独调试和替换。logs目录记录Agent的活动日志，排查问题时非常有用。

Python环境方面，这套模式对Python版本要求不高，3.8以上都能跑。核心依赖只有标准库里的hashlib、json、os、time，不需要额外安装第三方包。如果你想让代码更健壮，可以装一个filelock库来替代手写的文件锁：

pip install filelock

filelock的用法比手写锁更简洁，而且跨平台兼容性更好：

from filelock import FileLock lock = FileLock("shared/manifest.json.lock", timeout=30) with lock: # 在这里读写Manifest pass

3.2 Manifest的初始化与读取

任务开始时，第一个Agent负责初始化Manifest。初始化的工作包括：创建目录结构、生成task_id、写入初始的stages和空的files数组。

import json import os import uuid from datetime import datetime, timezone def init_manifest(shared_dir, stages_config): manifest_path = os.path.join(shared_dir, "manifest.json") if os.path.exists(manifest_path): raise FileExistsError("Manifest已存在，如需重新开始请先清理") os.makedirs(os.path.join(shared_dir, "files"), exist_ok=True) now = datetime.now(timezone.utc).isoformat() manifest = { "manifest_version": "1.0", "task_id": f"task_{uuid.uuid4().hex[:12]}", "created_at": now, "updated_at": now, "current_stage": stages_config[0]["name"], "stages": [ { "name": s["name"], "agent": s["agent"], "status": "pending", "started_at": None, "finished_at": None } for s in stages_config ], "files": [] } with open(manifest_path, "w", encoding="utf-8") as f: json.dump(manifest, f, ensure_ascii=False, indent=2) return manifest

stages_config是一个列表，按顺序定义每个阶段的名称和负责的Agent。比如：

stages_config = [ {"name": "data_collection", "agent": "collector_agent"}, {"name": "content_writing", "agent": "writer_agent"}, {"name": "format_check", "agent": "checker_agent"} ]

读取Manifest的代码更简单，但要注意加锁和校验：

def read_manifest(shared_dir): manifest_path = os.path.join(shared_dir, "manifest.json") with open(manifest_path, "r", encoding="utf-8") as f: manifest = json.load(f) return manifest

实际使用时，读取和更新通常要放在同一个锁里，避免读到旧数据后基于旧数据做更新，导致覆盖别人的修改。

3.3 Agent如何认领任务并更新Manifest

每个Agent启动后的第一件事，就是读Manifest，确认自己是不是当前该干活的角色。判断逻辑很简单：看current_stage对应的stage的agent字段是不是自己。

def claim_task(shared_dir, agent_name): lock = FileLock(os.path.join(shared_dir, "manifest.json.lock"), timeout=30) with lock: manifest = read_manifest(shared_dir) current = manifest["current_stage"] for stage in manifest["stages"]: if stage["name"] == current: if stage["agent"] != agent_name: return None # 不是自己的活 if stage["status"] == "in_progress": return None # 已经有人在干了 stage["status"] = "in_progress" stage["started_at"] = datetime.now(timezone.utc).isoformat() manifest["updated_at"] = stage["started_at"] update_manifest(shared_dir, manifest) return manifest return None

这段代码里有几个细节值得注意。第一，整个“读取-判断-更新”过程都在锁内完成，避免两个Agent同时判断自己该干活。第二，如果stage已经是in_progress状态，说明有另一个Agent正在处理，当前Agent应该退出而不是重复处理。第三，更新Manifest时同步更新updated_at字段，方便后续排查时间线。

Agent完成工作后，需要做三件事：把产出文件登记到files数组、把当前stage标记为completed、把下一个stage标记为in_progress并更新current_stage。

def complete_stage(shared_dir, agent_name, produced_files): lock = FileLock(os.path.join(shared_dir, "manifest.json.lock"), timeout=30) with lock: manifest = read_manifest(shared_dir) current = manifest["current_stage"] now = datetime.now(timezone.utc).isoformat() # 登记产出文件 for file_info in produced_files: full_path = os.path.join(shared_dir, "files", file_info["path"]) file_info["sha256"] = calculate_sha256(full_path) file_info["size_bytes"] = os.path.getsize(full_path) file_info["produced_by"] = agent_name file_info["status"] = "completed" manifest["files"].append(file_info) # 标记当前阶段完成 stage_index = None for i, stage in enumerate(manifest["stages"]): if stage["name"] == current: stage["status"] = "completed" stage["finished_at"] = now stage_index = i break # 激活下一个阶段 if stage_index is not None and stage_index + 1 < len(manifest["stages"]): next_stage = manifest["stages"][stage_index + 1] next_stage["status"] = "in_progress" next_stage["started_at"] = now manifest["current_stage"] = next_stage["name"] else: manifest["current_stage"] = "completed" manifest["updated_at"] = now update_manifest(shared_dir, manifest)

produced_files是一个列表，每个元素包含path和stage字段，path是相对于shared/files的路径。Agent在调用这个函数之前，需要确保文件已经写入到正确的位置。

3.4 文件校验与消费确认

下一个Agent在读取文件之前，必须先校验SHA256。这个步骤看起来繁琐，但能避免很多隐蔽的问题。

def verify_and_consume(shared_dir, file_path, expected_sha256): full_path = os.path.join(shared_dir, "files", file_path) if not os.path.exists(full_path): raise FileNotFoundError(f"文件不存在: {file_path}") actual_sha256 = calculate_sha256(full_path) if actual_sha256 != expected_sha256: raise ValueError( f"文件校验失败: {file_path}\n" f"期望: {expected_sha256}\n" f"实际: {actual_sha256}" ) return full_path

校验通过后，Agent可以放心地读取和处理文件。处理完成后，如果这个文件被消费了（比如被合并到另一个文件里），可以在Manifest里更新consumed_by字段，记录消费关系。这个字段不是必须的，但在排查“某个文件到底被谁用了”这类问题时非常有用。

实操心得：SHA256校验虽然可靠，但对于超大文件（比如几个GB的视频文件）来说，每次读取都全量计算校验值会很耗时。我的做法是对于超过100MB的文件，只在写入时计算一次，后续读取时用文件大小和修改时间做快速校验，只有在这两个值不匹配时才重新计算SHA256。这样在保证可靠性的同时，把校验开销降到了可接受的范围。

3.5 任务中断后的恢复逻辑

Manifest模式最大的优势之一就是中断可恢复。任务因为任何原因中断后，重新启动时只需要读Manifest，就能知道当前进行到哪一步了。

恢复逻辑分两种情况。第一种情况是当前stage是in_progress状态，说明上次有Agent正在处理但没完成。这时候需要判断：是重新处理，还是从断点继续？我的做法是看这个stage有没有产生部分文件。如果有部分文件已经登记在files数组里且状态是completed，说明部分工作已完成，可以从断点继续；如果没有，就重新开始这个stage。

def resume_task(shared_dir, agent_name): manifest = read_manifest(shared_dir) current = manifest["current_stage"] if current == "completed": print("任务已完成，无需恢复") return None for stage in manifest["stages"]: if stage["name"] == current: if stage["agent"] != agent_name: print(f"当前阶段由 {stage['agent']} 负责，本Agent无法恢复") return None if stage["status"] == "in_progress": # 检查是否有部分产出 partial_files = [ f for f in manifest["files"] if f["stage"] == current and f["status"] == "completed" ] if partial_files: print(f"发现 {len(partial_files)} 个已完成文件，从断点继续") return {"manifest": manifest, "resume_from": partial_files} else: print("无部分产出，重新开始当前阶段") return {"manifest": manifest, "resume_from": None} return {"manifest": manifest, "resume_from": None} return None

第二种情况是当前stage是pending状态，说明上一个stage完成后没有正确激活下一个stage，或者Manifest被手动修改过。这时候需要把当前stage标记为in_progress，然后正常开始处理。

4. 常见问题与排查技巧实录

4.1 Manifest损坏或丢失怎么办

Manifest是整套模式的核心，一旦损坏或丢失，整个任务的状态就丢了。我遇到过两次Manifest损坏的情况，一次是磁盘写入过程中断电，一次是手动编辑时JSON格式写错了。两次都靠备份恢复了。

我的做法是每次更新Manifest时，先把旧版本复制一份到backup目录，保留最近5个版本。这样即使当前Manifest损坏，也能从备份恢复。

import shutil def backup_manifest(shared_dir, keep=5): manifest_path = os.path.join(shared_dir, "manifest.json") backup_dir = os.path.join(shared_dir, "backup") os.makedirs(backup_dir, exist_ok=True) if not os.path.exists(manifest_path): return timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") backup_path = os.path.join(backup_dir, f"manifest_{timestamp}.json") shutil.copy2(manifest_path, backup_path) # 清理旧备份 backups = sorted([ f for f in os.listdir(backup_dir) if f.startswith("manifest_") ]) for old in backups[:-keep]: os.remove(os.path.join(backup_dir, old))

如果Manifest彻底丢失且没有备份，最后的办法是根据files目录里的实际文件重建Manifest。重建时只能恢复文件列表和SHA256，阶段状态需要人工判断。所以备份这件事，宁可多做不要少做。

4.2 Agent卡死或超时怎么处理

Agent卡死是另一个常见问题。可能是代码bug导致死循环，也可能是等待某个永远不来的资源。我的处理策略是给每个Agent设置超时，超时后自动释放锁并标记当前stage为failed。

import signal class TimeoutError(Exception): pass def timeout_handler(signum, frame): raise TimeoutError("Agent执行超时") def run_with_timeout(func, timeout_seconds=300): signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout_seconds) try: result = func() signal.alarm(0) return result except TimeoutError: signal.alarm(0) raise

超时后，Manifest里的stage状态需要从in_progress改为failed，并记录失败原因。下一个Agent启动时看到failed状态，可以选择重试或者跳过。这里要注意，signal.alarm只在Unix系统上有效，Windows上需要用threading.Timer来替代。

踩过的坑：有一次Agent卡死但没有设置超时，锁文件一直没释放，导致后续所有Agent都卡在获取锁这一步。后来我加了一个锁的过期机制：锁文件里写入获取时间，如果超过一定时间（比如10分钟）还没释放，就认为锁已过期，可以强制获取。这个机制救了我好几次。

4.3 文件校验失败的排查思路

SHA256校验失败通常有三种原因：文件被意外修改、文件写入不完整、Manifest里的校验值记录错误。排查时按这个顺序来。

先看文件大小。如果实际文件大小和Manifest里记录的size_bytes不一致，基本可以确定是写入不完整或文件被修改。如果大小一致但SHA256不匹配，那可能是内容被修改了但大小没变，或者Manifest里的校验值本身记错了。

再看文件修改时间。如果修改时间晚于Manifest里记录的完成时间，说明文件在登记之后被改动过。这时候需要查日志，看是哪个Agent在什么时间动了这个文件。

最后重新计算一次SHA256，确认不是计算过程出了问题。如果重新计算结果和Manifest里记录的一致，那说明之前的校验失败是偶发问题（比如读取时文件正在被写入），重试即可。

4.4 多Agent同时启动时的竞争处理

如果多个Agent同时启动，它们会同时尝试认领任务。虽然文件锁能保证只有一个Agent能成功更新Manifest，但其他Agent会收到“任务已被认领”的信号然后退出。这本身没问题，但如果Agent的启动逻辑是“启动后一直重试直到认领成功”，就可能出现活锁——多个Agent反复竞争同一个锁，谁也干不成活。

我的做法是给认领加一个随机退避：认领失败后，等待一个随机时间再重试，而不是立即重试。

import random import time def claim_with_backoff(shared_dir, agent_name, max_retries=5): for attempt in range(max_retries): result = claim_task(shared_dir, agent_name) if result is not None: return result wait = random.uniform(1, 5) * (attempt + 1) time.sleep(wait) return None

随机退避能有效打破活锁，让竞争变得公平。实测下来，三个Agent同时启动时，平均在2秒内就能完成认领，不会出现长时间空转。

4.5 Manifest交接模式的适用边界

说了这么多Manifest的好处，也得聊聊它不适合什么场景。如果你的任务满足以下任一条件，Manifest模式可能不是最优选择。

第一，高频并发写入。如果多个Agent需要每秒更新几十次Manifest，文件锁会成为瓶颈。这种场景更适合用数据库或内存队列。

第二，跨机器协作。Manifest模式假设所有Agent访问的是同一个文件系统。如果Agent分布在不同机器上，需要先解决文件系统共享的问题，这时候Manifest的优势就不明显了。

第三，超大规模文件管理。如果任务涉及几十万个文件，Manifest会变得非常大，每次读写都是全量操作，效率很低。这种场景需要分片Manifest或者用专门的元数据管理服务。

第四，强一致性要求。Manifest模式提供的是最终一致性，不是强一致性。如果业务要求任何时刻所有Agent看到的状态都完全一致，需要引入额外的协调机制。

我个人的经验是，Manifest模式最适合单机或共享存储环境下、Agent数量在3到10个之间、任务周期从几分钟到几小时、文件数量在几百到几千个的场景。超出这个范围，就需要考虑更重的方案了。

最后分享一个我在实际项目中总结的小技巧：在Manifest里加一个heartbeat字段，每个Agent在处理过程中定期更新这个字段的时间戳。这样其他Agent或监控程序就能通过心跳判断某个Agent是否还活着。如果心跳超过一定时间没更新，就可以认为该Agent已经失效，可以安全地接管它的工作。这个机制配合超时和锁过期，基本能覆盖绝大多数Agent异常退出的情况。