单个大模型做SEO任务容易顾此失彼。我设计了一个多Agent系统研究Agent、写作Agent、技术Agent各司其职通过共享任务队列协作。这篇文章分享架构设计和实现代码。一、为什么需要多Agent让一个LLM同时做研究、写作、技术审核效果通常不好上下文不够研究阶段收集的信息写作时已经忘记了一半角色冲突又要创意又要严谨模型容易精神分裂串行效率低研究→写作→审核一步错后面全废多Agent方案每个Agent专职一个角色通过消息队列协作。二、系统架构任务队列Redis/RabbitMQ ├── 研究AgentResearcher │ └─ 采集SERP → 分析竞品 → 输出brief ├── 写作AgentWriter │ └─ 读取brief → 生成文章 → 输出草稿 ├── 技术AgentTech SEO │ └─ 审核草稿 → 检查schema/内链/标题 → 输出修改建议 └── 发布AgentPublisher └─ 整合修改 → 生成最终HTML → 发布三、Agent实现3.1 研究Agent# agents/researcher.pyimportrequestsfromtypingimportDict,ListclassResearchAgent:def__init__(self,serp_key:str,llm_client):self.serp_keyserp_key self.llmllm_client self.base_urlhttps://api.serpbase.dev/google/searchasyncdefresearch(self,task:Dict)-Dict:研究任务给定关键词输出内容简报keywordtask[keyword]# 1. 采集SERPserp_dataawaitself._fetch_serp(keyword)# 2. 分析竞品competitor_analysisself._analyze_competitors(serp_data[organic][:5])# 3. 提取PAA和Relatedquestions[item.get(question,)foriteminserp_data.get(people_also_ask,[])]relatedserp_data.get(related_searches,[])[:10]# 4. 用LLM生成briefbriefawaitself._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)}}asyncdef_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}rrequests.post(self.base_url,headersheaders,jsonbody,timeout30)returnr.json()def_analyze_competitors(self,organic:List[Dict])-List[Dict]:分析前5名竞品analysis[]foriteminorganic: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:longiflen(item.get(snippet,))150elseshort})returnanalysisdef_detect_content_type(self,title:str)-str:title_lowertitle.lower()ifany(wintitle_lowerforwin[how to,guide,tutorial]):returnguideelifany(wintitle_lowerforwin[best,top,vs]):returnlisticleelifreviewintitle_lower:returnreviewreturnarticledef_estimate_length(self,serp_data:Dict)-int:snippets[item.get(snippet,)foriteminserp_data.get(organic,[])[:3]]avg_lengthsum(len(s)forsinsnippets)/len(snippets)ifsnippetselse0return2500ifavg_length150else1500asyncdef_generate_brief(self,keyword,competitors,questions,related)-str:promptf为{keyword}生成内容简报... # 省略详细prompt参考前面的content brief文章 returnself.llm.chat(prompt)3.2 写作Agent# agents/writer.pyclassWriterAgent:def__init__(self,llm_client):self.llmllm_clientasyncdefwrite(self,research_output:Dict)-Dict:基于brief写文章briefresearch_output[output][brief]keywordresearch_output[output][keyword]word_countresearch_output[output][suggested_word_count]# 构建写作promptpromptf你是一个专业的技术内容写手。请基于以下内容简报写一篇高质量的SEO文章。 目标关键词{keyword}目标字数{word_count}内容简报{brief}要求 1. 标题必须包含目标关键词 2. 使用H2/H3结构化内容 3. 包含具体的例子和代码片段 4. 语言自然不要过度优化 5. 结尾要有明确的CTA 请直接输出文章正文Markdown格式。articleself.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.pyclassTechAuditorAgent:def__init__(self,llm_client):self.llmllm_clientasyncdefaudit(self,writer_output:Dict,research_output:Dict)-Dict:技术SEO审核articlewriter_output[output][article]keywordresearch_output[output][keyword]issues[]# 1. 检查标题title_matchre.search(r^# (.)$,article,re.MULTILINE)iftitle_match:titletitle_match.group(1)ifkeyword.lower()notintitle.lower():issues.append(f标题未包含关键词 {keyword})iflen(title)60:issues.append(f标题过长 ({len(title)}字符)建议60)# 2. 检查H1/H2结构h2_countlen(re.findall(r^## ,article,re.MULTILINE))ifh2_count3:issues.append(fH2数量过少 ({h2_count})建议至少5个)# 3. 检查关键词密度word_countlen(article.split())keyword_countarticle.lower().count(keyword.lower())densitykeyword_count/word_count*100ifdensity3:issues.append(f关键词密度过高 ({density:.1f}%)建议2%)elifdensity0.5:issues.append(f关键词密度过低 ({density:.1f}%)建议1%)# 4. 检查内部链接机会internal_link_suggestionsself._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.pyimportasynciofromtypingimportDictclassSEOOrchestrator:def__init__(self):self.agents{researcher:ResearchAgent(...),writer:WriterAgent(...),tech_auditor:TechAuditorAgent(...),publisher:PublisherAgent(...)}self.task_queueasyncio.Queue()self.results{}asyncdefsubmit_task(self,task:Dict):提交新任务awaitself.task_queue.put(task)asyncdefrun_pipeline(self,task:Dict):运行完整流水线task_idtask[id]try:# Step 1: 研究print(f[{task_id}] Starting research...)research_resultawaitself.agents[researcher].research(task)self.results[f{task_id}_research]research_result# Step 2: 写作print(f[{task_id}] Starting writing...)writer_resultawaitself.agents[writer].write(research_result)self.results[f{task_id}_writer]writer_result# Step 3: 技术审核print(f[{task_id}] Starting tech audit...)audit_resultawaitself.agents[tech_auditor].audit(writer_result,research_result)self.results[f{task_id}_audit]audit_result# Step 4: 如果审核通过发布否则返回修改ifaudit_result[output][approved]:print(f[{task_id}] Publishing...)publish_resultawaitself.agents[publisher].publish(writer_result,audit_result)self.results[f{task_id}_publish]publish_resultreturnpublish_resultelse:print(f[{task_id}] Audit failed, returning for revision)return{status:needs_revision,audit_issues:audit_result[output][issues],draft:writer_result[output][article]}exceptExceptionase:return{status:error,error:str(e),task_id:task_id}asyncdefworker(self):持续处理队列中的任务whileTrue:taskawaitself.task_queue.get()resultawaitself.run_pipeline(task)print(fTask{task[id]}completed:{result[status]})self.task_queue.task_done()五、使用示例asyncdefmain():orchestratorSEOOrchestrator()# 提交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]fori,kwinenumerate(keywords):awaitorchestrator.submit_task({id:ftask_{i1},keyword:kw,priority:normal})# 启动3个worker并行处理workers[asyncio.create_task(orchestrator.worker())for_inrange(3)]# 等待所有任务完成awaitorchestrator.task_queue.join()# 取消workerforwinworkers:w.cancel()asyncio.run(main())六、实战效果多Agent系统跑了2周的测试指标单Agent多Agent文章通过率45%78%平均修改轮数3.21.4研究质量评分3.2/54.1/5技术合规率52%89%总耗时/篇25分钟18分钟多Agent的优势每个Agent专职输出质量更高审核Agent拦截问题减少返工并行处理整体效率更高七、总结多Agent SEO系统的核心价值专业化研究、写作、审核各司其职可扩展加新Agent如图片生成Agent不影响现有流程可调试哪个环节出问题一目了然可复用Agent可以独立复用到其他项目这套系统适合需要规模化内容生产的团队。单篇文章用单Agent就够了但批量生产时多Agent优势明显。多Agent系统的最大挑战是通信开销——Agent之间传递的信息格式要统一。建议定义好标准的Task和Result schema用Pydantic模型约束。否则AgentA的输出AgentB读不懂整个系统就崩了。
多Agent SEO协作系统：让AI工人分工做优化
单个大模型做SEO任务容易顾此失彼。我设计了一个多Agent系统研究Agent、写作Agent、技术Agent各司其职通过共享任务队列协作。这篇文章分享架构设计和实现代码。一、为什么需要多Agent让一个LLM同时做研究、写作、技术审核效果通常不好上下文不够研究阶段收集的信息写作时已经忘记了一半角色冲突又要创意又要严谨模型容易精神分裂串行效率低研究→写作→审核一步错后面全废多Agent方案每个Agent专职一个角色通过消息队列协作。二、系统架构任务队列Redis/RabbitMQ ├── 研究AgentResearcher │ └─ 采集SERP → 分析竞品 → 输出brief ├── 写作AgentWriter │ └─ 读取brief → 生成文章 → 输出草稿 ├── 技术AgentTech SEO │ └─ 审核草稿 → 检查schema/内链/标题 → 输出修改建议 └── 发布AgentPublisher └─ 整合修改 → 生成最终HTML → 发布三、Agent实现3.1 研究Agent# agents/researcher.pyimportrequestsfromtypingimportDict,ListclassResearchAgent:def__init__(self,serp_key:str,llm_client):self.serp_keyserp_key self.llmllm_client self.base_urlhttps://api.serpbase.dev/google/searchasyncdefresearch(self,task:Dict)-Dict:研究任务给定关键词输出内容简报keywordtask[keyword]# 1. 采集SERPserp_dataawaitself._fetch_serp(keyword)# 2. 分析竞品competitor_analysisself._analyze_competitors(serp_data[organic][:5])# 3. 提取PAA和Relatedquestions[item.get(question,)foriteminserp_data.get(people_also_ask,[])]relatedserp_data.get(related_searches,[])[:10]# 4. 用LLM生成briefbriefawaitself._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)}}asyncdef_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}rrequests.post(self.base_url,headersheaders,jsonbody,timeout30)returnr.json()def_analyze_competitors(self,organic:List[Dict])-List[Dict]:分析前5名竞品analysis[]foriteminorganic: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:longiflen(item.get(snippet,))150elseshort})returnanalysisdef_detect_content_type(self,title:str)-str:title_lowertitle.lower()ifany(wintitle_lowerforwin[how to,guide,tutorial]):returnguideelifany(wintitle_lowerforwin[best,top,vs]):returnlisticleelifreviewintitle_lower:returnreviewreturnarticledef_estimate_length(self,serp_data:Dict)-int:snippets[item.get(snippet,)foriteminserp_data.get(organic,[])[:3]]avg_lengthsum(len(s)forsinsnippets)/len(snippets)ifsnippetselse0return2500ifavg_length150else1500asyncdef_generate_brief(self,keyword,competitors,questions,related)-str:promptf为{keyword}生成内容简报... # 省略详细prompt参考前面的content brief文章 returnself.llm.chat(prompt)3.2 写作Agent# agents/writer.pyclassWriterAgent:def__init__(self,llm_client):self.llmllm_clientasyncdefwrite(self,research_output:Dict)-Dict:基于brief写文章briefresearch_output[output][brief]keywordresearch_output[output][keyword]word_countresearch_output[output][suggested_word_count]# 构建写作promptpromptf你是一个专业的技术内容写手。请基于以下内容简报写一篇高质量的SEO文章。 目标关键词{keyword}目标字数{word_count}内容简报{brief}要求 1. 标题必须包含目标关键词 2. 使用H2/H3结构化内容 3. 包含具体的例子和代码片段 4. 语言自然不要过度优化 5. 结尾要有明确的CTA 请直接输出文章正文Markdown格式。articleself.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.pyclassTechAuditorAgent:def__init__(self,llm_client):self.llmllm_clientasyncdefaudit(self,writer_output:Dict,research_output:Dict)-Dict:技术SEO审核articlewriter_output[output][article]keywordresearch_output[output][keyword]issues[]# 1. 检查标题title_matchre.search(r^# (.)$,article,re.MULTILINE)iftitle_match:titletitle_match.group(1)ifkeyword.lower()notintitle.lower():issues.append(f标题未包含关键词 {keyword})iflen(title)60:issues.append(f标题过长 ({len(title)}字符)建议60)# 2. 检查H1/H2结构h2_countlen(re.findall(r^## ,article,re.MULTILINE))ifh2_count3:issues.append(fH2数量过少 ({h2_count})建议至少5个)# 3. 检查关键词密度word_countlen(article.split())keyword_countarticle.lower().count(keyword.lower())densitykeyword_count/word_count*100ifdensity3:issues.append(f关键词密度过高 ({density:.1f}%)建议2%)elifdensity0.5:issues.append(f关键词密度过低 ({density:.1f}%)建议1%)# 4. 检查内部链接机会internal_link_suggestionsself._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.pyimportasynciofromtypingimportDictclassSEOOrchestrator:def__init__(self):self.agents{researcher:ResearchAgent(...),writer:WriterAgent(...),tech_auditor:TechAuditorAgent(...),publisher:PublisherAgent(...)}self.task_queueasyncio.Queue()self.results{}asyncdefsubmit_task(self,task:Dict):提交新任务awaitself.task_queue.put(task)asyncdefrun_pipeline(self,task:Dict):运行完整流水线task_idtask[id]try:# Step 1: 研究print(f[{task_id}] Starting research...)research_resultawaitself.agents[researcher].research(task)self.results[f{task_id}_research]research_result# Step 2: 写作print(f[{task_id}] Starting writing...)writer_resultawaitself.agents[writer].write(research_result)self.results[f{task_id}_writer]writer_result# Step 3: 技术审核print(f[{task_id}] Starting tech audit...)audit_resultawaitself.agents[tech_auditor].audit(writer_result,research_result)self.results[f{task_id}_audit]audit_result# Step 4: 如果审核通过发布否则返回修改ifaudit_result[output][approved]:print(f[{task_id}] Publishing...)publish_resultawaitself.agents[publisher].publish(writer_result,audit_result)self.results[f{task_id}_publish]publish_resultreturnpublish_resultelse:print(f[{task_id}] Audit failed, returning for revision)return{status:needs_revision,audit_issues:audit_result[output][issues],draft:writer_result[output][article]}exceptExceptionase:return{status:error,error:str(e),task_id:task_id}asyncdefworker(self):持续处理队列中的任务whileTrue:taskawaitself.task_queue.get()resultawaitself.run_pipeline(task)print(fTask{task[id]}completed:{result[status]})self.task_queue.task_done()五、使用示例asyncdefmain():orchestratorSEOOrchestrator()# 提交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]fori,kwinenumerate(keywords):awaitorchestrator.submit_task({id:ftask_{i1},keyword:kw,priority:normal})# 启动3个worker并行处理workers[asyncio.create_task(orchestrator.worker())for_inrange(3)]# 等待所有任务完成awaitorchestrator.task_queue.join()# 取消workerforwinworkers:w.cancel()asyncio.run(main())六、实战效果多Agent系统跑了2周的测试指标单Agent多Agent文章通过率45%78%平均修改轮数3.21.4研究质量评分3.2/54.1/5技术合规率52%89%总耗时/篇25分钟18分钟多Agent的优势每个Agent专职输出质量更高审核Agent拦截问题减少返工并行处理整体效率更高七、总结多Agent SEO系统的核心价值专业化研究、写作、审核各司其职可扩展加新Agent如图片生成Agent不影响现有流程可调试哪个环节出问题一目了然可复用Agent可以独立复用到其他项目这套系统适合需要规模化内容生产的团队。单篇文章用单Agent就够了但批量生产时多Agent优势明显。多Agent系统的最大挑战是通信开销——Agent之间传递的信息格式要统一。建议定义好标准的Task和Result schema用Pydantic模型约束。否则AgentA的输出AgentB读不懂整个系统就崩了。
