单个大模型做SEO任务，容易顾此失彼。我设计了一个多Agent系统：研究Agent、写作Agent、技术Agent各司其职，通过共享任务队列协作。这篇文章分享架构设计和实现代码。

一、为什么需要多Agent

让一个LLM同时做研究、写作、技术审核，效果通常不好：

• 上下文不够：研究阶段收集的信息，写作时已经"忘记"了一半
• 角色冲突：又要创意又要严谨，模型容易精神分裂
• 串行效率低：研究→写作→审核，一步错后面全废

多Agent方案：每个Agent专职一个角色，通过消息队列协作。

二、系统架构

任务队列（Redis/RabbitMQ）
    ├── 研究Agent（Researcher）
    │     └─ 采集SERP → 分析竞品 → 输出brief
    ├── 写作Agent（Writer）
    │     └─ 读取brief → 生成文章 → 输出草稿
    ├── 技术Agent（Tech SEO）
    │     └─ 审核草稿 → 检查schema/内链/标题 → 输出修改建议
    └── 发布Agent（Publisher）
          └─ 整合修改 → 生成最终HTML → 发布

三、Agent实现

3.1 研究Agent

# agents/researcher.py
import requests
from typing import Dict, List

class ResearchAgent:
    def __init__(self, serp_key: str, llm_client):
        self.serp_key = serp_key
        self.llm = llm_client
        self.base_url = "https://api.serpbase.dev/google/search"
    
    async def research(self, task: Dict) -> Dict:
        """研究任务：给定关键词，输出内容简报"""
        keyword = task["keyword"]
        
        # 1. 采集SERP
        serp_data = await self._fetch_serp(keyword)
        
        # 2. 分析竞品
        competitor_analysis = self._analyze_competitors(serp_data["organic"][:5])
        
        # 3. 提取PAA和Related
        questions = [item.get("question", "") for item in serp_data.get("people_also_ask", [])]
        related = serp_data.get("related_searches", [])[:10]
        
        # 4. 用LLM生成brief
        brief = await self._generate_brief(keyword, competitor_analysis, questions, related)
        
        return {
            "task_id": task["id"],
            "agent": "researcher",
            "status": "completed",
            "output": {
                "keyword": keyword,
                "brief": brief,
                "competitor_analysis": competitor_analysis,
                "must_answer_questions": questions,
                "related_topics": related,
                "suggested_word_count": self._estimate_length(serp_data)
            }
        }
    
    async def _fetch_serp(self, keyword: str) -> Dict:
        headers = {
            "X-API-Key": self.serp_key,
            "Content-Type": "application/json"
        }
        body = {
            "q": keyword,
            "hl": "en",
            "gl": "us",
            "page": 1
        }
        
        r = requests.post(self.base_url, headers=headers, json=body, timeout=30)
        return r.json()
    
    def _analyze_competitors(self, organic: List[Dict]) -> List[Dict]:
        """分析前5名竞品"""
        analysis = []
        for item in organic:
            analysis.append({
                "rank": item["rank"],
                "domain": item.get("display_link", ""),
                "title": item.get("title", ""),
                "content_type": self._detect_content_type(item.get("title", "")),
                "estimated_length": "long" if len(item.get("snippet", "")) > 150 else "short"
            })
        return analysis
    
    def _detect_content_type(self, title: str) -> str:
        title_lower = title.lower()
        if any(w in title_lower for w in ["how to", "guide", "tutorial"]):
            return "guide"
        elif any(w in title_lower for w in ["best", "top", "vs"]):
            return "listicle"
        elif "review" in title_lower:
            return "review"
        return "article"
    
    def _estimate_length(self, serp_data: Dict) -> int:
        snippets = [item.get("snippet", "") for item in serp_data.get("organic", [])[:3]]
        avg_length = sum(len(s) for s in snippets) / len(snippets) if snippets else 0
        return 2500 if avg_length > 150 else 1500
    
    async def _generate_brief(self, keyword, competitors, questions, related) -> str:
        prompt = f"""为"{keyword}"生成内容简报...
        # 省略详细prompt，参考前面的content brief文章
        """
        return self.llm.chat(prompt)

3.2 写作Agent

# agents/writer.py
class WriterAgent:
    def __init__(self, llm_client):
        self.llm = llm_client
    
    async def write(self, research_output: Dict) -> Dict:
        """基于brief写文章"""
        brief = research_output["output"]["brief"]
        keyword = research_output["output"]["keyword"]
        word_count = research_output["output"]["suggested_word_count"]
        
        # 构建写作prompt
        prompt = f"""你是一个专业的技术内容写手。请基于以下内容简报，写一篇高质量的SEO文章。

目标关键词：{keyword}
目标字数：{word_count}

内容简报：
{brief}

要求：
1. 标题必须包含目标关键词
2. 使用H2/H3结构化内容
3. 包含具体的例子和代码片段
4. 语言自然，不要过度优化
5. 结尾要有明确的CTA

请直接输出文章正文（Markdown格式）。"""
        
        article = self.llm.chat(prompt)
        
        return {
            "task_id": research_output["task_id"],
            "agent": "writer",
            "status": "completed",
            "output": {
                "article": article,
                "word_count": len(article.split()),
                "keyword": keyword
            }
        }

3.3 技术审核Agent

# agents/tech_auditor.py
class TechAuditorAgent:
    def __init__(self, llm_client):
        self.llm = llm_client
    
    async def audit(self, writer_output: Dict, research_output: Dict) -> Dict:
        """技术SEO审核"""
        article = writer_output["output"]["article"]
        keyword = research_output["output"]["keyword"]
        
        issues = []
        
        # 1. 检查标题
        title_match = re.search(r'^# (.+)$', article, re.MULTILINE)
        if title_match:
            title = title_match.group(1)
            if keyword.lower() not in title.lower():
                issues.append(f"标题未包含关键词 '{keyword}'")
            if len(title) > 60:
                issues.append(f"标题过长 ({len(title)}字符)，建议<60")
        
        # 2. 检查H1/H2结构
        h2_count = len(re.findall(r'^## ', article, re.MULTILINE))
        if h2_count < 3:
            issues.append(f"H2数量过少 ({h2_count})，建议至少5个")
        
        # 3. 检查关键词密度
        word_count = len(article.split())
        keyword_count = article.lower().count(keyword.lower())
        density = keyword_count / word_count * 100
        if density > 3:
            issues.append(f"关键词密度过高 ({density:.1f}%)，建议<2%")
        elif density < 0.5:
            issues.append(f"关键词密度过低 ({density:.1f}%)，建议>1%")
        
        # 4. 检查内部链接机会
        internal_link_suggestions = self._suggest_internal_links(article, keyword)
        
        return {
            "task_id": writer_output["task_id"],
            "agent": "tech_auditor",
            "status": "completed",
            "output": {
                "issues": issues,
                "internal_link_suggestions": internal_link_suggestions,
                "seo_score": max(0, 100 - len(issues) * 10),
                "approved": len(issues) <= 3
            }
        }
    
    def _suggest_internal_links(self, article: str, keyword: str) -> List[str]:
        # 简化实现：找文章中提到的其他主题
        # 实际应该查站内内容库
        return ["related-topic-1", "related-topic-2"]

四、任务调度器

# orchestrator.py
import asyncio
from typing import Dict

class SEOOrchestrator:
    def __init__(self):
        self.agents = {
            "researcher": ResearchAgent(...),
            "writer": WriterAgent(...),
            "tech_auditor": TechAuditorAgent(...),
            "publisher": PublisherAgent(...)
        }
        self.task_queue = asyncio.Queue()
        self.results = {}
    
    async def submit_task(self, task: Dict):
        """提交新任务"""
        await self.task_queue.put(task)
    
    async def run_pipeline(self, task: Dict):
        """运行完整流水线"""
        task_id = task["id"]
        
        try:
            # Step 1: 研究
            print(f"[{task_id}] Starting research...")
            research_result = await self.agents["researcher"].research(task)
            self.results[f"{task_id}_research"] = research_result
            
            # Step 2: 写作
            print(f"[{task_id}] Starting writing...")
            writer_result = await self.agents["writer"].write(research_result)
            self.results[f"{task_id}_writer"] = writer_result
            
            # Step 3: 技术审核
            print(f"[{task_id}] Starting tech audit...")
            audit_result = await self.agents["tech_auditor"].audit(writer_result, research_result)
            self.results[f"{task_id}_audit"] = audit_result
            
            # Step 4: 如果审核通过，发布；否则返回修改
            if audit_result["output"]["approved"]:
                print(f"[{task_id}] Publishing...")
                publish_result = await self.agents["publisher"].publish(writer_result, audit_result)
                self.results[f"{task_id}_publish"] = publish_result
                return publish_result
            else:
                print(f"[{task_id}] Audit failed, returning for revision")
                return {
                    "status": "needs_revision",
                    "audit_issues": audit_result["output"]["issues"],
                    "draft": writer_result["output"]["article"]
                }
        
        except Exception as e:
            return {
                "status": "error",
                "error": str(e),
                "task_id": task_id
            }
    
    async def worker(self):
        """持续处理队列中的任务"""
        while True:
            task = await self.task_queue.get()
            result = await self.run_pipeline(task)
            print(f"Task {task['id']} completed: {result['status']}")
            self.task_queue.task_done()

五、使用示例

async def main():
    orchestrator = SEOOrchestrator()
    
    # 提交10个写作任务
    keywords = [
        "docker compose tutorial",
        "kubernetes networking",
        "ci/cd best practices",
        "devops monitoring tools",
        "microservices architecture",
        "api gateway patterns",
        "container security",
        "git workflow strategies",
        "infrastructure as code",
        "observability stack"
    ]
    
    for i, kw in enumerate(keywords):
        await orchestrator.submit_task({
            "id": f"task_{i+1}",
            "keyword": kw,
            "priority": "normal"
        })
    
    # 启动3个worker并行处理
    workers = [asyncio.create_task(orchestrator.worker()) for _ in range(3)]
    
    # 等待所有任务完成
    await orchestrator.task_queue.join()
    
    # 取消worker
    for w in workers:
        w.cancel()

asyncio.run(main())

六、实战效果

多Agent系统跑了2周的测试：

多Agent的优势：

• 每个Agent专职，输出质量更高
• 审核Agent拦截问题，减少返工
• 并行处理，整体效率更高

七、总结

多Agent SEO系统的核心价值：

1. 专业化：研究、写作、审核各司其职
2. 可扩展：加新Agent（如图片生成Agent）不影响现有流程
3. 可调试：哪个环节出问题一目了然
4. 可复用：Agent可以独立复用到其他项目

这套系统适合需要规模化内容生产的团队。单篇文章用单Agent就够了，但批量生产时多Agent优势明显。

多Agent系统的最大挑战是"通信开销"——Agent之间传递的信息格式要统一。建议定义好标准的Task和Result schema，用Pydantic模型约束。否则AgentA的输出AgentB读不懂，整个系统就崩了。