从零理解 AI Agent：原理、架构与实践

2025 年被称作「AI Agent 元年」。从 LangChain 到 CrewAI，从 Function Calling 到 MCP 协议，AI Agent 正在从概念走向大规模落地。但 AI Agent 究竟是什么？它和 ChatGPT 有什么区别？本文从零开始，带你理解 AI Agent 的核心概念、技术架构，并手把手构建一个可运行的 Agent。

一、什么是 AI Agent？

1.1 一句话定义

AI Agent = 大语言模型 + 工具 + 记忆 + 规划能力

它不是一个只会聊天的机器人，而是一个能够自主完成复杂任务的智能体。Agent 可以像人类一样：理解任务、制定计划、使用工具、从错误中学习。

1.2 对比：Chatbot vs Agent

维度	Chatbot	AI Agent
交互模式	一问一答	多步骤自主执行
记忆	单次对话上下文	长期记忆 + 工作记忆
工具使用	❌ 不能	✅ 搜索、代码、API、文件
任务复杂度	简单问答	多步骤复杂任务
自主性	被动响应	主动规划、执行、反思
纠错能力	无法自我纠正	观察结果 → 调整策略

1.3 举一个具体的例子

用户需求：「帮我研究一下最近 3 个月 AI Agent 领域的融资情况，整理成报告发给我。」

• Chatbot：抱歉，我没有联网能力，无法获取实时数据。
• AI Agent：
  1. 调用搜索工具，查询「2026 AI Agent 融资」
  2. 打开 5 篇相关文章，提取关键数据
  3. 交叉验证数据准确性
  4. 发现某条数据有矛盾 → 重新搜索确认
  5. 整理成结构化报告（含表格和金额汇总）
  6. 保存为 PDF，发送给用户

这就是 Agent 与 Chatbot 的本质区别：自主决策 + 工具使用 + 多步执行。

二、AI Agent 的核心架构

一个完整的 AI Agent 由四个核心模块组成：

2.1 LLM — 大脑

Agent 的「思考」由大语言模型驱动。目前主流选型：

模型	优势	典型场景	参考成本
Claude Mythos	SWE-bench 93.9%、漏洞发现	复杂编程、安全审计	预览阶段（预测）
Claude 4 Opus	扩展思考、Agent 编码	深度推理、多步任务	$15/$75 per 1M tokens（推测）
GPT-5	原生Agent、幻觉大幅减少	通用 Agent、自主工作流	$5/$25（推测）
Gemini 2.5 Pro	1M 上下文、代码顶尖	文档分析、长文处理	$2.5/$10
DeepSeek-R1	开源推理、成本极低	数学、编程推理	¥1/¥4
Qwen 3 (通义千问)	国内最强开源、多模态	中文理解、企业场景	¥0.8/¥2
Kimi K2 (月之暗面)	超长上下文、搜索增强	长文分析、知识检索	¥1/¥4

2.2 记忆 — 海马体

Agent 需要记住上下文、用户偏好和历史经验：

记忆类型	存储内容	技术实现	典型方案
短期记忆	当前对话上下文	LLM Context Window	128K-1M tokens
长期记忆	用户偏好、历史事实	向量数据库	Milvus / Pinecone / Chroma
工作记忆	当前任务中间状态	结构化存储	JSON / SQLite

2.3 规划 — 前额叶

Agent 需要拆解复杂任务为可执行的步骤。主流规划策略：

ReAct 模式 (Reasoning + Acting) — 思考与行动交替进行：

思考：用户想知道今天天气 → 行动：调用 weather_api("北京")
观察：返回 {"temp": 25, "humidity": 60}
思考：还需要穿衣建议 → 行动：调用 fashion_advice(25)
观察：建议穿短袖
思考：可以回复了 → 输出："北京今天25°C，建议穿短袖"

Plan-and-Execute 模式 — 先制定完整计划，再逐步执行：

计划：完成天气查询任务
├── 步骤1：获取用户位置
├── 步骤2：调用天气 API
├── 步骤3：根据温度生成穿衣建议
└── 步骤4：格式化输出
→ 按计划逐步执行，每步检查结果

Reflection 模式 — 执行后自我审视，迭代改进：

行动：调用 search("2024 AI 融资")
观察：返回 5 条结果，部分数据时间不对
反思：搜索词不够精确 → 应该加限定词
行动：调用 search("2024年 AI Agent 融资 亿美元")
观察：返回 3 条精确结果，数据一致
反思：数据足够 → 可以输出
输出：整理后的融资报告

Reflection 的核心在于让 Agent 具备「元认知」——不仅做事，还能评估自己做得好不好，发现问题后自动调整策略。在编程、写作、数据分析等需要多次迭代的任务中效果显著。

2.4 工具 — Function Calling 深入解析

这是 Agent 最核心的能力。让我们看一个完整可运行的例子：

# 完整的 Function Calling 实现
import json
from openai import OpenAI

client = OpenAI(base_url="https://api.deepseek.com/v1", api_key="your-key")

# 步骤1：定义工具（JSON Schema 格式）
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学计算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '2+3*4'"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]

# 步骤2：实际的工具实现
def get_weather(city: str, unit: str = "celsius") -> dict:
    """模拟天气查询（实际应调用天气 API）"""
    weather_data = {
        "北京": {"temp": 25, "humidity": 60, "condition": "晴"},
        "上海": {"temp": 28, "humidity": 75, "condition": "多云"},
        "深圳": {"temp": 32, "humidity": 80, "condition": "阵雨"},
    }
    return weather_data.get(city, {"temp": 20, "humidity": 50, "condition": "未知"})

def calculate(expression: str) -> float:
    """安全计算数学表达式"""
    # 只允许数字和基本运算符
    import re
    if not re.match(r'^[\d\+\-\*\/\(\)\.\s]+$', expression):
        raise ValueError(f"不安全的表达式: {expression}")
    return eval(expression)

# 步骤3：Agent 循环 —— 这是 Agent 的核心
def agent_loop(user_query: str, max_steps: int = 5) -> str:
    """Agent 主循环：思考 → 行动 → 观察 → 循环"""
    messages = [
        {"role": "system", "content": "你是一个有用的助手，可以使用工具来完成任务。"},
        {"role": "user", "content": user_query}
    ]
  
    for step in range(max_steps):
        print(f"\n=== Step {step + 1} ===")
  
        # 调用 LLM
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=messages,
            tools=tools,
            temperature=0.1
        )
  
        choice = response.choices[0]
  
        # 如果有工具调用
        if choice.message.tool_calls:
            for tool_call in choice.message.tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(tool_call.function.arguments)
  
                print(f"  🔧 调用工具: {func_name}({func_args})")
  
                # 执行工具
                if func_name == "get_weather":
                    result = get_weather(**func_args)
                elif func_name == "calculate":
                    result = calculate(**func_args)
                else:
                    result = f"未知工具: {func_name}"
  
                print(f"  📊 返回结果: {result}")
  
                # 将工具结果加入对话
                messages.append(choice.message)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result, ensure_ascii=False)
                })
        else:
            # 没有工具调用，返回最终回答
            return choice.message.content
  
    return "达到最大步数限制"

# 运行
if __name__ == "__main__":
    result = agent_loop("北京今天天气怎么样？如果温度超过30度，帮我算一下体感温度（温度×1.5）")
    print(f"\n✅ Agent 最终回答: {result}")

执行流程：

=== Step 1 ===
  🔧 调用工具: get_weather({'city': '北京'})
  📊 返回结果: {'temp': 25, 'humidity': 60, 'condition': '晴'}

=== Step 2 ===
  （LLM 判断：25度不超过30，无需计算体感温度）
  
✅ Agent 最终回答: 北京今天晴天，温度25°C，湿度60%，天气不错！

三、主流 AI Agent 框架对比

3.1 LangChain — 最老牌

最成熟的 Agent 框架，生态最全。完整示例：

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 定义工具
@tool
def search(query: str) -> str:
    """搜索互联网信息"""
    return f"搜索结果：关于 '{query}' 的最新资讯..."

@tool
def calculator(expression: str) -> str:
    """执行数学运算，输入如 '2+3*4'"""
    return str(eval(expression))

# 创建 Agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [search, calculator]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手。"),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_openai_functions_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = executor.invoke({"input": "搜索最新的 AI 新闻，然后告诉我一共有几条"})
print(result["output"])

✅ 生态丰富 ⚠️ 抽象层太多，调试困难

3.2 LangGraph — 精确流程控制

LangChain 进化版，基于有向图的状态机：

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

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

def think(state: State) -> State:
    """推理节点"""
    response = llm.invoke(state["messages"])
    return {"messages": [response], "next_action": "act"}

def act(state: State) -> State:
    """行动节点 — 执行工具调用"""
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls"):
        for tc in last_msg.tool_calls:
            result = execute_tool(tc.name, tc.args)
            state["messages"].append(ToolMessage(content=str(result), tool_call_id=tc.id))
    return {"next_action": "decide"}

def decide(state: State) -> str:
    """决策：继续还是结束"""
    return "act" if state.get("need_tool") else END

graph = StateGraph(State)
graph.add_node("think", think)
graph.add_node("act", act)
graph.add_conditional_edges("think", decide)
graph.set_entry_point("think")
app = graph.compile()

✅ 可视化流程 ✅ 精确控制每条边 ⚠️ 学习曲线陡

3.3 CrewAI — 多 Agent 协作

模拟团队协作，每个 Agent 各司其职：

from crewai import Agent, Task, Crew, Process

# 研究员 Agent
researcher = Agent(
    role="资深研究员",
    goal="收集并分析最新的 {topic} 相关信息",
    backstory="你是一位拥有10年经验的技术研究员",
    tools=[search_tool],
    verbose=True
)

# 撰稿人 Agent
writer = Agent(
    role="技术撰稿人",
    goal="将研究结果整理成清晰易懂的报告",
    backstory="你擅长将复杂的技术概念转化为通俗文章",
    verbose=True
)

# 定义任务
research_task = Task(
    description="研究 {topic} 的最新进展，至少找3个关键发现",
    agent=researcher,
    expected_output="结构化的研究发现，包含数据和来源"
)
write_task = Task(
    description="基于研究结果撰写一份技术报告",
    agent=writer,
    expected_output="一份markdown格式的技术报告"
)

# 组建团队并执行
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff(inputs={"topic": "AI Agent 框架对比"})
print(result)

✅ 开箱即用 ✅ 角色扮演清晰 ⚠️ 灵活度受限

3.4 AutoGen (Microsoft)

微软出品，专注对话式协作：

from autogen import AssistantAgent, UserProxyAgent

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

assistant = AssistantAgent(
    name="助手",
    llm_config=config,
    system_message="你是一个Python专家。写代码前先解释思路。"
)

user = UserProxyAgent(
    name="用户代理",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "coding", "use_docker": False}
)

user.initiate_chat(
    assistant,
    message="写一个Python脚本，爬取HackerNews首页前10条新闻标题"
)

✅ 微软生态 ✅ 自动代码执行 ⚠️ 文档更新频繁

3.5 框架选型速查

场景	推荐框架	理由
简单单 Agent	LangChain	生态最全，文档丰富
精确流程控制	LangGraph	状态图，每条边可控
多 Agent 协作	CrewAI	角色扮演，开箱即用
对话式协作	AutoGen	自动代码执行，微软生态
完全自主	自建 + MCP	最大灵活性

四、MCP 协议 — Agent 的「USB-C 接口」

4.1 什么是 MCP？

MCP (Model Context Protocol) 是 Anthropic 提出的开放协议，定义了 LLM 与外部工具/数据源之间的标准接口。

就像 USB-C 统一了硬件接口，MCP 统一了 Agent 的工具接口。一个 MCP Server 写好，所有支持 MCP 的 Agent 都能调用。

4.2 MCP 架构

4.3 构建一个 MCP Server

# mcp_server.py — 一个最简单的 MCP Server
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
import mcp.server.stdio
import asyncio

server = Server("weather-server")

@server.list_tools()
async def handle_list_tools():
    """告诉客户端有哪些工具可用"""
    return [
        {
            "name": "get_weather",
            "description": "获取指定城市的天气",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        }
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    """实际执行工具调用"""
    if name == "get_weather":
        city = arguments["city"]
        # 这里接入真实的天气 API
        weather = {"北京": "晴 25°C", "上海": "多云 28°C"}
        return [{"type": "text", "text": weather.get(city, f"{city}: 未知")}]
    raise ValueError(f"未知工具: {name}")

async def main():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream, write_stream,
            InitializationCapabilities(
                sampling={}, roots={}
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

使用方式：在 Claude Desktop 的配置中注册这个 Server，Claude 就能自动发现并调用 get_weather 工具。

4.4 为什么 MCP 重要？

• 一次开发，到处使用：写一个 MCP Server，Claude、Cursor、自建 Agent 都能用
• 安全隔离：工具运行在独立进程，不影响主程序
• 生态爆发：GitHub 官方的 MCP Server、飞书、Notion、Docker 都在构建

五、手把手构建一个完整的 Agent

5.1 项目结构

my-agent/
├── main.py          # Agent 主循环
├── tools.py         # 工具定义
├── memory.py        # 记忆管理
├── config.py        # 配置
└── requirements.txt

5.2 记忆管理模块

# memory.py
import json
import chromadb
from chromadb.utils import embedding_functions

class MemoryManager:
    """管理 Agent 的短期和长期记忆"""
  
    def __init__(self):
        self.chroma_client = chromadb.PersistentClient(path="./agent_memory")
        self.ef = embedding_functions.DefaultEmbeddingFunction()
        self.collection = self.chroma_client.get_or_create_collection(
            name="agent_memory", embedding_function=self.ef)
  
        self.short_term = []  # 当前对话上下文
        self.work_memory = {}  # 当前任务状态
  
    def add_interaction(self, role: str, content: str):
        self.short_term.append({"role": role, "content": content})
        # 只保留最近 30 条
        if len(self.short_term) > 30:
            self.short_term = self.short_term[-30:]
  
    def save_long_term(self, key: str, value: str):
        """保存长期记忆"""
        existing = self.collection.get(ids=[key])
        if existing["ids"]:
            self.collection.update(ids=[key], documents=[value])
        else:
            self.collection.add(ids=[key], documents=[value])
  
    def recall(self, query: str, n: int = 3) -> list:
        """语义搜索长期记忆"""
        results = self.collection.query(query_texts=[query], n_results=n)
        return results.get("documents", [[]])[0]
  
    def get_context(self) -> dict:
        return {
            "short_term": self.short_term,
            "work_memory": self.work_memory
        }

5.3 完整的 Agent 主循环

# main.py
import json
from openai import OpenAI
from tools import ToolRegistry
from memory import MemoryManager
from config import LLM_CONFIG, MAX_STEPS

class AIAgent:
    """一个完整的 AI Agent 实现"""
  
    def __init__(self):
        self.client = OpenAI(**LLM_CONFIG)
        self.tools = ToolRegistry()
        self.memory = MemoryManager()
        self.step_count = 0
  
    def run(self, task: str) -> str:
        """运行 Agent 完成任务"""
        self.memory.add_interaction("user", task)
  
        while self.step_count < MAX_STEPS:
            self.step_count += 1
            print(f"\n{'='*40}\nStep {self.step_count}\n{'='*40}")
  
            # 1. 思考：调用 LLM
            response = self.client.chat.completions.create(
                model=LLM_CONFIG["model"],
                messages=self._build_messages(),
                tools=self.tools.get_definitions(),
                temperature=0.1
            )
  
            choice = response.choices[0]
            msg = choice.message
  
            # 2. 行动：执行工具或返回结果
            if msg.tool_calls:
                for tc in msg.tool_calls:
                    result = self.tools.execute(tc.function.name,
                                                json.loads(tc.function.arguments))
                    print(f"🔧 {tc.function.name} → {result}")
    
                    # 保存到工作记忆
                    self.memory.work_memory[tc.function.name] = result
                    self.memory.add_interaction("tool", str(result))
    
                    # 重要结果存入长期记忆
                    if self._is_important(result):
                        self.memory.save_long_term(
                            f"step_{self.step_count}",
                            json.dumps({"tool": tc.function.name, "result": result},
                                      ensure_ascii=False)
                        )
            else:
                # 3. 完成：返回最终结果
                self.memory.add_interaction("assistant", msg.content)
                return msg.content
  
        return "⚠️ 达到最大步数限制"
  
    def _build_messages(self) -> list:
        """构建发送给 LLM 的消息"""
        system_prompt = """你是一个自主 AI Agent。遵循 ReAct 模式：
        1. 分析用户任务
        2. 调用合适的工具
        3. 根据工具结果决定下一步
        4. 任务完成后输出总结
  
        可用的工作记忆：""" + json.dumps(self.memory.work_memory, ensure_ascii=False)
  
        messages = [{"role": "system", "content": system_prompt}]
        messages.extend(self.memory.short_term[-20:])  # 最近 20 条
        return messages
  
    def _is_important(self, result) -> bool:
        """判断结果是否需要存入长期记忆"""
        return isinstance(result, (dict, list)) and len(str(result)) > 100


# 运行
if __name__ == "__main__":
    agent = AIAgent()
  
    # 任务1：天气查询
    result = agent.run("北京今天天气怎么样？")
    print(f"\n✅ 结果: {result}")
  
    # 任务2：利用之前的记忆
    result = agent.run("再帮我查一下上海的温度，和北京对比")
    print(f"\n✅ 结果: {result}")

5.4 关键设计决策

决策点	选项	考量
模型选择	开源 vs API	成本、延迟、隐私
记忆方案	上下文 vs 向量库	任务长度、精度要求
工具数量	少而精 vs 多而全	5-8个为宜，过多导致选择困难
自主程度	全自动 vs 人工审批	高风险操作需人机协同
重试策略	无 / 线性 / 指数退避	工具调用失败时是否需要重试
并发	串行 / 并行	独立工具可并行调用提升速度

5.5 常见陷阱与对策

1. 过度工具化：给 Agent 配 50 个工具 → 它不知道该用哪个
   • ✅ 对策：按场景分组，每次只暴露 5-8 个相关工具
2. 无限循环：Agent 反复调用同一工具陷入死循环
   • ✅ 对策：设置 max_steps 上限，检测重复调用
3. 提示词过长：System Prompt 5000 字 → 吃掉大量 token
   • ✅ 对策：动态构建 prompt，只注入必要上下文
4. 忽视安全：允许 Agent 执行任意命令 → 💀
   • ✅ 对策：沙箱执行、参数白名单、人机审批

六、Agent 评测与性能优化

6.1 评测框架

评测 Agent 不能只看单轮对话质量，需要多维评估：

# 评测 Agent 的核心指标
def evaluate_agent(agent, test_cases: list[dict]) -> dict:
    """
    test_cases = [
        {"task": "...", "expected_tools": ["search"], "expected_steps": 3},
    ]
    """
    results = {"total": len(test_cases), "passed": 0, "metrics": {}}
    total_steps, total_tokens, total_time = 0, 0, 0
  
    for tc in test_cases:
        import time
        start = time.time()
        result = agent.run(tc["task"])
        elapsed = time.time() - start
  
        # 检查结果是否符合预期
        steps_ok = agent.step_count <= tc.get("expected_steps", 10)
  
        if steps_ok:
            results["passed"] += 1
  
        total_steps += agent.step_count
        total_time += elapsed
  
    results["metrics"] = {
        "accuracy": results["passed"] / results["total"],
        "avg_steps": total_steps / results["total"],
        "avg_time": total_time / results["total"],
    }
    return results

6.2 关键指标

关键指标：
├── 任务完成率：Agent 成功完成的比例（目标 > 90%）
├── 平均步数：完成任务需要的工具调用次数（越少越好）
├── Token 消耗：每次任务的成本（影响运营成本）
├── 首步准确率：第一步工具选择正确的比例
├── 错误恢复率：出错后能否自我纠正
└── 用户满意度：人工评分（1-5分）

6.3 优化方向

问题	优化手段
工具选择错误	优化工具描述、添加 few-shot 示例
步数过多	改进 Plan-and-Execute、减少无效调用
Token 消耗高	压缩 prompt、使用更小的模型
幻觉输出	强制引用来源、降低 temperature
响应慢	并行执行独立工具、使用更快的模型

七、未来趋势

7.1 2026-2028 关键方向

趋势	说明	代表项目
多 Agent 编排	数十个 Agent 协同完成端到端业务流程	LangGraph, CrewAI 2.0
物理世界 Agent	Agent 控制机器人、IoT 设备进入现实	Figure, Physical Intelligence
具身智能机器人	AI 驱动人形机器人商业化落地，工厂、家庭场景	特斯拉 Optimus, 宇树, 智元
Agent 经济网络	Agent 间自动交易、服务调用、支付结算	Agent-to-Agent Protocol, Stripe Agent SDK
全栈自主开发	Agent 从需求到部署全流程自主完成	Claude Code, Copilot Cowork
通用 Agent OS	操作系统级 Agent 平台，调度所有子 Agent	OpenAI Operator, Google Mariner
Agent 安全与治理	Agent 行为的审计、沙箱、合规框架	全球 AI 治理条约, 沙盒执行

7.2 实践案例：Hermes Agent 架构

7.3 构建 Agent 的经验教训

1. 先跑通再优化：不要一开始就追求完美架构，先让 Agent 能完成一个简单任务
2. 工具描述比模型更重要：清晰、具体的工具描述比换更好的模型效果提升更大
3. 人机协同是最好的安全网：高风险操作永远需要人类确认
4. 日志是关键：Agent 的每一步思考都要记录，否则出问题无从排查
5. 成本控制要趁早：Agent 的 token 消耗远高于单次对话，从第一天就要监控成本

总结

AI Agent 不是魔法，是 LLM + 工具 + 记忆 + 规划 的工程组合。入门路线图：

1. 第一周：理解 Function Calling，跑通一个单工具 Agent
2. 第二周：加入 ReAct 循环，让 Agent 自主多步执行
3. 第三周：引入记忆系统（短期+长期），让 Agent 「记住」历史
4. 第四周：探索 MCP 协议，构建可复用的工具生态
5. 持续：迭代优化、监控评测、扩展工具

最好的学习方式，是动手做一个。

本文所涉及的所有代码均可直接运行。只需要替换 api_key 和模型名称，即可在你的环境中体验 Agent 的完整流程。