timeline
    title Agent框架演化时间线
    2023-Q1 : AutoGen v0.1 发布（微软研究院）
    2023-Q3 : LangGraph 早期版本 / CrewAI v0.1
    2024-Q1 : LangGraph 公测 / CrewAI 快速迭代
    2024-Q3 : AutoGen v0.4 重构 / PydanticAI 发布
    2024-Q4 : LangGraph 1.0 GA
    2025-Q1 : OpenAI Agents SDK 发布 / AG2从AutoGen分叉
    2025-Q2 : Claude Agent SDK 发布
    2025-Q3 : Google ADK 1.0 / AWS Strands 开源
    2025-Q4 : Microsoft MAF 1.0 GA
    2026-Q1 : LangGraph v1.2 / Google ADK 2.0 / CrewAI v1.12
    2026-Q2 : AWS Strands 14M+ PyPI下载 / AG2 v0.12

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

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

def researcher(state: AgentState) -> AgentState:
    # 研究节点：调用搜索工具
    result = search_tool.invoke(state["messages"][-1])
    return {"messages": [result], "next_action": "write"}

def writer(state: AgentState) -> AgentState:
    # 写作节点：基于研究结果生成内容
    draft = llm.invoke(f"基于以下研究写文章: {state['messages'][-1]}")
    return {"messages": [draft], "next_action": "review"}

def reviewer(state: AgentState) -> AgentState:
    # 审核节点：质量检查
    quality = llm.invoke(f"审核这篇文章: {state['messages'][-1]}")
    if quality.score > 0.8:
        return {"messages": [quality.feedback], "next_action": "end"}
    return {"messages": [quality.feedback], "next_action": "write"}

# 构建有状态图
graph = StateGraph(AgentState)
graph.add_node("research", researcher)
graph.add_node("write", writer)
graph.add_node("review", reviewer)

graph.add_edge(START, "research")
graph.add_conditional_edges("research", lambda s: s["next_action"], {"write": "write"})
graph.add_conditional_edges("review", lambda s: s["next_action"], {"write": "write", "end": END})
graph.add_edge("write", "review")

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

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

search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()

researcher = Agent(
    role="高级市场研究员",
    goal="发现2026年AI Agent市场的关键趋势和机会",
    backstory="你是一位有10年经验的市场研究专家，擅长从海量信息中提炼洞察",
    tools=[search_tool, scrape_tool],
    memory=True,  # 启用记忆
    verbose=True
)

writer = Agent(
    role="技术内容总监",
    goal="将研究发现转化为引人入胜的分析报告",
    backstory="你是一位擅长将复杂技术概念通俗化的资深编辑",
    verbose=True
)

research_task = Task(
    description="调研2026年AI Agent市场的十大趋势",
    agent=researcher,
    expected_output="包含数据支撑的趋势分析报告"
)

write_task = Task(
    description="基于调研结果撰写深度分析文章",
    agent=writer,
    expected_output="3000字以上的专业分析文章"
)

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

result = crew.kickoff()

import autogen  # 实际导入的是AG2

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

researcher = autogen.AssistantAgent(
    name="Researcher",
    system_message="你是一个市场研究专家",
    llm_config={"config_list": config_list}
)

critic = autogen.AssistantAgent(
    name="Critic",
    system_message="你是一个严格的评审专家，找出分析中的漏洞",
    llm_config={"config_list": config_list}
)

writer = autogen.AssistantAgent(
    name="Writer",
    system_message="你根据研究和反馈撰写最终报告",
    llm_config={"config_list": config_list}
)

groupchat = autogen.GroupChat(
    agents=[researcher, critic, writer],
    messages=[],
    max_round=12,
    speaker_selection_method="auto"  # 或 "round_robin", "random"
)

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

user_proxy = autogen.UserProxyAgent(
    name="User", human_input_mode="NEVER",
    max_consecutive_auto_reply=3
)

user_proxy.initiate_chat(manager, message="分析2026年AI Agent投资机会")

from agents import Agent, Runner, function_tool, RunContext
from agents import GuardrailFunctionOutput, input_guardrail

@function_tool
def search_market(query: str) -> str:
    """搜索市场数据"""
    return f"搜索结果: {query}相关的2026年市场数据..."

@input_guardrail
async def check_relevance(ctx: RunContext, agent: Agent, input: str) -> GuardrailFunctionOutput:
    """确保输入与AI Agent相关"""
    is_relevant = "agent" in input.lower() or "AI" in input.lower()
    return GuardrailFunctionOutput(
        output_info={"relevant": is_relevant},
        tripwire_triggered=not is_relevant  # 触发护栏则终止
    )

researcher = Agent(
    name="MarketResearcher",
    instructions="你是一个AI市场研究专家，专注于收集和分析数据。",
    tools=[search_market],
    input_guardrails=[check_relevance],
    model="gpt-4o-2025-05-08"
)

writer = Agent(
    name="ReportWriter",
    instructions="你将研究结果整理为专业报告。",
    model="gpt-4o-2025-05-08"
)

# Handoff: researcher可以将控制权交给writer
researcher_handoff = researcher.as_tool(
    name="research_market",
    description="调用研究Agent进行市场调研"
)

result = await Runner.run(
    starting_agent=researcher,
    input="分析2026年AI Agent框架市场格局"
)

// Claude Agent SDK — TypeScript原生
import { Agent, Runner, tool, hook } from "@anthropic-ai/claude-agent-sdk";
import { z } from "zod";

const searchTool = tool({
  name: "search_market",
  description: "搜索AI Agent市场数据",
  parameters: z.object({
    query: z.string().describe("搜索关键词"),
    year: z.number().default(2026)
  }),
  execute: async ({ query, year }) => {
    return `${year}年${query}相关市场数据: 增长率47%, 市场规模$8.2B`;
  }
});

// Hook: 工具调用前的审计日志
const auditHook = hook("before_tool_call", async (ctx) => {
  console.log(`[AUDIT] Agent=${ctx.agent.name} Tool=${ctx.tool.name} Args=${JSON.stringify(ctx.args)}`);
  // 可选：拒绝特定工具调用
  if (ctx.tool.name === "delete_database") {
    throw new Error("此工具调用被安全策略阻止");
  }
});

const researcher = new Agent({
  name: "MarketResearcher",
  instructions: "你是一个严谨的市场分析专家。只基于可验证的数据进行分析。",
  tools: [searchTool],
  hooks: [auditHook],
  model: "claude-sonnet-4-6"
});

const result = await Runner.run(researcher, {
  input: "分析2026年Agent SDK市场竞争格局",
  maxTurns: 10
});

from google.adk import Agent, LlmAgent, Runner
from google.adk.tools import function_tool
from google.adk.a2a import A2AServer

@function_tool
def analyze_trend(market: str, year: int = 2026) -> dict:
    """分析市场趋势"""
    return {"market": market, "year": year, "growth": "47%", "confidence": 0.89}

researcher = LlmAgent(
    name="trend_researcher",
    model="gemini-2.5-pro",
    instruction="分析AI Agent市场趋势，提供数据支撑的洞察",
    tools=[analyze_trend]
)

# A2A协议：将Agent暴露为可被其他Agent发现的服务
a2a_server = A2AServer(
    agent=researcher,
    host="0.0.0.0",
    port=8080,
    description="AI市场趋势研究Agent"
)

# 其他Agent可以通过A2A协议发现并调用
# 无论对方使用什么框架（LangGraph/CrewAI/甚至OpenAI Agents）

from strands import Agent, tool

@tool
def search_market(query: str) -> str:
    """搜索市场趋势数据"""
    return f"2026年AI Agent市场: {query} — 增长率47%, 规模$8.2B"

@tool
def generate_chart(data: str, chart_type: str = "bar") -> str:
    """生成数据可视化图表"""
    return f"已生成{chart_type}图表: {data[:50]}..."

# 一个Agent = 一个提示词 + 一组工具，模型自动编排
agent = Agent(
    system_prompt="你是一个市场分析专家。使用搜索和可视化工具完成分析任务。",
    tools=[search_market, generate_chart]
)

# 不需要定义流程，模型自行决定先搜索再生成图表
result = agent("分析2026年AI Agent框架市场并生成可视化报告")

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
from dataclasses import dataclass

class MarketReport(BaseModel):
    """严格类型化的市场报告"""
    market_name: str
    growth_rate: float
    market_size_bn: float
    top_players: list[str]
    confidence: float
    risks: list[str]

@dataclass
class Dependencies:
    """依赖注入：类型安全的外部服务"""
    api_key: str
    db_connection: str  # 可替换为真实的数据库连接类型

market_agent = Agent(
    model="openai:gpt-4o",
    result_type=MarketReport,  # 输出严格类型化！
    deps_type=Dependencies,
    system_prompt="你是一个严谨的市场分析师。输出必须包含可验证的数据。"
)

@market_agent.tool
async def fetch_market_data(ctx: RunContext[Dependencies], market: str) -> str:
    """获取市场数据 — ctx.deps自动类型推断"""
    # ctx.deps.api_key 在IDE中有自动补全和类型检查
    return f"{market}市场数据: 增长47%, 规模$8.2B"

# 运行Agent — 输出自动验证为MarketReport类型
result = await market_agent.run(
    "分析2026年AI Agent框架市场",
    deps=Dependencies(api_key="sk-xxx", db_connection="postgresql://...")
)
print(result.data.market_name)     # IDE自动补全 ✅
print(result.data.growth_rate)     # 类型: float ✅
print(result.data.top_players[0])  # 类型: str ✅

graph TB
    subgraph "Layer 3: 编排层（Orchestration）"
        LG["LangGraph<br/>有状态图"]
        CR["CrewAI<br/>角色编排"]
        AG["AG2<br/>事件驱动"]
    end

    subgraph "Layer 2: SDK层（Vendor SDK）"
        OA["OpenAI Agents SDK<br/>轻量Handoff"]
        CA["Claude Agent SDK<br/>Hooks+护栏"]
        ADK["Google ADK<br/>图+A2A"]
        ST["AWS Strands<br/>模型驱动"]
    end

    subgraph "Layer 1: 基础层（Foundation）"
        PA["PydanticAI<br/>类型安全"]
        MCP["MCP协议<br/>工具互操作"]
        A2A["A2A协议<br/>Agent互操作"]
    end

    LG --> MCP
    CR --> MCP
    AG --> MCP
    OA -.-> MCP
    CA --> MCP
    ADK --> A2A
    ADK --> MCP
    ST -.-> MCP
    PA --> MCP

    style LG fill:#4CAF50,color:#fff
    style CR fill:#FF9800,color:#fff
    style ADK fill:#2196F3,color:#fff
    style CA fill:#9C27B0,color:#fff

flowchart LR
    subgraph "LangGraph: 有状态图执行"
        A1[用户输入] --> A2[状态初始化]
        A2 --> A3[节点1: 研究]
        A3 --> A4{条件分支}
        A4 -->|质量>0.8| A5[节点2: 写作]
        A4 -->|质量≤0.8| A3
        A5 --> A6[检查点保存]
        A6 --> A7[输出]
    end

    subgraph "CrewAI: 角色编排执行"
        B1[用户输入] --> B2[创建Crew]
        B2 --> B3[Agent1: 研究员]
        B3 --> B4[Agent2: 编辑]
        B4 --> B5[Agent3: 审核员]
        B5 --> B6[输出]
    end

    subgraph "Strands: 模型驱动执行"
        C1[用户输入] --> C2[LLM决策]
        C2 --> C3[工具A]
        C3 --> C4[LLM决策]
        C4 --> C5[工具B]
        C5 --> C6[LLM决策]
        C6 --> C7[输出]
    end