L23 工作流工程实战：让 Agent 跑得像一台"永动机"

• 难度 ⭐⭐⭐⭐⭐（高级）
• 预计阅读 32 分钟
• 代码量 含 checkpoint + 异步 + saga + event-sourcing 完整示例
• 先修 L6（多 Agent）+ L7（Planning）+ L16（MAF）+ L19（生产部署上）+ L20（生产部署下）

课程目标：理解 Agent 工作流与传统工作流（Temporal / Airflow）的本质差异，掌握状态持久化、长任务调度、事件溯源、跨系统集成四大工程模式，能设计一个能"扛得住半夜 3 点告警"的 Agent 工作流。

一、从一个真实事故讲起#

凌晨 3 点，某电商公司的客服 Agent 跑了 6 个小时后——

22:00  Agent 开始处理退款工单
22:05  调用支付系统——成功
22:10  调用订单系统——成功
23:00  调用发票系统——失败（超时）
23:00  Agent 抛异常，整个流程中断
       ←↑↑↑↑ 用户已退款、但订单还是"已付款"状态 ↑↑↑↑
23:30  SRE 收到告警
23:45  人工核对，发现 3 笔订单金额不对
00:30  客服主管打电话给用户道歉

发生了什么？

客服 Agent 是一个"工作流"——它跨多个系统、持续多个步骤、涉及真实资金。但它没有：

• ❌ 状态持久化——进程崩了就丢失中间状态

• ❌ 事务回滚——前面步骤成功了、后面失败了，无法撤销

• ❌ 可恢复性——半夜挂了没人能"接着干"

• ❌ 审计追踪——出错后找不到"谁做的、什么时候、为什么"

这就是 L23 要解决的问题：让 Agent 工作流像 Temporal 一样可靠。

二、用一个比喻：Agent 工作流 = 餐厅的"接力厨房"#

把 Agent 工作流比作一家高端餐厅的厨房：

L23 的目标：让你的 Agent 厨房像米其林餐厅一样——永远运转、出错可恢复、可审计。

三、Agent 工作流 vs 传统工作流：四大差异#

Temporal 博客 - Durable Execution 把"持久化执行"列为关键能力。Agent 工作流比传统工作流（Twitter、Airflow）复杂：

关键反转：

Agent 工作流的"状态"不只是"我在第几步"——还包括"LLM 现在怎么想、下一步要调什么工具、为什么这么决定"。

四、模式 1：状态持久化（Checkpoint）#

4.1 为什么需要？#

Agent 工作流可能运行几小时甚至几天。进程不能一直活着——必须能重启、迁移、扩展。

4.2 LangGraph 的 Checkpoint#

L5 LangGraph 用 checkpointer 持久化状态：

from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver

# 用 PostgreSQL 持久化
checkpointer = PostgresSaver.from_conn_string("postgresql://...")

workflow = StateGraph(MyState)
# ... 加节点和边 ...
app = workflow.compile(checkpointer=checkpointer)

# 第一次运行——执行到第 3 步后崩了
config = {"configurable": {"thread_id": "user-12345"}}
result = await app.send_input(config, {"input": "订会议室"})

# 进程重启后——从断点恢复
new_app = workflow.compile(checkpointer=checkpointer)
result = await new_app.send_input(config, {"input": None})  # 接着上次跑
# ↑↑↑↑ 不会从头开始 ↑↑↑↑

4.3 MAF 的 Thread Store#

L16 MAF 用 Foundry Thread Store：

from agent_framework.azure import AzureOpenAIChatClient

agent = AzureOpenAIChatClient(...).create_agent(
    instructions="...",
    name="BookingAgent"
)

# 创建持久化的 thread
thread = agent.get_new_thread()
await threadstore.persist(thread)  # 显式持久化

# 任意时刻恢复
restored_thread = await threadstore.load(thread.id)
await agent.run("我回来说下，刚才是订会议室", thread=restored_thread)
# ↑↑↑↑ 完整记忆延续 ↑↑↑↑

4.4 选型#

五、模式 2：长任务调度（Async / Saga）#

5.1 真实场景：30 天的"用户入职流程"#

新员工入职要经历 20 个步骤：签合同、申请电脑、申请门禁、申请邮箱、加入 Slack……每个步骤可能要等几天（HR 审批、IT 配硬件）。

如果一个 Agent 同步跑完——要在线 30 天？不可能。

5.2 异步 + Saga 模式#

# 步骤 1：启动工作流，立刻返回 task_id
task_id = await start_onboarding_workflow(user_id)
# return "你的入职流程已启动，task_id = abc-123"

# 工作流引擎异步驱动
async def onboarding_workflow(user_id: str):
    # Step 1: 签合同（同步：调用 DocuSign API）
    contract = await sign_contract(user_id)
    
    # Step 2: 申请电脑（异步：提交工单，等 IT 完成）
    request_id = await submit_it_request(user_id, "laptop")
    # ↑↑↑↑ 不等 IT 完成，先继续下一步 ↑↑↑↑
    
    # Step 3: 申请邮箱（异步）
    email = await create_email(user_id)
    
    # Step 4: 等 IT 完成（用 Saga + 补偿）
    laptop_status = await wait_for_it(request_id, timeout_days=7)
    if laptop_status == "delivered":
        # Step 5: 发欢迎邮件
        await send_welcome_email(user_id, email, contract)
    else:
        # Saga 补偿——撤销前面的步骤
        await cancel_email(email)
        await notify_hr(user_id, "IT 超时")

5.3 Saga 模式的三个原则#

1. 每步必须有补偿——成功做 Step 3action，失败做 compensating_action。

2. 不能有"无法撤销"的步骤——比如"已发货"必须能"召回"。

3. 最终一致性——不是强一致性，但可保证业务结果正确。

5.4 在退款 Agent 中的 Saga：#

用户申请退款
   ↓ (action) 调用支付系统退款
   ├─ 退款成功 → (action) 标记订单为"已退款"
   ├─ 退款成功但订单标记失败 → (compensation) 通知人工处理
   └─ 退款失败 → (compensation) 告知用户"退款失败，请重试"

六、模式 3：错误恢复（Retry / Dead Letter / HITL）#

6.1 错误分类#

6.2 指数退避重试#

import asyncio
import random

async def retry_with_backoff(coro, max_retries=5):
    """指数退避 + 抖动"""
    for attempt in range(max_retries):
        try:
            return await coro
        except TransientError as e:
            if attempt == max_retries - 1:
                raise
            # 2^attempt 秒 + 抖动（防 thundering herd）
            wait = (2 ** attempt) + random.uniform(0, 1)
            await asyncio.sleep(wait)

# 用法
result = await retry_with_backoff(
    lambda: refund_service.refund(order_id, amount)
)

6.3 死信队列（Dead Letter Queue）#

不是所有错误都该重试——比如"调用支付系统返回余额不足"。重试 100 次也没用。

async def call_payment_with_dlq(order_id, amount):
    try:
        return await payment.refund(order_id, amount)
    except InsufficientFunds:
        # 直接进死信队列，等人工处理
        await dlq.enqueue({
            "type": "refund_failed",
            "order_id": order_id,
            "amount": amount,
            "reason": "insufficient_funds",
            "needs_human": True
        })
        raise  # 重新抛，让工作流进入"停止回退"分支

6.4 HITL 介入#

L20 Foundry 部署 提到 @tool(approval_mode="require")——但生产实践要更细：

# 三级 HITL 策略
HITL_POLICY = {
    # 低风险：自动执行
    "search_doc": "auto",
    "query_order": "no",
    
    # 中风险：批量后审
    "create_ticket": "batch_review",  # 每 50 个一起审
    
    # 高风险：单笔必审
    "issue_refund": "single_review",  # 每笔都审
    "delete_data": "single_review",
    
    # 禁止操作：永远不自动
    "transfer_money": "forbidden"
}

def should_request_human_approval(tool_name, args, context):
    """决定是否需要人介入"""
    mode = HITL_POLICY.get(tool_name, "auto")
    
    if mode == "forbidden":
        raise ForbiddenToolError(f"{tool_name} 不能自动执行")
    
    if mode == "single_review":
        return True  # 必须人工审
    
    if mode == "batch_review":
        # 同类请求超过 N 个，一起审
        return context.get("batch_size", 0) >= 50
    
    return False  # auto

七、模式 4：跨系统集成（Webhook / Event-Driven）#

7.1 场景：Agent 需要等"外部事件"#

很多业务是事件驱动的：

• 用户付款了 → 发货

• 简历收到 → HR 审核

• 仓库温度超标 → 报警

Agent 不能一直 poll——必须用 Webhook / Event Bus。

7.2 Webhook 接收#

from fastapi import FastAPI, Request

app = FastAPI()

# 外部系统调我们
@app.post("/webhook/payment-completed")
async def on_payment_completed(req: Request):
    payload = await req.json()
    # payload = {"order_id": "123", "amount": 100, "status": "paid"}
    
    # 找到对应的工作流实例
    workflow_id = payload["workflow_id"]
    workflow = await workflowstore.load(workflow_id)
    
    # 唤醒等待中的工作流——从 "等支付" 步骤继续
    await workflow.resume({"payment_status": "paid"})
    return {"job": "resumed"}

7.3 Event-Driven 架构#

用户提交申请
   ↓
事件总线（EventBridge / Kafka / Service Bus）
   ↓
   ├─ Agent 1 监听"申请创建"事件 → 自动审核
   ├─ Agent 2 监听"待 HR"事件 → HR 审批
   └─ Agent 3 监听"已批准"事件 → 自动开通账号

# Agent 3：监听"已批准"事件
@event_bus.subscribe("employee.approved")
async def on_employee_approved(event):
    user_id = event["user_id"]
    
    # 创建账号（多步工作流）
    async with saga() as s:
        await s.step("create_email", lambda: create_email(user_id))
        await s.step("create_slack", lambda: create_slack(user_id))
        await s.step("assign_laptop", lambda: assign_laptop(user_id))
        
        if s.failed:
            await s.compensate_all()  # 自动回滚
        else:
            await send_welcome_email(user_id)

7.4 为什么不用 Polling？#

八、模式 5：事件溯源与审计（Event Sourcing）#

8.1 状态 vs 事件#

两种持久化思路：

# 思路 A：存状态
order = {
    "status": "shipped",
    "address": "北京",
    "tracking_id": "SF123"
}
# ↑↑↑↑ 现在什么状态 ↑↑↑↑

# 思路 B：存事件流
events = [
    {"type": "order.created", "ts": "...", "data": {...}},
    {"type": "address.changed", "ts": "...", "data": {...}},
    {"type": "warehouse.assigned", "ts": "...", "data": {...}},
    {"type": "shipped", "ts": "...", "data": {...}},
]
# ↑↑↑↑ 怎么一步步到这里的 ↑↑↑↑

8.2 事件溯源的好处#

审计 + 重放 + 调试 + LLM 决策解释：

# L22 的签名收据可以接入事件流
async def execute_with_audit(tool_call):
    """执行工具 + 写事件 + 签名"""
    receipt = await sign_receipt(
        agent_id=AGENT_ID,
        tool=tool_call.name,
        args=tool_call.args,
        result=result
    )
    
    await event_store.append({
        "type": "tool.executed",
        "ts": datetime.utcnow(),
        "agent_id": AGENT_ID,
        "tool": tool_call.name,
        "args_hash": hash(tool_call.args),
        "result_hash": hash(result),
        "receipt": receipt  # L22 的 Ed25519 签名
    })

8.3 用事件流调试 LLM#

事件流让"为什么 LLM 这么做"可解释：

# 查"Agent 为什么推荐 A 套餐"
events = await event_store.query(
    agent_id="recommendation-agent",
    session_id="user-12345",
    types=["llm.thought", "tool.executed"]
)

# 重放 LLM 的思考过程
for event in events:
    if event.type == "llm.thought":
        print(f"[{event.ts}] LLM 思考: {event.thought}")
    elif event.type == "tool.executed":
        print(f"[{event.ts}] 调用 {event.tool}({event.args})")

九、模式 6：可观测性（不只是 metrics）#

Honeycomb - The Three Pillars of Observability 提到：logs、metrics、traces 只是工具，不是目标。

Agent 工作流的可观测性需要四件事：

Reasoning Trace 是 Agent 独有的——它记录每一步 LLM 的 thought + tool_call + observation：

# LangSmith 自动记录
from langsmith import trace

@trace
async def my_agent_step(state):
    # LangSmith 自动捕获：
    # - LLM 调用（prompt、completion、tokens）
    # - 工具调用（args、result、latency）
    # - 工作流转换（from_state → to_state）
    return await llm.invoke(...)

十、综合案例：退款 Agent 的完整工作流#

把上面的所有模式串起来：

from temporalio import workflow, activity
from datetime import timedelta

@workflow.defn
class RefundWorkflow:
    """退款工作流——综合所有 L23 模式"""
    
    @workflow.run
    async def run(self, order_id: str, amount: float, user_id: str):
        # 1. 持久化当前状态（模式 1：Checkpoint）
        # Temporal 自动持久化——崩了能续
        
        # 2. Saga + 补偿（模式 2）
        async with saga() as s:
            # Step 1: 验证订单（重试 + 死信）
            try:
                order = await workflow.execute_activity(
                    "verify_order",
                    order_id,
                    start_to_close_timeout=timedelta(seconds=30),
                    retry_policy=RetryPolicy(
                        max_attempts=3,
                        initial_interval=timedelta(seconds=1),
                        backoff_coefficient=2.0
                    )
                )
            except ActivityFailure:
                await workflow.execute_activity("notify_user_invalid_order", user_id)
                return "rejected"
            
            # Step 2: 调用支付（高风险——HITL 审批）
            if await workflow.execute_activity(
                "requires_human_approval", 
                {"amount": amount}
            ):
                # 模式 3：HITL 介入
                approval = await workflow.wait_for_signal("approve_refund")
                if approval != "approved":
                    return "rejected_by_human"
            
            # Step 3: 执行退款
            payment_result = await workflow.execute_activity(
                "process_refund",
                {"order_id": order_id, "amount": amount}
            )
            
            # Step 4: 更新订单状态
            await workflow.execute_activity(
                "update_order_status", 
                {"order_id": order_id, "status": "refunded"}
            )
            
            # Step 5: 通知（event-driven 模式 4）
            await workflow.execute_activity(
                "send_notification",
                {"user_id": user_id, "type": "refund_success"}
            )
            
            # Step 6: 写事件（模式 5）
            await workflow.execute_activity(
                "audit_log",
                {
                    "event": "refund.completed",
                    "order_id": order_id,
                    "amount": amount,
                    "receipt": payment_result.receipt  # L22 签名
                }
            )
        
        return "completed"

这一个工作流用到了 6 个 L23 模式——这就是生产级 Agent 的标准模板。

十一、与现有课程的网络#

十二、动手时间：60 分钟搭一个"不会丢"的 Agent#

步骤 1：跑通 LangGraph Checkpoint（15 分钟）#

pip install langgraph langgraph-checkpoint-postgres

from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver

# 内存版 checkpoint——快速试
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
await app.ainvoke({"input": "..."}, config)
# 第二次调用同一个 thread_id——会延续状态

步骤 2：加 HITL 中断（20 分钟）#

from langgraph.checkpoint.memory import MemorySaver

workflow.add_node("dangerous_action", dangerous_action_node)

# 在危险操作前中断
app = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["dangerous_action"]
)

# 第一次调用——到 dangerous_action 前停下来
result = await app.ainvoke({"input": "..."}, config)
print(result["__interrupt__"])  # 提示用户确认

# 用户批准后——继续
await app.ainvoke(None, config)  # None = 接着上次跑

步骤 3：加死信队列（15 分钟）#

dead_letter = []

async def call_with_dlq(fn, *args, **kwargs):
    for attempt in range(3):
        try:
            return await fn(*args, **kwargs)
        except PermanentError as e:
            dead_letter.append({
                "fn": fn.__name__,
                "args": args,
                "error": str(e),
                "ts": datetime.utcnow()
            })
            raise
        except TransientError:
            await asyncio.sleep(2 ** attempt)

步骤 4：把签名收据（L22）接到事件流（10 分钟）#

from langgraph.checkpoint.memory import MemorySaver

audit_log = []

async def audited_invoke(tool_call):
    """调用工具 + 写事件 + L22 签名"""
    result = await call_with_dlq(...)
    
    receipt = await sign_receipt(
        tool=tool_call.name,
        args=tool_call.args,
        result=result,
        agent_id=AGENT_ID
    )
    
    audit_log.append({
        "ts": datetime.utcnow(),
        "type": "tool.executed",
        "tool": tool_call.name,
        "receipt": receipt
    })
    return result

十三、生产检查清单#

状态持久化#

长任务调度#

错误恢复#

跨系统集成#

审计#

十四、关键要点回顾#

核心原则：Agent 工作流不是"LLM 调几个 API"——它是一个分布式系统。需要按分布式系统的标准来设计：状态、事务、可观测、审计缺一不可。

十五、参考资源#

• Temporal - Durable Execution — 持久化执行的工作流引擎事实标准

• Microsoft Agent Framework Workflows — MAF 的 5 种编排模式

• LangGraph Persistence — StateGraph 的 checkpoint 实现

• Saga Pattern — 微服务事务模式

• Event Sourcing — Martin Fowler 的事件溯源定义

• Honeycomb - Observability 101 — 可观测性三大支柱

• 上一课：L22 安全的 Agent

十六、课程总结#

到这里你已经完成了 L0–L23——覆盖了从"什么是 Agent"到"Agent 工作流的工程实战"的完整路径。

💡 想知道怎么把 Agent 部署到 多模态 场景、怎么调试线上故障、怎么防御 Prompt 注入、怎么写测试保证质量？见 L24-L27 高级专题.

你现在拥有的能力：

1. 概念理解：ReAct、Function Calling、Multi-Agent、Context Engineering、Planning、Memory、Metacognition

2. 协议掌握：MCP / A2A / NLWeb 三大协议

3. 框架选型：LangGraph / AutoGen / CrewAI / smolagents / MAF / Semantic Kernel

4. 架构设计：从单 Agent 到多 Agent，从本地到云端，从原型到生产

5. 生产部署：可观测性、Foundry 托管、HITL、成本优化

6. 安全：签名收据、零信任、加密审计

7. 工作流工程：Checkpoint、Saga、HITL、Event-Driven、Event Sourcing

最后一句话——

Agent 工程不是关于"用什么框架"，而是关于"如何让 LLM 真正解决用户问题"。

框架会变、模型会变、协议会变，但"理解问题、设计可靠流程、持续评估改进"的能力是永恒的。

💬 有疑问？ 试试问我："Saga 的补偿怎么写？"、"HITL 该在哪些操作前插入？"、"Temporal vs LangGraph 怎么选？"——我会用代码或比喻为你解答。