graph TB
    subgraph 协议层
        MCP[MCP<br/>Anthropic · 工具发现]
        A2A[A2A<br/>Google · Agent通信]
        ACP[ACP<br/>IBM · 协作协议]
    end

    subgraph 编排层
        CrewAI[CrewAI 45.9K⭐<br/>角色编排]
        AG2[AG2/AutoGen<br/>对话协作]
        PydanticAI[Pydantic AI<br/>类型安全编排]
        Smolagents[Smolagents<br/>代码优先]
    end

    subgraph 基础设施层
        LangGraph[LangGraph 34.5M/月<br/>状态图运行时]
        OpenAISDK[OpenAI Agents SDK<br/>GPT原生运行时]
        ClaudeSDK[Claude Agent SDK<br/>Claude原生运行时]
        GoogleADK[Google ADK 1.0<br/>Gemini原生运行时]
    end

    CrewAI --> LangGraph
    CrewAI --> OpenAISDK
    AG2 --> LangGraph
    AG2 --> ClaudeSDK
    PydanticAI --> LangGraph
    Smolagents --> GoogleADK

    MCP -.-> LangGraph
    MCP -.-> OpenAISDK
    A2A -.-> GoogleADK
    A2A -.-> CrewAI
    ACP -.-> AG2
    ACP -.-> PydanticAI

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

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

def router(state: AgentState) -> str:
    """基于最后一条消息决定路由到哪个Agent"""
    last = state["messages"][-1]
    if "需要分析数据" in last:
        return "data_analyst"
    elif "需要写代码" in last:
        return "coder"
    return "responder"

graph = StateGraph(AgentState)
graph.add_node("router", router)
graph.add_node("data_analyst", analyst_agent)
graph.add_node("coder", code_agent)
graph.add_node("responder", response_agent)

graph.add_edge(START, "router")
graph.add_conditional_edges("router", router,
    {"data_analyst": "data_analyst",
     "coder": "coder",
     "responder": "responder"})
graph.add_edge("data_analyst", END)
graph.add_edge("coder", END)
graph.add_edge("responder", END)

app = graph.compile(checkpointer=SqliteSaver("state.db"))

from crewai import Agent, Task, Crew, Process, LLM

llm = LLM(model="ollama/deepseek-r1", base_url="http://localhost:11434")

researcher = Agent(
    role="市场调研分析师",
    goal="分析竞品的技术差异和市场份额",
    backstory="10年科技行业分析经验",
    llm=llm,
    tools=[search_tool, scrape_tool],
    memory=True,  # v1.12 层级记忆
    verbose=True
)

writer = Agent(
    role="技术内容策略师",
    goal="将调研结果转化为可执行的营销策略",
    backstory="前头部科技媒体主编",
    llm=llm,
    memory=True
)

research_task = Task(
    description="调研2026年AI Agent框架市场格局",
    agent=researcher,
    expected_output="结构化竞品分析报告"
)

write_task = Task(
    description="基于调研撰写营销策略",
    agent=writer,
    expected_output="5条可执行营销建议"
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    memory=True
)

result = crew.kickoff()

from openai import OpenAI
from agents import Agent, Runner, function_tool

@function_tool
def lookup_customer(customer_id: str) -> str:
    """查询客户信息"""
    return f"客户 {customer_id}: VIP用户, 累计消费$12,000"

triage_agent = Agent(
    name="客服分流",
    instructions="""你是一个客服分流Agent。
    根据用户问题判断需要转接到哪个专业Agent。
    - 技术问题 → tech_support
    - 账单问题 → billing
    - 其他 → general""",
    tools=[lookup_customer],
)

tech_agent = Agent(
    name="技术支持",
    instructions="解决技术问题，必要时查阅知识库。",
)

result = await Runner.run(
    starting_agent=triage_agent,
    input="我的API调用返回403错误",
    context={"customer_id": "C-12345"}
)

from claude_agent import Agent, Tool, Guardrail

# MCP原生工具集成
search_tool = Tool.from_mcp("mcp://search/web")
db_tool = Tool.from_mcp("mcp://postgres/query")

agent = Agent(
    name="数据分析助手",
    model="claude-sonnet-4-6",
    tools=[search_tool, db_tool],
    guardrails=[
        Guardrail(
            name="数据安全",
            condition="不允许删除或修改生产数据库记录",
            action="block_and_notify"
        ),
        Guardrail(
            name="成本控制",
            condition="单次工具调用成本不超过$0.50",
            action="require_human_approval"
        )
    ]
)

result = await agent.run(
    "分析过去30天的用户留存趋势",
    require_approval_for=["db_query"]  # 关键操作需人工确认
)

from google.adk import Agent, LlmAgent, LoopAgent
from google.adk.tools import FunctionTool

def analyze_document(doc_path: str) -> dict:
    """分析文档结构和关键信息"""
    return {"pages": 42, "topics": ["AI", "Agent", "架构"]}

def generate_summary(analysis: dict) -> str:
    """基于分析结果生成摘要"""
    return f"文档共{analysis['pages']}页，覆盖{len(analysis['topics'])}个主题"

analyzer = LlmAgent(
    name="文档分析器",
    model="gemini-2.5-pro",
    tools=[FunctionTool(analyze_document)],
)

summarizer = LlmAgent(
    name="摘要生成器",
    model="gemini-2.5-flash",
    tools=[FunctionTool(generate_summary)],
)

# 循环编排：分析→生成→评估→循环直到满意
pipeline = LoopAgent(
    name="文档处理流水线",
    sub_agents=[analyzer, summarizer],
    max_iterations=3,
    exit_condition=lambda state: state.get("quality_score", 0) > 0.85
)

# A2A跨框架通信
from google.adk.a2a import A2AClient
external_agent = A2AClient("https://other-framework-agent.example.com/a2a")
result = await pipeline.run(
    input="处理这份技术报告",
    external_agents=[external_agent]  # 跨框架Agent协作
)

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel

class AnalysisResult(BaseModel):
    sentiment: float  # -1.0 到 1.0
    topics: list[str]
    confidence: float

agent = Agent(
    "openai:gpt-4o",
    result_type=AnalysisResult,  # 结构化输出，编译时类型检查
    system_prompt="你是文本分析专家。返回结构化分析结果。"
)

@agent.tool
async def fetch_user_history(ctx: RunContext[None], user_id: str) -> str:
    """获取用户历史数据——参数类型在运行时验证"""
    return f"用户{user_id}最近浏览了5篇技术文章"

result = await agent.run("分析这段用户反馈的情感倾向")
# result.data 的类型是 AnalysisResult，IDE自动补全
print(result.data.sentiment)  # 类型安全，无需手动解析

from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel

agent = CodeAgent(
    model=HfApiModel("Qwen/Qwen2.5-72B-Instruct"),
    tools=[DuckDuckGoSearchTool()],
    additional_authorized_imports=["pandas", "numpy", "matplotlib"],
    max_steps=6,
)

# Agent自动生成并执行Python代码来解决问题
result = agent.run("""
    搜索2026年最受欢迎的AI Agent框架，
    用pandas创建DataFrame，
    生成柱状图展示GitHub Stars排名
""")
# result包含Agent生成的完整Python代码和执行结果

import ag2
from ag2 import AssistantAgent, GroupChat, GroupChatManager

config_list = [{"model": "gpt-4o", "api_key": "your-key"}]

planner = AssistantAgent(
    name="规划师",
    system_message="负责拆解复杂任务为子任务",
    llm_config={"config_list": config_list}
)

coder = AssistantAgent(
    name="程序员",
    system_message="负责实现子任务的代码",
    llm_config={"config_list": config_list}
)

reviewer = AssistantAgent(
    name="审查员",
    system_message="负责代码审查和质量把关",
    llm_config={"config_list": config_list}
)

group_chat = GroupChat(
    agents=[planner, coder, reviewer],
    messages=[],
    max_round=10,
    speaker_selection_method="round_robin"  # 轮流发言策略
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config={"config_list": config_list}
)

planner.initiate_chat(
    manager,
    message="设计一个日志分析系统，能自动检测异常并生成报告"
)

graph LR
    subgraph 企业决策层
        A{选型决策}
    end

    A -->|合规优先| B[Claude Agent SDK<br/>Guardrails + MCP]
    A -->|速度优先| C[OpenAI Agents SDK<br/>零摩擦 + Handoff]
    A -->|控制优先| D[LangGraph<br/>状态图 + 检查点]
    A -->|生态优先| E[Google ADK<br/>A2A + Gemini]
    A -->|成本优先| F[CrewAI<br/>本地模型 + 角色编排]

    B --> G{需要跨框架通信?}
    C --> G
    D --> G
    E --> G
    F --> G

    G -->|是| H[MCP + A2A 双协议栈]
    G -->|否| I[单一框架闭环]

    H --> J[协议层补充]
    I --> K[框架原生足够]

    subgraph 协议补充
        MCP_L[MCP: 工具发现]
        A2A_L[A2A: Agent通信]
        ACP_L[ACP: 协作协议]
    end

    J --> MCP_L
    J --> A2A_L
    J --> ACP_L