“

系列：S12 开发者教程 · 第 16 篇 | 难度：中级 | 阅读时间：15 分钟

Temporal 工作流编排指南

#引言

在企业级应用中，很多业务流程涉及多个步骤、多个系统的协作——审批流程、数据处理管道、订单履约链路。这些长时间运行的流程（Long-Running Process）需要持久化状态、错误重试、超时处理和可观测性。Temporal 正是为此而生的分布式工作流引擎。

coomia-dip 选择 Temporal 作为 Agent Runtime Layer（Agent Runtime）的工作流编排基础设施。本教程将带你从零开始，用 Python SDK 编写第一个 Temporal 工作流，集成到 coomia-dip 平台中。

#1. Temporal 核心概念

#1.1 为什么需要工作流引擎

假设你需要实现一个"客户贷款审批"流程：

接收申请
调用征信系统查询信用分
如果信用分 > 700，自动批准；否则进入人工审核
人工审核可能需要 1-7 天
审核通过后调用放款系统
发送通知

如果用传统代码实现，你需要处理：进程重启后状态恢复、网络超时重试、人工审核的长时间等待、每个步骤的幂等性。Temporal 将这些通用复杂性从业务代码中剥离出来。

#1.2 核心组件

Code

Temporal Server (已部署在 coomia-dip 中)
    |
    +-- Namespace: coomia-dip-workflows
    |       |
    |       +-- Task Queue: approval-tasks
    |       +-- Task Queue: data-pipeline-tasks
    |       +-- Task Queue: notification-tasks
    |
    +-- Worker (你的应用进程)
    |       |
    |       +-- 注册 Workflow 实现
    |       +-- 注册 Activity 实现
    |       +-- 轮询 Task Queue
    |
    +-- Client (发起工作流的代码)
            |
            +-- start_workflow()
            +-- signal_workflow()
            +-- query_workflow()

关键概念：

Workflow：业务流程的代码表示。看起来像普通函数，但具有持久化执行的能力
Activity：工作流中的单个步骤。与外部系统交互的代码放在 Activity 中
Worker：执行 Workflow 和 Activity 的进程。轮询 Task Queue 获取任务
Task Queue：Worker 和 Workflow 之间的任务分发通道

#1.3 Temporal 的魔法：事件溯源

Temporal 的核心原理是事件溯源（Event Sourcing）。每个 Workflow 的执行历史被完整记录。当 Worker 重启或 Workflow 需要恢复时，Temporal 通过重放事件历史来重建 Workflow 状态——你的代码会被重新执行，但 Activity 的结果从历史中获取而非重新调用。

这意味着：

代码即状态机：你写的是线性代码，Temporal 自动处理持久化
自动恢复：Worker 崩溃后，另一个 Worker 可以无缝接管
无限等待：Workflow 可以等待数天、数月甚至数年
完整审计：每个步骤的输入输出都被记录

#2. 环境搭建

#2.1 安装 Temporal Python SDK

Bash

pip install temporalio

#2.2 验证 Temporal Server 连接

在 coomia-dip 部署中，Temporal Server 已经作为基础设施运行：

Python

from temporalio.client import Client

async def check_connection():
    client = await Client.connect("localhost:7233")
    # 列出命名空间
    namespaces = await client.service_client.list_namespaces()
    for ns in namespaces:
        print(f"Namespace: {ns.namespace_info.name}")

import asyncio
asyncio.run(check_connection())

#3. 编写第一个工作流

#3.1 定义 Activity

Activity 是与外部系统交互的地方——API 调用、数据库查询、文件处理：

Python

from temporalio import activity
from dataclasses import dataclass
import httpx

@dataclass
class CreditCheckInput:
    customer_id: str
    application_id: str

@dataclass
class CreditCheckResult:
    score: int
    risk_level: str
    details: dict

@activity.defn
async def check_credit(input: CreditCheckInput) -> CreditCheckResult:
    activity.logger.info(f"Checking credit for customer {input.customer_id}")
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://credit-service:8080/api/check",
            json={"customer_id": input.customer_id},
            timeout=30.0,
        )
        response.raise_for_status()
        data = response.json()

    return CreditCheckResult(
        score=data["score"],
        risk_level=data["risk_level"],
        details=data.get("details", {}),
    )

@activity.defn
async def send_notification(customer_id: str, message: str) -> bool:
    activity.logger.info(f"Sending notification to {customer_id}")
    # 发送邮件/短信通知
    async with httpx.AsyncClient() as client:
        await client.post(
            "http://notification-service:8080/api/send",
            json={"customer_id": customer_id, "message": message},
        )
    return True

@activity.defn
async def process_disbursement(application_id: str, amount: float) -> str:
    activity.logger.info(f"Processing disbursement for {application_id}: {amount}")
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://disbursement-service:8080/api/disburse",
            json={"application_id": application_id, "amount": amount},
        )
        data = response.json()
    return data["transaction_id"]

#3.2 定义 Workflow

Python

from temporalio import workflow
from datetime import timedelta
from dataclasses import dataclass
from enum import Enum

class ApprovalStatus(str, Enum):
    PENDING = "PENDING"
    APPROVED = "APPROVED"
    REJECTED = "REJECTED"
    MANUAL_REVIEW = "MANUAL_REVIEW"

@dataclass
class LoanApplication:
    application_id: str
    customer_id: str
    amount: float
    purpose: str

@dataclass
class LoanResult:
    application_id: str
    status: ApprovalStatus
    transaction_id: str | None = None
    rejection_reason: str | None = None

@workflow.defn
class LoanApprovalWorkflow:
    def __init__(self):
        self._status = ApprovalStatus.PENDING
        self._manual_decision: ApprovalStatus | None = None
        self._reviewer_notes: str = ""

    @workflow.run
    async def run(self, application: LoanApplication) -> LoanResult:
        workflow.logger.info(f"Processing loan application {application.application_id}")

        # Step 1: 信用检查
        credit = await workflow.execute_activity(
            check_credit,
            CreditCheckInput(
                customer_id=application.customer_id,
                application_id=application.application_id,
            ),
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(
                maximum_attempts=3,
                initial_interval=timedelta(seconds=1),
            ),
        )

        # Step 2: 自动决策或转人工
        if credit.score >= 700 and application.amount <= 50000:
            self._status = ApprovalStatus.APPROVED
        elif credit.score < 500:
            self._status = ApprovalStatus.REJECTED
            await workflow.execute_activity(
                send_notification,
                args=[application.customer_id, "Your loan application has been rejected."],
                start_to_close_timeout=timedelta(seconds=10),
            )
            return LoanResult(
                application_id=application.application_id,
                status=ApprovalStatus.REJECTED,
                rejection_reason=f"Credit score {credit.score} below minimum",
            )
        else:
            # 进入人工审核
            self._status = ApprovalStatus.MANUAL_REVIEW
            await workflow.execute_activity(
                send_notification,
                args=[application.customer_id, "Your application is under manual review."],
                start_to_close_timeout=timedelta(seconds=10),
            )

            # 等待人工决策（最多 7 天）
            try:
                await workflow.wait_condition(
                    lambda: self._manual_decision is not None,
                    timeout=timedelta(days=7),
                )
            except asyncio.TimeoutError:
                self._status = ApprovalStatus.REJECTED
                return LoanResult(
                    application_id=application.application_id,
                    status=ApprovalStatus.REJECTED,
                    rejection_reason="Manual review timeout (7 days)",
                )

            if self._manual_decision == ApprovalStatus.REJECTED:
                return LoanResult(
                    application_id=application.application_id,
                    status=ApprovalStatus.REJECTED,
                    rejection_reason=self._reviewer_notes,
                )

            self._status = ApprovalStatus.APPROVED

        # Step 3: 放款
        transaction_id = await workflow.execute_activity(
            process_disbursement,
            args=[application.application_id, application.amount],
            start_to_close_timeout=timedelta(seconds=60),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

        # Step 4: 通知
        await workflow.execute_activity(
            send_notification,
            args=[application.customer_id, f"Loan approved! Transaction: {transaction_id}"],
            start_to_close_timeout=timedelta(seconds=10),
        )

        return LoanResult(
            application_id=application.application_id,
            status=ApprovalStatus.APPROVED,
            transaction_id=transaction_id,
        )

    @workflow.signal
    async def manual_review_decision(self, decision: str, notes: str):
        self._manual_decision = ApprovalStatus(decision)
        self._reviewer_notes = notes

    @workflow.query
    def get_status(self) -> str:
        return self._status.value

#3.3 启动 Worker

Python

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker

async def main():
    client = await Client.connect("localhost:7233")

    worker = Worker(
        client,
        task_queue="approval-tasks",
        workflows=[LoanApprovalWorkflow],
        activities=[check_credit, send_notification, process_disbursement],
    )

    print("Worker started, listening on 'approval-tasks'")
    await worker.run()

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

#3.4 启动工作流

Python

async def start_loan_workflow():
    client = await Client.connect("localhost:7233")

    application = LoanApplication(
        application_id="LOAN-2025-001",
        customer_id="CUST-123",
        amount=30000.0,
        purpose="Business expansion",
    )

    handle = await client.start_workflow(
        LoanApprovalWorkflow.run,
        application,
        id=f"loan-{application.application_id}",
        task_queue="approval-tasks",
    )

    print(f"Workflow started: {handle.id}")

    # 查询状态
    status = await handle.query(LoanApprovalWorkflow.get_status)
    print(f"Current status: {status}")

    # 发送人工审核信号
    await handle.signal(
        LoanApprovalWorkflow.manual_review_decision,
        "APPROVED",
        "Customer has good history",
    )

    # 等待结果
    result = await handle.result()
    print(f"Final result: {result}")

#4. 与 coomia-dip 集成

#4.1 在 Action 中触发工作流

Python

from ontology_sdk.functions import ontology_action

@ontology_action(
    name="submit_loan_application",
    parameters={
        "customerId": "string",
        "amount": "double",
        "purpose": "string",
    },
)
async def submit_loan(customer_id: str, amount: float, purpose: str) -> dict:
    client = await Client.connect("localhost:7233")
    application = LoanApplication(
        application_id=f"LOAN-{uuid.uuid4().hex[:8]}",
        customer_id=customer_id,
        amount=amount,
        purpose=purpose,
    )

    handle = await client.start_workflow(
        LoanApprovalWorkflow.run,
        application,
        id=f"loan-{application.application_id}",
        task_queue="approval-tasks",
    )

    return {
        "workflow_id": handle.id,
        "application_id": application.application_id,
        "status": "SUBMITTED",
    }

#4.2 工作流状态同步到 Ontology

Python

@activity.defn
async def sync_to_ontology(application_id: str, status: str, details: dict):
    platform = OntoPlatform(base_url="http://localhost:8080", token="service-token")
    platform.objects.update(
        object_type="LoanApplication",
        primary_key=application_id,
        properties={
            "status": status,
            "lastUpdated": datetime.now(timezone.utc).isoformat(),
            **details,
        },
    )

#5. 测试工作流

Python

import pytest
from temporalio.testing import WorkflowEnvironment

@pytest.mark.asyncio
async def test_auto_approve_high_credit():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client,
            task_queue="test-tasks",
            workflows=[LoanApprovalWorkflow],
            activities=[check_credit, send_notification, process_disbursement],
        ):
            result = await env.client.execute_workflow(
                LoanApprovalWorkflow.run,
                LoanApplication("TEST-001", "CUST-HIGH", 20000, "test"),
                id="test-auto-approve",
                task_queue="test-tasks",
            )
            assert result.status == ApprovalStatus.APPROVED
            assert result.transaction_id is not None

@pytest.mark.asyncio
async def test_manual_review_timeout():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client,
            task_queue="test-tasks",
            workflows=[LoanApprovalWorkflow],
            activities=[check_credit, send_notification, process_disbursement],
        ):
            handle = await env.client.start_workflow(
                LoanApprovalWorkflow.run,
                LoanApplication("TEST-002", "CUST-MID", 100000, "test"),
                id="test-timeout",
                task_queue="test-tasks",
            )

            # 快进 7 天
            await env.sleep(timedelta(days=7, seconds=1))

            result = await handle.result()
            assert result.status == ApprovalStatus.REJECTED
            assert "timeout" in result.rejection_reason.lower()

#6. 生产最佳实践

#6.1 工作流设计原则

Workflow 必须是确定性的：不要在 Workflow 中使用 random()、datetime.now() 或 I/O 操作
所有副作用放在 Activity 中：数据库操作、API 调用、文件读写都必须是 Activity
Workflow ID 要有业务含义：如 loan-LOAN-2025-001，便于查询和排重
Activity 必须幂等：同一个 Activity 可能会被重试执行多次

#6.2 错误处理

Python

from temporalio.common import RetryPolicy
from temporalio.exceptions import ActivityError, ApplicationError

# 配置重试策略
retry = RetryPolicy(
    initial_interval=timedelta(seconds=1),
    backoff_coefficient=2.0,
    maximum_interval=timedelta(minutes=5),
    maximum_attempts=5,
    non_retryable_error_types=["ValidationError"],
)

#6.3 监控

Temporal Web UI：访问 http://localhost:8233 查看工作流执行历史
Metrics：Temporal 暴露 Prometheus 指标，关注 workflow_task_schedule_to_start_latency
告警：设置工作流失败率和任务队列积压告警

#总结

本教程覆盖了 Temporal 工作流在 coomia-dip 中的完整使用方法：核心概念（Workflow/Activity/Worker/Signal/Query）、Python SDK 编程模型、与平台 Action 和 Ontology 的集成、测试策略和生产最佳实践。Temporal 让你用线性代码表达复杂的业务流程，而无需手动管理状态持久化、错误重试和超时处理。

下一篇：[S12-17] OSDK TypeScript 前端集成指南 上一篇：[S12-15] Flink CDC 实时数据同步指南