1. 先搞清楚 LangChain 和 LangGraph 到底能帮你解决什么问题如果你正在找一套能直接上手、把大语言模型LLM和你的数据、工具、业务流程结合起来的方案那 LangChain 和 LangGraph 就是目前最值得投入时间学习的框架之一。它们不是另一个需要你从零搭建的 AI 模型而是一个“连接器”和“编排器”核心价值在于帮你把 OpenAI、Anthropic、本地模型这些 LLM 能力和你自己的文档、数据库、API 接口、乃至复杂的多步骤任务逻辑高效、稳定地串联起来。很多人一上来就被“RAG”、“智能体”这些术语吓住或者陷入到无穷无尽的 API 调用细节里。其实LangChain 要解决的核心痛点非常具体当你有一个强大的 LLM但需要它基于你的私有数据回答问题RAG或者需要它按特定流程调用工具完成任务智能体时LangChain 提供了一套标准化的、可复用的组件和模式让你不用重复造轮子。而 LangGraph 则是在 LangChain 基础上专门为构建有状态、可循环、多角色协作的复杂智能体系统而设计的。所以这套教程的价值不在于教你最前沿的 AI 理论而在于提供一条从零到一构建可用应用的清晰路径。它适合已经了解 Python 基础、对 LLM API 有初步接触比如调用过 ChatGPT API但不知道如何将其工程化、产品化的开发者、产品经理或技术爱好者。学完之后你应该能独立搭建一个简单的文档问答机器人或者设计一个能自动执行多步骤任务的智能体流程。2. 学习前的环境准备与核心依赖确认在开始跟着任何教程敲代码之前先把环境理顺能避免至少一半的“跑不通”问题。LangChain 生态迭代很快但核心依赖相对稳定。基础环境要求Python: 推荐使用 Python 3.8 到 3.11 版本。3.12 及以上版本可能存在一些第三方库的兼容性问题新手建议先避开。包管理工具: 强烈建议使用pip配合virtualenv或conda创建独立的虚拟环境。这是保证项目依赖不冲突的最佳实践。代码编辑器: VS Code 或 PyCharm 均可确保有好的 Python 插件支持。核心依赖安装打开终端在你的项目虚拟环境中执行以下命令安装最核心的包pip install langchain langchain-community langchain-core这行命令安装了 LangChain 的核心框架、社区贡献的第三方集成以及核心抽象。这是构建大多数应用的基础。关键环境变量API KeysLangChain 本身不提供模型你需要接入一个 LLM 服务。对于学习和快速验证OpenAI 的 API 是最常见的选择。前往 OpenAI 平台注册并获取 API Key。在命令行中临时设置环境变量每次新开终端都需要export OPENAI_API_KEY你的-api-key或者在项目根目录创建.env文件写入OPENAI_API_KEY你的-api-key并使用python-dotenv包在代码中加载。这是更安全、更工程化的做法。pip install python-dotenv openai验证安装创建一个简单的test_env.py文件写入以下代码import os from langchain_openai import ChatOpenAI from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 llm ChatOpenAI(modelgpt-3.5-turbo) # 使用 gpt-3.5-turbo 验证成本更低 response llm.invoke(你好请用一句话介绍你自己。) print(response.content)如果能正常收到 LLM 的回复说明基础环境和 API 连接成功。如果报错优先检查OPENAI_API_KEY是否正确设置且有效。网络连接是否正常能否访问 OpenAI API。Python 和 pip 版本是否匹配。3. 构建你的第一个 RAG 应用从文档加载到智能问答RAG检索增强生成是 LangChain 最经典的应用场景。它的流程可以简化为加载你的文档 - 切分成片段 - 转换成向量并存储 - 提问时检索相关片段 - 连同问题和片段一起交给 LLM 生成答案。下面我们拆解每一步。3.1 文档加载与文本分割LangChain 提供了大量的DocumentLoader支持从 TXT、PDF、PPT、网页、Notion 等来源加载文档。这里以本地 TXT 文件为例。pip install pypdf # 如果你需要处理 PDF安装这个from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter # 1. 加载文档 loader TextLoader(./your_document.txt, encodingutf-8) documents loader.load() # 2. 分割文本 # 直接使用默认分割器可能不合适需要根据文档特点调整。 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个片段的最大字符数 chunk_overlap50, # 片段之间的重叠字符数保持上下文连贯 separators[\n\n, \n, 。, , , , , ] # 分割符优先级 ) split_docs text_splitter.split_documents(documents) print(f原始文档被切分为 {len(split_docs)} 个片段。)关键参数解析chunk_size: 太小会丢失上下文太大会降低检索精度并增加 LLM 处理负担。对于通用文本500-1000 是个不错的起点。chunk_overlap: 防止一个句子或关键信息被硬生生切断。通常设为chunk_size的 10%-20%。separators: 定义了分割的优先级。这里的意思是先按双换行分不行再按单换行分再按句号分……这样能尽可能在语义边界处切割。3.2 向量化与向量数据库存储文本片段需要转换成计算机能理解的“向量”一组数字这个过程叫嵌入Embedding。然后存入向量数据库以便后续快速检索。pip install chromadb langchain-openai tiktoken # Chroma 是一个轻量级、开源的向量数据库非常适合学习和原型开发。from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # 1. 初始化嵌入模型 # 使用 OpenAI 的 text-embedding-ada-002注意它和 Chat 模型是分开计费的。 embeddings OpenAIEmbeddings(modeltext-embedding-ada-002) # 2. 将分割后的文档转换为向量并存入 Chroma # persist_directory 指定向量数据库持久化到磁盘的路径 vectorstore Chroma.from_documents( documentssplit_docs, embeddingembeddings, persist_directory./chroma_db # 数据将保存在这个目录 ) vectorstore.persist() # 显式持久化 print(向量数据库已创建并保存。)重要提醒嵌入模型和聊天模型是独立的。OpenAIEmbeddings调用会产生额外的 API 费用。Chroma将向量数据保存在本地./chroma_db目录。首次运行会创建后续可以直接加载无需重新生成向量除非文档更新。生产环境可能会考虑Pinecone、Weaviate等托管服务但本地 Chroma 对于中小规模数据完全够用。3.3 构建检索链并进行问答现在我们已经有了一个“知识库”。接下来构建一个链条用户提问 - 从向量库检索相关片段 - 组合成提示词 - 发送给 LLM - 返回答案。from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI # 1. 从磁盘加载已存在的向量数据库 vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) # 2. 将向量数据库转换为一个检索器 retriever vectorstore.as_retriever( search_typesimilarity, # 相似度搜索 search_kwargs{k: 3} # 返回最相关的 3 个片段 ) # 3. 创建 LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # temperature0 使输出更确定 # 4. 创建检索问答链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最常用的类型将所有检索到的文档“塞”进上下文 retrieverretriever, return_source_documentsTrue # 返回参考来源便于验证 ) # 5. 进行提问 question 根据文档XX项目的核心目标是什么 result qa_chain.invoke({query: question}) print(f问题{question}) print(f答案{result[result]}) print(\n--- 参考来源 ---) for doc in result[source_documents]: print(f内容片段{doc.page_content[:200]}...) # 打印前200字符 print(f来源{doc.metadata}\n)链条类型chain_type选择stuff: 最简单直接将所有检索到的文档内容合并后一次性发送给 LLM。适用于检索片段总长度不超过 LLM 上下文窗口的情况。map_reduce: 先让 LLM 对每个片段单独总结再对总结进行归纳。适合处理大量文档但调用 API 次数多速度慢。refine: 迭代式处理用上一个片段的答案来完善下一个片段的答案。质量可能更高但更慢。map_rerank: 对每个片段打分并排序只选用高分片段。需要支持打分的模型。对于大多数入门和中等复杂度场景stuff是首选。你需要确保chunk_size * k不超过 LLM 的上下文限制如 GPT-3.5-turbo 是 16K tokens。4. 进阶到智能体用 LangChain 让 LLM 学会使用工具RAG 解决了“知识”问题智能体Agent则要解决“行动”问题。智能体的核心思想是LLM 作为“大脑”根据用户请求和当前状态决定下一步是直接回答还是调用某个工具如搜索、计算、查询数据库来获取信息然后继续思考直到得出最终答案。4.1 定义工具工具可以是任何可执行的函数它接收文本输入返回文本输出。LangChain 要求用tool装饰器来声明。from langchain.agents import tool import requests from datetime import datetime tool def get_current_time(tz: str Asia/Shanghai) - str: 获取指定时区的当前时间。输入应为时区名称例如 Asia/Shanghai 或 UTC。 # 这是一个简化示例实际应使用 pytz 或 zoneinfo 库 now datetime.now() return fThe current time in {tz} is approximately {now.strftime(%Y-%m-%d %H:%M:%S)}. tool def search_web(query: str) - str: 使用搜索引擎示例用 DuckDuckGo搜索网络信息。输入应为搜索关键词。 # 注意实际使用需要安装 duckduckgo-search 库并处理可能的不稳定 try: from duckduckgo_search import DDGS with DDGS() as ddgs: results list(ddgs.text(query, max_results3)) return \n.join([f{r[title]}: {r[body]} for r in results]) except ImportError: return Error: Please install duckduckgo-search package first. except Exception as e: return fSearch error: {str(e)} # 将工具放入列表供智能体使用 tools [get_current_time, search_web]每个工具都必须有清晰的名称和描述。LLM 正是通过这些描述来理解何时以及如何使用该工具。描述要尽可能准确。4.2 创建智能体执行器我们需要一个“执行器”来协调 LLM 的思考、工具调用和结果整合。这里使用 OpenAI 函数调用Function Calling作为智能体的底层机制这是目前最稳定、高效的方式。from langchain_openai import ChatOpenAI from langchain.agents import create_openai_functions_agent, AgentExecutor from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder # 1. 创建 LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 2. 创建提示词模板 # SYSTEM_MESSAGE 定义了智能体的角色和能力 SYSTEM_MESSAGE 你是一个有用的助手可以访问以下工具 {tools} 使用这些工具来回答用户的问题。如果你不需要使用工具也可以直接回答。 请始终以中文回复。 prompt ChatPromptTemplate.from_messages([ (system, SYSTEM_MESSAGE), MessagesPlaceholder(variable_namechat_history), # 预留对话历史的位置 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # 智能体思考过程 ]) # 3. 创建智能体 agent create_openai_functions_agent(llmllm, toolstools, promptprompt) # 4. 创建执行器 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 设置为 True 可以看到详细的思考步骤调试时非常有用 handle_parsing_errorsTrue, # 处理解析错误避免因格式问题直接崩溃 max_iterations5 # 限制最大迭代次数防止死循环 )4.3 运行智能体并观察其思考过程将verboseTrue后运行你会在控制台看到类似下面的输出这是理解智能体如何工作的关键。question 上海现在几点了顺便搜索一下今天的热点新闻。 result agent_executor.invoke({input: question, chat_history: []}) print(\n--- 最终答案 ---) print(result[output])控制台输出示例verbose 模式 Entering new AgentExecutor chain... Thought: 用户问了两个问题1. 上海当前时间。2. 今天的热点新闻。我需要使用工具来获取这些信息。 Action: get_current_time Action Input: {tz: Asia/Shanghai} Observation: The current time in Asia/Shanghai is approximately 2024-05-27 15:30:45. Thought: 我已经得到了上海的时间。现在需要搜索今天的热点新闻。我需要使用搜索工具。 Action: search_web Action Input: {query: 今日热点新闻 2024年5月27日} Observation: [搜索返回的三条新闻摘要...] Thought: 我现在有了时间和新闻信息可以综合起来回答用户了。 Action: Final Answer Final Answer: 上海现在是2024年5月27日下午3点30分左右。根据搜索今日的热点新闻有1. ...[新闻1摘要]。2. ...[新闻2摘要]。3. ...[新闻3摘要]。 Finished chain.通过verbose输出你可以清晰地看到智能体的“思考-行动-观察”循环。这对于调试工具描述是否清晰、LLM 是否错误理解指令至关重要。5. 构建复杂工作流引入 LangGraph 实现多智能体与状态管理当任务变得复杂需要多个步骤循环、分支判断或者多个“智能体”角色协作时基础的AgentExecutor就显得力不从心。这时就需要LangGraph。它将工作流抽象为“图”Graph节点代表步骤可以是工具调用、LLM调用或普通函数边代表步骤之间的流转条件。5.1 理解 LangGraph 的核心概念State: 一个共享的字典在整个工作流执行过程中传递和修改数据。这是 LangGraph 管理状态的核心。Node: 节点一个函数接收当前 State执行操作并返回更新后的 State。Edge: 边决定下一个执行哪个 Node。可以是固定流转也可以根据 State 中的某个值动态决定条件边。5.2 构建一个简单的审阅工作流假设我们有一个需求用户提交一段文案需要先后经过“拼写检查”和“风格优化”两个步骤。我们可以用 LangGraph 来编排。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END import operator # 1. 定义状态结构 class ReviewState(TypedDict): 工作流的状态定义 original_text: str # 原始文案 spell_checked_text: str # 拼写检查后的文案 final_text: str # 最终优化后的文案 feedback: list[str] # 收集各步骤的反馈信息 # 2. 定义节点函数 def node_spell_check(state: ReviewState) - ReviewState: 节点A模拟拼写检查 # 这里简化处理实际可以调用专门的拼写检查API或库 text state[original_text] # 假设我们只是做个简单替换模拟检查 checked_text text.replace(teh, the).replace(adn, and) feedback f拼写检查完成。修正了常见拼写错误。 return { spell_checked_text: checked_text, feedback: state[feedback] [feedback] } def node_style_optimize(state: ReviewState) - ReviewState: 节点B调用LLM进行风格优化 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) prompt f 请将以下文案优化得更专业、更吸引人。保持原意。 原文案{state[spell_checked_text]} 优化后的文案 response llm.invoke(prompt) optimized_text response.content feedback f风格优化完成。LLM已生成优化版本。 return { final_text: optimized_text, feedback: state[feedback] [feedback] } # 3. 构建图 workflow StateGraph(ReviewState) # 添加节点 workflow.add_node(spell_check, node_spell_check) workflow.add_node(style_optimize, node_style_optimize) # 设置边的连接关系spell_check - style_optimize - END workflow.add_edge(spell_check, style_optimize) workflow.add_edge(style_optimize, END) # 设置入口节点 workflow.set_entry_point(spell_check) # 编译图得到可执行对象 app workflow.compile()5.3 执行工作流并查看结果# 4. 初始化状态并执行 initial_state: ReviewState { original_text: Our product is teh best in the market adn we are sure you will love it., spell_checked_text: , final_text: , feedback: [] } # 执行工作流 final_state app.invoke(initial_state) print(原始文案:, final_state[original_text]) print(\n拼写检查后:, final_state[spell_checked_text]) print(\n最终优化文案:, final_state[final_text]) print(\n工作流反馈:, final_state[feedback])这个例子展示了线性工作流。LangGraph 更强大的地方在于支持条件边和循环。例如你可以在style_optimize节点后添加一个由 LLM 判断“文案是否足够好”的节点如果不够好就循环回style_optimize重新优化直到满足条件或达到最大循环次数。这正是在构建复杂、动态的智能体系统时所需要的。6. 实战避坑与生产化思考跟着教程跑通 Demo 只是第一步。要把 LangChain/LangGraph 用于实际项目以下几个坑点和优化方向必须提前考虑。6.1 常见错误与排查顺序API 密钥或网络问题任何与OpenAI、Anthropic等相关的调用失败首先检查环境变量OPENAI_API_KEY等是否设置正确以及网络是否能正常访问对应 API 地址对于国内用户这可能是个常见问题。版本兼容性问题LangChain 版本迭代快某些接口或参数名可能会变。如果代码报ImportError或AttributeError第一反应是去查阅对应版本的官方文档https://python.langchain.com/docs/而不是盲目搜索。提示词Prompt问题LLM 输出不符合预期比如不调用工具、格式错误。首先检查你的SYSTEM_MESSAGE和工具描述是否足够清晰。用verboseTrue查看 LLM 的原始思考过程往往能发现问题。向量检索效果差RAG 回答不准确。排查点文本分割chunk_size和chunk_overlap是否合适用print(split_docs)看看分割后的片段是否保持了语义完整性。检索策略search_kwargs{k: 3}中的k值是否太小可以尝试增加到 5 或 10。也可以试试search_typemmr最大边际相关性在相关性和多样性之间取得平衡。嵌入模型对于中文场景text-embedding-ada-002效果不错但也可以尝试专门的多语言或中文嵌入模型。智能体陷入死循环智能体不停调用同一个工具。务必设置AgentExecutor的max_iterations参数如 10。同时检查工具函数的返回值格式是否稳定LLM 能否正确解析。6.2 从 Demo 到生产环境的考量异步与并发LangChain 原生支持异步。对于需要处理大量请求的 Web 服务务必使用ainvoke、abatch等异步方法并结合asyncio提高吞吐量。# 异步调用示例 async def process_question(question): result await qa_chain.ainvoke({query: question}) return result缓存频繁调用相同的 LLM 请求例如相同的提示词会产生不必要的费用和延迟。使用LangChain的缓存组件如InMemoryCache或SQLiteCache可以显著提升性能。from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())日志与监控生产系统必须要有完善的日志。记录每一次 LLM 调用、工具调用的输入输出、耗时和 Token 使用量。这有助于成本核算、问题排查和效果优化。错误处理与重试网络波动、API 限流不可避免。使用tenacity等库为 LLM 和工具调用添加重试机制。在AgentExecutor中利用handle_parsing_errors等参数优雅处理部分错误。向量数据库的选择本地 Chroma 适合原型和中小数据量。当数据量极大百万级以上或需要高可用、分布式时需要考虑Pinecone、Weaviate、Qdrant等专业向量数据库服务。提示词工程与管理不要将提示词硬编码在代码中。考虑将其外置到配置文件、数据库或专门的提示词管理平台便于迭代和 A/B 测试。6.3 学习路径建议不要试图一次性掌握 LangChain 的所有模块。建议按以下路径循序渐进核心概念Model I/O(Prompt/LLM/OutputParser)Data Connection(Document Loader/Text Splitter/Vectorstore)Chains(LLMChain, SequentialChain)。重点突破深入掌握RetrievalQA链和OpenAI Functions Agent这是使用频率最高的两部分。探索生态根据需求学习LangChain Community里的各种集成如邮件工具、SQL 数据库工具、GitHub 工具等。进阶编排当遇到需要循环、分支、多角色协作的复杂场景时再深入学习LangGraph。关注官方动态LangChain 生态发展迅速关注其官方博客和 Discord了解新特性如 LangSmith 用于跟踪和评估 LangServe 用于部署的最佳实践。最终评判你是否掌握了 LangChain不是背下了多少 API而是能否独立设计并实现一个解决实际问题的、健壮的 AI 应用流程。从用一个清晰的提示词驱动 LLM到用 Chain 串联多个步骤再到用 Agent 动态决策最后用 Graph 管理复杂状态这条路径上的每一步都对应着真实项目中不断增长的需求复杂度。