你的 CrewAI 代码生成任务跑了 45 分钟后，在「提交到 GitHub」步骤失败了——错误信息是「git remote: Repository not found」。事实上，这个 GitHub 仓库的访问 token 早在任务开始前就已经过期了，一个简单的 pre-flight 检查本可以在 2 秒内发现这个问题，避免 45 分钟的无效工作。或者在 LangGraph 里，一个需要读取 100GB 数据集的分析 workflow 在启动后 3 小时才因磁盘空间不足报错——而这个问题在 df -h 输出里一眼就能发现。Pre-flight 被跳过的代价是：本可在秒级发现的问题，变成了在分钟到小时后才暴露的失败。
常见原因
1. Pre-flight 逻辑是「建议性」的而不是「强制性」的
Pre-flight 函数存在于代码库里，但调用方式是 if config.run_preflight: check_prerequisites()，在 CI 里 run_preflight=False 被设置为默认值（为了加快 CI 速度），结果所有自动触发的任务都跳过了 pre-flight。
怎么判断：检查 pre-flight 调用处的条件判断，是否存在可以绕过检查的配置开关。
2. 任务被「快速重启」时跳过了初始化流程
主任务失败后，用户点击了「Retry」按钮，框架认为「这是重试，不是首次启动」，跳过了 pre-flight（因为「初次启动时已经检查过了」）。但首次启动时的 pre-flight 结果已经过期了（如 token 在重试期间到期）。
怎么判断：查看重试/恢复路径的代码，是否有 if is_first_run: run_preflight() 的逻辑。
3. Pre-flight 检查的工具本身调用了 LLM，被当成「可选的」跳过
某些框架把所有前置步骤都建模为 Agent tool，当系统资源紧张或 Agent 认为「这个检查不重要」时，LLM 会决定跳过这个工具调用。这让本应强制执行的安全检查变成了 LLM 的自由裁量。
怎么判断：检查 pre-flight 是否被定义为 Agent 的 optional tool（可被 LLM 选择性调用），而不是强制执行的系统级函数。
4. 微服务架构下 pre-flight 只在协调器执行，不在 worker 执行
主协调器做了 pre-flight 检查，但实际执行任务的 worker 节点（可能在不同的机器上）没有独立执行 pre-flight，worker 的环境可能与协调器不同（缺少某些环境变量、磁盘空间不足、依赖库版本不同）。
怎么判断：在 worker 节点上手动执行 pre-flight 检查，看结果是否与协调器上的一致。
5. Pre-flight 的超时设置过短，检查被截断
Pre-flight 连接数据库的操作需要 3 秒，但超时设置是 2 秒，导致检查超时，框架把「超时」误判为「检查通过」（因为没有显式报错），任务正常启动。
怎么判断：对比 pre-flight 各子项的实际耗时与超时设置，是否有子项经常超时。
6. Pre-flight 检查清单随功能迭代而过时
系统新增了「读取 S3 的权限要求」，但 pre-flight 清单没有相应更新，任务启动时没有检查 S3 权限，等到实际访问 S3 时才发现权限不足。
怎么判断：对比当前 pre-flight 检查的项目列表与任务实际依赖的所有外部资源/权限，检查是否有遗漏项。
最短修复路径
Step 1：把 pre-flight 定义为不可跳过的系统级守门函数
from functools import wraps
import inspect
class PreflightFailed(Exception):
"""Pre-flight 失败，任务不允许启动。"""
class PreflightChecker:
"""所有 pre-flight 检查的注册中心，不通过任何检查就阻断任务启动。"""
_checks: list = []
@classmethod
def register(cls, name: str, required: bool = True):
"""注册 pre-flight 检查项，required=True 的项目失败时强制阻断任务。"""
def decorator(fn):
cls._checks.append({"name": name, "fn": fn, "required": required})
return fn
return decorator
@classmethod
def run_all(cls, context: dict) -> dict[str, bool]:
failures = []
results = {}
for check in cls._checks:
try:
result = check["fn"](context)
results[check["name"]] = True
print(f" [PASS] {check['name']}")
except Exception as e:
results[check["name"]] = False
print(f" [FAIL] {check['name']}: {e}")
if check["required"]:
failures.append(f"{check['name']}: {e}")
if failures:
raise PreflightFailed(
f"以下 {len(failures)} 个必要的 pre-flight 检查未通过，任务不允许启动：\n" +
"\n".join(f" - {f}" for f in failures)
)
return results
# 注册检查项（不能被跳过）
@PreflightChecker.register("github_token_valid", required=True)
def check_github_token(ctx: dict):
import requests
r = requests.get(
"https://api.github.com/user",
headers={"Authorization": f"token {ctx['github_token']}"},
timeout=5
)
r.raise_for_status()
@PreflightChecker.register("disk_space_sufficient", required=True)
def check_disk_space(ctx: dict):
import shutil
total, used, free = shutil.disk_usage(ctx.get("work_dir", "/tmp"))
required_gb = ctx.get("required_disk_gb", 10)
free_gb = free / (1024**3)
if free_gb < required_gb:
raise ValueError(f"磁盘空间不足：需要 {required_gb}GB，剩余 {free_gb:.1f}GB")Step 2：在任务入口强制调用 pre-flight，不允许跳过
def start_workflow(config: WorkflowConfig) -> WorkflowResult:
"""任务入口：pre-flight 是必须的，没有任何参数可以跳过它。"""
print("=== 执行 Pre-flight 检查 ===")
# 无条件执行，没有 if/else
PreflightChecker.run_all(config.to_context())
print("=== Pre-flight 全部通过，开始执行任务 ===")
return _execute_workflow(config)
# 重试路径也必须执行 pre-flight
def retry_workflow(failed_run_id: str, config: WorkflowConfig) -> WorkflowResult:
print("=== 重试任务，重新执行 Pre-flight 检查 ===")
# 重试时同样执行 pre-flight（因为环境可能已经变化）
PreflightChecker.run_all(config.to_context())
checkpoint = load_checkpoint(failed_run_id)
return _resume_workflow(checkpoint, config)Step 3：pre-flight 清单与任务资源声明绑定
class WorkflowConfig(BaseModel):
# 声明任务依赖的外部资源
required_permissions: list[str] = []
required_env_vars: list[str] = []
required_disk_gb: float = 5.0
required_tools: list[str] = []
def to_context(self) -> dict:
return self.model_dump()
# 通用 pre-flight：根据声明自动检查，不需要手动维护
@PreflightChecker.register("required_env_vars", required=True)
def check_env_vars(ctx: dict):
missing = [v for v in ctx.get("required_env_vars", []) if not os.environ.get(v)]
if missing:
raise ValueError(f"缺少环境变量：{', '.join(missing)}")Step 4：为 pre-flight 单独设置宽松超时
import asyncio
async def run_preflight_with_timeout(context: dict, timeout_seconds: int = 30):
"""Pre-flight 允许比实际工具调用更长的超时（因为检查必须完成）。"""
try:
async with asyncio.timeout(timeout_seconds):
return await PreflightChecker.run_all_async(context)
except asyncio.TimeoutError:
raise PreflightFailed(
f"Pre-flight 检查超时（{timeout_seconds}s），请检查网络连接或外部服务状态"
)
# 注意：超时时抛 PreflightFailed 而不是静默通过Step 5：在 CI 里测试 pre-flight 对各类失败的检测能力
# tests/test_preflight.py：验证 pre-flight 能检测到常见问题
# 模拟 GitHub token 过期
GITHUB_TOKEN=invalid_token python -c "
from workflow import start_workflow, WorkflowConfig
try:
start_workflow(WorkflowConfig(github_token='invalid'))
assert False, 'Pre-flight 应该失败但没有'
except PreflightFailed as e:
assert 'github_token_valid' in str(e)
print('PASS: 过期 token 被正确检测')
"预防建议
- Pre-flight 必须是系统级强制函数，不允许通过配置参数跳过（移除所有
if config.run_preflight开关） - 重试和恢复路径与首次启动路径一样，都必须执行 pre-flight
- 任务配置里声明所有外部依赖（权限、环境变量、磁盘空间、工具），pre-flight 根据声明自动检查
- Pre-flight 超时设置独立于任务超时，设置比实际检查耗时宽松 2-3 倍
- 每次新增外部依赖时，同步在 pre-flight 检查清单里添加对应的检查项（在代码 review 里强制要求）
- 在 worker 节点上独立执行 pre-flight，不只依赖协调器的检查
- 为每个 pre-flight 项写单元测试，覆盖「应该通过」和「应该失败」两种情况
常见问答 (FAQ)
Q: Pre-flight 需要联网检查（如验证 API token），这会不会让测试环境跑不起来？
A: 用接口注入（依赖注入）让 pre-flight 的外部调用可以被 mock：在测试里传入 check_github_token=lambda ctx: True（模拟总是通过）。生产代码里用真实的检查函数。这样测试不需要真实的网络访问，同时保证了生产代码的检查不被跳过。
Q: Pre-flight 失败后 Agent 应该做什么，直接报错退出吗？ A: Pre-flight 失败应该立即终止任务，返回结构化的错误信息（包含哪些检查失败、如何修复）。不要让 Agent 尝试「绕过」pre-flight 失败继续执行——如果前提条件没有满足，任务注定会在中途失败，提前失败比中途失败更好。
Q: 如何处理「pre-flight 偶发失败但实际上任务可以执行」的情况？ A: 这说明 pre-flight 的检查过于严格，或者检查的对象本身不稳定（如网络抖动导致 token 验证请求超时）。解决方案：对 pre-flight 本身加重试逻辑（如重试 3 次，每次等 2 秒），所有重试都失败后才认定为真正的 pre-flight 失败。不要把偶发失败当作「可以忽略」。
Q: 在 Temporal 里如何实现不可跳过的 pre-flight？
A: 在 workflow 的第一个 activity 里执行 pre-flight，如果失败则返回错误并终止 workflow（不继续调度后续 activity）。Temporal 的 activity retry 政策设置 maximum_attempts=3，3 次都失败后标记 workflow 为失败。不要把 pre-flight 放在 workflow 启动前的纯代码层（因为那里可以被绕过），而是作为 workflow 的第一个正式步骤。