尧图网页设计 
你是不是也遇到过这样的困惑看到各种AI Agent、RAG、LangChain的技术文章满天飞每个概念都懂一点但真要自己动手搭建一个能用的智能体系统却发现无从下手网上教程要么太零散要么直接甩给你一堆代码根本不解释为什么这么设计。这就是为什么我要写这篇教程。我见过太多开发者卡在“知道概念”到“能实战”的鸿沟里。这篇文章不是简单的概念罗列而是一套从零到一构建企业级AI Agent的完整工程化指南。我会带你亲手搭建一个融合了Agent核心思想、RAG知识库增强、MCP工具扩展以及LangGraph工作流编排的实战项目。读完本文你将彻底搞懂AI Agent的核心组件它不只是调用API而是由规划、记忆、工具使用、反思构成的完整智能体。RAG如何真正落地从文档处理、向量化到检索优化避开“向量数据库即RAG”的误区。MCP协议的价值如何用Model Context Protocol统一且安全地扩展Agent的能力边界。LangChain与LangGraph的分工前者是乐高积木后者是设计图纸如何组合使用。一个可运行的、模块化的企业级项目架构包含代码、配置、部署和调试全流程。我们从一个真实的场景开始构建一个“技术文档智能问答与任务执行Agent”。它不仅能回答你公司内部的技术文档问题RAG还能根据你的指令执行诸如创建JIRA工单、查询数据库、生成代码片段等操作Agent Tools。下面我们就从最根本的问题开始。1. 这篇文章真正要解决的问题为什么你的AI项目总是“Demo级”很多开发者尝试过AI项目但结果往往是本地跑通一个示例很开心一旦想接入真实数据、对接企业系统、处理复杂逻辑项目就迅速变得难以维护和扩展。问题通常出在以下几个方面架构混乱业务逻辑、AI调用、工具集成、状态管理全部混在一起代码像“意大利面条”。缺乏工程化没有清晰的模块划分、配置管理、错误处理和日志监控。对核心概念理解片面以为用了LangChain就是Agent装了向量数据库就是RAG忽略了记忆、规划、反思等关键机制。扩展性差每增加一个新工具或数据源就要大改核心代码。本文要解决的正是如何跨越从“玩具Demo”到“生产可用系统”的鸿沟。我们将采用分层和模块化的设计思想确保每一部分Agent核心、知识库、工具、工作流都能独立开发、测试和替换。无论你是想构建一个内部的效率助手还是一个面向客户的产品级智能体这套架构都能提供坚实的基础。2. 基础概念与核心原理拆解AI Agent的四大支柱在动手之前必须统一认知。一个完整的AI Agent智能体远不止一个“会聊天的AI”。我们可以将其类比为一个高级别的AI工程师实习生它具备以下核心能力核心组件通俗解释技术实现关键类比AI工程师实习生规划 (Planning)拆解复杂任务为可执行的步骤序列。Chain of Thought, ReAct框架 LangGraph的状态机。接到需求后先写技术方案拆解成子任务。工具使用 (Tool Use)调用外部API、数据库、系统命令来获取信息或执行动作。Function Calling, Tool Definition, MCP协议。会使用Git提交代码、用JIRA创建任务、查数据库文档。记忆 (Memory)保留对话历史、任务上下文和学到的知识。短期记忆对话缓存长期记忆向量数据库 LangGraph的持久化状态。记得刚才讨论的架构设计也记得公司项目的通用规范。反思 (Reflection)评估自身行动和结果进行修正和优化。Self-Critique, 验证步骤输出循环Loop控制。代码跑失败了会看日志、分析原因、重新尝试。而RAG、MCP、LangChain/LangGraph是帮助我们实现这些能力的“基础设施”RAG是Agent长期记忆和知识库的核心。它让Agent能“阅读”并理解你提供的私有文档如公司wiki、产品手册从而给出精准回答。其流程是文档加载 → 文本分割 → 向量化嵌入 → 存储到向量数据库 → 用户提问时检索相关片段 → 连同片段和问题交给大模型生成答案。MCP是Agent安全、标准化使用工具的协议。你可以把它想象成智能体的“USB标准接口”。任何工具查天气、操作数据库、调用内部API只要按照MCP协议实现一个Server就能被Agent即插即用无需为每个工具写一遍胶水代码。LangChain是一个丰富的“组件库”提供了连接大模型、处理文档、管理记忆、定义工具等大量标准化模块。它让你不用重复造轮子。LangGraph是一个基于图的“工作流编排引擎”。当任务步骤之间存在复杂依赖、循环或分支时比如先检索、再判断、不行就换关键词再检索用LangChain的简单链Chain会很吃力。LangGraph允许你以可视化或代码的方式定义状态State和节点Node精确控制Agent的执行流是实现复杂规划和反思的关键。理解了这些你就知道我们不是在堆砌时髦名词而是在用正确的工具解决正确的问题。接下来我们开始搭建环境。3. 环境准备与前置条件我们将使用Python作为开发语言。请确保你的环境满足以下要求操作系统Linux/macOS/Windows (WSL2推荐)Python版本3.10 或 3.11这是大多数AI库兼容性最好的版本包管理工具pip或poetry本文使用pip核心依赖langchainlangchain-community: 核心组件库。langgraph: 工作流编排。openai: 调用GPT等模型也可替换为langchain-anthropic,langchain-groq等。向量数据库客户端例如chromadb(轻量)或pinecone(云服务)。嵌入模型例如sentence-transformers(本地)或openai的text-embedding-ada-002(API)。mcpMCP协议Python SDK。fastapiuvicorn: 用于构建简单的工具服务或Agent Web接口。第一步创建项目并安装依赖# 创建项目目录 mkdir ai-agent-tutorial cd ai-agent-tutorial # 创建虚拟环境强烈推荐 python -m venv venv # 激活虚拟环境 # Linux/macOS: source venv/bin/activate # Windows: # venv\Scripts\activate # 安装核心依赖 pip install langchain langgraph langchain-community # 安装OpenAI如果你使用GPT pip install openai # 安装本地向量数据库和嵌入模型 pip install chromadb sentence-transformers # 安装MCP pip install mcp # 安装Web框架可选用于最后演示 pip install fastapi uvicorn第二步准备API密钥如果你使用OpenAI的模型需要设置环境变量。# Linux/macOS export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) # $env:OPENAI_API_KEYyour-api-key-here也可以创建.env文件管理使用python-dotenv库读取。环境就绪现在我们进入核心环节构建一个模块化的企业级AI Agent项目。4. 项目架构设计模块化与分层思想在写代码前先看设计图。一个健壮的系统源于清晰的架构。我们将项目分为以下层次和模块ai-agent-tutorial/ ├── config/ # 配置管理 │ ├── __init__.py │ ├── settings.py # 应用配置API密钥、模型选择等 │ └── prompts.py # 所有提示词模板 ├── core/ # 核心Agent逻辑 │ ├── __init__.py │ ├── agent.py # Agent核心类规划、执行、反思 │ └── state.py # LangGraph状态定义 ├── knowledge/ # RAG知识库模块 │ ├── __init__.py │ ├── loader.py # 文档加载器 │ ├── splitter.py # 文本分割器 │ ├── vector_store.py # 向量数据库封装 │ └── retriever.py # 检索器封装 ├── tools/ # 工具模块MCP及其他 │ ├── __init__.py │ ├── mcp_clients/ # MCP工具客户端 │ │ ├── __init__.py │ │ └── jira_client.py # 示例JIRA工具 │ └── custom_tools.py # 自定义非MCP工具 ├── workflows/ # LangGraph工作流定义 │ ├── __init__.py │ └── main_workflow.py # 主问答与执行工作流 ├── app.py # FastAPI应用入口可选 ├── requirements.txt └── .env.example这个结构的关键在于解耦。knowledge模块只关心文档处理和检索tools模块只关心如何调用外部服务core中的agent负责高层规划和决策workflows用LangGraph把它们粘合起来。任何一部分需要升级或替换影响范围都是可控的。5. 核心流程拆解与实现我们将按照“自底向上”的顺序实现先准备知识库和工具再构建Agent核心最后用工作流串联。5.1 第一步构建RAG知识库模块知识库是Agent的“长期记忆”。我们实现一个简单的本地知识库支持加载Markdown文档。文件knowledge/loader.pyimport os from typing import List from langchain_community.document_loaders import DirectoryLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document class KnowledgeLoader: 文档加载与处理类 def __init__(self, directory_path: str, glob_pattern**/*.md): self.directory_path directory_path self.glob_pattern glob_pattern def load_documents(self) - List[Document]: 加载目录下所有文档 if not os.path.exists(self.directory_path): raise FileNotFoundError(f知识库目录不存在: {self.directory_path}) loader DirectoryLoader( self.directory_path, globself.glob_pattern, loader_clsTextLoader, loader_kwargs{autodetect_encoding: True} ) documents loader.load() print(f已加载 {len(documents)} 个文档) return documents def split_documents(self, documents: List[Document], chunk_size1000, chunk_overlap200) - List[Document]: 分割文档为适合嵌入的片段 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) splits text_splitter.split_documents(documents) print(f文档分割为 {len(splits)} 个片段) return splits文件knowledge/vector_store.pyimport chromadb from chromadb.config import Settings from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings # 或者使用本地嵌入模型 from langchain.embeddings import HuggingFaceEmbeddings from typing import List import os class VectorStoreManager: 向量数据库管理类 def __init__(self, persist_directory: str ./chroma_db, embedding_model_name: str all-MiniLM-L6-v2): self.persist_directory persist_directory # 使用本地嵌入模型避免API调用和费用 self.embeddings HuggingFaceEmbeddings( model_namefsentence-transformers/{embedding_model_name} ) # 或者使用OpenAI嵌入需要API Key # self.embeddings OpenAIEmbeddings() def create_vector_store(self, documents: List): 创建并持久化向量存储 vectordb Chroma.from_documents( documentsdocuments, embeddingself.embeddings, persist_directoryself.persist_directory ) vectordb.persist() print(f向量数据库已创建并保存至: {self.persist_directory}) return vectordb def load_vector_store(self): 加载已存在的向量存储 if not os.path.exists(self.persist_directory): raise FileNotFoundError(f向量数据库目录不存在: {self.persist_directory}) vectordb Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings ) print(f向量数据库已从 {self.persist_directory} 加载) return vectordb def get_retriever(self, vectordb, search_typesimilarity, k4): 获取检索器 return vectordb.as_retriever( search_typesearch_type, search_kwargs{k: k} )使用示例初始化知识库创建一个脚本init_knowledge_base.pyimport sys sys.path.append(.) from knowledge.loader import KnowledgeLoader from knowledge.vector_store import VectorStoreManager def main(): # 1. 加载文档 loader KnowledgeLoader(directory_path./data/docs) # 假设你的文档放在./data/docs下 raw_docs loader.load_documents() # 2. 分割文档 split_docs loader.split_documents(raw_docs) # 3. 创建向量存储 vs_manager VectorStoreManager(persist_directory./chroma_db) vectordb vs_manager.create_vector_store(split_docs) # 4. 获取检索器备用 retriever vs_manager.get_retriever(vectordb) print(知识库初始化完成) if __name__ __main__: main()运行前请在项目根目录创建data/docs文件夹并放入一些Markdown格式的技术文档。运行此脚本后会在本地生成chroma_db目录里面就是向量化后的知识库。5.2 第二步通过MCP协议集成工具MCP的核心思想是标准化。我们以“查询天气”和“创建JIRA工单”为例展示如何集成MCP工具。首先你需要一个MCP Server。这里我们以调用一个公开的天气API为例模拟一个简单的MCP Server。概念澄清在实际生产中工具提供方如JIRA系统管理员会部署标准的MCP Server。作为Agent开发者你通常只需要作为Client去连接它。但为了演示我们写一个最简单的Server。文件tools/mcp_clients/weather_server.py(模拟MCP Server)# 这是一个极度简化的MCP Server示例用于演示概念。 # 真实MCP Server实现需遵循官方协议。 import json from typing import Dict, Any import requests from fastapi import FastAPI import uvicorn app FastAPI(titleMock Weather MCP Server) # 模拟MCP Server的工具列表端点 app.get(/tools) async def list_tools(): 列出此Server提供的工具 return { tools: [ { name: get_weather, description: 获取指定城市的当前天气情况, inputSchema: { type: object, properties: { city: {type: string, description: 城市名称如 Beijing} }, required: [city] } } ] } # 模拟MCP Server的工具调用端点 app.post(/tools/call) async def call_tool(tool_input: Dict[str, Any]): 调用工具 tool_name tool_input.get(name) arguments tool_input.get(arguments, {}) if tool_name get_weather: city arguments.get(city, Beijing) # 这里模拟一个API调用实际应调用真实天气API # 例如response requests.get(fhttps://api.weatherapi.com/v1/current.json?keyYOUR_KEYq{city}) # 为演示我们返回模拟数据 mock_data { city: city, temperature: 22°C, condition: Sunny, humidity: 65% } return { content: [ { type: text, text: f{city}的天气是{mock_data[condition]}温度{mock_data[temperature]}湿度{mock_data[humidity]}。 } ] } else: return {error: f未知工具: {tool_name}} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8001)运行python tools/mcp_clients/weather_server.py启动这个模拟Server。文件tools/mcp_clients/weather_client.py(MCP Client封装)import requests from typing import Dict, Any from langchain.tools import Tool class WeatherMCPClient: 封装对模拟天气MCP Server的调用 def __init__(self, server_url: str http://localhost:8001): self.server_url server_url def get_weather(self, city: str) - str: 调用远程MCP Server获取天气 try: payload { name: get_weather, arguments: {city: city} } response requests.post( f{self.server_url}/tools/call, jsonpayload, timeout10 ) response.raise_for_status() result response.json() # 解析MCP格式的返回内容 if content in result and len(result[content]) 0: return result[content][0][text] else: return f调用成功但返回格式异常: {result} except Exception as e: return f调用天气服务失败: {str(e)} def as_langchain_tool(self) - Tool: 转换为LangChain可用的Tool对象 return Tool( nameget_weather, funcself.get_weather, description获取指定城市的当前天气情况。输入应为城市名称如 Beijing。 ) # 示例如何集成到LangChain def get_weather_tool(): client WeatherMCPClient() return client.as_langchain_tool()这样我们就有了一个标准的、可被Agent调用的“天气工具”。对于JIRA、数据库等工具模式完全一样连接MCP Server → 封装调用 → 转换为LangChain Tool。这保证了工具集的整洁和可扩展性。5.3 第三步定义Agent核心与状态这是智能体的“大脑”。我们使用LangGraph来定义Agent的状态和执行逻辑。文件core/state.pyfrom typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): 定义Agent工作流的状态。 这是LangGraph中在各个节点间传递的上下文。 # 用户输入的问题 input: str # 从知识库检索到的相关文档 retrieved_docs: List[str] # Agent思考的步骤或规划 plan: List[str] # 工具调用的结果 tool_outputs: List[str] # 最终给用户的答案 final_answer: str # 迭代次数用于控制循环 iteration: Annotated[int, operator.add]文件core/agent.pyfrom langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_react_agent from langchain.memory import ConversationBufferMemory from langchain import hub from core.state import AgentState import os class CoreAgent: Agent核心封装规划、执行和工具调用逻辑 def __init__(self, tools, model_namegpt-3.5-turbo): # 初始化大语言模型 self.llm ChatOpenAI( modelmodel_name, temperature0, # 降低随机性使输出更确定 api_keyos.getenv(OPENAI_API_KEY) ) self.tools tools # 从LangChain Hub拉取一个优秀的ReAct提示词模板 # 你也可以自定义放在 config/prompts.py 中 self.prompt hub.pull(hwchase17/react) # 创建具有ReAct推理能力的Agent self.agent create_react_agent( llmself.llm, toolsself.tools, promptself.prompt ) # 创建执行器并开启详细日志便于调试 self.agent_executor AgentExecutor( agentself.agent, toolsself.tools, verboseTrue, handle_parsing_errorsTrue, # 优雅处理解析错误 max_iterations5, # 防止无限循环 early_stopping_methodgenerate # 达到满意结果时提前停止 ) def run(self, input_text: str) - str: 执行单轮问答 try: result self.agent_executor.invoke({input: input_text}) return result.get(output, Agent未返回明确结果。) except Exception as e: return fAgent执行过程中出错: {str(e)}这个CoreAgent类封装了一个基于ReAct框架的、能使用工具的智能体。但它还缺少与知识库RAG的深度集成以及更复杂的工作流如先检索再判断是否需要工具。这正是LangGraph要发挥作用的地方。5.4 第四步用LangGraph编排高级工作流现在我们把知识库检索、工具调用和Agent推理组合成一个智能的工作流。这个工作流能决定用户的问题应该直接查知识库回答还是需要调用工具执行任务文件workflows/main_workflow.pyfrom typing import Literal from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor, ToolInvocation from langchain_core.messages import HumanMessage, AIMessage from core.state import AgentState from knowledge.vector_store import VectorStoreManager from core.agent import CoreAgent import json class MainWorkflow: 主工作流结合RAG检索和工具调用的决策流 def __init__(self, retriever, tools): self.retriever retriever self.tools tools self.tool_executor ToolExecutor(tools) # 初始化一个简单的LLM用于路由决策 from langchain_openai import ChatOpenAI self.router_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 构建图 self.graph self._build_graph() self.compiled_graph self.graph.compile() def _retrieve_node(self, state: AgentState) - AgentState: 节点1从知识库检索相关文档 print(f[检索节点] 正在检索与问题相关的文档: {state[input]}) docs self.retriever.get_relevant_documents(state[input]) state[retrieved_docs] [doc.page_content for doc in docs] print(f[检索节点] 检索到 {len(docs)} 个相关片段。) return state def _route_question(self, state: AgentState) - Literal[answer_from_knowledge, use_agent_tools]: 路由决策判断问题类型 question state[input] docs state[retrieved_docs] # 构建路由提示词 router_prompt f 你是一个智能路由助手。请分析用户的问题和检索到的知识库内容决定下一步操作。 用户问题: {question} 检索到的知识库内容前200字符: {docs[0][:200] if docs else 无相关内容} 请根据以下规则判断 1. 如果知识库内容**明确、直接**回答了用户问题选择 answer_from_knowledge。 2. 如果用户问题涉及**执行具体操作**如创建任务、查询实时信息、计算等或者知识库内容**无法回答**选择 use_agent_tools。 只返回 answer_from_knowledge 或 use_agent_tools 中的一个。 response self.router_llm.invoke(router_prompt) decision response.content.strip().lower() print(f[路由决策] 问题类型判断为: {decision}) return decision def _answer_from_knowledge_node(self, state: AgentState) - AgentState: 节点2基于知识库生成答案 print([知识库回答节点] 正在综合知识库内容生成答案...) question state[input] context \n\n.join(state[retrieved_docs]) # 使用LLM综合上下文生成答案 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo) prompt f 请根据以下提供的上下文信息回答用户的问题。如果上下文信息不足以回答问题请如实告知。 上下文信息 {context} 用户问题{question} 请给出专业、准确的回答 answer llm.invoke(prompt) state[final_answer] answer.content print(f[知识库回答节点] 答案生成完成。) return state def _agent_tools_node(self, state: AgentState) - AgentState: 节点3调用Agent工具来回答问题/执行任务 print([Agent工具节点] 正在调用工具处理复杂任务...) # 这里可以复用之前定义的CoreAgent或者直接调用工具执行器 # 简化演示我们直接让一个Agent执行器来处理 from core.agent import CoreAgent agent CoreAgent(toolsself.tools) result agent.run(state[input]) state[final_answer] result print(f[Agent工具节点] 工具调用完成。) return state def _build_graph(self): 构建LangGraph图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(retrieve, self._retrieve_node) workflow.add_node(answer_from_knowledge, self._answer_from_knowledge_node) workflow.add_node(agent_tools, self._agent_tools_node) # 设置入口点 workflow.set_entry_point(retrieve) # 设置条件边检索后根据路由决策走不同分支 workflow.add_conditional_edges( retrieve, self._route_question, { answer_from_knowledge: answer_from_knowledge, use_agent_tools: agent_tools, } ) # 设置终点 workflow.add_edge(answer_from_knowledge, END) workflow.add_edge(agent_tools, END) return workflow def run(self, question: str) - dict: 运行工作流 initial_state: AgentState { input: question, retrieved_docs: [], plan: [], tool_outputs: [], final_answer: , iteration: 0 } final_state self.compiled_graph.invoke(initial_state) return final_state这个工作流实现了智能路由无论什么问题先检索知识库。用一个轻量级LLM判断知识库能直接回答吗如果能走分支A合成答案如果不能或需要执行操作走分支B启动能使用工具的Agent。这比单一的RAG或单一的Agent更强大、更灵活。6. 运行结果与效果验证让我们把所有模块组装起来进行端到端的测试。文件run_agent.py(主运行脚本)import sys sys.path.append(.) from knowledge.vector_store import VectorStoreManager from tools.mcp_clients.weather_client import get_weather_tool from workflows.main_workflow import MainWorkflow def main(): print( 启动企业级AI Agent系统 ) # 1. 加载知识库 print(1. 加载向量知识库...) vs_manager VectorStoreManager(persist_directory./chroma_db) try: vectordb vs_manager.load_vector_store() retriever vs_manager.get_retriever(vectordb) print( 知识库加载成功。) except Exception as e: print(f 知识库加载失败: {e}。请先运行 init_knowledge_base.py 初始化。) return # 2. 准备工具集 print(2. 初始化工具集...) weather_tool get_weather_tool() # 这里可以添加更多工具如JIRA工具、数据库查询工具等 tools [weather_tool] print(f 已加载 {len(tools)} 个工具。) # 3. 初始化工作流 print(3. 初始化主工作流...) workflow MainWorkflow(retrieverretriever, toolstools) print( 系统初始化完成\n) # 4. 交互式问答 print(请输入您的问题输入 quit 退出:) while True: user_input input(\n ) if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input.strip(): continue print(\n *50) print(f[用户问题] {user_input}) print(*50) try: result workflow.run(user_input) print(f\n[系统回答] {result.get(final_answer, 未生成答案)}) except Exception as e: print(f\n[错误] 处理问题时发生异常: {e}) if __name__ __main__: main()运行与验证启动知识库确保已运行init_knowledge_base.py并生成了./chroma_db。启动模拟工具Server在另一个终端运行python tools/mcp_clients/weather_server.py。运行主程序python run_agent.py。预期交互示例 启动企业级AI Agent系统 1. 加载向量知识库... 知识库加载成功。 2. 初始化工具集... 已加载 1 个工具。 3. 初始化主工作流... 系统初始化完成 请输入您的问题输入 quit 退出: 我们公司的API网关超时设置默认是多少秒 [用户问题] 我们公司的API网关超时设置默认是多少秒 [检索节点] 正在检索与问题相关的文档: 我们公司的API网关超时设置默认是多少秒 [检索节点] 检索到 3 个相关片段。 [路由决策] 问题类型判断为: answer_from_knowledge [知识库回答节点] 正在综合知识库内容生成答案... [知识库回答节点] 答案生成完成。 [系统回答] 根据公司技术文档API网关的默认超时设置为30秒。如果您的服务需要更长处理时间建议在网关配置中单独调整该路由的超时参数。 今天北京天气怎么样 [用户问题] 今天北京天气怎么样 [检索节点] 正在检索与问题相关的文档: 今天北京天气怎么样 [检索节点] 检索到 1 个相关片段。 [路由决策] 问题类型判断为: use_agent_tools [Agent工具节点] 正在调用工具处理复杂任务... Entering new AgentExecutor chain... 我需要调用天气查询工具来获取北京今天的天气。 Action: get_weather Action Input: {city: Beijing} Observation: Beijing的天气是Sunny温度22°C湿度65%。 Thought:我已经获取到了北京的天气信息可以给出最终答案了。 Final Answer: 北京今天的天气是晴天气温22摄氏度湿度65%。 [Agent工具节点] 工具调用完成。 [系统回答] 北京今天的天气是晴天气温22摄氏度湿度65%。你可以看到系统能自动判断问题类型并选择最合适的路径来回答。对于知识库已有的信息它直接检索回答对于需要实时信息的天气查询它成功调用了MCP工具。7. 常见问题与排查思路在实践过程中你可能会遇到以下问题问题现象可能原因排查方式解决方案运行init_knowledge_base.py时报No module named langchain虚拟环境未激活或依赖未安装。在终端输入pip list | grep langchain。激活虚拟环境并运行pip install -r requirements.txt。知识库检索结果不相关1. 文档分割块太大或太小。2. 嵌入模型不适合中文或领域文本。3. 检索参数k不合适。1. 检查分割后的片段长度。2. 尝试不同的嵌入模型。3. 调整检索器的k值。1. 调整chunk_size和chunk_overlap。2. 换用text-embedding-ada-002或paraphrase-multilingual模型。3. 尝试similarity_score_threshold检索。Agent陷入循环不输出最终答案1.max_iterations设置过高。2. 工具描述不清晰导致LLM无法正确选择。3. ReAct提示词不适合当前任务。1. 观察Agent执行日志看它在重复什么动作。2. 检查工具的描述是否准确。1. 适当降低max_iterations。2. 优化工具的描述使其更精确。3. 尝试不同的提示词模板或自定义提示词。调用MCP工具超时或失败1. MCP Server未启动或地址错误。2. 网络问题。3. 请求/响应格式不符合MCP协议。1. 用curl或Postman直接测试MCP Server端点。2. 检查客户端代码中的URL和端口。1. 确保MCP Server正常运行。2. 检查防火墙和网络配置。3. 对照MCP官方协议检查数据格式。LangGraph工作流编译错误1. 状态State定义与节点返回值不匹配。2. 图的结构有误如未连接END。1. 仔细检查AgentState的字段类型。2. 使用workflow.get_graph().draw_mermaid()可视化图结构。1. 确保每个节点都返回一个完整的、或部分更新的State字典。2. 确保所有路径最终都指向END。生产环境部署后性能差1. 向量检索未使用索引优化。2. LLM API调用延迟高。3. 未做缓存。1. 监控各环节耗时。2. 使用异步调用。3. 考虑更高效的向量数据库。1. 对频繁查询的向量数据建立索引。2. 对LLM回答和检索结果进行缓存。3. 考虑使用更快的本地模型或优化后的API模型。8. 最佳实践与工程建议要将这个Demo升级为生产级系统你需要关注以下几点配置中心化将所有API密钥、模型名称、服务器地址等配置项移出代码使用环境变量或配置管理工具如pydantic-settings管理。日志与监控为关键节点检索、工具调用、LLM请求添加结构化日志。监控Token消耗、响应时间和错误率。缓存策略LLM缓存对相同提示词的LLM请求进行缓存节省成本和时间。向量检索缓存对常见问题的检索结果进行缓存。可以使用langchain.cache或redis实现。异步优化对于IO密集型操作如调用多个工具、网络请求使用异步asyncio来提升并发性能。测试与评估单元测试测试每个独立模块如文档加载、工具调用。集成测试测试整个工作流。评估体系定义评估指标如答案相关性、事实准确性、工具调用正确率并定期用测试集进行评估。安全与权限工具权限不是所有用户都能调用所有工具。实现基于用户或角色的工具权限控制。输入输出过滤对用户输入和模型输出进行安全检查防止Prompt注入或输出有害内容。数据脱敏确保知识库和工具调用不泄露敏感信息。可观测性考虑集成像LangSmith这样的平台它可以可视化跟踪每个LangChain/LangGraph调用的详细步骤、耗时和中间结果是调试和优化Agent的利器。版本管理对提示词、工作流图、知识库文档进行版本控制便于回滚和A/B测试。9. 总结与后续学习方向通过这个完整的项目我们实现了一个模块化、可扩展、具备决策能力的企业级AI Agent原型。它不仅仅是调用API而是融合了RAG知识、MCP工具、LangGraph流程的智能系统。本文带你走通了几个关键环节清晰的概念理解明白了Agent、RAG、MCP、LangChain、LangGraph各自扮演的角色。工程化的项目结构学会了如何用分层的代码结构来组织复杂的AI应用。核心模块的实现亲手实现了知识库构建、MCP工具封装、Agent核心和工作流编排。智能路由决策用LangGraph构建了一个能自动选择“查知识库”还是“调用工具”的智能工作流。如果你想继续深入可以从以下几个方向探索更复杂的工具尝试集成真实的JIRA、Confluence、数据库或GitLab的MCP Server让Agent能执行更真实的办公任务。高级记忆机制实现对话历史记忆短期以及基于用户反馈更新知识库的能力长期学习。多Agent协作使用LangGraph创建多个具有不同专长的Agent如一个负责分析一个负责执行一个负责检查让它们协同完成复杂项目。前端界面使用Gradio或Streamlit快速构建一个Web界面或使用FastAPI构建更正式的API服务。部署上线考虑使用Docker容器化你的应用并部署到云服务器或Kubernetes集群。AI Agent的开发是一场结合了软件工程、提示词工程和大模型理解的综合实践。希望这篇教程能成为你探索这个激动人心领域的坚实起点。建议收藏本文并在实际项目中尝试扩展每个模块你将收获远超Demo的实战经验。 从零构建企业级AI Agent：融合RAG、MCP与LangGraph的工程化实战
你是不是也遇到过这样的困惑看到各种AI Agent、RAG、LangChain的技术文章满天飞每个概念都懂一点但真要自己动手搭建一个能用的智能体系统却发现无从下手网上教程要么太零散要么直接甩给你一堆代码根本不解释为什么这么设计。这就是为什么我要写这篇教程。我见过太多开发者卡在“知道概念”到“能实战”的鸿沟里。这篇文章不是简单的概念罗列而是一套从零到一构建企业级AI Agent的完整工程化指南。我会带你亲手搭建一个融合了Agent核心思想、RAG知识库增强、MCP工具扩展以及LangGraph工作流编排的实战项目。读完本文你将彻底搞懂AI Agent的核心组件它不只是调用API而是由规划、记忆、工具使用、反思构成的完整智能体。RAG如何真正落地从文档处理、向量化到检索优化避开“向量数据库即RAG”的误区。MCP协议的价值如何用Model Context Protocol统一且安全地扩展Agent的能力边界。LangChain与LangGraph的分工前者是乐高积木后者是设计图纸如何组合使用。一个可运行的、模块化的企业级项目架构包含代码、配置、部署和调试全流程。我们从一个真实的场景开始构建一个“技术文档智能问答与任务执行Agent”。它不仅能回答你公司内部的技术文档问题RAG还能根据你的指令执行诸如创建JIRA工单、查询数据库、生成代码片段等操作Agent Tools。下面我们就从最根本的问题开始。1. 这篇文章真正要解决的问题为什么你的AI项目总是“Demo级”很多开发者尝试过AI项目但结果往往是本地跑通一个示例很开心一旦想接入真实数据、对接企业系统、处理复杂逻辑项目就迅速变得难以维护和扩展。问题通常出在以下几个方面架构混乱业务逻辑、AI调用、工具集成、状态管理全部混在一起代码像“意大利面条”。缺乏工程化没有清晰的模块划分、配置管理、错误处理和日志监控。对核心概念理解片面以为用了LangChain就是Agent装了向量数据库就是RAG忽略了记忆、规划、反思等关键机制。扩展性差每增加一个新工具或数据源就要大改核心代码。本文要解决的正是如何跨越从“玩具Demo”到“生产可用系统”的鸿沟。我们将采用分层和模块化的设计思想确保每一部分Agent核心、知识库、工具、工作流都能独立开发、测试和替换。无论你是想构建一个内部的效率助手还是一个面向客户的产品级智能体这套架构都能提供坚实的基础。2. 基础概念与核心原理拆解AI Agent的四大支柱在动手之前必须统一认知。一个完整的AI Agent智能体远不止一个“会聊天的AI”。我们可以将其类比为一个高级别的AI工程师实习生它具备以下核心能力核心组件通俗解释技术实现关键类比AI工程师实习生规划 (Planning)拆解复杂任务为可执行的步骤序列。Chain of Thought, ReAct框架 LangGraph的状态机。接到需求后先写技术方案拆解成子任务。工具使用 (Tool Use)调用外部API、数据库、系统命令来获取信息或执行动作。Function Calling, Tool Definition, MCP协议。会使用Git提交代码、用JIRA创建任务、查数据库文档。记忆 (Memory)保留对话历史、任务上下文和学到的知识。短期记忆对话缓存长期记忆向量数据库 LangGraph的持久化状态。记得刚才讨论的架构设计也记得公司项目的通用规范。反思 (Reflection)评估自身行动和结果进行修正和优化。Self-Critique, 验证步骤输出循环Loop控制。代码跑失败了会看日志、分析原因、重新尝试。而RAG、MCP、LangChain/LangGraph是帮助我们实现这些能力的“基础设施”RAG是Agent长期记忆和知识库的核心。它让Agent能“阅读”并理解你提供的私有文档如公司wiki、产品手册从而给出精准回答。其流程是文档加载 → 文本分割 → 向量化嵌入 → 存储到向量数据库 → 用户提问时检索相关片段 → 连同片段和问题交给大模型生成答案。MCP是Agent安全、标准化使用工具的协议。你可以把它想象成智能体的“USB标准接口”。任何工具查天气、操作数据库、调用内部API只要按照MCP协议实现一个Server就能被Agent即插即用无需为每个工具写一遍胶水代码。LangChain是一个丰富的“组件库”提供了连接大模型、处理文档、管理记忆、定义工具等大量标准化模块。它让你不用重复造轮子。LangGraph是一个基于图的“工作流编排引擎”。当任务步骤之间存在复杂依赖、循环或分支时比如先检索、再判断、不行就换关键词再检索用LangChain的简单链Chain会很吃力。LangGraph允许你以可视化或代码的方式定义状态State和节点Node精确控制Agent的执行流是实现复杂规划和反思的关键。理解了这些你就知道我们不是在堆砌时髦名词而是在用正确的工具解决正确的问题。接下来我们开始搭建环境。3. 环境准备与前置条件我们将使用Python作为开发语言。请确保你的环境满足以下要求操作系统Linux/macOS/Windows (WSL2推荐)Python版本3.10 或 3.11这是大多数AI库兼容性最好的版本包管理工具pip或poetry本文使用pip核心依赖langchainlangchain-community: 核心组件库。langgraph: 工作流编排。openai: 调用GPT等模型也可替换为langchain-anthropic,langchain-groq等。向量数据库客户端例如chromadb(轻量)或pinecone(云服务)。嵌入模型例如sentence-transformers(本地)或openai的text-embedding-ada-002(API)。mcpMCP协议Python SDK。fastapiuvicorn: 用于构建简单的工具服务或Agent Web接口。第一步创建项目并安装依赖# 创建项目目录 mkdir ai-agent-tutorial cd ai-agent-tutorial # 创建虚拟环境强烈推荐 python -m venv venv # 激活虚拟环境 # Linux/macOS: source venv/bin/activate # Windows: # venv\Scripts\activate # 安装核心依赖 pip install langchain langgraph langchain-community # 安装OpenAI如果你使用GPT pip install openai # 安装本地向量数据库和嵌入模型 pip install chromadb sentence-transformers # 安装MCP pip install mcp # 安装Web框架可选用于最后演示 pip install fastapi uvicorn第二步准备API密钥如果你使用OpenAI的模型需要设置环境变量。# Linux/macOS export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) # $env:OPENAI_API_KEYyour-api-key-here也可以创建.env文件管理使用python-dotenv库读取。环境就绪现在我们进入核心环节构建一个模块化的企业级AI Agent项目。4. 项目架构设计模块化与分层思想在写代码前先看设计图。一个健壮的系统源于清晰的架构。我们将项目分为以下层次和模块ai-agent-tutorial/ ├── config/ # 配置管理 │ ├── __init__.py │ ├── settings.py # 应用配置API密钥、模型选择等 │ └── prompts.py # 所有提示词模板 ├── core/ # 核心Agent逻辑 │ ├── __init__.py │ ├── agent.py # Agent核心类规划、执行、反思 │ └── state.py # LangGraph状态定义 ├── knowledge/ # RAG知识库模块 │ ├── __init__.py │ ├── loader.py # 文档加载器 │ ├── splitter.py # 文本分割器 │ ├── vector_store.py # 向量数据库封装 │ └── retriever.py # 检索器封装 ├── tools/ # 工具模块MCP及其他 │ ├── __init__.py │ ├── mcp_clients/ # MCP工具客户端 │ │ ├── __init__.py │ │ └── jira_client.py # 示例JIRA工具 │ └── custom_tools.py # 自定义非MCP工具 ├── workflows/ # LangGraph工作流定义 │ ├── __init__.py │ └── main_workflow.py # 主问答与执行工作流 ├── app.py # FastAPI应用入口可选 ├── requirements.txt └── .env.example这个结构的关键在于解耦。knowledge模块只关心文档处理和检索tools模块只关心如何调用外部服务core中的agent负责高层规划和决策workflows用LangGraph把它们粘合起来。任何一部分需要升级或替换影响范围都是可控的。5. 核心流程拆解与实现我们将按照“自底向上”的顺序实现先准备知识库和工具再构建Agent核心最后用工作流串联。5.1 第一步构建RAG知识库模块知识库是Agent的“长期记忆”。我们实现一个简单的本地知识库支持加载Markdown文档。文件knowledge/loader.pyimport os from typing import List from langchain_community.document_loaders import DirectoryLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document class KnowledgeLoader: 文档加载与处理类 def __init__(self, directory_path: str, glob_pattern**/*.md): self.directory_path directory_path self.glob_pattern glob_pattern def load_documents(self) - List[Document]: 加载目录下所有文档 if not os.path.exists(self.directory_path): raise FileNotFoundError(f知识库目录不存在: {self.directory_path}) loader DirectoryLoader( self.directory_path, globself.glob_pattern, loader_clsTextLoader, loader_kwargs{autodetect_encoding: True} ) documents loader.load() print(f已加载 {len(documents)} 个文档) return documents def split_documents(self, documents: List[Document], chunk_size1000, chunk_overlap200) - List[Document]: 分割文档为适合嵌入的片段 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) splits text_splitter.split_documents(documents) print(f文档分割为 {len(splits)} 个片段) return splits文件knowledge/vector_store.pyimport chromadb from chromadb.config import Settings from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings # 或者使用本地嵌入模型 from langchain.embeddings import HuggingFaceEmbeddings from typing import List import os class VectorStoreManager: 向量数据库管理类 def __init__(self, persist_directory: str ./chroma_db, embedding_model_name: str all-MiniLM-L6-v2): self.persist_directory persist_directory # 使用本地嵌入模型避免API调用和费用 self.embeddings HuggingFaceEmbeddings( model_namefsentence-transformers/{embedding_model_name} ) # 或者使用OpenAI嵌入需要API Key # self.embeddings OpenAIEmbeddings() def create_vector_store(self, documents: List): 创建并持久化向量存储 vectordb Chroma.from_documents( documentsdocuments, embeddingself.embeddings, persist_directoryself.persist_directory ) vectordb.persist() print(f向量数据库已创建并保存至: {self.persist_directory}) return vectordb def load_vector_store(self): 加载已存在的向量存储 if not os.path.exists(self.persist_directory): raise FileNotFoundError(f向量数据库目录不存在: {self.persist_directory}) vectordb Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings ) print(f向量数据库已从 {self.persist_directory} 加载) return vectordb def get_retriever(self, vectordb, search_typesimilarity, k4): 获取检索器 return vectordb.as_retriever( search_typesearch_type, search_kwargs{k: k} )使用示例初始化知识库创建一个脚本init_knowledge_base.pyimport sys sys.path.append(.) from knowledge.loader import KnowledgeLoader from knowledge.vector_store import VectorStoreManager def main(): # 1. 加载文档 loader KnowledgeLoader(directory_path./data/docs) # 假设你的文档放在./data/docs下 raw_docs loader.load_documents() # 2. 分割文档 split_docs loader.split_documents(raw_docs) # 3. 创建向量存储 vs_manager VectorStoreManager(persist_directory./chroma_db) vectordb vs_manager.create_vector_store(split_docs) # 4. 获取检索器备用 retriever vs_manager.get_retriever(vectordb) print(知识库初始化完成) if __name__ __main__: main()运行前请在项目根目录创建data/docs文件夹并放入一些Markdown格式的技术文档。运行此脚本后会在本地生成chroma_db目录里面就是向量化后的知识库。5.2 第二步通过MCP协议集成工具MCP的核心思想是标准化。我们以“查询天气”和“创建JIRA工单”为例展示如何集成MCP工具。首先你需要一个MCP Server。这里我们以调用一个公开的天气API为例模拟一个简单的MCP Server。概念澄清在实际生产中工具提供方如JIRA系统管理员会部署标准的MCP Server。作为Agent开发者你通常只需要作为Client去连接它。但为了演示我们写一个最简单的Server。文件tools/mcp_clients/weather_server.py(模拟MCP Server)# 这是一个极度简化的MCP Server示例用于演示概念。 # 真实MCP Server实现需遵循官方协议。 import json from typing import Dict, Any import requests from fastapi import FastAPI import uvicorn app FastAPI(titleMock Weather MCP Server) # 模拟MCP Server的工具列表端点 app.get(/tools) async def list_tools(): 列出此Server提供的工具 return { tools: [ { name: get_weather, description: 获取指定城市的当前天气情况, inputSchema: { type: object, properties: { city: {type: string, description: 城市名称如 Beijing} }, required: [city] } } ] } # 模拟MCP Server的工具调用端点 app.post(/tools/call) async def call_tool(tool_input: Dict[str, Any]): 调用工具 tool_name tool_input.get(name) arguments tool_input.get(arguments, {}) if tool_name get_weather: city arguments.get(city, Beijing) # 这里模拟一个API调用实际应调用真实天气API # 例如response requests.get(fhttps://api.weatherapi.com/v1/current.json?keyYOUR_KEYq{city}) # 为演示我们返回模拟数据 mock_data { city: city, temperature: 22°C, condition: Sunny, humidity: 65% } return { content: [ { type: text, text: f{city}的天气是{mock_data[condition]}温度{mock_data[temperature]}湿度{mock_data[humidity]}。 } ] } else: return {error: f未知工具: {tool_name}} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8001)运行python tools/mcp_clients/weather_server.py启动这个模拟Server。文件tools/mcp_clients/weather_client.py(MCP Client封装)import requests from typing import Dict, Any from langchain.tools import Tool class WeatherMCPClient: 封装对模拟天气MCP Server的调用 def __init__(self, server_url: str http://localhost:8001): self.server_url server_url def get_weather(self, city: str) - str: 调用远程MCP Server获取天气 try: payload { name: get_weather, arguments: {city: city} } response requests.post( f{self.server_url}/tools/call, jsonpayload, timeout10 ) response.raise_for_status() result response.json() # 解析MCP格式的返回内容 if content in result and len(result[content]) 0: return result[content][0][text] else: return f调用成功但返回格式异常: {result} except Exception as e: return f调用天气服务失败: {str(e)} def as_langchain_tool(self) - Tool: 转换为LangChain可用的Tool对象 return Tool( nameget_weather, funcself.get_weather, description获取指定城市的当前天气情况。输入应为城市名称如 Beijing。 ) # 示例如何集成到LangChain def get_weather_tool(): client WeatherMCPClient() return client.as_langchain_tool()这样我们就有了一个标准的、可被Agent调用的“天气工具”。对于JIRA、数据库等工具模式完全一样连接MCP Server → 封装调用 → 转换为LangChain Tool。这保证了工具集的整洁和可扩展性。5.3 第三步定义Agent核心与状态这是智能体的“大脑”。我们使用LangGraph来定义Agent的状态和执行逻辑。文件core/state.pyfrom typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): 定义Agent工作流的状态。 这是LangGraph中在各个节点间传递的上下文。 # 用户输入的问题 input: str # 从知识库检索到的相关文档 retrieved_docs: List[str] # Agent思考的步骤或规划 plan: List[str] # 工具调用的结果 tool_outputs: List[str] # 最终给用户的答案 final_answer: str # 迭代次数用于控制循环 iteration: Annotated[int, operator.add]文件core/agent.pyfrom langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_react_agent from langchain.memory import ConversationBufferMemory from langchain import hub from core.state import AgentState import os class CoreAgent: Agent核心封装规划、执行和工具调用逻辑 def __init__(self, tools, model_namegpt-3.5-turbo): # 初始化大语言模型 self.llm ChatOpenAI( modelmodel_name, temperature0, # 降低随机性使输出更确定 api_keyos.getenv(OPENAI_API_KEY) ) self.tools tools # 从LangChain Hub拉取一个优秀的ReAct提示词模板 # 你也可以自定义放在 config/prompts.py 中 self.prompt hub.pull(hwchase17/react) # 创建具有ReAct推理能力的Agent self.agent create_react_agent( llmself.llm, toolsself.tools, promptself.prompt ) # 创建执行器并开启详细日志便于调试 self.agent_executor AgentExecutor( agentself.agent, toolsself.tools, verboseTrue, handle_parsing_errorsTrue, # 优雅处理解析错误 max_iterations5, # 防止无限循环 early_stopping_methodgenerate # 达到满意结果时提前停止 ) def run(self, input_text: str) - str: 执行单轮问答 try: result self.agent_executor.invoke({input: input_text}) return result.get(output, Agent未返回明确结果。) except Exception as e: return fAgent执行过程中出错: {str(e)}这个CoreAgent类封装了一个基于ReAct框架的、能使用工具的智能体。但它还缺少与知识库RAG的深度集成以及更复杂的工作流如先检索再判断是否需要工具。这正是LangGraph要发挥作用的地方。5.4 第四步用LangGraph编排高级工作流现在我们把知识库检索、工具调用和Agent推理组合成一个智能的工作流。这个工作流能决定用户的问题应该直接查知识库回答还是需要调用工具执行任务文件workflows/main_workflow.pyfrom typing import Literal from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor, ToolInvocation from langchain_core.messages import HumanMessage, AIMessage from core.state import AgentState from knowledge.vector_store import VectorStoreManager from core.agent import CoreAgent import json class MainWorkflow: 主工作流结合RAG检索和工具调用的决策流 def __init__(self, retriever, tools): self.retriever retriever self.tools tools self.tool_executor ToolExecutor(tools) # 初始化一个简单的LLM用于路由决策 from langchain_openai import ChatOpenAI self.router_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 构建图 self.graph self._build_graph() self.compiled_graph self.graph.compile() def _retrieve_node(self, state: AgentState) - AgentState: 节点1从知识库检索相关文档 print(f[检索节点] 正在检索与问题相关的文档: {state[input]}) docs self.retriever.get_relevant_documents(state[input]) state[retrieved_docs] [doc.page_content for doc in docs] print(f[检索节点] 检索到 {len(docs)} 个相关片段。) return state def _route_question(self, state: AgentState) - Literal[answer_from_knowledge, use_agent_tools]: 路由决策判断问题类型 question state[input] docs state[retrieved_docs] # 构建路由提示词 router_prompt f 你是一个智能路由助手。请分析用户的问题和检索到的知识库内容决定下一步操作。 用户问题: {question} 检索到的知识库内容前200字符: {docs[0][:200] if docs else 无相关内容} 请根据以下规则判断 1. 如果知识库内容**明确、直接**回答了用户问题选择 answer_from_knowledge。 2. 如果用户问题涉及**执行具体操作**如创建任务、查询实时信息、计算等或者知识库内容**无法回答**选择 use_agent_tools。 只返回 answer_from_knowledge 或 use_agent_tools 中的一个。 response self.router_llm.invoke(router_prompt) decision response.content.strip().lower() print(f[路由决策] 问题类型判断为: {decision}) return decision def _answer_from_knowledge_node(self, state: AgentState) - AgentState: 节点2基于知识库生成答案 print([知识库回答节点] 正在综合知识库内容生成答案...) question state[input] context \n\n.join(state[retrieved_docs]) # 使用LLM综合上下文生成答案 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo) prompt f 请根据以下提供的上下文信息回答用户的问题。如果上下文信息不足以回答问题请如实告知。 上下文信息 {context} 用户问题{question} 请给出专业、准确的回答 answer llm.invoke(prompt) state[final_answer] answer.content print(f[知识库回答节点] 答案生成完成。) return state def _agent_tools_node(self, state: AgentState) - AgentState: 节点3调用Agent工具来回答问题/执行任务 print([Agent工具节点] 正在调用工具处理复杂任务...) # 这里可以复用之前定义的CoreAgent或者直接调用工具执行器 # 简化演示我们直接让一个Agent执行器来处理 from core.agent import CoreAgent agent CoreAgent(toolsself.tools) result agent.run(state[input]) state[final_answer] result print(f[Agent工具节点] 工具调用完成。) return state def _build_graph(self): 构建LangGraph图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(retrieve, self._retrieve_node) workflow.add_node(answer_from_knowledge, self._answer_from_knowledge_node) workflow.add_node(agent_tools, self._agent_tools_node) # 设置入口点 workflow.set_entry_point(retrieve) # 设置条件边检索后根据路由决策走不同分支 workflow.add_conditional_edges( retrieve, self._route_question, { answer_from_knowledge: answer_from_knowledge, use_agent_tools: agent_tools, } ) # 设置终点 workflow.add_edge(answer_from_knowledge, END) workflow.add_edge(agent_tools, END) return workflow def run(self, question: str) - dict: 运行工作流 initial_state: AgentState { input: question, retrieved_docs: [], plan: [], tool_outputs: [], final_answer: , iteration: 0 } final_state self.compiled_graph.invoke(initial_state) return final_state这个工作流实现了智能路由无论什么问题先检索知识库。用一个轻量级LLM判断知识库能直接回答吗如果能走分支A合成答案如果不能或需要执行操作走分支B启动能使用工具的Agent。这比单一的RAG或单一的Agent更强大、更灵活。6. 运行结果与效果验证让我们把所有模块组装起来进行端到端的测试。文件run_agent.py(主运行脚本)import sys sys.path.append(.) from knowledge.vector_store import VectorStoreManager from tools.mcp_clients.weather_client import get_weather_tool from workflows.main_workflow import MainWorkflow def main(): print( 启动企业级AI Agent系统 ) # 1. 加载知识库 print(1. 加载向量知识库...) vs_manager VectorStoreManager(persist_directory./chroma_db) try: vectordb vs_manager.load_vector_store() retriever vs_manager.get_retriever(vectordb) print( 知识库加载成功。) except Exception as e: print(f 知识库加载失败: {e}。请先运行 init_knowledge_base.py 初始化。) return # 2. 准备工具集 print(2. 初始化工具集...) weather_tool get_weather_tool() # 这里可以添加更多工具如JIRA工具、数据库查询工具等 tools [weather_tool] print(f 已加载 {len(tools)} 个工具。) # 3. 初始化工作流 print(3. 初始化主工作流...) workflow MainWorkflow(retrieverretriever, toolstools) print( 系统初始化完成\n) # 4. 交互式问答 print(请输入您的问题输入 quit 退出:) while True: user_input input(\n ) if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input.strip(): continue print(\n *50) print(f[用户问题] {user_input}) print(*50) try: result workflow.run(user_input) print(f\n[系统回答] {result.get(final_answer, 未生成答案)}) except Exception as e: print(f\n[错误] 处理问题时发生异常: {e}) if __name__ __main__: main()运行与验证启动知识库确保已运行init_knowledge_base.py并生成了./chroma_db。启动模拟工具Server在另一个终端运行python tools/mcp_clients/weather_server.py。运行主程序python run_agent.py。预期交互示例 启动企业级AI Agent系统 1. 加载向量知识库... 知识库加载成功。 2. 初始化工具集... 已加载 1 个工具。 3. 初始化主工作流... 系统初始化完成 请输入您的问题输入 quit 退出: 我们公司的API网关超时设置默认是多少秒 [用户问题] 我们公司的API网关超时设置默认是多少秒 [检索节点] 正在检索与问题相关的文档: 我们公司的API网关超时设置默认是多少秒 [检索节点] 检索到 3 个相关片段。 [路由决策] 问题类型判断为: answer_from_knowledge [知识库回答节点] 正在综合知识库内容生成答案... [知识库回答节点] 答案生成完成。 [系统回答] 根据公司技术文档API网关的默认超时设置为30秒。如果您的服务需要更长处理时间建议在网关配置中单独调整该路由的超时参数。 今天北京天气怎么样 [用户问题] 今天北京天气怎么样 [检索节点] 正在检索与问题相关的文档: 今天北京天气怎么样 [检索节点] 检索到 1 个相关片段。 [路由决策] 问题类型判断为: use_agent_tools [Agent工具节点] 正在调用工具处理复杂任务... Entering new AgentExecutor chain... 我需要调用天气查询工具来获取北京今天的天气。 Action: get_weather Action Input: {city: Beijing} Observation: Beijing的天气是Sunny温度22°C湿度65%。 Thought:我已经获取到了北京的天气信息可以给出最终答案了。 Final Answer: 北京今天的天气是晴天气温22摄氏度湿度65%。 [Agent工具节点] 工具调用完成。 [系统回答] 北京今天的天气是晴天气温22摄氏度湿度65%。你可以看到系统能自动判断问题类型并选择最合适的路径来回答。对于知识库已有的信息它直接检索回答对于需要实时信息的天气查询它成功调用了MCP工具。7. 常见问题与排查思路在实践过程中你可能会遇到以下问题问题现象可能原因排查方式解决方案运行init_knowledge_base.py时报No module named langchain虚拟环境未激活或依赖未安装。在终端输入pip list | grep langchain。激活虚拟环境并运行pip install -r requirements.txt。知识库检索结果不相关1. 文档分割块太大或太小。2. 嵌入模型不适合中文或领域文本。3. 检索参数k不合适。1. 检查分割后的片段长度。2. 尝试不同的嵌入模型。3. 调整检索器的k值。1. 调整chunk_size和chunk_overlap。2. 换用text-embedding-ada-002或paraphrase-multilingual模型。3. 尝试similarity_score_threshold检索。Agent陷入循环不输出最终答案1.max_iterations设置过高。2. 工具描述不清晰导致LLM无法正确选择。3. ReAct提示词不适合当前任务。1. 观察Agent执行日志看它在重复什么动作。2. 检查工具的描述是否准确。1. 适当降低max_iterations。2. 优化工具的描述使其更精确。3. 尝试不同的提示词模板或自定义提示词。调用MCP工具超时或失败1. MCP Server未启动或地址错误。2. 网络问题。3. 请求/响应格式不符合MCP协议。1. 用curl或Postman直接测试MCP Server端点。2. 检查客户端代码中的URL和端口。1. 确保MCP Server正常运行。2. 检查防火墙和网络配置。3. 对照MCP官方协议检查数据格式。LangGraph工作流编译错误1. 状态State定义与节点返回值不匹配。2. 图的结构有误如未连接END。1. 仔细检查AgentState的字段类型。2. 使用workflow.get_graph().draw_mermaid()可视化图结构。1. 确保每个节点都返回一个完整的、或部分更新的State字典。2. 确保所有路径最终都指向END。生产环境部署后性能差1. 向量检索未使用索引优化。2. LLM API调用延迟高。3. 未做缓存。1. 监控各环节耗时。2. 使用异步调用。3. 考虑更高效的向量数据库。1. 对频繁查询的向量数据建立索引。2. 对LLM回答和检索结果进行缓存。3. 考虑使用更快的本地模型或优化后的API模型。8. 最佳实践与工程建议要将这个Demo升级为生产级系统你需要关注以下几点配置中心化将所有API密钥、模型名称、服务器地址等配置项移出代码使用环境变量或配置管理工具如pydantic-settings管理。日志与监控为关键节点检索、工具调用、LLM请求添加结构化日志。监控Token消耗、响应时间和错误率。缓存策略LLM缓存对相同提示词的LLM请求进行缓存节省成本和时间。向量检索缓存对常见问题的检索结果进行缓存。可以使用langchain.cache或redis实现。异步优化对于IO密集型操作如调用多个工具、网络请求使用异步asyncio来提升并发性能。测试与评估单元测试测试每个独立模块如文档加载、工具调用。集成测试测试整个工作流。评估体系定义评估指标如答案相关性、事实准确性、工具调用正确率并定期用测试集进行评估。安全与权限工具权限不是所有用户都能调用所有工具。实现基于用户或角色的工具权限控制。输入输出过滤对用户输入和模型输出进行安全检查防止Prompt注入或输出有害内容。数据脱敏确保知识库和工具调用不泄露敏感信息。可观测性考虑集成像LangSmith这样的平台它可以可视化跟踪每个LangChain/LangGraph调用的详细步骤、耗时和中间结果是调试和优化Agent的利器。版本管理对提示词、工作流图、知识库文档进行版本控制便于回滚和A/B测试。9. 总结与后续学习方向通过这个完整的项目我们实现了一个模块化、可扩展、具备决策能力的企业级AI Agent原型。它不仅仅是调用API而是融合了RAG知识、MCP工具、LangGraph流程的智能系统。本文带你走通了几个关键环节清晰的概念理解明白了Agent、RAG、MCP、LangChain、LangGraph各自扮演的角色。工程化的项目结构学会了如何用分层的代码结构来组织复杂的AI应用。核心模块的实现亲手实现了知识库构建、MCP工具封装、Agent核心和工作流编排。智能路由决策用LangGraph构建了一个能自动选择“查知识库”还是“调用工具”的智能工作流。如果你想继续深入可以从以下几个方向探索更复杂的工具尝试集成真实的JIRA、Confluence、数据库或GitLab的MCP Server让Agent能执行更真实的办公任务。高级记忆机制实现对话历史记忆短期以及基于用户反馈更新知识库的能力长期学习。多Agent协作使用LangGraph创建多个具有不同专长的Agent如一个负责分析一个负责执行一个负责检查让它们协同完成复杂项目。前端界面使用Gradio或Streamlit快速构建一个Web界面或使用FastAPI构建更正式的API服务。部署上线考虑使用Docker容器化你的应用并部署到云服务器或Kubernetes集群。AI Agent的开发是一场结合了软件工程、提示词工程和大模型理解的综合实践。希望这篇教程能成为你探索这个激动人心领域的坚实起点。建议收藏本文并在实际项目中尝试扩展每个模块你将收获远超Demo的实战经验。