LangChain与LangGraph实战:从零构建企业级AI智能体应用

发布时间:2026/8/25 18:50:32
LangChain与LangGraph实战:从零构建企业级AI智能体应用 这次我们来看一套专注于 LangChain 和 LangGraph 的实战视频教程。这套教程号称是“B站讲得最好的”实战课程内容覆盖了 Agent智能体、RAG检索增强生成、MCP模型上下文协议等当前最热门的技术栈。对于想要从零开始构建企业级 AI 应用或者希望深入理解 LangGraph 如何编排复杂 Agent 工作流的开发者来说这是一个非常值得关注的系统性学习资源。这套教程的核心价值在于“实战”。它不满足于仅仅讲解概念而是通过具体的项目案例带你一步步搭建可运行的智能体系统。课程重点涵盖了如何将 LangChain 的基础组件与 LangGraph 的流程编排能力结合实现具备长期记忆、工具调用和复杂决策能力的多智能体应用并深入整合 RAG 和 MCP 协议来扩展应用的能力边界。无论你是想开发一个能自动处理文档的智能助手还是构建一个可以协调多个 AI 模型完成复杂任务的 Agent 系统这套教程提供的路径都极具参考意义。本文将为你深度拆解这套教程可能涵盖的核心内容、技术栈要点以及自学实践的关键路径。我们将重点关注1课程的核心技术模块与学习路线2本地开发环境的搭建与工具链准备3基于 LangGraph 构建多智能体的实战流程4集成 RAG 与 MCP 来增强 Agent 能力的实现方法5项目部署与调试的常见问题。即使没有直接观看原视频你也可以根据本文梳理的框架和实操要点在自己的开发环境中复现核心功能构建起属于自己的 AI Agent 应用。1. 核心能力速览教程内容覆盖本教程并非一个可直接部署的软件而是一套教育内容。因此下表梳理的是教程预计覆盖的技术能力和实战项目特点帮助你判断其学习价值。能力项说明与教程重点核心教学栈LangChain应用框架 LangGraph工作流编排 LLM如 OpenAI GPT, Ollama 本地模型关键技术主题Agent智能体开发、RAG检索增强生成系统、MCP模型上下文协议集成、多智能体协作实战项目类型本地知识库问答助手、自动化任务处理智能体、多工具协调 Agent、具备长期记忆的对话系统开发环境门槛主流操作系统Win/Mac/Linux、Python 环境、能访问 API 或运行本地大模型需一定显存/内存学习前置要求基础 Python 编程能力、对 AI 应用开发有基本了解、熟悉命令行操作课程产出物可运行的 Python 项目代码、配置好的工作流、部署可用的 AI Agent 应用原型适合人群AI 应用开发者、全栈工程师、对构建企业级智能助手感兴趣的技术人员、LangChain 初学者与进阶者2. 适用场景与使用边界这套教程的目标是赋能开发者构建能够解决实际问题的 AI 应用。以下是其主要的适用场景企业级智能助手开发学习如何利用 RAG 构建连接内部知识库如产品手册、公司制度的问答机器人提供精准、有据可查的答案。自动化工作流编排掌握使用 LangGraph 将多个 AI 步骤如分析、决策、执行和工具调用如搜索、数据库查询、发送邮件串联起来实现自动化流程。复杂任务分解与协作构建多智能体Multi-Agent系统让不同的 Agent 专精于特定任务如一个负责分析需求一个负责编写代码一个负责检查结果并协同完成复杂项目。集成第三方工具与服务通过 MCP 协议或自定义工具让 Agent 能够安全、可控地访问外部系统如 Figma、数据库、Jira极大扩展 AI 的能力边界。原型验证与快速迭代为你的 AI 产品想法提供一个快速实现和验证的技术框架降低从概念到可运行原型的技术门槛。使用边界与注意事项非“一键部署”产品这是一套教程你需要具备学习能力和动手实践意愿跟随步骤编写和调试代码。依赖底层 LLM 能力最终应用的效果很大程度上取决于所选用的基础大模型如 GPT-4、Claude 或本地模型的能力和成本。数据安全与隐私在构建 RAG 系统或让 Agent 访问内部数据时必须高度重视数据安全。确保知识库文档已脱敏API 密钥妥善管理避免敏感信息泄露。工具调用风险当 Agent 被授权调用外部工具如发送邮件、修改数据时必须设计严格的权限控制和操作确认机制防止未经授权的操作。3. 环境准备与前置条件在开始跟随教程实践之前你需要准备好以下开发环境。教程可能会提供更详细的配置说明但以下是一个通用的、必须的基础清单。操作系统Windows 10/11, macOS, 或 Linux 发行版如 Ubuntu。确保系统有足够的磁盘空间存放项目代码和可能的模型文件。Python 环境推荐使用 Python 3.10 或 3.11。避免使用 Python 3.12 的早期版本以免遇到某些库的兼容性问题。强烈建议使用虚拟环境如venv或conda来隔离项目依赖。代码编辑器或 IDEVisual Studio Code (VSCode) 是绝佳选择配合 Python 扩展。PyCharm 或其他你熟悉的 IDE 亦可。版本控制安装 Git用于克隆教程可能提供的示例代码库和管理你自己的项目。包管理工具确保pip已更新至最新版。大模型访问权限方案一API 调用你需要准备一个或多个大模型的 API Key。最常见的是OpenAI API Key用于 GPT 系列模型。也可能涉及 Anthropic Claude、Google Gemini 等。准备好相应的账户和额度。方案二本地运行如果你希望完全本地运行需要部署本地大模型服务例如使用Ollama。这要求你的机器具备足够的内存通常 8GB 用于 7B 参数模型和显存如果使用 GPU 加速。对于教程入门使用 API 方式更为简便和稳定。网络环境如果使用云端 API需要保证能稳定访问相关服务。4. 安装部署与启动方式由于是教程不存在统一的“启动命令”但每个实战项目都会有类似的初始化步骤。以下是开始一个典型 LangChain LangGraph 项目的通用流程。步骤 1创建项目并初始化虚拟环境# 创建项目目录 mkdir langgraph-agent-tutorial cd langgraph-agent-tutorial # 创建 Python 虚拟环境以 venv 为例 python -m venv venv # 激活虚拟环境 # Windows (PowerShell) .\venv\Scripts\Activate.ps1 # Linux/macOS source venv/bin/activate步骤 2安装核心依赖库教程的核心将围绕以下几个库展开通过pip安装。# 安装 LangChain 全家桶核心 pip install langchain langchain-community langchain-core # 安装 LangGraph用于构建有状态的、多步骤的工作流 pip install langgraph # 安装用于向量数据库和文本拆分的库RAG 必需 pip install chromadb langchain-chroma tiktoken # 安装 OpenAI 库如果使用 GPT 系列模型 pip install openai # 安装环境变量管理库用于安全存储 API Key pip install python-dotenv步骤 3配置环境变量在项目根目录创建.env文件用于存储敏感信息切勿提交到 Git。# .env 文件示例 OPENAI_API_KEYsk-your-openai-api-key-here # 其他可能的 API Key如 ANTHROPIC_API_KEY, GROQ_API_KEY 等在代码中通过dotenv加载from dotenv import load_dotenv load_dotenv() # 这会从 .env 文件加载环境变量 import os openai_api_key os.getenv(OPENAI_API_KEY)步骤 4获取并运行示例代码教程可能会提供一个 GitHub 仓库。使用 Git 克隆代码并运行主程序。# 假设教程代码仓库地址 git clone https://github.com/example/langgraph-tutorial.git cd langgraph-tutorial # 安装项目特定依赖通常有一个 requirements.txt pip install -r requirements.txt # 运行主示例脚本 python main.py如果教程是分章节的则每个章节可能是一个独立的脚本或 Jupyter Notebook。5. 功能测试与效果验证教程的实战部分会引导你逐步构建并测试多个核心功能。以下是你可以预期学习和验证的关键模块。5.1 基础 LangChain 链与提示工程测试目的验证环境配置正确能够调用 LLM 并得到基础响应。操作步骤编写一个简单的脚本使用 LangChain 的ChatOpenAI或ChatOllama模型。构建一个提示模板PromptTemplate询问一个简单问题。通过LCELLangChain Expression Language将提示和模型组合成链chain并调用。输入示例from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate model ChatOpenAI(modelgpt-4o-mini, temperature0) prompt ChatPromptTemplate.from_template(请用一句话解释什么是{concept}) chain prompt | model response chain.invoke({concept: RAG}) print(response.content)预期结果成功输出一个关于 RAG 的简洁解释。这证明你的 API 密钥、网络和基础库工作正常。5.2 构建第一个简单 Agent工具调用测试目的让 AI 学会使用一个自定义工具例如计算器或网络搜索。操作步骤使用tool装饰器定义一个 Python 函数作为工具如计算平方。将工具绑定到 LLM。使用create_react_agent或类似方法创建一个 Agent 执行器。向 Agent 提出一个需要用到该工具的问题。输入示例from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import tool from langchain_openai import ChatOpenAI tool def square_calculator(number: float) - float: 计算一个数的平方。 return number ** 2 tools [square_calculator] model ChatOpenAI(modelgpt-4o-mini, temperature0) agent create_react_agent(model, tools) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) result agent_executor.invoke({input: 请问 12.5 的平方是多少}) print(result[output])预期结果Agent 应能识别出需要调用square_calculator工具并正确输出156.25。控制台的verbose日志会显示其思考过程。5.3 使用 LangGraph 构建多步骤工作流测试目的实现一个包含条件判断和多个节点的工作流例如一个审批流程模拟。操作步骤定义工作流的状态StateGraph。创建多个节点Node每个节点是一个函数处理状态的一部分。定义边Edge决定流程如何根据条件在不同节点间流转。编译图并运行。输入示例一个简易的质数判断流程from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END import operator class AgentState(TypedDict): number: int is_prime: bool message: str def check_even(state: AgentState): if state[number] % 2 0 and state[number] 2: return {is_prime: False, message: 是大于2的偶数不是质数。} return state def check_small_number(state: AgentState): if state[number] in [2, 3, 5, 7]: return {is_prime: True, message: 是小于10的质数。} return state def default_check(state: AgentState): # 简化逻辑实际需要更复杂的算法 return {is_prime: True, message: 经过简单检查可能是质数需进一步验证。} workflow StateGraph(AgentState) workflow.add_node(check_even, check_even) workflow.add_node(check_small, check_small_number) workflow.add_node(default, default_check) workflow.set_entry_point(check_even) workflow.add_conditional_edges( check_even, lambda x: END if not x.get(is_prime, True) else check_small ) workflow.add_edge(check_small, default) workflow.add_edge(default, END) app workflow.compile() result app.invoke({number: 9}) print(result)预期结果根据输入的数字工作流会经过不同的节点并输出对应的判断结果和消息。这验证了 LangGraph 编排复杂逻辑的能力。5.4 实现 RAG 问答系统测试目的从本地文档构建知识库并实现基于知识库的精准问答。操作步骤准备知识文档如 TXT、PDF、MD 文件。使用文本分割器RecursiveCharacterTextSplitter将文档切块。使用嵌入模型OpenAIEmbeddings或OllamaEmbeddings将文本块转换为向量。将向量存入向量数据库如Chroma。构建一个检索链用户问题 - 检索相关文档块 - 组合上下文 - LLM 生成答案。输入示例核心代码片段from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_chroma import Chroma from langchain.chains import create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain from langchain_core.prompts import ChatPromptTemplate # 1. 加载与分割文档 loader TextLoader(./knowledge.txt) docs loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) splits text_splitter.split_documents(docs) # 2. 创建向量存储 embeddings OpenAIEmbeddings() vectorstore Chroma.from_documents(documentssplits, embeddingembeddings, persist_directory./chroma_db) retriever vectorstore.as_retriever() # 3. 构建提示模板和链 prompt ChatPromptTemplate.from_template( 基于以下上下文回答问题。如果你不知道答案就说不知道。 上下文{context} 问题{input} ) llm ChatOpenAI(modelgpt-4o-mini) question_answer_chain create_stuff_documents_chain(llm, prompt) rag_chain create_retrieval_chain(retriever, question_answer_chain) # 4. 提问 response rag_chain.invoke({input: 你们公司的产品保修期是多久}) print(response[answer])预期结果系统应从knowledge.txt文档中检索到与“保修期”相关的信息并生成准确的答案而不是依赖 LLM 的通用知识。5.5 集成 MCP 服务器以文件系统为例测试目的让 Agent 能够通过 MCP 协议安全地访问外部资源如读取本地文件列表。操作步骤了解 MCP 协议概念。教程可能会使用mcp客户端库或演示如何与 MCP 服务器通信。启动或连接一个 MCP 服务器例如一个暴露了文件系统读取工具的服务器。在 LangChain/LangGraph 中将这些 MCP 工具作为普通Tool集成到 Agent 中。测试 Agent 调用这些工具的能力。输入示例概念性代码具体依赖教程实现# 假设已配置好 MCP 客户端并获取了工具列表 from langchain.tools import Tool from mcp_client import Client # 连接 MCP 服务器 async with Client.connect_to_server(...) as client: # 获取服务器提供的工具 mcp_tools_descriptions await client.list_tools() # 将 MCP 工具包装成 LangChain Tool 对象 langchain_tools [] for tool_desc in mcp_tools_descriptions: async def tool_func(**kwargs): result await client.call_tool(tool_desc.name, argumentskwargs) return result langchain_tools.append(Tool(nametool_desc.name, descriptiontool_desc.description, functool_func)) # 使用这些工具创建 Agent # ... (后续步骤与 5.2 类似)预期结果Agent 能够成功调用由 MCP 服务器提供的工具例如列出指定目录下的文件。这验证了 Agent 能力通过标准化协议进行扩展。6. 接口 API 与批量任务一个成熟的 AI Agent 应用最终需要以服务的形式提供能力。教程的高级部分可能会涉及如何将构建好的智能体封装成 API 服务并处理批量任务。6.1 使用 FastAPI 封装 Agent 为 Web API启动方式创建一个main.py文件使用 FastAPI 框架。核心代码示例from fastapi import FastAPI, HTTPException from pydantic import BaseModel from your_agent_module import create_agent_executor # 导入你构建的 Agent app FastAPI(titleLangGraph Agent API) agent_executor create_agent_executor() # 初始化你的智能体 class QueryRequest(BaseModel): input: str session_id: str | None None # 用于支持多轮对话会话 app.post(/chat) async def chat_with_agent(request: QueryRequest): try: result agent_executor.invoke({input: request.input}) return {response: result[output], session_id: request.session_id} except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)请求与响应请求POST /chatBody 为 JSON{input: 你的问题, session_id: abc123}响应{response: 智能体的回答, session_id: abc123}6.2 批量任务处理对于需要处理大量文档或问题的场景需要设计批量任务队列。实现思路目录扫描编写脚本扫描./batch_input/目录下的所有文件如.txt,.pdf。任务队列使用concurrent.futures的ThreadPoolExecutor或ProcessPoolExecutor实现并发处理或者使用更专业的任务队列如CeleryRedis。结果保存将每个任务的处理结果如摘要、答案保存到对应的输出文件或数据库中。简单并发示例import os from concurrent.futures import ThreadPoolExecutor from your_processing_function import process_single_file input_dir ./batch_input output_dir ./batch_output os.makedirs(output_dir, exist_okTrue) def process_file(filename): input_path os.path.join(input_dir, filename) output_path os.path.join(output_dir, fresult_{filename}) result process_single_file(input_path) # 调用你的处理逻辑 with open(output_path, w, encodingutf-8) as f: f.write(result) return output_path file_list [f for f in os.listdir(input_dir) if f.endswith(.txt)] with ThreadPoolExecutor(max_workers4) as executor: # 控制并发数 futures [executor.submit(process_file, f) for f in file_list] for future in concurrent.futures.as_completed(futures): print(f处理完成: {future.result()})7. 资源占用与性能观察开发过程中的性能关注点与运行一个本地大模型服务不同更多集中在代码执行效率、API 调用成本和响应延迟上。API 调用成本与延迟观察点每次调用agent_executor.invoke()时背后可能产生多次 LLM API 调用思考、工具调用、总结。在verboseTrue模式下控制台会打印每次调用你可以清楚看到 Token 消耗和步骤数。优化建议优化提示词减少不必要的思考循环为工具调用设置超时和重试对于简单任务考虑使用更便宜、更快的模型如gpt-4o-mini代替gpt-4o。RAG 检索性能观察点向量数据库检索速度。当知识库文档很大时检索可能成为瓶颈。优化建议调整文本分割的chunk_size和chunk_overlap为向量数据库建立索引考虑使用更高效的嵌入模型或本地嵌入。LangGraph 工作流复杂度观察点工作流中节点的数量、条件分支的复杂度。过于复杂的图可能难以调试和维护。优化建议将大图拆分为多个子图使用StateGraph的检查点Checkpoint功能来持久化状态支持长时间运行和恢复。内存占用观察点如果使用本地模型如通过 Ollama则需要关注模型加载后的内存和显存占用。使用nvidia-smiGPU或任务管理器内存进行监控。优化建议根据硬件条件选择合适尺寸的模型对于纯 CPU 推理确保有足够的交换空间。8. 常见问题与排查方法在学习和实践过程中你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案导入 LangChain 库失败提示缺少模块1. 未安装特定子包如langchain-openai,langchain-chroma。2. 虚拟环境未激活或包版本冲突。1. 检查错误信息中缺失的模块名。2. 运行pip list | grep langchain查看已安装包。1. 使用pip install langchain-openai等命令安装指定子包。2. 确保在正确的虚拟环境中并尝试pip install --upgrade或根据requirements.txt重装。运行 Agent 时报错OpenAI API相关1. API Key 未设置或错误。2. 网络问题无法访问 API。3. 账户余额不足或速率限制。1. 检查.env文件是否正确加载环境变量名是否正确。2. 使用curl或ping测试网络连通性。3. 登录 OpenAI 后台检查用量和额度。1. 确认load_dotenv()在代码开头执行且 Key 无误。2. 配置网络代理如需。3. 充值或等待限制解除或切换备用 API。RAG 系统返回的答案与知识库无关1. 文档未正确加载或分割。2. 向量数据库检索失败或未命中。3. 嵌入模型不匹配或效果差。1. 打印splits变量检查文档内容是否被正确分割。2. 检查retriever.get_relevant_documents(“你的问题”)返回的结果。3. 尝试不同的chunk_size或嵌入模型。1. 确保文档路径正确文本分割参数合理。2. 调整检索器的search_kwargs如k值。3. 更换嵌入模型或对检索结果进行重排序rerank。LangGraph 工作流陷入死循环或状态错误1. 条件边conditional_edges的逻辑判断有误。2. 节点函数未正确更新状态。3. 图的结构定义有循环依赖。1. 在节点函数中添加print语句输出状态变化。2. 使用app.get_graph().draw_mermaid_png()需安装pygraphviz可视化图结构检查流向。1. 仔细检查条件边的判断函数确保其返回正确的下一个节点名或END。2. 确保每个节点都返回一个字典来更新状态。3. 简化图结构避免循环或使用检查点中断。工具调用失败Agent 无法使用工具1. 工具函数定义不符合tool装饰器要求。2. 工具描述description不够清晰LLM 无法理解何时调用。3. Agent 类型选择不当。1. 检查工具函数的参数是否有类型注解文档字符串是否清晰。2. 让 LLM 描述它认为工具是做什么的看是否与你的意图一致。3. 尝试使用create_react_agent或create_openai_tools_agent等不同的 Agent 创建方式。1. 严格按照 LangChain 工具定义规范编写函数。2. 优化工具描述明确指出工具的输入、输出和适用场景。3. 参考官方文档为你的任务选择合适的 Agent 类型。批量处理时程序崩溃或内存溢出1. 并发数 (max_workers) 设置过高。2. 单个任务处理消耗资源过大。3. 未正确处理异常导致整个程序停止。1. 监控系统资源CPU、内存使用情况。2. 在单个任务函数内部添加 try-catch记录错误日志而不中断全局。1. 降低并发数或使用ProcessPoolExecutor隔离内存。2. 优化单个任务的处理逻辑例如流式处理大文件。3. 在任务函数和主循环中都做好异常捕获和日志记录。9. 最佳实践与使用建议基于这套教程的学习路径和项目开发经验以下建议能帮助你更高效、更稳健地构建 AI Agent 应用从简到繁迭代开发不要一开始就设计极其复杂的多智能体图。从一个能跑通的简单链Chain开始逐步添加工具、引入状态、拆分成多个节点。每步都进行充分测试。善用日志与可视化始终在创建 Agent 或图时设置verboseTrue这能让你清晰看到 LLM 的思考过程、工具调用和状态流转。对于 LangGraph可视化工具是调试神器。提示词工程是核心Agent 的表现很大程度上取决于给它的指令System Prompt和工具描述。花时间精心设计提示词明确角色、规则和输出格式。可以使用PromptTemplate进行模块化管理。为工具调用设置“护栏”特别是当工具能执行写操作如发邮件、写文件时一定要在工具函数内部进行参数验证、权限检查和操作确认。可以考虑让工具返回一个待确认的操作计划由用户或另一个监督 Agent 批准后再执行。管理好配置与密钥所有 API Key、数据库连接字符串等敏感信息必须通过.env文件管理并确保.env在.gitignore中。对于不同的环境开发、测试、生产使用不同的配置文件。设计可观测性在生产环境中不仅要记录结果还要记录 AI 的决策链路Chain of Thought。这有助于分析错误、优化流程和审计。可以考虑集成像LangSmith这样的追踪平台。性能与成本平衡在项目初期就关注 API 调用成本和响应时间。对于内部知识库问答可以缓存常见的问答对。对于复杂工作流评估是否所有步骤都需要调用最强大的也是最贵的模型。法律与伦理合规确保你的 Agent 应用遵守数据隐私法规如 GDPR。如果处理用户数据需明确告知并获得同意。避免构建可能用于生成虚假信息、进行欺诈或侵犯他人权益的 Agent 系统。这套 LangChain LangGraph 实战教程为你提供了一条从入门到构建高级 AI Agent 应用的清晰路径。其最大的价值在于将抽象的概念Agent, RAG, MCP与具体的、可运行的代码相结合。通过跟随教程你不仅能理解这些技术如何工作更能掌握如何将它们组合起来解决真实世界的问题。建议你按照“环境搭建 - 基础链与 Agent - LangGraph 工作流 - RAG 集成 - MCP 扩展 - API 封装”的顺序进行学习与实践。过程中最可能遇到的挑战不是代码语法而是对 AI 决策逻辑的设计和调试。多利用verbose输出和可视化工具像调试普通程序一样耐心地观察和调整 AI 的“思考”过程。当你成功构建出第一个能自动处理特定任务的智能体时你会发现AI 应用的开发范式已经发生了根本性的改变。接下来的方向可以是深入优化现有 Agent 的可靠性和效率探索更复杂的多智能体协作模式或是将你的智能体接入更多的业务系统和数据源真正释放其生产力价值。