多Agent编排实战：用CrewAI构建自动写作流水线

一、背景：为什么需要多Agent编排？

单个 LLM 调用有两个根本局限：一是上下文窗口有限（即使是 128K，放不下整个企业知识库），二是单一模型不擅长所有子任务（研究、写作、审校、SEO 优化各自需要不同的"人格"和工具集）。

多 Agent 编排的核心思路，是把一个复杂任务拆成多个子任务，每个子任务由一个"专业 Agent"执行，通过消息传递实现协作。CrewAI 是目前 Python 生态中最成熟的编排框架之一，其核心概念是 Crew（团队） 和 Task（任务）。

二、系统设计：四级写作流水线

我们以"自动生成技术博客"为目标，设计以下 Agent 团队：

三、环境搭建

# 创建虚拟环境
python -m venv crewai-env && source crewai-env/bin/activate

# 安装核心依赖
pip install crewai crewai-tools "openai>=1.0" requests beautifulsoup4

# 验证
python -c "import crewai; print(crewai.__version__)"

四、代码实战：构建四级Agent团队

4.1 定义各 Agent 角色

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool
from openai import OpenAI

# 统一 LLM 配置（对接本地或云端 DeepSeek-V3）
llm_config = {
    "model": "deepseek-v3",
    "api_key": "sk-internal",
    "base_url": "http://toukenai-llm:8000/v1",
    "temperature": 0.3,
}

# 研究员：负责资料收集
researcher = Agent(
    role="高级技术研究员",
    goal="围绕给定主题，搜集权威资料、最新进展和行业案例，输出结构化的研究简报",
    backstory="""你曾在多家科技媒体担任技术记者，擅长从 ArXiv、GitHub Trending、技术博客中提取高价值信息。
你的工作标准是：每条信息必须有来源， speculative（推测性）内容必须标注。""",
    tools=[SerperDevTool(), WebsiteSearchTool()],
    llm_config=llm_config,
    verbose=True,
    max_iter=5,
)

# 作者：根据研究简报写初稿
writer = Agent(
    role="资深技术作者",
    goal="将研究简报转化为 2000-3000 字的技术博客，包含代码示例、实操步骤、效果数据",
    backstory="""你是一位写过多篇 10k+ 阅读技术博客的作者。你遵循"具体 > 抽象"原则，
包含真实命令、真实报错、真实排障过程。读者的 Mn 时间是 15 分钟，你必须在开头 3 句内说清价值。""",
    llm_config=llm_config,
    verbose=True,
)

# 编辑：结构与逻辑优化
editor = Agent(
    role="技术编辑",
    goal="优化文章结构，补强逻辑断层，确保以"问题 → 方案 → 验证"主线推进",
    backstory="""你有 8 年技术出版经验，最讨厌"泛泛而谈"和"结构松散"。你的检查清单：
1. 标题是否准确反映内容？2. 每个 H2 是否有实质贡献？3. 代码块是否有注释？4. 数据是否可验证？""",
    llm_config=llm_config,
    verbose=True,
)

# SEO 审校
seo_reviewer = Agent(
    role="SEO 审校专家",
    goal="优化 meta description、标题关键词密度、可读性评分到 Flesch 60+",
    backstory="""你熟悉 Google / 百度 SEO 算法，知道技术文章怎样在搜索结果中获得高点击率。
你的金科玉律是：标题必须包含核心关键词 + 用户收益，meta description 150 字内讲清"你能获得什么"。""",
    llm_config=llm_config,
    verbose=True,
)

4.2 定义 Task 链

# Task 1：研究
research_task = Task(
    description=(
        "请为以下主题做深度研究：{topic}。"
        "要求：① 找 3-5 篇权威参考（官方文档、知名博客、论文）；"
        "② 提取关键技术参数和实测数据；"
        "③ 列出 2-3 个实际可运行的代码示例；"
        "④ 输出 Markdown 格式研究简报。"
    ),
    expected_output="结构化研究简报，含来源链接、要点列表、代码片段",
    agent=researcher,
)

# Task 2：写作
writing_task = Task(
    description=(
        "基于以下研究简报，撰写一篇面向中级开发者的技术博客：\n\n{research_task.output}"
        "要求：① 标题要直击痛点；② 4 个 H2 模块（背景/方案/效果/总结）；"
        "③ 每个模块含至少一个可运行的代码块；④ 字数 2000-3000 中文；⑤ 无营销话术。"
    ),
    expected_output="完整 Markdown 草稿",
    agent=writer,
    context=[research_task],    # 依赖研究员输出
)

# Task 3：编辑
editing_task = Task(
    description=(
        "请审查并优化以下技术博客：\n\n{writing_task.output}\n\n"
        "只输出修改后的终稿，不要提修改建议。"
    ),
    expected_output="优化后的终稿 Markdown",
    agent=editor,
    context=[writing_task],
)

# Task 4：SEO 审校
seo_task = Task(
    description=(
        "请为以下终稿优化 SEO：\n\n{editing_task.output}\n\n"
        "输出包含：① 建议标题；② Meta description（≤150字符）；"
        "③ 关键词密度分析；④ 最终 Markdown。"
    ),
    expected_output="SEO 优化后的文章 + meta 信息",
    agent=seo_reviewer,
    context=[editing_task],
)

4.3 启动 Crew

crew = Crew(
    agents=[researcher, writer, editor, seo_reviewer],
    tasks=[research_task, writing_task, editing_task, seo_task],
    process=Process.sequential,   # 按顺序执行
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "DeepSeek-V3 本地部署与 vLLM 优化实践"})
print(result.raw)

五、关键代码：状态管理与人机协作

5.1 输出校验与人工介入点

多 Agent 流水线不能完全黑箱。必须在关键节点插入人工确认：

from pydantic import BaseModel, ValidationError

class ArticleDraft(BaseModel):
    title: str
    word_count: int
    code_blocks: int
    has_references: bool

def validate_draft(raw_output: str) -> ArticleDraft:
    """将 LLM 输出解析为 Pydantic 模型，失败则要求重写"""
    try:
        return ArticleDraft.model_validate_json(raw_output)
    except ValidationError as e:
        # 触发重写
        raise ValueError(f"草稿格式不合规，要求重写：{e}")

# 在写作 Task 后插入人工确认
human_task = Task(
    description="请确认以下大纲是否满足要求：{writing_task.output}。如满足，输出 'APPROVED'，否则列出修改意见。",
    expected_output="APPROVED 或修改意见",
    agent=editor,
    context=[writing_task],
    human_input=True,   # CrewAI 会暂停等待控制台输入
)

六、高级主题：动态路由与回退机制

真实生产环境需要处理三类异常：

1. LLM 超时：单个 Agent 调用超过 120 秒，自动切换到备用 LLM（如 DeepSeek-V3 失败则 fallback 到 Qwen2.5-72B）；
2. 输出质量不达标：通过 Pydantic 校验，失败次数 ≥ 2 次则自动升级为人工处理；
3. 成本预算：每个 Task 的 token 消耗达到阈值后，截断后续 Agent 并触发告警。

from crewai.utilities.events import CrewEvent

class CostTrackingHandler(CrewEvent):
    MAX_COST_USD = 2.0

    def on_task_complete(self, task):
        cost = task.cost_usage.total_cost
        if cost > self.MAX_COST_USD:
            self.crew.stop()  # 立即终止流水线
            raise RuntimeError(f"成本超限：当前 ${cost:.2f}，上限 ${self.MAX_COST_USD}")

七、效果与最佳实践

投肯智能内测数据（每天产出 5 篇）：