Mistral Workflows 实战：用 Temporal 引擎在 30 分钟把 AI 流程从 PoC 推到生产

💡 一句话总结：Mistral Workflows 不是又一个 Agent 框架，是把 Netflix / Stripe 早就在用的 Temporal durable execution 引擎给 AI 流程做了 LLM-native 封装。LangGraph 是『可用』，Workflows 是『可生产』。

为什么需要又一个编排引擎

过去 12 个月，我们看着团队拿 LangGraph 跑的 Agent 频繁出问题：

• worker 重启丢上下文，长流程必须重头跑（一次 RAG + LLM 调用成本 $5-10）
• checkpoint 写到 SQLite 后跨多个 worker 节点不一致
• 异常处理代码量超过业务代码本身，超时/重试/补偿散落在各处
• 业务方想看『现在跑到哪一步』，只能让开发者打日志再 grep

LangGraph 1.2 在 5/14 加了 per-node timeout、error recovery、graceful shutdown，但底层仍是『进程内状态』模型——只要 K8s 把 pod 杀掉，进程外的状态就只能祈祷 checkpoint 写全了。

Mistral 的判断是：AI 流程已经到了需要 Temporal 这种 durable execution 引擎托底的阶段。4 月 28 日他们把 Workflows 推向公开预览时，Le Chat 已经在生产跑了几个月，单日执行数百万次。

30 分钟跑通一个发票审批流

我们从一个真实场景起步：每天上千张发票需要解析、和 ERP 对账、超过阈值送人工审批、最终入账。这种流程有三个让传统编排崩溃的点：

1. 长持续：等待人工审批可能 8 小时也可能 3 天
2. 混合 IO：要调 LLM（解析）、ERP API（查 PO）、Slack（通知）、数据库（落账）
3. 必须可重放：审计要求每张发票的处理路径都能追溯

项目初始化

pip install mistralai mistralai-workflows temporalio
mistral-workflows init invoice-approval
cd invoice-approval

CLI 会生成：

invoice-approval/
├── workflows/
│   └── invoice.py          # 你的 workflow 定义
├── activities/
│   ├── llm.py              # LLM 调用 activity
│   ├── erp.py              # ERP 接入 activity
│   └── notify.py           # 通知 activity
├── helm/
│   └── values.yaml         # Worker 部署
├── studio.yaml             # Studio 注册元数据
└── .env.example

写第一个 workflow

from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy

with workflow.unsafe.imports_passed_through():
    from activities.llm import parse_invoice
    from activities.erp import lookup_po, post_to_gl
    from activities.notify import slack_request_approval

@workflow.defn
class InvoiceApproval:
    @workflow.run
    async def run(self, invoice_url: str) -> dict:
        retry_3x = RetryPolicy(maximum_attempts=3)

        parsed = await workflow.execute_activity(
            parse_invoice,
            invoice_url,
            start_to_close_timeout=timedelta(minutes=2),
            retry_policy=retry_3x,
        )

        po = await workflow.execute_activity(
            lookup_po,
            parsed["po_number"],
            start_to_close_timeout=timedelta(seconds=10),
        )

        if parsed["amount"] > 10000:
            decision = await workflow.execute_activity(
                slack_request_approval,
                {"invoice": parsed, "po": po},
                start_to_close_timeout=timedelta(hours=72),
            )
            if decision != "approved":
                return {"status": "rejected", "reason": decision}

        gl_entry = await workflow.execute_activity(
            post_to_gl,
            {"invoice": parsed, "po": po},
            start_to_close_timeout=timedelta(seconds=30),
        )

        return {"status": "posted", "gl_id": gl_entry["id"]}

注意三个关键点：

• workflow.execute_activity 包住所有外部 IO，框架会把每次调用的入参出参写入 event log
• slack_request_approval 的 timeout 是 72 小时，但 worker 不需要一直在跑——Temporal 自动 park，外部信号到达时再 wake up
• workflow 函数本身必须 deterministic，不能直接调 datetime.now() 或 requests.get()，要包成 activity

activity 实现示例

from temporalio import activity
from mistralai import Mistral

mistral = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

@activity.defn
async def parse_invoice(invoice_url: str) -> dict:
    activity.logger.info(f"parsing {invoice_url}")
    resp = await mistral.chat.complete_async(
        model="mistral-large-2026",
        messages=[
            {"role": "system", "content": "提取发票字段，返回 JSON"},
            {"role": "user", "content": [
                {"type": "document_url", "document_url": invoice_url}
            ]},
        ],
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content)

activity 可以失败、可以重试、可以超时，但每次重试都是『从 activity 边界重新开始』，workflow 主流程的状态不会丢。

部署 worker 到 K8s

# helm/values.yaml
worker:
  image: registry.company.com/invoice-approval:0.1.0
  replicas: 3
  env:
    MISTRAL_API_KEY:
      valueFrom:
        secretKeyRef:
          name: mistral-keys
          key: api-key
    ERP_BASE_URL: https://erp.internal.company.com
  resources:
    requests:
      cpu: 500m
      memory: 1Gi

helm install invoice-worker mistral-workflows/worker -f helm/values.yaml

worker 启动后会自动连接 Mistral 托管的 Temporal cluster，注册自己能处理的 task queue。Mistral 的控制平面看到 worker 上线，就把所有标记 invoice-approval 的 workflow 实例分发过来。

Studio 里发生了什么

打开 studio.mistral.ai，进入 Workflows tab，能看到刚部署的 InvoiceApproval。每次执行的页面包含：

• Timeline：每个 activity 的开始时间、耗时、状态、入参出参
• Replay：可以在 UI 里手动 replay 任意历史执行，输入相同参数验证 deterministic
• Errors：失败的 activity 会高亮，点击进去能看 stack trace 和 retry 历史
• Audit log：谁在什么时间触发了哪个执行，导出 CSV 给合规

最有用的是『signal』和『query』两个接口。signal 让外部系统（比如 Slack bot）能往运行中的 workflow 发消息（『审批结果是 approved』）；query 让运维能问『这个 workflow 现在在哪一步』而不影响主流程。

Le Chat 集成：让业务人员自然语言触发

在 studio.yaml 里声明：

name: invoice-approval
display_name: 发票审批
description: 自动解析发票、对账、超阈值审批、入账
trigger:
  type: chat_action
  examples:
    - "帮我跑一遍今天的发票审批"
    - "审批这张发票 https://files/inv-2026-05-001.pdf"
inputs:
  - name: invoice_url
    type: url
    description: 发票 PDF 链接
permissions:
  roles:
    - finance.operator
    - finance.manager

部署后业务人员在 Le Chat 里输入『审批这张发票 https://…』，Le Chat 自动识别意图、参数提取、触发 workflow、流式显示执行进度。审批环节的 Slack 通知也会附带『回到 Le Chat 查看完整 timeline』的链接。

容错策略的三层防御

生产部署后我们踩过的坑，浓缩成三条 best practice：

第一层：activity 内部用 RetryPolicy 处理瞬时故障

RetryPolicy(
    initial_interval=timedelta(seconds=1),
    backoff_coefficient=2.0,
    maximum_attempts=5,
    non_retryable_error_types=["InvalidInvoiceFormat"],
)

业务校验错误（发票格式不对）不要重试，会浪费 LLM tokens；网络/超时错误指数退避重试。

第二层：workflow 主流程用 try/except 处理业务异常

try:
    parsed = await workflow.execute_activity(parse_invoice, ...)
except ActivityError as e:
    await workflow.execute_activity(
        notify_ops, f"发票解析失败：{e}",
        start_to_close_timeout=timedelta(seconds=10),
    )
    return {"status": "manual_review", "reason": str(e)}

第三层：用 saga 模式做补偿

如果『入账』成功但『更新 ERP 状态』失败，需要回滚入账。Workflows 推荐显式写补偿 activity 而不是依赖事务：

compensations = []
try:
    gl_id = await workflow.execute_activity(post_to_gl, ...)
    compensations.append(lambda: workflow.execute_activity(reverse_gl, gl_id))

    await workflow.execute_activity(update_erp_status, parsed["po_number"])
except Exception:
    for comp in reversed(compensations):
        await comp()
    raise

成本和性能账

跑了一个月的真实数据（每天约 800 张发票）：

按工程师 $80/h 算，LangGraph 方案隐性成本 $3250，Workflows 方案 $2140，单月节省 $1100，年化 $13K。规模到日万级发票后差距会进一步拉大。

什么时候不该用 Workflows

不是所有 AI 流程都需要 durable execution：

• 单轮聊天 / RAG 问答：响应时间秒级，无副作用，用普通 API 就好
• 纯实验性 demo：写 @workflow.defn 的开发成本比 LangGraph 高 30%，PoC 阶段没必要
• 极致延迟敏感：Temporal 每个 activity 边界有 5-20ms 的持久化开销，做 streaming 聊天会感知到

判断标准简单：『如果这个流程跑失败一次，业务方会要求你排查到底跑了什么』，就该上 Workflows。

下一步

Mistral 在 5/22 announce 了 Workflows 的几个新特性：

• Python SDK 1.1：支持本地 mistral-workflows dev 起开发 server，不再需要连远端 Temporal cluster
• Workflow Templates：官方提供发票审批、合同审查、HR 入职、客服 triage 四个开源模板，clone 即用
• MCP 集成：workflow 可以直接调用 MCP server 暴露的工具，等于把 OpenClaw 生态接进来

要上手，先从 mistral-workflows init 跑个 hello-world，再 fork 一个模板改造，比从零搭强。文档在 mistral.ai/docs/workflows，社区频道在 Discord 的 #workflows 子频道。

Agent 框架的『可用性竞赛』2024-2025 卷完了，2026 的关键词是『可生产』。Workflows 是这条路线上目前最清晰的答案。