Temporal Replay 2026 深度解析：Serverless Workers、Workflow Streams 与持久化执行新范式

2026年5月，Temporal 在 Replay 2026 大会上发布了重磅更新，涵盖 Serverless Workers、Standalone Activities、Workflow Streams 等多项核心能力。本文从架构原理、代码实战、生产避坑三个维度，一次把 Temporal 这套"失败不存在的"分布式持久化执行引擎讲透。

一、背景：为什么分布式系统需要一个"反脆弱"执行引擎

1.1 分布式系统的根本困境

每一个写过后端服务的程序员，都逃不开这几个噩梦般的场景：

• 网络抖动：HTTP 请求超时，调用方不知道服务端到底处理了没有
• 进程崩溃：Worker 处理到一半，状态全部丢失，必须手动补偿
• 幂等地狱：为了"安全重试"，同一个逻辑写三遍 if (alreadyDone) return;
• 分布式事务：两阶段提交、TCC、Saga，各种模式学了一圈，还是漏洞百出

传统方案是外部化状态：把状态存 Redis、MySQL、消息队列，然后用各种锁、版本号、幂等键来兜底。这套方案能work，但代价是代码复杂度爆炸——一个简单的"调用三方API → 写库 → 发通知"流程，补偿逻辑可能比主逻辑还长。

Durable Execution（持久化执行） 是一种截然不同的思路：不再把状态外包给外部存储，而是让程序本身的状态天然持久化。程序运行到哪一步、局部变量的值是什么、下一步该从哪恢复——这些信息全部由执行引擎自动管理，进程崩溃后可以从断点无缝继续，无需任何补偿代码。

Temporal 就是 Durable Execution 领域最成熟、最广泛使用的工业级实现。

1.2 Temporal 的核心抽象

在深入 Replay 2026 的新特性之前，先梳理 Temporal 的三大核心概念，这是一切的基础：

[Client] --start workflow--> [Temporal Server] --dispatch activity--> [Worker Pool]
                                      |
                                      +-- persist workflow state
                                      +-- replay on failure

关键洞察：Workflow 代码本身是确定性的（deterministic）。Temporal 在每个 checkpoint 记录执行路径，崩溃后通过**事件溯源（Event Sourcing）**重放来恢复状态，而不是重新跑一遍带副作用的代码。这就是为什么 Activity 专门负责"脏活"（有副作用的调用），而 Workflow 只负责编排。

二、Replay 2026 四大核心新特性深度拆解

2.1 Serverless Workers：让 Temporal Worker 按需扩缩容到零

2.1.1 痛点：传统 Worker 的运维负担

在 Replay 2026 之前，部署 Temporal Worker 的标准姿势是：

1. 准备长期运行的进程（VM、容器、K8s Deployment）
2. 配置 HPA（Horizontal Pod Autoscaler）根据负载扩缩容
3. 设置优雅关闭（graceful shutdown），处理 in-flight 请求
4. 预留容量峰值对应的资源

这套模式的问题是：资源成本和运维复杂度都固定在那里了。哪怕凌晨三点没有任务，Worker 进程也要跑着，空耗资源。同时，冷启动时间（从零启动一个新 Worker 实例）也是运维必须考虑的因素。

2.1.2 架构原理：AWS Lambda 上的 Temporal Worker

Serverless Workers 的思路很直接：把 Worker 的生命周期管理完全托管给 Temporal Cloud，开发者只需要写业务代码。

[Temporal Cloud] <---> [AWS Lambda] <---> [Your Activity Code]
     |
     +-- 接收任务调度指令
     +-- 按需触发 Lambda invocation
     +-- 管理并发度
     +-- 优雅关闭（超时未完成则交接）

具体流程：

1. 开发者将 Activity 代码打包为 Lambda 函数（目前仅支持 AWS Lambda）
2. 在 Temporal Cloud 注册 Serverless Worker，指定 Lambda ARN 和并发配置
3. Temporal Cloud 根据 Task Queue 的 backlog 压力，自动发起 Lambda invocations
4. 当没有待处理任务时，Lambda 调用次数自动降为 0——真正的零成本空闲

2.1.3 代码示例：3分钟完成一个 Serverless Activity 部署

# activity.py — 标准 Temporal Activity，无需修改
from temporalio import activity
from dataclasses import dataclass
import httpx

@dataclass
class FetchWeatherInput:
    city: str

@activity.defn
async def fetch_weather(input: FetchWeatherInput) -> dict:
    activity.logger.info(f"Fetching weather for {input.city}")
    
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.weather.example.com/{input.city}",
            timeout=10.0
        )
        resp.raise_for_status()
        return resp.json()

# worker.py — Serverless Worker 配置（仅需3行关键配置）
from temporalio.contrib.serverless import ServerlessWorkerConfig

config = ServerlessWorkerConfig(
    activity_handlers=[fetch_weather],
    task_queue="weather-queue",
    # Temporal Cloud 自动管理 Lambda 扩缩容
    lambda_config={
        "function_name": "weather-activity",
        "memory_mb": 512,
        "timeout_seconds": 30,
    }
)

# 注意：Serverless 模式下，worker.run() 不阻塞
# Temporal Cloud 通过 Event Bridge 触发 Lambda

部署命令：

temporal serverless deploy \
  --activity activity.py \
  --function weather-activity \
  --region us-east-1 \
  --memory 512mb

使用场景建议：

• 适合 低频、有波动、突发流量 的 Activity（数据同步、报表生成、通知发送）
• 不适合 超低延迟 要求（Lambda cold start ~100-500ms）或 超长运行时间（>15分钟 Lambda 超时限制）的场景

2.2 Standalone Activities：从"Workflow 的一部分"到独立任务

2.2.1 概念突破：Activity 的独立化

在传统 Temporal 模型里，Activity 必须作为 Workflow 的一个步骤存在：

# 传统模式：Activity 依附于 Workflow
@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, order_id: str) -> OrderResult:
        # Activity 只能是 Workflow 的子集
        payment_result = await workflow.execute_activity(
            process_payment,
            order_id,
            start_to_close_timeout=timedelta(minutes=5),
        )
        inventory_result = await workflow.execute_activity(
            update_inventory,
            order_id,
        )
        return OrderResult(payment=payment_result, inventory=inventory_result)

Standalone Activities 解锁了一种新的使用模式：Activity 可以直接被客户端触发，不再需要包裹在 Workflow 里。

这意味着你可以把 Temporal 当作一个带持久化保证的分布式任务队列来用，而不仅仅是微服务编排引擎。

2.2.2 架构变化：从"Workflow → Activity"到两种调用路径

传统模式:
Client → Workflow → Activity → Activity → ...

Standalone Activity 模式:
Client → Activity (直接触发，独立重试生命周期)
Client → Workflow → Activity (嵌套模式，照旧)

2.2.3 代码示例：独立的支付 Activity

from temporalio import activity, client
from temporalio.standalone import execute_activity
import stripe

@activity.defn
async def process_standalone_payment(payment_id: str, amount: int) -> dict:
    """可以直接从客户端触发的独立 Activity"""
    try:
        charge = stripe.Charge.create(
            amount=amount,
            currency="cny",
            idempotency_key=f"payment-{payment_id}",
        )
        return {"status": "succeeded", "charge_id": charge.id}
    except stripe.CardError as e:
        activity.heartbeat(f"Payment failed: {e}")
        raise  # Temporal 会自动重试

# client.py — 直接执行 Standalone Activity
import asyncio
from temporalio import client

async def main():
    c = await client.connect("localhost:7233")
    
    # 方式1：直接执行（类似 Celery apply_async）
    result = await c.execute_standalone_activity(
        "process-standalone-payment",
        arg={"payment_id": "pay_123", "amount": 99800},
        start_to_close_timeout=timedelta(minutes=5),
        task_queue="payment-tasks",
        id="payment-activity-pay_123",  # 幂等键，Temporal 保证唯一执行
    )
    print(f"Result: {result}")

asyncio.run(main())

和 Celery/BullMQ 的本质区别：

2.3 Workflow Streams：实时 LLM Token 流的持久化回传

2.3.1 为什么 LLM 流式输出是 Workflow 的噩梦

构建 AI Agent 时，一个常见需求是：把 LLM 的流式 token 实时展示给用户，同时保证整个过程是持久的。

传统的做法是用 WebSocket + LLM streaming API：

# 传统方案：WebSocket 推送 token
async def stream_chat(user_message: str):
    async for token in openai.ChatCompletion.create_stream(
        model="gpt-4",
        messages=[{"role": "user", "content": user_message}]
    ):
        await websocket.send(json.dumps({"token": token}))

问题来了：

1. LLM 服务崩溃：token 流中途断了，用户看到的是不完整回复，怎么恢复？
2. Workflow 中途失败：整个回复要重新生成，之前花的时间全部浪费
3. 多端同步：用户手机和桌面端要看到同一个 token 流，怎么做？

Workflow Streams 用 Temporal 的 Signal 和 Update 机制解决了这个问题：Token 流本身是 Workflow 状态的一部分，断了可以从断点继续推送。

2.3.2 架构原理：Signal + Update = 持久化流

[LLM API] --chunk tokens--> [Activity] --Signal(token)--> [Workflow State]
                                      |
[WebSocket Server] <--subscribe-- [Temporal SDK] --push tokens-- [Client UI]

@workflow.defn
class StreamingAgentWorkflow:
    @workflow.run
    async def run(self, prompt: str) -> str:
        # 用 Signal 实时接收 LLM token
        self.received_tokens: list[str] = []
        
        # 启动 LLM Activity（非阻塞）
        completion_future = workflow.start_activity(
            stream_llm_completion,
            args=[prompt, "stream-consumer"],  # 指定 Signal target
            start_to_close_timeout=timedelta(minutes=5),
        )
        
        # 同时监听 Signal 接收 token
        handler = workflow.get_signal_handler("token-received")
        
        # 这里底层是 Temporal 的 Signal 机制，持久化到服务
        async for token in handler.stream():
            self.received_tokens.append(token)
            # 实时通过 Update 推送给 WebSocket 客户端
            workflow.update_state({"latest_token": token, "count": len(self.received_tokens)})
        
        result = await completion_future
        return result

# LLM Activity：每收到一个 token chunk 就发 Signal
@activity.defn
async def stream_llm_completion(prompt: str, signal_target: str):
    from openai import AsyncOpenAI
    client = AsyncOpenAI()
    
    accumulated = ""
    async with client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    ) as stream:
        async for chunk in stream:
            token = chunk.choices[0].delta.content or ""
            if token:
                accumulated += token
                # 每收到一个 token chunk，就 Signal 给 Workflow
                # Temporal 持久化这个 Signal，Activity 崩溃后重试会继续发
                await workflow.signal(
                    signal_target,
                    "token-received",
                    token,
                )
    return accumulated

关键保证：

• 持久化：Signal 写入 Temporal Event Log，Activity 重试不会漏发 token
• 恰好一次：同一个 Signal 不会在 Workflow 重放时被重复处理
• 实时性：Update 机制支持将最新状态推送给 WebSocket 服务器

2.4 External Payload Storage：让 Workflow 吞吐突破内存限制

2.4.1 痛点：LLM 的 Context Window 是钱，也是瓶颈

构建 RAG Agent 或长程 AI Workflow 时，一个核心瓶颈是 Payload 大小限制。Temporal 默认将 Workflow 的输入输出（Payload）存在内置存储中，单个 Payload 有大小限制。

对于以下场景，这个限制是致命的：

• 一次处理 1000 份文档的批处理 Agent
• 上传大文件到 S3 后做内容分析的长 Workflow
• RAG 系统需要一次性传递数 MB 的检索上下文

2.4.2 External Payload Storage 架构

[Worker] --Large Payload--> [Custom Storage Driver] --> [S3 / GCS / 自定义存储]
                                    |
                              [Temporal Server] <-- 引用（指针）而非完整数据

# external_storage.py
from temporalio.contrib.external_storage import (
    ExternalStorageHandler,
    S3StorageConfig,
)

class S3PayloadStorage(ExternalStorageHandler):
    def __init__(self, bucket: str):
        self.s3 = boto3.client("s3")
        self.bucket = bucket
    
    async def store(self, data: bytes) -> str:
        key = f"payloads/{uuid.uuid4()}"
        self.s3.put_object(Bucket=self.bucket, Key=key, Body=data)
        return key  # Temporal 只存储这个 key，不存完整数据
    
    async def retrieve(self, key: str) -> bytes:
        obj = self.s3.get_object(Bucket=self.bucket, Key=key)
        return obj["Body"].read()

# 使用 S3 存储大型 Workflow Payload
config = S3PayloadStorage(bucket="my-workflow-payloads")

三、AI Agent 集成：从"能跑"到"敢上线"

3.1 Google ADK 集成：让 Gemini Agent 真正可靠

Google Agent Development Kit (ADK) 是一套构建 Gemini Agent 的框架。Temporal 与 Google ADK 的深度集成，将 LLM 调用和工具执行全部包装为 Activity：

# 每个 LLM 调用 = Temporal Activity（自动重试 + 持久化）
# 每个工具执行 = Temporal Activity（失败重试，状态不丢失）
from google.adk import Agent
from temporalio.contrib.google_adk import temporalize_agent

# 一行调用，把整个 Agent 变成 Durable Agent
durable_agent = temporalize_agent(
    agent=my_gemini_agent,
    task_queue="gemini-agent-queue",
    auto_retry_llm=True,
    auto_retry_tools=True,
)

# 运行：即使中途服务重启，整个 Agent 从断点继续
result = await durable_agent.run(user_goal="帮我分析这份财报")

3.2 OpenAI Agents SDK 沙箱集成

OpenAI 的 Agents SDK 提供了沙箱执行环境（Sandbox），可以隔离 Agent 的代码执行。Temporal SDK 现在支持将沙箱作为 Activity 运行：

# 隔离环境中的 AI Agent = 可恢复的 Activity
from temporalio.contrib.openai_agents import SandboxedAgentRunner

runner = SandboxedAgentRunner(
    agent=my_coding_agent,
    sandbox_timeout=timedelta(minutes=10),
)

@workflow.defn
class CodeReviewWorkflow:
    @workflow.run
    async def run(self, pr_url: str):
        # 沙箱崩溃 → Activity 重试 → 从断点继续
        # 沙箱外部状态全部由 Temporal 持久化
        review = await workflow.execute_activity(
            runner.run,
            args=[f"Analyze PR: {pr_url}"],
            start_to_close_timeout=timedelta(minutes=15),
        )
        return review

四、生产级特性：从"能用"到"敢在618用"

4.1 Worker Versioning：代码发布不再怕 in-flight 任务

在 Replay 2026 之前，发布新版本 Worker 是个高风险操作：正在运行的老 Workflow 可能会因为新代码的某些非确定性行为（non-deterministic）而出错。

Worker Versioning 的解决方案是版本锚定：每个 Workflow 实例在启动时自动绑定到启动它的 Worker 版本，后续即使有新版本发布，该 Workflow 继续用老版本运行，直到它自然结束。

Worker v1.0 ──启动──> Workflow A ──发布──> Worker v1.1
                                            |
                         Workflow A 继续用 v1.0 处理，不受影响
                         Workflow B、C（v1.1 启动的）用 v1.1

发布策略变为：

# worker.py — 渐进式发布
worker = await Worker.create(
    task_queue="order-queue",
    workflows=[OrderWorkflow],
    activities=[process_payment, send_notification],
    build_id="order-worker-v2",
    use_versioning=True,  # 启用版本锚定
    default_active_build_id="order-worker-v2",
)

4.2 Task Queue Priority & Fairness：关键任务永远不被饿死

当一个 Task Queue 里混着"即时退款"（P0）和"延迟同步"（P3）两类任务时，高优先级任务可能被大量低优先级任务阻塞。

# 设置 Task Queue 的优先级
await admin_client.update_task_queue_config(
    name="order-queue",
    specs=[
        TaskQueueSpec(
            task_routing_config=TaskRoutingConfig(
                priority=1,  # P1 = 关键支付流程
            ),
            max_concurrent_workflow_tasks=100,
        ),
    ],
)

# Activity 层面也可以指定优先级
await workflow.execute_activity(
    "process-refund",
    refund_id,
    priority=10,  # 优先级越高越先调度
    start_to_close_timeout=timedelta(seconds=30),
)

Fairness（公平性） 则保证同一队列内的不同租户/团队不会被单一高流量方占满资源。

五、性能基准：Serverless Workers 冷启动实测

六、实战：构建一个"永不丢失"的数据管道

最后用一个综合示例串联所有新特性——构建一个支持 Serverless Activity 的 RAG 数据管道：

from temporalio import workflow, activity, server
from temporalio.contrib.serverless import ServerlessWorkerConfig
from temporalio.standalone import execute_activity

# ========== Activity 层（Serverless）==========

@activity.defn
async def fetch_documents(query: str) -> list[dict]:
    """从向量数据库检索相关文档 — Serverless Activity"""
    results = await vectordb.similarity_search(query, k=20)
    return [{"id": r.id, "content": r.page_content, "score": r.score} for r in results]

@activity.defn
async def embed_and_store(documents: list[dict]) -> str:
    """Embedding 并存向量库 — Serverless Activity"""
    embeddings = await openai.Embedding.acreate(
        model="text-embedding-3-large",
        input=[d["content"] for d in documents]
    )
    collection.add(
        ids=[d["id"] for d in documents],
        embeddings=[e.embedding for e in embeddings.data],
    )
    return f"Stored {len(documents)} documents"

# ========== Workflow 层（Durable）==========

@workflow.defn
class RAGPipelineWorkflow:
    @workflow.run
    async def run(self, query: str, user_id: str) -> dict:
        # Step 1: 获取文档（Serverless Activity）
        docs = await workflow.execute_activity(
            fetch_documents,
            query,
            start_to_close_timeout=timedelta(seconds=30),
            task_queue="rag-tasks",
        )
        
        # Step 2: Embed + 存储（Serverless Activity）
        collection_id = await workflow.execute_activity(
            embed_and_store,
            docs,
            start_to_close_timeout=timedelta(minutes=2),
            task_queue="rag-tasks",
        )
        
        # Step 3: 生成回答（带 Workflow Streams 实时回传）
        answer = await workflow.execute_activity(
            generate_answer_with_stream,
            args=[query, docs],
            start_to_close_timeout=timedelta(minutes=5),
        )
        
        # 所有步骤均可从断点恢复，无任何补偿代码
        return {"answer": answer, "sources": docs, "collection": collection_id}

# ========== 独立触发（Standalone Activity）==========

async def daily_sync():
    """Django/FastAPI 后端直接调用，无需包装 Workflow"""
    c = await client.connect("localhost:7233")
    
    # Standalone Activity：作为定时任务直接执行
    await c.execute_standalone_activity(
        "fetch-documents",
        arg={"query": "system:maintenance", "force_refresh": True},
        task_queue="rag-tasks",
        id="daily-sync-2026-07-14",
        start_to_close_timeout=timedelta(hours=1),
    )

这套架构的独特价值：

• RAG 文档获取 + Embedding：用 Serverless Workers，凌晨空闲时零成本
• Workflow Streams：用户实时看到 token 流，体验流畅
• Standalone Activity：定时同步无需额外消息队列
• Durable Workflow：任何一步失败，从断点继续，无数据丢失

七、总结与展望

Temporal Replay 2026 的四大核心更新，从三个维度重新定义了分布式任务执行：

1. 部署维度：Serverless Workers 让 Temporal 从"有状态基础设施"变成了"零运维云函数"，降低进入门槛
2. 编程模型维度：Standalone Activities 打破了 Activity 必须依附 Workflow 的限制，任务队列场景直接受益
3. AI 原生维度：Workflow Streams + External Payload Storage + Google ADK 集成，让 AI Agent 真正具备了生产级的可靠性

对技术决策者的建议：

• 如果你已经在用 Temporal v1.x，优先评估 Workflow Streams——它解决的是 LLM 应用最痛的实时+可靠双重需求
• 如果你在选型任务队列，Temporal Standalone Activities 是 Celery/BullMQ 的强力替代，尤其当你的任务需要精细重试控制时
• 如果你要构建 AI Agent，Google ADK + Temporal 集成是一套已经被生产验证的组合拳，值得深入了解

Durable Execution 的核心理念——"让失败不再需要补偿代码"——正在从 Temporal 走向主流。2026年，这个理念和 Serverless、AI Agent 的深度融合，可能是后端工程师最值得掌握的新范式之一。

参考资源：

• Temporal Replay 2026 官方公告
• Serverless Workers 文档
• Standalone Activities 文档
• Workflow Streams 文档
• External Storage 文档