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

定义一个任务："调研 AI 编程工具市场，输出一份分析报告"
  ↓
1. 研究员 Agent：搜索信息，收集数据
2. 分析师 Agent：分析数据，提炼观点
3. 写手 Agent：整理成报告
  ↓
最终输出：一份结构化的分析报告

agentflow/
├── agentflow.py          # CLI + 工作流引擎
├── agents.py             # Agent 定义
└── requirements.txt

# agentflow.py
from openai import OpenAI
import os, json, click
from dotenv import load_dotenv
load_dotenv()

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)

class Agent:
    """工作流中的单个 Agent。"""
    def __init__(self, name, role, instruction):
        self.name = name
        self.role = role
        self.instruction = instruction

    def run(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.5 if "写手" in self.name else 0.3,
            max_tokens=2048,
        )
        return resp.choices[0].message.content

class AgentWorkflow:
    """Agent 工作流引擎。"""
    def __init__(self):
        self.agents = []
        self.context = ""

    def add_agent(self, agent):
        self.agents.append(agent)

    def run(self, task):
        results = []
        for i, agent in enumerate(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})
        return results

# 预定义的工作流
def create_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 结构，包含：概述、关键发现、趋势分析、建议。"))
    return wf

def create_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输出不合规的代码和建议。"))
    return wf

@click.group()
def cli(): pass

@cli.command()
@click.argument("task")
def research(task):
    """多 Agent 协作研究。"""
    wf = create_research_workflow()
    results = wf.run(task)
    for r in results:
        print(f"\n{'='*60}")
        print(f"📋 {r['agent']}")
        print(f"{'='*60}")
        print(r['result'])

@cli.command()
@click.argument("file", type=click.Path(exists=True))
def review(file):
    """多 Agent 代码审查。"""
    with open(file, encoding="utf-8") as f:
        code = f.read()
    wf = create_code_review_workflow()
    results = wf.run(code)
    for r in results:
        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("翻译并发布这篇技术文章")