Agent工作流实例入门指南：从零搭建高可靠自动化流程

1次阅读

没有评论

共计 2776 个字符，预计需要花费 7 分钟才能阅读完成。

Agent 工作流可以类比为快递配送系统：

触发器（Trigger）：相当于用户的下单请求，当特定事件（如新订单生成）或时间条件（如每小时整点）发生时启动工作流
执行器（Agent）：如同快递分拣中心，负责接收事件并执行业务逻辑（状态检查、数据处理等）
动作（Action）：类似快递员送货，是工作流的终端操作（写数据库、发通知等）

与传统脚本的关键差异在于：

状态感知 ：Agent 能感知前序步骤的执行结果（如通过context_id 追踪流程）
失败恢复：支持从特定步骤重试而非全量重启
资源隔离：单个 Agent 崩溃不会导致整个系统瘫痪

典型问题场景：

长事务阻塞：单个耗时任务占用数据库连接，导致后续步骤超时
状态雪崩：未处理的异常引发级联故障（如：支付状态未回滚导致库存不一致）
幽灵执行：崩溃后的自动重启导致重复执行关键操作

from typing import Optional, Dict
from datetime import timedelta
from pydantic import BaseModel

class OrderEvent(BaseModel):
    order_id: str
    status: str
    context_id: Optional[str] = None  # 工作流上下文标识

class OrderSyncAgent:
    def __init__(self, max_retries: int = 3):
        self.retry_policy = {1: timedelta(seconds=1),
            2: timedelta(seconds=5),
            3: timedelta(seconds=30)
        }

    def handle_event(self, event: OrderEvent) -> Dict:
        """
        核心事件处理逻辑（含错误回退）:param event: 订单事件数据
        :return: 执行结果（含重试建议）"""
        try:
            # 步骤 1：检查事件是否已处理（幂等性设计）if self._is_duplicate(event):
                return {'status': 'skipped', 'reason': 'duplicate'}

            # 步骤 2：同步订单状态（伪代码）sync_result = external_api.sync_status(
                order_id=event.order_id,
                status=event.status
            )

            # 步骤 3：记录执行上下文
            self._save_context(
                context_id=event.context_id,
                step="order_synced",
                data=sync_result
            )

            return {'status': 'success', 'data': sync_result}
        except TemporaryError as e:  # 可重试异常
            return {
                'status': 'retry',
                'delay': self.retry_policy.get(e.retry_count, timedelta(minutes=5))
            }
        except BusinessError as e:  # 业务异常需人工干预
            self._trigger_alert(f"订单同步失败: {event.order_id}")
            return {'status': 'failed', 'error': str(e)}

关键设计要点：

事件去重 ：通过_is_duplicate 方法检查事件指纹（如：order_id+status 的 MD5）
错误分类：区分临时性错误（网络抖动）和业务错误（订单不存在）
上下文传递 ：使用context_id 贯穿整个工作流生命周期

错误类型	重试次数	退避策略	最终处理方式
网络超时	3	指数退避(1s/5s/30s)	人工介入
数据库死锁	2	固定间隔(10s)	自动回滚
第三方 API 限流	5	随机退避(5-15s)	进入降级流程

import logging

# 工作流引擎日志（DEBUG 级别存入单独文件）engine_logger = logging.getLogger('workflow.engine')
engine_handler = logging.FileHandler('/var/log/workflow_engine.log')
engine_logger.addHandler(engine_handler)

# 业务日志（INFO 级别输出到控制台）biz_logger = logging.getLogger('workflow.business')
biz_logger.setLevel(logging.INFO)
console_handler = logging.StreamHandler()
biz_logger.addHandler(console_handler)

扩展性限制：有状态的 Agent 无法水平扩展
恢复复杂性：崩溃后难以重建内存状态
调试困难：状态不一致问题难以追踪

网络超时：所有外部调用必须设置 connect/read 双超时
CPU 超时：长时间计算任务需添加中断检查点
死锁检测：数据库事务配置锁等待阈值
依赖超时：组合服务设置级联超时（如：总耗时≤2s）

# 超时控制示例
from concurrent.futures import TimeoutError

def sync_with_timeout(order_id: str, timeout: float) -> bool:
    try:
        with ThreadPoolExecutor() as executor:
            future = executor.submit(external_api.sync_status, order_id)
            return future.result(timeout=timeout)
    except TimeoutError:
        metrics.counter("timeout_errors", tags=["method:sync_status"])
        return False