Agent开发核心组件进阶全总结：从理论到代码实战

引言

掌握Agent的核心组件，是区分“调用接口的程序员”与“Agent工程师”的关键分水岭。调用API只是入门，真正理解核心循环、记忆系统、RAG、工具调用和规划范式的工程细节，才能构建出稳定、可用的Agent系统。

本文把Agent开发的五大核心组件拆解为可操作的知识点，每个组件都配有完整的Python代码示例和深入分析，适合已经完成前置基础学习、准备进入框架实战的开发者。

一、Agent核心循环

1.1 什么是Agent核心循环

Agent核心循环是Agent的运行引擎。它的基本逻辑是：LLM思考当前该做什么 → 选择一个动作（通常是调用某个工具） → 获取观察结果 → 再思考下一步 → 循环，直到任务完成。

这个过程可以用ReAct范式概括：让AI边思考边行动的交互模式，让大语言模型能够像人类一样推理→行动→观察→再推理，逐步解决复杂问题。

一个经典的Agent循环包括：理解用户目标 → 任务规划 → 选择工具并执行操作 → 观察结果并修正 → 生成结果或进入下一轮行动。

1.2 核心循环的代码实现

以下是基于Anthropic API风格的ReAct循环核心代码结构：

python

import anthropic client = anthropic.Anthropic() MODEL = "claude-sonnet-4-6" SYSTEM = "你是一个有用的助手。你可以使用工具来完成任务。" def agent_loop(query: str) -> str: """ Agent核心循环：ReAct范式 思考 → 行动 → 观察 → 再思考，直到模型不再调用工具 """ messages = [{"role": "user", "content": query}] while True: # Step 1: 将消息和工具定义发给LLM response = client.messages.create( model=MODEL, system=SYSTEM, messages=messages, tools=TOOLS, # 工具定义列表（JSON Schema） max_tokens=8000, ) # Step 2: 追加助手响应到消息历史 messages.append({"role": "assistant", "content": response.content}) # Step 3: 检查退出条件 # 如果模型没有调用工具（stop_reason != "tool_use"），任务完成 if response.stop_reason != "tool_use": # 提取最终答案 for block in response.content: if hasattr(block, "text"): return block.text return "任务完成" # Step 4: 执行工具调用，收集结果 results = [] for block in response.content: if block.type == "tool_use": # 根据工具名称分发到对应的执行函数 output = execute_tool(block.name, block.input) results.append({ "type": "tool_result", "tool_use_id": block.id, "content": output, }) # Step 5: 将工具结果作为用户消息追加，回到Step 1 messages.append({"role": "user", "content": results})

1.3 退出条件的三种设计方式

Agent循环必须有明确的退出条件，否则会出现无限循环：

方式一：基于stop_reason（推荐）。当模型返回的stop_reason不等于tool_use时，说明模型认为任务已完成，没有更多工具需要调用。这是最自然的退出方式。

方式二：基于迭代次数上限。设置最大迭代次数（如10次），达到上限后强制退出并返回当前结果。这可以防止死循环消耗过多Token。

方式三：基于显式终止指令。在提示词中要求模型在任务完成时输出特定的终止标记（如[TASK_COMPLETE]），循环检测到该标记后退出。

1.4 核心循环的常见失败模式

失败模式一：无限Agent循环（IALs）。Agent可能在没有有效终止条件的情况下反复触发LLM调用或工具调用。例如：Agent反复调用同一个工具而不改变参数；多个Agent之间的交接形成了循环——Agent A交给B，B又交回给A。一项针对6549个LLM Agent仓库的研究中，IAL-Scan检测出了74个潜在问题，人工审查确认了47个项目中的68个无限循环失败，精确率达到91.9%。

失败模式二：上下文爆炸。每一轮Thought-Action-Observation都会增加上下文长度。如果工具返回的内容很长（如搜索结果、文件内容），上下文会迅速膨胀。当窗口填满时，模型开始“遗忘”早期内容，导致推理质量急剧下降。

失败模式三：工具选择错误。模型可能选择了不合适的工具，或者为工具传入了错误的参数。更糟糕的是，模型可能反复尝试同一个失败的工具调用，因为它在上下文中看到了之前的失败尝试，而这反过来“引导”它继续尝试同样的方式。

二、记忆系统

2.1 为什么Agent需要记忆系统

没有记忆的Agent，每次对话都是“失忆”的。你告诉它你的名字，下一轮对话它就忘了。你让它记住项目规范，切换会话后它就重新开始。记忆系统的核心价值在于让Agent跨轮次、跨会话保持状态和知识。

区分短期记忆与长期记忆是构建有状态Agent的基础。

2.2 短期记忆：上下文窗口管理

短期记忆依托上下文窗口与会话缓存，通过滑动窗口机制控制Token消耗。

python

class ShortTermMemory: """短期记忆：管理当前会话的上下文窗口""" def __init__(self, max_tokens: int = 8000): self.messages = [] self.max_tokens = max_tokens def add_message(self, role: str, content: str): """添加消息并检查Token预算""" self.messages.append({"role": role, "content": content}) self._truncate_if_needed() def _truncate_if_needed(self): """滑动窗口：保留最近的消息，丢弃最旧的""" while self._count_tokens() > self.max_tokens and len(self.messages) > 2: # 保留system prompt（第一条）和最近的消息 self.messages.pop(1) # 删除最早的用户消息 def _count_tokens(self) -> int: """估算Token数量（简化版）""" return sum(len(m["content"]) // 2 for m in self.messages) def get_context(self) -> list: return self.messages

但简单的滑动窗口存在一个核心问题：被截断的历史信息永久丢失。如果Agent在第1轮对话中获知了关键约束（如“用户要求用英文回答”），当这条消息被截断后，Agent就失去了这个约束。因此，需要长期记忆来弥补。

2.3 长期记忆：向量数据库持久化

长期记忆依赖向量数据库（Chroma、Milvus、Pinecone等），实现跨会话的知识持久化。核心流程是：将对话轮次提炼为具有嵌入向量的可修正事实，通过语义搜索在后续会话中检索相关记忆。

以下是基于ChromaDB的实现示例：

python

import chromadb from sentence_transformers import SentenceTransformer class LongTermMemory: """长期记忆：基于向量数据库的语义记忆""" def __init__(self, collection_name: str = "agent_memory"): self.client = chromadb.Client() self.collection = self.client.get_or_create_collection(collection_name) self.encoder = SentenceTransformer("BAAI/bge-large-zh-v1.5") def store(self, content: str, metadata: dict = None): """存储记忆：将文本向量化并存入向量库""" embedding = self.encoder.encode(content).tolist() self.collection.add( embeddings=[embedding], documents=[content], metadatas=[metadata or {}], ids=[f"mem_{self.collection.count()}"] ) def retrieve(self, query: str, top_k: int = 3) -> list: """检索记忆：根据当前对话上下文查找相关记忆""" query_embedding = self.encoder.encode(query).tolist() results = self.collection.query( query_embeddings=[query_embedding], n_results=top_k ) return results["documents"][0] if results["documents"] else []

2.4 分层记忆架构

生产级Agent系统通常采用分层记忆架构，不同层级的记忆有不同的生命周期和访问频率：

2.5 对话结束时的记忆生成

CogniWeave项目展示了一种自动记忆生成的方式：当会话结束时，系统自动生成短期和长期记忆，并合并到历史记录中。这种方式确保Agent在每次对话结束后都能从经验中学习。

python

class MemoryManager: """统一的记忆管理器""" def __init__(self): self.short_term = ShortTermMemory(max_tokens=8000) self.long_term = LongTermMemory() def add_interaction(self, user_input: str, agent_output: str): """记录一次交互""" self.short_term.add_message("user", user_input) self.short_term.add_message("assistant", agent_output) def on_session_end(self): """会话结束时：从短期记忆中提取关键信息存入长期记忆""" # 提取关键事实（可以用LLM做摘要） key_facts = extract_key_facts(self.short_term.messages) for fact in key_facts: self.long_term.store(fact, metadata={"source": "session_end"}) # 清空短期记忆 self.short_term.messages = [] def get_context_for_query(self, query: str) -> list: """为当前查询组装上下文""" # 短期记忆：最近的对话 recent = self.short_term.get_context() # 长期记忆：语义相关的历史 relevant = self.long_term.retrieve(query, top_k=3) return recent + [{"role": "system", "content": f"相关记忆：{r}"} for r in relevant]

三、RAG（检索增强生成）

3.1 为什么需要RAG

RAG是解决模型幻觉与私有知识接入的核心技术。仅依赖向量相似度检索，在实际业务中常面临：召回不准（语义相近但主题无关的文档被误召回）、排序不佳（前k个结果中关键文档位置靠后）、泛化性弱（对专有名词、缩写、长尾问题处理乏力）。

3.2 完整链路

RAG的完整链路包括六个步骤：文档解析 → 文本切分 → Embedding向量化 → 混合检索（向量+关键词）→ 重排序（Rerank）→ 引用溯源。

3.3 文档加载与分块

分块策略的核心是在语义完整性与检索粒度之间取得平衡。chunk_size=512是一个常用的起点，chunk_overlap=50可以避免上下文断裂。针对中文文档，分隔符优先级需要调整，加入中文标点。

python

from langchain.document_loaders import PyPDFDirectoryLoader from langchain.text_splitter import RecursiveCharacterTextSplitter # 加载文档 loader = PyPDFDirectoryLoader("./docs/") raw_docs = loader.load() # 中文优化的文本切分 text_splitter = RecursiveCharacterTextSplitter( chunk_size=512, chunk_overlap=50, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) chunks = text_splitter.split_documents(raw_docs) print(f"共生成 {len(chunks)} 个文本块")

3.4 混合检索（向量 + BM25）

混合检索的核心思想是：向量检索擅长语义匹配，BM25擅长关键词精确匹配，两者互补。实践中发现，混合检索架构可以通过查询扩展提升20-30%的召回率，通过重排序提升15-25%的精确率。

python

import numpy as np from rank_bm25 import BM25Okapi import jieba class HybridRetriever: """混合检索器：向量检索 + BM25关键词检索""" def __init__(self, vector_retriever, bm25_retriever, vector_weight=0.6, bm25_weight=0.4): self.vector_retriever = vector_retriever self.bm25_retriever = bm25_retriever self.vector_weight = vector_weight self.bm25_weight = bm25_weight def retrieve(self, query: str, top_k: int = 5) -> list: # 向量检索（语义匹配） vector_results = self.vector_retriever.retrieve(query, top_k=top_k * 2) # BM25检索（关键词匹配，中文需分词） tokenized_query = list(jieba.cut(query)) bm25_scores = self.bm25_retriever.get_scores(tokenized_query) bm25_top_indices = np.argsort(bm25_scores)[-top_k * 2:][::-1] bm25_results = [(self.bm25_retriever.documents[i], bm25_scores[i]) for i in bm25_top_indices] # 分数归一化后加权融合[reference:12] vector_normalized = self._normalize_scores( [r[1] for r in vector_results] ) bm25_normalized = self._normalize_scores( [r[1] for r in bm25_results] ) # 合并结果 merged = {} for (doc, _), score in zip(vector_results, vector_normalized): merged[doc] = merged.get(doc, 0) + score * self.vector_weight for (doc, _), score in zip(bm25_results, bm25_normalized): merged[doc] = merged.get(doc, 0) + score * self.bm25_weight # 按融合分数排序 sorted_results = sorted(merged.items(), key=lambda x: x[1], reverse=True) return sorted_results[:top_k] def _normalize_scores(self, scores: list) -> list: """Min-Max归一化""" if not scores: return [] min_s, max_s = min(scores), max(scores) if max_s == min_s: return [1.0] * len(scores) return [(s - min_s) / (max_s - min_s) for s in scores]

3.5 重排序（Rerank）

重排序器使用交叉编码器（Cross-Encoder）对候选集精细打分。与向量检索的“双塔”结构不同，Cross-Encoder将查询和文档拼接后一起输入模型，能够捕捉更细粒度的语义交互。

python

from sentence_transformers import CrossEncoder class Reranker: """重排序器：使用Cross-Encoder对候选文档精细打分""" def __init__(self, model_name: str = "BAAI/bge-reranker-large"): self.model = CrossEncoder(model_name) def rerank(self, query: str, documents: list, top_k: int = 3) -> list: """对候选文档进行重排序""" # 构建query-document对 pairs = [(query, doc) for doc, _ in documents] # Cross-Encoder打分 scores = self.model.predict(pairs) # 按分数排序 scored_docs = list(zip([doc for doc, _ in documents], scores)) scored_docs.sort(key=lambda x: x[1], reverse=True) return scored_docs[:top_k]

3.6 完整的RAG流程

python

class RAGPipeline: """完整的RAG流水线""" def __init__(self): self.retriever = HybridRetriever(...) self.reranker = Reranker() def query(self, question: str) -> str: # Step 1: 混合检索（高召回） candidates = self.retriever.retrieve(question, top_k=10) # Step 2: 重排序（高精度） reranked = self.reranker.rerank(question, candidates, top_k=3) # Step 3: 组装Prompt并生成 context = "\n\n".join([doc for doc, _ in reranked]) prompt = f"""基于以下资料回答问题。如果资料中没有相关信息，请明确说明。 资料： {context} 问题：{question} """ return call_llm(prompt)

四、工具调用与MCP协议

4.1 Function Calling：工具调用的基础

Function Calling让Agent具备“手”的能力，能够调用搜索API、数据库查询、代码解释器等外部工具。它是Agent的地基——所有花哨的Agent架构都建立在它之上。

核心机制：用JSON Schema描述函数的样子（名字、参数、要填哪些），模型在对话中自主决定“要不要调、调哪个、传什么参数”，返回结构化参数让你去执行。

4.2 工具定义的最佳实践

工具描述是你能控制的最大杠杆。Spell out the trigger——“当用户询问天气时使用此工具”——模型用这个来决定何时调用。

python

# 工具定义示例（OpenAI Function Calling格式） tools = [ { "type": "function", "function": { "name": "search_web", "description": "搜索网络获取最新信息。当用户询问当前事件、实时数据或你不确定的信息时使用此工具。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词，不超过100字" }, "max_results": { "type": "integer", "description": "返回结果数量上限", "default": 5 } }, "required": ["query"] } } }, { "type": "function", "function": { "name": "calculator", "description": "执行数学计算。当用户需要进行加减乘除、百分比、幂运算等数学计算时使用。", "parameters": { "type": "object", "properties": { "expression": { "type": "string", "description": "数学表达式，如 '(3+5)*2'" } }, "required": ["expression"] } } } ]

关键控制参数：

tool_choice控制模型“是否调用工具、调用哪个”的开关。可选值包括auto（默认，模型自己定）、required（强制至少调一次）、指定某个函数。

strict模式确保模型返回的参数严格符合Schema定义，不允许额外字段。

4.3 MCP协议：工具调用的标准化

MCP（Model Context Protocol）是由Anthropic推动的开放标准，为LLM应用提供标准化接口以连接和交互外部数据源和工具，现已捐赠给Linux基金会。

MCP的核心价值：你写一个Server暴露工具、资源和提示词；任何MCP Host（Claude Code、VS Code、Cursor等）都能连接它，让模型使用这些工具。

4.4 MCP Server编写示例

以下是一个完整的MCP Server示例（TypeScript SDK v2）：

typescript

import { McpServer } from '@modelcontextprotocol/server'; import { serveStdio } from '@modelcontextprotocol/server/stdio'; import * as z from 'zod/v4'; serveStdio(() => { const server = new McpServer({ name: 'weather', version: '1.0.0' }); server.registerTool( 'get-forecast', { description: 'Get the weather forecast for a city', inputSchema: z.object({ city: z.string() }) }, async ({ city }) => ({ content: [{ type: 'text', text: `Sunny in ${city} all week.` }] }) ); return server; });

SDK会在你的处理函数运行前，用z.object(...)schema验证每一次调用。

4.5 MCP客户端接入

在Python中接入MCP Server的示例：

python

from tina.mcp import MCPClient # 创建MCP客户端 mcp = MCPClient() # 添加stdio类型的MCP服务 mcp.add_server( server_id="weather", config={ "command": "uvx", "args": ["mcp-weather-server"] } ) # 调用MCP工具 result = await mcp.call_tool("weather", "get-forecast", {"city": "北京"})

4.6 Function Calling vs MCP vs Skills

Function Calling是AI Agent调用工具的基础能力，也是后面两个能够存在的基础。MCP提供了标准化接口，Skills则允许更细致地用文字定义指令、脚本和资源。

4.7 工具调用的最佳实践

用JSON Schema把工具描述干净。想让AI真正干活，先把手上的工具用JSON Schema描述干净，比换更贵的模型收益大。

明确触发条件。在工具描述中说明“何时使用此工具”，让模型能够准确判断调用时机。

参数验证前置。在工具执行前用Pydantic校验参数，避免模型传入错误类型的参数。

错误处理与重试。工具执行失败时，返回结构化的错误信息，让模型能够理解失败原因并尝试修正。

五、规划范式

5.1 三种核心规划范式

Agent的规划范式决定了它如何思考和执行任务。三种主流范式各有优劣：

5.2 ReAct范式深入分析

ReAct的核心思想是：先思考当前要做什么 → 选择一个动作（通常是调用某个工具）→ 获取观察结果 → 再思考下一步 → 循环，直到任务完成。

ReAct的失败模式：

注意力锁定（Attention Latch）。在仅解码器自回归Transformer模型中，存在一种系统性的失败模式——当推理链需要超过3跳合成时，ReAct基线的成功率会崩溃到0.1%。

死循环。模型可能反复调用同一个工具而不改变参数，因为它在上下文中看到了之前的失败尝试，而这反过来“引导”它继续尝试同样的方式。

5.3 Plan-and-Execute范式

Plan-and-Execute的核心思想是：先制定完整的执行计划，再逐步执行。与ReAct的“边想边做”不同，Plan-and-Execute在开始行动之前就有了全局视角。

python

from langgraph.graph import StateGraph, START, END def plan_and_execute_agent(query: str): """Plan-and-Execute Agent""" # Step 1: 规划阶段 plan_prompt = f"""你是一个任务规划专家。请将以下任务拆解为具体步骤： 任务：{query} 请输出一个步骤列表，每个步骤包含： - 步骤编号 - 步骤描述 - 所需工具（如果有） """ plan = call_llm(plan_prompt) # Step 2: 执行阶段 results = [] for step in parse_plan(plan): result = execute_step(step) results.append(result) # Step 3: 汇总阶段 summary_prompt = f"""基于以下执行结果，给出最终答案： 计划：{plan} 执行结果：{results} 最终答案： """ return call_llm(summary_prompt)

Plan-and-Execute的优势：在规划与执行之间提供了检查点，让你可以在执行之前审查计划。当执行过程中获得新信息时，还可以触发重新规划（Replanning），而不是继续沿着错误路径前进。

Plan-and-Execute的适用场景：需求相对简单、任务步骤可以预先确定、追求执行效率的场景。缺点是如果初始计划不准确，后续执行会偏离方向。

5.4 Reflection范式

Reflection Loop的核心流程是：生成初始答案 → 自我批判 → 修正生成。无论使用ReAct还是Plan-and-Execute，都建议与Reflection Loop结合使用，它对于最终的输出质量至关重要。

python

class ReflectionAgent: """Reflection Agent：生成 → 反思 → 改进""" def run(self, question: str, max_iterations: int = 3) -> str: # Step 1: 生成初始答案 answer = call_llm(f"请回答以下问题：{question}") for i in range(max_iterations): # Step 2: 反思——批判当前答案 reflection_prompt = f"""请审查以下回答，找出其中的错误、遗漏或不准确之处： 问题：{question} 回答：{answer} 请以JSON格式输出： {{"score": 0-10分, "issues": ["问题1", "问题2"], "suggestions": "改进建议"}} """ reflection = call_llm(reflection_prompt) parsed = parse_json(reflection) # Step 3: 如果分数足够高，停止迭代 if parsed["score"] >= 9: break # Step 4: 基于反思改进答案 answer = call_llm(f"""基于以下反馈改进你的回答： 原回答：{answer} 反馈：{parsed['suggestions']} 请给出改进后的回答：""") return answer

Reflection的适用场景：对输出质量要求高的任务，如代码生成、文档撰写、复杂分析。缺点是每次迭代都会额外消耗LLM调用成本。

5.5 混合模式：生产级Agent的最佳实践

在实际生产环境中，通常需要混合使用多种范式：

ReAct + Reflection：先用ReAct完成任务的初步执行，再用Reflection检查结果质量。这种方式兼顾了执行效率和输出质量。

Plan-and-Execute + ReAct：先用Plan-and-Execute制定高层计划，每个步骤内部用ReAct执行具体操作。这种方式适合复杂的多步骤任务。

范式路由（Paradigm Routing）：根据任务类型动态选择范式。简单任务直接用ReAct，复杂长任务用Plan-and-Execute，对质量要求高的用Reflection。一项研究提出了“Select-then-Solve”的范式路由方法，在推理时根据任务特征动态选择最优范式。

六、实践任务：用Python裸写一个ReAct循环的Agent

6.1 任务目标

用纯Python（不依赖任何Agent框架）实现一个ReAct循环的Agent，接入一个搜索工具，让它能自主判断“何时搜索、何时直接回答”。

6.2 完整代码

python

""" 最小化ReAct Agent实现 不依赖任何Agent框架，纯Python + OpenAI API """ import json import httpx from typing import Optional # ============ 工具定义 ============ TOOLS = [ { "type": "function", "function": { "name": "search_web", "description": "搜索网络获取信息。当用户询问你不确定的信息、需要实时数据、或需要验证事实时使用此工具。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词" } }, "required": ["query"] } } } ] def execute_tool(name: str, args: dict) -> str: """工具执行分发器""" if name == "search_web": return search_web(args["query"]) return f"未知工具: {name}" def search_web(query: str) -> str: """模拟搜索工具（实际项目中替换为真实API）""" # 这里可以接入SerpAPI、Tavily、Bing Search等 return f"关于'{query}'的搜索结果：这是一段模拟的搜索结果内容。" # ============ LLM调用 ============ async def call_llm(messages: list) -> dict: """调用LLM并返回响应""" async with httpx.AsyncClient() as client: resp = await client.post( "https://api.openai.com/v1/chat/completions", headers={ "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" }, json={ "model": "gpt-4o", "messages": messages, "tools": TOOLS, "tool_choice": "auto", "temperature": 0.1 }, timeout=60.0 ) return resp.json() # ============ ReAct核心循环 ============ async def agent_loop(query: str, max_iterations: int = 10) -> str: """ ReAct Agent核心循环 流程：思考 → 行动 → 观察 → 再思考 退出条件：模型不再调用工具 或 达到最大迭代次数 """ messages = [ {"role": "system", "content": "你是一个有用的助手。你可以使用搜索工具来获取信息。当你不确定答案时，使用搜索工具。"}, {"role": "user", "content": query} ] for i in range(max_iterations): # Step 1: 调用LLM（思考阶段） response = await call_llm(messages) choice = response["choices"][0] message = choice["message"] # Step 2: 检查是否有工具调用（行动阶段） tool_calls = message.get("tool_calls", []) if not tool_calls: # 没有工具调用，返回最终答案 return message.get("content", "任务完成") # Step 3: 执行每个工具调用并收集结果（观察阶段） messages.append(message) # 追加助手消息（含tool_calls） for tool_call in tool_calls: function_name = tool_call["function"]["name"] function_args = json.loads(tool_call["function"]["arguments"]) # 执行工具 result = execute_tool(function_name, function_args) # 追加工具结果 messages.append({ "role": "tool", "tool_call_id": tool_call["id"], "content": result }) # 循环回到Step 1（再思考） return "达到最大迭代次数，任务未完成" # ============ 入口 ============ import asyncio async def main(): print("=== ReAct Agent 测试 ===") queries = [ "今天北京天气怎么样？", # 应该触发搜索 "1+1等于几？", # 应该直接回答 "2026年诺贝尔物理学奖得主是谁？" # 应该触发搜索 ] for q in queries: print(f"\n用户：{q}") answer = await agent_loop(q) print(f"Agent：{answer}") if __name__ == "__main__": asyncio.run(main())

6.3 代码解析

退出条件的设计：循环在max_iterations（默认10次）内运行。每次迭代检查模型是否返回了tool_calls——如果没有，说明模型认为可以直接回答，任务完成，退出循环。

消息历史的管理：每轮迭代的完整流程是：追加助手消息（含tool_calls）→ 执行工具 → 追加工具结果 → 回到Step 1。这保证了模型在下一轮能看到完整的推理链和工具结果。

工具执行的分发：execute_tool函数根据工具名称分发到具体的执行函数。在实际项目中，这里可以接入真实的搜索API、数据库查询、代码执行器等。

6.4 测试与验证

运行上述代码后，观察Agent的行为：

• 当用户问“今天北京天气怎么样？”时，Agent应该判断需要搜索，调用search_web工具。

• 当用户问“1+1等于几？”时，Agent应该直接回答，不调用工具。

• 当用户问“2026年诺贝尔物理学奖得主是谁？”时，Agent应该判断需要搜索（因为这是训练数据之后的信息）。

七、总结

Agent核心组件进阶可以概括为五个模块：

1. 核心循环：ReAct范式是Agent的运行引擎，退出条件的设计决定系统是否会死循环。核心是“思考 → 行动 → 观察 → 再思考”的循环，直到模型不再调用工具。

2. 记忆系统：短期记忆依赖上下文窗口和滑动窗口，长期记忆依赖向量数据库。分层记忆架构（工作记忆、情景记忆、语义记忆）是生产级Agent的标准配置。

3. RAG：从文档解析到混合检索再到重排序的完整链路，核心是“高召回 + 高精度”的两阶段策略。向量检索和BM25互补，Cross-Encoder重排序提升精确率。

4. 工具调用与MCP：Function Calling是基础能力，MCP是标准化协议。工具描述的质量直接决定Agent的调用准确性。

5. 规划范式：ReAct适合路径未知的排查任务，Plan-and-Execute适合复杂长任务，Reflection适合对质量要求高的场景。生产级Agent通常混合使用多种范式。

把这五个模块理解透彻，并通过实践任务亲手实现一个ReAct Agent，你就真正跨过了“调用接口的程序员”与“Agent工程师”之间的分水岭。