我怎么搭建 AI 内容 Agent — 完整工作流

开场

你现在读的这篇文章，从选题到初稿到审核到排版，全部由一套 AI Agent Pipeline 生产。这个系统处理了 300 多个选题，覆盖 6 个内容系列，中文优先，支持文章和推文两种格式。这篇文章把整套系统的架构、每个 Agent 的职责、质量控制机制和实际运行数据完全公开。

问题背景

一个人做内容创作，瓶颈不是写作本身，而是写作的完整流程：

1. 调研：一篇技术文章需要 1-2 小时的调研——看文档、对比价格、找数据
2. 写作：2-3 小时的初稿，还要保持语气一致性
3. 审核：自己审自己的文章，很难发现问题
4. 排版：YAML frontmatter、文件命名、目录结构——枯燥但必须准确

如果要做 300 篇内容，按人工效率（一天一篇），需要整整一年。用 Agent Pipeline，我用了两周完成了 80% 的初稿，剩下的时间做人工审核和迭代。

核心架构

Pipeline 总览

Topic Source (300+ topics)
        ↓
┌──────────────────────────────────────────┐
│              Content Pipeline             │
│                                          │
│  ┌──────────┐    ┌──────────┐            │
│  │ Research  │───→│ Chinese  │            │
│  │  Agent    │    │  Draft   │            │
│  │ (调研)    │    │  Agent   │            │
│  └──────────┘    └────┬─────┘            │
│                       │                  │
│                  ┌────▼─────┐            │
│                  │ Quality  │            │
│                  │ Reviewer │            │
│                  │ (审核)   │            │
│                  └────┬─────┘            │
│                       │                  │
│              ┌────────┴────────┐         │
│              │                 │         │
│         score >= 75      score < 75      │
│              │                 │         │
│        ┌─────▼─────┐   ┌──────▼──────┐  │
│        │  Format    │   │  Revision   │  │
│        │  Agent     │   │  Agent      │  │
│        │  (排版)    │   │  (修改)     │  │
│        └─────┬─────┘   └──────┬──────┘  │
│              │                │          │
│              │           (回到审核,      │
│              │            最多2轮)       │
│              ↓                          │
│       Final Output                      │
└──────────────────────────────────────────┘

设计原则

1. 单一职责：每个 Agent 只做一件事——调研的不写作，写作的不排版
2. 质量门控：审核 Agent 用评分机制卡质量，低于 75 分自动触发修改
3. 中文优先：所有内容先写中文，不是从英文翻译过来的
4. 批量处理：推文 10 篇一批，文章 5 篇一批，并行执行

实现细节

Agent 1: Research Agent（调研员）

调研 Agent 的核心任务：根据选题，用 WebSearch 收集事实、价格、数据，输出结构化的调研报告。

import anthropic
import json

class ResearchAgent:
    """调研 Agent：收集事实数据"""

    def __init__(self):
        self.client = anthropic.Anthropic()

    async def research(self, topic: dict) -> dict:
        """对一个选题做调研"""
        system_prompt = """你是一个专业的技术调研员。

任务：根据给定选题，收集最新的事实数据。

输出格式（JSON）：
{
  "key_facts": [
    {"fact": "描述", "source": "来源", "date": "数据日期"}
  ],
  "pricing_data": {
    "产品名": {"价格": "xxx", "来源": "url"}
  },
  "statistics": [
    {"metric": "指标", "value": "数值", "context": "背景"}
  ],
  "competitor_info": [...],
  "technical_details": [...]
}

规则：
1. 只输出可验证的事实，不要推测
2. 标注每条数据的来源
3. 价格数据必须标注币种和时间
4. 如果某类信息找不到，该字段留空数组"""

        response = self.client.messages.create(
            model="claude-sonnet-4-5-20250514",
            max_tokens=4096,
            system=system_prompt,
            messages=[{
                "role": "user",
                "content": f"""选题信息：
标题: {topic['title']}
描述: {topic['description']}
系列: {topic['series']}
关键词: {', '.join(topic['keywords'])}"""
            }]
        )

        return json.loads(response.content[0].text)

Agent 2: Chinese Draft Agent（中文写手）

这是最关键的 Agent。它接收调研数据和写作规则，输出中文文章。

class ChineseDraftAgent:
    """中文写作 Agent"""

    def __init__(self, series_rules: str, brand_voice: str, anti_ai_patterns: list[str]):
        self.client = anthropic.Anthropic()
        self.series_rules = series_rules
        self.brand_voice = brand_voice
        self.banned_phrases = anti_ai_patterns

    async def write(self, topic: dict, research_data: dict) -> str:
        """基于调研数据写文章"""
        system_prompt = f"""你是 Jessie Qin 的 AI 写作助手。

## 作者背景
- CS PhD + NYU Stern Master
- Senior Member of Technical Staff, Generative AI
- 一人独角兽俱乐部创始人
- 12 年美国生活经验，中英文双母语思维

## 写作规则
{self.series_rules}

## 品牌调性
{self.brand_voice}

## 禁用表达（出现任何一个都算质量不合格）
{json.dumps(self.banned_phrases, ensure_ascii=False)}

## 核心要求
1. 用原生中文写作，不是英文翻译
2. 第一人称，基于实战经验
3. 2000-3000 字
4. 必须包含代码示例
5. 必须包含生产数据（延迟、成本、准确率）
6. 技术术语保留英文：Agent, RAG, LLM, token 等
7. 每个段落有信息密度，不堆废话

## 文章结构
开场 hook → 问题背景 → 核心架构 → 实现细节(含代码) → 实战经验 → 总结(3 takeaway)"""

        response = self.client.messages.create(
            model="claude-sonnet-4-5-20250514",
            max_tokens=8192,
            system=system_prompt,
            messages=[{
                "role": "user",
                "content": f"""选题：{topic['title']}
描述：{topic['description']}

调研数据：
{json.dumps(research_data, ensure_ascii=False, indent=2)}

请写出完整的文章。"""
            }]
        )

        return response.content[0].text

Agent 3: Quality Reviewer（质量审核员）

class QualityReviewer:
    """质量审核 Agent"""

    def __init__(self, rubric: dict, anti_ai_patterns: list[str]):
        self.client = anthropic.Anthropic()
        self.rubric = rubric
        self.anti_ai_patterns = anti_ai_patterns

    async def review(self, article: str, topic: dict) -> dict:
        """审核文章质量"""

        # 第一步：规则检查（不需要 LLM）
        rule_issues = self._rule_check(article)

        # 第二步：LLM 评估
        review_prompt = f"""你是一个严格的内容质量审核员。

## 评分标准（每项 20 分，共 100 分）

1. 内容深度 (20分)
   - 有具体数据和案例？
   - 有代码示例？
   - 有生产环境数据？

2. 结构清晰度 (20分)
   - 逻辑流畅？
   - 过渡自然？
   - 层次分明？

3. 品牌一致性 (20分)
   - 符合第一人称实战者视角？
   - 语气干练、不废话？
   - 没有 AI 味道？

4. 技术准确性 (20分)
   - 代码语法正确？
   - 数据和描述一致？
   - 术语使用准确？

5. 原创价值 (20分)
   - 有独特观点或经验？
   - 不是泛泛而谈的教程？
   - 读者有收获？

输出 JSON：
{{
  "total_score": 82,
  "dimensions": {{
    "content_depth": {{"score": 18, "feedback": "..."}},
    "structure": {{"score": 16, "feedback": "..."}},
    "brand_voice": {{"score": 17, "feedback": "..."}},
    "technical_accuracy": {{"score": 15, "feedback": "..."}},
    "originality": {{"score": 16, "feedback": "..."}}
  }},
  "critical_issues": ["..."],
  "improvement_suggestions": ["..."]
}}

选题：{topic['title']}

文章内容：
{article}"""

        response = self.client.messages.create(
            model="claude-sonnet-4-5-20250514",
            max_tokens=2048,
            messages=[{"role": "user", "content": review_prompt}]
        )

        llm_review = json.loads(response.content[0].text)

        # 合并规则检查和 LLM 评估
        if rule_issues:
            llm_review["critical_issues"].extend(rule_issues)
            # 每个禁用词出现扣 5 分
            penalty = len(rule_issues) * 5
            llm_review["total_score"] = max(0, llm_review["total_score"] - penalty)

        return llm_review

    def _rule_check(self, article: str) -> list[str]:
        """基于规则的检查（禁用词等）"""
        issues = []
        for phrase in self.anti_ai_patterns:
            if phrase in article:
                issues.append(f"包含禁用词: '{phrase}'")
        return issues

Agent 4: Revision Agent（修改员）

class RevisionAgent:
    """修改 Agent：根据审核意见修改文章"""

    def __init__(self):
        self.client = anthropic.Anthropic()

    async def revise(
        self, article: str, review: dict, attempt: int
    ) -> str:
        """根据审核意见修改文章"""
        system_prompt = f"""你是一个文章修改专家。

你收到了一篇文章和审核意见。请按照审核意见修改文章。

## 修改规则
1. 只修改审核意见指出的问题，不要大改没问题的部分
2. 保持原文的整体结构和风格
3. 如果审核意见要求增加内容，要自然融入不要生硬插入
4. 禁用词必须全部替换
5. 这是第 {attempt}/2 次修改，请认真处理每个问题

审核意见：
{json.dumps(review, ensure_ascii=False, indent=2)}"""

        response = self.client.messages.create(
            model="claude-sonnet-4-5-20250514",
            max_tokens=8192,
            system=system_prompt,
            messages=[{
                "role": "user",
                "content": f"请修改以下文章：\n\n{article}"
            }]
        )

        return response.content[0].text

Agent 5: Format Agent（排版员）

class FormatAgent:
    """格式化 Agent：生成最终文件"""

    def __init__(self):
        self.client = anthropic.Anthropic()

    async def format_article(
        self, article: str, topic: dict, quality_score: int
    ) -> str:
        """添加 frontmatter，验证格式，输出最终文件"""

        # 计算字数
        word_count = len(article.replace(" ", "").replace("\n", ""))

        frontmatter = f"""---
title: "{topic['title']}"
date: 2026-03-07
series: {topic['series']}
topic_id: {topic['topic_id']}
lang: zh
format: article
word_count: {word_count}
status: draft
quality_score: {quality_score}
images: []
tags: {json.dumps(topic['tags'], ensure_ascii=False)}
twitter_summary: ""
---"""

        return f"{frontmatter}\n\n{article}"

Pipeline 编排

class ContentPipeline:
    """内容生产 Pipeline：编排所有 Agent"""

    def __init__(self, series_config: dict):
        self.research = ResearchAgent()
        self.writer = ChineseDraftAgent(
            series_rules=series_config["rules"],
            brand_voice=series_config["voice"],
            anti_ai_patterns=series_config["banned_phrases"]
        )
        self.reviewer = QualityReviewer(
            rubric=series_config["rubric"],
            anti_ai_patterns=series_config["banned_phrases"]
        )
        self.reviser = RevisionAgent()
        self.formatter = FormatAgent()
        self.max_revision_cycles = 2

    async def produce(self, topic: dict) -> dict:
        """生产一篇文章的完整流程"""

        # 1. 调研
        research_data = await self.research.research(topic)

        # 2. 写初稿
        draft = await self.writer.write(topic, research_data)

        # 3. 审核 + 修改循环
        current_draft = draft
        for cycle in range(self.max_revision_cycles + 1):
            review = await self.reviewer.review(current_draft, topic)

            if review["total_score"] >= 75:
                # 通过审核
                final = await self.formatter.format_article(
                    current_draft, topic, review["total_score"]
                )
                return {
                    "status": "approved",
                    "content": final,
                    "score": review["total_score"],
                    "revision_cycles": cycle
                }

            if cycle < self.max_revision_cycles:
                # 没通过，修改后重新审核
                current_draft = await self.reviser.revise(
                    current_draft, review, cycle + 1
                )

        # 两次修改后仍未通过，标记需要人工审核
        return {
            "status": "needs_human_review",
            "content": current_draft,
            "score": review["total_score"],
            "last_review": review
        }

实战经验

生产数据

这套系统跑了 300+ 个选题，分为推文和文章两种格式：

指标	推文系列 (D+F, 90篇)	文章系列 (A+B+C+E, 210篇)
单篇调研耗时	8s	25s
单篇写作耗时	12s	45s
单篇审核耗时	5s	18s
单篇总成本	$0.06	$0.22
一次通过率	78%	68%
修改一次后通过率	95%	91%
需人工审核率	5%	9%
批量处理（并行）	10 篇/批，约 2 分钟	5 篇/批，约 4 分钟

300 篇内容总 API 成本：约 $58。时间成本：两周（包括系统搭建 + 人工审核迭代）。

踩过的坑

坑 1：写手 Agent 的 "创意过度"。 最初写手 Agent 会自己编造生产数据（"延迟降低了 47%"）。解决方案：在 system prompt 中强调 "所有数据必须来自调研报告，不得编造"，并且审核 Agent 专门检查数据是否有来源。

坑 2：审核 Agent 给自家 Agent 的文章打高分。 用同一个模型既写文章又审核，存在 "自我偏好" 的问题。解决方案：审核 Agent 用不同的 temperature（0.3 vs 写作的 0.7），并且在 prompt 中强调 "你和写手是不同的人，请客观评价"。

坑 3：并行执行时的 rate limit。 5 篇文章同时调用 Claude API，很容易触发 rate limit（Claude API 的 Sonnet 级别默认 4,000 RPM）。解决方案：加入 semaphore 控制并发数，同时用指数退避处理 429 错误。

坑 4：禁用词漏网。 审核 Agent 用 LLM 做检查时，偶尔会漏掉禁用词。解决方案：禁用词用正则匹配做 hard check，LLM 审核只做语义和质量评估。规则能解决的事情不要交给 LLM。

总结

三条核心 takeaway：

1. Agent Pipeline 的价值不在于单个 Agent 的能力，而在于流程的自动化——调研、写作、审核、修改、排版，每一步都不难，但串起来的流程管理才是真正节省时间的地方。

2. 质量门控是 Agent Pipeline 的核心——没有审核 Agent 的 Pipeline 是一个批量生产垃圾的机器。75 分的阈值 + 最多 2 轮修改 + 人工兜底，这三层门控保证了输出质量。

3. 规则检查和 LLM 检查要配合使用——禁用词、格式验证、数据完整性用规则检查（确定性、零成本）；语气一致性、内容深度、原创性用 LLM 评估（灵活但有波动）。两者互补。

如果你也想搭建内容生产 Agent，建议从最简单的两步开始——一个写作 Agent + 一个审核 Agent。先用 10 篇内容验证质量，调好 prompt 后再加调研和排版环节。

你在用 AI 做内容生产吗？什么环节效率提升最明显？欢迎讨论。