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

AI 工具开发实战（10）：开发一个 AI Agent 自动化工作流——让多个 Agent 协作完成复杂任务（完结篇）

news 2026/7/13 16:09:09

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()client=OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"),base_url="https://api.deepseek.com/v1",)classAgent:"""工作流中的单个 Agent。"""def__init__(self,name,role,instruction):self.name=name self.role=role self.instruction=instructiondefrun(self,context,task):prompt=f"""## 你的角色{self.role}## 任务{self.instruction.format(task=task,context=context)}## 当前上下文{context}请完成你的任务，输出结果。"""resp=client.chat.completions.create(model="deepseek-chat",messages=[{"role":"user","content":prompt}],temperature=0.5if"写手"inself.nameelse0.3,max_tokens=2048,)returnresp.choices[0].message.contentclassAgentWorkflow:"""Agent 工作流引擎。"""def__init__(self):self.agents=[]self.context=""defadd_agent(self,agent):self.agents.append(agent)defrun(self,task):results=[]fori,agentinenumerate(self.agents):print(f"🤖 [{i+1}/{len(self.agents)}]{agent.name}工作中...")result=agent.run(self.context,task)self.context+=f"\n\n##{agent.name}输出\n{result}"results.append({"agent":agent.name,"result":result})returnresults# 预定义的工作流defcreate_research_workflow():wf=AgentWorkflow()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():wf=AgentWorkflow()wf.add_agent(Agent("安全审查员","安全专家","审查代码安全问题：注入、XSS、密钥泄露。""代码：{task}\n输出有问题的行号和修复建议。"))wf.add_agent(Agent("性能审查员","性能优化专家","审查性能问题：N+1查询、内存泄漏、低效算法。""上下文：{context}\n代码：{task}\n输出具体问题和优化方案。"))wf.add_agent(Agent("代码规范审查员","代码规范专家","审查代码规范：命名、注释、结构。""上下文：{context}\n代码：{task}\n输出不合规的代码和建议。"))returnwf@click.group()defcli():pass@cli.command()@click.argument("task")defresearch(task):"""多 Agent 协作研究。"""wf=create_research_workflow()results=wf.run(task)forrinresults:print(f"\n{'='*60}")print(f"📋{r['agent']}")print(f"{'='*60}")print(r['result'])@cli.command()@click.argument("file",type=click.Path(exists=True))defreview(file):"""多 Agent 代码审查。"""withopen(file,encoding="utf-8")asf:code=f.read()wf=create_code_review_workflow()results=wf.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 research"2026年AI编程工具市场"# 🤖 [1/3] 研究员 工作中...# 🤖 [2/3] 分析师 工作中...# 🤖 [3/3] 写手 工作中...# 输出完整分析报告# 多 Agent 代码审查python agentflow.py review src/api/auth.py# 🤖 [1/3] 安全审查员 工作中...# 🤖 [2/3] 性能审查员 工作中...# 🤖 [3/3] 代码规范审查员 工作中...# 输出三维度审查报告

自定义工作流

# 创建自己的 Agent 协作工作流wf=AgentWorkflow()wf.add_agent(Agent("翻译员","...","..."))wf.add_agent(Agent("润色员","...","..."))wf.add_agent(Agent("校对员","...","..."))wf.run("翻译并发布这篇技术文章")