使用 Temporal 和 CrewAI 构建容错 AI 智能体工作流

目前，AI 智能体（AI Agents）领域充斥着各种令人惊叹的 Demo，但遗憾的是，这些 Demo 在面对生产环境的严苛要求时往往显得力不从心。大多数 AI 智能体框架的设计初衷是快速开发，而非系统耐用性。它们通常将状态保存在进程内存中，将“人工审批”视为一种可选的 UI 交互而非强制性的逻辑门控，并且对于凌晨两点发生的 LLM API 超时完全没有应对方案。对于在 n1n.ai 上构建应用的开发者来说，目标不仅仅是调用一个大模型，而是建立一个能够可靠管理这些调用的系统。

在真实的生产部署中，系统状态的生命周期必须超越进程本身。一个文档生成或审批工作流可能需要等待人工审核员数天时间。传统的 Python 脚本、甚至是一些由用户自行管理状态的 LangGraph 实例，在等待期间如果遇到容器重启或崩溃，其上下文就会丢失，除非你投入大量精力去开发一套复杂的持久化层。而 Temporal 与 CrewAI 的结合，正是解决这一痛点的关键方案。

企业级 AI 的挑战：跨越 Demo 陷阱

当你准备大规模部署 AI 智能体——使用通过 n1n.ai 提供的 DeepSeek-V3 或 Claude 3.5 Sonnet 等顶级模型时，你会面临三个不可回避的障碍：

1. 瞬时 API 故障：频率限制（Rate Limits）和 5xx 错误是常态。零散的重试逻辑会导致状态不一致。
2. 长周期状态维持：人工在环（Human-in-the-loop, HITL）不只是一个简单的 Webhook，它是一个分布式系统问题。系统必须在不丢失上下文的情况下，无限期阻塞并等待外部信号。
3. 版本迭代：你无法在不破坏执行历史的情况下，直接更新一个已经运行了三天的在线工作流代码。

方案对比：编排策略的演进

架构设计：Temporal 为骨架，CrewAI 为大脑

核心设计原则是将 编排（Orchestration） 与 推理（Reasoning） 分离。Temporal 充当编排器，是状态和执行顺序的唯一事实来源；CrewAI 充当推理单元，是执行具体任务的无状态工人。这种分离确保了即便一个 CrewAI 智能体在处理复杂的 RAG（检索增强生成）任务时耗时数分钟，整个工作流依然是持久可靠的。通过 n1n.ai 接入 LLM，可以确保推理过程的高效与稳定。

1. 确定性与沙箱机制

Temporal 的工作流函数必须是确定性的。它并不像传统代码那样直接运行，而是通过重放事件历史来重建状态。因此，任何非确定性的操作（如 LLM 调用、网络请求）都必须被隔离在 Activity 中。由于 CrewAI 会进行网络调用并管理内部状态，它不能直接写在 Workflow 逻辑里，必须封装在 Activity 中运行。

# 必须通过特定的方式导入非确定性库
with workflow.unsafe.imports_passed_through():
    from activities.sop_activity import generate_sop_phase_activity
    from activities.fix_sop_activity import fix_sop_with_crew_activity

2. 解耦多智能体工作流

一个常见的错误是将整个 CrewAI 的 “Crew”（例如包含 Writer 和 Reviewer）打包进一个 Activity。如果 Reviewer 环节由于 LLM 响应异常失败了，整个 Activity 会重试，导致 Writer 重新运行，这会白白浪费你在 n1n.ai 上的 Token 成本。正确的做法是将其解耦：

async def _call_fix_decomposed(self, sop_text, failures, human_feedback=""):
    # 第一阶段：Writer 智能体
    writer_result = await workflow.execute_activity(
        writer_task_activity,
        args=[sop_text, failures, human_feedback],
        start_to_close_timeout=timedelta(minutes=7),
        retry_policy=LLM_RETRY_POLICY,
    )

    # 第二阶段：Reviewer 智能体
    # 如果这里失败了，重试只会从 Reviewer 开始，不会重新运行 Writer
    reviewer_result = await workflow.execute_activity(
        reviewer_task_activity,
        args=[writer_result.text],
        start_to_close_timeout=timedelta(minutes=7),
        retry_policy=LLM_RETRY_POLICY,
    )

3. 实现“人工门控”

在企业级流程中，人工审批是一个硬性门控。利用 Temporal 的 Signal（信号）机制，我们可以让工作流无限期暂停。与轮询数据库不同，这是基于推送的，且不存在竞态条件。

@workflow.signal
def approve_step(self, feedback: str) -> None:
    self._step_feedback = feedback
    self._signal_received = True

# 在工作流循环中等待
await workflow.wait_condition(lambda: self._signal_received)

这种模式允许工作流在等待人工操作时（无论是几分钟还是几天），即便后端服务重启或迁移，也能精准地从暂停点恢复。状态被安全地存储在 Temporal 的持久化层中。

进阶版本控制：workflow.patched 模式

AI 智能体开发中最难的一点是在工作流运行中途更新提示词（Prompt）或智能体结构。如果你直接修改正在运行的代码，Temporal 的重放机制会发现新代码路径与历史记录不符，从而抛出错误。Temporal 通过显式补丁解决了这个问题：

if workflow.patched("split-writer-reviewer"):
    # 新逻辑：拆分后的智能体调用
    return await self._call_fix_decomposed(sop_text, failures, human_feedback)
else:
    # 旧逻辑：保持原有的单 Activity 调用，确保旧实例不崩溃
    return await workflow.execute_activity(fix_sop_with_crew_activity, ...)

结合 n1n.ai 实现集中式可靠性

通过 n1n.ai 统一路由你的 LLM 请求，你可以获得极高的 API 稳定性。结合 Temporal 的 RetryPolicy，你可以构建一个对网络波动几乎免疫的系统：

LLM_RETRY_POLICY = RetryPolicy(
    maximum_attempts=5,              # 最大重试 5 次
    initial_interval=timedelta(seconds=2),
    backoff_coefficient=2.0,         # 指数级退避
    maximum_interval=timedelta(seconds=60),
)

这套策略确保了如果 LLM 供应商出现短暂抖动，Temporal 会自动进行指数级退避重试，而你无需在业务逻辑中编写任何复杂的 try/except 块。

总结

构建真正可落地生产的 AI 智能体，需要跳出简单的“脚本循环”思维。通过 Temporal 管理状态，CrewAI 管理推理，你可以构建一个可持久、可审计且易于扩展的系统。而在底层 LLM 基础设施方面，n1n.ai 提供了支撑这些复杂工作流所需的卓越性能和稳定性。

立即在 n1n.ai 获取免费 API 密钥。