第 73 章

案例：多 Agent 协作内容创作系统

第七十三章：案例：多 Agent 协作内容创作系统

章节导语

一篇真正优秀的内容，绝不只是"好文章"——它需要准确的信息、引人入胜的叙事、精准的 SEO 布局、适合不同平台的传播格式，还需要经过严格的编辑审查。这些工作如果由一个人完成，往往顾此失彼；如果由多个人协作，则需要大量沟通协调成本。本章将设计一个多 Agent 协作内容创作系统，由四个专用 Agent（研究员、作家、编辑、SEO 专家）协作完成，Orchestrator Agent 负责调度与质量把关。给定一个主题，系统将自动输出：完整文章 + 配图描述 + SEO 元数据 + 社媒传播版本。

73.1 需求与系统目标

内容创作的典型工作流

传统内容创作流程（人工）：
主题确定 → 研究（1-2天）→ 写作（2-4小时）→ 编辑审查（1-2小时）
→ SEO 优化（30分钟）→ 配图/社媒适配（30分钟）
总计：约 1.5-2 天

目标：多 Agent 自动化
总计：约 15-30 分钟（含质量评分 ≥ 80 分的迭代）

输出规格

输出物	规格	格式
主体文章	1500-3000 字，结构完整	Markdown
配图描述	3-5 张图的 AI 生成提示词	JSON
SEO 元数据	Title/Description/Keywords	JSON
Twitter/X 线程	5-8 条推文	文本列表
LinkedIn 版本	800 字专业版	Markdown
质量评分报告	各维度评分	JSON

73.2 多 Agent 系统架构

Agent 角色设计

┌─────────────────────────────────────────────────────────────┐
│                     Orchestrator Agent                       │
│  （任务调度 + 质量把关 + 迭代控制）                           │
│  负责：拆解任务 → 分配给专用 Agent → 聚合结果 → 质量评估     │
└────────────────────────────┬────────────────────────────────┘
                             │ 协调
          ┌──────────────────┼──────────────────┐
          ▼                  ▼                  ▼
┌─────────────────┐ ┌─────────────────┐ ┌──────────────────┐
│  研究员 Agent   │ │   作家 Agent    │ │  编辑 Agent      │
│                 │ │                 │ │                  │
│ 职责：          │ │ 职责：          │ │ 职责：           │
│ - 主题背景调研  │ │ - 生成初稿      │ │ - 事实核查       │
│ - 关键数据收集  │ │ - 叙事结构设计  │ │ - 逻辑一致性检查 │
│ - 竞品内容分析  │ │ - 案例故事写作  │ │ - 语言风格统一   │
│ - 关键词挖掘    │ │ - 配图描述生成  │ │ - 反馈修改意见   │
└─────────────────┘ └─────────────────┘ └──────────────────┘
                              │
                              ▼
                   ┌─────────────────┐
                   │  SEO 专家 Agent │
                   │                 │
                   │ 职责：          │
                   │ - 关键词优化    │
                   │ - Meta标签撰写  │
                   │ - 社媒改写      │
                   │ - 发布策略建议  │
                   └─────────────────┘

数据流设计

[输入：主题]
    ↓
[研究员] → 输出：研究摘要（事实、数据、视角、关键词）
    ↓
[作家] 读取研究摘要 → 输出：初稿文章 + 配图描述
    ↓
[编辑] 读取初稿 → 输出：修改意见 + 修订稿
    ↓
[SEO专家] 读取修订稿 → 输出：SEO元数据 + 社媒版本
    ↓
[Orchestrator] 质量评估 → 
    如果分数 < 80 → 请求修改
    如果分数 ≥ 80 → 输出最终内容包

73.3 完整实现代码

共享数据结构

# content_system/models.py
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class ResearchBrief:
    """研究员输出的研究摘要"""
    topic: str
    background: str                    # 主题背景
    key_facts: List[str]              # 关键事实列表
    statistics: List[dict]            # 数据统计（含来源）
    target_keywords: List[str]        # 目标关键词
    competitor_analysis: str          # 竞品内容分析
    recommended_angle: str            # 推荐切入角度
    sources: List[str]               # 参考来源

@dataclass
class DraftArticle:
    """作家输出的初稿"""
    title: str
    hook: str                         # 开篇钩子
    sections: List[dict]              # [{heading, content}]
    conclusion: str
    call_to_action: str
    image_prompts: List[dict]         # AI配图提示词
    word_count: int = 0

@dataclass
class EditedArticle:
    """编辑输出的修订稿"""
    title: str
    content: str                      # 完整修订后文章
    editor_notes: List[str]          # 编辑意见
    fact_check_results: List[dict]   # 事实核查结果
    quality_score: dict              # 各维度评分

@dataclass
class ContentPackage:
    """最终内容包（所有输出物）"""
    topic: str
    article: str                     # 最终文章
    seo_meta: dict                   # SEO元数据
    image_prompts: List[dict]        # 配图提示词
    twitter_thread: List[str]        # Twitter线程
    linkedin_post: str               # LinkedIn版本
    quality_report: dict             # 质量评分报告
    generation_time_seconds: float

Orchestrator Agent

# content_system/orchestrator.py
import os
import json
import time
from openai import OpenAI
from .models import ResearchBrief, DraftArticle, EditedArticle, ContentPackage
from .agents import researcher, writer, editor, seo_expert

client = OpenAI(
    base_url=os.getenv("HERMES_BASE_URL", "http://localhost:11434/v1"),
    api_key=os.getenv("HERMES_API_KEY", "ollama"),
)
MODEL = os.getenv("HERMES_MODEL", "nous-hermes-2-mixtral-8x7b-dpo")

ORCHESTRATOR_PROMPT = """你是内容创作系统的总指挥，负责协调多个专业 Agent 完成高质量内容创作。

你的职责：
1. 接收主题，制定创作计划
2. 按序调用研究员、作家、编辑、SEO专家
3. 评估每个阶段的输出质量
4. 如果质量不达标（<80分），要求相关 Agent 修改
5. 最终汇总输出完整内容包

质量评分标准：
- 信息准确性（25分）：事实是否有来源支撑
- 叙事流畅度（20分）：是否引人入胜，逻辑清晰
- 原创价值（20分）：是否有独特视角和洞察
- SEO 友好度（15分）：关键词自然融入
- 读者价值（20分）：是否真正帮助读者

最低发布标准：总分 ≥ 80 分，且信息准确性 ≥ 20 分"""

ORCHESTRATOR_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "run_researcher",
            "description": "调用研究员 Agent 收集主题相关信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "topic": {"type": "string"},
                    "research_focus": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "重点研究方向"
                    }
                },
                "required": ["topic"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "run_writer",
            "description": "调用作家 Agent 基于研究摘要写作",
            "parameters": {
                "type": "object",
                "properties": {
                    "research_brief": {"type": "object"},
                    "style": {
                        "type": "string",
                        "enum": ["technical", "conversational", "narrative", "academic"],
                        "default": "conversational"
                    },
                    "target_length": {"type": "integer", "default": 2000}
                },
                "required": ["research_brief"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "run_editor",
            "description": "调用编辑 Agent 审查和修订文章",
            "parameters": {
                "type": "object",
                "properties": {
                    "draft": {"type": "object"},
                    "research_brief": {"type": "object"},
                    "edit_focus": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "重点审查方面"
                    }
                },
                "required": ["draft", "research_brief"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "run_seo_expert",
            "description": "调用 SEO 专家生成优化建议和多平台版本",
            "parameters": {
                "type": "object",
                "properties": {
                    "article": {"type": "object"},
                    "target_keywords": {
                        "type": "array",
                        "items": {"type": "string"}
                    },
                    "platforms": {
                        "type": "array",
                        "items": {"type": "string"},
                        "default": ["twitter", "linkedin"]
                    }
                },
                "required": ["article", "target_keywords"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "evaluate_quality",
            "description": "对内容进行质量评分",
            "parameters": {
                "type": "object",
                "properties": {
                    "article_content": {"type": "string"},
                    "research_brief": {"type": "object"}
                },
                "required": ["article_content", "research_brief"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "compile_final_package",
            "description": "编译最终内容包并输出",
            "parameters": {
                "type": "object",
                "properties": {
                    "article": {"type": "string"},
                    "seo_data": {"type": "object"},
                    "social_versions": {"type": "object"},
                    "image_prompts": {"type": "array"},
                    "quality_report": {"type": "object"}
                },
                "required": ["article", "seo_data", "quality_report"]
            }
        }
    }
]


def run_content_system(topic: str, style: str = "conversational") -> ContentPackage:
    """运行完整内容创作系统"""
    start_time = time.time()
    print(f"[Orchestrator] 开始创作：{topic}")
    
    messages = [
        {"role": "system", "content": ORCHESTRATOR_PROMPT},
        {
            "role": "user",
            "content": f"""请为以下主题创作完整的内容包：

**主题：** {topic}
**写作风格：** {style}
**输出要求：**
1. 完整文章（1500-3000字）
2. 3-5 张 AI 配图提示词
3. SEO 元数据（title/description/keywords）
4. Twitter 线程（5-8条）
5. LinkedIn 版本（800字）
6. 质量评分报告

请协调各专业 Agent 完成此任务，确保最终质量分数 ≥ 80 分。"""
        }
    ]
    
    final_package = None
    max_iterations = 20
    iteration_count = 0
    
    while iteration_count < max_iterations:
        iteration_count += 1
        
        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=ORCHESTRATOR_TOOLS,
            tool_choice="auto",
            temperature=0.4,
        )
        
        message = response.choices[0].message
        messages.append(message)
        
        if not message.tool_calls:
            # Orchestrator 认为任务完成
            break
        
        for tool_call in message.tool_calls:
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            print(f"[Orchestrator] 调用：{name}")
            
            result = _dispatch_orchestrator_tool(name, args)
            
            if name == "compile_final_package" and result.get("success"):
                final_package = result.get("package")
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })
    
    generation_time = time.time() - start_time
    print(f"[Orchestrator] 完成！耗时 {generation_time:.1f} 秒")
    
    return final_package or {"status": "failed", "iterations": iteration_count}


def _dispatch_orchestrator_tool(name: str, args: dict) -> dict:
    """工具调度"""
    if name == "run_researcher":
        return researcher.run(**args)
    elif name == "run_writer":
        return writer.run(**args)
    elif name == "run_editor":
        return editor.run(**args)
    elif name == "run_seo_expert":
        return seo_expert.run(**args)
    elif name == "evaluate_quality":
        return _evaluate_quality(**args)
    elif name == "compile_final_package":
        return _compile_package(**args)
    return {"error": f"未知工具: {name}"}

研究员 Agent

# content_system/agents/researcher.py
import os, json
from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("HERMES_BASE_URL"), api_key=os.getenv("HERMES_API_KEY")
)
MODEL = os.getenv("HERMES_MODEL")

RESEARCHER_PROMPT = """你是一位专业的内容研究员。
你的任务是为给定主题进行深入研究，输出结构化的研究摘要供作家参考。

研究原则：
- 收集多角度视角，不偏向单一观点
- 注重数据和案例的具体性
- 识别主题的独特切入点
- 关注目标受众的痛点和需求

输出格式（JSON）：
{
  "background": "主题背景...",
  "key_facts": ["事实1", "事实2"],
  "statistics": [{"data": "...", "source": "..."}],
  "target_keywords": ["关键词1", "关键词2"],
  "competitor_analysis": "竞品内容分析...",
  "recommended_angle": "推荐切入角度",
  "sources": ["url1", "url2"]
}"""

RESEARCHER_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_topic",
            "description": "搜索主题相关信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "type": {"type": "string", "enum": ["general", "news", "academic"]}
                },
                "required": ["query"]
            }
        }
    }
]

def run(topic: str, research_focus: list = None) -> dict:
    """运行研究员 Agent"""
    messages = [
        {"role": "system", "content": RESEARCHER_PROMPT},
        {"role": "user", "content": f"请研究主题：{topic}\n重点关注：{research_focus or '全面覆盖'}"}
    ]
    
    # 简化版：直接让 LLM 生成研究摘要
    # 实际生产环境应接入搜索 API
    response = client.chat.completions.create(
        model=MODEL, messages=messages, temperature=0.3,
        response_format={"type": "json_object"}
    )
    
    try:
        brief = json.loads(response.choices[0].message.content)
        brief["topic"] = topic
        return {"success": True, "brief": brief}
    except Exception as e:
        return {"success": False, "error": str(e)}

作家 Agent

# content_system/agents/writer.py
import os, json
from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("HERMES_BASE_URL"), api_key=os.getenv("HERMES_API_KEY")
)
MODEL = os.getenv("HERMES_MODEL")

WRITER_PROMPT = """你是一位优秀的内容作家，擅长将复杂信息写成引人入胜的文章。

写作原则：
- 开篇必须有力，3秒内抓住读者注意力
- 每个章节都要有实质内容，不堆砌废话
- 用故事和案例让抽象概念具体化
- 结尾要有清晰的行动号召

同时为每个关键部分生成 AI 配图提示词（英文，Midjourney/DALL-E 格式）"""

def run(research_brief: dict, style: str = "conversational", target_length: int = 2000) -> dict:
    """运行作家 Agent"""
    brief_str = json.dumps(research_brief, ensure_ascii=False, indent=2)
    
    messages = [
        {"role": "system", "content": WRITER_PROMPT},
        {
            "role": "user",
            "content": f"""基于以下研究摘要，写一篇{target_length}字的{style}风格文章：

{brief_str}

输出格式（JSON）：
{{
  "title": "文章标题",
  "hook": "开篇钩子（前100字）",
  "sections": [
    {{"heading": "章节标题", "content": "章节内容"}}
  ],
  "conclusion": "结论",
  "call_to_action": "行动号召",
  "image_prompts": [
    {{"position": "header", "prompt": "英文提示词", "description": "图片描述"}}
  ]
}}"""
        }
    ]
    
    response = client.chat.completions.create(
        model=MODEL, messages=messages, temperature=0.7,
        max_tokens=4000, response_format={"type": "json_object"}
    )
    
    try:
        draft = json.loads(response.choices[0].message.content)
        # 计算字数
        full_text = draft.get("hook", "") + " ".join(
            s.get("content", "") for s in draft.get("sections", [])
        ) + draft.get("conclusion", "")
        draft["word_count"] = len(full_text)
        return {"success": True, "draft": draft}
    except Exception as e:
        return {"success": False, "error": str(e)}

编辑 Agent

# content_system/agents/editor.py
import os, json
from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("HERMES_BASE_URL"), api_key=os.getenv("HERMES_API_KEY")
)
MODEL = os.getenv("HERMES_MODEL")

EDITOR_PROMPT = """你是一位严格的资深内容编辑。
你的任务：事实核查、逻辑审查、语言润色、质量把关。

审查清单：
□ 所有统计数据是否有来源？
□ 逻辑是否自洽，有无前后矛盾？
□ 语言是否流畅，有无生硬翻译感？
□ 标题是否吸引人？
□ 开篇钩子是否够有力？
□ 结论是否与正文呼应？

请输出：修改意见 + 修订后完整文章 + 质量评分（各维度）"""

def run(draft: dict, research_brief: dict, edit_focus: list = None) -> dict:
    """运行编辑 Agent"""
    messages = [
        {"role": "system", "content": EDITOR_PROMPT},
        {
            "role": "user",
            "content": f"""请审查并修订以下文章：

原稿：
{json.dumps(draft, ensure_ascii=False, indent=2)}

参考研究摘要（用于事实核查）：
{json.dumps(research_brief, ensure_ascii=False, indent=2)}

{'重点关注：' + str(edit_focus) if edit_focus else ''}

输出（JSON格式）：
{{
  "title": "最终标题",
  "content": "修订后完整文章（Markdown格式）",
  "editor_notes": ["修改意见1", "修改意见2"],
  "fact_check_results": [{{"claim": "...", "status": "verified/unverified/corrected"}}],
  "quality_scores": {{
    "accuracy": 0-25,
    "narrative": 0-20,
    "originality": 0-20,
    "seo_readiness": 0-15,
    "reader_value": 0-20
  }},
  "total_score": 0-100
}}"""
        }
    ]
    
    response = client.chat.completions.create(
        model=MODEL, messages=messages, temperature=0.2,
        max_tokens=5000, response_format={"type": "json_object"}
    )
    
    try:
        edited = json.loads(response.choices[0].message.content)
        return {"success": True, "edited": edited}
    except Exception as e:
        return {"success": False, "error": str(e)}

SEO 专家 Agent

# content_system/agents/seo_expert.py
import os, json
from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("HERMES_BASE_URL"), api_key=os.getenv("HERMES_API_KEY")
)
MODEL = os.getenv("HERMES_MODEL")

SEO_PROMPT = """你是一位数字营销和 SEO 专家，同时擅长多平台内容运营。

SEO 优化原则：
- Title: 60字符以内，包含核心关键词，有吸引力
- Description: 120-160字符，包含关键词，有行动号召
- 关键词密度：核心词 1-2%，语义相关词自然分布

社媒平台写作风格：
- Twitter：简洁有力，每条 ≤ 280字符，第一条最重要，末条有CTA
- LinkedIn：专业深度，可用表情符号，适当空行，重视第一段"""

def run(article: dict, target_keywords: list, platforms: list = None) -> dict:
    """运行 SEO 专家 Agent"""
    if platforms is None:
        platforms = ["twitter", "linkedin"]
    
    article_content = article.get("content", "")
    title = article.get("title", "")
    
    messages = [
        {"role": "system", "content": SEO_PROMPT},
        {
            "role": "user",
            "content": f"""为以下文章生成 SEO 优化和多平台版本：

标题：{title}
目标关键词：{target_keywords}
文章内容：
{article_content[:3000]}...（节选）

需要输出（JSON格式）：
{{
  "seo_meta": {{
    "title": "SEO标题",
    "description": "Meta描述",
    "keywords": ["关键词列表"],
    "og_title": "社交分享标题",
    "og_description": "社交分享描述"
  }},
  "twitter_thread": [
    "第1条推文（最重要，包含钩子）",
    "第2条推文",
    "...",
    "最后一条（CTA）"
  ],
  "linkedin_post": "LinkedIn版本全文（800字，Markdown格式）",
  "keyword_suggestions": ["追加建议关键词"]
}}"""
        }
    ]
    
    response = client.chat.completions.create(
        model=MODEL, messages=messages, temperature=0.5,
        max_tokens=3000, response_format={"type": "json_object"}
    )
    
    try:
        result = json.loads(response.choices[0].message.content)
        return {"success": True, "seo_data": result}
    except Exception as e:
        return {"success": False, "error": str(e)}

73.4 质量评分机制

# content_system/quality.py

def evaluate_content_quality(article_content: str, research_brief: dict) -> dict:
    """多维度内容质量评估"""
    scores = {}
    
    # 1. 信息准确性（25分）
    # 检查文章中是否引用了研究摘要中的关键事实
    key_facts = research_brief.get("key_facts", [])
    fact_matches = sum(1 for fact in key_facts if _fuzzy_contains(article_content, fact))
    scores["accuracy"] = min(25, int(25 * fact_matches / max(len(key_facts), 1)))
    
    # 2. 结构完整性（检查是否有标题、段落、结论）
    has_headings = article_content.count("##") >= 2
    has_conclusion = any(kw in article_content.lower() for kw in ["结论", "总结", "conclusion"])
    scores["structure"] = 15 if (has_headings and has_conclusion) else 8
    
    # 3. 长度适当性
    word_count = len(article_content)
    if 1500 <= word_count <= 4000:
        scores["length"] = 10
    elif 800 <= word_count < 1500:
        scores["length"] = 6
    else:
        scores["length"] = 3
    
    # 4. 关键词覆盖度（15分）
    keywords = research_brief.get("target_keywords", [])
    kw_hits = sum(1 for kw in keywords if kw.lower() in article_content.lower())
    scores["keyword_coverage"] = min(15, int(15 * kw_hits / max(len(keywords), 1)))
    
    # 5. 可读性（20分）：平均句子长度
    sentences = [s.strip() for s in article_content.split("。") if s.strip()]
    avg_sentence_len = sum(len(s) for s in sentences) / max(len(sentences), 1)
    if 20 <= avg_sentence_len <= 60:
        scores["readability"] = 20
    elif avg_sentence_len < 20 or avg_sentence_len > 100:
        scores["readability"] = 10
    else:
        scores["readability"] = 15
    
    total = sum(scores.values())
    
    return {
        "scores": scores,
        "total": total,
        "grade": "A" if total >= 85 else "B" if total >= 70 else "C",
        "pass": total >= 80,
        "recommendations": _generate_recommendations(scores)
    }


def _fuzzy_contains(text: str, fact: str) -> bool:
    """宽松匹配：检查文本是否包含事实的主要关键词"""
    words = [w for w in fact.split() if len(w) > 2]
    return sum(1 for w in words if w in text) >= len(words) * 0.6


def _generate_recommendations(scores: dict) -> list:
    recs = []
    if scores.get("accuracy", 0) < 18:
        recs.append("增加更多具体数据和来源引用")
    if scores.get("structure", 0) < 12:
        recs.append("完善文章结构：添加清晰的章节标题和结论")
    if scores.get("keyword_coverage", 0) < 10:
        recs.append("自然融入更多目标关键词")
    if scores.get("readability", 0) < 15:
        recs.append("调整句子长度，提高可读性")
    return recs

73.5 完整运行示例

# run.py
from content_system.orchestrator import run_content_system
import json

# 运行内容创作系统
result = run_content_system(
    topic="AI Agent 如何改变软件开发工作流",
    style="conversational"
)

if result and result.get("article"):
    print("=" * 60)
    print("最终文章：")
    print(result["article"])
    print("\nSEO 元数据：")
    print(json.dumps(result["seo_meta"], ensure_ascii=False, indent=2))
    print("\nTwitter 线程：")
    for i, tweet in enumerate(result.get("twitter_thread", []), 1):
        print(f"{i}. {tweet}")
    print(f"\n质量评分：{result['quality_report']['total']}/100")

本章小结

本章设计并实现了一个生产级多 Agent 协作内容创作系统：

角色分工：研究员/作家/编辑/SEO专家各司其职，避免单一 Agent 质量均值回归
Orchestrator 模式：中央调度器负责协调、质量把关和迭代控制
数据流设计：明确的接口契约使各 Agent 可独立测试和替换
质量闭环：内置评分机制确保输出达到可发布标准

多 Agent 系统的核心价值不在于并行加速（实际上协作 Agent 往往是串行的），而在于专业化分工带来的质量提升——就像一篇优秀文章需要作者、编辑、事实核查员的协作，而不只是一个人的独自努力。

思考题

如何设计 Agent 间的通信协议，使其既灵活又不会产生信息损耗？
当编辑 Agent 的意见与作家 Agent 的风格产生冲突时，Orchestrator 如何仲裁？
如何扩展这个系统支持视频脚本、播客内容等非文字形式？
多 Agent 系统的调试和可观测性如何设计？

本章评分

4.7 / 5 (3 评分)

案例：多 Agent 协作内容创作系统

第七十三章：案例：多 Agent 协作内容创作系统

73.1 需求与系统目标

内容创作的典型工作流

输出规格

73.2 多 Agent 系统架构

Agent 角色设计

数据流设计

73.3 完整实现代码

共享数据结构

Orchestrator Agent

研究员 Agent

作家 Agent

编辑 Agent

SEO 专家 Agent

73.4 质量评分机制

73.5 完整运行示例

本章小结

思考题

💬 留言讨论