当前位置：首页 > news >正文

第五章：Execution Flow Framework

news 来源：原创 2025/5/25 10:15:46

Chapter 5: Execution Flow Framework

从消息记忆到执行流程：如何让多个AI“同事”协同完成复杂任务？

在上一章的消息与记忆系统中，我们已经能让AI记住之前的对话内容。但你是否想过：如果用户要求“预订从北京到上海的高铁，并安排附近的酒店”，这需要协调多个代理（比如票务查询、酒店预订、时间规划），如何让这些“AI同事”有条不紊地合作呢？就像给交通指挥中心配备一套智能调度系统！本章将介绍如何通过执行流程框架管理复杂的多步骤任务。

核心问题：如何让多个AI代理“像交响乐团一样协作”？

想象你是一名指挥家，需要让不同乐器演奏家（代理）配合完成一首乐曲：

小提琴代理：负责查询高铁时刻表
大提琴代理：负责筛选酒店选项
指挥代理：协调两者并整合最终方案

执行流程框架就像这位指挥家，通过以下核心机制实现：
✅ 自动规划任务步骤
✅ 同时执行多个代理任务
✅ 在超时或出错时优雅退出

核心概念分解

1. FlowFactory：流程工厂的“总调度台”

就像汽车工厂根据需求生产不同车型，FlowFactory根据任务类型创建对应的流程：

# 创建规划型流程
flow = FlowFactory.create_flow(flow_type=FlowType.PLANNING, agents=代理们)

2. FlowType：流程的“指挥风格”

支持多种协作模式：

规划型：先制定详细步骤再执行（适合复杂任务）
并发型：同时执行多个步骤（适合并行任务）
串行型：按顺序一步步执行（适合简单任务）

# 选择并发型执行多个代理任务
flow = FlowFactory.create_flow(flow_type=FlowType.PARALLEL)

3. 超时管理：防止流程“卡在进行时”

设置全局超时时间，超过后自动终止并清理资源：

try:await asyncio.wait_for(流程执行(), timeout=3600)  # 最长执行1小时
except asyncio.TimeoutError:print("任务超时！已终止")

如何用执行流程框架解决问题？

场景：用户要求“预订北京到上海的高铁并安排酒店”

步骤1：创建代理团队

from app.agent import TicketAgent, HotelAgentagents = {"ticket": TicketAgent(),  # 负责查询高铁"hotel": HotelAgent()     # 负责酒店预订
}

步骤2：定义流程类型并执行

# 创建并发型流程，同时处理票务和酒店
flow = FlowFactory.create_flow(flow_type=FlowType.PARALLEL,agents=agents
)# 执行任务并等待结果
result = await flow.execute("预订北京到上海的高铁并安排酒店")

步骤3：查看整合结果

print(result)
# 输出示例：
# "高铁已预订：G101次（10:00发车），酒店：上海外滩希尔顿（距车站1公里）"

内部实现揭秘：流程的“指挥过程”

流程分解（用简单序列图演示）

关键代码解析

1. 流程工厂创建逻辑（app/flow/flow_factory.py）

class FlowFactory:@staticmethoddef create_flow(flow_type, agents):if flow_type == FlowType.PARALLEL:return ParallelFlow(agents)  # 并行流程类elif flow_type == FlowType.PLANNING:return PlanningFlow(agents)  # 规划型流程类# 其他类型...

2. 并行执行示例（app/flow/parallel_flow.py）

class ParallelFlow:async def execute(self, prompt):tasks = []for agent in self.agents.values():tasks.append(asyncio.create_task(agent.run(prompt)))  # 同时启动所有代理任务results = await asyncio.gather(*tasks, return_exceptions=True)return self._merge_results(results)  # 整合结果

深入理解执行流程框架

通过本章，你已掌握：
✅ 如何通过FlowFactory创建不同流程类型
✅ 并行与串行执行的差异与适用场景
✅ 超时机制如何避免任务无限挂起

小结与展望

通过执行流程框架，我们实现了：
✅ 多代理任务的有序协调
✅ 根据需求灵活选择执行策略
✅ 自动化资源清理与超时管理

下一章我们将探讨工具与LLM集成，学习如何让AI调用真实世界的工具（比如天气API、浏览器）——就像给指挥家配备更多乐器，让整个“交响乐团”能力更强！现在你可以尝试修改代码，为流程添加更多代理（比如时间规划代理）了！