基于LangGraph的GEO多Agent协同编排系统

                    ┌─────────────────────────┐
                    │   用户查询: "2024年AI  │
                    │   营销趋势分析"          │
                    └────────────┬────────────┘
                                 │
                    ┌────────────▼────────────┐
                    │     Orchestrator        │
                    │   (LangGraph StateGraph)│
                    └────────────┬────────────┘
                                 │
          ┌──────────────────────┼──────────────────────┐
          │                      │                      │
    ┌─────▼─────┐         ┌─────▼─────┐         ┌─────▼─────┐
    │Search     │         │Analysis   │         │Writing    │
    │Agent      │────────│Agent      │────────│Agent      │
    │(联网搜索)  │         │(信息提炼)  │         │(内容生成)  │
    └───────────┘         └───────────┘         └─────┬─────┘
                                                       │
                    ┌──────────────────────────────────┘
                    │
          ┌─────────▼─────────┐
          │   Review Agent    │
          │   (质量审核)       │
          └─────────┬─────────┘
                    │
          ┌─────────▼─────────────────────────┐
          │         Publish Agent              │
          │   (多平台分发: 公众号/知乎/官网)     │
          └───────────────────────────────────┘

"""
geo_multi_agent.py - 基于LangGraph的GEO多Agent协同编排系统
技术栈: Python / langgraph / langchain / pydantic / asyncio / typing
场景: 多Agent协同完成GEO内容生成与分发全流程
"""
import os
import json
import time
import asyncio
import hashlib
import logging
from typing import List, Dict, Optional, Any, Literal, TypedDict, Annotated
from datetime import datetime
from enum import Enum
from dataclasses import dataclass, field
from pydantic import BaseModel, Field

# LangGraph
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

# LangChain
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser

# 模拟LLM（实际使用时替换为真实模型）
class MockLLM: 2014.baike.tongsou.com 
    """模拟LLM，用于演示"""

    def __init__(self, model_name: str = "mock-gpt-4"):
        self.model_name = model_name
        self.call_count = 0

    def invoke(self, messages: list, **kwargs) -> AIMessage:
        self.call_count += 1
        last_msg = messages[-1] if messages else None
        content = last_msg.content if last_msg else ""

        # 根据内容关键词返回模拟响应
        if "搜索" in content or "search" in content.lower():
            response = self._mock_search_response(content)
        elif "分析" in content or "analyze" in content.lower():
            response = self._mock_analysis_response(content)
        elif "写作" in content or "write" in content.lower():
            response = self._mock_writing_response(content)
        elif "审核" in content or "review" in content.lower():
            response = self._mock_review_response(content)
        elif "发布" in content or "publish" in content.lower():
            response = self._mock_publish_response(content)
        else:
            response = f"[MockLLM] Received: {content[:100]}..."

        return AIMessage(content=response)

    def _mock_search_response(self, query: str) -> str: 2013.baike.tongsou.com 
        return json.dumps({
            "query": query,
            "results": [
                {
                    "title": f"2024年AI营销趋势报告 - 来源{i+1}",
                    "url": f"https://example.com/ai-marketing-2024-{i+1}",
                    "snippet": f"2024年AI营销领域呈现{i+1}大趋势：个性化推荐、自动化内容生成、智能客服...",
                    "source": f"媒体{i+1}",
                    "published_date": "2024-01-15",
                }
                for i in range(5)
            ],
            "total_results": 5,
        }, ensure_ascii=False)

    def _mock_analysis_response(self, content: str) -> str:
        return json.dumps({
            "key_points": [
                "AI内容生成效率提升300%",
                "个性化营销成为主流",
                "智能客服覆盖率达85%",
                "营销自动化平台竞争加剧",
                "数据隐私合规要求提升",
            ],
            "sentiment": "positive",
            "confidence": 0.92,
            "categories": ["AI营销", "趋势分析", "行业报告"],
            "summary": "2024年AI营销领域呈现高速增长态势，内容生成、个性化推荐、智能客服三大方向表现突出。",
        }, ensure_ascii=False)

    def _mock_writing_response(self, content: str) -> str: 2012.baike.tongsou.com 
        return """# 2024年AI营销趋势深度分析

## 引言

随着大语言模型技术的成熟，AI营销正在经历前所未有的变革。本文基于最新行业数据，深度解析2024年AI营销的五大核心趋势。

## 趋势一：AI内容生成效率革命

根据最新研究，AI内容生成工具已将营销团队的内容生产效率提升**300%**。从社交媒体文案到长篇文章，从视频脚本到广告创意，AI正在成为营销人的得力助手。

> "AI不是取代营销人，而是让营销人从重复劳动中解放出来，专注于更有价值的创意和策略工作。"

## 趋势二：个性化营销成为标配

85%的消费者期望品牌提供个性化体验。AI通过分析用户行为数据，能够实现：
- **千人千面**的内容推荐
- **实时动态**的价格策略
- **精准触达**的广告投放

## 趋势三：智能客服全面普及

智能客服覆盖率已达85%，平均响应时间缩短至3秒以内，客户满意度提升40%。

## 趋势四：营销自动化平台竞争白热化

HubSpot、Marketo、国内的神策、Convertlab等玩家纷纷加码AI能力，平台间的差异化竞争日益激烈。

## 趋势五：数据隐私合规成为底线

随着《个人信息保护法》等法规的实施，合规使用数据、保护用户隐私成为AI营销的底线要求。

## 结语

2024年是AI营销的爆发之年。拥抱AI，但不盲目追随；善用数据，但坚守合规底线——这才是营销人的正确姿态。

---
*本文基于公开行业报告整理，仅供参考。*
"""

    def _mock_review_response(self, content: str) -> str:
        return json.dumps({
            "passed": True,
            "score": 92,
            "issues": [],
            "suggestions": [
                "可以补充更多具体案例数据",
                "建议增加图表可视化",
            ],
            "compliance_check": {
                "sensitive_words": False,
                "copyright_risk": False,
                "factual_accuracy": True,
            },
            "readability_score": 85,
        }, ensure_ascii=False)

    def _mock_publish_response(self, content: str) -> str:
        return json.dumps({
            "platforms": [
                {
                    "name": "微信公众号",
                    "status": "published",
                    "url": "https://mp.weixin.qq.com/s/xxx",
                    "published_at": "2024-01-20T10:30:00",
                },
                {
                    "name": "知乎专栏",
                    "status": "published",
                    "url": "https://zhuanlan.zhihu.com/p/xxx",
                    "published_at": "2024-01-20T10:31:00",
                },
                {
                    "name": "企业官网博客",
                    "status": "published",
                    "url": "https://blog.example.com/ai-marketing-2024",
                    "published_at": "2024-01-20T10:32:00",
                },
            ],
            "total_published": 3,
            "failed": 0,
        }, ensure_ascii=False)


# ==================== 模块1: 状态定义 ====================

class AgentStatus(Enum): 2011.baike.tongsou.com 
    """Agent执行状态"""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class AgentResult: 2010.baike.tongsou.com 
    """单个Agent的执行结果"""
    agent_name: str = ""
    status: AgentStatus = AgentStatus.PENDING
    output: Any = None
    error: str = ""
    start_time: float = 0.0
    end_time: float = 0.0
    duration_ms: float = 0.0
    metadata: Dict[str, Any] = field(default_factory=dict)

    def mark_start(self): 2009.baike.tongsou.com 
        self.status = AgentStatus.RUNNING
        self.start_time = time.time(2008.baike.tongsou.com)

    def mark_complete(self, output: Any = None):
        self.status = AgentStatus.COMPLETED
        self.output = output
        self.end_time = time.time(2007.baike.tongsou.com)
        self.duration_ms = (self.end_time - self.start_time) * 1000

    def mark_failed(self, error: str): 2006.baike.tongsou.com 
        self.status = AgentStatus.FAILED
        self.error = error
        self.end_time = time.time(2005.baike.tongsou.com)
        self.duration_ms = (self.end_time - self.start_time) * 1000


class GeoAgentState(TypedDict): 2004.baike.tongsou.com 
    """
    LangGraph状态定义
    整个工作流中流转的核心数据结构
    """
    # 用户输入
    query: str
    # 消息历史（LangGraph要求）
    messages: Annotated[list, add_messages]
    # 各Agent结果
    search_result: Optional[Dict[str, Any]]
    analysis_result: Optional[Dict[str, Any]]
    writing_result: Optional[str]
    review_result: Optional[Dict[str, Any]]
    publish_result: Optional[Dict[str, Any]]
    # 执行状态追踪
    agent_results: Dict[str, AgentResult]
    # 当前执行步骤
    current_step: str
    # 是否需要人工审核
    needs_human_review: bool
    # 错误信息
    errors: List[str]


# ==================== 模块2: 搜索Agent ====================

class SearchAgent: 2002.baike.tongsou.com 
    """
    搜索Agent
    负责联网搜索、收集资料
    """

    def __init__(self, llm: MockLLM = None):
        self.llm = llm or MockLLM(2001.baike.tongsou.com)
        self.name = "search_agent"

    def run(self, state: GeoAgentState) -> GeoAgentState:
        """执行搜索"""
        result = AgentResult(agent_name=self.name)
        result.mark_start()

        try:
            query = state.get("query", "")
            messages = [
                SystemMessage(content="你是一个专业的信息搜索助手。请根据用户查询，搜索相关信息并返回结构化的搜索结果。"),
                HumanMessage(content=f"请搜索以下主题的相关信息：{query}"),
            ]

            response = self.llm.invoke(messages)
            search_data = json.loads(response.content)

            state["search_result"] = search_data
            state["messages"].append(HumanMessage(content=f"搜索完成，找到{search_data.get('total_results', 0)}条结果"))
            result.mark_complete(output=search_data)

        except Exception as e:
            result.mark_failed(str(e))
            state.setdefault("errors", []).append(f"SearchAgent failed: {e}")

        state.setdefault("agent_results", {})[self.name] = result
        state["current_step"] = "analysis"
        return state


# ==================== 模块3: 分析Agent ====================

class AnalysisAgent:
    """
    分析Agent
    负责从搜索结果中提炼关键信息、做情感分析、分类打标
    """

    def __init__(self, llm: MockLLM = None):
        self.llm = llm or MockLLM()
        self.name = "analysis_agent"

    def run(self, state: GeoAgentState) -> GeoAgentState:
        """执行分析"""
        result = AgentResult(agent_name=self.name)
        result.mark_start()

        try:
            search_result = state.get("search_result")
            if not search_result:
                raise ValueError("No search results available")

            messages = [
                SystemMessage(content="你是一个专业的内容分析师。请分析搜索结果，提炼关键要点、情感倾向、分类标签和摘要。"),
                HumanMessage(content=f"请分析以下搜索结果：{json.dumps(search_result, ensure_ascii=False)}"),
            ]

            response = self.llm.invoke(messages)
            analysis_data = json.loads(response.content)

            state["analysis_result"] = analysis_data
            state["messages"].append(HumanMessage(content="分析完成"))
            result.mark_complete(output=analysis_data)

        except Exception as e:
            result.mark_failed(str(e))
            state.setdefault("errors", []).append(f"AnalysisAgent failed: {e}")

        state.setdefault("agent_results", {})[self.name] = result
        state["current_step"] = "writing"
        return state


# ==================== 模块4: 写作Agent ====================

class WritingAgent:
    """
    写作Agent
    负责根据分析结果生成完整文章
    """

    def __init__(self, llm: MockLLM = None, style: str = "professional"):
        self.llm = llm or MockLLM()
        self.style = style
        self.name = "writing_agent"

    def run(self, state: GeoAgentState) -> GeoAgentState:
        """执行写作"""
        result = AgentResult(agent_name=self.name)
        result.mark_start()

        try:
            analysis = state.get("analysis_result")
            if not analysis:
                raise ValueError("No analysis results available")

            key_points = analysis.get("key_points", [])
            summary = analysis.get("summary", "")

            prompt = f"""
            请根据以下分析结果，撰写一篇专业的行业分析文章。

            风格：{self.style}
            核心要点：{json.dumps(key_points, ensure_ascii=False)}
            摘要：{summary}

            要求：
            1. 结构清晰，有引言、正文、结语
            2. 数据准确，引用规范
            3. 语言流畅，专业但不晦涩
            4. 字数控制在1500-2000字
            """

            messages = [
                SystemMessage(content="你是一个专业的科技内容写作者。请根据提供的分析结果撰写文章。"),
                HumanMessage(content=prompt),
            ]

            response = self.llm.invoke(messages)
            article = response.content

            state["writing_result"] = article
            state["messages"].append(HumanMessage(content="文章撰写完成"))
            result.mark_complete(output=article)

        except Exception as e:
            result.mark_failed(str(e))
            state.setdefault("errors", []).append(f"WritingAgent failed: {e}")

        state.setdefault("agent_results", {})[self.name] = result
        state["current_step"] = "review"
        return state


# ==================== 模块5: 审核Agent ====================

class ReviewAgent:
    """
    审核Agent
    负责内容质量审核、合规检查
    """

    def __init__(self, llm: MockLLM = None, auto_approve_threshold: float = 90.0):
        self.llm = llm or MockLLM()
        self.auto_approve_threshold = auto_approve_threshold
        self.name = "review_agent"

    def run(self, state: GeoAgentState) -> GeoAgentState:
        """执行审核"""
        result = AgentResult(agent_name=self.name)
        result.mark_start()

        try:
            article = state.get("writing_result")
            if not article:
                raise ValueError("No article to review")

            messages = [
                SystemMessage(content="你是一个专业的内容审核员。请对文章进行质量审核，包括：内容准确性、合规性、可读性、敏感词检查。返回JSON格式的审核结果。"),
                HumanMessage(content=f"请审核以下文章：\n\n{article[:2000]}"),
            ]

            response = self.llm.invoke(messages)
            review_data = json.loads(response.content)

            state["review_result"] = review_data
            state["messages"].append(HumanMessage(content="审核完成"))

            # 判断是否需要人工审核
            score = review_data.get("score", 0)
            state["needs_human_review"] = score < self.auto_approve_threshold

            result.mark_complete(output=review_data)

        except Exception as e:
            result.mark_failed(str(e))
            state.setdefault("errors", []).append(f"ReviewAgent failed: {e}")
            state["needs_human_review"] = True

        state.setdefault("agent_results", {})[self.name] = result
        state["current_step"] = "publish"
        return state


# ==================== 模块6: 发布Agent ====================

class PublishAgent:
    """
    发布Agent
    负责将审核通过的内容分发到各平台
    """

    def __init__(self, llm: MockLLM = None, platforms: List[str] = None):
        self.llm = llm or MockLLM()
        self.platforms = platforms or ["微信公众号", "知乎专栏", "企业官网博客"]
        self.name = "publish_agent"

    def run(self, state: GeoAgentState) -> GeoAgentState:
        """执行发布"""
        result = AgentResult(agent_name=self.name)
        result.mark_start()

        try:
            # 检查审核状态
            review = state.get("review_result")
            if not review or not review.get("passed", False):
                raise ValueError("Article not approved for publishing")

            article = state.get("writing_result")
            if not article:
                raise ValueError("No article to publish")

            messages = [
                SystemMessage(content="你是一个内容发布助手。请将文章发布到指定平台，返回发布结果。"),
                HumanMessage(content=f"请将文章发布到以下平台：{json.dumps(self.platforms, ensure_ascii=False)}"),
            ]

            response = self.llm.invoke(messages)
            publish_data = json.loads(response.content)

            state["publish_result"] = publish_data
            state["messages"].append(HumanMessage(content="发布完成"))
            result.mark_complete(output=publish_data)

        except Exception as e:
            result.mark_failed(str(e))
            state.setdefault("errors", []).append(f"PublishAgent failed: {e}")

        state.setdefault("agent_results", {})[self.name] = result
        state["current_step"] = "completed"
        return state


# ==================== 模块7: LangGraph状态图编排 ====================

def build_geo_workflow() -> StateGraph:
    """
    构建GEO多Agent工作流
    使用LangGraph定义状态图
    """
    # 初始化Agent
    llm = MockLLM()
    search_agent = SearchAgent(llm)
    analysis_agent = AnalysisAgent(llm)
    writing_agent = WritingAgent(llm, style="professional")
    review_agent = ReviewAgent(llm, auto_approve_threshold=85.0)
    publish_agent = PublishAgent(llm)

    # 创建状态图
    workflow = StateGraph(GeoAgentState)

    # 添加节点
    workflow.add_node("search", search_agent.run)
    workflow.add_node("analyze", analysis_agent.run)
    workflow.add_node("write", writing_agent.run)
    workflow.add_node("review", review_agent.run)
    workflow.add_node("publish", publish_agent.run)

    # 设置入口点
    workflow.set_entry_point("search")

    # 定义边（顺序执行）
    workflow.add_edge("search", "analyze")
    workflow.add_edge("analyze", "write")
    workflow.add_edge("write", "review")

    # 审核后根据结果决定下一步
    def route_after_review(state: GeoAgentState) -> str:
        if state.get("needs_human_review", False):
            return "human_review"  # 需要人工审核
        return "publish"

    workflow.add_conditional_edges(
        "review",
        route_after_review,
        {
            "human_review": END,  # 暂停等待人工
            "publish": "publish",
        },
    )

    workflow.add_edge("publish", END)

    return workflow


# ==================== 模块8: 并行执行优化 ====================

class ParallelSearchAgent:
    """
    并行搜索Agent
    同时从多个来源搜索信息，提升效率
    """

    def __init__(self, llm: MockLLM = None):
        self.llm = ll