Agent Frameworks Reason与Temporal的生存之道：面向智能体的持久执行方案

2026年，那些正在构建智能体系统的工程师们被问及他们所使用的框架时，几乎所有人都提到了LangGraph、CrewAI、AutoGen或这些技术的某种组合。这些工具本身并无优劣之分，问题在于整个讨论都发生在技术栈的错误层级上。

实际生产环境中的智能体系统总是呈现出某种重复的模式。凌晨3点惊醒值班工程师的，往往并非决定智能体行为逻辑的框架本身，而几乎总是同一类故障：某个流程在处理过程中突然终止；由于重试操作不具备幂等性，导致某一步被执行了两次；容器被驱逐时，正在运行的智能体任务在40分钟时中断且无法恢复；又或者工作流状态存储在Python变量中，而容器回收时该变量也随之消失。

LangGraph、CrewAI 和 AutoGen 均无法端到端处理该类任务，而这正是其设计初衷，因为它们的存在目的是协调大语言模型的调用与工具的使用。生产级智能体系统真正的问题在于缺乏稳定的执行能力，这是一个更为古老且发展缓慢的领域，比“智能体”这一概念的出现还要早大约十年。尽管拥有该技术的 Temporal 几乎未在智能体框架的讨论中出现，但它却已悄然成为越来越多严肃智能体应用背后的核心支撑。下文将介绍这三个流行框架的优势所在、各自抽象层的局限性，以及为何这个不太流行的第四种选择值得企业投入。

LangGraph

LangGraph的贡献在于将智能体转化为基于类型化状态的显式图结构，而非隐式的提示链。这具有实际价值，因为控制流变得可检查且可测试。例如，分诊工作流程可以如下所示：

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

class TriageState(TypedDict):
    ticket_id: str
    category: Literal["billing", "technical", "abuse", "unknown"]
    priority: int
    assigned_queue: str | None

def classify_node(state: TriageState) -> TriageState:
    category = llm_classify(state["ticket_id"])  # the only probabilistic step
    return {**state, "category": category}

def route_node(state: TriageState) -> str:
    # deterministic, testable, no LLM involved
    return {
        "billing": "billing_queue",
        "technical": "tech_queue",
        "abuse": "trust_and_safety_queue",
        "unknown": "human_review",
    }[state["category"]]

graph = StateGraph(TriageState)
graph.add_node("classify", classify_node)
graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route_node)
app = graph.compile()

这是个好主意，表述也很清晰。问题出现在某些内容需要存活的时间超过一次 invoke() 调用及其检查点之后。比如跨服务之间的协调、持续数天的工作流，或是 LangGraph 无法干预的执行步骤：Spark 作业、通过 Slack 处理并在数天后才返回结果的审批队列，或是由其他团队按自身节奏部署的服务。虽然检查点机制确实存在，但它仅适用于 LangGraph 自身的执行模型。一旦将任务交给与大语言模型无关的基础设施，持久性就不得不通过手动方式来实现。

CrewAI

只需几行代码即可定义一个多智能体团队：

from crewai import Agent, Task, Crew

researcher = Agent(role="Research Analyst", goal="Find relevant market data", tools=[search_tool])
writer = Agent(role="Report Writer", goal="Turn research into a client-ready summary")

crew = Crew(agents=[researcher, writer], tasks=[
    Task(description="Research {company}'s Q2 earnings", agent=researcher),
    Task(description="Write a 3-paragraph summary for the client", agent=writer),
])
crew.kickoff(inputs={"company": "Acme Corp"})

没有哪种工具能比CrewAI更快地在20分钟内将一个初步构想转化为多智能体工作流并生成可运行的演示版本。代价则体现在之后。它使用kickoff()函数来启动任务，却不存在暂停数小时或数天的有效方式，除了各工具自身的重试机制外没有其他重试功能，且除了开发者手动保存的数据外没有持久化状态。它仅能协调大语言模型的推理过程，并非真正的工作流引擎，虽然很多团队会无意中将其当作工作流引擎使用。

AutoGen

最简单的AutoGen配置是将助手与用户代理结合使用：

import autogen

assistant = autogen.AssistantAgent(name="assistant", llm_config={"config_list": config_list})
user_proxy = autogen.UserProxyAgent(name="user_proxy", human_input_mode="NEVER")
user_proxy.initiate_chat(assistant, message="Summarize this quarter's incident reports.")

AutoGen的对话式、自动生成风格非常适合探索性工作。其2026版本的工具界面更为结构化，采用了事件驱动模式，甚至倾向于将流程协调与对话功能分开。不过其底层仍然以消息记录的形式存在，而非持久化的结构化执行日志。当承载对话的进程终止时，留下的只是部分聊天记录，而非可以继续运行的计算结果。

共有的缺陷

这三种技术在新功能方面表现都很出色：能够让大语言模型进行推理、调用工具以及与其他由大语言模型驱动的智能体协作。但对于那些更基础的问题——即确保多步骤流程在出现崩溃、重启、超时或底层系统故障时仍能正常完成——它们并未试图解决。这并非它们的过错，期望它们做到这一点其实属于分类错误。

考虑一个实际的工作流程，而非演示场景：一名员工负责核对供应商发票。他首先从PDF文件中提取各项明细，将其与ERP系统中的采购订单进行比对，若差异超过设定阈值则会标记出来供人工审核，之后可能需要等待数天才能得到回复，之后才能将已批准的记录录入会计系统。出现错误时必须重新尝试，且不得重复录入数据。在流程进行到一半时重新部署系统，也不得导致流程从头开始或数据悄无声息地丢失。

┌────────────┐     ┌───────────────┐     ┌─────────────────┐     ┌───────────────┐
│  Extract   │ --> │ Match against │ --> │  Human approval │ --> │  Post to ERP  │
│  line items│     │ purchase order│     │  (can take days)│     │  (idempotent) │
└────────────┘     └───────────────┘     └─────────────────┘     └───────────────┘
      ▲                                          │
      └───────────── retry on failure ───────────┘

LangGraph、CrewAI或AutoGen的设计初衷并非保证在第三天重启容器时，当人们仍在评估第14项内容时该架构依然能正常运行。这并非关于大语言模型的编排，而是关乎持久执行能力；那些在这方面表现出色的系统——以Temporal为首——多年来一直承担着各种并不引人注目的后台工作，比如支付处理、订单履行以及基于Saga模式的微服务，早在“agent”这个词被用来描述这类程序之前就已经如此。

Temporal的工作原理

Temporal的模型看似简单，但实际上颇具巧思。工作流以普通代码中的函数形式编写，包含循环、条件判断和函数调用，Temporal会将该函数的执行历史持久化保存。如果流程在某个时刻崩溃，Temporal会从已保存的事件历史中重新执行该函数直至崩溃点，然后以确定性的方式继续运行，而不会重复触发副作用。其精妙之处在于，诸如调用大型语言模型、访问API或向数据库写入数据之类的副作用都被封装在被称为“活动”的独立函数中。Temporal会以可配置的延迟策略重新尝试这些活动，而工作流函数仅负责协调它们。

下面的发票对账代理将大型语言模型的调用视为多个活动中的一个。首先，这些与外部世界交互的活动包括：

# activities.py — the parts that touch the outside world (LLMs, APIs, DBs)
from temporalio import activity
from dataclasses import dataclass

@dataclass
class Invoice:
    invoice_id: str
    vendor: str
    line_items: list[dict]

@activity.defn
async def extract_line_items(pdf_url: str) -> list[dict]:
    # calls a vision-capable LLM, or a dedicated OCR model — doesn't matter to the workflow
    text = await download_and_ocr(pdf_url)
    return await llm_extract_structured(text, schema=LINE_ITEM_SCHEMA)

@activity.defn
async def match_purchase_order(invoice: Invoice) -> dict:
    po = await erp_client.get_purchase_order(invoice.vendor)
    discrepancies = diff_line_items(invoice.line_items, po["line_items"])
    return {"po_id": po["id"], "discrepancies": discrepancies}
@activity.defn
async def post_to_erp(invoice_id: str, po_id: str) -> dict:
    # idempotency is enforced with a deterministic key derived from workflow inputs,
    # not a random UUID generated per attempt — this is the same lesson as always
    return await erp_client.post_journal_entry(
        idempotency_key=f"invoice-post-{invoice_id}",
        invoice_id=invoice_id,
        po_id=po_id,
    )

@activity.defn
async def notify_human_reviewer(invoice_id: str, discrepancies: list[dict]) -> None:
    await slack_client.post_message(
        channel="#ap-review",
        text=f"Invoice {invoice_id} has {len(discrepancies)} discrepancies needing review.",
    )

接下来是工作流，它承担了LangGraph或CrewAI通常所扮演的协调角色，只不过它的稳定性是内置的，无需依赖检查点机制：

# workflows.py
from temporalio import workflow
from datetime import timedelta

@workflow.defn
class InvoiceReconciliationWorkflow:
    def __init__(self) -> None:
        self._approval_received = False
        self._approved = False

    @workflow.signal
    async def submit_human_decision(self, approved: bool) -> None:
        # A signal is how the outside world talks to a running, possibly-days-old workflow
        self._approval_received = True
        self._approved = approved

    @workflow.run
    async def run(self, pdf_url: str, invoice_id: str, vendor: str) -> dict:
        line_items = await workflow.execute_activity(
            extract_line_items, pdf_url,
            start_to_close_timeout=timedelta(minutes=2),
            retry_policy=workflow.RetryPolicy(maximum_attempts=3),
        )

        invoice = Invoice(invoice_id=invoice_id, vendor=vendor, line_items=line_items)
        match_result = await workflow.execute_activity(
            match_purchase_order, invoice,
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=workflow.RetryPolicy(maximum_attempts=5),
        )

        if match_result["discrepancies"]:
            await workflow.execute_activity(
                notify_human_reviewer, invoice_id, match_result["discrepancies"],
                start_to_close_timeout=timedelta(seconds=10),
            )
            # This is the line that would be genuinely hard to write safely in the
            # other three frameworks: wait, durably, for up to 7 days, for a signal
            # that might arrive from a Slack button, a web form, anything —
            # while the workflow's entire state sits safely in Temporal's history,
            # costing zero compute while it waits.
            await workflow.wait_condition(
                lambda: self._approval_received,
                timeout=timedelta(days=7),
            )
            if not self._approved:
                return {"status": "rejected", "invoice_id": invoice_id}

        result = await workflow.execute_activity(
            post_to_erp, invoice_id, match_result["po_id"],
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=workflow.RetryPolicy(maximum_attempts=5),
        )
        return {"status": "posted", "invoice_id": invoice_id, "erp_result": result}

有几点细节值得注意，因为它们正是导致智能体系统表现不稳定的故障模式，而这些其实是运行时的固有属性，并非需要人们特意去添加的功能。

七天的等待是免费的，且能抵御一切情况。 workflow.wait_condition不会让任何线程、进程或容器持续运行超过一周。Temporal会将工作流的状态存储在自身持久化的存储中，因此在此期间工作者可能会被重新部署、缩容至零或完全崩溃。当信号最终到达时，Temporal会回放历史记录并从那个await点继续执行。若要在LangGraph检查点器或CrewAI团队系统中实现相同功能，基本上就意味着要重新构建Temporal的事件溯源模型。

重试是自动进行的，范围有限，且绝不会重复已完成的工作。假设在match_purchase_order成功执行后post_to_erp失败了。Temporal不会重新运行match_purchase_order，因为其结果已经保存在持久历史记录中。以下是将这一理念的简单理解与正确做法进行对比：

# BEFORE — naive retry loop, the kind you write at 11pm under deadline pressure
async def reconcile_invoice_naive(pdf_url, invoice_id, vendor):
    for attempt in range(3):
        try:
            line_items = await extract_line_items(pdf_url)
            match_result = await match_purchase_order(Invoice(invoice_id, vendor, line_items))
            if match_result["discrepancies"]:
                await notify_human_reviewer(invoice_id, match_result["discrepancies"])
                approved = await wait_for_approval_somehow(invoice_id)  # good luck
                if not approved:
                    return {"status": "rejected"}
            return await post_to_erp(invoice_id, match_result["po_id"])
        except Exception:
            continue  # retries EVERYTHING from the top, including re-extracting the PDF

那种简单的实现方式会在每次重试时都重复进行大语言模型提取操作，为本来已经成功的步骤浪费令牌和金钱；而如果需要让某个功能在重新部署后依然存在，wait_for_approval_somehow也找不到合适的解决方案。Temporal的版本则无需通过额外代码来解决这些问题，由于其执行结构的设计，这些问题根本不会出现。

纯Python语言意味着代码可测试。Temporal配备了测试框架，能够在单元测试中以毫秒级时间跳过timedelta(days=7)的等待时间，而且无需连接网络即可对各种功能进行模拟：

import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

@pytest.mark.asyncio
async def test_rejected_invoice_does_not_post_to_erp():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client, task_queue="test-queue",
            workflows=[InvoiceReconciliationWorkflow],
            activities=[extract_line_items, match_purchase_order, notify_human_reviewer, post_to_erp],
        ):
            handle = await env.client.start_workflow(
                InvoiceReconciliationWorkflow.run,
                args=["fake.pdf", "INV-001", "Acme"],
                id="test-invoice-001",
                task_queue="test-queue",
            )
            await handle.signal(InvoiceReconciliationWorkflow.submit_human_decision, False)
            result = await handle.result()
            assert result["status"] == "rejected"

注意其中缺失的部分：没有真正的七天睡眠测试，没有对检查点内部机制的深入探究，也没有关于重启的手动推理过程。耐久性属于支撑性基础设施，并非每次测试都需要重新验证的要素。

同时使用这两层技术

公平的解决方案并非舍弃 LangGraph 而选择 Temporal。二者作用于不同的层级，2026 年最优秀的实际应用方案正是将它们结合使用：在单一步骤中利用 LangGraph 或 CrewAI 进行推理，再通过 Temporal 构建连接多日、多服务流程的工作流层。某个任务可以这样封装一个有界推理步骤：

@activity.defn
async def run_agent_reasoning_step(ticket_text: str) -> dict:
    # LangGraph handles the bounded, single-turn reasoning —
    # classify, decide, draft — inside an activity that Temporal
    # retries and times out like any other side effect.
    result = triage_graph.invoke({"ticket_id": ticket_text, "audit_log": []})
    return {"category": result["category"], "priority": result["priority"]}

@workflow.defn
class SupportTicketWorkflow:
    @workflow.run
    async def run(self, ticket_id: str, ticket_text: str) -> dict:
        triage = await workflow.execute_activity(
            run_agent_reasoning_step, ticket_text,
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=workflow.RetryPolicy(maximum_attempts=3),
        )
        if triage["priority"] >= 8:
            await workflow.execute_activity(page_oncall_engineer, ticket_id, ...)
            await workflow.wait_condition(lambda: self._acknowledged, timeout=timedelta(minutes=15))
        # ... routing, SLA timers, escalation chains — all durable, all testable

对于那些代理系统需要承受生产环境压力的团队，这正是推荐的架构。基于大语言模型的原生框架能够处理其擅长的任务，即在有限步骤内进行推理；而稳定的执行引擎则负责其设计初衷的任务，确保整个流程——无论持续数分钟还是数周，也无论涉及大语言模型调用还是人工审批——都能正确完成或明确失败。两者无需互相替代对方的功能。

时间成本过高的场景

Temporal存在诸多缺陷，因此并不适合作为大多数智能体任务的默认选择。它需要额外维护一个分布式系统，无论是自托管服务器还是Temporal Cloud服务，再加上工作进程和任务队列，其基础设施复杂度远高于运行LangGraph的Python进程。此外，工作流代码还必须遵循严格的确定性规则：直接调用datetime.now()、使用未经控制的随机性元素，或在工作流函数中直接进行I/O操作都是被禁止的。团队需要时间来适应这些规则，而违反规则则会在重放时引发难以理解的非确定性错误。对于那些只需单个进程即可完成的短任务，比如通过一次LLM调用来回答问题或总结文档，持久化执行其实只是不必要的负担，因为根本不存在可能因崩溃而受损的多步骤流程。

决策规则

当工作流可能持续超过单个进程的生命周期、需要在不同服务间协调或等待人工处理，或是涉及那些绝不能被重复处理的对象——如资金、库存或不应被多次发送的邮件时，应选择耐用的执行方式作为核心。让首选的LLM框架在其某个步骤中作为组件运行即可。如果所有操作都在一个请求-响应周期内完成，那就无需使用它，因为添加这一组件只会解决尚不存在的问题。

许多关于“该使用哪种智能体框架”的讨论都忽视了一个更重要的问题：系统的哪一层真正需要加强防护？LangGraph、CrewAI和AutoGen正在就如何构建大语言模型推理机制及多智能体协作方式进行有益的探讨。然而，智能体系统能否在部署失败、网络分区，或是有人需要三天时间才点击“批准”时依然正常运行，这是另一个问题；实际上，在“智能体AI”这一概念出现之前，工作流引擎就已经通过实现可靠的执行功能解决了这个问题。那些流行的框架只是调整了演示中可见的部分，而那个被忽视的要素则决定了一个月后系统是否依然能够正常运转。