LangChain教程

简介

LangChain 是一个基于大型语言模型（LLMs）的开源框架，旨在帮助开发者构建端到端的语言模型应用程序。他是基于Python语言的框架，我们今天来初步尝试一下。

基础概念

LangChain的核心组件

LangChain框架的核心模块主要包括模型输入输出(Model I/O)、数据连接(Data Connection)、
链(Chains)、记忆(Memory)、代理(Agents)和回调(Callbacks)等六个模块。

模型输入输出模块

• 提供了语言模型和大语言模型的接口，可以将文本格式化为模型需要的输入（与大模型交互，向大模型发送文本，并接收返回的信息）。

数据连接模块

• 提供了文档加载器和文档转换器等工具，用于将非结构化文本转换为可处理的数据（也许称为文本加工模块更加符合）。

链模块

• 提供了各种类型的链，如基础链、路由链和顺序链等，用于组合和连接不同的功能。（比如需要查询外部接口、加工输入信息，我们可以用链将这些任务串起来）

记忆模块

• 用于在链之间存储和传递信息，实现对话的上下文感知能力。（将对话数据存储到nosql或者sql中）

代理模块

• 通过使用LLM来自动决策和执行动作，完成任务。（也许可以叫助理模块，我们将资源和需要执行的任务目标告诉程序，程序自动帮助我们实现任务）

回调模块

• 提供了连接到LLM申请的各个阶段的功能，用于日志记录、监控和流传输等任务。

文本 & 聊天消息

文本

与LLM交互的自然语言方式

my_text = "每个星期有几天"

聊天消息

类似文本，但指定了消息类型（系统、人类、人工智能）

• 系统（system）：有用的背景消息，告诉人工智能该做什么
• 人类（human）：用户的消息
• 人工智能（AI）：人工智能响应的内容消息

提示词 & 提示词模版

提示词（Prompts）

提示词是引导语言模型生成特定输出的输入文本。在LangChain中，提示词通过模板实现动态化和结构化，使开发者能够分离文本逻辑与数据，提高复用性和可维护性。

提示词模板（Prompt Templates）

模板是包含占位符的文本框架，运行时动态填充变量生成最终提示词。例如：

from langchain.prompts import PromptTemplate
template = "请为一家{industry}公司起一个名字。"
prompt = PromptTemplate(input_variables=["industry"], template=template)
formatted_prompt = prompt.format(industry="人工智能")  # 输出：请为一家人工智能公司起一个名字。[2](@ref)

文档加载器 & 文本分割器

文档加载器（Document Loaders）

支持从多种来源加载文本（如PDF、CSV、数据库、网页爬虫等），例如：

from langchain.document_loaders import PyPDFLoader
loader = PyPDFLoader("example.pdf")
documents = loader.load()  # 加载PDF文档内容[5,8](@ref)

文本分割器（Text Splitters）

将长文本分割为适合模型处理的片段，保持语义连贯性：

from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = splitter.split_documents(documents)  # 分割为1000字符的块，重叠200字符[5,7](@ref)

示例选择器

简介

示例选择器的作用是从一组预定义的示例中，根据输入动态筛选最相关的子集，帮助语言模型更好地理解任务并生成准确回答。（对提示词做加工）

内置选择器类型

类型	选择策略	适用场景
LengthBasedExampleSelector	根据输入长度调整示例数量（长输入选较少示例，短输入选更多）	控制上下文窗口长度
MMRExampleSelector	基于最大边际相关性（Max Marginal Relevance），平衡相关性与多样性	信息检索、推荐系统
SemanticSimilarityExampleSelector	通过嵌入模型（如余弦相似度）选择语义最相似的示例	需要高语义匹配的任务
NGramOverlapExampleSelector	基于 N-gram 重叠度选择示例（如词或字符序列匹配）	文本匹配、翻译任务

输出解析器

核心功能

输出解析器用于将语言模型（LLM）生成的非结构化文本转换为结构化数据（如JSON、列表、Pydantic对象等），主要作用包括：

• 结构化转换：将模型输出的文本解析为程序可处理的格式（如字典、列表、日期对象等）
• 数据验证：通过Pydantic等工具确保输出符合预定义的字段类型和约束（如评分必须在1-10之间）
• 错误修复：自动修复格式错误的输出（如无效JSON），通过OutputFixingParser重新调用LLM修正
• 模型指导：通过get_format_instructions()生成格式说明，引导LLM输出符合预期的结构

类型

解析器类型	类名	核心功能
字符串解析器	StrOutputParser	提取模型输出的原始文本，不做额外处理
逗号分隔列表解析器	CommaSeparatedListOutputParser	将逗号分隔的文本转换为Python列表
JSON解析器	JsonOutputParser	将JSON格式文本转换为Python字典或列表，支持Pydantic验证
Pydantic解析器	PydanticOutputParser	基于Pydantic模型定义输出结构，支持嵌套字段和强类型验证
结构化输出解析器	StructuredOutputParser	通过ResponseSchema自定义字段，生成结构化字典
日期时间解析器	DatetimeOutputParser	将文本解析为Python datetime对象，支持多种时间格式
枚举解析器	EnumOutputParser	限制输出为预定义的枚举值（如颜色类型）
自动修复解析器	OutputFixingParser	包装其他解析器，在解析失败时调用LLM修复错误
XML解析器	XMLOutputParser	解析XML格式的输出为字典或对象

使用场景与最佳实践

使用场景

• 数据提取：从模型回答中提取结构化字段（如电影信息、产品参数）
• API集成：将LLM输出转换为JSON格式，供其他系统调用
• 多语言支持：结合翻译链处理跨语言输出

最佳实践

• 错误处理：使用OutputFixingParser或RetryWithErrorParser增强鲁棒性
• 嵌套结构：通过Pydantic模型定义复杂数据结构，并在提示中提供清晰示例
• 性能优化：对高频解析结果缓存，减少LLM重复调用

环境搭建

langchain是一个基于python语言的框架，开发环境有多种选择，langchain官方推荐使用JupyterNotebook，但我使用的是Anaconda+Pycharm，Anaconda用来管理python环境，Pycharm是Jetbrain专门用于开发python的软件。

使用Jupyter 搭建环境

地址：https://www.devbean.net/2024/07/langchain-01-setup-jupyter/

使用Anaconda搭建环境

首先下载Anaconda并安装，然后修改.condarc文件，配置镜像源，这里使用的是清华镜像源

channels:
  - defaults
show_channel_urls: true
default_channels:
  - https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main
  - https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/r
  - https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/msys2
custom_channels:
  conda-forge: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud
  pytorch: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud

然后进入Anaconda PowerShell Prompt，输入命令创建环境

# 列出所有环境
conda env list
# 创建虚拟环境
conda create -n langchain_env python=3.12 # 名称为‘langchain_env’ 使用的python版本为3.12
# 激活虚拟环境
conda activate langchain_env
# 安装核心依赖
pip install langchain-core langchain-community langchain-openai langchain-deepseek
# 安装requirements.txt文件中的依赖（可以使用requirements.txt文件来统一管理依赖）
pip install -r requirements.txt

接着我们打开PyCharm，新建一个Python项目，我的叫‘langchain-demo’；新建一个requirements.txt文件来存放项目的python依赖，使用命令“pip install -r requirements.txt”安装依赖；

aiohappyeyeballs==2.6.1
aiohttp==3.12.12
aiosignal==1.3.2
annotated-types==0.6.0
anyio==4.7.0
attrs==25.3.0
black==23.12.1
certifi==2025.4.26
charset-normalizer==3.4.2
click==8.1.8
colorama==0.4.6
dataclasses-json==0.6.7
distro==1.9.0
fastapi>=0.115.0,<0.116.0  # 允许补丁更新，避免次要版本升级[2,8](@ref)
frozenlist==1.7.0
greenlet==3.1.1
h11==0.16.0
httpcore==1.0.9
httptools==0.6.4
httpx==0.28.1
httpx-sse==0.4.0
idna==3.7
jiter==0.9.0
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.25
langchain-community==0.3.24
langchain-core==0.3.60
langchain-deepseek==0.1.3
langchain-openai==0.3.17
langchain-text-splitters==0.3.8
langgraph==0.4.8
langgraph-checkpoint==2.0.26
langgraph-prebuilt==0.2.2
langgraph-sdk==0.1.70
langsmith==0.3.42
marshmallow==3.26.1
multidict==6.4.4
mypy_extensions==1.1.0
numpy==2.3.0
openai==1.79.0
orjson==3.10.18
ormsgpack==1.10.0
packaging==24.2
propcache==0.3.2
pydantic==2.10.3
pydantic-settings==2.6.1
pydantic_core==2.27.1
PyMySQL==1.1.1
python-dotenv==1.1.0
python-multipart==0.0.20
PyYAML==6.0.2
regex==2024.11.6
requests==2.32.4
requests-toolbelt==1.0.0
setuptools==78.1.1
sniffio==1.3.0
SQLAlchemy==2.0.39
starlette==0.38.2
tenacity==9.1.2
tiktoken==0.9.0
tqdm==4.67.1
typing-inspect==0.9.0
typing_extensions==4.12.2
urllib3==2.4.0
uv==0.7.5
uvicorn==0.32.1
watchfiles==1.0.5
websockets==15.0.1
wheel==0.45.1
xxhash==3.5.0
yarl==1.20.1
zstandard==0.23.0

新建一个.env文件，用来存放需要使用到的环境变量（配置信息）；

ALIYUN_API_KEY="sk-xxx"
ALIYUN_MODEL_NAME="text-embedding-v1"
DEEPSEEK_API_KEY="sk-xxx"
DEEPSEEK_MODEL="deepseek-chat"

我们引入了uvicorn，可以使用命令”uvicorn app.main:app –reload“启动项目

基础教程

简单的交互

我们使用deepseek大模型，首先在deepseek开发平台注册并创建api_key，在项目中创建model/deepseekModel.py，将大模型服务注册为一个对象

import os
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv

# 加载环境变量（需在.env文件中设置DEEPSEEK_API_KEY）
load_dotenv()

# 获取deepseek_api_key
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")

# 初始化DeepSeek模型
model = ChatDeepSeek(
    model="deepseek-chat",  # 可选模型：deepseek-chat（对话）或deepseek-reasoner（推理）
    api_key=DEEPSEEK_API_KEY,
    temperature=0.5,  # 控制输出随机性（0-1，越高越具创造性）
    max_tokens=512
)

我们再创建一个helloworld.py，再其中调用大模型服务

# 查看langchain版本
# import langchain
#
# print(langchain.__version__)

from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser

# 引入deepseek大模型
from model.deepseekModel import model

# 构建消息
message = [
    SystemMessage(content="你是一个翻译助手，请将下面的内容翻译为英文"),
    HumanMessage(content="你好，langchain，今天过得怎么样？")
]

# 调用deepseek
response = model.invoke(message)
print(response)

# 文本解析器，提取返回中的文本信息
parser = StrOutputParser()
parser_response = parser.invoke(response)
print("提取到的文本信息：" + parser_response)

# 链式调用，将上面两个步骤串连起来，得到一个调用链，使用链调用并处理返回的信息
chain = model | parser
chain_response = chain.invoke(message)
print("链式调用获取到的信息：" + chain_response)

可以看到代码还是比较简单的，稍微解释一下链式调用，在java中有一种设计模式叫‘责任链’，简单来说就是将复杂的任务切割为相对简单的一个个业务模块，再由流程控制器调用；使用时就可以灵活的调用其中的某几个模块，业务执行时就像链条一样，按照串联顺序执行。

构建聊天机器人

刚才我们已经实现了与大模型的对话，但是多进行几次对话我们会发现两个不足的地方，第一，每次发送请求返回的延迟都有点高，造成这个问题的原因是现在AI大模型将文本全部生成后再返回给我们；第二，现在AI大模型记不住对话信息，所以每次发送请求都相当于第一句对话；

现在我们想进一步构建一个聊天机器人，弥补我们刚才说的两个问题；所以需要更进一步实现两个功能：流式访问和记忆对话；流式访问是指我们让AI大模型返回的信息采用流式返回（生成词语及返回，不需要全部生成后返回），这样就可以避免返回时间过长的问题；记忆对话是指我们将对话信息保存下来，并在下一次发送对话请求时将整个对话信息发送给AI大模型，这样AI大模型就可以像人类一样与我们进行交谈；代码如下：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate, \
    MessagesPlaceholder
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables import RunnableWithMessageHistory

# 引入deepseek大模型
from model.DeepseekModel import deepseek_model

# 使用提示词模版构建消息
messageTemplate = ChatPromptTemplate.from_messages(
    [
        SystemMessagePromptTemplate.from_template("你是一个对话助手，请用{language}友好回答问题"),
        HumanMessagePromptTemplate.from_template("{text}"),
        # 标识要传给大模型的用户输入信息的名称（这里就表示要将占位符{text}的内容传给大模型）
        MessagesPlaceholder(variable_name="text")
    ]
)

# 文本解析器，提取返回中的文本信息
parser = StrOutputParser()

# 链式调用，将步骤串连起来，得到一个调用链，使用链调用并处理返回的信息
chain = messageTemplate | deepseek_model | parser

# 保存聊天记录的存储结构
store = {}

# 根据session_id获取对话记录
# session_id: str 入参及参数类型
# BaseChatMessageHistory 返回值的参数类型
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory();
    return store[session_id]


# 构造能保存历史聊天记录的对象
chat_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    # 与提示词模版中的占位符
    input_messages_key="text"
)

# 通过config对象（json）设置对话的session_id
config = {"configurable": {"session_id": "100"}}

# 使用流式返回
for res1 in chat_with_history.stream(
        {
            "language": "中文",
            "text": "我的名字叫张三"
        },
        config=config,
):
    print(res1, end="|")

for res2 in chat_with_history.stream(
        {
            "language": "中文",
            "text": "我的名字叫什么"
        },
        config=config
):
    print(res2, end="|")

构建向量存储和检索器

对于向量存储的概念，我们之前的文章打造私有知识库中，我们已经介绍过向量存储的概念，现在来看看在langchain中如何实现；首先，在python中引入阿里云百炼的向量模型，我们先在.evn文件中添加api_key和model_name，

ALIYUN_API_KEY="sk-xxx"
ALIYUN_MODEL_NAME="text-embedding-v1"

再初始化一个阿里云百炼向量模型的对象

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

# 加载 .env 文件
load_dotenv()
aliyun_key = os.getenv("ALIYUN_API_KEY")
model_name = os.getenv("ALIYUN_MODEL_NAME", "text-embedding-v1")

embeddings = DashScopeEmbeddings(
    dashscope_api_key=aliyun_key,
    model=model_name  # 阿里云官方embedding模型
)

我们编写一个相似度查询的demo

from langchain_chroma import Chroma
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

from langchain.globals import set_debug, get_debug

from model.DeepseekModel import deepseek_model

# 开启调试模式
set_debug(True)
# 获取当前调试状态
debug_status = get_debug()
print("Debug mode:", debug_status)

# 模拟文档数据
documents = [
    Document(
        page_content="猫是柔软可爱的动物，但相对独立",
        metadata={"source": "常见动物宠物文档"},
    ),
    Document(
        page_content="狗是人类很早开始的动物伴侣，具有团队能力",
        metadata={"source": "常见动物宠物文档"},
    ),
    Document(
        page_content="金鱼是我们常常喂养的观赏动物之一，活泼灵动",
        metadata={"source": "鱼类宠物文档"},
    ),
    Document(
        page_content="鹦鹉是猛禽，但能够模仿人类的语言",
        metadata={"source": "飞禽宠物文档"},
    ),
    Document(
        page_content="乌龟是小朋友比较喜欢的宠物，而且很好喂养，但是不太亲人",
        metadata={"source": "常见动物宠物文档"},
    ),
]

# 引入阿里云百炼的文本向量模型
from model.AliyunBailianEmbeddings import embeddings

# 将文档添加到向量存储中
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings
)

# 按照相似度搜索 返回纯文档列表
res1 = vectorstore.similarity_search("cat")
print(res1)
print("========================")

# 按照分数相似度搜索，分数越小越相似
# 返回文档和对应的相似度分数，信息更丰富
res2 = vectorstore.similarity_search_with_score("cat")
print(res2)
print("========================")

# 检索器
# k=1 返回一条数据
retriever = RunnableLambda(vectorstore.similarity_search).bind(k=1)
res3 = retriever.batch(["猫", "鲨鱼"])
print(res3)
print("========================")

# 构建信息
message = """
仅使用提供的上下文回答下面的问题：
{question}
上下文：
{context}
"""

# 构建提示词模版
rag_prompt_template = ChatPromptTemplate.from_messages(("human", message))

# 文本解析器，提取返回中的文本信息
parser = StrOutputParser()

# {"question": RunnablePassthrough(), "context": retriever} 是一个RunnableParallel 对象
# 它用于并行执行多个 Runnable 组件，并将结果合并成一个字典（map）输出。
chain = {"question": RunnablePassthrough(), "context": retriever} | rag_prompt_template | deepseek_model | parser

res4 = chain.invoke("请为我介绍一下猫这种生物")
print(res4)

可以看到，代码的基本逻辑就是将文档（docs）添加到向量存储（vectorstore）中，然后利用RunnableLambda构建一个检索器（retriever），将检索器添加到执行链（chain）中，这样执行就可以相似度检索（执行链会将用户对话向量化，与向量存储中的内容进行向量比较，返回最接近的内容）

构建对话式 RAG

我们前面介绍了向量存储，这里我们进一步实现对话式RAG；关于RAG的知识我们在文章langchain4j实现智能助手中介绍过，现在我们看看在langchain中如何实现，代码如下：

import os

import bs4
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain_chroma import Chroma
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_text_splitters import RecursiveCharacterTextSplitter

from model.AliyunBailianEmbeddings import embeddings
from model.DeepseekModel import deepseek_model

os.environ[
    "USER_AGENT"] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"

# 1. 加载、分块和索引博客内容
# 加载
loader = WebBaseLoader(
    # 博客的地址
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    # 只提取某几类class中的信息
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
    # 添加headers参数
    header_template={
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    },
)

docs = loader.load()

# 分块
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)

split_documents = text_splitter.split_documents(docs)

# 向量存储
vectorstore = Chroma.from_documents(documents=split_documents, embedding=embeddings)

# 创建检索器
retriever = vectorstore.as_retriever()

contextualize_q_system_prompt = (
    "给定一个聊天历史记录和最新的用户问题，"
    "可能会涉及聊天历史记录中的上下文，"
    "制定一个独立的问题，可以在没有聊天历史记录的情况下理解。"
    "如果需要，重新构造问题，否则原样返回。"
)

contextualize_q_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", contextualize_q_system_prompt),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ]
)

# 构建历史感知检索器
history_aware_retriever = create_history_aware_retriever(deepseek_model, retriever, contextualize_q_prompt)

# 2. 将检索器融入问答链

# 创建聊天记录存储器（注意这里与前文的向量存储不是一回事）
chat_store = {}

# 获取历史聊天记录
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in chat_store:
        chat_store[session_id] = ChatMessageHistory()
    return chat_store[session_id]

# 构建提示词
system_prompt = (
    "您是一个用于回答问题的助手。"
    "使用检索到的上下文来回答问题。"
    "如果不知道答案，请说不知道。"
    "最多使用三句话，保持回答简洁。"
    "\n\n"
    "{context}"
)
human_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system_prompt),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ]
)

# 创建问答链
question_answer_chain = create_stuff_documents_chain(deepseek_model, human_prompt)
rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)

# 构建对话链
conversational_rag_chain = RunnableWithMessageHistory(rag_chain, get_session_history,
                                                      input_messages_key="input",
                                                      output_messages_key="answer",
                                                      history_messages_key="chat_history", )
# 调用
message1 = {"input": "什么是任务分解？"}
config1 = {
    "configurable": {"session_id": "abc123"}
}
response1 = conversational_rag_chain.invoke(input=message1, config=config1)
print(response1["answer"])

message2 = {"input": "常见的做法有哪些？"}
config2 = config = {"configurable": {"session_id": "abc123"}}

response2 = conversational_rag_chain.invoke(input=message2, config=config2)
print(response2["answer"])

可以看到，我们的工作大致分为5个部分：1. 加载资源并将资源分块，利用资源创建向量存储和历史感知检索器； 2. 构建聊天对话存储 3. 构建提示词模版和提示词 4. 构建执行链 5. 调用执行链

构建在 SQL 数据上的问答系统

在上文’构建对话式RAG‘中我们使用的资源来自网络上的博客，而我们平时经常使用的数据库，也是一个重要的资源；我们来看看如何使用SQL数据库，代码如下：

from operator import itemgetter

from langchain.chains.sql_database.query import create_sql_query_chain
from langchain_community.tools import QuerySQLDataBaseTool
from langchain_community.utilities import SQLDatabase
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough

from model.DeepseekModel import deepseek_model

db = SQLDatabase.from_uri("sqlite:///D:/Data/chinook/chinook.db")

# 数据库测试
# print(db.dialect)
# print(db.get_usable_table_names())
# print(db.run("SELECT * FROM artists LIMIT 10;"))

# 大模型对问题返回SQL语句
sql_query_chain = create_sql_query_chain(deepseek_model, db)
message = {"question": "有多少名员工"}
response1 = sql_query_chain.invoke(message)
# 返回的sql语句如下：“SQLQuery: SELECT COUNT(*) AS "EmployeeCount" FROM "employees";”
sql = "";
if ": " in response1:
    sql = response1.split(": ", 1)[1].strip()

print(sql)
print("=========================")


class ExecuteTool(QuerySQLDataBaseTool):
    def _run(self, query: str) -> str:
        # 去除可能的SQLQuery:前缀
        if query.startswith("SQLQuery:"):
            query = query.split(":", 1)[1].strip()
        return super()._run(query)


execute_tool = ExecuteTool(db=db)
chain = sql_query_chain | execute_tool

response2 = chain.invoke(message)
print(response2)
print("=========================")

answer_prompt = PromptTemplate.from_template(
    """给定以下用户问题，相应的 SQL 查询和 SQL 结果，回答用户问题。
    问题：{question}
    SQL 查询：{query}
    SQL 结果：{result}
    答案："""
)

chain = (
        RunnablePassthrough.assign(query=sql_query_chain).assign(
            result=itemgetter("query") | execute_tool
        )
        | answer_prompt
        | deepseek_model
        | StrOutputParser()
)

response3 = chain.invoke({"question": "有多少员工"})
print(response3)
print("=========================")

可以看到，我们的代码大致分为3个部分：1. 引入数据库db后，利用其与AI大模型构建查询链，当我们输入问题，程序就能返回对应的SQL语句；2. 我们对返回的SQL语句进行格式处理后，构建SQL的执行器；3. 构建提示词；4. 将查询链、数据库执行器、提示词、AI大模型、文本处理器串联为最终的执行链；5. 调用执行链

进阶教程（如何使用代理）

代理（Agent）

单独来看，语言模型无法采取行动 - 它们只能输出文本。 LangChain的一个重要用途是创建代理。 代理是使用AI大模型作为推理引擎的系统，用于确定要采取的行动以及这些行动的输入应该是什么。 然后，可以将这些行动的结果反馈给代理，并确定是否需要更多行动，或者是否可以结束。我们来构建一个代理，该代理可以与多个不同的工具进行交互：一个是本地数据库，另一个是搜索引擎。我们可以使用tavily_search，这是langchain已经集成的一款搜索引擎，当然我们先申请其api_key，将其填入.env文件，程序就可以自动加载；实现agent的代码如下：

from dotenv import load_dotenv
from langchain_community.tools import TavilySearchResults
from langgraph.prebuilt import chat_agent_executor
from langchain_core.messages import HumanMessage

from model.DeepseekModel import deepseek_model

load_dotenv()
# 引入tavily天气查询（注册https://app.tavily.com/home，申请apikey，将其添加到.env文件）
tavily_search = TavilySearchResults(max_results=1)

# 将tavily_search包装为数组
tools = [tavily_search]
# 将工具绑定至ai大模型
model_bind_tavily = deepseek_model.bind_tools(tools)

# 创建一个agent，并用Agent来执行工具的使用
agent = chat_agent_executor.create_tool_calling_executor(deepseek_model, tools)

config = {"configurable": {"thread_id": "abc123"}}

for chunk in agent.stream({"messages": [HumanMessage(content="昆明今天的天气如何？")]}, config):
    print(chunk)
    print("-----")

我们将搜索引擎包装为数组，再将其与AI大模型绑定，然后将其传入agent的执行器中，再传入消息和id；执行程序时，代理就会自动使用搜索引擎搜索相关资源，再经过自然语言加工，返回经过加工的回答

利用代理构建对话式RAG

在上文’构建对话式 RAG‘中，我们利用链式方法构建了一个对话式RAG，现在既然我们学些了代理（agent），我们来看看如何使用代理构建对话式RAG，代码如下：

import bs4
from langchain.tools.retriever import create_retriever_tool
from langchain_chroma import Chroma
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.messages import HumanMessage
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import chat_agent_executor

from model.AliyunBailianEmbeddings import embeddings
from model.DeepseekModel import deepseek_model

# 1. 加载、分块和索引博客内容，创建一个索引器
# 加载
loader = WebBaseLoader(
    # 博客的地址
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    # 只提取某几类class中的信息
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
    # 添加headers参数
    header_template={
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    },
)

docs = loader.load()

# 分块
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)

split_documents = text_splitter.split_documents(docs)

# 向量存储
vectorstore = Chroma.from_documents(documents=split_documents, embedding=embeddings)

# 创建检索器
retriever = vectorstore.as_retriever()

tool = create_retriever_tool(
    retriever,
    "blog_post_retriever",
    "搜索并返回自主代理博客文章摘录。",
)
tools = [tool]

# response = tool.invoke("任务分解")
# print(response)

config = {"configurable": {"thread_id": "abc123"}}
agent_executor = chat_agent_executor.create_tool_calling_executor(deepseek_model, tools, checkpointer=MemorySaver())

for chunk in agent_executor.stream(
        {"messages": [HumanMessage(content="什么是任务分解")]}, config
):
    print(chunk)
    print("-----")

for chunk in agent_executor.stream(
        {"messages": [HumanMessage(content="我刚才问的问题是什么")]}, config
):
    print(chunk)
    print("-----")

可以看出，代理方式和链式方式最大的不同是在代理方式中我们不明确规定实现的逻辑，而是将获取资源的‘工具‘交给AI大模型，由AI大模型决定如何利用这些’工具‘来实现业务

利用代理构建在 SQL 数据上的问答系统

在上文’构建在 SQL 数据上的问答系统‘中，我们利用链式方式构建了一个以SQL为资源的问答系统，现在我们看看如何用agent实现，代码如下：

from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain_community.utilities import SQLDatabase
from langchain_core.messages import HumanMessage
from langchain_core.prompts import PromptTemplate
from langgraph.prebuilt import chat_agent_executor

from model.DeepseekModel import deepseek_model

system_prompt = PromptTemplate.from_template(
    """您是一个专门与SQL数据库交互的代理。
    给定一个输入问题，创建一个语法正确的SQLite查询来运行，然后查看查询的结果并返回答案。
    如果第一次查询不成功，最多尝试3次不同的查询方案。
    除非用户指定他们希望获得的特定数量的示例，否则始终将查询限制在最多5个结果。
    您可以按相关列对结果进行排序，以返回数据库中最有趣的示例。
    永远不要查询特定表的所有列，只根据问题要求的相关列进行查询。
    您可以使用与数据库交互的工具。
    只使用以下工具。只使用以下工具返回的信息来构建您的最终答案。
    在执行查询之前，您必须仔细检查您的查询。如果在执行查询时出现错误，请重新编写查询并重试。
    不要对数据库进行任何DML语句（INSERT、UPDATE、DELETE、DROP等）。
    首先，您应该始终查看数据库中的表，以了解您可以查询的内容。
    不要跳过此步骤。
    然后，您应该查询最相关表的模式。"""
)

db = SQLDatabase.from_uri("sqlite:///D:/Data/chinook/chinook.db")
toolkit = SQLDatabaseToolkit(db=db, llm=deepseek_model)
tools = toolkit.get_tools()
agent_executor = chat_agent_executor.create_tool_calling_executor(deepseek_model,
                                                                  tools,
                                                                  prompt=system_prompt)

for s in agent_executor.stream(
        {"messages": [HumanMessage(content="Which country's customers spent the most?")]}
):
    print(s)
    print("----")

可以看到，我们没有规划具体的执行步骤，依旧是利用工具与AI大模型构建代理，然后将问题提交给代理处理

更多

关于更多的langchain使用方法，可以参考：https://www.aidoczh.com/langchain/v0.2/docs/introduction/