LangChain实战：30分钟构建RAG文档问答与AI智能体
1. 从“胶水代码”到“智能应用流水线”：我为什么选择LangChain
如果你最近在捣鼓大语言模型（LLM），想把ChatGPT、Claude或者本地部署的Llama、Qwen这些“大脑”真正用起来，而不是仅仅停留在聊天窗口里，那你大概率已经听过或者正在被“LangChain”这个词包围。我第一次接触它时，感觉就像回到了Web开发早期，每个项目都要自己手写一堆连接数据库、处理表单、管理会话的“胶水代码”，繁琐且重复。LangChain的出现，就是为了解决这个痛点：它不是一个新模型，而是一个框架，一套工具链，或者说，一个专门为构建基于LLM的应用而设计的“智能应用流水线”。
简单来说，LangChain帮你把“调用大模型API”这件简单的事，升级成了“构建一个可靠、可扩展、具备复杂逻辑的AI应用”这件系统工程。它把常见的模式抽象成组件，比如如何把用户的问题和你的知识库（文档）结合起来（这就是RAG），如何让大模型学会使用工具（比如查天气、执行计算），如何管理多轮对话的上下文，以及如何把多个步骤串联成一个自动化的工作流。没有它，你可能需要自己处理提示词工程、上下文窗口管理、工具调用解析、异步流式输出等一系列令人头疼的细节。有了它，你可以像搭积木一样，快速组合出功能强大的AI智能体（Agent）或者问答系统。
网上很多人争论LangChain和它的“兄弟”LangGraph，或者和Dify、CrewAI这些后起之秀有什么区别。我的看法是，LangChain更像是一个底层工具箱和标准件库，它提供了最大的灵活性和控制力，适合开发者深入定制；而Dify、CrewAI等则是在这个工具箱基础上封装好的一体化解决方案或高级工作台，开箱即用，但定制性相对受限。对于想真正理解LLM应用架构、并拥有完全掌控权的开发者而言，从LangChain入门是必经之路。最近甚至看到有说法，OpenAI内部团队用类似LangChain的方法论，在5个月内零手写代码产出了百万行级别的系统，这虽然无从考证，但足以说明模块化、链式编排的思想在AI工程化中的巨大价值。
那么，这篇快速入门的目标就是：让你在30分钟内，绕过那些复杂的概念堆砌，直接动手搭建起两个最核心、最实用的LangChain应用场景——文档问答（RAG）和智能体（Agent），并理解其背后的运作原理。我们会用最少的依赖，最多的注释，带你走通整个流程。
2. 环境搭建与核心概念“祛魅”
在开始写代码之前，我们需要一个清晰的地图。LangChain的体系看似庞大，但核心就是几个概念，一旦理解，后面就是组合使用的问题。
2.1 极简环境准备：只安装必要的
我不建议一开始就pip install langchain[all]，那会引入大量你可能暂时用不到的依赖。我们聚焦核心。
# 1. 安装LangChain核心包 pip install langchain-core langchain # 2. 安装社区集成包，这里以OpenAI为例（你需要有自己的API Key） # 如果你用国产模型，比如DeepSeek，可以安装 `langchain-deepseek` 或类似社区包 pip install langchain-openai # 3. 安装文本处理和向量数据库客户端（用于RAG示例） # 我们选用轻量级的ChromaDB作为本地向量库，以及文本分割器 pip install chromadb langchain-chroma tiktoken # 4. 可选但推荐：安装用于Agent示例的工具调用模拟包 pip install langchain-community安装完成后，建议你准备好一个LLM的API Key。本文以OpenAI GPT-3.5-turbo为例，因为它最通用。如果你没有，可以使用开源的Ollama本地运行一个模型（如Llama 3.1），只需将后续代码中的ChatOpenAI替换为ChatOllama并指定模型名称即可。
2.2 五大核心概念，五分钟掌握
LangChain的文档里概念很多，但入门只需抓住这五个：
模型 I/O (Model I/O)：这是最底层的一环，负责与大模型对话。主要包括：
- LLM： 纯文本补全模型（如早期的GPT-3）。
- ChatModel： 专为对话设计的模型（如GPT-3.5-turbo, Claude）。这是我们最常用的。
- 提示词模板 (PromptTemplate)： 避免在代码中硬编码提示词。你可以创建一个模板，把
{topic}这样的占位符在运行时替换成具体内容。
检索 (Retrieval)： 这是RAG（检索增强生成）的核心。当模型需要回答超出其训练数据（或最新）的问题时，就从你自己的知识库（如文档、数据库）中查找相关信息，然后连同问题和信息一起发给模型。关键组件有：
- 文档加载器 (Document Loader)： 从PDF、网页、Notion等处加载文档。
- 文本分割器 (Text Splitter)： 将长文档切成模型上下文窗口能容纳的小块。
- 向量存储 (Vector Store)： 将文本块转换成向量（嵌入）并存储，实现相似性搜索。
链 (Chain)： LangChain的灵魂。它不是简单的顺序调用，而是将多个组件（模型、提示词、工具等）按特定逻辑组合成一个可执行的工作流。最简单的链是
LLMChain（提示词 + 模型），复杂的链可以包含条件判断、循环等。智能体 (Agent)： 链的升级版。智能体的核心是让模型学会自主决策使用哪些工具。你给模型一套工具（如计算器、搜索引擎API、数据库查询），和一个目标，模型会自己规划步骤：“要解决这个问题，我需要先查天气，再用结果进行计算。” Agent = LLM + 工具集 + 决策逻辑。
记忆 (Memory)： 让对话或应用拥有“记忆”能力，记住之前交互的内容。可以是简单的对话缓冲区，也可以是更复杂的、基于向量存储的长期记忆。
注意： 你可能还听过
LangGraph，它是基于LangChain构建的，用于描述有状态、多环节、可能循环或分支的复杂工作流。如果把Chain比作一条直线，LangGraph就是一张流程图。对于入门来说，我们先掌握Chain和Agent就够了。
理解了这些，我们就可以开始实战了。下面两个例子，将分别对应检索（RAG）和智能体这两个最核心的应用。
3. 实战一：构建你的第一个RAG文档问答系统
RAG是目前LangChain最火的应用场景。假设你有一份公司内部的产品手册PDF，你想让AI根据这份手册来回答问题。
3.1 步骤拆解与代码实现
整个过程分为四步：加载文档 -> 分割文本 -> 向量化存储 -> 检索问答。
# 导入必要的模块 import os from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.document_loaders import TextLoader # 示例用文本，实际可用PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_chroma import Chroma from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate # 步骤1: 设置你的OpenAI API Key (关键！请替换成你自己的) os.environ["OPENAI_API_KEY"] = "sk-你的真实api-key" # 初始化一个性价比高的聊天模型 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 步骤2: 加载与分割文档 # 假设我们有一个 `product_manual.txt` 文件 loader = TextLoader("./product_manual.txt") documents = loader.load() # 使用递归字符分割器，它尝试按段落、句子、单词等自然边界分割 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个块大约500字符 chunk_overlap=50, # 块之间重叠50字符，避免信息被割裂 separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""] # 分割优先级 ) chunks = text_splitter.split_documents(documents) print(f"原始文档被分割成了 {len(chunks)} 个文本块。") # 步骤3: 向量化并存入向量数据库 # 使用OpenAI的嵌入模型将文本转换为向量 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 将分割后的文本块存入ChromaDB。`persist_directory` 参数让数据持久化到磁盘 vector_store = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db" # 数据将保存在这个目录 ) vector_store.persist() # 确保写入磁盘 print("向量数据库已创建并持久化。") # 步骤4: 创建检索问答链 # 首先，定义一个提示词模板，告诉模型如何利用检索到的上下文 qa_prompt = PromptTemplate.from_template( """请根据以下上下文信息来回答问题。如果你不知道答案，就说不知道，不要编造。 上下文： {context} 问题：{question} 答案：""" ) # 创建检索器，从向量库中搜索最相关的3个文本块 retriever = vector_store.as_retriever(search_kwargs={"k": 3}) # 构建RetrievalQA链，它封装了检索+问答的过程 qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # “stuff”模式：将所有检索到的上下文塞进提示词。还有`map_reduce`、`refine`等处理长上下文的方式。 retriever=retriever, chain_type_kwargs={"prompt": qa_prompt}, # 使用我们自定义的提示词 return_source_documents=True # 返回检索到的源文档，便于调试 ) # 步骤5: 提问！ question = "我们产品的主要优势是什么？" result = qa_chain.invoke({"query": question}) print(f"问题：{question}") print(f"答案：{result['result']}") print("\n--- 检索到的参考来源 ---") for i, doc in enumerate(result['source_documents'][:2]): # 打印前两个来源 print(f"[来源{i+1}] {doc.page_content[:200]}...") # 只打印前200字符3.2 关键细节与避坑指南
文本分割是门艺术：
chunk_size和chunk_overlap没有银弹。500/50是一个通用起点。如果答案总是支离破碎，尝试减小chunk_size；如果答案缺乏连贯性，尝试增大overlap。对于中文，separators里加入句号、逗号很重要。嵌入模型的选择：示例用了OpenAI的嵌入模型，需要计费。对于本地或低成本方案，可以考虑开源的
BAAI/bge-small-zh-v1.5等模型，配合langchain-huggingface包。关键点：问答用的LLM和生成嵌入的模型最好在语义空间上对齐（例如都用OpenAI系或都用BGE系），否则检索精度可能下降。chain_type的选择：stuff：最简单直接，把所有检索到的上下文拼接起来发给模型。适合上下文总长度不超过模型限制的情况。map_reduce：先对每个文本块单独生成答案（Map），再汇总所有答案生成最终答案（Reduce）。适合处理大量文档，但成本高、可能丢失全局信息。refine：迭代式处理，用第一个块生成初始答案，然后用后续块不断“优化”这个答案。通常质量更高，但速度慢。对于入门，stuff足够了。当你发现提示词因上下文过长而被截断时，再考虑后两者。
向量数据库持久化：示例中
Chroma数据保存到了本地./chroma_db。这意味着下次启动程序时，你可以直接加载已有的数据库，无需重新嵌入，节省大量时间和API费用：# 第二次及以后运行，直接加载 vector_store = Chroma(persist_directory="./chroma_db", embedding_function=embeddings) retriever = vector_store.as_retriever()这是生产级应用必须考虑的一步。
4. 实战二：创建一个能使用工具的AI智能体（Agent）
智能体让AI从“答题者”变成了“执行者”。我们创建一个能进行简单数学计算和获取当前日期的智能体。
4.1 定义工具与初始化Agent
LangChain提供了多种Agent类型，我们使用最通用、功能最强的ReAct范式Agent。
from langchain.agents import AgentExecutor, create_react_agent from langchain import hub from langchain.tools import Tool from datetime import datetime import math # 步骤1: 自定义工具函数 # 工具1: 一个能计算平方根的函数 def calculate_sqrt(input_str: str) -> str: """计算一个数的平方根。输入应该是一个数字字符串。""" try: number = float(input_str) if number < 0: return "错误：不能计算负数的平方根。" result = math.sqrt(number) return f"{number} 的平方根是 {result:.4f}" except ValueError: return "错误：请输入一个有效的数字。" # 工具2: 一个能返回当前日期和时间的函数 def get_current_time(input_str: str = "") -> str: """返回当前的日期和时间。输入参数被忽略。""" now = datetime.now() # 忽略输入参数是LangChain工具定义的一个常见模式 return f"当前日期和时间是：{now.strftime('%Y-%m-%d %H:%M:%S')}" # 步骤2: 将函数包装成LangChain Tool对象 # `func`参数指向我们的函数，`description`至关重要，Agent靠它来决定是否使用该工具。 tools = [ Tool( name="SquareRootCalculator", func=calculate_sqrt, description="""在需要计算一个非负数的平方根时使用。输入应该是一个数字字符串。例如，如果问题是‘16的平方根是多少？’，输入就是‘16’。""" ), Tool( name="CurrentTime", func=get_current_time, description="""在用户询问当前时间、今天日期或类似关于现在时刻的问题时使用。此工具不需要输入参数。""" ) ] # 步骤3: 从LangChain Hub拉取一个优秀的ReAct提示词模板 # 这是一个社区维护的、经过优化的提示词，比我们自己写要可靠得多。 prompt = hub.pull("hwchase17/react") # 步骤4: 创建ReAct Agent # `create_react_agent` 将模型、工具和提示词组合成一个Agent对象。 agent = create_react_agent(llm, tools, prompt) # 步骤5: 创建Agent执行器，它负责运行Agent并处理工具调用循环 agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 设置为True，可以看到Agent的思考过程！这对调试和学习至关重要。 handle_parsing_errors=True, # 当Agent输出无法解析为工具调用时，自动处理错误 max_iterations=5, # 限制最大迭代次数，防止陷入死循环 early_stopping_method="generate" # 当Agent认为任务完成时，可以提前停止 ) # 步骤6: 运行Agent！ print("=== Agent 思考过程演示 ===") result = agent_executor.invoke({ "input": "请先告诉我现在的时间，然后计算一下25的平方根。" }) print(f"\n最终答案：{result['output']}")当你运行这段代码，并将verbose=True时，会在控制台看到类似下面的精彩输出，这正是ReAct（Reasoning + Acting）思想的体现：
> Entering new AgentExecutor chain... 我需要按顺序回答两个问题：当前时间，和25的平方根。 我有两个工具：CurrentTime 可以获取当前时间，SquareRootCalculator 可以计算平方根。 首先，我应该获取当前时间。 Action: CurrentTime Action Input: Observation: 当前日期和时间是：2024-05-27 14:30:15 好的，我已经知道时间了。现在需要计算25的平方根。 Action: SquareRootCalculator Action Input: 25 Observation: 25 的平方根是 5.0000 现在我有了两个信息：时间和计算结果。我可以给出最终答案了。 Final Answer: 当前时间是2024年5月27日 14:30:15。25的平方根是5。 > Finished chain. 最终答案：当前时间是2024年5月27日 14:30:15。25的平方根是5。4.2 Agent开发的核心心法
工具描述是灵魂：
description字段必须清晰、无歧义地说明工具的用途、适用场景和输入格式。Agent完全依赖这个描述来做决策。写得模糊，Agent就会用错或不用。善用
verbose=True：在开发阶段，务必打开这个选项。它能让你亲眼看到Agent的思考链（Chain of Thought），理解它为什么做出某个决策，在哪里卡住了。这是调试Agent最强大的手段。处理解析错误：
handle_parsing_errors=True是救命稻草。有时模型输出格式不符合工具调用规范，这个设置能防止整个程序崩溃，让模型重试。更高级的做法是自定义一个错误处理回调函数。设置迭代限制：
max_iterations必须设置。防止Agent陷入“思考-调用-再思考”的死循环，消耗大量Token。从简单工具开始：先让Agent能稳定调用一两个简单工具，再逐步增加复杂度。不要一开始就给它十几种工具，那会大大增加决策难度和出错概率。
5. 进阶：流式输出、复杂链与调试技巧
当你掌握了基本用法，接下来会遇到一些实际开发中的挑战。
5.1 实现流式输出，提升用户体验
在Web应用中，让答案一个字一个字地“流”出来，体验远好于等待长时间后一次性显示。LangChain对流式输出有很好的支持。
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler # 方法1: 在模型层面启用流式（输出原始Token） streaming_llm = ChatOpenAI( model="gpt-3.5-turbo", temperature=0, streaming=True, # 关键参数 callbacks=[StreamingStdOutCallbackHandler()] # 回调函数，将流输出到标准输出 ) # 注意：这种方式下，`invoke`会边生成边打印。但如果你用Agent或复杂链，流式可能只体现在模型生成部分，中间步骤还是会一次性输出。 # 方法2: 对于Chain，使用 `astream` 或 `astream_events` (异步) # 这是更现代、更推荐的方式，可以流式输出整个链的每一步。 import asyncio async def stream_qa_chain(): qa_chain = RetrievalQA.from_chain_type(llm=streaming_llm, retriever=retriever, chain_type="stuff") async for chunk in qa_chain.astream({"query": "产品优势是什么？"}): # `chunk` 是一个字典，包含中间状态和最终输出 if "result" in chunk: print(chunk["result"], end="", flush=True) # 逐块打印结果 # asyncio.run(stream_qa_chain())关于你搜索词中提到的“流式输出吞掉reasoning-content字段”，这通常发生在使用某些特定Agent或复杂事件流时。解决方案是使用astream_events并正确过滤事件类型，或者检查回调函数的处理逻辑，确保不是只捕获了最终输出而忽略了中间推理步骤。
5.2 构建自定义链：串联多个步骤
当内置链不够用时，你需要用LCEL（LangChain Expression Language）来定义自己的链。LCEL使用管道符|，非常直观。
from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough # 假设我们想先让模型把用户问题改写成更好的搜索查询，再用这个查询去检索，最后回答。 # 1. 定义两个提示词模板 rewrite_prompt = ChatPromptTemplate.from_template( "你是一个专业的搜索查询改写助手。请将以下用户问题改写成更适合用于向量数据库检索的简短关键词查询。\n原问题：{question}\n改写后的查询：" ) answer_prompt = ChatPromptTemplate.from_template( "基于以下上下文回答问题：\n上下文：{context}\n问题：{original_question}\n答案：" ) # 2. 用LCEL组合链 custom_chain = ( { "original_question": RunnablePassthrough(), # 传递原始问题 "enhanced_query": rewrite_prompt | llm | StrOutputParser(), # 改写问题 } | { "context": lambda x: retriever.invoke(x["enhanced_query"]), # 用改写后的问题检索 "original_question": lambda x: x["original_question"], # 继续传递原始问题 } | answer_prompt # 组合上下文和原始问题，形成最终提示词 | llm # 发送给模型 | StrOutputParser() # 解析输出 ) result = custom_chain.invoke("你们公司产品的售后服务政策怎么样？") print(result)LCEL的优势在于声明式和组合性。每个步骤清晰可见，而且这些Runnable对象本身可以嵌套组合，构建出极其复杂的工作流。
5.3 必须掌握的调试与日志记录技巧
LangChain应用一旦复杂，调试起来可能像黑盒。以下几个方法是我的必备工具箱：
verbose=True无处不在： 不仅在AgentExecutor，在LLMChain、RetrievalQA等对象初始化时也可以设置，它会打印内部的LLM调用和输入输出。使用回调函数： LangChain提供了强大的回调系统。你可以自定义回调来记录每次LLM调用的提示词、完成词、Token用量等。
from langchain.callbacks import FileCallbackHandler import logging logging.basicConfig(level=logging.INFO, filename='langchain.log') handler = FileCallbackHandler('langchain.log') llm = ChatOpenAI(..., callbacks=[handler])手动检查中间结果： 对于RAG，经常需要检查检索到的文档是否相关。可以在调用链之前，先单独测试检索器：
test_docs = retriever.invoke("你们的产品优势") for doc in test_docs: print(doc.page_content[:300]) print("---")简化问题定位： 如果链不工作，先绕过链，直接测试最基础的组件：模型能正常响应吗？提示词模板格式化对吗？检索器能返回文档吗？一步步隔离问题。
6. 生态、选型与未来学习路径
当你完成上面两个实战，你已经掌握了LangChain最核心的60%功能。接下来，你可以根据兴趣深入：
深入RAG： 研究更高级的检索策略，如
MultiQueryRetriever（生成多个查询以提升召回率）、ContextualCompressionRetriever（在检索后对文档进行压缩摘要）。学习ParentDocumentRetriever来处理文档层次结构。深入Agent： 尝试
OpenAI ToolsAgent（直接利用GPT-4等模型的原生函数调用能力，格式更稳定）。给你的Agent添加记忆，让它能在多轮对话中记住上下文。探索Plan-and-Execute类型的Agent，进行更复杂的任务分解。探索LangGraph： 当你需要处理有循环、有状态、多角色协作的工作流时（例如一个模拟辩论的Agent系统，或者一个需要反复审核修改的写作流程），LangGraph是你的下一个台阶。它用图的方式来定义和控制流程。
关注部署与生产化： 使用
LangServe快速将你的链或Agent部署为API服务。用LangSmith（LangChain官方平台）来跟踪、监控、调试和评估你的所有LLM调用，这是团队协作和项目上线的神器。
关于选型，最后再总结一下：
- LangChain vs LangGraph： LangChain是基础和标准库，LangGraph是用于复杂工作流的扩展库。先学好LangChain。
- LangChain vs Dify/CrewAI： 如果你需要快速搭建一个标准化的AI应用（如客服机器人、知识库）且不想写太多代码，Dify这类低代码平台很棒。如果你要构建高度定制、逻辑复杂、需要深度集成到现有系统的AI能力，LangChain提供的编程控制能力是不可替代的。CrewAI则更聚焦于多智能体协作场景。
学习资源方面，除了官方文档，多关注GitHub上的示例项目，并在自己的项目中大胆实践。从解决一个具体的小问题开始，比如“用LangChain自动总结我每天收到的邮件”，在实战中成长是最快的。记住，这个领域变化飞快，保持动手和阅读最新博客、论文的习惯，比死记硬背API更重要。