智能体工作流（Agentic Workflow）应用案例的项目集合。通过多个精心设计的案例，演示如何使用 AI 智能体协作完成复杂任务，涵盖内容创作、信息检索、数据处理等多个领域。
Agentic Workflow（智能体工作流）是一种新兴的 AI 应用模式，通过多个专门化的 AI 智能体协作，完成单个 AI 难以胜任的复杂任务。本项目收集并实现了多个典型的 Agentic Workflow 案例，每个案例都展示了不同的协作模式和应用场景。
- 🔄 多智能体协作：展示不同智能体间的协作模式
- 🎨 多样化场景：涵盖内容创作、信息检索、数据分析等领域
- 🛠️ 模块化设计：每个案例都是独立的模块，易于理解和扩展
- 📚 详细文档：每个案例都有完整的说明文档和使用指南
- 🔧 易于扩展：提供统一的基础设施，便于添加新案例
基于 Actor-Critic 模式的智能反思工作流系统，通过参与者和评论者的多轮协作，实现内容的持续改进和优化。
核心特性：
- Actor-Critic 协作模式
- 多轮迭代优化
- 完整状态追溯
- 智能反馈机制
应用场景：
- 学术论文写作和改进
- 技术文档的迭代优化
- 创意内容的反思式创作
详细文档： reflection/README.md
graph LR A[用户输入主题] --> B[Actor生成初稿] B --> C[Critic评审反馈] C --> D[Actor修订内容] D --> E{达到循环次数?} E -->|否| C E -->|是| F[输出最终结果] style A fill:#e1f5fe style F fill:#c8e6c9 style B fill:#fff3e0 style C fill:#fce4ec style D fill:#fff3e0 工作流程：
- 初始创作：Actor 根据主题生成初始草稿
- 专业评审：Critic 提供同行评议级别的反馈
- 迭代改进：Actor 根据反馈系统性修订内容
- 循环优化：重复评审-修订过程直到达到预设轮次
- 状态追溯：完整记录每轮的草稿和评审历史
基于多智能体协作的网络信息检索和摘要生成系统，通过搜索、抓取、摘要三个智能体的协作，实现从查询到摘要的完整信息处理流程。
核心特性：
- 三阶段流水线处理
- 智能查询优化
- 并发网页抓取
- AI 驱动摘要生成
应用场景：
- 新闻事件快速调研
- 学术研究背景收集
- 市场分析和竞品调研
- 信息整理和摘要生成
详细文档： web_access/README.md
graph LR A[用户查询] --> B[搜索智能体] B --> C[抓取智能体] C --> D[摘要智能体] D --> E[结构化摘要] B --> F[SERP API] F --> G[搜索结果] G --> C C --> H[并发抓取] H --> I[内容清理] I --> D D --> J[AI摘要] J --> K[引用管理] K --> E style A fill:#e1f5fe style E fill:#c8e6c9 style B fill:#fff3e0 style C fill:#f3e5f5 style D fill:#e8f5e8 工作流程：
- 智能搜索：AI 优化查询词，调用搜索 API 获取结果
- 并发抓取：多线程抓取网页内容，智能去噪处理
- 摘要生成：AI 分析内容生成结构化摘要，自动添加引用
基于语义路由和意图识别的智能旅行规划系统，通过协调器智能体和专业子智能体的协作，实现自然语言查询的智能路由和统一服务整合。
核心特性：
- 智能意图识别和语义路由
- 多领域专业智能体协作
- 统一的旅行服务整合
- 自然语言查询处理
应用场景：
- 智能旅行助手和规划
- 多服务统一入口平台
- 客服机器人智能路由
- 意图识别和分类系统
详细文档： semantic_router/README.md
graph LR A[用户查询] --> B[协调器智能体] B --> C[意图识别] C --> D{路由选择} D -->|航班搜索| E[航班搜索智能体] D -->|酒店搜索| F[酒店搜索智能体] D -->|租车搜索| G[租车搜索智能体] D -->|未知意图| H[错误处理] E --> I[WebAccess调用] F --> J[WebAccess调用] G --> K[WebAccess调用] I --> L[航班信息] J --> M[酒店信息] K --> N[租车信息] L --> O[结果整合] M --> O N --> O O --> P[综合响应] style A fill:#e1f5fe style P fill:#c8e6c9 style B fill:#fff3e0 style C fill:#f3e5f5 style E fill:#e8f5e8 style F fill:#e8f5e8 style G fill:#e8f5e8 工作流程：
- 查询接收：用户提交自然语言旅行查询
- 意图识别：协调器分析查询内容，识别用户意图（航班/酒店/租车）
- 智能路由：根据识别的意图路由到相应的专业子智能体
- 信息获取：子智能体优化查询并调用 WebAccess 获取相关信息
- 结果整合：协调器整合子智能体结果，生成统一的综合响应
基于语义路由和意图识别的智能旅行规划系统，通过协调器智能体和专业子智能体的协作，实现自然语言查询的智能路由和统一服务整合。
核心特性：
- 智能意图识别和语义路由
- 多领域专业智能体协作
- 统一的旅行服务整合
- 自然语言查询处理
应用场景：
- 智能旅行助手和规划
- 多服务统一入口平台
- 客服机器人智能路由
- 意图识别和分类系统
详细文档： parallel_delegation/README.md
graph LR A[用户查询] --> B[协调器智能体] B --> C[意图识别] C --> D{路由选择} D -->|航班搜索| E[航班搜索智能体] D -->|酒店搜索| F[酒店搜索智能体] D -->|租车搜索| G[租车搜索智能体] D -->|未知意图| H[错误处理] E --> I[WebAccess调用] F --> J[WebAccess调用] G --> K[WebAccess调用] I --> L[航班信息] J --> M[酒店信息] K --> N[租车信息] L --> O[结果整合] M --> O N --> O O --> P[综合响应] style A fill:#e1f5fe style P fill:#c8e6c9 style B fill:#fff3e0 style C fill:#f3e5f5 style E fill:#e8f5e8 style F fill:#e8f5e8 style G fill:#e8f5e8 工作流程：
- 查询接收：用户提交自然语言旅行查询
- 意图识别：协调器分析查询内容，识别用户意图（航班/酒店/租车）
- 智能路由：根据识别的意图路由到相应的专业子智能体
- 信息获取：子智能体优化查询并调用 WebAccess 获取相关信息
- 结果整合：协调器整合子智能体结果，生成统一的综合响应
基于协调器-委托智能体模式的动态分片信息处理系统，通过智能分片和异步并发处理，实现大量实体数据的批量信息获取和整合。
核心特性：
- 动态分片处理机制
- 协调器-委托智能体协作
- 异步并发优化
- 智能信息获取集成
应用场景：
- 批量人物信息收集
- 企业信息调研
- 学术研究支持
- 知识图谱构建
详细文档： dynamic_sharding/README.md
graph LR A[实体列表] --> B[协调器] B --> C[动态分片] C --> D[委托智能体1] C --> E[委托智能体2] C --> F[委托智能体N] D --> G[WebAccess] E --> H[WebAccess] F --> I[WebAccess] G --> J[实体信息1] H --> K[实体信息2] I --> L[实体信息N] J --> M[结果整合] K --> M L --> M M --> N[最终报告] style A fill:#e1f5fe style N fill:#c8e6c9 style B fill:#fff3e0 style C fill:#f3e5f5 style D fill:#e8f5e8 style E fill:#e8f5e8 style F fill:#e8f5e8 工作流程：
- 动态分片：协调器根据配置将实体列表分割为多个分片
- 并发处理：为每个分片创建委托智能体，异步并发处理
- 信息获取：每个委托智能体调用 WebAccess 获取实体详细信息
- 结果整合：收集所有分片结果，生成完整的信息报告
import asyncio from dynamic_sharding.main import run def run_dynamic_sharding_example(): """运行动态分片工作流示例""" # 配置参数 input_file = "dynamic_sharding/data/entities.txt" output_file = "dynamic_sharding/data/entity_info.txt" shard_size = 3 # 每个分片包含3个实体 print(f"开始执行动态分片工作流...") print(f"输入文件：{input_file}") print(f"输出文件：{output_file}") print(f"分片大小：{shard_size}") try: # 执行异步工作流 asyncio.run(run(input_file, output_file, shard_size)) print("✅ 动态分片工作流执行完成！") print(f"结果已保存到：{output_file}") # 读取并显示结果摘要 with open(output_file, 'r', encoding='utf-8') as f: content = f.read() print(f"生成内容长度：{len(content)} 字符") except Exception as e: print(f"❌ 工作流执行失败：{str(e)}") if __name__ == "__main__": run_dynamic_sharding_example()基于任务分解模式的智能文档处理系统，通过协调器和子任务代理的协作，将复杂的文档分析任务分解为多个专门的子任务，并行执行以提高处理效率和准确性。
核心特性：
- 任务分解机制
- 并行处理优化
- 专业分工子任务
- 结果智能整合
应用场景：
- 学术论文和研究报告分析
- 新闻文章和媒体内容处理
- 法律文档和合同分析
- 技术文档和规范解析
详细文档： task_decomposition/README.md
graph TD A[文档输入] --> B[CoordinatorAgent<br/>协调器] B --> C[任务分解] C --> D[SubTaskAgent_1<br/>命名实体提取] C --> E[SubTaskAgent_2<br/>引用信息提取] C --> F[SubTaskAgent_3<br/>数值数据提取] C --> G[SubTaskAgent_4<br/>关键术语提取] C --> H[SubTaskAgent_5<br/>外部来源提取] D --> I[并行处理] E --> I F --> I G --> I H --> I I --> J[结果整合] J --> K[最终报告输出] style A fill:#e1f5fe style K fill:#c8e6c9 style B fill:#fff3e0 style C fill:#f3e5f5 style D fill:#e8f5e8 style E fill:#e8f5e8 style F fill:#e8f5e8 style G fill:#e8f5e8 style H fill:#e8f5e8 工作流程：
- 任务分解：协调器将文档分析任务分解为 5 个专门的子任务
- 并行执行：多个子任务代理同时处理各自领域的信息提取
- 专业处理：每个子任务专注于特定类型的信息提取（实体、引用、数值、术语、来源）
- 结果整合：将所有子任务结果合并为完整的文档分析报告
import asyncio from task_decomposition.main import TaskDecomposition def run_task_decomposition_example(): """运行任务分解工作流示例""" async def main(): # 创建任务分解实例 task_decomp = TaskDecomposition() print("开始执行任务分解工作流...") print(f"输入文件：{task_decomp.input_file}") print(f"输出文件：{task_decomp.output_file}") try: # 执行任务分解流程 await task_decomp.run() print("✅ 任务分解工作流执行完成！") print(f"结果已保存到：{task_decomp.output_file}") except Exception as e: print(f"❌ 工作流执行失败：{str(e)}") asyncio.run(main()) if __name__ == "__main__": run_task_decomposition_example()基于动态任务分解模式的智能文档处理系统，通过协调器和子任务代理的协作，使用大型语言模型动态生成专门的子任务，将复杂的文档分析任务智能分解为 10 个独立的并行子任务，以提高处理效率和分析深度。
核心特性：
- 动态任务分解机制
- AI 智能生成子任务
- 自适应分析策略
- 深度并行处理优化
应用场景：
- 文学作品和小说分析
- 学术论文和研究报告处理
- 新闻文章和媒体内容分析
- 历史文档和档案研究
详细文档： dynamic_decomposition/README.md
基于有向无环图（DAG）的智能文档处理系统，通过协调器和多个专门的智能体协作，按照预定义的依赖关系和执行顺序，自动化处理复杂的文档分析和处理任务。
核心特性：
- DAG编排任务调度
- 智能体协作处理
- 异步并行执行
- 动态配置加载
- 结果自动传递
应用场景：
- 复杂文档处理流水线
- 多阶段数据分析任务
- 内容生成和编译系统
- 自动化报告生成
- 多源信息整合和分析
详细文档： dag_orchestration/README.md
graph TD A[YAML配置文件] --> B[CoordinatorAgent<br/>协调器] B --> C[任务依赖解析] C --> D[CollectAgent<br/>文档收集] D --> E[PreprocessAgent<br/>预处理] E --> F[ExtractAgent<br/>信息提取] F --> G[CompileAgent<br/>内容编译] G --> H[SummarizeAgent<br/>总结生成] D --> I[ChatModel<br/>AI模型接口] E --> I F --> I G --> I H --> I I --> J[任务结果存储] J --> K[最终报告输出] style A fill:#e1f5fe style K fill:#c8e6c9 style B fill:#fff3e0 style C fill:#f3e5f5 style D fill:#e8f5e8 style E fill:#e8f5e8 style F fill:#e8f5e8 style G fill:#e8f5e8 style H fill:#e8f5e8 工作流程：
- 配置加载：从YAML文件加载DAG定义，解析任务依赖关系
- 任务调度：协调器按依赖顺序执行任务，支持无依赖任务的并行执行
- 智能体协作：5个专门智能体按序处理文档收集、预处理、信息提取、内容编译和总结生成
- 结果传递：上游任务的输出自动作为下游任务的输入，形成完整的处理链
- 报告生成：生成最终的综合分析报告
import asyncio from dag_orchestration.main import DagOrchestrationAgent def run_dag_orchestration_example(): """运行DAG编排工作流示例""" async def main(): # 创建DAG编排实例 dag_agent = DagOrchestrationAgent() print("开始执行DAG编排工作流...") print(f"DAG配置文件：{dag_agent.dag_file_path}") print(f"输出报告文件：{dag_agent.report_file_path}") try: # 执行DAG编排流程 await dag_agent.run() print("✅ DAG编排工作流执行完成！") print(f"最终报告已保存到：{dag_agent.report_file_path}") # 读取并显示结果摘要 with open(dag_agent.report_file_path, 'r', encoding='utf-8') as f: content = f.read() print(f"生成报告长度：{len(content)} 字符") except Exception as e: print(f"❌ 工作流执行失败：{str(e)}") asyncio.run(main()) if __name__ == "__main__": run_dag_orchestration_example()graph TD A[文档输入] --> B[CoordinatorAgent<br/>协调器] B --> C[AI模型动态任务分解] C --> D[SubTaskAgent_1<br/>动态任务1] C --> E[SubTaskAgent_2<br/>动态任务2] C --> F[SubTaskAgent_3<br/>动态任务3] C --> G[SubTaskAgent_4<br/>动态任务4] C --> H[SubTaskAgent_5<br/>动态任务5] C --> I[SubTaskAgent_6<br/>动态任务6] C --> J[SubTaskAgent_7<br/>动态任务7] C --> K[SubTaskAgent_8<br/>动态任务8] C --> L[SubTaskAgent_9<br/>动态任务9] C --> M[SubTaskAgent_10<br/>动态任务10] D --> N[并行处理] E --> N F --> N G --> N H --> N I --> N J --> N K --> N L --> N M --> N N --> O[结果整合] O --> P[最终报告输出] style A fill:#e1f5fe style P fill:#c8e6c9 style B fill:#fff3e0 style C fill:#f3e5f5 style D fill:#e8f5e8 style E fill:#e8f5e8 style F fill:#e8f5e8 style G fill:#e8f5e8 style H fill:#e8f5e8 style I fill:#e8f5e8 style J fill:#e8f5e8 style K fill:#e8f5e8 style L fill:#e8f5e8 style M fill:#e8f5e8 工作流程：
- 动态任务分解：协调器使用 AI 模型根据文档内容智能生成 10 个最适合的子任务
- 并行执行：10 个动态生成的子任务代理同时处理各自的专门分析任务
- 自适应处理：每个子任务根据文档特点自动调整分析重点和策略
- 结果整合：将所有动态子任务结果合并为完整的文档分析报告
import asyncio from dynamic_decomposition.main import DynamicDecomposition def run_dynamic_decomposition_example(): """运行动态分解工作流示例""" async def main(): # 创建动态分解实例 dynamic_decomp = DynamicDecomposition() print("开始执行动态分解工作流...") print(f"输入文件：{dynamic_decomp.input_file}") print(f"输出文件：{dynamic_decomp.output_file}") try: # 执行动态分解流程 await dynamic_decomp.run() print("✅ 动态分解工作流执行完成！") print(f"结果已保存到：{dynamic_decomp.output_file}") except Exception as e: print(f"❌ 工作流执行失败：{str(e)}") asyncio.run(main()) if __name__ == "__main__": run_dynamic_decomposition_example()git clone https://github.com/your-username/agentic_workflow_case.git cd agentic_workflow_casepip install -r requirements.txt创建 .env 文件并配置必要的 API 密钥：
# AI模型API配置 AGENTLY_API_KEY=your_agently_api_key DOUBAO_API_KEY=your_doubao_api_key # 搜索API配置（用于web_access案例） SERPAPI_API_KEY=your_serpapi_key- Python 3.8+
- 8GB+ RAM（推荐）
- 稳定的网络连接
agently>=0.1.0 requests>=2.28.0 beautifulsoup4>=4.11.0 python-dotenv>=0.19.0 loguru>=0.6.0| 服务 | 用途 | 必需性 | 获取方式 |
|---|---|---|---|
| 豆包 API | AI 模型调用 | 必需 | 豆包开放平台 |
| SERP API | 网络搜索 | web_access 案例必需 | SerpApi |
agentic_workflow_case/ ├── README.md # 项目主文档 ├── requirements.txt # 依赖包列表 ├── .env.example # 环境变量模板 ├── utils/ # 公共工具模块 │ ├── ChatModel.py # AI模型接口封装 │ ├── logger.py # 日志配置 │ ├── save_to_disk.py # 文件保存工具 │ └── manage.py # 项目管理工具 ├── reflection/ # 反思工作流案例 │ ├── README.md # 案例详细文档 │ ├── main.py # 主程序入口 │ ├── actor.py # Actor智能体实现 │ ├── critic.py # Critic智能体实现 │ ├── prompts.py # 提示词定义 │ └── data/ # 数据存储目录 ├── web_access/ # 网络搜索工作流案例 │ ├── README.md # 案例详细文档 │ ├── main.py # 主程序入口 │ ├── search.py # 搜索智能体 │ ├── scrape.py # 抓取智能体 │ ├── summarize.py # 摘要智能体 │ ├── serp.py # 搜索API客户端 │ ├── prompts.py # 提示词定义 │ └── data/ # 数据存储目录 ├── parallel_delegation/ # 并行委托工作流案例 │ ├── README.md # 案例详细文档 │ ├── main.py # 主程序入口 │ ├── coordinator.py # 旅行规划协调器 │ ├── flight_search.py # 航班搜索代理 │ ├── hotel_search.py # 酒店搜索代理 │ ├── car_rental_search.py # 租车搜索代理 │ └── prompts.py # 提示词定义 ├── semantic_router/ # 语义路由工作流案例 │ ├── main.py # 主程序入口 │ ├── coordinator.py # 协调器智能体 │ ├── hotel_search.py # 酒店搜索智能体 │ ├── flight_search.py # 航班搜索智能体 │ ├── car_rental_search.py # 租车搜索智能体 │ └── prompts.py # 提示词定义 ├── dynamic_sharding/ # 动态分片工作流案例 │ ├── README.md # 案例详细文档 │ ├── main.py # 主程序入口 │ ├── coordinator.py # 协调器智能体 │ ├── delegate.py # 委托智能体 │ ├── message.py # 消息传递结构 │ └── data/ # 数据存储目录 ├── task_decomposition/ # 任务分解工作流案例 │ ├── README.md # 案例详细文档 │ ├── main.py # 主程序入口 │ ├── coordinator.py # 协调器代理 │ └── delegates.py # 子任务代理 ├── dynamic_decomposition/ # 动态分解工作流案例 │ ├── README.md # 案例详细文档 │ ├── main.py # 主程序入口 │ ├── coordinator.py # 协调器代理 │ └── delegates.py # 子任务代理 ├── dag_orchestration/ # DAG编排工作流案例 │ ├── README.md # 案例详细文档 │ ├── main.py # 主程序入口 │ ├── coordinator.py # DAG协调器实现 │ └── agents/ # 智能体模块目录 │ ├── collect.py # 文档收集智能体 │ ├── preprocess.py # 预处理智能体 │ ├── extract.py # 信息提取智能体 │ ├── compile.py # 编译智能体 │ └── summarize.py # 总结智能体 ├── test/ # 测试文件目录 └── logs/ # 日志文件目录 本项目采用 MIT 许可证。详见 LICENSE 文件。
- 项目维护者: justin.郑
- 邮箱: 3907721@qq.com
- Issues: GitHub Issues
⭐ 如果这个项目对你有帮助，请给我们一个星标！
🔄 持续更新中，更多精彩的 Agentic Workflow 案例即将到来...