Agent框架三范式决战：从命令式到协议原生

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

class AgentState(TypedDict):
    messages: list
    next_action: str
    retry_count: int

def router(state: AgentState) -> str:
    if state["retry_count"] > 3:
        return "escalate"
    if state["next_action"] == "search":
        return "search_agent"
    return "respond"

graph = StateGraph(AgentState)
graph.add_node("planner", plan_step)
graph.add_node("search_agent", search_step)
graph.add_node("respond", respond_step)
graph.add_node("escalate", escalate_step)

graph.add_edge(START, "planner")
graph.add_conditional_edges("planner", router)
graph.add_edge("search_agent", "planner")  # 循环回到规划
graph.add_edge("respond", END)
graph.add_edge("escalate", END)

app = graph.compile(
    checkpointer=SqliteSaver("agent.db"),
    interrupt_before=["escalate"],  # 人机审批门控
)

from microsoft.agent_framework import Agent, AgentRuntime
from microsoft.agent_framework.teams import Team, RoundRobin

researcher = Agent(
    name="researcher",
    instructions="You research topics thoroughly...",
    tools=[search_tool, scrape_tool],
)

writer = Agent(
    name="writer", 
    instructions="You write clear summaries...",
)

# AutoGen风格的对话式多Agent
team = Team(
    agents=[researcher, writer],
    mode=RoundRobin(max_turns=6),
)

# Semantic Kernel风格的DAG编排
from microsoft.agent_framework.workflows import WorkflowBuilder
workflow = (
    WorkflowBuilder()
    .add_step("research", researcher, output_key="findings")
    .add_step("write", writer, input_key="findings", output_key="draft")
    .add_step("review", reviewer, input_key="draft")
    .build()
)

runtime = AgentRuntime()
result = await runtime.run(workflow, input="Analyze Q1 market trends")

from google.adk import Agent, Runner
from google.adk.a2a import A2ASkill, A2ACard

class MarketAnalyst(Agent):
    name = "market_analyst"
    description = "Analyzes market trends and generates reports"

    skills = [
        A2ASkill(
            id="analyze-trends",
            name="Market Trend Analysis",
            description="Analyze market data for trends",
            input_schema={"type": "object", "properties": {
                "sector": {"type": "string"},
                "timeframe": {"type": "string"}
            }}
        )
    ]

    async def handle_task(self, task):
        # 处理来自远程Agent的A2A请求
        data = await self.fetch_market_data(task.params)
        analysis = await self.llm.analyze(data)
        return analysis

# 发布A2A Agent Card供发现
card = A2ACard(
    agent=MarketAnalyst(),
    endpoint="https://my-agent.run.app/a2a",
    capabilities=["streaming", "long-running"],
)
card.publish()  # 注册到A2A发现服务

import { Agent, Workflow, step } from "@mastra/core";
import { z } from "zod";

const researcher = new Agent({
  name: "researcher",
  instructions: "Research topics deeply...",
  model: openai("gpt-4.1"),
});

const writer = new Agent({
  name: "writer",
  instructions: "Write compelling summaries...",
  model: openai("gpt-4.1"),
});

// 声明式工作流 - 像定义Docker Compose一样定义Agent拓扑
const researchWorkflow = new Workflow({
  name: "research-pipeline",
  triggerSchema: z.object({ topic: z.string() }),
})
  .step(step({
    id: "research",
    execute: async ({ context }) => {
      return researcher.generate(`Research: ${context.triggerData.topic}`);
    },
  }))
  .step(step({
    id: "write",
    execute: async ({ context }) => {
      return writer.generate(`Summarize: ${context.researchStep.result}`);
    },
  }))
  .after("research")
  .run("write");

// Mastra 1.0的Agent Network - 自动发现和编排
const network = new AgentNetwork({
  agents: [researcher, writer, reviewer],
  model: openai("gpt-4.1"), // 路由用LLM
});

from agents import Agent, Runner, function_tool, handoff

@function_tool
def search_web(query: str) -> str:
    """Search the web for information."""
    return web_search(query)

@function_tool 
def analyze_data(data_path: str) -> str:
    """Analyze a data file."""
    return data_analyzer.run(data_path)

researcher = Agent(
    name="Researcher",
    instructions="You research topics using web search.",
    tools=[search_web],
)

analyst = Agent(
    name="Analyst", 
    instructions="You analyze data and provide insights.",
    tools=[analyze_data],
)

coordinator = Agent(
    name="Coordinator",
    instructions="Route tasks to specialist agents.",
    handoffs=[
        handoff(researcher, tool_name_override="delegate_research"),
        handoff(analyst, tool_name_override="delegate_analysis"),
    ],
)

# Runner是核心 - 执行Agent Loop直到完成
result = await Runner.run(
    coordinator,
    input="Research AI trends and analyze the market data",
    max_turns=20,
    context=RunContextWrapper(user_id="user123"),
)

# 2026年4月更新：Sandbox Execution
result = await Runner.run(
    coordinator,
    input="Run this analysis code safely",
    sandbox=SandboxConfig(
        allowed_commands=["python", "pip"],
        memory_limit="512MB",
        timeout=60,
    ),
)

from crewai import Agent, Task, Crew, Process, Flow
from crewai.tools import RagTool

# 用自然语言定义Agent角色
researcher = Agent(
    role="Senior Market Researcher",
    goal="Uncover cutting-edge developments in {topic}",
    backstory="""You are a seasoned researcher at a top think tank.
    Known for your ability to find the most relevant information.""",
    tools=[search_tool, scrape_tool],
    llm="gpt-4.1",
    memory=True,  # CrewAI内置记忆
)

writer = Agent(
    role="Tech Content Strategist",
    goal="Transform complex findings into compelling narratives",
    backstory="You are a famous tech writer with 1M followers.",
)

# 定义任务流程
research_task = Task(
    description="Research the latest in {topic}",
    expected_output="A detailed report with 5 sections",
    agent=researcher,
)

write_task = Task(
    description="Write an article based on the research",
    expected_output="A 2000-word article in markdown",
    agent=writer,
    context=[research_task],  # 显式依赖
)

# CrewAI v1.14的Flow - 声明式工作流
from crewai.flow import Flow, listen, start, router

class ResearchFlow(Flow):
    @start()
    def begin_research(self):
        return {"topic": self.inputs["topic"]}

    @listen("begin_research")
    @router()
    def route_topic(self, inputs):
        if "tech" in inputs["topic"]:
            return "deep_tech"
        return "general"

    @listen("deep_tech")
    def deep_tech_research(self, inputs):
        return researcher.execute(inputs)

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

result = crew.kickoff(inputs={"topic": "AI agent frameworks 2026"})

graph TB
    subgraph "范式一：命令式 Imperative"
        A1[OpenAI Agents SDK] --> B1[Agent Loop]
        A1 --> C1[Handoff转交]
        B1 --> D1[函数调用链]
        C1 --> D1
        D1 --> E1["控制力 ★★★★★<br/>可靠性 ★★★<br/>互操作性 ★★"]
    end

    subgraph "范式二：声明式 Declarative"
        A2[LangGraph Pregel] --> B2[图编译]
        A3[Mastra Workflow] --> B3[DAG描述]
        A4[CrewAI Flow] --> B4[角色+Flow]
        B2 --> D2[超级步执行]
        B3 --> D2
        B4 --> D2
        D2 --> E2["控制力 ★★★<br/>可靠性 ★★★★★<br/>互操作性 ★★★"]
    end

    subgraph "范式三：协议原生 Protocol-Native"
        A5[Google ADK] --> B5[A2A内置]
        A6[Microsoft MAF 1.0] --> B6[ACP适配]
        B5 --> D3[服务发现+RPC]
        B6 --> D3
        D3 --> E3["控制力 ★★<br/>可靠性 ★★★★<br/>互操作性 ★★★★★"]
    end

    E1 -.->|"2026趋势：<br/>收敛为混合范式"| F[混合架构]
    E2 -.-> F
    E3 -.-> F

flowchart LR
    START[开始选型] --> Q1{任务复杂度?}
    Q1 -->|"单步/简单链"| Q2{需要OpenAI绑定?}
    Q1 -->|"多步/有条件分支"| Q3{团队语言?}
    Q1 -->|"多系统/跨组织"| Q4{协议要求?}

    Q2 -->|"是"| S1[OpenAI Agents SDK]
    Q2 -->|"否/需多模型"| S2[PydanticAI]

    Q3 -->|"Python"| Q5{需要图可视化调试?}
    Q3 -->|"TypeScript"| S3[Mastra]
    Q3 -->|"C#/企业"| S4[Microsoft MAF 1.0]

    Q5 -->|"是"| S5[LangGraph]
    Q5 -->|"否/需快速原型"| S6[CrewAI]

    Q4 -->|"A2A"| S7[Google ADK]
    Q4 -->|"ACP/MCP"| S4
    Q4 -->|"混合"| S8["LangGraph +<br/>协议适配层"]