Temporal AI 工作流编排 — 可靠的多步骤 AI 流水线

TL;DR #

Temporal 是一个持久化执行平台，让构建可靠的 AI 工作流变得极其简单。无需与 Kubernetes CronJob、死信队列和手动重试逻辑搏斗，你只需将 Python 函数装饰为 Temporal 的 workflow 和 activity。Temporal 保证恰好一次执行、自动指数退避重试和开箱即用的完整可观测性。

Temporal 是什么？ #

Temporal 是用于以规模运行容错工作流的开源分布式系统。其核心提供持久化执行——你的代码在 Temporal 管理的基础设施中运行，它自动处理故障、重试、检查点和状态持久化。

对于 AI 工作负载，这意味着：

因限流而失败的 LLM 推理调用会自动重试并退避
多步微调流水线在容器崩溃后仍能存活且不会丢失进度
Agent 编排中每一步的输出都被持久化并可检查
训练任务在 GPU 故障后从上次检查点恢复

传统 AI 编排的问题 #

考虑一个典型的 AI 流水线：

[加载数据] → [预处理] → [文档嵌入] → [向量库索引] → [测试检索] → [通知团队]

使用传统工具（Airflow、Celery、cron 脚本），每个步骤都需要：

网络超时的自定义错误处理
故障恢复的手动检查点
分布式 worker 间的状态管理
调试的可观测性仪表板

Temporal 通过让你的 Python 代码天然可恢复来消除所有这些。如果第 3 步崩溃，Temporal 仅重启第 3 步并使用完全相同的输入——第 1-2 步从历史记录重放。

Temporal vs 替代方案 #

特性	Temporal	Airflow	Celery + Redis	Kubernetes CronJob
代码即工作流定义	✅（Python 装饰器）	❌（DAG YAML/Python）	❌（仅是任务队列）	❌（Shell 脚本）
自动重试	✅（可配置策略）	⚠️（基础）	⚠️（手动配置）	❌（无）
状态持久化	✅（内置）	⚠️（外部数据库）	❌（内存中）	❌
恰好一次语义	✅	❌	❌	❌
交互式调试	✅（Web UI + CLI）	⚠️（有限）	❌	❌
ML 友好集成	✅（原生）	⚠️（插件）	❌	❌

快速开始 #

第一步：安装 Temporal 栈 #

# 选项 A：Docker Compose（本地开发推荐）
git clone https://github.com/temporalio/docker-compose.git
cd docker-compose
docker compose up -d

# 选项 B：Temporal Cloud（托管，无需管理基础设施）
# 在 cloud.temporal.io 注册并创建命名空间

# 验证服务器正在运行
temporal cluster health

默认 Docker Compose 设置包括：

Temporal Server（gRPC API + 历史）
Temporal UI（localhost:8233）
Elasticsearch（搜索/索引）
Temporal Frontend（端口 7233）

第二步：安装 Python SDK #

pip install temporalio

第三步：第一个 Workflow #

import asyncio
from temporalio import worker, workflow, activity
from temporalio.client import Client
from temporalio.common import RetryPolicy

# 定义 activity（各个步骤）
@activity.defn
async def load_dataset(dataset_name: str):
    """加载并验证数据集。"""
    print(f"正在加载数据集: {dataset_name}")
    data = {"samples": 10000, "features": 128}
    activity.info(f"已加载 {data['samples']} 个样本")
    return data

@activity.defn
async def preprocess(data: dict):
    """清洗和规范化数据。"""
    print("正在预处理数据...")
    processed = {
        "cleaned_samples": data["samples"],
        "normalized": True,
        "feature_count": data["features"]
    }
    return processed

@activity.defn
async def train_model(preprocessed_data: dict, epochs: int = 10):
    """在预处理数据上训练模型。"""
    print(f"正在训练模型 {epochs} 轮...")
    metrics = {
        "final_loss": 0.0234,
        "final_accuracy": 0.9456,
        "epochs_trained": epochs
    }
    activity.info(f"训练完成: accuracy={metrics['final_accuracy']:.4f}")
    return metrics

@activity.defn
async def deploy_model(metrics: dict):
    """将训练的模型部署到生产环境。"""
    print("正在将模型部署到生产环境...")
    deployment = {
        "model_id": f"model-{metrics['final_accuracy']:.4f}",
        "status": "deployed",
        "endpoint": "https://api.example.com/v1/predict"
    }
    activity.info(f"模型已部署: {deployment['model_id']}")
    return deployment

# 定义 workflow
@workflow.defn
class MLTrainingPipeline:
    @workflow.run
    async def run(self, dataset_name: str, epochs: int = 10) -> dict:
        # 每个步骤是一个 activity 调用
        data = await workflow.execute_activity(
            load_dataset,
            dataset_name,
            retry=RetryPolicy(max_attempts=3)
        )
        
        processed = await workflow.execute_activity(
            preprocess,
            data,
            retry=RetryPolicy(max_attempts=2)
        )
        
        metrics = await workflow.execute_activity(
            train_model,
            processed,
            epochs,
            retry=RetryPolicy(max_attempts=3, initial_interval=10)
        )
        
        deployment = await workflow.execute_activity(
            deploy_model,
            metrics,
            retry=RetryPolicy(max_attempts=2)
        )
        
        return deployment

第四步：运行 Worker 和 Client #

# worker.py
import asyncio
from temporalio.worker import Worker
from my_workflow import MLTrainingPipeline, load_dataset, preprocess, train_model, deploy_model

async def main():
    worker = Worker(
        client,  # Temporal Client 实例
        task_queue="ml-pipeline",
        workflows=[MLTrainingPipeline],
        activities=[load_dataset, preprocess, train_model, deploy_model]
    )
    print("Worker 已启动。按 Ctrl+C 退出。")
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

AI 专用工作流模式 #

模式一：带回退的 LLM 链 #

链式多个 LLM 调用，自动回退到更便宜的模型：

from temporalio import workflow, activity

@activity.defn
async def generate_with_gpt4(prompt: str) -> str:
    """先尝试 GPT-4。"""
    response = await call_openai(prompt, model="gpt-4o")
    return response

@activity.defn
async def generate_with_claude(prompt: str) -> str:
    """回退到 Claude。"""
    response = await call_anthropic(prompt, model="claude-sonnet-4")
    return response

@activity.defn
async def generate_with_local(prompt: str) -> str:
    """最后手段：本地模型。"""
    response = await call_ollama(prompt, model="llama3.2")
    return response

@workflow.defn
class ResilientLLMChain:
    @workflow.run
    async def run(self, prompt: str) -> dict:
        try:
            result = await workflow.execute_activity(
                generate_with_gpt4, prompt,
                timeout=timedelta(minutes=5),
                retry=RetryPolicy(max_attempts=2)
            )
            model_used = "gpt-4o"
        except Exception:
            try:
                result = await workflow.execute_activity(
                    generate_with_claude, prompt,
                    timeout=timedelta(minutes=5),
                    retry=RetryPolicy(max_attempts=2)
                )
                model_used = "claude-sonnet-4"
            except Exception:
                result = await workflow.execute_activity(
                    generate_with_local, prompt,
                    timeout=timedelta(minutes=10),
                    retry=RetryPolicy(max_attempts=3)
                )
                model_used = "local-llama"
        
        return {"response": result, "model_used": model_used, "fallback_chain": True}

模式二：异步多 Agent 编排 #

并行运行多个 AI Agent，然后聚合结果：

@activity.defn
async def agent_research(query: str) -> dict:
    """研究 Agent：从网络收集信息。"""
    results = await search_web(query)
    return {"type": "research", "sources": len(results), "summary": summarize(results)}

@activity.defn
async def agent_analysis(research_data: dict) -> dict:
    """分析 Agent：评估发现。"""
    analysis = await analyze_findings(research_data["summary"])
    return {"type": "analysis", "confidence": analysis["confidence_score"]}

@activity.defn
async def agent_synthesis(research: dict, analysis: dict) -> dict:
    """综合 Agent：将研究和综合分析成报告。"""
    report = await synthesize_report(research, analysis)
    return {"type": "synthesis", "report_length": len(report)}

@workflow.defn
class MultiAgentResearch:
    @workflow.run
    async def run(self, query: str) -> dict:
        research_handle = workflow.execute_activity(
            agent_research, query, start_to_close_timeout=timedelta(minutes=5)
        )
        
        research_result = await research_handle
        analysis_handle = workflow.execute_activity(
            agent_analysis, research_result, start_to_close_timeout=timedelta(minutes=3)
        )
        
        analysis_result = await analysis_handle
        
        final_report = await workflow.execute_activity(
            agent_synthesis, research_result, analysis_result,
            start_to_close_timeout=timedelta(minutes=5)
        )
        
        return final_report

模式三：带检查点恢复的 ML 训练 #

在任何故障后自动从上次检查点恢复训练：

@activity.defn
async def save_checkpoint(epoch: int, model_state: dict) -> str:
    """将训练检查点保存到持久化存储。"""
    checkpoint_path = f"s3://my-bucket/checkpoints/epoch_{epoch}.pt"
    await upload_to_s3(model_state, checkpoint_path)
    activity.info(f"检查点已保存: {checkpoint_path}")
    return checkpoint_path

@activity.defn
async def load_checkpoint(checkpoint_path: str) -> dict:
    """从检查点加载模型状态。"""
    model_state = await download_from_s3(checkpoint_path)
    activity.info(f"检查点已加载: {checkpoint_path}")
    return model_state

@workflow.defn
class ResumableTraining:
    @workflow.run
    async def run(self, dataset_url: str, total_epochs: int, lr: float = 0.001) -> dict:
        checkpoint_path = workflow.info().get_memo_field("last_checkpoint")
        
        if checkpoint_path:
            model_state = await workflow.execute_activity(
                load_checkpoint, checkpoint_path,
                start_to_close_timeout=timedelta(minutes=2)
            )
            start_epoch = int(checkpoint_path.split("_")[-1].split(".")[0])
            activity.info(f"从第 {start_epoch} 轮恢复")
        else:
            model_state = initialize_model(dataset_url)
            start_epoch = 0
        
        for epoch in range(start_epoch, total_epochs):
            result = await workflow.execute_activity(
                train_epoch, model_state, epoch, lr,
                start_to_close_timeout=timedelta(minutes=30),
                retry=RetryPolicy(max_attempts=3, backoff_coefficient=2.0)
            )
            
            model_state = result["state"]
            
            if (epoch + 1) % 5 == 0:
                cp_path = await workflow.execute_activity(
                    save_checkpoint, epoch + 1, model_state,
                    start_to_close_timeout=timedelta(minutes=5)
                )
                workflow.set_memo({"last_checkpoint": cp_path})
        
        return {"final_state": model_state, "total_epochs": total_epochs}

模式四：流式 LLM 输出 #

在 workflow 中处理 LLM 的流式响应：

@activity.defn
async def stream_llm_response(prompt: str, max_tokens: int = 1024) -> list[str]:
    """从 LLM 流式传输 token 并以列表返回。"""
    tokens = []
    async for token in call_streaming_api(prompt, max_tokens):
        tokens.append(token)
        await asyncio.sleep(0.01)
    return tokens

AI 工作流的高级功能 #

基于信号的工作流控制 #

从外部信号工作流以取消、更新优先级或注入新数据：

@workflow.defn
class PriorityWorkflow:
    def __init__(self):
        self.priority = "normal"
        self.cancel_requested = False
    
    @workflow.signal
    def set_priority(self, new_priority: str):
        self.priority = new_priority
        workflow.logger.info(f"优先级已更改为 {new_priority}")
    
    @workflow.signal
    def cancel_workflow(self):
        self.cancel_requested = True
        workflow.logger.info("取消请求已发送")
    
    @workflow.run
    async def run(self, task_data: dict) -> dict:
        while not self.cancel_requested:
            result = await process_task(task_data, self.priority)
            await asyncio.sleep(0.1)
        return {"status": "cancelled", "partial_result": result}

子工作流实现模块化设计 #

将复杂流水线分解为嵌套子工作流：

@workflow.defn
class DataPreparation:
    @workflow.run
    async def run(self, raw_data: dict) -> dict:
        cleaned = await workflow.execute_activity(clean_data, raw_data)
        validated = await workflow.execute_activity(validate_data, cleaned)
        return validated

@workflow.defn
class FullMLPipeline:
    @workflow.run
    async def run(self, raw_data: dict, model_config: dict) -> dict:
        prepared_data = await workflow.child_execute(DataPreparation.run, raw_data)
        trained_model = await workflow.child_execute(ModelTraining.run, prepared_data, model_config)
        eval_results = await workflow.child_execute(ModelEvaluation.run, trained_model)
        return eval_results

查询工作流状态 #

检查运行中的工作流而不停止它们：

client = await Client.connect("localhost:7233")
handle = client.get_workflow_handle("training-job-001")

state = await handle.query(lambda wf: wf.current_state)
print(f"当前状态: {state}")

info = await handle.describe()
print(f"状态: {info.status}")
print(f"开始时间: {info.start_time}")

监控和调试 #

Temporal Web UI #

在 http://localhost:8233 访问内置 Web UI：

查看所有运行中和已完成的工作流
检查每个 activity 的输入/输出数据
逐步重放工作流历史
按 ID、状态或自定义属性搜索工作流

CLI 调试 #

# 列出所有工作流
temporal workflow list --namespace default

# 描述特定工作流
temporal workflow describe --workflow-id training-job-001

# 显示工作流历史（执行跟踪）
temporal workflow show --workflow-id training-job-001

# 在工作流中重置到特定点
temporal workflow reset --workflow-id training-job-001 --reset-point LastAutoClose

# 终止运行中的工作流
temporal workflow terminate --workflow-id training-job-001 --reason "用户请求"

结构化日志 #

import structlog
from temporalio import activity

logger = structlog.get_logger()

@activity.defn
async def train_with_logging(model_config: dict) -> dict:
    logger.info("training_start", config=model_config)
    
    for epoch in range(10):
        loss = perform_training_epoch(model_config)
        logger.info("epoch_complete", epoch=epoch, loss=loss, learning_rate=model_config["lr"])
    
    logger.info("training_complete", final_loss=loss)
    return {"final_loss": loss}

日志出现在 Temporal UI 中，并可导出到 Elasticsearch、Datadog 或任何 SIEM。

成本优化 #

长运行任务的 Activity Heartbeat #

通过报告进度防止浪费计算：

@activity.defn
async def long_training_job(config: dict):
    for epoch in range(100):
        activity.heartbeat(f"第 {epoch}/100 轮完成")
        loss = train_one_epoch(config)
    return {"final_loss": loss}

右侧大小 Worker 资源 #

worker = Worker(
    client, task_queue="ml-workers",
    workflows=[MLTrainingPipeline],
    activities=[train_model, evaluate_model],
    max_concurrent_activities=50,
    max_concurrent_workflow_tasks=100,
)

成本对比 #

方案	月成本（每月 100 个训练任务）	运维开销
Kubernetes + CronJob	$800（常驻节点）+ 20 小时/月 DevOps	高
AWS Batch	$450（抢占式实例）+ 10 小时/月配置	中
Temporal Cloud	$200（计算）+ $0 运维	无
自托管 Temporal	$150（2 台小 VM）+ 5 小时/月维护	低

未来方向 #

Temporal 的 AI 路线图 #

Temporal 正在积极构建 AI 专用功能：

原生 LLM activity 模板：常见 LLM 操作（聊天、补全、嵌入）的预构建 activity，内置重试和限流处理
向量记忆：内置向量存储以跨执行持久化工作流上下文
Agent SDK：第一类多 Agent 编排支持，含共享内存和通信协议
GPU 感知调度：与 GPU 集群的原生集成以优化 ML 工作负载
Temporal Studio 增强：实时工作流可视化，带 ML 指标叠加

何时使用 Temporal #

选择 Temporal 当：

AI 流水线有多个依赖步骤
你需要保证执行（崩溃时不丢失任务）
你想交互式调试工作流
你的团队重视 Python 原生开发
你需要复杂模式（重试、超时、并行、子工作流）

考虑更简单的替代方案当：

你有单步任务——直接使用 cron 或 API 调用
你需要实时流式传输——Temporal 是批处理导向的
你更喜欢可视化 DAG 编辑器——考虑 Apache Airflow
你已深度投入 AWS Step Functions——原生集成可能更简单

社区动态 #

工作流编排领域持续演进。2026 年的 notable developments 包括：

Temporal Cloud 扩展到 5 个区域，含 GPU 优化的 worker 节点
开源 Temporal 添加对 Python 3.12 和 PyPy 的原生支持
社区集成：LangChain、LlamaIndex 和 CrewAI 都发布了官方 Temporal 连接器
企业采用：Scale AI 和 Hugging Face 等主要 AI 公司使用 Temporal 进行生产 ML 流水线

Temporal 社区已增长到超过 50,000 GitHub star，来自构建生产 AI 系统的公司有活跃贡献。生态系统包括流行 ML 框架的连接器、监控集成和常见 AI 工作流模式的模板仓库。

FAQ #

Q: Temporal 如何处理 LLM 限流？ #

使用 Temporal 的重试策略配合指数退避。配置 initial_interval、maximum_interval 和 backoff_coefficient 以实现礼貌的重试策略：

retry=RetryPolicy(
    initial_interval=timedelta(seconds=1),
    maximum_interval=timedelta(minutes=5),
    backoff_coefficient=2.0,
    maximum_attempts=5
)

这自然地在遇到限流时节流请求，不同于盲目冲击 API 的简单重试循环。

Q: 我可以在抢占式实例上运行 Temporal worker 吗？ #

可以。Temporal 的架构为此设计。worker 可以随时来去——如果 worker 在 activity 中途死亡，Temporal 检测心跳超时并在另一个可用 worker 上重新调度该 activity。这使得 Temporal 非常适合成本优化的抢占式实例部署。

Q: 如何在 Temporal 中处理 LLM 流式输出？ #

虽然 Temporal activity 传统上是请求-响应模式，但你可以使用流式模式：在 activity 执行期间在内存中收集流式 token，然后返回完整结果。对于端到端的真实流式传输，结合 Temporal（用于工作流持久性）与轮询工作流状态的 WebSocket 端点。

Q: Temporal 工作流的最大持续时间是多少？ #

Temporal 工作流可以无限期运行——没有硬性超时。有记录的最长 Temporal 工作流连续运行了 14 个月，处理了数百万事件。 practical 起见，在 individual activity 上设置合理的超时，并对长运行操作使用心跳。

Q: Temporal 能与无服务器 GPU（Modal、RunPod）一起使用吗？ #

可以。Temporal worker 可以在任何地方运行——EC2、GKE、EKS 甚至 serverless container。将 Temporal worker 与 Modal function 或 RunPod 实例一起部署。关键洞察：Temporal 管理工作流协调，而实际的 GPU 计算在成本最低的地方发生。

参考资料 #

加入我们的 Telegram 群组获取实时 AI 工具讨论和部署技巧：t.me/dibi8