框架生态¶

更新日期：2026-04-15

一、Agent 架构层次¶

flowchart LR
    user["User"]
    llm["LLM Core<br/>(Claude/GPT/Kimi)"]
    plan["Planning<br/>(任务拆解)"]
    mem["Memory<br/>(working/episodic/<br/>semantic/long-term)"]
    tools["Tools<br/>(MCP / function call)"]
    env["Environment<br/>(API/sandbox/file)"]

    user --> llm
    llm --> plan
    plan --> tools
    tools --> env
    env --> mem
    mem --> llm

    classDef stage fill:#fff,stroke:#cc785c,color:#1a1a1a;
    class user,llm,plan,mem,tools,env stage

5 层抽象（自顶向下）：

Application — 用户界面、对话历史
Agent Core — LLM + 规划循环（ReAct / Reflexion）
Memory — 短期 / 长期 / 情景 / 语义记忆
Tools — MCP server / function call / API
Runtime — sandbox / VM / shell

二、主流 Agent 框架对比¶

框架	开发者	Stars	定位	特点	为什么选 / 适用场景
LangChain/LangGraph	LangChain	95K+	最完整生态，社区插件最多	用图 (Graph) 定义工作流，支持条件分支、循环、人工审批等复杂 Agent 模式	需要高度自定义工作流、多步骤 Agent pipeline 时首选；学习曲线较陡
CrewAI	crewAIInc	45K+	角色化多 Agent 协作框架	每个 Agent 定义角色 (role)、目标 (goal)、工具，自动协调执行	模拟真实团队分工（研究员+写手+审核），快速搭建多 Agent 原型
OpenAI Agents SDK	OpenAI	-	OpenAI 官方 Agent 开发工具	内置 Handoff、Guardrails、Tracing，深度集成 GPT 系列模型	已绑定 OpenAI 生态、需要 GPT function calling + 内置安全栏时优先
Anthropic Agent SDK	Anthropic	-	Anthropic 官方 Agent 开发工具	原生支持 MCP、Tool Use、长上下文，Claude 模型效果最佳	使用 Claude 系列模型时的最优路径，MCP 集成最顺畅
Google ADK	Google	-	Google 官方 Agent Development Kit	与 Gemini / Vertex AI / A2A 深度集成	Google Cloud 用户、需要 A2A 多 Agent 互通时首选
AutoGen	Microsoft	35K+	研究导向的对话式多 Agent	Agent 之间通过自然语言对话协商任务，灵活度高	学术研究、需要 Agent 自由协商 / 辩论式推理的场景
LlamaIndex	LlamaIndex	38K+	数据检索 + Agent 一体化	内置丰富的 data connector 和 index，擅长 RAG 密集型 Agent	核心需求是"从大量文档中检索并回答"时最顺手
Smolagents	HuggingFace	-	极轻量级 Agent 库	几行代码即可创建 Agent，支持多模型后端	简单工具调用、PoC 验证、不想引入重框架时使用

框架	开发者	Stars	定位	特点	为什么选 / 适用场景
PydanticAI	Pydantic	-	类型安全优先的 Agent 框架	用 Pydantic 模型严格定义输入输出，编译期类型检查	工程规范要求高、需要 schema 验证和 IDE 自动补全的生产项目
Haystack	deepset	17K+	企业级 RAG + Agent 框架	Pipeline 式设计，组件可插拔，适合生产部署	企业级部署、需要可观测性和组件化运维时选用
LlamaBot	-	-	极简 Python-native Agent	纯 Python 装饰器定义 Agent，零依赖	个人脚本、学习入门，不需要任何额外抽象

三、MCP (Model Context Protocol) 深入¶

3.1 概念¶

2024 年末 Anthropic 提出的开放协议，2026 年已成为 Agent 工具集成的事实标准。所有主流厂商（OpenAI、Google、Anthropic）都支持。

flowchart LR
    client["MCP Client<br/>(Claude Desktop /<br/>Cursor / VSCode)"]
    transport["Transport<br/>(stdio / HTTP+SSE)"]
    server["MCP Server<br/>(filesystem / git /<br/>postgres / brave / ...)"]
    res["资源<br/>(tools / resources /<br/>prompts)"]

    client <--> transport
    transport <--> server
    server --> res

    classDef stage fill:#fff,stroke:#cc785c,color:#1a1a1a;
    class client,transport,server,res stage

3.2 核心概念¶

概念	含义	例子
Server	暴露能力的进程	filesystem-server / git-server / db-server
Client	LLM 应用	Claude Desktop / Cursor / Cline
Transport	通信通道	stdio（本地）/ HTTP+SSE（远程）
Tools	可调用函数	`read_file(path)` / `search_db(query)`
Resources	可读取数据	文件内容 / DB 表 / API endpoint
Prompts	模板化 prompt	`code_review_prompt(file)` 返回填好的指令
Sampling	Server 反向调 LLM	server 在执行 tool 时让 LLM 帮做子决策

跟 OpenAI function calling 比，MCP 主要赢在跨厂商通用 + 资源类型更丰富（不只 tools）。

3.3 MCP Server 实现¶

# 用 Python SDK 实现一个 MCP Server
from mcp.server import Server
from mcp.types import Tool, Resource, Prompt

app = Server("my-weather-server")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_weather",
            description="Get current weather for a location",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "units": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["city"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_weather":
        city = arguments["city"]
        units = arguments.get("units", "celsius")
        # 调用真实 API
        weather = await fetch_weather_api(city, units)
        return [TextContent(type="text", text=str(weather))]

@app.list_resources()
async def list_resources():
    return [
        Resource(
            uri="weather://cities",
            name="Supported cities",
            mimeType="text/plain"
        )
    ]

@app.read_resource()
async def read_resource(uri: str):
    if uri == "weather://cities":
        return "Beijing, Shanghai, Tokyo, NYC, London, Paris, ..."

# 启动: python weather_server.py
# 配置 Claude Desktop 的 config.json 指向这个 server

3.4 MCP Client 使用¶

# LLM 应用集成 MCP
from mcp.client import Client, StdioTransport

# 连接到 server
transport = StdioTransport(command=["python", "weather_server.py"])
client = await Client.connect(transport)

# 列出可用工具
tools = await client.list_tools()
# [{'name': 'get_weather', 'description': '...', 'inputSchema': {...}}]

# 将工具转换为 LLM API 格式
llm_tools = [{"type": "function", "function": t} for t in tools]

# LLM 调用
response = llm.chat(messages, tools=llm_tools)

# 如果 LLM 返回工具调用
if response.tool_calls:
    for call in response.tool_calls:
        result = await client.call_tool(call.name, call.arguments)
        messages.append({"role": "tool", "content": result})
    # 继续 LLM 对话...

3.5 MCP 生态¶

Server	维护方	用途
`filesystem`	官方	读写本地文件
`git`	官方	git status / diff / commit 等
`postgres`	官方	查询 DB schema + 执行 SELECT
`brave-search`	官方	Web search via Brave API
`slack`	官方	发消息 / 读 channel
`puppeteer`	官方	浏览器自动化
`obsidian`	社区	操作 Obsidian vault
`linear`	社区	Linear issue 管理
`kubernetes`	社区	k8s 操作
自家业务	自建	内部 API / DB

截至 2026.04 公开 MCP server 已超 200+ 个。Anthropic 维护官方索引。

四、A2A (Agent-to-Agent) 协议¶

Google 在 2025 年提出的 Agent 间通信协议。解决多 Agent 系统中 Agent 如何发现和调用彼此的问题。

4.1 核心概念¶

flowchart LR
    a1["Agent A<br/>(Researcher)"]
    a2["Agent B<br/>(Writer)"]
    a3["Agent C<br/>(Reviewer)"]
    bus["A2A bus<br/>(messaging + auth)"]
    reg["Agent Registry<br/>(discovery)"]

    a1 <--> bus
    a2 <--> bus
    a3 <--> bus
    bus --> reg

    classDef stage fill:#fff,stroke:#cc785c,color:#1a1a1a;
    class a1,a2,a3,bus,reg stage

A2A 抽象：

Agent Card：描述 agent 能力（类似 MCP server 的 tool list）
Message / Task：agent 间发起任务或对话
Auth + ACL：跨 agent 调用的鉴权（OAuth / token）
Streaming：长任务进度回传

4.2 A2A vs MCP¶

维度	MCP	A2A
通信对象	LLM ↔ tool/resource	Agent ↔ Agent
状态	session-scoped	task-scoped
用途	给单 LLM 配置工具	多 LLM 协作
主推方	Anthropic
协议	JSON-RPC over stdio/SSE	gRPC / HTTP（待核实）
生态成熟度（2026.04）	高	低（早期）
跨厂商	已	推进中

实操：单模型 + 工具用 MCP，多模型协作（如 CrewAI / LangGraph 多 agent）用 A2A 或框架内置消息总线。

五、Agent 核心能力模块¶

flowchart LR
    mem["Memory<br/>4 类"]
    plan["Planning<br/>(ReAct/ToT/Reflexion)"]
    tools["Tools<br/>(MCP/native)"]
    obs["Observation<br/>处理"]

    mem --- plan
    plan --- tools
    tools --- obs
    obs --- mem

    classDef stage fill:#fff,stroke:#cc785c,color:#1a1a1a;
    class mem,plan,tools,obs stage

5.1 记忆系统¶

class AgentMemory:
    def __init__(self):
        # 短期记忆: 当前对话上下文
        self.working_memory = []

        # 长期记忆: 向量数据库
        self.long_term = VectorDB()

        # 情景记忆: 结构化的 (task, observation, action, result)
        self.episodic = []

        # 语义记忆: 抽象知识
        self.semantic = KnowledgeGraph()

    def store(self, item):
        self.working_memory.append(item)
        # 重要事件存入长期
        if self.is_important(item):
            self.long_term.insert(item)

    def retrieve(self, query):
        # 组合多种记忆
        recent = self.working_memory[-10:]
        relevant = self.long_term.search(query, top_k=5)
        episodes = self.find_similar_episodes(query)
        return combine(recent, relevant, episodes)

5.2 规划系统¶

class AgentPlanner:
    """常见三种规划范式（参考 J3 reasoning-modes 章节）"""

    def react_loop(self, task, max_steps=10):
        """ReAct: thought → action → observation 单线"""
        history = []
        for _ in range(max_steps):
            thought = self.llm.think(task, history)
            action = self.llm.decide_action(thought)
            obs = self.execute(action)
            history.append((thought, action, obs))
            if action.is_final():
                return action.answer
        return "step limit"

    def tot(self, task, beam=3, depth=5):
        """Tree of Thoughts: BFS over thought tree"""
        frontier = [(task, [])]
        for _ in range(depth):
            cands = []
            for state, chain in frontier:
                for thought in self.llm.sample_thoughts(state, beam):
                    score = self.evaluator(state, thought)
                    cands.append((state + thought, chain + [thought], score))
            frontier = sorted(cands, key=lambda x: -x[2])[:beam]
        return max(frontier, key=lambda x: x[2])

    def reflexion(self, task, max_attempts=3):
        """失败后让模型 reflect 再重试"""
        for attempt in range(max_attempts):
            result = self.try_solve(task)
            if self.success(result):
                return result
            reflection = self.llm.reflect(task, result)
            task = self.augment_with_reflection(task, reflection)

不同 task 适用不同范式：标准任务 ReAct，复杂搜索类 ToT，需要从失败学习的 Reflexion。

5.3 工具执行¶

class ToolExecutor:
    def __init__(self, tools, sandbox=True):
        self.tools = tools
        self.sandbox = sandbox

    async def execute(self, tool_call):
        if self.sandbox:
            # 在沙箱中执行
            return await docker_exec(tool_call)
        else:
            return await direct_exec(tool_call)

    def safety_check(self, tool_call):
        # 白名单检查
        if tool_call.name not in self.allowed_tools:
            raise PermissionError()
        # 参数合理性检查
        if not validate_schema(tool_call.arguments):
            raise ValueError()

六、框架选型决策¶

flowchart TB
    start["要做 Agent 系统"]
    q1{"主要场景"}

    start --> q1

    q1 -->|"快速 PoC / 个人脚本"| smol["Smolagents / Pydantic AI<br/>极轻量"]
    q1 -->|"标准 Agent 工作流"| oa["OpenAI/Anthropic SDK<br/>一份代码绑模型"]
    q1 -->|"复杂条件分支 / DAG"| lg["LangGraph<br/>(图式定义)"]
    q1 -->|"角色化多 Agent"| crew["CrewAI<br/>研究员+写手+审核"]
    q1 -->|"研究 / 学术"| ag["AutoGen<br/>对话式协商"]
    q1 -->|"RAG-heavy"| li["LlamaIndex<br/>data connector 多"]
    q1 -->|"企业 prod"| hay["Haystack / 自家 SDK<br/>可观测性"]

    classDef stage fill:#fff,stroke:#cc785c,color:#1a1a1a;
    classDef decision fill:#f5f3eb,stroke:#bdb9ab,color:#1a1a1a;
    class start,smol,oa,lg,crew,ag,li,hay stage
    class q1 decision

生产建议: - 快速原型: OpenAI/Anthropic Agents SDK - 灵活度高: LangGraph - 多 Agent: CrewAI (开始) → LangGraph (需要更多控制时迁移)

七、Agent 的关键挑战¶

挑战	描述	缓解策略	为什么难解决
Hallucination	LLM 生成看似合理但实际虚构的事实或工具参数	RAG 注入真实文档 + 工具返回值交叉验证	LLM 本质是概率生成模型，无法保证事实一致性；Agent 场景中幻觉会被工具放大
错误循环	Agent 反复执行相同失败操作，无法跳出死循环	设置最大步数限制；检测重复 action 序列后强制终止或回退	LLM 缺乏全局状态感知，局部决策看起来"合理"但全局无进展
工具滥用	调用危险工具（删除文件、发送邮件等）造成不可逆后果	工具白名单 + 敏感操作需人工审批 (human-in-the-loop)	LLM 不具备后果评估能力，可能因 prompt 引导执行危险操作
成本失控	长时运行的 Agent 消耗大量 API token，费用远超预期	设置 token 预算上限；用小模型做规划，大模型做关键决策	Agent 的循环-反思机制天然倾向多轮调用，每轮都带完整上下文
状态丢失	复杂多步任务中上下文窗口超限，早期信息被截断	关键信息压缩/摘要存入长期记忆；分层记忆架构	当前 LLM 上下文有限（4K-1M），但 Agent 任务可能积累远超上限的历史
安全隐患	恶意用户通过 Prompt Injection 劫持 Agent 行为	输入过滤 + 工具沙箱隔离 + 输出审计	Agent 同时处理用户输入和外部数据，攻击面比纯聊天大得多

参考文献¶

[1] Yao et al. ReAct: Synergizing Reasoning and Acting. ICLR 2023. 论文
[2] Wang et al. Plan-and-Solve Prompting. ACL 2023. 论文
[3] Shinn et al. Reflexion. NeurIPS 2023. 论文
[4] Yao et al. Tree of Thoughts. NeurIPS 2023. 论文
[5] MCP Specification
[6] MCP Servers Repository
[7] A2A Protocol
[8] LangGraph Docs
[9] CrewAI Docs

↑ 上级 · H. 应用层