AI Agent工作流搭建实战：从零构建高可用自动化流程

1次阅读

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

在医疗报告生成场景中，典型的处理流程包含：影像数据预处理(DICOM 解析)、AI 模型推理、报告结构化生成、医生审核反馈循环。当用线性脚本实现时，会遇到三个典型问题：

异步依赖管理失控：CT 影像预处理完成后才能启动肺部结节检测模型，但脚本难以优雅处理这种跨进程依赖
错误恢复成本高：当血液分析模型失败时，需要手动清理半成品数据才能重试
监控盲区：无法直观掌握各环节耗时分布，难以定位性能瓶颈

维度	Airflow	Luigi	自建 DAG 引擎
学习曲线	陡峭(需学 Operator)	中等	灵活可控
调度粒度	分钟级	任务级	可自定义
强依赖	Celery/K8s	本地文件系统	无
适用场景	企业级定时任务	数据管道	高定制 AI 流程

采用 NetworkX 构建轻量级 DAG 引擎，关键组件包括：

from typing import Protocol, runtime_checkable
import networkx as nx

@runtime_checkable
class WorkflowNode(Protocol):
    node_id: str
    max_retries: int

    def execute(self, context: dict) -> dict:
        ...

class DAGEngine:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.metrics = PrometheusClient()

    def add_node(self, node: WorkflowNode) -> None:
        self.graph.add_node(node.node_id, instance=node)

    def add_dependency(self, from_node: str, to_node: str) -> None:
        if nx.has_path(self.graph, to_node, from_node):
            raise ValueError(f"循环依赖检测: {from_node} -> {to_node}")
        self.graph.add_edge(from_node, to_node)

from dataclasses import dataclass
from typing import Any, Optional
import time

@dataclass
class BaseNode:
    node_id: str
    max_retries: int = 3
    retry_delay: float = 1.0

    def __call__(self, context: dict[str, Any]) -> dict[str, Any]:
        last_error: Optional[Exception] = None

        for attempt in range(self.max_retries + 1):
            try:
                result = self._execute(context)
                self._log_success(context)
                return result
            except Exception as e:
                last_error = e
                if attempt < self.max_retries:
                    time.sleep(self.retry_delay * (attempt + 1))

        raise RuntimeError(f"节点 {self.node_id} 重试 {self.max_retries} 次仍失败") from last_error

    def _execute(self, context: dict[str, Any]) -> dict[str, Any]:
        raise NotImplementedError

    def _log_success(self, context: dict[str, Any]) -> None:
        # 自动脱敏敏感字段
        safe_ctx = {k: '*****' if 'password' in k.lower() else v 
            for k, v in context.items()}
        logger.info(f"节点 {self.node_id} 执行完成: {safe_ctx}")

from concurrent.futures import ThreadPoolExecutor
from collections import deque

class ParallelExecutor:
    def __init__(self, max_workers: int = 4):
        self.ready_queue = deque()
        self.running = set()
        self.lock = threading.Lock()
        self.executor = ThreadPoolExecutor(max_workers=max_workers)

    def schedule(self, dag: nx.DiGraph) -> None:
        # 基于拓扑排序的任务调度
        for node in nx.topological_generations(dag):
            with self.lock:
                self.ready_queue.extend(node)

        while self.ready_queue or self.running:
            if self.ready_queue and len(self.running) < self.executor._max_workers:
                node = self.ready_queue.popleft()
                future = self.executor.submit(
                    self._wrap_execution,
                    dag.nodes[node]['instance']
                )
                future.add_done_callback(self._on_complete)
                self.running.add(future)
            time.sleep(0.1)

    def _wrap_execution(self, node: WorkflowNode) -> None:
        # Prometheus 监控埋点
        with self.metrics.timer(f'node_{node.node_id}_duration'):
            return node.execute({})

    def _on_complete(self, future) -> None:
        with self.lock:
            self.running.remove(future)
            if future.exception():
                self.metrics.incr('workflow_errors')

使用生成 10 万节点的随机 DAG 进行基准测试：

import numpy as np

def benchmark():
    graph = nx.gnp_random_graph(100000, 0.0001, directed=True)
    dag = nx.DiGraph([(u, v) for (u, v) in graph.edges() if u < v])

    # 内存跟踪
    import tracemalloc
    tracemalloc.start()

    start = time.perf_counter()
    list(nx.topological_sort(dag))  # 关键路径
    elapsed = time.perf_counter() - start

    current, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()

    print(f"拓扑排序耗时: {elapsed:.2f}s")
    print(f"内存占用峰值: {peak / 1024**2:.2f} MB")

典型结果：
– 拓扑排序耗时：1.87s
– 内存占用：142.31MB

def check_memory_leak():
    import tracemalloc

    tracemalloc.start()
    snapshot1 = tracemalloc.take_snapshot()

    # 执行可疑代码
    leaky_list = []
    for _ in range(1000):
        leaky_list.append(np.zeros(1024))

    snapshot2 = tracemalloc.take_snapshot()
    top_stats = snapshot2.compare_to(snapshot1, 'lineno')

    for stat in top_stats[:3]:
        print(stat)

循环依赖检测
使用 NetworkX 的 has_path 在添加边时实时检查
对于批量导入，建议实现基于 Tarjan 的 SCC(强连通分量)算法
幂等性保证
每个节点实现 fingerprint() 方法计算输入特征哈希

持久化执行记录到 SQLite，关键逻辑：

def ensure_idempotent(node: WorkflowNode, ctx: dict) -> bool:
    fp = hashlib.md5(node.node_id.encode() + 
        json.dumps(ctx, sort_keys=True).encode()).hexdigest()

    with sqlite3.connect('workflow.db') as conn:
        cursor = conn.execute(
            "SELECT 1 FROM executions WHERE fingerprint = ?", 
            (fp,)
        )
        if cursor.fetchone():
            return False
        conn.execute("INSERT INTO executions VALUES (?, datetime('now'))",
            (fp,)
        )
    return True