Agent 框架对比：LangGraph, CrewAI, AutoGen

# Agent 框架的核心抽象层
from abc import ABC, abstractmethod
from typing import Any, Dict, List

class BaseAgent(ABC):
    """所有 Agent 框架的共同抽象"""

    @abstractmethod
    def execute(self, task: str, context: Dict = None) -> Any:
        pass

class BaseTool(ABC):
    """工具接口的统一抽象"""

    @property
    @abstractmethod
    def name(self) -> str:
        pass

    @abstractmethod
    def run(self, *args, **kwargs) -> Any:
        pass

class AgentFramework(ABC):
    """框架级别的统一抽象"""

    @abstractmethod
    def create_agent(self, role: str, tools: List[BaseTool]) -> BaseAgent:
        pass

    @abstractmethod
    def create_workflow(self, agents: List[BaseAgent], flow: str) -> Any:
        pass

# 各框架的设计哲学对比（伪代码）

# LangGraph 风格：图 = 节点 + 边
class LangGraphStyle:
    def build(self):
        graph = StateGraph(MyState)
        graph.add_node("research", research_node)
        graph.add_node("write", write_node)
        graph.add_edge("research", "write")  # 明确的控制流
        graph.add_conditional_edges("write", should_continue)
        return graph.compile()

# CrewAI 风格：角色 + 任务 + 流程
class CrewAIStyle:
    def build(self):
        crew = Crew(
            agents=[researcher, writer],
            tasks=[research_task, write_task],
            process=Process.sequential  # 或 Process.hierarchical
        )
        return crew.kickoff()

# AutoGen 风格：对话驱动
class AutoGenStyle:
    def build(self):
        groupchat = GroupChat(agents=[user_proxy, assistant, critic], messages=[])
        manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)
        user_proxy.initiate_chat(manager, message="帮我写一份报告")

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    current_step: str
    retry_count: int

def research_node(state: AgentState) -> AgentState:
    print("执行研究节点...")
    return {"messages": ["研究完成"], "current_step": "writing"}

def writing_node(state: AgentState) -> AgentState:
    print("执行写作节点...")
    return {"messages": ["写作完成"], "current_step": "review"}

def should_retry(state: AgentState) -> str:
    if state["retry_count"] < 3:
        return "rewrite"
    return END

# 构建图
graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("write", writing_node)
graph.add_node("rewrite", writing_node)

graph.set_entry_point("research")
graph.add_edge("research", "write")
graph.add_conditional_edges("write", should_retry, {
    "rewrite": "rewrite",
    END: END,
})

app = graph.compile()
result = app.invoke({
    "messages": [], "current_step": "start", "retry_count": 0
})

# LangGraph 的检查点持久化
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

class ArticleState(TypedDict):
    topic: str
    outline: list
    draft: str
    feedback: str

def generate_outline(state: ArticleState) -> ArticleState:
    return {"outline": ["引言", "主体", "结论"]}

def generate_draft(state: ArticleState) -> ArticleState:
    return {"draft": "这是草稿内容..."}

def review_draft(state: ArticleState) -> str:
    if "好" in state.get("feedback", ""):
        return "accept"
    return "revise"

graph = StateGraph(ArticleState)
graph.add_node("outline", generate_outline)
graph.add_node("draft", generate_draft)

graph.set_entry_point("outline")
graph.add_edge("outline", "draft")
graph.add_edge("draft", END)

# 添加检查点保存器
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# 执行并自动保存状态
config = {"configurable": {"thread_id": "article-001"}}
result = app.invoke({"topic": "LangGraph教程", "outline": [], "draft": "", "feedback": ""}, config)

# 从检查点恢复
snapshot = app.get_state(config)
print(f"当前节点: {snapshot.next}")

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

# 定义角色
researcher = Agent(
    role="高级研究分析师",
    goal="深入调查 {topic} 的最新趋势和数据",
    backstory="你是一位拥有 10 年经验的研究分析师，擅长从海量信息中提取关键洞察",
    tools=[SerperDevTool()],
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role="科技内容作家",
    goal="根据研究结果撰写高质量的 {topic} 分析报告",
    backstory="你是顶尖科技媒体的资深编辑，擅长将复杂的技术概念转化为通俗易懂的文章",
    tools=[],
    verbose=True,
)

editor = Agent(
    role="内容编辑",
    goal="审查和润色报告，确保质量和一致性",
    backstory="你是拥有 15 年经验的主编，对内容质量有着极高的标准",
    tools=[],
    verbose=True,
)

# 定义任务
research_task = Task(
    description="研究 {topic} 的最新发展、市场规模、主要参与者",
    expected_output="包含关键发现的结构化研究报告",
    agent=researcher,
)

write_task = Task(
    description="基于研究报告撰写 2000 字的深度分析文章",
    expected_output="完整的分析报告，包含引言、分析和建议",
    agent=writer,
)

edit_task = Task(
    description="审查文章的事实准确性、逻辑性和可读性",
    expected_output="润色后的最终版本",
    agent=editor,
)

# 组建 Crew 并执行
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.sequential,
    verbose=2,
)

result = crew.kickoff(inputs={"topic": "AI Agent 框架"})

# CrewAI 的层级流程（Hierarchical Process）
from crewai import Agent, Task, Crew, Process

# 定义 Manager Agent（自动创建，也可以手动指定）
manager = Agent(
    role="项目经理",
    goal="协调整个项目流程，分配任务给团队成员",
    backstory="你是一位经验丰富的项目经理，擅长统筹规划和质量控制",
    allow_delegation=True,
)

analyst = Agent(
    role="数据分析师",
    goal="分析数据并生成洞察",
    backstory="你精通数据分析和可视化",
)

developer = Agent(
    role="开发工程师",
    goal="根据分析结果构建原型",
    backstory="你是全栈开发工程师，擅长快速原型开发",
)

# 使用层级流程
crew = Crew(
    agents=[manager, analyst, developer],
    tasks=[
        Task(description="分析用户数据集", expected_output="分析报告", agent=analyst),
        Task(description="构建数据可视化原型", expected_output="可运行的原型", agent=developer),
    ],
    process=Process.hierarchical,
    manager_llm=manager,  # 指定 Manager 使用的 LLM
    verbose=True,
)

# 层级模式下，Manager 会：
# 1. 接收项目目标
# 2. 将任务分配给合适的 Agent
# 3. 审查结果
# 4. 决定是否需要返工或进入下一阶段
result = crew.kickoff()

import autogen

# 配置 LLM
config_list = [
    {"model": "gpt-4", "api_key": "sk-xxx"},
    {"model": "qwen-plus", "api_key": "sk-xxx", "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1"},
]
llm_config = {"config_list": config_list, "temperature": 0.7}

# 创建 Agent
user_proxy = autogen.UserProxyAgent(
    name="User",
    human_input_mode="TERMINATE",  # 需要人工确认才终止
    max_consecutive_auto_reply=5,
    code_execution_config={"work_dir": "coding", "use_docker": False},
)

coder = autogen.AssistantAgent(
    name="Coder",
    llm_config=llm_config,
    system_message="你是一位资深 Python 工程师，擅长编写高质量代码。",
)

reviewer = autogen.AssistantAgent(
    name="Reviewer",
    llm_config=llm_config,
    system_message="你是代码审查专家。检查代码的正确性、可读性和最佳实践。指出问题但不直接修改代码。",
)

# 创建群聊
groupchat = autogen.GroupChat(
    agents=[user_proxy, coder, reviewer],
    messages=[],
    max_round=15,
    speaker_selection_method="auto",  # auto / round_robin / manual
)

manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config,
)

# 启动对话
user_proxy.initiate_chat(
    manager,
    message="请用 Python 实现一个快速排序算法，包含类型注解和单元测试。",
)

# AutoGen 2.0 新 API 风格（应用层抽象）
# 注意：需要安装 autogen-agentchat >= 0.4.0
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 创建模型客户端
model_client = OpenAIChatCompletionClient(model="gpt-4")

# 创建 Agent
planner = AssistantAgent(
    name="Planner",
    model_client=model_client,
    system_message="你是项目规划师，负责任务分解和优先级排序。",
)

executor = AssistantAgent(
    name="Executor",
    model_client=model_client,
    system_message="你是执行者，负责编写和运行代码。",
)

evaluator = AssistantAgent(
    name="Evaluator",
    model_client=model_client,
    system_message="你是评估者，负责审查结果并给出反馈。",
)

# 使用 RoundRobin 团队（轮流发言）
team = RoundRobinGroupChat(
    [planner, executor, evaluator],
    termination_condition=MaxMessageTermination(max_messages=20),
)

# 运行
async def run_team():
    stream = team.run_stream(task="构建一个数据分析 Pipeline")
    async for message in stream:
        print(message)

# 同一个任务：多步数据处理，三种框架的架构差异

# === LangGraph: 显式图结构 ===
from langgraph.graph import StateGraph

class DataState(TypedDict):
    raw_data: str
    cleaned_data: str
    analysis: str
    report: str

graph = StateGraph(DataState)
graph.add_node("fetch", fetch_node)
graph.add_node("clean", clean_node)
graph.add_node("analyze", analyze_node)
graph.add_node("report", report_node)
graph.add_edge("fetch", "clean")
graph.add_edge("clean", "analyze")
graph.add_edge("analyze", "report")
graph.set_entry_point("fetch")
app = graph.compile()
# 执行路径完全确定：fetch -> clean -> analyze -> report

# === CrewAI: 角色驱动 ===
from crewai import Agent, Task, Crew, Process

fetcher = Agent(role="数据获取专家", goal="获取原始数据")
cleaner = Agent(role="数据清洗专家", goal="清洗和标准化数据")
analyst = Agent(role="数据分析专家", goal="执行统计分析")
reporter = Agent(role="报告撰写人", goal="撰写分析报告")

crew = Crew(
    agents=[fetcher, cleaner, analyst, reporter],
    tasks=[
        Task(description="获取数据", agent=fetcher),
        Task(description="清洗数据", agent=cleaner),
        Task(description="分析数据", agent=analyst),
        Task(description="撰写报告", agent=reporter),
    ],
    process=Process.sequential,
)
# 执行路径由任务顺序决定，但每个 Agent 自主决定如何完成

# === AutoGen: 对话驱动 ===
import autogen
# Agent 自主对话决定谁做什么、怎么做
# 没有预定义的流程图
groupchat = autogen.GroupChat(
    agents=[user_proxy, data_engineer, analyst, reviewer],
    messages=[], max_round=20,
)
# 执行路径完全动态，取决于对话内容

# 三种框架的状态/上下文管理对比

# LangGraph: 全局共享状态（TypedDict）
class LGState(TypedDict):
    messages: list
    results: dict
    step_count: int
# 所有节点读写同一份状态，支持 Annotated 合并

# CrewAI: 任务上下文链式传递
# Task N 的输出 → Task N+1 的上下文
# 每个 Agent 有自己的"记忆"（对话历史）
# 通过 context 参数显式指定依赖
task2 = Task(
    description="分析数据",
    context=[task1],  # 依赖 task1 的输出
    agent=analyst,
)

# AutoGen: 消息历史
# 每个 Agent 维护自己的消息列表
# GroupChat 维护全局消息列表
# 新消息自动广播给所有参与者
# 通过 system_message 控制 Agent 行为
# 通过 function_map 注册可调用工具

# 性能对比：同一任务的 LLM 调用次数估算

# 场景：研究 + 写作 + 审查，3 步流程
# 假设每步需要 1 次 LLM 调用（不含重试）

benchmark = {
    "LangGraph": {
        "llm_calls": 3,       # 每节点 1 次调用
        "token_estimate": "约 8K",  # 精确控制，无多余调用
        "avg_latency": "~5s",       # 顺序执行，无额外开销
        "parallel_support": True,    # 支持并行节点
    },
    "CrewAI_Sequential": {
        "llm_calls": 3,       # 每任务 1 次调用
        "token_estimate": "约 9K",  # 上下文传递增加少量 token
        "avg_latency": "~6s",
        "parallel_support": False,   # Sequential 不支持并行
    },
    "CrewAI_Hierarchical": {
        "llm_calls": 5,       # 3 任务 + 2 次 Manager 调度
        "token_estimate": "约 14K",  # Manager 额外调用
        "avg_latency": "~9s",
        "parallel_support": True,    # Manager 可并行分配
    },
    "AutoGen_GroupChat": {
        "llm_calls": "8-15",  # 动态对话，难以预估
        "token_estimate": "约 20K+",  # 群聊对话消耗大
        "avg_latency": "~12s+",
        "parallel_support": False,   # 群聊本质是串行
    },
}

# LangGraph 的并行节点执行
from langgraph.graph import StateGraph
from typing import TypedDict
import asyncio

class ParallelState(TypedDict):
    query: str
    web_results: str
    db_results: str
    final_report: str

async def web_search(state: ParallelState) -> dict:
    # 模拟网络搜索（可与其他节点并行）
    await asyncio.sleep(2)
    return {"web_results": "网络搜索结果..."}

async def db_query(state: ParallelState) -> dict:
    # 模拟数据库查询（与 web_search 并行）
    await asyncio.sleep(1.5)
    return {"db_results": "数据库查询结果..."}

def combine(state: ParallelState) -> dict:
    return {"final_report": f"Web: {state['web_results']}
DB: {state['db_results']}"}

graph = StateGraph(ParallelState)
graph.add_node("search", web_search)
graph.add_node("query", db_query)
graph.add_node("combine", combine)

graph.set_entry_point("search")
# 并行边：search 和 query 同时执行
graph.add_edge("search", "combine")
graph.add_edge("query", "combine")
graph.add_edge("combine", "__end__")

app = graph.compile()
# search 和 query 会被异步并行执行

# 场景一：内容创作 Pipeline（CrewAI 最佳实践）
from crewai import Agent, Task, Crew, Process

# 角色定义——越具体越好
researcher = Agent(
    role="科技趋势研究员",
    goal="发现并分析 {topic} 领域的最新技术趋势",
    backstory="你在顶级科技媒体工作 8 年，对行业趋势有敏锐的洞察力",
    tools=[SerperDevTool()],
)

writer = Agent(
    role="深度报道撰稿人",
    goal="撰写引人入胜的 {topic} 深度报道",
    backstory="你擅长用故事化的方式讲述技术，让复杂的概念变得通俗易懂",
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[
        Task(
            description="研究 {topic} 的 3 个最新趋势，每个趋势包含：技术原理、市场影响、代表公司",
            expected_output="结构化的趋势研究报告，每个趋势 200 字",
            agent=researcher,
        ),
        Task(
            description="基于研究报告撰写一篇面向技术管理者的深度文章",
            expected_output="1500 字的深度报道，包含具体案例和行动建议",
            agent=writer,
            context=[research_task],
        ),
    ],
    process=Process.sequential,
)
result = crew.kickoff(inputs={"topic": "AI Agent 框架"})

# 场景三：数据处理流水线（LangGraph 最佳实践）
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class DataPipelineState(TypedDict):
    raw_data: str
    cleaned_data: dict
    transformed_data: dict
    validation_result: str
    errors: Annotated[list, operator.add]

def extract_node(state: DataPipelineState) -> DataPipelineState:
    """数据提取节点：从原始数据源获取"""
    raw = state["raw_data"]
    # 实际场景：读取数据库/API/文件
    return {"raw_data": raw, "errors": []}

def clean_node(state: DataPipelineState) -> DataPipelineState:
    """数据清洗节点：去重、缺失值处理、类型转换"""
    cleaned = {"records": 1000, "removed_duplicates": 23, "filled_missing": 15}
    return {"cleaned_data": cleaned}

def validate_node(state: DataPipelineState) -> str:
    """验证节点：检查数据质量"""
    quality_score = 0.95
    if quality_score > 0.9:
        return "approved"
    return "rejected"

def transform_node(state: DataPipelineState) -> DataPipelineState:
    """数据转换节点：特征工程、聚合"""
    return {"transformed_data": {"features": 50, "aggregations": 12}}

graph = StateGraph(DataPipelineState)
graph.add_node("extract", extract_node)
graph.add_node("clean", clean_node)
graph.add_node("validate", validate_node)
graph.add_node("transform", transform_node)

graph.set_entry_point("extract")
graph.add_edge("extract", "clean")
graph.add_edge("clean", "validate")
graph.add_conditional_edges("validate", validate_node, {
    "approved": "transform",
    "rejected": END,
})
graph.add_edge("transform", END)

app = graph.compile()