AI 工具开发实战10开发一个 AI Agent 自动化工作流——让多个 Agent 协作完成复杂任务系列最后一篇做一个最有野心的工具——AI Agent 工作流引擎。不是写一个 Agent而是做一个框架让你可以定义多个 Agent 角色让它们协作完成复杂任务。框架做什么定义一个任务调研 AI 编程工具市场输出一份分析报告 ↓ 1. 研究员 Agent搜索信息收集数据 2. 分析师 Agent分析数据提炼观点 3. 写手 Agent整理成报告 ↓ 最终输出一份结构化的分析报告项目结构agentflow/ ├── agentflow.py # CLI 工作流引擎 ├── agents.py # Agent 定义 └── requirements.txt核心实现# agentflow.pyfromopenaiimportOpenAIimportos,json,clickfromdotenvimportload_dotenv load_dotenv()clientOpenAI(api_keyos.getenv(DEEPSEEK_API_KEY),base_urlhttps://api.deepseek.com/v1,)classAgent:工作流中的单个 Agent。def__init__(self,name,role,instruction):self.namename self.rolerole self.instructioninstructiondefrun(self,context,task):promptf## 你的角色{self.role}## 任务{self.instruction.format(tasktask,contextcontext)}## 当前上下文{context}请完成你的任务输出结果。respclient.chat.completions.create(modeldeepseek-chat,messages[{role:user,content:prompt}],temperature0.5if写手inself.nameelse0.3,max_tokens2048,)returnresp.choices[0].message.contentclassAgentWorkflow:Agent 工作流引擎。def__init__(self):self.agents[]self.contextdefadd_agent(self,agent):self.agents.append(agent)defrun(self,task):results[]fori,agentinenumerate(self.agents):print(f [{i1}/{len(self.agents)}]{agent.name}工作中...)resultagent.run(self.context,task)self.contextf\n\n##{agent.name}输出\n{result}results.append({agent:agent.name,result:result})returnresults# 预定义的工作流defcreate_research_workflow():wfAgentWorkflow()wf.add_agent(Agent(研究员,信息收集专家,搜索并收集关于「{task}」的关键信息、数据、趋势。输出格式按要点列出 5-10 个关键发现。))wf.add_agent(Agent(分析师,数据分析专家,基于研究员的发现分析「{task}」的核心趋势和洞察。上下文{context}输出格式3-5 个核心洞察每个包含趋势分析和影响评估。))wf.add_agent(Agent(写手,技术写作专家,基于研究员和分析师的工作撰写一份关于「{task}」的综合报告。上下文{context}输出格式Markdown 结构包含概述、关键发现、趋势分析、建议。))returnwfdefcreate_code_review_workflow():wfAgentWorkflow()wf.add_agent(Agent(安全审查员,安全专家,审查代码安全问题注入、XSS、密钥泄露。代码{task}\n输出有问题的行号和修复建议。))wf.add_agent(Agent(性能审查员,性能优化专家,审查性能问题N1查询、内存泄漏、低效算法。上下文{context}\n代码{task}\n输出具体问题和优化方案。))wf.add_agent(Agent(代码规范审查员,代码规范专家,审查代码规范命名、注释、结构。上下文{context}\n代码{task}\n输出不合规的代码和建议。))returnwfclick.group()defcli():passcli.command()click.argument(task)defresearch(task):多 Agent 协作研究。wfcreate_research_workflow()resultswf.run(task)forrinresults:print(f\n{*60})print(f{r[agent]})print(f{*60})print(r[result])cli.command()click.argument(file,typeclick.Path(existsTrue))defreview(file):多 Agent 代码审查。withopen(file,encodingutf-8)asf:codef.read()wfcreate_code_review_workflow()resultswf.run(code)forrinresults:print(f\n{*60})print(f{r[agent]})print(f{*60})print(r[result])if__name____main__:cli()使用方式# 多 Agent 研究模式python agentflow.py research2026年AI编程工具市场# [1/3] 研究员 工作中...# [2/3] 分析师 工作中...# [3/3] 写手 工作中...# 输出完整分析报告# 多 Agent 代码审查python agentflow.py review src/api/auth.py# [1/3] 安全审查员 工作中...# [2/3] 性能审查员 工作中...# [3/3] 代码规范审查员 工作中...# 输出三维度审查报告自定义工作流# 创建自己的 Agent 协作工作流wfAgentWorkflow()wf.add_agent(Agent(翻译员,...,...))wf.add_agent(Agent(润色员,...,...))wf.add_agent(Agent(校对员,...,...))wf.run(翻译并发布这篇技术文章)总结Agent 工作流引擎的核心就两层Agent 定义角色 指令模板工作流编排顺序执行上下文传递10 篇文章10 个工具从翻译到代码审查再到多 Agent 协作——这个系列的目标就是让每个开发者都能用 AI 做出实用工具。本文是《AI 开发者工具链实战》系列的第 10 篇完结篇全系列 10 篇已完结