graph TB
    subgraph VendorSDK["原厂SDK阵营"]
        OAI["OpenAI Agents SDK<br/>轻量·Handoff原生"]
        CLA["Claude Agent SDK<br/>MCP原生·Managed Agents"]
        GOO["Google ADK 2.0<br/>图运行时·A2A原生"]
    end
    subgraph OpenFramework["开源框架阵营"]
        LG["LangGraph v1.2<br/>图式状态机"]
        CR["CrewAI v1.14.4<br/>角色编排·AMP平台"]
        MS["Microsoft Agent Framework 1.0<br/>AutoGen+SK统一"]
        PA["PydanticAI v1.70<br/>类型安全·模型无关"]
    end
    subgraph Protocols["互操作协议层"]
        MCP["MCP<br/>97M月下载"]
        A2A["A2A<br/>50+伙伴"]
        ACP["ACP<br/>IBM Research"]
    end
    OAI -.-> MCP
    CLA -.-> MCP
    GOO -.-> A2A
    LG -.-> MCP
    CR -.-> MCP
    MS -.-> MCP
    PA -.-> MCP
    GOO -.-> ACP

from agents import Agent, Runner

researcher = Agent(
    name="Researcher",
    instructions="You research topics thoroughly.",
    tools=[web_search, read_file],
)

writer = Agent(
    name="Writer", 
    instructions="You write engaging articles.",
    output_type=ArticleSchema,
)

# Handoff: researcher完成后自动交给writer
triage = Agent(
    name="Triage",
    instructions="Route to researcher then writer.",
    handoffs=[researcher, writer],
)

result = await Runner.run(triage, messages=[{"role": "user", "content": "Write about quantum computing"}])

import { Agent, MCPClient } from "@anthropic-ai/claude-agent-sdk";

const agent = new Agent({
  model: "claude-sonnet-4-6",
  tools: [
    // MCP原生工具
    await MCPClient.connect("filesystem", { root: "/workspace" }),
    await MCPClient.connect("github", { repo: "my-org/my-repo" }),
  ],
  maxTurns: 50,
  sandbox: {
    mode: "self-hosted",  // 工具执行在本地沙箱
    allowedCommands: ["git", "npm", "python"],
  },
  guardrails: {
    maxFileWritesPerTurn: 10,
    blockedPaths: ["/etc", "~/.ssh"],
  }
});

const result = await agent.run("Refactor the authentication module");

from google.adk import Agent, Workflow, Graph

# 定义Agent
data_analyst = Agent(
    name="data_analyst",
    model="gemini-3.5-pro",
    instruction="Analyze data and produce insights.",
)

report_gen = Agent(
    name="report_gen", 
    model="gemini-3.5-flash",
    instruction="Generate formatted reports from analysis.",
)

# ADK 2.0图式工作流
workflow = Graph()
workflow.add_node("analyze", data_analyst)
workflow.add_node("generate", report_gen)
workflow.add_node("review", HumanApproval())  # 人工审批节点
workflow.add_edge("analyze", "generate")
workflow.add_edge("generate", "review")
workflow.add_condition("review", {
    "approved": END,
    "revise": "analyze",
})

app = Workflow(runtime="cloud-run", graph=workflow)
result = await app.run({"data": quarterly_sales})

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver

class AgentState(TypedDict):
    messages: list[BaseMessage]
    next_action: str
    research_data: dict
    revision_count: int

def researcher(state: AgentState) -> AgentState:
    # 深度研究节点
    data = research_tool(state["messages"][-1].content)
    return {"research_data": data, "next_action": "write"}

def writer(state: AgentState) -> AgentState:
    # 写作节点
    article = write_article(state["research_data"])
    return {"messages": [HumanMessage(content=article)]}

def reviewer(state: AgentState) -> AgentState:
    # 审核节点 - 可触发循环
    quality = evaluate(state["messages"][-1].content)
    if quality < 0.8 and state["revision_count"] < 3:
        return {"next_action": "research", "revision_count": state["revision_count"] + 1}
    return {"next_action": "end"}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("research", researcher)
graph.add_node("write", writer)
graph.add_node("review", reviewer)
graph.add_edge("research", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", lambda s: s["next_action"], {
    "research": "research",  # 循环回研究节点
    "end": END,
})

# PostgreSQL持久化 + 流式输出
checkpointer = PostgresSaver(conn_string)
app = graph.compile(checkpointer=checkpointer)
async for event in app.astream_events(
    {"messages": [HumanMessage(content="Deep dive into quantum error correction")]},
    config={"configurable": {"thread_id": "research-001"}}
):
    print(event)  # 实时流式事件

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Senior Research Analyst",
    goal="Uncover cutting-edge developments in AI",
    backstory="You're a veteran analyst at a top tech think tank.",
    tools=[search_tool, scrape_tool],
    verbose=True,
)

writer = Agent(
    role="Tech Content Strategist",
    goal="Craft compelling content on AI advancements",
    backstory="You're a former tech journalist who knows how to engage audiences.",
    verbose=True,
)

research_task = Task(
    description="Analyze the latest breakthroughs in {topic}",
    expected_output="A detailed analysis report with key findings",
    agent=researcher,
)

write_task = Task(
    description="Write an engaging article based on the research findings",
    expected_output="A polished 2000-word article with citations",
    agent=writer,
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,  # 顺序执行
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "quantum computing in 2026"})

from microsoft.agentframework import Agent, AgentGroupChat, ProcessBuilder

# AutoGen风格：事件驱动多Agent对话
analyst = Agent(name="Analyst", instructions="Analyze financial data")
advisor = Agent(name="Advisor", instructions="Provide investment advice")
reviewer = Agent(name="Reviewer", instructions="Review advice for compliance")

group_chat = AgentGroupChat(
    agents=[analyst, advisor, reviewer],
    termination_condition=MaxMessages(10),
    selection_strategy="round_robin",
)

# Semantic Kernel风格：结构化进程编排
process = ProcessBuilder()
process.add_step("collect_data", analyst)
process.add_step("generate_advice", advisor)
process.add_step("compliance_review", reviewer, 
    approval_required=True)  # 需要人工审批
process.add_fork("approved", [
    ("notify_client", notification_agent),
    ("update_portfolio", portfolio_agent),
])

# 两种范式可以混用
result = await process.execute(
    input={"query": "Q1 portfolio analysis"},
    fallback=group_chat,  # 进程失败时回退到自由对话
)

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel

class ResearchOutput(BaseModel):
    findings: list[str]
    confidence: float
    sources: list[str]

research_agent = Agent(
    "openai:gpt-4o",  # 模型无关：可以换成 anthropic:claude-sonnet-4-6
    output_type=ResearchOutput,
    system_prompt="You are a thorough research analyst.",
)

@research_agent.tool
async def search_web(ctx: RunContext[None], query: str) -> str:
    """Search the web for information."""  # 自动成为工具描述
    return await web_search(query)

@research_agent.tool
async def read_paper(ctx: RunContext[None], url: str) -> str:
    """Read and summarize an academic paper."""
    return await paper_reader(url)

# 类型安全的结果
result = await research_agent.run("Analyze transformer architecture variants")
print(result.output.findings)  # IDE知道这是list[str]
print(result.output.confidence)  # IDE知道这是float

graph LR
    subgraph Philosophy["架构哲学光谱"]
        direction LR
        A["声明式<br/>CrewAI<br/>'告诉AI做什么'"] --> B["图式<br/>LangGraph/ADK 2.0<br/>'画出执行流程'"]
        B --> C["事件驱动<br/>Microsoft AF<br/="'响应消息流'"]
        C --> D["类型驱动<br/>PydanticAI<br/>'定义输入输出'"]
        D --> E["轻量原厂<br/>OpenAI/Anthropic<br/>'最少概念'"]
    end

graph TB
    subgraph ProtocolStack["协议栈分层"]
        TOOLS["工具接入层<br/>MCP — 97M下载/月<br/>事实标准"]
        A2AL["Agent发现层<br/>A2A — 50+伙伴<br/>跨组织协作"]
        COMMS["Agent通信层<br/>ACP — IBM Research<br/>低延迟消息"]
    end
    TOOLS --> A2AL --> COMMS