最近在尝试将大语言模型LLM应用到实际业务中时你是否遇到过这样的困境模型本身很强大但让它读取你的私有文档、调用外部API、或者记住多轮对话内容却异常困难网上资料要么是零散的代码片段要么是晦涩的理论讲解想要构建一个可用的AI应用总感觉无从下手。如果你正为此烦恼那么你来对地方了。LangChain 正是为解决这些“最后一公里”问题而生的框架。本文将为你提供一份从零开始的 LangChain 实战指南摒弃华而不实的宣传专注于可复现的代码和清晰的逻辑。无论你是刚接触 AI 应用开发的在校学生还是希望将 LLM 能力集成到现有系统的工程师都能通过本文构建起坚实的知识体系并亲手搭建出你的第一个智能应用。1. LangChain 是什么为什么你需要它在深入代码之前我们首先要理解 LangChain 的核心价值。简单来说LangChain 是一个用于开发由语言模型驱动的应用程序的框架。它不是一个模型而是一个“粘合剂”和“工具箱”。想象一下强大的 LLM如 GPT-4、ChatGLM、通义千问是一个聪明但“与世隔绝”的大脑。它知识渊博却无法直接读取你电脑里的 PDF不能帮你查询今天的天气也无法记住十分钟前你们聊过什么。LangChain 的作用就是为这个大脑安装上“眼睛”文档读取、“手”工具调用和“记忆”历史记录。LangChain 主要解决以下核心问题数据接入如何让 LLM 处理它训练数据之外的信息比如你的公司知识库、本地文档、数据库。能力扩展如何让 LLM 执行它本身做不到的动作比如计算、搜索网页、调用业务 API。对话连贯如何让 LLM 在多轮对话中保持上下文理解指代实现连贯的交互。流程编排如何将读取文档、调用工具、生成回答等多个步骤组合成一个稳定、可控的自动化流程。市面上也有其他工具如 Dify、LangFlow它们更偏向于低代码/可视化搭建。而LangChain 的优势在于其灵活性和可编程性它提供了丰富的底层组件允许开发者进行深度定制和复杂逻辑的构建更适合集成到生产级应用中。2. 环境准备搭建你的第一个 LangChain 项目“工欲善其事必先利其器”。在开始编写激动人心的 AI 应用之前我们需要一个干净、可复现的开发环境。本节将详细指导你完成从零开始的环境搭建。2.1 基础环境配置首先确保你的操作系统上已经安装了Python。LangChain 支持 Python 3.8 及以上版本。我们推荐使用Python 3.10它在兼容性和性能上都有较好的表现。步骤 1检查 Python 环境打开你的终端Windows 上是 CMD 或 PowerShellMac/Linux 上是 Terminal输入以下命令python --version # 或 python3 --version如果显示版本号大于等于 3.8则可以进行下一步。如果没有安装请前往 Python 官网 下载并安装。步骤 2创建虚拟环境为了避免不同项目间的依赖冲突强烈建议使用虚拟环境。这里我们使用venvPython 内置。# 创建一个新的项目目录 mkdir my-langchain-project cd my-langchain-project # 创建虚拟环境环境文件夹名为 venv python -m venv venv # 激活虚拟环境 # 在 Windows 上 venv\Scripts\activate # 在 Mac/Linux 上 source venv/bin/activate激活后你的命令行提示符前通常会显示(venv)表示你已进入该虚拟环境。2.2 安装核心依赖在激活的虚拟环境中使用pip安装 LangChain。由于 LangChain 生态庞大我们通常安装其核心包以及可能用到的社区集成包。# 安装 LangChain 核心包 pip install langchain # 安装 OpenAI 集成包如果你使用 OpenAI 的模型如 GPT-3.5/4 pip install openai # 安装用于解析文档的常用工具 pip install langchain-community # 安装文本嵌入模型库用于文档向量化后续RAG会用到 pip install sentence-transformers # 安装向量数据库客户端以Chroma为例轻量易用 pip install chromadb # 安装用于读取PDF/TXT等文件的文档加载器 pip install pypdf安装完成后可以通过pip list | findstr langchainWindows或pip list | grep langchainMac/Linux来确认安装成功。2.3 配置 API 密钥大多数 LLM 服务如 OpenAI、智谱AI、百度千帆都需要 API 密钥。请务必妥善保管你的密钥不要将其提交到公开的代码仓库如 GitHub。我们以 OpenAI 为例你也可以使用其他兼容 OpenAI API 的模型服务。创建一个名为.env的文件来存储密钥并使用python-dotenv库来读取它。# 安装 python-dotenv pip install python-dotenv在项目根目录创建.env文件内容如下# .env OPENAI_API_KEY你的-openai-api-key-在这里然后创建一个 Python 文件test_env.py来测试环境是否就绪# test_env.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI # 加载 .env 文件中的环境变量 load_dotenv() # 初始化一个聊天模型 llm ChatOpenAI( modelgpt-3.5-turbo, temperature0, # 控制创造性0表示更确定性的输出 api_keyos.getenv(OPENAI_API_KEY) # 从环境变量读取密钥 ) # 进行一个简单的调用 response llm.invoke(你好请用中文简单介绍一下你自己。) print(response.content)运行这个脚本python test_env.py如果看到模型返回了一段中文的自我介绍恭喜你你的 LangChain 基础环境已经成功搭建。3. LangChain 核心概念与组件拆解LangChain 的设计是模块化的理解其核心抽象是高效使用的关键。我们将逐一拆解最重要的几个概念。3.1 Model I/O与模型对话的桥梁这是最基础的模块负责与各种 LLM 进行交互。主要包含三个部分Prompt 提示词模板用于结构化地组织输入给模型的信息。Model 模型本身如ChatOpenAI,ChatGoogleGenerativeAI等。Output Parser 输出解析器用于将模型非结构化的文本输出解析成我们程序可以处理的结构化数据如 JSON、列表。示例使用 PromptTemplate 和 OutputParserfrom langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI # 1. 定义提示词模板 prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个专业的翻译助手。), (user, 请将以下英文句子翻译成中文{input_text}) ]) # 2. 初始化模型 model ChatOpenAI(modelgpt-3.5-turbo) # 3. 定义输出解析器这里简单地将内容转为字符串 output_parser StrOutputParser() # 4. 将三者链接成一个链Chain chain prompt_template | model | output_parser # 5. 调用链 result chain.invoke({input_text: Hello, LangChain! This is a powerful framework for building LLM applications.}) print(f翻译结果{result}) # 输出翻译结果你好LangChain这是一个用于构建大型语言模型应用程序的强大框架。这个|运算符是 LangChain v0.1.0 后引入的 LCELLangChain Expression Language语法它让组件的连接变得非常直观和灵活。3.2 Retrieval让模型拥有“外部记忆”检索Retrieval是构建知识库问答系统的核心。其核心流程被称为RAGRetrieval-Augmented Generation检索增强生成。加载从源PDF、网页、数据库加载文档。分割将长文档切分成语义相关的小块Chunks。嵌入使用嵌入模型Embedding Model将文本块转换为数值向量。存储将向量存储到向量数据库Vector Store中。检索当用户提问时将问题也转换为向量并在向量数据库中查找最相似的文本块。生成将检索到的相关文本块作为上下文与问题一起交给 LLM 生成最终答案。示例创建一个简单的本地知识库from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI # 1. 加载文档这里用文本文件示例 loader TextLoader(./my_document.txt, encodingutf-8) documents loader.load() # 2. 分割文档 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块的最大字符数 chunk_overlap50 # 块之间的重叠字符数保持上下文连贯 ) texts text_splitter.split_documents(documents) # 3. 创建向量存储 embeddings OpenAIEmbeddings() # 需要 OPENAI_API_KEY vectorstore Chroma.from_documents(documentstexts, embeddingembeddings, persist_directory./chroma_db) # persist_directory 参数会将向量数据库保存到本地磁盘 # 4. 创建检索器 retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 检索最相关的3个块 # 5. 创建问答链 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 将检索到的所有内容“塞”进提示词 retrieverretriever, return_source_documentsTrue # 返回参考来源 ) # 6. 进行问答 query 我的文档中提到了哪些主要项目 result qa_chain.invoke({query: query}) print(f答案{result[result]}) print(f参考来源{result[source_documents]})3.3 Agents赋予模型使用工具的能力智能体Agent是 LangChain 中最具魅力的部分。它让 LLM 能够自主决策根据用户目标选择并调用合适的工具Tool来完成任务。你可以把 Agent 看作一个“调度员”LLM 是它的“大脑”Tools 是它的“双手”。一个典型的 Agent 工作流程是用户输入 - LLM 思考 - 选择工具 - 执行工具 - 观察结果 - 再次思考 - ... - 生成最终回答。示例创建一个能查询天气和计算的智能体from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.tools import Tool from langchain_community.utilities import WikipediaAPIWrapper import math # 1. 定义自定义工具 def calculate_sqrt(input: str) - str: 计算一个数的平方根。 try: number float(input) return str(math.sqrt(number)) except ValueError: return 输入错误请提供一个数字。 # 2. 封装工具列表 tools [ Tool( nameWikipedia, funcWikipediaAPIWrapper().run, # 使用维基百科查询工具 description当你需要查询关于人物、地点、公司、历史事件等通用事实信息时非常有用。输入应该是一个明确的搜索主题。 ), Tool( nameSquareRootCalculator, funccalculate_sqrt, description计算一个正数的平方根。输入应该是一个数字。 ) ] # 3. 创建提示词模板这是驱动Agent思考的“系统指令” prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以调用工具来回答问题。 如果你需要查询事实信息请使用Wikipedia工具。 如果你需要计算平方根请使用SquareRootCalculator工具。 请用中文回答用户。在最终回答前请清晰展示你的思考过程。), (user, {input}), (agent_scratchpad, {agent_scratchpad}), # 这是Agent记录其思考和工具调用结果的地方 ]) # 4. 初始化LLM需要支持Function Calling的模型如gpt-3.5-turbo或gpt-4 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 5. 创建Agent agent create_openai_tools_agent(llm, tools, prompt) # 6. 创建Agent执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 7. 运行Agent result agent_executor.invoke({input: 请先告诉我爱因斯坦的主要贡献然后计算256的平方根。}) print(result[output])运行上述代码你会看到verboseTrue模式下 Agent 详细的思考步骤Thought、行动Action和观察Observation最终给出结合了工具调用结果的答案。4. 完整实战构建个人智能文档问答助手现在我们将综合运用前面所学的知识构建一个完整的、带 Web 界面的个人文档问答助手。这个项目将涵盖文档加载、向量化存储、检索和问答链。4.1 项目结构设计首先创建清晰的项目目录。my_doc_qa_assistant/ ├── app.py # 主应用文件使用Streamlit构建界面 ├── core/ # 核心逻辑模块 │ ├── __init__.py │ ├── vector_store.py # 向量存储相关操作 │ └── qa_chain.py # 问答链构建 ├── docs/ # 存放待处理的文档PDF, TXT等 ├── chroma_db/ # 向量数据库持久化目录自动生成 ├── requirements.txt # 项目依赖 └── .env # 环境变量API密钥4.2 编写核心逻辑1. 安装额外依赖在requirements.txt中添加langchain langchain-openai langchain-community chromadb sentence-transformers pypdf python-dotenv streamlit streamlit-chat运行pip install -r requirements.txt。2. 实现向量存储模块 (core/vector_store.py)# core/vector_store.py import os from typing import List from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.schema import Document class VectorStoreManager: def __init__(self, persist_directory: str ./chroma_db): self.persist_directory persist_directory # 使用OpenAI的嵌入模型也可替换为 HuggingFaceEmbeddings 以本地运行 self.embeddings OpenAIEmbeddings() self.text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, separators[\n\n, \n, 。, , , , , , ] ) def load_and_split_documents(self, file_paths: List[str]) - List[Document]: 加载并分割文档 all_docs [] for file_path in file_paths: ext os.path.splitext(file_path)[-1].lower() try: if ext .pdf: loader PyPDFLoader(file_path) elif ext .txt: loader TextLoader(file_path, encodingutf-8) elif ext .md: loader UnstructuredMarkdownLoader(file_path) else: print(f暂不支持 {ext} 格式的文件: {file_path}) continue docs loader.load() print(f成功加载 {file_path}, 共 {len(docs)} 页/段。) all_docs.extend(docs) except Exception as e: print(f加载文件 {file_path} 时出错: {e}) # 分割文档 split_docs self.text_splitter.split_documents(all_docs) print(f文档分割完成共得到 {len(split_docs)} 个文本块。) return split_docs def create_vector_store(self, documents: List[Document], collection_name: str my_docs): 创建并持久化向量存储 vectorstore Chroma.from_documents( documentsdocuments, embeddingself.embeddings, persist_directoryself.persist_directory, collection_namecollection_name ) vectorstore.persist() # 显式持久化到磁盘 print(f向量存储已创建并保存至 {self.persist_directory}) return vectorstore def get_retriever(self, collection_name: str my_docs, search_kwargs: dict None): 获取检索器 if search_kwargs is None: search_kwargs {k: 4} # 默认返回4个最相关片段 vectorstore Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings, collection_namecollection_name ) return vectorstore.as_retriever(search_kwargssearch_kwargs)3. 实现问答链模块 (core/qa_chain.py)# core/qa_chain.py from langchain_openai import ChatOpenAI from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain.memory import ConversationBufferMemory from core.vector_store import VectorStoreManager class QASystem: def __init__(self, vector_store_manager: VectorStoreManager): self.vs_manager vector_store_manager self.llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.2) # 稍高的temperature使回答更自然 self.memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue, output_keyresult) self.qa_chain None def initialize_chain(self, collection_name: str my_docs): 初始化带记忆的问答链 retriever self.vs_manager.get_retriever(collection_namecollection_name) # 自定义提示词模板引导模型基于上下文回答 prompt_template 你是一个专业的文档分析助手请严格根据提供的上下文来回答问题。如果上下文没有明确信息请直接说“根据提供的资料我无法回答这个问题”不要编造信息。 上下文 {context} 历史对话 {chat_history} 问题{question} 基于上下文的回答 PROMPT PromptTemplate( templateprompt_template, input_variables[context, chat_history, question] ) self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, retrieverretriever, chain_type_kwargs{ prompt: PROMPT, memory: self.memory }, return_source_documentsTrue ) print(问答链初始化完成。) def ask(self, question: str): 提问并获取答案 if not self.qa_chain: raise ValueError(请先调用 initialize_chain 方法初始化问答链。) result self.qa_chain.invoke({query: question}) return { answer: result[result], source_docs: result.get(source_documents, []) } def clear_memory(self): 清空对话记忆 self.memory.clear()4.3 构建 Web 界面 (app.py)我们将使用 Streamlit 快速构建一个交互式界面。# app.py import streamlit as st from streamlit_chat import message import os from dotenv import load_dotenv from core.vector_store import VectorStoreManager from core.qa_chain import QASystem # 加载环境变量 load_dotenv() # 页面配置 st.set_page_config(page_title我的智能文档助手, page_icon, layoutwide) st.title( 个人智能文档问答助手) st.markdown(上传你的文档PDF/TXT/MD然后就可以针对文档内容进行提问了) # 初始化会话状态 if vs_manager not in st.session_state: st.session_state.vs_manager VectorStoreManager() if qa_system not in st.session_state: st.session_state.qa_system QASystem(st.session_state.vs_manager) if messages not in st.session_state: st.session_state.messages [] if vector_store_ready not in st.session_state: st.session_state.vector_store_ready False # 侧边栏文档上传与处理 with st.sidebar: st.header(文档管理) uploaded_files st.file_uploader( 选择文档, type[pdf, txt, md], accept_multiple_filesTrue ) collection_name st.text_input(知识库名称可选, valuemy_docs) if st.button(处理并创建知识库, typeprimary): if uploaded_files: with st.spinner(正在处理文档请稍候...): # 保存上传的文件到临时目录 temp_dir ./temp_docs os.makedirs(temp_dir, exist_okTrue) file_paths [] for uploaded_file in uploaded_files: file_path os.path.join(temp_dir, uploaded_file.name) with open(file_path, wb) as f: f.write(uploaded_file.getbuffer()) file_paths.append(file_path) # 加载、分割并创建向量存储 docs st.session_state.vs_manager.load_and_split_documents(file_paths) if docs: st.session_state.vs_manager.create_vector_store(docs, collection_name) st.session_state.qa_system.initialize_chain(collection_name) st.session_state.vector_store_ready True st.success(f知识库 {collection_name} 创建成功共处理 {len(docs)} 个文本块。) else: st.error(未能从上传的文件中提取出有效文本。) else: st.warning(请先上传至少一个文档。) st.divider() if st.button(清空对话历史): st.session_state.messages [] st.session_state.qa_system.clear_memory() st.rerun() # 主界面聊天区域 chat_container st.container() with chat_container: # 显示历史消息 for i, msg in enumerate(st.session_state.messages): if msg[role] user: message(msg[content], is_userTrue, keyfuser_{i}) else: message(msg[content], keyfai_{i}) # 输入区域 if st.session_state.vector_store_ready: with st.form(keychat_form, clear_on_submitTrue): user_input st.text_area( 请输入你的问题, keyinput, height80) submit_button st.form_submit_button(label发送, typeprimary) if submit_button and user_input: # 添加用户消息到历史 st.session_state.messages.append({role: user, content: user_input}) with st.spinner(正在思考...): try: # 调用问答系统 response st.session_state.qa_system.ask(user_input) answer response[answer] sources response[source_docs] # 添加AI回复到历史 st.session_state.messages.append({role: assistant, content: answer}) # 显示来源可选 if sources: with st.expander(查看回答依据的来源): for idx, doc in enumerate(sources): st.caption(f**片段 {idx1}** (相关性: ...)) st.text(doc.page_content[:300] ...) except Exception as e: st.error(f出错了{e}) st.rerun() # 刷新界面以显示新消息 else: st.info( 请在左侧上传文档并创建知识库然后开始提问。)4.4 运行与验证确保你的.env文件已正确配置OPENAI_API_KEY。在项目根目录下将你的文档如project_plan.pdf放入docs/文件夹或者直接在 Web 界面上传。在终端运行streamlit run app.py浏览器会自动打开http://localhost:8501。在侧边栏上传文档点击“处理并创建知识库”。处理完成后在主界面输入关于文档内容的问题即可获得基于文档的精准回答。5. 常见问题与排查思路在实际开发中你可能会遇到各种问题。下面是一些常见问题的排查指南。问题现象可能原因解决思路ModuleNotFoundError: No module named ‘langchain_community’LangChain 版本 0.1.0 后部分组件被移入独立包。运行pip install langchain-community。其他类似错误如langchain-openai同理。调用 OpenAI API 超时或连接错误1. 网络问题如代理设置。2. API 密钥错误或余额不足。3. 服务端问题。1. 检查网络如有需要在代码中配置openai.proxy。2. 在 OpenAI 官网检查密钥状态和余额。3. 查看 OpenAI Status 。向量检索结果不相关1. 文档分割策略不当块太大或太小。2. 嵌入模型不匹配或效果差。3. 检索参数k设置不合理。1. 调整chunk_size和chunk_overlap尝试 500-1500 字符。2. 尝试不同的嵌入模型如text-embedding-3-small。3. 调整search_kwargs如{“k”: 3, “score_threshold”: 0.5}。Agent 陷入循环或调用错误工具1. 工具描述description不够清晰。2. LLM 的temperature过高导致决策不稳定。3. 提示词Prompt指令不明确。1. 为每个工具编写精确、无歧义的描述。2. 将temperature设为 0 或一个较低的值如 0.1。3. 在系统提示词中明确约束 Agent 的行为和工具使用条件。处理长文档时内存溢出OOM1. 一次性加载所有文档内容到内存。2. 嵌入模型在 CPU 上运行处理大量数据慢。1. 使用流式或分页加载文档。2. 考虑使用更轻量的嵌入模型或将向量化过程分批进行。3. 对于超大文档先进行摘要或关键信息提取。Streamlit 应用上传文件后找不到Streamlit 每次交互都会重新运行脚本上传的文件对象是临时的。必须像示例中那样将上传的文件内容立即保存到本地目录如./temp_docs/中后续处理都基于这个本地文件路径。6. 最佳实践与进阶建议掌握了基础之后以下建议能帮助你将 LangChain 应用到更严肃的生产环境中。6.1 提示词工程优化提示词是影响模型表现的最关键因素之一。结构化使用ChatPromptTemplate明确区分system、user、assistant角色。具体化给出明确的指令、格式要求和示例Few-shot。上下文管理在 RAG 中明确告诉模型“请根据以下上下文回答”并设置拒绝回答未知问题的指令。迭代不要指望一次写出完美提示词根据输出结果不断调整和优化。6.2 生产环境部署考量依赖与版本锁定使用requirements.txt或poetry精确锁定所有包的版本避免因依赖更新导致应用崩溃。API 密钥管理永远不要将密钥硬编码在代码中。使用环境变量、密钥管理服务如 AWS Secrets Manager或配置文件。异步与性能对于高并发场景使用 LangChain 的异步接口ainvoke,astream以提高吞吐量。日志与监控为关键步骤如模型调用、工具执行、向量检索添加详细的日志记录。监控 Token 消耗、响应时间和错误率。错误处理与重试网络请求和模型调用都可能失败。实现健壮的重试机制和优雅的降级策略例如当主要模型不可用时回退到更简单的规则引擎。6.3 超越基础 RAG基础的“加载-分割-向量化-检索”流程可能不足以应对复杂场景。混合检索结合向量检索语义相似和关键词检索如 BM25提升召回率。重排序在初步检索出多个片段后使用一个更小的、专门的重排序模型对结果进行精排将最相关的放在前面。元数据过滤在向量存储时为每个片段添加元数据如来源文件、章节、创建日期。检索时可以结合元数据过滤如“只检索来自某份报告第三章节的内容”。Agentic RAG让 Agent 来管理 RAG 流程。例如先让 Agent 判断用户问题是否需要检索知识库如果需要再生成一个优化的查询语句去检索最后综合检索结果生成答案。6.4 探索 LangChain 生态LangSmithLangChain 官方提供的调试、测试、监控和评估平台。可以可视化地追踪链的每一步执行是开发和优化复杂应用的神器。LangGraph用于构建有状态、多智能体工作流的库。如果你需要设计复杂的、带循环和条件分支的 AI 应用流程如一个模拟会议的多角色辩论系统LangGraph 提供了比基础 Chain 更强大的抽象。社区集成LangChain 社区提供了数百种与不同工具、数据库、API 的集成。在着手自己造轮子前先去 LangChain Integrations 页面看看是否有现成的方案。学习 LangChain 的最佳路径是“做中学”。从一个明确的小目标开始比如“让模型总结我上传的周报”实现它遇到问题解决问题然后逐步增加复杂度。这个框架的边界就是你的想象力边界祝你构建出令人惊叹的 AI 应用。
最近在尝试将大语言模型LLM应用到实际业务中时你是否遇到过这样的困境模型本身很强大但让它读取你的私有文档、调用外部API、或者记住多轮对话内容却异常困难网上资料要么是零散的代码片段要么是晦涩的理论讲解想要构建一个可用的AI应用总感觉无从下手。如果你正为此烦恼那么你来对地方了。LangChain 正是为解决这些“最后一公里”问题而生的框架。本文将为你提供一份从零开始的 LangChain 实战指南摒弃华而不实的宣传专注于可复现的代码和清晰的逻辑。无论你是刚接触 AI 应用开发的在校学生还是希望将 LLM 能力集成到现有系统的工程师都能通过本文构建起坚实的知识体系并亲手搭建出你的第一个智能应用。1. LangChain 是什么为什么你需要它在深入代码之前我们首先要理解 LangChain 的核心价值。简单来说LangChain 是一个用于开发由语言模型驱动的应用程序的框架。它不是一个模型而是一个“粘合剂”和“工具箱”。想象一下强大的 LLM如 GPT-4、ChatGLM、通义千问是一个聪明但“与世隔绝”的大脑。它知识渊博却无法直接读取你电脑里的 PDF不能帮你查询今天的天气也无法记住十分钟前你们聊过什么。LangChain 的作用就是为这个大脑安装上“眼睛”文档读取、“手”工具调用和“记忆”历史记录。LangChain 主要解决以下核心问题数据接入如何让 LLM 处理它训练数据之外的信息比如你的公司知识库、本地文档、数据库。能力扩展如何让 LLM 执行它本身做不到的动作比如计算、搜索网页、调用业务 API。对话连贯如何让 LLM 在多轮对话中保持上下文理解指代实现连贯的交互。流程编排如何将读取文档、调用工具、生成回答等多个步骤组合成一个稳定、可控的自动化流程。市面上也有其他工具如 Dify、LangFlow它们更偏向于低代码/可视化搭建。而LangChain 的优势在于其灵活性和可编程性它提供了丰富的底层组件允许开发者进行深度定制和复杂逻辑的构建更适合集成到生产级应用中。2. 环境准备搭建你的第一个 LangChain 项目“工欲善其事必先利其器”。在开始编写激动人心的 AI 应用之前我们需要一个干净、可复现的开发环境。本节将详细指导你完成从零开始的环境搭建。2.1 基础环境配置首先确保你的操作系统上已经安装了Python。LangChain 支持 Python 3.8 及以上版本。我们推荐使用Python 3.10它在兼容性和性能上都有较好的表现。步骤 1检查 Python 环境打开你的终端Windows 上是 CMD 或 PowerShellMac/Linux 上是 Terminal输入以下命令python --version # 或 python3 --version如果显示版本号大于等于 3.8则可以进行下一步。如果没有安装请前往 Python 官网 下载并安装。步骤 2创建虚拟环境为了避免不同项目间的依赖冲突强烈建议使用虚拟环境。这里我们使用venvPython 内置。# 创建一个新的项目目录 mkdir my-langchain-project cd my-langchain-project # 创建虚拟环境环境文件夹名为 venv python -m venv venv # 激活虚拟环境 # 在 Windows 上 venv\Scripts\activate # 在 Mac/Linux 上 source venv/bin/activate激活后你的命令行提示符前通常会显示(venv)表示你已进入该虚拟环境。2.2 安装核心依赖在激活的虚拟环境中使用pip安装 LangChain。由于 LangChain 生态庞大我们通常安装其核心包以及可能用到的社区集成包。# 安装 LangChain 核心包 pip install langchain # 安装 OpenAI 集成包如果你使用 OpenAI 的模型如 GPT-3.5/4 pip install openai # 安装用于解析文档的常用工具 pip install langchain-community # 安装文本嵌入模型库用于文档向量化后续RAG会用到 pip install sentence-transformers # 安装向量数据库客户端以Chroma为例轻量易用 pip install chromadb # 安装用于读取PDF/TXT等文件的文档加载器 pip install pypdf安装完成后可以通过pip list | findstr langchainWindows或pip list | grep langchainMac/Linux来确认安装成功。2.3 配置 API 密钥大多数 LLM 服务如 OpenAI、智谱AI、百度千帆都需要 API 密钥。请务必妥善保管你的密钥不要将其提交到公开的代码仓库如 GitHub。我们以 OpenAI 为例你也可以使用其他兼容 OpenAI API 的模型服务。创建一个名为.env的文件来存储密钥并使用python-dotenv库来读取它。# 安装 python-dotenv pip install python-dotenv在项目根目录创建.env文件内容如下# .env OPENAI_API_KEY你的-openai-api-key-在这里然后创建一个 Python 文件test_env.py来测试环境是否就绪# test_env.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI # 加载 .env 文件中的环境变量 load_dotenv() # 初始化一个聊天模型 llm ChatOpenAI( modelgpt-3.5-turbo, temperature0, # 控制创造性0表示更确定性的输出 api_keyos.getenv(OPENAI_API_KEY) # 从环境变量读取密钥 ) # 进行一个简单的调用 response llm.invoke(你好请用中文简单介绍一下你自己。) print(response.content)运行这个脚本python test_env.py如果看到模型返回了一段中文的自我介绍恭喜你你的 LangChain 基础环境已经成功搭建。3. LangChain 核心概念与组件拆解LangChain 的设计是模块化的理解其核心抽象是高效使用的关键。我们将逐一拆解最重要的几个概念。3.1 Model I/O与模型对话的桥梁这是最基础的模块负责与各种 LLM 进行交互。主要包含三个部分Prompt 提示词模板用于结构化地组织输入给模型的信息。Model 模型本身如ChatOpenAI,ChatGoogleGenerativeAI等。Output Parser 输出解析器用于将模型非结构化的文本输出解析成我们程序可以处理的结构化数据如 JSON、列表。示例使用 PromptTemplate 和 OutputParserfrom langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI # 1. 定义提示词模板 prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个专业的翻译助手。), (user, 请将以下英文句子翻译成中文{input_text}) ]) # 2. 初始化模型 model ChatOpenAI(modelgpt-3.5-turbo) # 3. 定义输出解析器这里简单地将内容转为字符串 output_parser StrOutputParser() # 4. 将三者链接成一个链Chain chain prompt_template | model | output_parser # 5. 调用链 result chain.invoke({input_text: Hello, LangChain! This is a powerful framework for building LLM applications.}) print(f翻译结果{result}) # 输出翻译结果你好LangChain这是一个用于构建大型语言模型应用程序的强大框架。这个|运算符是 LangChain v0.1.0 后引入的 LCELLangChain Expression Language语法它让组件的连接变得非常直观和灵活。3.2 Retrieval让模型拥有“外部记忆”检索Retrieval是构建知识库问答系统的核心。其核心流程被称为RAGRetrieval-Augmented Generation检索增强生成。加载从源PDF、网页、数据库加载文档。分割将长文档切分成语义相关的小块Chunks。嵌入使用嵌入模型Embedding Model将文本块转换为数值向量。存储将向量存储到向量数据库Vector Store中。检索当用户提问时将问题也转换为向量并在向量数据库中查找最相似的文本块。生成将检索到的相关文本块作为上下文与问题一起交给 LLM 生成最终答案。示例创建一个简单的本地知识库from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI # 1. 加载文档这里用文本文件示例 loader TextLoader(./my_document.txt, encodingutf-8) documents loader.load() # 2. 分割文档 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块的最大字符数 chunk_overlap50 # 块之间的重叠字符数保持上下文连贯 ) texts text_splitter.split_documents(documents) # 3. 创建向量存储 embeddings OpenAIEmbeddings() # 需要 OPENAI_API_KEY vectorstore Chroma.from_documents(documentstexts, embeddingembeddings, persist_directory./chroma_db) # persist_directory 参数会将向量数据库保存到本地磁盘 # 4. 创建检索器 retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 检索最相关的3个块 # 5. 创建问答链 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 将检索到的所有内容“塞”进提示词 retrieverretriever, return_source_documentsTrue # 返回参考来源 ) # 6. 进行问答 query 我的文档中提到了哪些主要项目 result qa_chain.invoke({query: query}) print(f答案{result[result]}) print(f参考来源{result[source_documents]})3.3 Agents赋予模型使用工具的能力智能体Agent是 LangChain 中最具魅力的部分。它让 LLM 能够自主决策根据用户目标选择并调用合适的工具Tool来完成任务。你可以把 Agent 看作一个“调度员”LLM 是它的“大脑”Tools 是它的“双手”。一个典型的 Agent 工作流程是用户输入 - LLM 思考 - 选择工具 - 执行工具 - 观察结果 - 再次思考 - ... - 生成最终回答。示例创建一个能查询天气和计算的智能体from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.tools import Tool from langchain_community.utilities import WikipediaAPIWrapper import math # 1. 定义自定义工具 def calculate_sqrt(input: str) - str: 计算一个数的平方根。 try: number float(input) return str(math.sqrt(number)) except ValueError: return 输入错误请提供一个数字。 # 2. 封装工具列表 tools [ Tool( nameWikipedia, funcWikipediaAPIWrapper().run, # 使用维基百科查询工具 description当你需要查询关于人物、地点、公司、历史事件等通用事实信息时非常有用。输入应该是一个明确的搜索主题。 ), Tool( nameSquareRootCalculator, funccalculate_sqrt, description计算一个正数的平方根。输入应该是一个数字。 ) ] # 3. 创建提示词模板这是驱动Agent思考的“系统指令” prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以调用工具来回答问题。 如果你需要查询事实信息请使用Wikipedia工具。 如果你需要计算平方根请使用SquareRootCalculator工具。 请用中文回答用户。在最终回答前请清晰展示你的思考过程。), (user, {input}), (agent_scratchpad, {agent_scratchpad}), # 这是Agent记录其思考和工具调用结果的地方 ]) # 4. 初始化LLM需要支持Function Calling的模型如gpt-3.5-turbo或gpt-4 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 5. 创建Agent agent create_openai_tools_agent(llm, tools, prompt) # 6. 创建Agent执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 7. 运行Agent result agent_executor.invoke({input: 请先告诉我爱因斯坦的主要贡献然后计算256的平方根。}) print(result[output])运行上述代码你会看到verboseTrue模式下 Agent 详细的思考步骤Thought、行动Action和观察Observation最终给出结合了工具调用结果的答案。4. 完整实战构建个人智能文档问答助手现在我们将综合运用前面所学的知识构建一个完整的、带 Web 界面的个人文档问答助手。这个项目将涵盖文档加载、向量化存储、检索和问答链。4.1 项目结构设计首先创建清晰的项目目录。my_doc_qa_assistant/ ├── app.py # 主应用文件使用Streamlit构建界面 ├── core/ # 核心逻辑模块 │ ├── __init__.py │ ├── vector_store.py # 向量存储相关操作 │ └── qa_chain.py # 问答链构建 ├── docs/ # 存放待处理的文档PDF, TXT等 ├── chroma_db/ # 向量数据库持久化目录自动生成 ├── requirements.txt # 项目依赖 └── .env # 环境变量API密钥4.2 编写核心逻辑1. 安装额外依赖在requirements.txt中添加langchain langchain-openai langchain-community chromadb sentence-transformers pypdf python-dotenv streamlit streamlit-chat运行pip install -r requirements.txt。2. 实现向量存储模块 (core/vector_store.py)# core/vector_store.py import os from typing import List from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.schema import Document class VectorStoreManager: def __init__(self, persist_directory: str ./chroma_db): self.persist_directory persist_directory # 使用OpenAI的嵌入模型也可替换为 HuggingFaceEmbeddings 以本地运行 self.embeddings OpenAIEmbeddings() self.text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, separators[\n\n, \n, 。, , , , , , ] ) def load_and_split_documents(self, file_paths: List[str]) - List[Document]: 加载并分割文档 all_docs [] for file_path in file_paths: ext os.path.splitext(file_path)[-1].lower() try: if ext .pdf: loader PyPDFLoader(file_path) elif ext .txt: loader TextLoader(file_path, encodingutf-8) elif ext .md: loader UnstructuredMarkdownLoader(file_path) else: print(f暂不支持 {ext} 格式的文件: {file_path}) continue docs loader.load() print(f成功加载 {file_path}, 共 {len(docs)} 页/段。) all_docs.extend(docs) except Exception as e: print(f加载文件 {file_path} 时出错: {e}) # 分割文档 split_docs self.text_splitter.split_documents(all_docs) print(f文档分割完成共得到 {len(split_docs)} 个文本块。) return split_docs def create_vector_store(self, documents: List[Document], collection_name: str my_docs): 创建并持久化向量存储 vectorstore Chroma.from_documents( documentsdocuments, embeddingself.embeddings, persist_directoryself.persist_directory, collection_namecollection_name ) vectorstore.persist() # 显式持久化到磁盘 print(f向量存储已创建并保存至 {self.persist_directory}) return vectorstore def get_retriever(self, collection_name: str my_docs, search_kwargs: dict None): 获取检索器 if search_kwargs is None: search_kwargs {k: 4} # 默认返回4个最相关片段 vectorstore Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings, collection_namecollection_name ) return vectorstore.as_retriever(search_kwargssearch_kwargs)3. 实现问答链模块 (core/qa_chain.py)# core/qa_chain.py from langchain_openai import ChatOpenAI from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain.memory import ConversationBufferMemory from core.vector_store import VectorStoreManager class QASystem: def __init__(self, vector_store_manager: VectorStoreManager): self.vs_manager vector_store_manager self.llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.2) # 稍高的temperature使回答更自然 self.memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue, output_keyresult) self.qa_chain None def initialize_chain(self, collection_name: str my_docs): 初始化带记忆的问答链 retriever self.vs_manager.get_retriever(collection_namecollection_name) # 自定义提示词模板引导模型基于上下文回答 prompt_template 你是一个专业的文档分析助手请严格根据提供的上下文来回答问题。如果上下文没有明确信息请直接说“根据提供的资料我无法回答这个问题”不要编造信息。 上下文 {context} 历史对话 {chat_history} 问题{question} 基于上下文的回答 PROMPT PromptTemplate( templateprompt_template, input_variables[context, chat_history, question] ) self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, retrieverretriever, chain_type_kwargs{ prompt: PROMPT, memory: self.memory }, return_source_documentsTrue ) print(问答链初始化完成。) def ask(self, question: str): 提问并获取答案 if not self.qa_chain: raise ValueError(请先调用 initialize_chain 方法初始化问答链。) result self.qa_chain.invoke({query: question}) return { answer: result[result], source_docs: result.get(source_documents, []) } def clear_memory(self): 清空对话记忆 self.memory.clear()4.3 构建 Web 界面 (app.py)我们将使用 Streamlit 快速构建一个交互式界面。# app.py import streamlit as st from streamlit_chat import message import os from dotenv import load_dotenv from core.vector_store import VectorStoreManager from core.qa_chain import QASystem # 加载环境变量 load_dotenv() # 页面配置 st.set_page_config(page_title我的智能文档助手, page_icon, layoutwide) st.title( 个人智能文档问答助手) st.markdown(上传你的文档PDF/TXT/MD然后就可以针对文档内容进行提问了) # 初始化会话状态 if vs_manager not in st.session_state: st.session_state.vs_manager VectorStoreManager() if qa_system not in st.session_state: st.session_state.qa_system QASystem(st.session_state.vs_manager) if messages not in st.session_state: st.session_state.messages [] if vector_store_ready not in st.session_state: st.session_state.vector_store_ready False # 侧边栏文档上传与处理 with st.sidebar: st.header(文档管理) uploaded_files st.file_uploader( 选择文档, type[pdf, txt, md], accept_multiple_filesTrue ) collection_name st.text_input(知识库名称可选, valuemy_docs) if st.button(处理并创建知识库, typeprimary): if uploaded_files: with st.spinner(正在处理文档请稍候...): # 保存上传的文件到临时目录 temp_dir ./temp_docs os.makedirs(temp_dir, exist_okTrue) file_paths [] for uploaded_file in uploaded_files: file_path os.path.join(temp_dir, uploaded_file.name) with open(file_path, wb) as f: f.write(uploaded_file.getbuffer()) file_paths.append(file_path) # 加载、分割并创建向量存储 docs st.session_state.vs_manager.load_and_split_documents(file_paths) if docs: st.session_state.vs_manager.create_vector_store(docs, collection_name) st.session_state.qa_system.initialize_chain(collection_name) st.session_state.vector_store_ready True st.success(f知识库 {collection_name} 创建成功共处理 {len(docs)} 个文本块。) else: st.error(未能从上传的文件中提取出有效文本。) else: st.warning(请先上传至少一个文档。) st.divider() if st.button(清空对话历史): st.session_state.messages [] st.session_state.qa_system.clear_memory() st.rerun() # 主界面聊天区域 chat_container st.container() with chat_container: # 显示历史消息 for i, msg in enumerate(st.session_state.messages): if msg[role] user: message(msg[content], is_userTrue, keyfuser_{i}) else: message(msg[content], keyfai_{i}) # 输入区域 if st.session_state.vector_store_ready: with st.form(keychat_form, clear_on_submitTrue): user_input st.text_area( 请输入你的问题, keyinput, height80) submit_button st.form_submit_button(label发送, typeprimary) if submit_button and user_input: # 添加用户消息到历史 st.session_state.messages.append({role: user, content: user_input}) with st.spinner(正在思考...): try: # 调用问答系统 response st.session_state.qa_system.ask(user_input) answer response[answer] sources response[source_docs] # 添加AI回复到历史 st.session_state.messages.append({role: assistant, content: answer}) # 显示来源可选 if sources: with st.expander(查看回答依据的来源): for idx, doc in enumerate(sources): st.caption(f**片段 {idx1}** (相关性: ...)) st.text(doc.page_content[:300] ...) except Exception as e: st.error(f出错了{e}) st.rerun() # 刷新界面以显示新消息 else: st.info( 请在左侧上传文档并创建知识库然后开始提问。)4.4 运行与验证确保你的.env文件已正确配置OPENAI_API_KEY。在项目根目录下将你的文档如project_plan.pdf放入docs/文件夹或者直接在 Web 界面上传。在终端运行streamlit run app.py浏览器会自动打开http://localhost:8501。在侧边栏上传文档点击“处理并创建知识库”。处理完成后在主界面输入关于文档内容的问题即可获得基于文档的精准回答。5. 常见问题与排查思路在实际开发中你可能会遇到各种问题。下面是一些常见问题的排查指南。问题现象可能原因解决思路ModuleNotFoundError: No module named ‘langchain_community’LangChain 版本 0.1.0 后部分组件被移入独立包。运行pip install langchain-community。其他类似错误如langchain-openai同理。调用 OpenAI API 超时或连接错误1. 网络问题如代理设置。2. API 密钥错误或余额不足。3. 服务端问题。1. 检查网络如有需要在代码中配置openai.proxy。2. 在 OpenAI 官网检查密钥状态和余额。3. 查看 OpenAI Status 。向量检索结果不相关1. 文档分割策略不当块太大或太小。2. 嵌入模型不匹配或效果差。3. 检索参数k设置不合理。1. 调整chunk_size和chunk_overlap尝试 500-1500 字符。2. 尝试不同的嵌入模型如text-embedding-3-small。3. 调整search_kwargs如{“k”: 3, “score_threshold”: 0.5}。Agent 陷入循环或调用错误工具1. 工具描述description不够清晰。2. LLM 的temperature过高导致决策不稳定。3. 提示词Prompt指令不明确。1. 为每个工具编写精确、无歧义的描述。2. 将temperature设为 0 或一个较低的值如 0.1。3. 在系统提示词中明确约束 Agent 的行为和工具使用条件。处理长文档时内存溢出OOM1. 一次性加载所有文档内容到内存。2. 嵌入模型在 CPU 上运行处理大量数据慢。1. 使用流式或分页加载文档。2. 考虑使用更轻量的嵌入模型或将向量化过程分批进行。3. 对于超大文档先进行摘要或关键信息提取。Streamlit 应用上传文件后找不到Streamlit 每次交互都会重新运行脚本上传的文件对象是临时的。必须像示例中那样将上传的文件内容立即保存到本地目录如./temp_docs/中后续处理都基于这个本地文件路径。6. 最佳实践与进阶建议掌握了基础之后以下建议能帮助你将 LangChain 应用到更严肃的生产环境中。6.1 提示词工程优化提示词是影响模型表现的最关键因素之一。结构化使用ChatPromptTemplate明确区分system、user、assistant角色。具体化给出明确的指令、格式要求和示例Few-shot。上下文管理在 RAG 中明确告诉模型“请根据以下上下文回答”并设置拒绝回答未知问题的指令。迭代不要指望一次写出完美提示词根据输出结果不断调整和优化。6.2 生产环境部署考量依赖与版本锁定使用requirements.txt或poetry精确锁定所有包的版本避免因依赖更新导致应用崩溃。API 密钥管理永远不要将密钥硬编码在代码中。使用环境变量、密钥管理服务如 AWS Secrets Manager或配置文件。异步与性能对于高并发场景使用 LangChain 的异步接口ainvoke,astream以提高吞吐量。日志与监控为关键步骤如模型调用、工具执行、向量检索添加详细的日志记录。监控 Token 消耗、响应时间和错误率。错误处理与重试网络请求和模型调用都可能失败。实现健壮的重试机制和优雅的降级策略例如当主要模型不可用时回退到更简单的规则引擎。6.3 超越基础 RAG基础的“加载-分割-向量化-检索”流程可能不足以应对复杂场景。混合检索结合向量检索语义相似和关键词检索如 BM25提升召回率。重排序在初步检索出多个片段后使用一个更小的、专门的重排序模型对结果进行精排将最相关的放在前面。元数据过滤在向量存储时为每个片段添加元数据如来源文件、章节、创建日期。检索时可以结合元数据过滤如“只检索来自某份报告第三章节的内容”。Agentic RAG让 Agent 来管理 RAG 流程。例如先让 Agent 判断用户问题是否需要检索知识库如果需要再生成一个优化的查询语句去检索最后综合检索结果生成答案。6.4 探索 LangChain 生态LangSmithLangChain 官方提供的调试、测试、监控和评估平台。可以可视化地追踪链的每一步执行是开发和优化复杂应用的神器。LangGraph用于构建有状态、多智能体工作流的库。如果你需要设计复杂的、带循环和条件分支的 AI 应用流程如一个模拟会议的多角色辩论系统LangGraph 提供了比基础 Chain 更强大的抽象。社区集成LangChain 社区提供了数百种与不同工具、数据库、API 的集成。在着手自己造轮子前先去 LangChain Integrations 页面看看是否有现成的方案。学习 LangChain 的最佳路径是“做中学”。从一个明确的小目标开始比如“让模型总结我上传的周报”实现它遇到问题解决问题然后逐步增加复杂度。这个框架的边界就是你的想象力边界祝你构建出令人惊叹的 AI 应用。
实战应用)
