temporal

Temporal 工作流编排

用于构建持久、可靠的分布式系统的专家级知识，基于 Temporal.io。涵盖 Python SDK（temporalio）、TypeScript SDK（@temporalio/*）、核心平台概念以及 AI/大模型编排模式。

参考资料

文件	阅读时机
core-concepts.md	架构、确定性、超时、重试策略、高级功能
python-sdk.md	Python 工作流/活动实现、装饰器、测试、客户端/Worker 设置
typescript-sdk.md	TypeScript 工作流/活动实现、沙箱、打包、测试
ai-patterns.md	AI Agent 编排、大模型流水线、人机协同、框架集成

平台概述

Temporal 提供持久执行——通过基于事件的重放，即使进程崩溃、基础设施故障或部署，应用状态也能得以保留。该平台由 Temporal Service（服务端 + 持久化）和应用代码（Workers + Clients）组成。Workers 在用户基础设施上执行 Workflow 和 Activity 代码；Temporal Service 负责协调任务分发、状态持久化和计时器管理。

核心原语

原语	作用
Workflow	确定性编排函数；状态通过事件历史持久化
Activity	非确定性工作单元（I/O、API 调用、副作用）
Worker	轮询任务队列并执行工作流和活动的进程
Task Queue	将工作从服务路由到 Workers 的命名队列
Signal	向运行中的工作流发送异步即发即弃消息
Query	同步只读状态检查（不产生事件历史）
Update	带返回值和验证的同步状态变更
Child Workflow	由另一个工作流启动的工作流；拥有独立的事件历史
Schedule	由服务端管理的周期性工作流执行

确定性契约

工作流代码在恢复时根据事件历史重放。给定相同的 Events，每次重放都会产生相同序列的 Commands。非确定性操作（I/O、随机数、系统时间、线程操作）应放在 Activities 中。

在工作流中安全使用	替代以下内容
workflow.now() / Date.now()（由沙箱替换）	datetime.now() / 原生 Date.now()
workflow.random() / Math.random()（由沙箱替换）	random.random() / 原生 Math.random()
workflow.uuid4() / 来自 @temporalio/workflow 的 uuid4()	uuid.uuid4() / crypto.randomUUID()
workflow.sleep() / 来自 @temporalio/workflow 的 sleep()	asyncio.sleep() / setTimeout()
执行 Activity	直接进行网络/文件 I/O

超时模型

超时	范围	适用于
Workflow Execution Timeout	整个链（包括 Continue-As-New 运行）	Workflow
Workflow Run Timeout	链中的单次运行	Workflow
Workflow Task Timeout	单个 SDK 处理步骤（默认：10s）	Workflow
Start-to-Close Timeout	单次 Activity 尝试	Activity
Schedule-to-Close Timeout	整个 Activity 执行过程，包括重试	Activity
Schedule-to-Start Timeout	在队列中等待 Worker 接手的时间	Activity
Heartbeat Timeout	长时间运行的 Activity 两次心跳之间的间隔	Activity

Start-to-Close Timeout 是检测 Worker 故障的主要机制，也是最重要的 Activity 超时设置。

重试策略默认值

属性	默认值
Initial Interval	1 秒
Backoff Coefficient	2.0
Maximum Interval	Initial Interval 的 100 倍
Maximum Attempts	不限
Non-Retryable Error Types	无

Activities 默认会重试。Workflows 默认不会重试。

SDK 快速参考

Python

包： temporalio（pip install temporalio）

from temporalio import workflow, activity
from temporalio.client import Client
from temporalio.worker import Worker

@activity.defn
async def greet(name: str) -> str:
    return f"Hello, {name}!"

@workflow.defn
class GreetingWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            greet, name,
            start_to_close_timeout=timedelta(seconds=10),
        )

# Client
client = await Client.connect("localhost:7233")
result = await client.execute_workflow(
    GreetingWorkflow.run, "World",
    id="greeting-1", task_queue="my-queue",
)

# Worker
worker = Worker(client, task_queue="my-queue",
    workflows=[GreetingWorkflow], activities=[greet])
await worker.run()

Python 装饰器： @workflow.defn、@workflow.run、@workflow.init、@workflow.signal、@workflow.query、@workflow.update、@activity.defn

沙箱： 工作流代码在隔离的沙箱中运行。导入活动类型和数据类时，请使用 with workflow.unsafe.imports_passed_through():。

TypeScript

包： @temporalio/client、@temporalio/worker、@temporalio/workflow、@temporalio/activity、@temporalio/common

// activities.ts
export async function greet(name: string): Promise<string> {
  return `Hello, ${name}!`;
}

// workflows.ts
import { proxyActivities } from '@temporalio/workflow';
import type * as activities from './activities';

const { greet } = proxyActivities<typeof activities>({
  startToCloseTimeout: '10s',
});

export async function greetingWorkflow(name: string): Promise<string> {
  return await greet(name);
}

// client.ts
const handle = await client.workflow.start(greetingWorkflow, {
  workflowId: 'greeting-1', taskQueue: 'my-queue', args: ['World'],
});

// worker.ts
const worker = await Worker.create({
  workflowsPath: require.resolve('./workflows'),
  activities, taskQueue: 'my-queue',
});
await worker.run();

关键模式： 工作流仅从活动中导入类型（import type * as activities）。实际实现会注册到 Worker 上。通过 proxyActivities<typeof activities>() 访问活动。

沙箱： 工作流代码会通过 Webpack 打包到 V8 隔离区中。Math.random()、Date.now() 和 setTimeout 会替换为确定性版本。工作流代码无法使用 Node.js 内置模块（fs、path、crypto、http）。

模式选择指南

场景	模式	详情
调用外部 API、查询数据库、文件 I/O	活动	具有自动重试机制的非确定性工作
编排多个步骤并维护状态	工作流	通过持久状态进行确定性协调
在多个服务／团队间拆分工作	子工作流	独立事件历史，由独立 Worker 执行
事件历史接近 50K 个事件	Continue-As-New	在保留工作流标识的同时创建新的历史记录
周期性执行	Schedule	由服务端管理，支持暂停和重叠策略
等待外部输入	Signal + wait_condition／condition	持久化等待，零计算开销
从外部读取工作流状态	Query	不产生事件历史开销
修改状态并返回响应	Update	经过验证、同步执行并有记录
多步骤操作并支持回滚	Saga（补偿列表）	失败时按相反顺序执行补偿操作
长时间运行并报告进度	心跳活动	检测故障并支持恢复进度
外部完成（人工审批）	异步活动完成	使用任务令牌进行带外完成
AI Agent 循环	工作流 + 大模型活动	持久化 Agent 状态，并重试大模型调用
并行 AI 处理	扇出子工作流／活动	并发执行并汇总结果
人类参与的 AI	Signal／Update 处理器	持久化审批关卡

AI 编排概览

Temporal 是用于生产级 AI Agent 系统的基础设施。核心模式是：工作流维护持久化的 Agent 状态和对话历史，而活动处理非确定性的大模型调用、工具调用和外部 API 请求。

AI 工作负载的关键特性：

• 已完成的活动结果通过事件溯源进行持久化——发生崩溃后，系统会重放已记录的大模型响应，而非重新查询（节省成本并保持确定性）
• 重试策略会自动处理速率限制（429）、瞬时故障和 API 超时
• Signal 支持设置人工参与的审批关卡，且无需消耗计算资源进行等待
• 任务队列会将推理工作路由到配备 GPU 的 Worker
• 存储为工作流状态的对话历史会自动持久化

框架集成

框架	集成类型
OpenAI Agents SDK	自定义 Runner；activity_as_tool 助手
Pydantic AI	TemporalAgent 封装器；自动将 I/O 转移到活动中
Vercel AI SDK	Worker 的 AiSDKPlugin；将 generateText() 封装在活动中
Anthropic Claude API	在活动中直接使用 AsyncAnthropic 客户端；max_retries=0（由 Temporal 处理重试）
LangChain	由活动封装的链执行
Model Context Protocol	持久化 MCP：每个工具都由一个工作流提供支持

AI 专用配置

LLM 客户端库内置的重试逻辑会与 Temporal 的重试管理冲突。请禁用客户端级重试，并交由 Temporal 处理：

# Python - Anthropic
client = AsyncAnthropic(max_retries=0)

# Python - OpenAI
client = AsyncOpenAI(max_retries=0)

// TypeScript - OpenAI
const client = new OpenAI({ maxRetries: 0 });

LLM 活动的推荐重试策略：

RetryPolicy(
    initial_interval=timedelta(seconds=1),
    backoff_coefficient=2.0,
    maximum_interval=timedelta(seconds=120),
    maximum_attempts=50,
    non_retryable_error_types=["InvalidInput", "AuthenticationError"],
)

使用 ApplicationError / ApplicationFailure 上的 next_retry_delay，解析 429 响应中的 Retry-After 标头。

开发环境

# Install Temporal CLI
brew install temporal

# Start local dev server (gRPC :7233, Web UI :8233)
temporal server start-dev

# Persistent storage between restarts
temporal server start-dev --db-filename temporal.db

# Common CLI commands
temporal workflow start --task-queue my-queue --type MyWorkflow --input '{"key":"val"}'
temporal workflow list
temporal workflow describe --workflow-id my-id
temporal workflow signal --workflow-id my-id --name my-signal --input '"data"'
temporal workflow query --workflow-id my-id --name my-query
temporal workflow cancel --workflow-id my-id

版本控制策略

两个 SDK 都提供 patched() / deprecate_patch() 机制，可在现有工作流执行仍在运行时演进工作流代码：

1. 添加补丁：根据 patched("change-id") 进行分支——新执行采用新路径，正在重放的执行采用旧路径
2. 弃用补丁：所有旧执行完成后，调用 deprecate_patch("change-id") 并移除旧代码路径
3. 移除补丁：保留期结束后，完全移除弃用调用