LLM、MCP协议、AA协议、RAG、智能体(AI Agent) 图解详细讲解

随着大语言模型（LLM）技术的飞速发展，如何让模型更智能、更可控、能与外部世界交互，成了核心挑战。本文将从实战角度出发，用代码和图示（文字描述）带你彻底搞懂 LLM、MCP 协议、AA 协议、RAG 和 AI Agent 这五个核心概念。—## 1. LLM 基础与大模型调用LLM（Large Language Model）是智能体的“大脑”，它基于 Transformer 架构，通过海量文本训练获得语言理解和生成能力。实战中我们常用 OpenAI、Claude 或本地模型。### 1.1 基础 LLM 调用示例python# 示例1：使用 OpenAI API 调用 LLMimport openaifrom openai import OpenAI# 初始化客户端（请替换为你的 API Key）client = OpenAI(api_key="sk-your-api-key")def call_llm(prompt: str) -> str: """ 调用 GPT-4 模型生成回复 :param prompt: 用户输入的提示词 :return: 模型生成的文本 """ response = client.chat.completions.create( model="gpt-4", messages=[ {"role": "system", "content": "你是一个有用的助手。"}, {"role": "user", "content": prompt} ], temperature=0.7, max_tokens=500 ) return response.choices[0].message.content# 测试调用if __name__ == "__main__": user_input = "请用一句话解释什么是LLM？" result = call_llm(user_input) print(f"LLM回复: {result}")图解说明： 用户输入 → [System Prompt & User Message] → LLM (GPT-4) → 生成文本 → 返回结果—## 2. MCP 协议：模型与工具的通信桥梁MCP（Model Context Protocol）是一种让 LLM 与外部工具（如数据库、API、文件系统）安全交互的协议。它定义了“工具描述”的格式，让模型知道有哪些能力可用。### 2.1 MCP 核心概念- 工具 (Tool)：可被 LLM 调用的函数，如查询天气、计算数学。- 函数描述 (Function Description)：用 JSON Schema 描述工具的输入输出。- 调用循环：LLM 生成工具调用请求 → 执行工具 → 返回结果给 LLM。### 2.2 MCP 实战代码：集成天气查询python# 示例2：实现一个 MCP 风格的天气查询工具import jsonimport requestsfrom typing import Dict, Any# 定义工具描述（MCP 格式）TOOL_DESCRIPTION = { "name": "get_weather", "description": "获取指定城市的实时天气信息", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称，如北京、上海" } }, "required": ["city"] }}def get_weather(city: str) -> Dict[str, Any]: """ 模拟天气查询工具（实际可对接真实 API） :param city: 城市名称 :return: 天气信息字典 """ # 这里使用模拟数据，实际可集成 openweathermap 等 API weather_data = { "北京": {"温度": "15°C", "天气": "晴朗", "湿度": "45%"}, "上海": {"温度": "20°C", "天气": "多云", "湿度": "60%"}, "广州": {"温度": "28°C", "天气": "阵雨", "湿度": "80%"} } return weather_data.get(city, {"错误": f"未找到 {city} 的天气数据"})# MCP 调用处理器：将 LLM 请求路由到正确工具def mcp_tool_handler(llm_request: Dict) -> str: """ 处理 LLM 发起的工具调用请求 :param llm_request: 包含 tool_name 和参数的字典 :return: 工具执行结果（字符串） """ tool_name = llm_request.get("tool_name") parameters = llm_request.get("parameters", {}) if tool_name == "get_weather": city = parameters.get("city", "北京") result = get_weather(city) return json.dumps(result, ensure_ascii=False) else: return json.dumps({"error": f"未知工具: {tool_name}"})# 模拟 LLM 调用 MCP 工具if __name__ == "__main__": # 假设 LLM 决定调用天气工具 llm_decision = { "tool_name": "get_weather", "parameters": {"city": "上海"} } tool_result = mcp_tool_handler(llm_decision) print(f"MCP 工具返回: {tool_result}")图解说明： LLM → 分析用户问题 → 决定调用工具 → 生成 MCP 请求（工具名+参数） → 执行工具 → 返回结果给 LLM → 生成最终回复—## 3. AA 协议：智能体间的通信标准AA（Agent-to-Agent）协议定义了多个智能体如何协作完成任务。每个智能体有自己的角色和能力，通过标准化的消息格式交换信息。### 3.1 AA 协议消息格式json{ "sender": "agent_id", "receiver": "agent_id", "message_type": "request/response/broadcast", "payload": { "task": "具体任务内容" }, "context": { "session_id": "会话ID" }}### 3.2 实战：两个智能体协作实现“旅游规划”python# 示例3：AA 协议风格的智能体协作import jsonfrom typing import Dict, Listclass BaseAgent: """基础智能体类""" def __init__(self, agent_id: str, role: str): self.agent_id = agent_id self.role = role def send_message(self, receiver: str, message_type: str, payload: Dict) -> Dict: """发送 AA 协议消息""" return { "sender": self.agent_id, "receiver": receiver, "message_type": message_type, "payload": payload, "context": {"session_id": "sess-001"} } def receive_message(self, message: Dict) -> str: """处理接收到的消息（子类重写）""" raise NotImplementedErrorclass TravelPlannerAgent(BaseAgent): """旅游规划智能体 - 负责制定行程""" def receive_message(self, message: Dict) -> str: if message["message_type"] == "request": task = message["payload"].get("task", "") if "景点" in task or "行程" in task: return f"【规划智能体】已规划：北京3日经典游（故宫、长城、颐和园）" else: return f"【规划智能体】请提供更具体的需求"class WeatherAgent(BaseAgent): """天气智能体 - 负责查询天气""" def receive_message(self, message: Dict) -> str: if message["message_type"] == "request": city = message["payload"].get("city", "北京") # 模拟天气查询 weather_info = {"北京": "晴朗 15°C", "上海": "多云 20°C"} return f"【天气智能体】{city}天气：{weather_info.get(city, '未知')}" return "【天气智能体】无法处理该请求"# 测试 AA 协议协作if __name__ == "__main__": planner = TravelPlannerAgent("agent-001", "规划师") weather = WeatherAgent("agent-002", "天气顾问") # 用户请求：规划一个北京的旅游行程 user_request = {"task": "请帮我规划北京3日游行程，并查询天气"} # 第一步：规划智能体处理 msg_to_planner = planner.send_message( receiver="agent-001", message_type="request", payload={"task": user_request["task"]} ) plan_result = planner.receive_message(msg_to_planner) print(plan_result) # 第二步：天气智能体查询 msg_to_weather = weather.send_message( receiver="agent-002", message_type="request", payload={"city": "北京"} ) weather_result = weather.receive_message(msg_to_weather) print(weather_result) # 第三步：汇总结果（模拟） final_result = f"【汇总】完整旅游方案：\n{plan_result}\n{weather_result}" print(final_result)图解说明： 用户 → 主智能体 → [AA协议消息] → 规划智能体 → [返回行程] → [AA协议消息] → 天气智能体 → [返回天气] → 汇总结果 → 用户—## 4. RAG：让 LLM 拥有实时知识RAG（Retrieval-Augmented Generation）通过从外部知识库检索相关信息，增强 LLM 的回答准确性和时效性。核心流程：用户问题 → 向量检索 → 拼接上下文 → LLM 生成答案。### 4.1 实战：基于 FAISS 的简易 RAG 系统python# 示例4：RAG 检索增强生成实现import numpy as npfrom sentence_transformers import SentenceTransformerimport faissclass SimpleRAG: """简易 RAG 系统""" def __init__(self, documents: List[str]): self.encoder = SentenceTransformer('all-MiniLM-L6-v2') # 嵌入模型 self.documents = documents self._build_index() def _build_index(self): """构建 FAISS 向量索引""" embeddings = self.encoder.encode(self.documents) dimension = embeddings.shape[1] self.index = faiss.IndexFlatL2(dimension) # L2距离索引 self.index.add(embeddings.astype(np.float32)) def retrieve(self, query: str, top_k: int = 2) -> List[str]: """检索最相关的文档""" query_embedding = self.encoder.encode([query]) distances, indices = self.index.search(query_embedding.astype(np.float32), top_k) return [self.documents[i] for i in indices[0]] def generate(self, query: str) -> str: """基于检索结果生成答案（模拟 LLM）""" relevant_docs = self.retrieve(query) context = "\n".join(relevant_docs) # 实际项目中这里调用 LLM API prompt = f"基于以下信息回答问题：\n{context}\n\n问题：{query}\n答案：" # 模拟 LLM 生成 return f"[生成答案] 根据检索到的信息：{context[:50]}... 回答：{query} 的相关内容是..."# 测试 RAGif __name__ == "__main__": # 知识库文档 docs = [ "MCP协议是模型上下文协议，用于LLM与工具交互。", "AA协议是智能体间通信协议，支持多Agent协作。", "RAG技术通过检索增强生成，提高LLM准确性。", "AI Agent是能自主执行任务的智能系统。" ] rag_system = SimpleRAG(docs) query = "什么是MCP协议？" result = rag_system.generate(query) print(f"用户问题: {query}") print(f"RAG回答: {result}")图解说明： 用户问题 → 向量化 → [FAISS索引] → 检索相关文档 → 拼接上下文 → LLM → 生成答案—## 5. AI Agent：自主执行任务的智能体AI Agent 是集成了 LLM、工具、记忆和规划能力的自主系统。它能理解复杂任务、分解步骤、调用工具、处理反馈并最终完成目标。### 5.1 实战：一个完整的 AI Agent 示例python# 示例5：构建一个简单的 AI Agentimport jsonfrom typing import Dict, List, Callableclass Tool: """工具类""" def __init__(self, name: str, func: Callable, description: str): self.name = name self.func = func self.description = descriptionclass AIAgent: """简易 AI Agent""" def __init__(self, tools: List[Tool]): self.tools = {tool.name: tool for tool in tools} self.memory = [] # 短期记忆 def think(self, task: str) -> List[str]: """ 思考并规划任务步骤（模拟 LLM 推理） 实际项目中这里调用 LLM 进行任务分解 """ # 模拟 LLM 的任务分解能力 if "天气" in task and "行程" in task: return ["get_weather", "plan_trip"] elif "天气" in task: return ["get_weather"] else: return ["default_reply"] def execute(self, plan: List[str], task: str) -> str: """执行规划好的步骤""" results = [] for step in plan: if step in self.tools: tool = self.tools[step] # 简单参数解析（实际需更复杂） if "天气" in task: result = tool.func("北京") else: result = tool.func(task) results.append(result) self.memory.append(f"执行工具 {step}: {result}") else: results.append(f"[默认回复] 无法处理步骤: {step}") return "\n".join(results) def run(self, task: str) -> str: """运行 Agent 完整流程""" print(f"[Agent] 接收任务: {task}") plan = self.think(task) print(f"[Agent] 规划步骤: {plan}") result = self.execute(plan, task) print(f"[Agent] 执行结果:\n{result}") return result# 定义工具函数def get_weather_tool(city: str) -> str: return f"天气工具: {city} 今天晴，15-25°C"def plan_trip_tool(task: str) -> str: return f"行程工具: 已规划3日游行程（故宫、长城）"# 创建 Agentif __name__ == "__main__": weather_tool = Tool("get_weather", get_weather_tool, "查询天气") trip_tool = Tool("plan_trip", plan_trip_tool, "规划行程") agent = AIAgent(tools=[weather_tool, trip_tool]) # 测试复杂任务 agent.run("查询北京的天气并规划一个行程")图解说明： 用户输入 → Agent接收 → [LLM推理] → 任务分解 → [工具1] → [工具2] → 汇总结果 → 返回用户 ↓ [记忆存储]—## 6. 总结本文从实战角度讲解了 LLM、MCP 协议、AA 协议、RAG 和 AI Agent 的核心概念与实现：| 技术 | 核心作用 | 实战要点 ||------|---------|----------|| LLM | 语言理解与生成 | API 调用、Prompt 设计 || MCP | 模型与工具交互 | 工具描述 JSON Schema、执行循环 || AA | 多智能体协作 | 消息格式、角色分工 || RAG | 知识增强 | 向量检索、文档分块 || AI Agent | 自主任务执行 | 任务规划、工具调用、记忆管理 |未来趋势：这些技术正在融合——AI Agent 内部使用 MCP 调用工具，通过 RAG 获取知识，多个 Agent 通过 AA 协议协作，最终由 LLM 驱动决策。掌握这些技术栈，你就能构建出真正智能、实用的 AI 应用。建议读者在实际项目中逐步集成这些组件，从简单的 LLM 调用开始，逐步加入 RAG 和 Agent 能力。