Agno 的 Workflow工作流是一个用于编排确定性、多步骤 AI 流程的核心组件。你可以将工作流理解为一条智能流水线它将复杂的业务逻辑分解为一系列可预测、可重复、可审计的步骤Workflow与 Agent/Team 的区别理解何时使用 Workflow可以帮你为任务选择最合适的工具。特性Agent (智能体)Team (团队)Workflow (工作流)核心比喻单一专家专注执行特定任务项目组多名专家协作解决开放式问题自动化流水线按预设步骤执行已知流程任务类型单一、领域明确的任务需要协作、分工的复杂任务步骤固定、可预测的批量化任务执行逻辑自主推理决策动态协调分工预定义的流程适用场景初步原型、简单任务研究、规划等需要推理的任务数据处理、报告生成等固定流程核心概念与架构Workflow 采用步骤Step为核心的基础架构主要包含以下几个概念Workflow(工作流)顶层的编排器负责管理整个执行过程。Step(步骤)工作的最小单元封装了一个具体的执行逻辑。每个步骤都定义一个明确的executor执行器它可以是Agent (智能体)由 AI 驱动完成特定任务。Team (团队)多个 Agent 的协作组。Function (自定义函数)用于数据清洗、格式转换等纯 Python 逻辑。控制流组件 (Control Flow)用于编排步骤的执行顺序。Parallel(并行)同时执行多个步骤。Condition(条件)根据特定条件决定是否执行某个步骤。Loop(循环)重复执行一组步骤直到满足条件。Router(路由器)根据当前结果动态路由到不同的后续步骤基础用法第一个Workflowfrom agno.agent import Agent from agno.models.deepseek import DeepSeek from agno.workflow import Step, Workflow from dotenv import load_dotenv from agno.tools.baidusearch import BaiduSearchTools load_dotenv() #定义大模型 deepseek DeepSeek(iddeepseek-chat) #定义agent search_agent Agent( nameResearcher, modeldeepseek, tools[BaiduSearchTools()], description负责研究主题收集相关信息, instructions[ 使用搜索工具查找最新信息, 总结3-5个关键要点, 提供信息来源 ], markdownTrue ) writer_agent Agent( nameWriter, modeldeepseek, description负责撰写文章, instructions[ 基于研究结果撰写500字文章, 使用清晰的结构引言、正文、结论, 使用Markdown格式 ], markdownTrue ) simple_workflow Workflow( nameSimple Article Generator, description简单的文章生成工作流, steps[ Step(agentsearch_agent), Step(agentwriter_agent) ] ) if __name__ __main__: topic agno框架在agent上的运用 result simple_workflow.run(topic) print(result.content)使用steps组织from agno.agent import Agent from agno.models.deepseek import DeepSeek from agno.workflow import Step, Steps, Workflow from dotenv import load_dotenv from agno.tools.baidusearch import BaiduSearchTools load_dotenv() #定义大模型 deepseek DeepSeek(iddeepseek-chat) #定义agent search_agent Agent( nameResearcher, modeldeepseek, tools[BaiduSearchTools()], description负责研究主题收集相关信息, instructions[ 使用搜索工具查找最新信息, 总结3-5个关键要点, 提供信息来源 ], markdownTrue ) writer_agent Agent( nameWriter, modeldeepseek, description负责撰写文章, instructions[ 基于研究结果撰写500字文章, 使用清晰的结构引言、正文、结论, 使用Markdown格式 ], markdownTrue ) step1 Step(agentsearch_agent,namecollect) step2 Step(agentwriter_agent,namewrite) steps Steps(steps[step1, step2],name完整的研究报告流程) simple_workflow Workflow( nameSimple Article Generator, description简单的文章生成工作流, steps[ steps ] ) if __name__ __main__: topic agno框架在agent上的运用 result simple_workflow.run(topic) print(result.content)混合组织Workflow的强大之处在于可以混合不同类型的执行器from agno.agent import Agent from agno.models.deepseek import DeepSeek from agno.team import Team, TeamMode from agno.workflow import Step, StepOutput, Workflow from dotenv import load_dotenv from agno.tools.baidusearch import BaiduSearchTools from agno.tools.yfinance import YFinanceTools load_dotenv() #定义大模型 deepseek DeepSeek(iddeepseek-chat) #定义数据预处理函数 def preprocess_data(data): 预处理函数清洗和标准化数据 Args: step_input: StepInput对象包含上一步的输出 Returns: StepOutput: 预处理后的数据 raw_data data.input cleaned_data raw_data.replace( , ) return StepOutput(contentf已清洗数据{cleaned_data}) # 2. 定义市场研究Agent market_researcher Agent( nameMarket Researcher, modeldeepseek , tools[BaiduSearchTools()], description市场研究专家, instructions[收集市场数据, 分析竞争对手] ) # 3. 定义市场分析Agent # 3. 定义分析Team web_analyst Agent( nameWeb Analyst, tools[BaiduSearchTools()], description网络信息分析师 ) financial_analyst Agent( nameFinancial Analyst, tools[YFinanceTools()], description金融数据分析师 ) analysis_team Team( modeTeamMode.coordinate, members[web_analyst, financial_analyst], modeldeepseek, description多领域分析团队, instructions[综合分析多个维度, 提供全面的见解] ) # 4. 定义报告生成Agent report_generator Agent( nameReport Generator, modeldeepseek, description报告撰写专家, instructions[ 生成结构化的分析报告, 包含执行摘要、详细分析和建议, 使用专业的商业语言 ] ) # 5. 组合成Workflow mixed_workflow Workflow( nameMarket Analysis Pipeline, description混合执行的市场分析流水线, steps[ market_researcher, # Step 1: Agent preprocess_data, # Step 2: Function analysis_team, # Step 3: Team report_generator, # Step 4: Agent ] ) if __name__ __main__: topic 分析电动汽车市场的竞争格局 result mixed_workflow.run(topic) print(result.content)使用Pydantic模型使用Pydantic模型定义Workflow的输入和输from typing import List, Optional from agno.agent import Agent from agno.models.deepseek import DeepSeek from agno.team import Team, TeamMode from agno.workflow import Step, StepOutput, Workflow from dotenv import load_dotenv from agno.tools.baidusearch import BaiduSearchTools from agno.tools.yfinance import YFinanceTools from pydantic import BaseModel, Field from agno.tools.crawl4ai import Crawl4aiTools load_dotenv() #定义大模型 deepseek DeepSeek(iddeepseek-chat) # 1. 定义输入模型 class ResearchRequest(BaseModel): 研究请求输入模型 topic: str Field(..., description研究主题, min_length5) depth: int Field(default1, description研究深度(1-5), ge1, le5) source: Optional[List[str]] Field(defaultNone, description数据源) class ResearchFinding(BaseModel): 单个研究发现 url:str Field(..., description信息源URL) summary: str Field(..., description关键信息摘要) class ResearchReport(BaseModel): 完整的研究报告 topic: str Field(..., description研究主题) key_findings: List[ResearchFinding] Field( ..., description关键发现列表 ) overall_summary: str Field(..., description总体概述) confidence_score: float Field(..., description置信度(0-1), ge0, le1) structured_researcher Agent( namestructured_researcher, modeldeepseek, tools[BaiduSearchTools(), Crawl4aiTools()], description结构化研究专家, output_schemaResearchReport, # 指定输出模型 markdownTrue ) validated_workflow Workflow( namevalidated_workflow, input_schemaResearchRequest, # 指定输入验证模型 steps[Step(agentstructured_researcher)], debug_modeTrue ) request ResearchRequest( topic量子计算的商业应用, depth4, sources[arxiv.org, nature.com] ) result validated_workflow.run(request) print(result.content)条件workflowfrom typing import List from agno.agent import Agent from agno.models.deepseek import DeepSeek from agno.workflow import Router, Step, StepInput, Workflow from dotenv import load_dotenv load_dotenv() #定义大模型 deepseek DeepSeek(iddeepseek-chat) content_classifier Agent( nameContent Classifier, modeldeepseek, description内容分类专家, instructions[ 分析输入内容的类型, 分类为technical技术、business商业、creative创意, 返回分类结果和置信度 ] ) # 2. 定义不同类型的处理Agent technical_processor Agent( nameTechnical Processor, modeldeepseek, description技术内容处理专家, instructions[ 使用专业技术术语, 包含代码示例和技术细节, 添加技术参考链接 ] ) business_processor Agent( nameBusiness Processor, modeldeepseek, description商业内容处理专家, instructions[ 使用商业语言, 强调ROI和商业价值, 包含市场分析 ] ) creative_processor Agent( nameCreative Processor, modeldeepseek, description创意内容处理专家, instructions[ 使用富有创意的表达, 讲述引人入胜的故事, 激发读者情感共鸣 ] ) technical_processor Step( nametechnical, agenttechnical_processor, description技术内容处理 ) business_processor Step( namebusiness, agentbusiness_processor, description商业内容处理 ) creative_processor Step( namecreative, agentcreative_processor, description创意内容处理 ) def content_router(step_input: StepInput) - List[Step]: 根据分类结果路由到不同的处理器 Args: step_input: 分类Agent的输入 Returns: str: 路由目标的步骤名称 content step_input.input.lower() if technical in content or 技术 in content: return [technical_processor] elif business in content or 商业 in content: return [business_processor] else: return [creative_processor] workflow Workflow( nameContent Processing with Routing, description根据内容类型自动选择处理流程, steps[ Step( nameclassify, agentcontent_classifier, description分类内容类型 ), # Step 2: 路由根据分类结果选择处理路径 Router( selectorcontent_router, choices[technical_processor, business_processor, creative_processor] ) ] ) if __name__ __main__: resp workflow.run(使用搜索工具查找最新信息) print(resp.content)并行workflowfrom typing import List from agno.tools.yfinance import YFinanceTools from agno.agent import Agent from agno.models.deepseek import DeepSeek from agno.workflow import Parallel, Router, Step, StepInput, Workflow from dotenv import load_dotenv from agno.tools.baidusearch import BaiduSearchTools load_dotenv() #定义大模型 deepseek DeepSeek(iddeepseek-chat) fundamental_analyst Agent( nameFundamental Analyst, modeldeepseek, tools[YFinanceTools(enable_company_infoTrue, enable_analyst_recommendationsTrue)], description基本面分析专家, instructions[ 分析公司财务状况, 评估估值水平, 提供基本面评级 ] ) technical_analyst Agent( nameTechnical Analyst, modeldeepseek, tools[YFinanceTools(enable_stock_priceTrue, enable_technical_indicatorsTrue)], description技术分析专家, instructions[ 分析价格走势, 识别技术形态, 提供技术指标信号 ] ) sentiment_analyst Agent( nameSentiment Analyst, modeldeepseek, tools[BaiduSearchTools()], description情绪分析专家, instructions[ 收集最新新闻, 分析市场情绪, 评估舆论倾向 ] ) risk_analyst Agent( nameRisk Analyst, modeldeepseek, description风险评估专家, instructions[ 识别潜在风险因素, 评估风险等级, 提供风险控制建议 ] ) # 2. 定义汇总Agent synthesis_agent Agent( nameSynthesis Agent, modeldeepseek, description综合分析专家, instructions[ 整合所有分析结果, 提供综合投资建议, 给出明确的操作建议买入/持有/卖出, 标注关键风险点 ] ) parallel_workflow Workflow( nameParallel Stock Analysis, description并行多维度股票分析, steps[ Parallel( Step(namefundamental, agentfundamental_analyst), Step(nametechnical, agenttechnical_analyst), Step(namesentiment, agentsentiment_analyst), Step(namerisk, agentrisk_analyst), nameparallel_analysis ), # Step 2: 汇总并行结果 Step( namesynthesis, agentsynthesis_agent, description综合所有分析提供最终建议 ) ] ) resp parallel_workflow.run(分析湖南黄金的股票) print(resp.content)循环work flowfrom typing import List from agno.tools.yfinance import YFinanceTools from agno.agent import Agent from agno.models.deepseek import DeepSeek from agno.workflow import Loop, Parallel, Router, Step, StepInput, StepOutput, Workflow from dotenv import load_dotenv from agno.tools.baidusearch import BaiduSearchTools from pydantic import BaseModel, Field load_dotenv() #定义大模型 deepseek DeepSeek(iddeepseek-chat) # 1. 定义写作Agent writer_agent Agent( nameWriter, modeldeepseek, description内容写作专家, instructions[ 根据给定主题撰写文章, 使用清晰的结构, 确保内容准确性 ] ) # 2. 定义评审Agent带结构化输出 class ReviewResult(BaseModel): 评审结果模型 quality_score: float Field(..., description质量评分(0-100), ge0, le100) needs_improvement: bool Field(..., description是否需要改进) feedback: str Field(..., description具体反馈意见) suggestions: List[str] Field(..., description改进建议列表) reviewer_agent Agent( nameReviewer, modeldeepseek, description内容评审专家, output_schemaReviewResult, instructions[ 评估文章质量准确性、可读性、结构、深度, 如果质量分数低于80分标记needs_improvementTrue, 提供具体的改进建议 ] ) # 3. 定义改进Agent improver_agent Agent( nameImprover, modeldeepseek, description内容改进专家, instructions[ 根据评审反馈改进文章, 保持原文的核心思想, 针对性解决指出的问题 ] ) # 4. 定义循环条件 def should_continue_iteration(step_output: List[StepOutput]) - bool: 判断是否继续迭代 Args: step_output: 评审步骤的输出 Returns: bool: True表示继续迭代False表示停止 try: for output in step_output: if output.step_name review: if isinstance(output.content, ReviewResult): result: ReviewResult output.content return not (result.needs_improvement and result.quality_score 80) else: # 如果文本中包含需要改进则继续 return True except: pass return True iterative_workflow Workflow( nameIterative Writing Workflow, description迭代式写作改进工作流, steps[ Step( nameinitial_writing, agentwriter_agent, description撰写初稿 ), Loop( namequality_improvement_loop, steps[ Step( namereview, agentreviewer_agent, description评审文章质量 ), Step( nameimprove, agentimprover_agent, description根据反馈改进文章 ) ], end_conditionshould_continue_iteration, max_iterations3 ) ], debug_modeTrue, ) if __name__ __main__: resp iterative_workflow.run(深度学习在自然语言处理中的应用) print(resp.content)注意end_condition是在每次完整的迭代循环体执行完毕之后被调用而不是在每个单独的步骤Step之后。