零基础从入门到精通 LangChain（保姆级超详细教程）下篇

糕692

965人浏览 · 2026-04-16 23:12:39

糕692 · 2026-04-16 23:12:39 发布

前言：从基础Demo到生产级落地，这一篇就够了

在上篇中，我们已经掌握了LangChain的所有基础组件，学会了“搭积木”——搞懂了每个核心组件的作用、用法，能跑通简单的对话应用、完成文档的加载与拆分。

而这一篇，我们要完成的是“造房子”——把这些零散的积木，组合成完整的、可落地的、能解决真实业务问题的生产级大模型应用。我们会从RAG检索增强生成这个最核心的落地场景开始，一步步深入到工具调用、记忆管理、AI Agent、LangGraph可控工作流，最后落地到生产级工程化最佳实践，彻底打通从入门到精通的全链路。

下篇完整知识体系

RAG核心实现：从零搭建生产级检索增强生成系统，从基础链路到进阶优化，彻底解决大模型幻觉与私有数据接入问题
工具调用：给大模型装上“手脚”，突破原生能力边界，实现自定义工具与内置工具的全场景调用
记忆模块：让大模型拥有“长期记忆”，实现连贯多轮对话，适配长对话场景与多用户隔离
Agent开发：让大模型拥有自主思考、规划、执行、反思的能力，完成复杂多步骤任务
LangGraph工作流：构建可控、可观测、可调试的生产级复杂Agent，解决传统Agent不可控、易死循环的痛点
生产级工程化：从Demo到落地的最后一公里，包括项目规范、接口封装、部署方案、安全防护、性能优化

话不多说，我们直接进入核心内容！

一、RAG 核心实现：从零搭建生产级检索增强生成系统

RAG（Retrieval-Augmented Generation，检索增强生成）是目前LangChain最核心、落地最广泛的应用场景，没有之一。它完美解决了大模型的三大致命痛点：幻觉问题、知识过时问题、私有数据接入问题。

1.1 先彻底搞懂：RAG的核心原理与完整链路

用最通俗的类比：RAG就像给大模型配备了「专属私人图书馆+智能检索专员」。

离线建库环节：把你的私有文档（PDF、Word、企业知识库、业务数据）做加载、拆分、向量化，存入向量数据库，相当于把图书分类归档到图书馆，给每一页内容建立精准的语义索引。
在线问答环节：用户提问时，检索专员先把用户问题转换成向量，在图书馆里找到语义最相关的内容，再把「用户问题+检索到的权威上下文」一起打包给大模型，强制大模型严格基于检索内容回答，从根源上杜绝胡编乱造。

RAG标准全链路（必须牢记）

用户提问 → 问题向量化 → 向量数据库相似性检索 → 检索结果重排/过滤 → 构建提示词（问题+相关上下文） → 大模型生成回答 → 返回给用户

上篇我们学习的「文档加载与拆分」，就是RAG的前置离线环节，是整个系统的基础。

1.2 核心前置组件：嵌入模型与向量数据库

1.2.1 嵌入模型（Embedding Model）

嵌入模型是RAG的“眼睛”，核心作用是把文本转换成固定维度的高维向量，语义越相似的文本，向量在空间中的距离越近。
比如“LangChain怎么搭建RAG”和“如何用LangChain实现检索增强生成”，两句话字面不同但语义完全一致，嵌入模型会将其转换为距离极近的向量，实现精准检索。

LangChain对所有主流嵌入模型做了统一接口封装，换模型仅需修改实例化代码，业务逻辑完全无需改动。

嵌入模型完整接入代码

① OpenAI嵌入模型（通用首选，性价比最高）

from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings

load_dotenv()

# 实例化嵌入模型，text-embedding-3-small是目前性价比最高的通用模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=1536  # 向量维度，small模型默认1536维
)

# 测试：文本转向量
text = "LangChain是一个大语言模型应用开发框架"
vector = embeddings.embed_query(text)

print(f"向量维度：{len(vector)}")
print(f"向量前10个值：{vector[:10]}")

② 国内嵌入模型替代方案（通义千问，国内用户零门槛）

from dotenv import load_dotenv
from langchain_community.embeddings import DashScopeEmbeddings
import os

load_dotenv()

# 实例化通义千问嵌入模型，免费额度充足，国内访问无延迟
embeddings = DashScopeEmbeddings(
    model="text-embedding-v2",
    dashscope_api_key=os.getenv("DASHSCOPE_API_KEY")
)

# 测试转换
text = "LangChain是一个大语言模型应用开发框架"
vector = embeddings.embed_query(text)
print(f"向量维度：{len(vector)}")

1.2.2 向量数据库

向量数据库是RAG的“专属图书馆”，专门用于存储文本对应的向量，核心能力是毫秒级相似性检索，能在百万级、亿级向量中，快速找到与查询向量最相关的TopN个结果。

新手入门首选：Chroma，轻量级本地向量数据库，无需部署任何服务，纯Python安装即可使用，完美适配LangChain，零基础零门槛。
生产级环境推荐：Milvus、Pinecone、Weaviate、PGVector（PostgreSQL插件），支持分布式部署、亿级向量检索、高并发访问。

Chroma向量数据库完整实现代码

# 先安装依赖：pip install chromadb langchain-chroma
from dotenv import load_dotenv
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

load_dotenv()

# 1. 初始化嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# ----------------------
# 离线环节：文档建库
# ----------------------
# 2. 加载并拆分文档（完全复用上篇的成熟方案）
loader = PyPDFLoader("./test.pdf")
documents = loader.load()

# 中文优化版文本拆分器
text_splitter = RecursiveCharacterTextSplitter(
    separators=["\n\n", "\n", "。", "！", "？", " ", ""],
    chunk_size=500,
    chunk_overlap=100,
    length_function=len
)
split_docs = text_splitter.split_documents(documents)

# 3. 创建Chroma向量数据库，持久化存储到本地
# persist_directory：本地持久化路径，重启程序数据不丢失
vector_store = Chroma(
    collection_name="my_rag_db",  # 集合名称，相当于数据库的表
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

# 批量添加文档到向量数据库
vector_store.add_documents(documents=split_docs)
print(f"成功添加{len(split_docs)}个文档块到向量数据库")

# ----------------------
# 在线环节：相似性检索
# ----------------------
# 4. 测试检索：用户提问，返回Top3最相关的文档块
query = "LangChain怎么搭建RAG系统？"
retrieve_docs = vector_store.similarity_search(query, k=3)

print(f"检索到{len(retrieve_docs)}个相关文档块")
print("-"*50)
for i, doc in enumerate(retrieve_docs):
    print(f"第{i+1}个相关文档：")
    print(f"内容：{doc.page_content}")
    print(f"来源：{doc.metadata}")
    print("-"*50)

运行这段代码，你就完成了RAG最核心的「建库+检索」环节，本地会生成chroma_db文件夹，所有向量和文档都会持久化存储，重启程序不会丢失。

1.3 完整的基础RAG问答链实现

现在，我们用LCEL把「检索→提示词模板→大模型→输出解析器」完整串联，实现一个可直接使用、能杜绝幻觉的RAG问答系统，每一行都有详细注释。

from dotenv import load_dotenv
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

load_dotenv()

# ----------------------
# 1. 初始化核心组件
# ----------------------
# 嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 大模型（temperature=0，最大化确定性，杜绝幻觉）
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 加载已建好的向量数据库
vector_store = Chroma(
    collection_name="my_rag_db",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)
# 转换为检索器，k=3表示每次检索Top3相关文档
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

# ----------------------
# 2. 定义RAG专属提示词模板（防幻觉核心）
# ----------------------
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """
你是一个专业、严谨的知识库问答助手，必须严格遵守以下规则：
1. 只能使用下方提供的【检索上下文】回答用户问题，绝对不允许使用自身预训练知识。
2. 如果【检索上下文】中没有相关内容，必须直接回答："抱歉，我在知识库中没有找到相关内容，无法为您解答。"，绝对禁止编造信息。
3. 回答必须准确、清晰、简洁，保留原文专业术语和数据，不篡改原意。
4. 回答结束后，必须标注内容来源的文件名和页码。

【检索上下文】：
{context}
"""),
    ("human", "{question}")
])

# ----------------------
# 3. 辅助函数：把检索到的多个文档拼接为完整上下文
# ----------------------
def format_docs(docs):
    return "\n\n".join([
        f"文档内容：{doc.page_content}\n来源：{doc.metadata.get('source', '未知')}，第{doc.metadata.get('page', '未知')}页" 
        for doc in docs
    ])

# ----------------------
# 4. 用LCEL构建完整RAG链
# ----------------------
# 链路逻辑：
# 1. 传入用户问题question
# 2. 分支1：question传入retriever检索，结果通过format_docs拼接为context
# 3. 分支2：question通过RunnablePassthrough原封不动传递
# 4. context和question传入提示词模板，再传给大模型，最后解析输出
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | model
    | StrOutputParser()
)

# ----------------------
# 5. 测试RAG问答链
# ----------------------
question = "LangChain搭建RAG系统的核心步骤是什么？"
result = rag_chain.invoke(question)

print(f"用户问题：{question}")
print(f"助手回答：{result}")

恭喜你！运行这段代码，你就从零搭建了一个完整的、可落地的RAG知识库问答系统，这就是市面上90%AI知识库产品的核心原型。

1.4 RAG进阶优化：从Demo到生产级的核心技巧

基础RAG能跑通，但在实际生产环境中，会遇到检索不精准、上下文冗余、长文档语义丢失、召回率不足等问题，这里给大家讲6个生产级必备的优化方案，每个都有可直接运行的代码。

1.4.1 多查询检索（MultiQueryRetriever）：解决用户提问表述偏差

用户的提问往往是口语化、表述不精准的，单query检索很容易漏掉相关内容。MultiQueryRetriever会让大模型基于用户原始问题，生成3-5个不同表述的查询语句，同时执行检索，合并所有结果，大幅提升内容召回率。

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI

# 初始化大模型
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 创建多查询检索器
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vector_store.as_retriever(search_kwargs={"k": 3}),
    llm=model,
    include_original=True  # 保留原始查询，确保核心内容不丢失
)

# 测试检索
query = "LangChain RAG怎么用？"
retrieve_docs = multi_query_retriever.invoke(query)
print(f"多查询检索到{len(retrieve_docs)}个相关文档")

1.4.2 重排（Rerank）：解决检索结果排序不精准问题

相似性检索是基于向量距离的，Top1的文档不一定是语义最匹配的。重排的核心逻辑是：先用向量检索召回Top10-20个候选文档，再用专门的重排模型，重新计算每个文档与用户问题的语义相关性，只保留Top3-5最匹配的结果，大幅提升检索精准度。

# 安装依赖：pip install langchain-cohere
from langchain.retrievers.contextual_compression import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
from dotenv import load_dotenv
import os

load_dotenv()

# 基础检索器：先召回Top10候选文档
base_retriever = vector_store.as_retriever(search_kwargs={"k": 10})

# 初始化重排模型，Cohere Rerank支持多语言，免费额度充足
compressor = CohereRerank(
    model="rerank-multilingual-v3.0",
    cohere_api_key=os.getenv("COHERE_API_KEY"),
    top_n=3  # 重排后仅保留Top3最相关文档
)

# 创建带重排的检索器
rerank_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)

# 测试检索
query = "LangChain搭建RAG系统的核心步骤是什么？"
retrieve_docs = rerank_retriever.invoke(query)
print(f"重排后保留{len(retrieve_docs)}个最相关文档")

1.4.3 父文档检索器（ParentDocumentRetriever）：解决长文档语义丢失问题

基础拆分把文档拆成小块，检索时只能拿到碎片化内容，丢失上下文语义。父文档检索器的逻辑是：拆分为「父文档（大的完整段落）」和「子文档（小的检索块）」，子文档用于精准检索，检索命中后返回对应的完整父文档，既保证了检索精准度，又保留了完整的上下文语义。

from langchain.retrievers import ParentDocumentRetriever
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.storage import InMemoryStore
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# 1. 初始化核心组件
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma(collection_name="parent_child_db", embedding_function=embeddings)
store = InMemoryStore()  # 存储完整父文档

# 2. 定义两级拆分器
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)  # 父文档：大段落
child_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)    # 子文档：检索块

# 3. 创建父文档检索器
parent_retriever = ParentDocumentRetriever(
    vectorstore=vector_store,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

# 4. 添加文档
loader = PyPDFLoader("./test.pdf")
documents = loader.load()
parent_retriever.add_documents(documents)

# 5. 测试检索：检索到子文档，返回完整父文档
retrieve_docs = parent_retriever.invoke("LangChain RAG的核心原理")
print(f"检索到的父文档长度：{len(retrieve_docs[0].page_content)}")

除此之外，还有HyDE假设文档嵌入、混合检索（关键词+向量）、上下文压缩、元数据过滤等优化方案，核心目标都是提升检索的召回率和精准度，为大模型提供更高质量的上下文。

二、工具调用：给大模型装上“手脚”，突破原生能力边界

原生大模型有很多天生的能力短板：不会做精准的数学计算、无法获取实时信息、不能调用外部API、无法操作数据库、不能读写本地文件。而工具调用（Tool Calling），就是给大模型装上“手脚”，让大模型能自主调用外部工具，突破原生能力的边界。

2.1 工具调用的核心原理

工具调用的本质，是让大模型具备「自主决策能力」，完整流程如下：

决策判断：收到用户问题后，大模型先判断：这个问题能否用自身原生能力解决？如果不能，需要调用什么工具？
参数生成：如果需要调用工具，大模型会生成符合工具要求的参数，输出标准化的工具调用指令；
工具执行：程序接收到指令后，执行对应的工具，拿到工具返回的结果；
结果生成：把工具返回的结果和用户原始问题一起传给大模型，大模型基于工具结果生成最终回答。

LangChain已经把整个流程做了标准化封装，你只需要定义工具，剩下的逻辑都由框架处理。

2.2 工具的3种定义方式

LangChain里定义工具的方式有3种，从简单到复杂，覆盖所有业务场景。

2.2.1 方式一：@tool装饰器自定义工具（最简单、最常用）

这是日常开发中最推荐的方式，只需要给Python函数加一个@tool装饰器，加上详细的注释，就能把函数转换成大模型可调用的工具。大模型会自动读取函数的注释、参数类型、返回值，生成正确的调用指令，零额外配置。

完整示例：自定义3个高频工具

from dotenv import load_dotenv
from langchain_core.tools import tool
import requests
from datetime import datetime

load_dotenv()

# ----------------------
# 工具1：精准数学计算器
# ----------------------
@tool
def calculator(expression: str) -> str:
    """
    精准的数学计算器，仅用于计算数学表达式，支持加减乘除、括号、幂运算、百分比计算。
    注意：必须传入合法的数学表达式，禁止传入非数学内容。
    
    参数:
        expression: 合法的数学表达式字符串，例如"1+2*3"、"(10+5)/3"
    返回:
        计算结果的字符串
    """
    try:
        # 生产环境建议使用sympy等安全计算库，避免eval安全风险
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

# ----------------------
# 工具2：实时天气查询
# ----------------------
@tool
def get_weather(city: str) -> str:
    """
    查询指定城市的实时天气信息，包括温度、天气状况、风力。
    注意：必须传入中文城市名称，禁止传入拼音、英文或非城市名称。
    
    参数:
        city: 中文城市名称，例如"北京"、"上海"、"广州"
    返回:
        天气信息的完整字符串
    """
    try:
        # 免费天气API，生产环境可替换为商用API
        url = f"https://wttr.in/{city}?format=4"
        response = requests.get(url, timeout=10)
        if response.status_code == 200:
            return response.text.strip()
        else:
            return f"查询失败，无法获取{city}的天气信息"
    except Exception as e:
        return f"天气查询错误：{str(e)}"

# ----------------------
# 工具3：当前时间获取
# ----------------------
@tool
def get_current_time() -> str:
    """
    获取当前系统的北京时间，精确到分钟。
    不需要传入任何参数。
    """
    return datetime.now().strftime("%Y年%m月%d日 %H:%M")

# 把所有工具放入列表，统一管理
tools = [calculator, get_weather, get_current_time]

# 查看工具的元数据，大模型就是通过这些信息理解工具的
print("工具名称：", calculator.name)
print("工具描述：", calculator.description)
print("工具参数规范：", calculator.args)

2.2.2 方式二：使用LangChain内置工具

LangChain已经内置了上百个开箱即用的工具，覆盖搜索引擎、文件操作、数据库、Shell命令、GitHub、企业微信、飞书等场景，直接导入即可使用，无需自己开发。

示例：内置AI专用搜索引擎Tavily

# 安装依赖：pip install tavily-python
from langchain_community.tools.tavily_search import TavilySearchResults
from dotenv import load_dotenv
import os

load_dotenv()

# 初始化搜索引擎工具，Tavily是专为AI Agent设计的搜索引擎，免费额度充足
search_tool = TavilySearchResults(
    max_results=3,
    tavily_api_key=os.getenv("TAVILY_API_KEY")
)

# 测试工具
result = search_tool.invoke("2026年LangChain最新稳定版本号")
print("搜索结果：", result)

2.3 工具调用全流程实现

现在，我们把定义好的工具绑定到大模型上，实现完整的工具调用流程，支持大模型自主决策、多工具轮询调用。

LangChain对所有支持函数调用的大模型，都提供了统一的bind_tools方法，OpenAI、通义千问、豆包、文心一言等主流模型用法完全一致。

完整可运行代码：工具调用全流程

from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, ToolMessage
import requests

load_dotenv()

# ----------------------
# 1. 复用前面定义的3个工具
# ----------------------
@tool
def calculator(expression: str) -> str:
    """精准的数学计算器，仅用于计算数学表达式，支持加减乘除、括号、幂运算。"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

@tool
def get_weather(city: str) -> str:
    """查询指定城市的实时天气信息，必须传入中文城市名称。"""
    try:
        url = f"https://wttr.in/{city}?format=4"
        response = requests.get(url, timeout=10)
        return response.text.strip() if response.status_code == 200 else f"查询{city}天气失败"
    except Exception as e:
        return f"天气查询错误：{str(e)}"

@tool
def get_current_time() -> str:
    """获取当前北京时间，精确到分钟，无需传入参数。"""
    from datetime import datetime
    return datetime.now().strftime("%Y年%m月%d日 %H:%M")

tools = [calculator, get_weather, get_current_time]

# ----------------------
# 2. 实例化大模型，绑定工具
# ----------------------
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 核心：把工具绑定到大模型，让大模型知道自己可以调用这些工具
model_with_tools = model.bind_tools(tools)

# ----------------------
# 3. 测试1：无需调用工具的问题
# ----------------------
print("测试1：无需调用工具的问题")
messages = [HumanMessage(content="你好，请问LangChain是什么？")]
response = model_with_tools.invoke(messages)
print("AI回复：", response.content)
print("是否调用工具：", response.tool_calls)
print("-"*50)

# ----------------------
# 4. 测试2：需要调用工具的问题
# ----------------------
print("测试2：需要调用工具的问题")
messages = [HumanMessage(content="北京今天的天气怎么样？")]
response = model_with_tools.invoke(messages)
print("AI回复：", response.content)
print("工具调用指令：", response.tool_calls)
print("-"*50)

# ----------------------
# 5. 完整工具调用闭环：执行工具→返回结果→生成最终回答
# ----------------------
print("完整工具调用闭环演示")
# 第一步：用户提问
messages = [HumanMessage(content="356乘以789加上1234的结果是多少？")]
# 第二步：调用大模型，生成工具调用指令
ai_response = model_with_tools.invoke(messages)
# 把AI回复加入消息列表，保留上下文
messages.append(ai_response)

# 第三步：遍历工具调用指令，执行对应工具
for tool_call in ai_response.tool_calls:
    # 根据工具名称匹配对应的工具
    selected_tool = {
        "calculator": calculator,
        "get_weather": get_weather,
        "get_current_time": get_current_time
    }[tool_call["name"].lower()]
    # 执行工具，传入参数
    tool_output = selected_tool.invoke(tool_call["args"])
    # 把工具返回结果加入消息列表
    messages.append(ToolMessage(tool_output, tool_call_id=tool_call["id"]))

# 第四步：把包含工具结果的消息列表传给大模型，生成最终回答
final_response = model_with_tools.invoke(messages)
print("最终回答：", final_response.content)

运行这段代码，你会清晰看到工具调用的完整闭环：

对于无需工具的问题，大模型直接回答，不会触发工具调用；
对于需要工具的问题，大模型会生成精准的调用指令，执行工具后，基于工具结果生成最终回答，完美解决大模型数学计算不准、实时信息缺失的问题。

三、记忆模块：让大模型拥有“长期记忆”，实现连贯多轮对话

原生大模型是无状态的，每一次调用都是独立的，它不会记住上一轮对话的内容，这对于对话机器人、AI Agent来说，是致命的短板。

LangChain的记忆模块（Memory），就是专门解决这个问题的核心组件，它能自动记录对话上下文，在每一轮对话时，把历史对话一起传给大模型，让大模型拥有“长期记忆”，实现连贯的多轮对话。

3.1 记忆模块的核心设计逻辑

记忆模块的核心，是自动化的对话历史管理，核心能力包括：

自动存储每一轮的用户消息（HumanMessage）和AI回复（AIMessage）；
每一轮对话前，自动加载历史对话，拼接到提示词中；
支持多种记忆策略，解决长对话上下文溢出的问题；
支持多用户隔离与持久化，重启程序记忆不丢失。

3.2 4种常用记忆类型与完整实现

LangChain提供了多种记忆类型，适配不同的业务场景，这里我们讲4种最常用、生产级落地最多的方案。

3.2.1 基础缓存记忆：ConversationBufferMemory

这是最基础的记忆类型，会把所有对话历史原封不动地存储下来，每一轮都把完整历史传给大模型，适合短对话、客服机器人等场景，零配置开箱即用。

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

load_dotenv()

# 1. 初始化大模型
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# 2. 定义带记忆的提示词模板
# MessagesPlaceholder("chat_history") 是历史对话的占位符，会自动填充历史消息
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好、贴心的对话助手，会记住和用户的所有对话内容，回答亲切自然。"),
    MessagesPlaceholder(variable_name="chat_history"),  # 历史对话占位符
    ("human", "{input}")
])

# 3. 构建基础对话链
chain = prompt | model | StrOutputParser()

# 4. 定义对话历史存储，实现多用户隔离
# 生产环境可替换为Redis、SQLite、MySQL等持久化存储
store = {}

def get_session_history(session_id: str):
    """根据会话ID获取对话历史，不同会话ID的记忆完全隔离"""
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 5. 给链加上记忆能力
chain_with_memory = RunnableWithMessageHistory(
    runnable=chain,
    get_session_history=get_session_history,
    input_messages_key="input",  # 用户输入的key
    history_messages_key="chat_history"  # 历史对话的key，必须和模板占位符一致
)

# ----------------------
# 测试多轮对话
# ----------------------
# 第一轮对话
print("第一轮对话：")
response1 = chain_with_memory.invoke(
    {"input": "你好，我叫小明，今年25岁，在广州工作"},
    config={"configurable": {"session_id": "user_xiaoming"}}  # 会话ID，区分不同用户
)
print("AI回复：", response1)
print("-"*50)

# 第二轮对话，无需重复告知信息，AI会记住
print("第二轮对话：")
response2 = chain_with_memory.invoke(
    {"input": "请问我叫什么名字，在哪里工作，今年多大？"},
    config={"configurable": {"session_id": "user_xiaoming"}}
)
print("AI回复：", response2)

运行这段代码，你会发现AI完美记住了你的个人信息，实现了连贯的多轮对话。其中session_id是多用户隔离的核心，不同的session_id对应完全独立的记忆，完美适配多用户场景。

3.2.2 窗口记忆：ConversationBufferWindowMemory

基础缓存记忆会存储所有历史对话，对话轮次多了之后，会超出大模型的上下文窗口，导致报错和成本飙升。
窗口记忆只会记住最近K轮的对话，自动丢弃更早的对话，完美控制上下文长度，适合长对话、闲聊机器人等场景。

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.messages import trim_messages

load_dotenv()

model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# 提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的对话助手，会记住和用户的最近对话内容。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}")
])

# 核心：定义消息修剪器，仅保留最近3轮对话，控制上下文长度
trimmer = trim_messages(
    max_tokens=800,  # 历史对话的最大token数
    strategy="last",  # 保留最后面的消息
    token_counter=model,
    include_system=True,
    allow_partial=False,
    start_on="human",
)

# 构建带修剪的对话链
from langchain_core.runnables import RunnablePassthrough
chain = (
    RunnablePassthrough.assign(chat_history=lambda x: trimmer.invoke(x["chat_history"]))
    | prompt
    | model
    | StrOutputParser()
)

# 对话历史存储
store = {}
def get_session_history(session_id: str):
    from langchain_community.chat_message_histories import ChatMessageHistory
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 加上记忆能力
chain_with_window_memory = RunnableWithMessageHistory(
    runnable=chain,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

3.2.3 摘要记忆：ConversationSummaryMemory（长对话最优方案）

窗口记忆会丢弃早期对话，若用户问起早期内容，AI会完全遗忘。摘要记忆的核心逻辑是：不原封不动存储所有对话，而是用大模型自动把历史对话生成摘要，仅把摘要传给大模型，既保留了核心信息，又严格控制了上下文长度，是长对话场景的最优方案。

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import SQLChatMessageHistory

load_dotenv()

model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# 提示词模板，加入对话历史摘要
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的对话助手，结合对话历史摘要和最近对话内容，回答用户的问题。\n对话历史摘要：{summary}"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}")
])

# 用SQLite做持久化存储，重启程序记忆不会丢失
def get_session_history(session_id: str):
    return SQLChatMessageHistory(
        session_id=session_id,
        connection="sqlite:///chat_history.db"  # 本地SQLite数据库文件
    )

3.3 记忆的持久化

上面的示例用的是内存存储，重启程序后记忆就会丢失，生产环境必须做持久化，LangChain支持多种成熟的持久化方案：

SQLite/SQL数据库：用SQLChatMessageHistory，适合中小规模场景，零额外部署，开箱即用；
Redis：适合高并发、分布式场景，性能极高，支持过期时间管理；
PostgreSQL：适合大规模多用户场景，支持复杂查询和权限管理；
MongoDB：适合非结构化对话数据存储，灵活性极高。

仅需把代码中的ChatMessageHistory替换为对应的持久化存储类，用法完全一致，零额外开发成本。

四、Agent 开发：让大模型拥有自主思考与行动能力

Agent（智能体）是LangChain最核心的进阶能力，也是大模型应用的未来方向。

我们前面学的链（Chain），是预定义好的固定流程，无论什么问题，都按固定的步骤执行；而Agent，是让大模型作为“大脑”，自主思考、规划、决策、执行、反思，根据用户的问题，自己决定下一步要做什么、调用什么工具，完成复杂的多步骤任务。

4.1 Agent的核心原理：ReAct框架

目前最主流、最稳定、落地最多的Agent框架是ReAct（Reasoning + Acting），也就是「思考-行动-观察」循环，也是LangChain官方推荐的标准实现方式。

ReAct的完整循环逻辑：

思考（Thought）：大模型收到用户问题后，先拆解任务，思考：我要完成这个目标，第一步需要做什么？需要调用什么工具？
行动（Action）：大模型决定执行的动作，也就是调用对应的工具，传入正确的参数；
观察（Observation）：执行工具，拿到工具返回的结果，也就是观察到的信息；
循环迭代：大模型基于观察到的信息，继续思考下一步要做什么，直到任务完成，生成最终回答。

简单来说，ReAct Agent就像人解决问题的过程：先想第一步做什么，做完看结果，再想第二步做什么，直到把问题彻底解决。

4.2 从零搭建第一个ReAct Agent（完整可运行代码）

我们用前面学的工具、记忆、大模型，从零搭建一个完整的ReAct Agent，支持多工具调用、多轮对话、记忆能力，每一行都有详细注释。

from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain import hub
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
import requests

load_dotenv()

# ----------------------
# 1. 定义Agent需要用到的工具
# ----------------------
@tool
def calculator(expression: str) -> str:
    """精准的数学计算器，仅用于计算数学表达式，支持加减乘除、括号、幂运算。"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

@tool
def get_weather(city: str) -> str:
    """查询指定城市的实时天气信息，必须传入中文城市名称。"""
    try:
        url = f"https://wttr.in/{city}?format=4"
        response = requests.get(url, timeout=10)
        return response.text.strip() if response.status_code == 200 else f"查询{city}天气失败"
    except Exception as e:
        return f"天气查询错误：{str(e)}"

@tool
def tavily_search(query: str) -> str:
    """联网搜索引擎，用于查询实时信息、新闻、百科、最新数据等，所有未知信息都可以用这个工具查询。"""
    from langchain_community.tools.tavily_search import TavilySearchResults
    search_tool = TavilySearchResults(max_results=3)
    result = search_tool.invoke(query)
    return str(result)

tools = [calculator, get_weather, tavily_search]

# ----------------------
# 2. 初始化大模型
# ----------------------
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# ----------------------
# 3. 定义ReAct Agent的提示词模板
# 这里使用LangChain官方Hub的成熟ReAct模板，也可以完全自定义
# ----------------------
prompt = hub.pull("hwchase17/react-chat")

# ----------------------
# 4. 创建ReAct Agent
# ----------------------
agent = create_react_agent(model, tools, prompt)

# ----------------------
# 5. 创建Agent执行器，负责管理Agent的思考-行动-观察循环
# ----------------------
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 开启详细日志，能看到Agent完整的思考、行动、观察过程，新手必开
    handle_parsing_errors=True,  # 自动处理解析错误，避免Agent卡死
    max_iterations=10,  # 最大循环次数，避免Agent无限死循环
)

# ----------------------
# 6. 给Agent加上记忆能力，实现多轮对话
# ----------------------
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

agent_with_memory = RunnableWithMessageHistory(
    runnable=agent_executor,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

# ----------------------
# 测试Agent：复杂多步骤任务
# ----------------------
print("Agent测试：复杂多步骤任务")
response = agent_with_memory.invoke(
    {"input": "请问上海今天的天气怎么样？适合穿什么衣服？另外，2026年世界GDP排名前3的国家是哪些？"},
    config={"configurable": {"session_id": "agent_test_001"}}
)
print("-"*50)
print("Agent最终回答：", response["output"])

运行这段代码，你会在终端里看到Agent完整的「思考-行动-观察」循环过程：

首先，Agent拆解用户的两个问题，判断第一个问题需要调用get_weather工具，第二个问题需要调用tavily_search工具；
先调用get_weather工具，拿到上海的实时天气结果；
再调用tavily_search工具，拿到2026年世界GDP排名的最新数据；
最后，把两个结果整合，生成完整的最终回答。

这就是Agent的强大之处：它能自主拆解复杂任务，分步骤执行，不需要你预定义任何流程，完全由大模型自主决策，完成人工无法预定义的复杂任务。

4.3 Agent进阶：RAG + Agent 融合

实际业务中，我们经常需要Agent既能读取私有知识库，又能联网、调用工具，这就是RAG + Agent的融合方案，仅需把RAG检索器做成一个工具，加入Agent的工具列表即可。

# 把RAG检索器封装为Agent工具
@tool
def rag_knowledge_search(query: str) -> str:
    """
    专属知识库检索工具，用于查询LangChain、RAG、大模型应用开发相关的专业知识，
    所有相关的技术问题，必须优先使用这个工具查询，禁止使用自身预训练知识。
    """
    from langchain_chroma import Chroma
    from langchain_openai import OpenAIEmbeddings
    
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vector_store = Chroma(
        collection_name="my_rag_db",
        embedding_function=embeddings,
        persist_directory="./chroma_db"
    )
    retriever = vector_store.as_retriever(search_kwargs={"k": 3})
    docs = retriever.invoke(query)
    return "\n\n".join([doc.page_content for doc in docs])

# 把RAG工具加入Agent的工具列表
tools.append(rag_knowledge_search)

这样，Agent就拥有了私有知识库的能力，遇到相关技术问题，会优先去你的专属知识库查询，完美结合了RAG的精准性和Agent的自主性。

五、LangGraph 工作流：构建可控、可观测、生产级的复杂Agent

传统的ReAct Agent虽然强大，但有一个致命的痛点：不可控。大模型的自主决策很容易出现死循环、调用错误工具、参数错误、偏离任务目标等问题，在生产环境中很难落地。

为了解决这个问题，LangChain官方推出了LangGraph，一个基于图结构的Agent开发框架，它用「节点+边」的图结构，来定义Agent的工作流，支持循环、条件分支、状态管理、错误处理，让Agent的每一步都可控、可观测、可调试，是目前生产级Agent开发的首选方案。

5.1 LangGraph的核心概念

LangGraph的核心是状态机，整个工作流的所有数据，都存在一个共享的State（状态）里，每个节点执行完之后，会更新State，然后根据State决定下一步走哪个节点。

必须牢记的核心概念：

State（状态）：整个工作流的共享数据存储，包含对话历史、工具调用结果、用户输入、中间结果等，所有节点都能读取和更新State。
Node（节点）：工作流的执行单元，是一个Python函数，接收State作为输入，执行具体操作（调用大模型、执行工具、处理数据），返回更新后的State。
Edge（边）：节点之间的连接，定义工作流的流转顺序，比如节点A执行完之后，必须执行节点B。
Conditional Edge（条件边）：根据State的内容，动态决定下一步走哪个节点，是实现循环、分支判断的核心。
START/END节点：工作流的起点和终点，START是入口，END是结束。

5.2 环境搭建

首先安装LangGraph依赖：

pip install langgraph

5.3 从零用LangGraph构建ReAct Agent（完整可运行代码）

我们用LangGraph，从零构建一个和上面功能完全一致的ReAct Agent，但是每一步都完全可控、可调试、可观测，彻底解决传统Agent的不可控痛点。

from dotenv import load_dotenv
from typing import Annotated, TypedDict, List
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
import requests

load_dotenv()

# ----------------------
# 1. 定义工具，和前面完全一致
# ----------------------
@tool
def calculator(expression: str) -> str:
    """精准的数学计算器，仅用于计算数学表达式，支持加减乘除、括号、幂运算。"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

@tool
def get_weather(city: str) -> str:
    """查询指定城市的实时天气信息，必须传入中文城市名称。"""
    try:
        url = f"https://wttr.in/{city}?format=4"
        response = requests.get(url, timeout=10)
        return response.text.strip() if response.status_code == 200 else f"查询{city}天气失败"
    except Exception as e:
        return f"天气查询错误：{str(e)}"

tools = [calculator, get_weather]

# ----------------------
# 2. 定义State（状态）：整个工作流的共享数据
# ----------------------
class State(TypedDict):
    # messages是对话历史，add_messages表示新消息会追加到列表，而非覆盖
    messages: Annotated[List[BaseMessage], add_messages]

# ----------------------
# 3. 定义节点：大模型调用节点
# ----------------------
# 实例化大模型，绑定工具
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
model_with_tools = model.bind_tools(tools)

def chatbot(state: State):
    """大模型调用节点：接收状态，调用大模型，返回更新后的消息列表"""
    return {"messages": [model_with_tools.invoke(state["messages"])]}

# ----------------------
# 4. 定义工具执行节点：用LangGraph内置的ToolNode，无需自己开发
# ----------------------
tool_node = ToolNode(tools=tools)

# ----------------------
# 5. 构建状态图
# ----------------------
# 初始化状态图，传入State类型
graph_builder = StateGraph(State)

# 把节点添加到图中
graph_builder.add_node("chatbot", chatbot)  # 大模型思考节点
graph_builder.add_node("tools", tool_node)  # 工具执行节点

# 定义边：设置工作流的流转顺序
# 1. 从START节点进入chatbot节点（工作流启动后，先执行大模型思考）
graph_builder.add_edge(START, "chatbot")

# 2. 定义条件边：chatbot执行完后，根据结果决定下一步
# tools_condition是LangGraph内置的条件判断：
# - 如果大模型返回了工具调用指令，流转到tools节点
# - 如果没有工具调用指令，流转到END节点，结束工作流
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)

# 3. 定义边：tools节点执行完后，必须回到chatbot节点，让大模型基于工具结果继续思考
graph_builder.add_edge("tools", "chatbot")

# ----------------------
# 6. 编译图，生成可执行的工作流
# ----------------------
app = graph_builder.compile()

# ----------------------
# 测试LangGraph Agent
# ----------------------
print("LangGraph Agent测试")
# 用户提问
user_input = "广州今天的天气怎么样？温度乘以2是多少？"
# 执行工作流，传入初始状态
events = app.stream(
    {"messages": [HumanMessage(content=user_input)]},
    stream_mode="values"  # 流式输出每一步的状态
)

# 遍历每一步的状态，打印完整执行过程
for event in events:
    event["messages"][-1].pretty_print()

运行这段代码，你会看到完整的可控执行过程：

首先执行chatbot节点，大模型收到问题，决定先调用get_weather工具查询广州天气；
然后执行tools节点，调用get_weather工具，拿到天气结果；
回到chatbot节点，大模型基于天气结果，决定调用calculator工具计算温度乘以2；
再次执行tools节点，调用calculator工具，拿到计算结果；
回到chatbot节点，大模型判断任务完成，无工具调用，生成最终回答，结束工作流。

整个过程完全可控，每一步的流转都由我们定义的图结构决定，不会出现死循环、偏离任务的问题，完美解决了传统Agent的不可控痛点。

5.4 LangGraph生产级进阶能力

LangGraph还有很多生产级的进阶能力，是企业级落地的核心：

持久化检查点（Checkpoint）：支持工作流中断、恢复、回溯，比如执行到一半暂停，下次重启继续执行，适合长任务、人工审核场景；
人机交互（Human-in-the-Loop）：支持在工作流中加入人工审核环节，比如工具调用前需要人工确认，适合金融、政务等高风险场景；
分支与并行执行：支持多个节点并行执行，比如同时调用多个工具、同时执行多个检索任务，大幅提升执行效率；
子图：支持把复杂的工作流拆分成多个子图，模块化开发，便于维护和迭代；
超时与错误处理：支持节点超时、重试、错误捕获与降级，生产级高可用必备。

六、生产级工程化最佳实践：从Demo到落地的最后一公里

很多同学能写出跑通的Demo，但不知道怎么把它变成能上线、能给用户用的生产级应用，这一部分，我们就讲透从Demo到落地的所有核心要点，包括项目结构、接口封装、部署方案、安全防护、性能优化。

6.1 项目结构规范

一个规范的项目结构，是可维护性、可扩展性的基础，这里给大家推荐LangChain官方推荐的生产级项目结构：

langchain-production-project/
├── .env                # 环境变量、密钥配置（必须加入.gitignore，禁止提交到Git）
├── .gitignore          # Git忽略文件
├── requirements.txt    # 依赖清单，固定版本号
├── Dockerfile          # Docker部署文件
├── pyproject.toml      # 项目配置文件
├── README.md           # 项目说明文档
├── app/                # 应用核心代码
│   ├── __init__.py
│   ├── main.py         # 应用入口，FastAPI接口
│   ├── config.py       # 配置管理
│   ├── chains/         # 链的定义
│   │   ├── __init__.py
│   │   ├── rag_chain.py
│   │   └── chat_chain.py
│   ├── agents/         # Agent和LangGraph工作流定义
│   │   ├── __init__.py
│   │   ├── react_agent.py
│   │   └── workflow.py
│   ├── tools/          # 自定义工具
│   │   ├── __init__.py
│   │   ├── weather.py
│   │   └── search.py
│   ├── utils/          # 工具函数
│   │   ├── __init__.py
│   │   ├── logger.py
│   │   └── security.py
│   └── prompts/        # 提示词模板管理
│       ├── __init__.py
│       ├── rag_prompt.py
│       └── agent_prompt.py
├── tests/              # 单元测试
│   ├── __init__.py
│   ├── test_rag.py
│   └── test_agent.py
└── data/               # 本地数据、文档存储

6.2 用FastAPI封装HTTP接口

Demo只能在本地运行，要给前端、其他服务调用，必须封装成标准化的HTTP接口。FastAPI是目前Python生态里性能最高、最适合AI应用的Web框架，完美适配LangChain的异步调用，自带接口文档，零配置开箱即用。

完整示例：封装RAG问答生产级接口

# app/main.py
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import uvicorn
import os

load_dotenv()

# 初始化FastAPI应用
app = FastAPI(
    title="LangChain RAG 生产级API",
    version="1.0.0",
    description="基于LangChain构建的知识库问答API"
)

# 配置CORS跨域，支持前端调用
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境请替换为前端域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 定义请求和响应的数据模型
class RagQueryRequest(BaseModel):
    question: str
    session_id: str
    top_k: int = 3

class RagQueryResponse(BaseModel):
    question: str
    answer: str
    source_documents: list

# 全局变量，存储核心组件
rag_chain = None
source_documents = []

# 应用启动时，一次性初始化所有核心组件，避免每次请求重复初始化
@app.on_event("startup")
async def startup_event():
    global rag_chain, embeddings, vector_store, retriever
    # 1. 初始化嵌入模型
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    # 2. 加载向量数据库
    vector_store = Chroma(
        collection_name="my_rag_db",
        embedding_function=embeddings,
        persist_directory="./chroma_db"
    )
    # 3. 初始化检索器
    retriever = vector_store.as_retriever(search_kwargs={"k": 3})
    # 4. 初始化大模型
    model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    # 5. 定义提示词模板
    rag_prompt = ChatPromptTemplate.from_messages([
        ("system", """
你是一个专业的知识库问答助手，必须严格基于检索到的上下文回答问题，不知道就说不知道，禁止编造内容。
【检索上下文】：{context}
"""),
        ("human", "{question}")
    ])
    # 6. 文档格式化函数
    def format_docs(docs):
        global source_documents
        source_documents = docs
        return "\n\n".join([doc.page_content for doc in docs])
    # 7. 构建RAG链
    rag_chain = (
        {"context": retriever | format_docs, "question": RunnablePassthrough()}
        | rag_prompt
        | model
        | StrOutputParser()
    )

# 定义RAG问答接口
@app.post("/api/rag/query", response_model=RagQueryResponse)
async def rag_query(request: RagQueryRequest):
    try:
        # 动态设置检索top_k
        retriever.search_kwargs["k"] = request.top_k
        # 调用RAG链
        answer = rag_chain.invoke(request.question)
        # 返回标准化结果
        return RagQueryResponse(
            question=request.question,
            answer=answer,
            source_documents=[
                {"content": doc.page_content, "metadata": doc.metadata} 
                for doc in source_documents
            ]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"接口调用失败：{str(e)}")

# 健康检查接口，用于服务监控
@app.get("/api/health")
async def health_check():
    return {"status": "ok", "message": "LangChain RAG API 运行正常"}

# 启动服务
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行这个文件，你就会得到一个生产级的RAG HTTP接口，自带Swagger接口文档（访问http://localhost:8000/docs即可查看），支持跨域、参数校验、异常处理，前端可直接调用。

6.3 Docker部署

生产环境部署首选Docker，它能彻底解决环境不一致的问题，一次打包，到处运行，无需在服务器上配置任何Python环境。

Dockerfile完整示例

# 基础镜像，使用Python3.11 slim版本，体积小、稳定性高
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 设置时区为北京时间
ENV TZ=Asia/Shanghai
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone

# 复制依赖文件
COPY requirements.txt .

# 安装依赖，使用国内镜像源，提升构建速度
RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 复制项目代码
COPY . .

# 暴露服务端口
EXPOSE 8000

# 启动命令
CMD ["python", "app/main.py"]

requirements.txt示例（固定版本号，避免版本冲突）

langchain==0.2.0
langchain-openai==0.1.7
langchain-community==0.2.0
langchain-text-splitters==0.2.0
langchain-chroma==0.1.1
langgraph==0.0.66
python-dotenv==1.0.1
fastapi==0.111.0
uvicorn==0.29.0
pypdf==4.2.0
beautifulsoup4==4.12.3
tiktoken==0.7.0
requests==2.32.2

构建和运行命令

# 1. 构建Docker镜像
docker build -t langchain-rag-api:v1.0 .

# 2. 运行容器，把.env文件挂载进去，避免密钥打包到镜像中
docker run -d --name langchain-rag -p 8000:8000 -v ./.env:/app/.env langchain-rag-api:v1.0

6.4 生产级核心优化要点

6.4.1 日志与可观测性

生产环境必须有完整的日志和可观测性，推荐使用LangSmith，它是LangChain官方推出的调试和可观测性平台，能追踪每一次链、Agent的调用，看到每一步的输入输出、耗时、错误，完美解决AI应用调试难、排错难的问题。

仅需在.env文件中加入以下配置，即可自动接入LangSmith：

LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=你的LangSmith API密钥
LANGCHAIN_PROJECT=你的项目名称

6.4.2 安全防护

AI应用有很多特有的安全风险，生产环境必须做好全链路防护：

提示词注入防护：用langchain-guardrails或自定义规则，检测并拦截恶意提示词注入；
数据脱敏：用户输入和文档中的敏感信息（手机号、身份证、银行卡、商业机密），必须先脱敏再处理；
权限控制：接口必须加鉴权，比如API Key、JWT，禁止未授权访问，不同用户隔离知识库；
输入校验：严格校验用户输入的长度、格式，避免恶意输入导致的报错和安全问题；
工具调用白名单：只允许Agent调用白名单内的工具，禁止执行高危操作（删除文件、执行Shell命令、修改数据库）。

6.4.3 性能与成本优化

缓存：用Redis缓存高频问题的检索结果和大模型回答，减少重复调用，降低成本，提升响应速度；
异步调用：所有大模型、工具、数据库调用，都用异步方式，大幅提升接口的并发能力；
批量处理：批量处理文档、批量调用大模型，提升处理效率；
模型分级：简单任务用小模型（比如gpt-3.5-turbo），复杂任务用大模型，大幅降低调用成本；
限流熔断：用Redis实现接口限流，避免大流量打垮服务，同时控制月度调用成本。

下篇总结 & 附加篇预告

恭喜你！坚持看完了下篇的全部内容，现在的你，已经完全掌握了LangChain从入门到精通的所有核心能力，从基础的RAG系统，到自主决策的AI Agent，再到可控的LangGraph工作流，最后到生产级的工程化落地，已经能独立开发完整的、可落地的商业级大模型应用了。

下篇我们学到了什么？

彻底搞懂了RAG的核心原理，从零搭建了生产级的检索增强生成系统，掌握了多种进阶优化技巧，从根源上解决了大模型幻觉和私有数据接入问题；
学会了工具调用的完整实现，能自定义工具、使用内置工具，给大模型装上了“手脚”，突破了原生能力的边界；
掌握了记忆模块的多种实现方式，让大模型拥有了长期记忆，实现了连贯的多轮对话，支持多用户隔离和持久化；
从零搭建了ReAct Agent，让大模型拥有了自主思考、规划、执行的能力，能完成复杂的多步骤任务；
学会了用LangGraph构建可控、可观测的生产级Agent，彻底解决了传统Agent不可控、易死循环的痛点；
掌握了生产级工程化的所有核心要点，包括项目结构、接口封装、Docker部署、安全防护、性能优化，完成了从Demo到落地的最后一公里。

附加篇预告

在附加篇里，我会给大家整理《LangChain面试八股文全集》，覆盖从基础到进阶的所有高频面试考点，包括核心概念、RAG、Agent、LangGraph、工程化、常见问题排查，每一个问题都有标准的满分答案，不管是面试求职，还是巩固知识点，都非常有用。

互动环节

如果大家在学习过程中遇到任何问题，都可以在评论区留言，我会一一解答！
也可以在评论区说说你想用LangChain做的业务场景，我会在附加篇里补充对应的面试题和实战案例！

如果这篇教程对你有帮助，欢迎点赞、收藏、关注，后续会更新附加篇的面试八股文全集，以及更多LangChain的生产级实战案例！

DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

告别千万级学费：AI编程双剑合璧 Cursor + Claude Code

DeepSeek技术社区

2026年AI协同底座深度评测：让外部Agent真正跑通企业业务流

2026年AI协同底座深度评测：打通外部Agent与企业业务流过去两年，主流外部Agent（如Cursor、ClaudeCode、Codex、GeminiCLI）在单点任务上表现卓越，但始终受限于“本地工具”定位，难以融入团队协作流。通过飞书aily协同底座的实践，实现了外部Agent与企业业务流的无缝衔接，核心逻辑为：分工明确：外部Agent专注专业领域计算（如代码生成、日志分析），协同底座

DeepSeek技术社区

2026年AI协同底座深度评测｜多外部Agent落地企业业务流的上下文管理指南

我自己作为互联网公司的技术负责人，同时也是玩了快两年的AI Power User，Codex、Cursor、Claude Code、Gemini CLI这些主流外部Agent我几乎都深度用过，单拎出来每一个的单点能力都足够惊艳：Cursor写业务代码的补全准确率能到80%以上，Claude Code处理几十万行的日志文件不用拆分，Codex爬取公开行业数据的效率比我手下两个运营加起来还高。

DeepSeek技术社区

所有评论(0)

查看更多评论

糕692

@2502_92311356

已为社区贡献8条内容

零基础从入门到精通 LangChain（保姆级超详细教程）下篇

糕692

前言：从基础Demo到生产级落地，这一篇就够了

下篇完整知识体系

一、RAG 核心实现：从零搭建生产级检索增强生成系统

1.1 先彻底搞懂：RAG的核心原理与完整链路

RAG标准全链路（必须牢记）

1.2 核心前置组件：嵌入模型与向量数据库

1.2.1 嵌入模型（Embedding Model）

嵌入模型完整接入代码

1.2.2 向量数据库

Chroma向量数据库完整实现代码

1.3 完整的基础RAG问答链实现

1.4 RAG进阶优化：从Demo到生产级的核心技巧

1.4.1 多查询检索（MultiQueryRetriever）：解决用户提问表述偏差

1.4.2 重排（Rerank）：解决检索结果排序不精准问题

1.4.3 父文档检索器（ParentDocumentRetriever）：解决长文档语义丢失问题

二、工具调用：给大模型装上“手脚”，突破原生能力边界

2.1 工具调用的核心原理

2.2 工具的3种定义方式

2.2.1 方式一：@tool装饰器自定义工具（最简单、最常用）

2.2.2 方式二：使用LangChain内置工具

2.3 工具调用全流程实现

三、记忆模块：让大模型拥有“长期记忆”，实现连贯多轮对话

3.1 记忆模块的核心设计逻辑

3.2 4种常用记忆类型与完整实现

3.2.1 基础缓存记忆：ConversationBufferMemory

3.2.2 窗口记忆：ConversationBufferWindowMemory

3.2.3 摘要记忆：ConversationSummaryMemory（长对话最优方案）

3.3 记忆的持久化

四、Agent 开发：让大模型拥有自主思考与行动能力

4.1 Agent的核心原理：ReAct框架

4.2 从零搭建第一个ReAct Agent（完整可运行代码）

4.3 Agent进阶：RAG + Agent 融合

五、LangGraph 工作流：构建可控、可观测、生产级的复杂Agent

5.1 LangGraph的核心概念

5.2 环境搭建

5.3 从零用LangGraph构建ReAct Agent（完整可运行代码）

5.4 LangGraph生产级进阶能力

六、生产级工程化最佳实践：从Demo到落地的最后一公里

6.1 项目结构规范

6.2 用FastAPI封装HTTP接口

6.3 Docker部署

6.4 生产级核心优化要点

6.4.1 日志与可观测性

6.4.2 安全防护

6.4.3 性能与成本优化

下篇总结 & 附加篇预告

下篇我们学到了什么？

附加篇预告

互动环节

所有评论(0)

温馨提示：您尚未绑定手机号

糕692