引言

Agent系统本质上是一个工作流编排系统——理解意图、检索知识、调用工具、评估结果、生成回复，每一步都是工作流中的一个节点。选择合适的工作流引擎，直接决定了Agent系统的可靠性、可观测性和开发效率。

2026年，工作流引擎领域已形成了清晰的格局。Temporal凭借其强大的状态管理和重试机制成为Agent系统的热门选择，Airflow在数据处理管道中依然占有一席之地，而自研引擎则在对性能和灵活性有极致要求的场景中仍有市场。

Agent工作流的特殊需求

与传统数据处理工作流不同，Agent工作流有其独特的需求特征：

需求维度传统工作流Agent工作流
执行时长分钟到小时秒到分钟
分支复杂度低（线性DAG）高（动态分支、循环）
人机交互罕见频繁（澄清、确认）
失败处理重试或告警重新规划、降级策略
状态管理简单复杂（对话历史、中间结果）
实时性批处理实时或近实时
动态性静态DAG运行时动态生成

三大方案深度对比

Temporal：Agent工作流的最佳搭档

from temporalio import workflow, activity
from datetime import timedelta

@activity.defn
async def understand_intent(user_input: str) -> dict:
    """意图理解活动"""
    # 调用LLM进行意图分类
    result = await llm_client.classify(user_input)
    return {
        "intent": result.intent,
        "confidence": result.confidence,
        "entities": result.entities
    }

@activity.defn
async def retrieve_memory(query: str, top_k: int = 5) -> list:
    """记忆检索活动"""
    memories = await vector_db.search(query, top_k=top_k)
    return [{"content": m.text, "score": m.score} for m in memories]

@activity.defn
async def execute_tool(tool_name: str, params: dict) -> dict:
    """工具执行活动"""
    tool = tool_registry.get(tool_name)
    result = await tool.run(**params)
    return result

@activity.defn
async def generate_response(prompt: str, context: dict) -> str:
    """响应生成活动"""
    response = await llm_client.generate(prompt, **context)
    return response

@workflow.defn
class AgentWorkflow:
    """Agent主工作流"""
    
    @workflow.run
    async def run(self, user_input: str) -> str:
        # Step 1: 意图理解
        intent_result = await workflow.execute_activity(
            understand_intent,
            user_input,
            start_to_close_timeout=timedelta(seconds=10),
            retry_policy=RetryPolicy(
                initial_interval=timedelta(seconds=1),
                maximum_interval=timedelta(seconds=10),
                maximum_attempts=3
            )
        )
        
        # 需要澄清时，等待用户输入
        if intent_result["confidence"] < 0.6:
            clarification = await workflow.wait_for_signal(
                "user_clarification",
                timeout=timedelta(minutes=5)
            )
            intent_result = await workflow.execute_activity(
                understand_intent,
                clarification,
                start_to_close_timeout=timedelta(seconds=10)
            )
        
        # Step 2: 并行检索记忆和执行工具
        memory_task = workflow.execute_activity(
            retrieve_memory,
            user_input,
            start_to_close_timeout=timedelta(seconds=5)
        )
        tool_task = workflow.execute_activity(
            execute_tool,
            intent_result["intent"],
            intent_result["entities"],
            start_to_close_timeout=timedelta(seconds=30)
        )
        
        memories, tool_results = await asyncio.gather(memory_task, tool_task)
        
        # Step 3: 生成响应
        prompt = build_prompt(user_input, memories, tool_results)
        response = await workflow.execute_activity(
            generate_response,
            prompt,
            {"intent": intent_result["intent"]},
            start_to_close_timeout=timedelta(seconds=15)
        )
        
        return response

Temporal的优势在Agent场景中极为突出：

  1. 内置状态持久化：工作流状态自动持久化，进程崩溃后可恢复
  2. 信号机制：原生支持人机交互（等待用户澄清）
  3. 精确重试：Activity级别重试，不影响已完成步骤
  4. 超时控制：多层级超时（start_to_close、schedule_to_close等）
  5. 可观测性：内置UI展示工作流执行历史

Airflow：传统但成熟

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.python_branch import BranchPythonOperator
from datetime import datetime, timedelta

def understand_intent(**context):
    user_input = context["params"]["user_input"]
    result = llm_client.classify(user_input)
    # 通过XCom传递结果
    context["task_instance"].xcom_push(key="intent", value=result)
    return result["intent"]

def route_by_confidence(**context):
    intent = context["task_instance"].xcom_pull(
        task_ids="understand_intent", key="intent"
    )
    if intent["confidence"] < 0.6:
        return "ask_clarification"
    return "execute_tools"

agent_dag = DAG(
    "agent_workflow",
    default_args={
        "owner": "agent-team",
        "retries": 3,
        "retry_delay": timedelta(seconds=5),
    },
    schedule_interval=None,  # 事件触发
    start_date=datetime(2026, 1, 1),
    catchup=False,
)

with agent_dag:
    intent_task = PythonOperator(
        task_id="understand_intent",
        python_callable=understand_intent,
        params={"user_input": "{{ dag_run.conf['user_input'] }}"},
    )
    
    branch_task = BranchPythonOperator(
        task_id="route_by_confidence",
        python_callable=route_by_confidence,
    )
    
    clarify_task = PythonOperator(
        task_id="ask_clarification",
        python_callable=lambda: print("Asking user..."),
    )
    
    tools_task = PythonOperator(
        task_id="execute_tools",
        python_callable=execute_tools,
    )
    
    response_task = PythonOperator(
        task_id="generate_response",
        python_callable=generate_response,
    )
    
    intent_task >> branch_task
    branch_task >> [clarify_task, tools_task]
    [clarify_task, tools_task] >> response_task

Airflow的局限在Agent场景中很明显：

  • XCom传递大量数据效率低
  • 不原生支持等待外部信号
  • 工作流定义是静态的，难以动态调整
  • 重试粒度是Task级别，不够精细

自研引擎：极致控制

class AgentWorkflowEngine:
    """轻量级自研工作流引擎"""
    
    def __init__(self):
        self.steps = {}
        self.state_store = StateStore()
    
    def step(self, name: str, timeout: float = 30.0, retries: int = 3):
        """注册工作流步骤"""
        def decorator(fn):
            self.steps[name] = WorkflowStep(
                name=name,
                handler=fn,
                timeout=timeout,
                max_retries=retries
            )
            return fn
        return decorator
    
    async def execute(self, workflow_id: str, initial_input: dict) -> dict:
        """执行工作流"""
        state = await self.state_store.load(workflow_id)
        
        if state is None:
            state = WorkflowState(
                workflow_id=workflow_id,
                current_step="start",
                context=initial_input,
                status="running"
            )
            await self.state_store.save(state)
        
        while state.current_step != "end" and state.status == "running":
            step = self.steps.get(state.current_step)
            if not step:
                raise ValueError(f"Unknown step: {state.current_step}")
            
            try:
                result = await asyncio.wait_for(
                    step.handler(state.context),
                    timeout=step.timeout
                )
                state.context.update(result)
                state.current_step = result.get("next_step", "end")
                
            except asyncio.TimeoutError:
                logger.warning(f"Step {step.name} timed out")
                step.retry_count += 1
                if step.retry_count >= step.max_retries:
                    state.status = "failed"
                    state.error = f"Step {step.name} exceeded max retries"
                # 指数退避
                await asyncio.sleep(2 ** step.retry_count)
            
            await self.state_store.save(state)
        
        return state.context

选型决策框架

                    ┌──────────────────────┐
                    │ 是否需要人机交互？     │
                    └──────┬───────┬───────┘
                         Yes      No
                           │       │
                    ┌──────▼──┐  ┌─▼──────────────┐
                    │Temporal │  │ 是否需要复杂DAG？│
                    └─────────┘  └──┬───────┬─────┘
                                  Yes      No
                                    │    ┌───▼────────────┐
                           ┌────────▼──┐ │ 是否对延迟有    │
                           │ Airflow或 │ │ 极致要求(<10ms)?│
                           │ Temporal  │ └──┬──────┬──────┘
                           └───────────┘   Yes     No
                                             │  ┌───▼──────┐
                                        ┌────▼┐ │ Temporal  │
                                        │自研 │ │ 或Airflow │
                                        │引擎 │ └──────────┘
                                        └─────┘

核心维度对比

维度TemporalAirflow自研引擎
开发效率⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐
运行性能⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐
状态持久化内置有限需自建
动态工作流原生支持不支持完全控制
人机交互Signal机制不支持需自建
可观测性UI+APIUI需自建
运维复杂度中（需Temporal Server）高（Scheduler+Worker+DB）低
生态成熟度高极高低
学习曲线中等陡峭N/A

迁移路径建议

从Airflow迁移到Temporal

# 迁移映射表
MIGRATION_MAP = {
    "DAG": "Workflow",
    "Task": "Activity",
    "XCom": "Workflow State (自动管理)",
    "BranchOperator": "Python条件分支",
    "Sensor": "Workflow.wait_for_signal()",
    "retry_delay": "RetryPolicy.initial_interval",
    "retries": "RetryPolicy.maximum_attempts",
    "execution_timeout": "start_to_close_timeout"
}

混合架构

在实际生产中，混合使用多种引擎也是一种有效策略：

class HybridOrchestrator:
    """混合编排器——按场景选择引擎"""
    
    async def route_workflow(self, request: dict) -> str:
        if request["type"] == "realtime_chat":
            # 实时对话——使用Temporal
            return await self.temporal_client.start_workflow(
                AgentChatWorkflow.run,
                request["input"],
                id=request["session_id"],
                task_queue="agent-realtime"
            )
        elif request["type"] == "batch_analysis":
            # 批量分析——使用Airflow
            return await self.airflow_client.trigger_dag(
                dag_id="batch_analysis",
                conf=request
            )
        elif request["type"] == "high_frequency_simple":
            # 高频简单任务——自研引擎
            return await self.custom_engine.execute(request)

总结

对于2026年的Agent系统，Temporal是工作流引擎的首选——它的状态持久化、信号机制和精确重试能力几乎完美匹配Agent工作流的需求。Airflow适合已有Airflow基础设施的团队用于批处理场景。自研引擎只在对延迟有极致要求（<10ms）或工作流逻辑极度定制化的场景下才值得投入。

无论选择哪种方案，核心原则是：工作流引擎应该解决Agent编排的复杂度，而不是成为新的复杂度来源。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。