基于Agent流程图的高效任务编排：从设计到落地实践

1次阅读

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

在微服务和自动化任务场景中，开发者经常遇到以下典型问题：

依赖地狱：任务之间存在复杂的先后关系，手动管理容易出错
状态追踪困难：分布式环境下难以直观查看任务执行进度
容错性差：单个节点失败可能导致整个流程阻塞

这些问题在 ETL 数据处理、机器学习流水线等场景尤为突出。传统解决方案（如直接编码调用链）往往导致代码臃肿且难以维护。

对比三种主流方案：

方案类型	优点	缺点
DSL 脚本	灵活度高	学习成本高，调试困难
低代码平台	可视化操作	扩展性有限
Agent 流程图(DAG)	拓扑结构清晰	需实现执行引擎

有向无环图 (Directed Acyclic Graph, DAG) 因其天然适合表达任务依赖关系，成为流程编排的理想选择。Airflow、Kubeflow 等知名项目均采用此模型。

from abc import ABC, abstractmethod
from typing import Dict, Any

class BaseAgent(ABC):
    """
    Agent 基类定义
    Attributes:
        agent_id: 节点唯一标识
        upstream: 上游依赖节点 ID 列表
    """
    def __init__(self, agent_id: str, upstream: list = None):
        self.agent_id = agent_id
        self.upstream = upstream or []

    @abstractmethod
    def execute(self, context: Dict[str, Any]) -> Any:
        """执行核心业务逻辑"""
        pass

    def __repr__(self):
        return f"<Agent {self.agent_id}>"

from collections import deque

def topological_sort(agents: list[BaseAgent]) -> list[BaseAgent]:
    """
    基于 Kahn 算法的拓扑排序
    返回：按执行顺序排列的 Agent 列表
    """
    # 构造入度表和邻接表
    in_degree = {a.agent_id: 0 for a in agents}
    graph = {a.agent_id: [] for a in agents}

    for agent in agents:
        for dep in agent.upstream:
            graph[dep].append(agent.agent_id)
            in_degree[agent.agent_id] += 1

    # 初始化队列
    queue = deque([aid for aid, count in in_degree.items() if count == 0])
    sorted_agents = []

    # 拓扑排序主逻辑
    while queue:
        current = queue.popleft()
        sorted_agents.append(next(a for a in agents if a.agent_id == current))

        for neighbor in graph[current]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)

    if len(sorted_agents) != len(agents):
        raise ValueError("存在循环依赖")

    return sorted_agents

class CircuitBreaker:
    """
    简易断路器实现
    连续失败超过阈值后暂停执行
    """
    def __init__(self, max_failures=3, reset_timeout=60):
        self.failure_count = 0
        self.max_failures = max_failures
        self.reset_timeout = reset_timeout
        self.last_failure_time = None

    def is_available(self) -> bool:
        if self.last_failure_time and \
           (time.time() - self.last_failure_time) > self.reset_timeout:
            self.reset()
        return self.failure_count < self.max_failures

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()

    def reset(self):
        self.failure_count = 0
        self.last_failure_time = None

批处理：对数据库 /API 调用进行批量操作
示例：将多个数据插入合并为批量 INSERT
效果：减少网络往返开销
异步 IO：使用 asyncio 提升 I / O 密集型任务吞吐量
注意：需确保 Agent 实现是线程安全的
缓存策略：
本地缓存：高频读取但很少变更的数据
分布式缓存：跨节点共享状态

在前文拓扑排序实现中，如果最终排序结果数量小于节点总数，则说明存在循环依赖。可通过深度优先搜索 (DFS) 进行更细致的检测：

def has_cycle(graph: Dict[str, list[str]]) -> bool:
    visited = set()
    recursion_stack = set()

    def dfs(node):
        if node in recursion_stack:
            return True
        if node in visited:
            return False

        visited.add(node)
        recursion_stack.add(node)

        for neighbor in graph.get(node, []):
            if dfs(neighbor):
                return True

        recursion_stack.remove(node)
        return False

    return any(dfs(node) for node in graph)