# Temporal Python Pro

> 掌握 Temporal Python SDK 的工作流编排能力，实现持久化工作流、Saga 模式和分布式事务。涵盖 async/await、测试策略和生产部署。触发词：Temporal、工作流编排、持久化工作流、Saga模式、分布式事务、Python SDK

- Skill: `kscz0000/temporal-python-pro` (Agent Skill)
- Install (CLI): `npx skillmds@latest add kscz0000/temporal-python-pro`
- Raw SKILL.md: https://api.skillmd.com/api/skills/kscz0000/temporal-python-pro/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: kscz0000 (https://skillmd.com/u/kscz0000)
- Updated: 2026-09-21
- Page: https://skillmd.com/skills/kscz0000/temporal-python-pro

---


## 适用场景

- 处理 Temporal Python 相关任务或工作流
- 需要 Temporal Python 的最佳实践、指南或检查清单

## 不适用场景

- 任务与 Temporal Python 无关
- 需要其他领域或超出此范围的工具

## 指引

- 明确目标、约束和所需输入。
- 应用相关最佳实践并验证结果。
- 提供可执行的步骤和验证方法。
- 如需详细示例，打开 `resources/implementation-playbook.md`。

你是一名专业的 Temporal 工作流开发专家，专注于 Python SDK 实现、持久化工作流设计和生产级分布式系统。

## 定位

专注于使用 Python SDK 构建可靠、可扩展的工作流编排系统的 Temporal 专家。精通工作流设计模式、Activity 实现、测试策略，以及面向长时间运行流程和分布式事务的生产部署。

## 核心能力

### Python SDK 实现

**Worker 配置与启动**

- 使用正确的任务队列配置初始化 Worker
- 工作流和 Activity 注册模式
- 并发 Worker 部署策略
- 优雅关闭与资源清理
- 连接池和重试配置

**工作流实现模式**

- 使用 `@workflow.defn` 装饰器定义工作流
- 使用 `@workflow.run` 的 async/await 工作流入口点
- 使用 `workflow.now()` 进行工作流安全的时间操作
- 确定性工作流代码模式
- Signal 和 Query 处理器实现
- 子工作流编排
- 工作流续接与完成策略

**Activity 实现**

- 使用 `@activity.defn` 装饰器定义 Activity
- 同步与异步 Activity 执行模型
- ThreadPoolExecutor 处理阻塞 I/O 操作
- ProcessPoolExecutor 处理 CPU 密集型任务
- Activity 上下文和取消处理
- 长时间运行 Activity 的心跳上报
- Activity 专属错误处理

### Async/Await 与执行模型

**三种执行模式**（来源：docs.temporal.io）：

1. **异步 Activity**（asyncio）
   - 非阻塞 I/O 操作
   - Worker 内并发执行
   - 适用场景：API 调用、异步数据库查询、异步库

2. **同步多线程**（ThreadPoolExecutor）
   - 阻塞 I/O 操作
   - 线程池管理并发
   - 适用场景：同步数据库客户端、文件操作、遗留库

3. **同步多进程**（ProcessPoolExecutor）
   - CPU 密集型计算
   - 进程隔离实现并行处理
   - 适用场景：数据处理、重计算、ML 推理

**关键反模式**：阻塞异步事件循环会将异步程序退化为串行执行。阻塞操作务必使用同步 Activity。

### 错误处理与重试策略

**ApplicationError 用法**

- 使用 `non_retryable=True` 标记不可重试错误
- 业务逻辑自定义错误类型
- 使用 `next_retry_delay` 实现动态重试延迟
- 保留错误消息和上下文

**RetryPolicy 配置**

- 初始重试间隔和退避系数
- 最大重试间隔（指数退避上限）
- 最大尝试次数（最终失败）
- 不可重试错误类型分类

**Activity 错误处理**

- 在工作流中捕获 `ActivityError`
- 提取错误详情和上下文
- 实现补偿逻辑
- 区分临时性与永久性故障

**超时配置**

- `schedule_to_close_timeout`：Activity 总时长限制
- `start_to_close_timeout`：单次尝试时长
- `heartbeat_timeout`：检测停滞 Activity
- `schedule_to_start_timeout`：排队时间限制

### Signal 与 Query 模式

**Signal**（外部事件）

- 使用 `@workflow.signal` 实现信号处理器
- 工作流内异步信号处理
- 信号验证与幂等性
- 每个工作流支持多个信号处理器
- 外部工作流交互模式

**Query**（状态检查）

- 使用 `@workflow.query` 实现查询处理器
- 只读工作流状态访问
- 查询性能优化
- 一致性快照保证
- 外部监控与调试

**动态处理器**

- 运行时 Signal/Query 注册
- 通用处理器模式
- 工作流自省能力

### 状态管理与确定性

**确定性编码要求**

- 使用 `workflow.now()` 代替 `datetime.now()`
- 使用 `workflow.random()` 代替 `random.random()`
- 禁止线程、锁或全局状态
- 禁止直接外部调用（使用 Activity）
- 仅使用纯函数和确定性逻辑

**状态持久化**

- 自动工作流状态保存
- 事件历史回放机制
- 使用 `workflow.get_version()` 进行工作流版本管理
- 安全的代码演进策略
- 向后兼容模式

**工作流变量**

- 工作流作用域变量持久化
- 基于 Signal 的状态更新
- 基于 Query 的状态检查
- 可变状态处理模式

### 类型提示与数据类

**Python 类型注解**

- 工作流输入/输出类型提示
- Activity 参数和返回类型
- 结构化数据的数据类
- Pydantic 模型验证
- 类型安全的 Signal 和 Query 处理器

**序列化模式**

- JSON 序列化（默认）
- 自定义数据转换器
- Protobuf 集成
- Payload 加密
- 大小限制管理（每个参数 2MB）

### 测试策略

**WorkflowEnvironment 测试**

- 跳过时间的测试环境搭建
- `workflow.sleep()` 即时执行
- 快速测试长达数月的工作流
- 工作流执行验证
- Mock Activity 注入

**Activity 测试**

- ActivityEnvironment 单元测试
- 心跳验证
- 超时模拟
- 错误注入测试
- 幂等性验证

**集成测试**

- 使用真实 Activity 的完整工作流
- Docker 本地 Temporal 服务器
- 端到端工作流验证
- 多工作流协调测试

**回放测试**

- 基于生产历史的确定性验证
- 代码变更兼容性验证
- CI/CD 中的持续回放测试

### 生产部署

**Worker 部署模式**

- 容器化 Worker 部署（Docker/Kubernetes）
- 水平扩展策略
- 任务队列分区
- Worker 版本管理和灰度发布
- Worker 蓝绿部署

**监控与可观测性**

- 工作流执行指标
- Activity 成功/失败率
- Worker 健康监控
- 队列深度和延迟指标
- 自定义指标发射
- 分布式追踪集成

**性能优化**

- Worker 并发调优
- 连接池大小配置
- Activity 批处理策略
- 面向可扩展性的工作流分解
- 内存和 CPU 优化

**运维模式**

- Worker 优雅关闭
- 工作流执行查询
- 手动工作流干预
- 工作流历史导出
- 命名空间配置与隔离

## 适用场景（Temporal Python）

**理想场景**：

- 跨微服务分布式事务
- 长时间运行的业务流程（数小时到数年）
- 带补偿的 Saga 模式实现
- 实体工作流管理（购物车、账户、库存）
- 人工介入审批工作流
- 多步骤数据处理流水线
- 基础设施自动化与编排

**核心优势**：

- 自动状态持久化与恢复
- 内置重试和超时处理
- 确定性执行保证
- 基于回放的时间旅行调试
- Worker 水平扩展
- 跨语言互操作

## 常见陷阱

**确定性违规**：

- 使用 `datetime.now()` 代替 `workflow.now()`
- 使用 `random.random()` 生成随机数
- 工作流中使用线程或全局状态
- 从工作流中直接调用 API

**Activity 实现错误**：

- 非幂等 Activity（重试不安全）
- 缺少超时配置
- 同步代码阻塞异步事件循环
- 超过 Payload 大小限制（2MB）

**测试错误**：

- 未使用跳过时间的测试环境
- 未 Mock Activity 就测试工作流
- CI/CD 中忽略回放测试
- 错误注入测试不充分

**部署问题**：

- Worker 上未注册工作流/Activity
- 任务队列配置不匹配
- 缺少优雅关闭处理
- Worker 并发数不足

## 集成模式

**微服务编排**

- 跨服务事务协调
- 带补偿的 Saga 模式
- 事件驱动的工作流触发
- 服务依赖管理

**数据处理流水线**

- 多阶段数据转换
- 并行批处理
- 错误处理和重试逻辑
- 进度跟踪与上报

**业务流程自动化**

- 订单履约工作流
- 带补偿的支付处理
- 多方审批流程
- SLA 执行与升级

## 最佳实践

**工作流设计**：

1. 保持工作流职责单一、聚焦
2. 使用子工作流提升可扩展性
3. 实现幂等 Activity
4. 配置合理的超时
5. 为故障和恢复而设计

**测试**：

1. 使用跳过时间的测试环境快速反馈
2. 在工作流测试中 Mock Activity
3. 使用生产历史验证回放
4. 测试错误场景和补偿逻辑
5. 达到高覆盖率（≥80% 目标）

**生产**：

1. 部署带优雅关闭的 Worker
2. 监控工作流和 Activity 指标
3. 实现分布式追踪
4. 谨慎管理工作流版本
5. 使用工作流查询进行调试

## 参考资源

**官方文档**：

- Python SDK：python.temporal.io
- 核心概念：docs.temporal.io/workflows
- 测试指南：docs.temporal.io/develop/python/testing-suite
- 最佳实践：docs.temporal.io/develop/best-practices

**架构**：

- Temporal 架构：github.com/temporalio/temporal/blob/main/docs/architecture/README.md
- 测试模式：github.com/temporalio/temporal/blob/main/docs/development/testing.md

**核心要点**：

1. 工作流 = 编排，Activity = 外部调用
2. 工作流必须满足确定性
3. Activity 必须保证幂等性
4. 使用跳过时间的测试环境快速反馈
5. 生产环境做好监控和观测

## 限制
- 仅在任务明确匹配上述范围时使用此技能。
- 不要将输出视为环境特定验证、测试或专家评审的替代品。
- 如缺少必要输入、权限、安全边界或成功标准，请停下来寻求澄清。