目录

一、引言：为什么需要LangChain
    1.1 原生LLM的痛点
    1.2 环境准备与依赖安装
二、快速上手：跑通第一个LangChain应用
    2.1 申请API Key并配置环境变量
    2.2 步骤1：定义大模型
    2.3 步骤2：定义消息列表
    2.4 步骤3：调用大模型
    2.5 步骤4：输出解析
    2.6 步骤5：链式执行
    2.7 核心概念：Runnable接口与LCEL
三、聊天模型核心能力
    3.1 定义聊天模型
        3.1.1 ChatOpenAI
        3.1.2 invoke()调用
        3.1.3 init_chat_model工厂函数
        3.1.4 ChatOllama本地部署
    3.2 聊天模型——调用工具
        3.2.1 创建工具
        3.2.2 绑定工具
        3.2.3 工具调用
        3.2.4 强制模型调用工具
        3.2.5 工具属性
        3.2.6 将工具输出传递给聊天模型
        3.2.7 LangChain提供的工具：TavilySearch
    3.3 聊天模型——结构化输出
        3.3.1 with_structured_output()方法
        3.3.2 返回Pydantic对象
        3.3.3 返回TypedDict
        3.3.4 返回JSON
        3.3.5 选择输出格式
        3.3.6 实用场景
    3.4 聊天模型——流式传输
        3.4.1 stream()同步传输
        3.4.2 astream()异步传输
        3.4.3 使用StrOutputParser解析模型的输出
        3.4.4 自定义流式输出解析器
        3.4.5 深度探索流式传输
    3.5 使用LangSmith跟踪LLM应用
四、核心组件（Components）
    4.1 消息（Messages）
    4.2 提示词模板（PromptTemplate）
    4.3 少样本提示（few-shotting）
    4.4 输出解析器（Output parsers）
    4.5 文档加载器（Document loaders）
    4.6 文本分割器（Text splitters）
    4.7 文本向量（Embeddings）与向量存储
    4.8 检索器（Retrievers）与RAG实战
五、总结

---

一、引言：为什么需要LangChain

LangChain是一个用于开发由大语言模型（LLM）驱动的应用程序的框架。尽管大模型十分强大，但直接使用原生LLM可能会存在一些问题，例如将其当作搜索引擎去使用时，LLM生成的答案可能要比其他搜索引擎查到的答案更符合你的预期，但要在复杂场景下使用，如将LLM嵌入应用程序时会遭遇全新难题：

• 简单提示词（Prompt）得到的答案经常出现幻觉？
• 提示词结构是否可以统一规范？
• 如何实现开发过程中大模型的轻松、灵活切换？
• 大模型输出是非结构化的，怎样与要求结构化数据的程序接口交互？
• 如何克服预训练模型知识陈旧的问题，引入实时更新？
• 如何连接模型与外部工具或系统，执行具体任务？

LangChain框架的核心目标就是应对这些挑战。它通过将自然语言处理流程拆解为标准化组件，让开发者能够自由组合并高效定制工作流。

本文我将从快速上手出发，逐步深入聊天模型的四大核心能力（定义模型、工具调用、结构化输出、流式传输），再系统拆解LangChain的八大核心组件（消息、提示词模板、少样本提示、输出解析器、文档加载器、文本分割器、文本向量、检索器），最终串联起RAG（检索增强生成）的全流程，让你彻底掌握LangChain的核心能力。

1.1 原生LLM的痛点

总结起来，直接调用原生LLM构建应用时主要面临以下障碍：

痛点	说明
幻觉	简单提示词得到的答案经常出现"一本正经地胡说八道"
提示词不规范	提示词结构缺乏统一规范，难以维护
模型切换困难	更换模型提供方（OpenAI/DeepSeek/Ollama等）需重写大量代码
输出非结构化	模型输出是自然语言字符串，无法直接对接程序接口
知识陈旧	训练数据有截止日期，无法获取实时信息
无法连接外部系统	LLM本身无法执行计算、查询数据库、调用API

1.2 环境准备与依赖安装

本示例环境基于 Python 3.13，第一部分依赖的 requirements.txt 如下：

langchain-openai==0.3.33
langchain==0.3.27
langchain-deepseek==0.1.4
langchain-ollama==0.3.6
langchain-tavily==0.2.12
langchain-chroma==0.2.5
langchain-community==0.3.22
nltk==3.9.2
langchain-redis==0.2.4
unstructured==0.18.15
markdown==3.9
redisvl==0.10.0

安装命令：

pip install -r requirements.txt

第二部分依赖：

pinecone==7.3.0
langchain-pinecone==0.2.12

注意：

• 这部分需要学习到【4.7.3.4 Pinecone 向量存储】后才可安装。原因是安装 langchain-pinecone 时，会同时安装依赖包 simsimd，安装此包后会导致【4.7.3.3 Redis 向量存储】的MMR搜索不可用，该问题目前官方尚未解决。
• 若安装 langchain-pinecone 高版本（如 0.2.3，支持Python 3.13），可能会无法成功。原因是其依赖包 numpy 要求 numpy<2.0.0,>=1.26.4，而该范围的 numpy 不支持 Python 3.13，该问题目前官方尚未解决。

二、快速上手：跑通第一个LangChain应用

本节主要演示如何使用LangChain接入大模型组件，以及与大模型进行简单对话的能力。掌握了基本用法后，将引出LangChain相关前置概念，以便后续深入学习LangChain标准化组件。

2.1 申请API Key并配置环境变量

1. 申请API Key

以OpenAI为例，官网地址：https://platform.openai.com/ 。注册并登录成功后，点击 settings 图标，选择 API keys 配置页面，点击 “Create new secret key” 按钮新增API Key，并将API Key在本地保存好，后续接入ChatGPT时需要使用。

2. 配置环境变量

将API Key配置在环境变量中主要是为了保证其隐私性。由于API Key比较隐私，为避免在程序中暴露，可提前将各个API Key配置在环境变量中，这样在程序中就可以通过获取对应环境变量拿到API Key，保证了API Key的隐私。

2.2 步骤1：定义大模型

首先安装OpenAI包：

pip install -U langchain-openai

定义大模型的核心代码：

# 定义大模型
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-5-mini")

2.3 步骤2：定义消息列表

核心代码：

# 定义消息列表
from langchain_core.messages import HumanMessage, SystemMessage
messages = [
    SystemMessage(content="Translate the following from English into Chinese"),
    HumanMessage(content="hi!"),
]

参数说明：

• SystemMessage：表示系统角色消息，系统消息通常作为输入消息序列中的第一条传入，是用来启动AI行为的消息。
• HumanMessage：表示用户角色消息，是来自用户的、从用户传递到模型的消息。

2.4 步骤3：调用大模型

model 是LangChain Runnable（可运行）接口的实例，这意味着 model 提供了一个标准接口供我们与之交互。要简单地调用模型，我们可以将消息列表传递给 .invoke 方法。

使用 .invoke 方法进行大模型调用，核心代码：

result = model.invoke(messages)
print(result)

输出结果（调试可以看见 result 类型为 AIMessage）：

content='你好!' additional_kwargs={'refusal': None} response_metadata=
{'token_usage': {'completion_tokens': 2, 'prompt_tokens': 20, 'total_tokens': 
22, ...}, 'model_name': 'gpt-5-mini-2024-07-18', 'system_fingerprint': 'fp_560af6e559', 
'id': 'chatcmpl-C5l29mi6KkQSFO3qaElDJD5Lj9nBn', 'service_tier': 'default', 
'finish_reason': 'stop', 'logprobs': None} id='run--777ef0be-55c2-411a-a4ab-
104a45c22f20-0' usage_metadata={'input_tokens': 20, 'output_tokens': 2, 
'total_tokens': 22, ...}

输出说明：

• AIMessage：来自AI的消息，从聊天模型返回，作为对提示（输入）的响应。
  • content：消息的内容。
  • additional_kwargs：与消息关联的其他有效负载数据。对于来自AI的消息，可能包括模型提供程序编码的工具调用。
  • response_metadata：响应元数据。例如：响应标头、logprobs、令牌计数、模型名称。侧重于"响应"本身的信息，比如这次请求的ID、使用的模型版本、以及服务提供商返回的所有原始元数据。它主要用于调试、日志记录和获取请求的上下文信息。
  • usage_metadata：消息的使用元数据，例如令牌计数。侧重于"资源消耗"的量化信息，即这次请求消耗了多少Token。它主要用于成本计算、监控和预算控制。

2.5 步骤4：输出解析

若只想输出聊天模型返回的结果字符串，可以使用 StrOutputParser 输出解析器组件，将大模型输出结果解析为最可能的字符串。核心代码：

# 定义str字符串输出解析器
from langchain_core.output_parsers import StrOutputParser
parser = StrOutputParser()
print(parser.invoke(result))

输出结果：

你好!

2.6 步骤5：链式执行

通过上述步骤，无论是调用大模型，还是输出解析，我们发现，每次都调用了一个 invoke() 方法，最终才会得到我们想要的结果。

对于LangChain，它给我们提供了链式执行的能力，即我们只需要定义各个"组件"，将它们"链起来"，一次性执行即可得到最终效果。

核心代码如下：

# 定义大模型
model = ChatOpenAI(model="gpt-5-mini")
# 定义消息列表
messages = [
    SystemMessage(content="Translate the following from English into Chinese"),
    HumanMessage(content="hi!"),
]
# 定义输出解析器
parser = StrOutputParser()
# 定义链
chain = model | parser
# 执行链
result = chain.invoke(messages)
print(result)

输出结果：

你好!

快速上手完整代码参考：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.output_parsers import StrOutputParser
# 定义大模型
model = ChatOpenAI(model="gpt-5-mini")
# 定义消息列表
messages = [
    SystemMessage(content="Translate the following from English into Chinese"),
    HumanMessage(content="hi!"),
]
# 定义输出解析器
parser = StrOutputParser()
# 定义链
chain = model | parser
# 执行链
result = chain.invoke(messages)
print(result)

2.7 核心概念：Runnable接口与LCEL

2.7.1 Runnable接口

Runnable接口是使用LangChain Components（组件）的基础。

概念说明——Components（组件）：用来帮助我们在构建应用程序时，提供了一系列的核心构建块，例如语言模型、输出解析器、检索器、编译的LangGraph图等。

Runnable定义了一个标准接口，允许Runnable组件：

• Invoked（调用）：单个输入转换为输出。
• Batched（批处理）：多个输入被有效地转换为输出。
• Streamed（流式传输）：输出在生成时进行流式传输。
• Inspected（检查）：可以访问有关 Runnable 的输入、输出和配置的原理图信息。
• Composed（组合）：可以组合多个Runnable，以使用LCEL协同工作以创建复杂的管道。

因此，在快速上手示例中，我们定义的语言模型（model）、输出解析器（StrOutputParser）都是Runnable接口的实例！它们都使用了Invoked（调用）的能力，回顾一下：

# 语言模型（model）
model = ChatOpenAI(model="gpt-5-mini")
result = model.invoke(messages)   # 语言模型是 Runnable 接口实例，允许invoke调用

# 输出解析器（StrOutputParser）
parser = StrOutputParser()
parser.invoke(result)             # 输出解析器是 Runnable 接口实例，允许invoke调用

2.7.2 LangChain Expression Language（LCEL）

LangChain Expression Language（LCEL）：采用声明性方法，从现有Runnable对象构建新的Runnable对象。

通过LCEL构建出的新的Runnable对象，被称为 RunnableSequence，表示可运行序列。RunnableSequence 就是一种链（参考2.6节）。通过调试就能发现，chain 的类型就是 RunnableSequence。

重要的是，RunnableSequence 也是Runnable接口的实例，它实现了完整的Runnable接口，因此它可以用与任何其他Runnable相同的姿势使用。回顾一下：

chain = model | parser
chain.invoke(messages)  # 链是 Runnable 接口实例，允许invoke调用

可以看到，LCEL其实是一种编排解决方案，它使LangChain能够以优化的方式处理链的运行时执行。任何两个Runnable实例都可以"链"在一起成序列。上一个可运行对象的 .invoke() 调用的输出作为输入传递给下一个可运行对象。方法就是使用 |（管道/运算符）：

chain = model | parser

它通过两个 Runnable 对象去创建一个 RunnableSequence。实际上LangChain重载了 | 运算符，使用 | 运算符就相当于：

from langchain_core.runnables import RunnableSequence
chain = RunnableSequence(first=model, last=parser)

除此之外，可以使用 .pipe 方法代替，这也相当于 | 运算符：

chain = model.pipe(parser)

在Unix/Linux系统中，pipe() 系统调用和 | 管道操作符都用于实现进程间通信，这里同样也是迁移过来的用法。

三、聊天模型核心能力

3.1 定义聊天模型

大语言模型（LLM）在各种与语言相关的任务（例如文本生成、翻译、摘要、问答等）中表现出色。现代LLM通常通过聊天模型接口访问，该接口将消息列表作为输入，并返回消息作为输出，而不是使用纯文本。

这里需要注意 LLM 与 LangChain 中聊天模型的关系：

• 在LangChain的官方文档中，认为 LLM 大多数是纯文本补全模型。这些纯文本模型封装的API接受一个字符串提示作为输入，并输出一个字符串补全结果（实际上LLM还包括多模态输入）。OpenAI的GPT-5就是作为LLM来实现的。
• LangChain中的聊天模型通常由LLM提供支持，但经过专门调整以用于对话。关键在于，它们不是接受单个字符串作为输入，而是接受聊天消息列表，并返回一条AI消息作为输出。

3.1.1 方式1：ChatOpenAI

class langchain_openai.chat_models.base.ChatOpenAI 是LangChain为OpenAI的聊天模型（如 gpt-5、gpt-5-mini）提供的具体实现类。其继承了 class langchain_openai.chat_models.base.BaseChatOpenAI，且 BaseChatOpenAI 实现了标准的Runnable接口。

ChatOpenAI常用初始化参数说明：

参数名	参数描述
model	要使用的OpenAI模型的名称
temperature	采样温度，温度值越高，AI回答越天马行空；温度越低，回答越保守靠谱
max_tokens	要生成的最大令牌数
timeout	请求超时时间
max_retries	最大重试次数
openai_api_key / api_key	OpenAI API密钥。如果未传入，将从环境变量中读取 OPENAI_API_KEY
base_url	API请求的基本URL
organization	OpenAI组织ID。如果未传入，将从env var OPENAI_ORG_ID 中读取

示例：

from langchain_openai import ChatOpenAI
model = ChatOpenAI(
    model="gpt-5-mini",
    temperature=0,
    max_tokens=None,
    timeout=None,
    max_retries=2,
    # api_key="...",
    # base_url="...",
    # organization="...",
    # other params...
)

若使用其它与OpenAI兼容的大模型，例如DeepSeek，则可以使用以下定义方式：

import os
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key=OPENAI_API_KEY,
    model="deepseek-chat",
    # ...
)

参数说明：

• base_url：出于与OpenAI兼容考虑，要将 base_url 设置为 https://api.deepseek.com/v1 来使用，但注意，此处 v1 与模型版本无关。
• openai_api_key：需要单独申请DeepSeek的API Key，然后重新进行环境变量配置。DeepSeek API Key申请地址：https://platform.deepseek.com/api_keys 。

3.1.2 invoke()调用

介绍方式2之前，需要先来了解一下关于Runnable接口中的 .invoke() 调用。该方法是将单个输入转换为对应的输出。例如对于聊天模型来说，就是根据用户的问题输入，输出相应的答案。

invoke() 方法定义：

abstractmethod invoke(
    input: Input,
    config: RunnableConfig | None = None,
    **kwargs: Any,
) → Output

请求参数：

• input：输入一个Runnable实例
• config（默认空）：用于 Runnable 的配置。

返回值：

• 返回一个Runnable实例

class RunnableConfig 常用参数说明：

参数名	参数描述
configurable	通过 .configurable_fields() 在此Runnable或子Runnable上配置的属性的运行时值
run_id	针对此调用运行的跟踪器的唯一标识符。如果未提供，将生成新的UUID
run_name	此调用的跟踪器运行的名称。默认为类的名称
metadata	此次调用和任何子调用的元数据。键是字符串，值是JSON。类型：dict[str, Any]

3.1.3 方式2：init_chat_model

上面的 ChatOpenAI 用于明确创建OpenAI聊天模型的实例。而 init_chat_model() 是一个工厂函数，它可以初始化多种支持的聊天模型（如OpenAI、Anthropic、FireworksAI等），不仅仅是OpenAI的聊天模型。

init_chat_model() 函数定义：

langchain.chat_models.base.init_chat_model(
    model: str,
    *,
    model_provider: str | None = None,
    configurable_fields: Literal[None] = None,
    config_prefix: str | None = None,
    **kwargs: Any,
) → BaseChatModel

init_chat_model() 常用参数说明：

参数名	参数描述
model	要使用的模型的名称
model_provider	模型提供方。支持的model_provider值和相应的集成包有：openai -> langchain-openai；anthropic -> langchain-anthropic；google_genai -> langchain-google-genai；ollama -> langchain-ollama；deepseek -> langchain-deepseek 等。如果未指定，将尝试从模型推断model_provider
configurable_fields	设置哪些模型参数是可配置的。若配置为：None：没有可配置的字段；'any'：所有字段都是可配置的，类似 api_key、base_url 等可以在运行时更改；Union[List[str], Tuple[str, ...]]：指定的字段是可配置的
config_prefix	配置为非空字符串，则模型将在运行时通过查找 config["configurable"]["{config_prefix}_{param}"] 字段设置配置项；配置为空字符串，那么模型将可以通过 config["configurable"]["{param}"] 字段设置配置项
temperature	采样温度，温度值越高，AI回答越天马行空；温度越低，回答越保守靠谱
max_tokens	要生成的最大令牌数
timeout	请求超时时间
max_retries	最大重试次数
openai_api_key / api_key	OpenAI API密钥。如果未传入，将从环境变量中读取 OPENAI_API_KEY
base_url	API请求的基本URL

init_chat_model() 返回值说明：函数返回一个与指定的model_name和model_provider相对应的 BaseChatModel（如 ChatOpenAI、ChatAnthropic 等）。注意要是模型可配置，则返回一个聊天模型模拟器，该模拟器在传入配置后，于运行时才会初始化底层模型。

示例1：基本用法

使用不同的模型提供方，需要安装为其各自包，与设置各自的API Key环境变量！例如：

• OpenAI环境变量配置为：OPENAI_API_KEY="your_openai_api_key"，安装命令：pip install -U langchain-openai
• Anthropic环境变量配置为：ANTHROPIC_API_KEY="your_anthropic_api_key"，安装命令：pip install -U langchain-anthropic
• DeepSeek环境变量配置为：DEEPSEEK_API_KEY="your_deepseek_api_key"，安装命令：pip install -U langchain-deepseek
• Google VertexAI环境变量配置为：GOOGLE_API_KEY="your_google_api_key"，安装命令：pip install -U langchain-google-vertexai

from langchain.chat_models import init_chat_model
# 返回 langchain_openai.ChatOpenAI 实例
gpt_model = init_chat_model("gpt-5-mini", model_provider="openai", temperature=0)
# 返回 langchain_deepseek.ChatDeepSeek 实例
deepseek_model = init_chat_model("deepseek-chat", model_provider="deepseek", temperature=0)
# 由于所有模型集成都实现了ChatModel接口，因此可以以相同的方式使用它们。
print("gpt-5-mini: " + gpt_model.invoke("what's your name").content + "\n")
print("deepseek-chat: " + deepseek_model.invoke("what's your name").content + "\n")

输出：

gpt-5-mini: I'm called ChatGPT. How can I assist you today?
deepseek-chat: I'm DeepSeek Chat! 😊 You can call me DeepSeek or just Chat if you'd like. I'm here to help with anything you need—ask me anything! 🚀

示例2：创建可配置模型

# 可配置模型模拟器
configurable_model_1 = init_chat_model(temperature=0)
# 动态修改配置，初始化模型并调用
configurable_result_1 = configurable_model_1.invoke(
    "what's your name", config={"configurable": {"model": "gpt-5-mini"}}
)
print("configurable1: " + configurable_result_1.content + "\n")

输出：

configurable1: I'm called ChatGPT. How can I assist you today?

示例3：具有默认值的可配置模型

# 可配置模型模拟器
configurable_model_2 = init_chat_model(
    model="gpt-5-mini",
    temperature=0,
    configurable_fields=("model", "model_provider", "temperature", "max_tokens"),
    config_prefix="first",
)
# 动态修改配置，初始化模型并调用
configurable_result_2 = configurable_model_2.invoke(
    "what's your name?",
    config={
        "configurable": {
            "first_model": "deepseek-chat",
            "first_temperature": 0.5,
            "first_max_tokens": 100,
        }
    },
)
print("configurable2: " + configurable_result_2.content + "\n")

输出：

configurable2: My name is DeepSeek Chat! 😊 I'm here to help you with any questions or topics you're curious about. How can I assist you today?

3.1.4 ChatOllama本地部署

若想使用ChatOllama，需要先安装Ollama包：

pip install -U langchain_ollama

class langchain_ollama.chat_models.base.ChatOllama 是LangChain为通过Ollama部署的聊天模型提供的具体实现类。ChatOllama 同样也实现了标准的Runnable接口。

ChatOllama常用初始化参数说明：

参数名	参数描述
model	要使用的Ollama模型的名称
temperature	采样温度，温度值越高，AI回答越天马行空；温度越低，回答越保守靠谱
timeout	请求超时时间
base_url	API请求的基本URL
num_ctx	设置用于生成下一个令牌的上下文窗口的大小（默认值：2048）
num_gpu	要使用的GPU数量。在macOS上，默认为1表示启用金属支持，默认为0表示禁用

示例：

from langchain_ollama import ChatOllama
ollama_model = ChatOllama(model="deepseek-r1:70b", base_url='http://192.168.100.220:11434')
result = ollama_model.invoke("what's your name?")
print(result)

输出：

content="<think>\n\n</think>\n\nGreetings! I'm DeepSeek-R1, an artificial intelligence assistant created by DeepSeek. I'm at your service and would be delighted to assist you with any inquiries or tasks you may have." 
additional_kwargs={} response_metadata={'model': 'deepseek-r1:70b', 
'created_at': '2025-08-20T06:10:54.9742632Z', 'done': True, 'done_reason': 'stop', 
'total_duration': 9158375400, 'load_duration': 103256900, 
'prompt_eval_count': 8, 'prompt_eval_duration': 1303997800, 'eval_count': 44, 
'eval_duration': 7739543700, 'model_name': 'deepseek-r1:70b'} id='run--c62716da-3c5c-4ddf-8e76-da27dc9291a7-0' usage_metadata={'input_tokens': 8, 
'output_tokens': 44, 'total_tokens': 52}

3.2 聊天模型——调用工具

工具调用根本作用是让大语言模型（LLM）具备与外部世界交互的能力。

LLM本身是一个封闭的知识系统，其能力受限于其训练数据（存在滞后性）和内在的文本生成逻辑。它无法执行直接计算、查询实时信息、操作数据库或调用任何外部API。工具调用打破了这层壁垒，其作用具体体现在：

1. 扩展能力边界：模型可以借助工具完成它自身无法完成的任务，如执行数学计算、搜索网络、查询数据库等。
2. 保证信息实时性：通过调用搜索工具或数据库查询工具，LLM可以获取最新的、训练数据中不存在的信息，避免回答过时或"一本正经地胡说八道"。
3. 处理复杂任务：将一个复杂的用户请求（如"分析我上个月的消费趋势"）分解成多个步骤，并依次调用不同的工具（如"从数据库获取数据" -> “用Python进行数据分析” -> “生成图表”）来协同完成。协调这件事更体现在Agent智能体上。
4. 连接现有系统：可以将企业内部已有的系统、API和数据库封装成工具，让LLM成为一个用自然语言驱动的统一接口，极大地提升了自动化和集成能力。

在LangChain中，聊天模型提供了额外的功能：工具调用。它能使LLM与外部服务、API和数据库进行交互。工具调用还可用于从非结构化数据中提取结构化信息并执行各种其他任务。例如，当我们希望获取当前天气情况时，由于LLM无法获取实时信息，此时我们就可以借助工具，通过外部服务进行搜索完成查询；再例如，当我们希望获取数据库表中的数据时，由于LLM无法直接获取表数据，此时我们就可以借助工具，通过与数据库交互完成查询。

3.2.1 创建工具

3.2.1.1 使用 @tool 装饰器创建工具

在LangChain中，实现了一个 @tool 装饰器来创建工具，@tool 装饰器是自定义工具的最简单方法。如下所示：

from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two integers.
    Args:
        a: First integer
        b: Second integer
    """
    return a * b

print(multiply.invoke({"a": 2, "b": 3}))  # 输出：6
print(multiply.name)                      # 输出：multiply
print(multiply.description)               # 输出：Multiply two ...省略...b: Second integer
print(multiply.args)                      # 输出：{'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}

可以看出，工具通过 @tool 加 Python函数 实现，其中：

• 该装饰器默认使用函数名称作为工具名称。
• 该装饰器将使用函数的文档字符串作为工具的描述。

因此，函数名、类型提示和文档字符串都是传递给工具 Schema 的一部分，不可缺失。定义好的描述是使模型良好运行的重要部分。

什么是Schema？

想象以不同的方式，在JSON中表示有关一个人的信息。

示例1：

{
  "name": "张小红",
  "birthday": "1732年2月22日",
  "address": "陕西省西安市雁塔区"
}

示例2：

{
  "surname": "王",
  "given_name": "刚",
  "birthday": "1732-02-22",
  "address": {
    "district": "萧山区",
    "city": "杭州市",
    "province": "浙江省",
    "country": "中国"
  }
}

这两种表述同样有效，尽管示例2显然比示例1更正式。记录的设计在很大程度上取决于其在应用程序中的预期用途，因此这里没有正确或错误的答案。但是，当应用程序说"给我一个人的JSON记录"时，重要的是要确切地知道该如何组织记录。例如，我们需要知道需要哪些字段，以及如何表示这些值。这就是 JSON Schema 的用武之地。将其转换成编码方式，则为以下内容（此JSON Schema片段描述了上述第二个示例的结构）：

{
  "type": "object",
  "properties": {
    "surname": { "type": "string" },
    "given_name": { "type": "string" },
    "birthday": { "type": "string", "format": "date" },
    "address": {
       "type": "object",
       "properties": {
         "district": { "type": "string" },
         "city": { "type": "string" },
         "province": { "type": "string" },
         "country": { "type" : "string" }
       }
    }
  }
}

若用此 JSON Schema "验证"示例1，那么示例1是不符合当前 JSON Schema 的，但是示例2可以"验证"通过。

注意，JSON Schema 是数据本身，而不是计算机程序，它只是一种"描述其他数据结构"的声明格式。简明扼要地描述数据的表面结构，并根据数据自动验证数据很容易。但是，由于 JSON Schema 不能包含任意代码，因此无法表达数据元素之间的关系存在某些约束。因此，对于足够复杂的数据格式，任何"验证工具"都可能有两个验证阶段：一个在schema（或结构）级别，一个在语义级别。后一种检查可能需要使用更通用的编程语言来实现。

最后回答一下问题：Schema 就是描述其他数据结构的声明格式，用于自动验证数据而存在。

有了以上概念铺垫，对于工具schema，它将从函数名、类型提示和文档字符串中获取相关属性，以此来声明一个工具，包括其名称、描述、输入参数、输出类型等等。这里需要说明的是，若是简单定义工具，如上述示例，工具schema需要解析Google风格的文档字符串去获取【参数描述】。

什么是Google风格的文档字符串？ Google风格是Python文档字符串的一种写作规范。它并非Python语言官方强制要求，而是由Google为其内部Python项目制定的规范，后来因为其极高的可读性和简洁性而在整个Python社区中变得非常流行。它使用 Args:、Returns: 等关键字，参数描述简洁明了，如下所示：

def fetch_data(url, retries=3):
    """从给定的URL获取数据。
    Args:
        url (str): 要从中获取数据的URL。
        retries (int, optional): 失败时重试的次数。默认为3。
    Returns:
        dict: 从URL解析的JSON响应。
    """
    # ... 函数实现 ...

除了这种方式，还有其他方式可以让工具schema获取相关工具声明需要的内容。下面再展示其他常用的工具定义模式。

3.2.1.2 模式1：依赖Pydantic类

若使用 @tool 定义工具时，没有提供文档字符串，则会报错：

from langchain_core.tools import tool

@tool
def add(a: int, b: int) -> int:
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    return a * b

点击运行，报错：ValueError: Function must have a docstring if description not provided.

此时，在LangChain中，可以使用Pydantic类，提供运行时数据验证和类型检查。通过 Field(description="...") 添加字段描述，LangChain会自动提取。注意，除非提供默认值，否则所有字段都是required。如下所示：

from pydantic import BaseModel, Field

class AddInput(BaseModel):
    """Add two integers."""
    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")

class MultiplyInput(BaseModel):
    """Multiply two integers."""
    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")

完整代码如下所示：

# pydantic 数据验证
from pydantic import BaseModel, Field

class AddInput(BaseModel):
    """Add two integers."""
    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")

class MultiplyInput(BaseModel):
    """Multiply two integers."""
    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")

# 定义工具
from langchain_core.tools import tool

@tool(args_schema=AddInput)
def add(a: int, b: int) -> int:
    # 未提供描述
    return a + b

@tool(args_schema=MultiplyInput)
def multiply(a: int, b: int) -> int:
    # 未提供描述
    return a * b

注意是代码中 @tool 的 args_schema 参数，它表示工具函数在未提供描述、文档字符串等需要传递给工具 Schema 的内容时，依赖Pydantic类使用 args_schema 参数，定义并提供工具输入参数的schema。默认为None。

点击运行，不会报错，且将来运行时会进行数据验证。因此，我们再次印证了函数名、类型提示和文档字符串都是传递给工具 Schema 的一部分，不可缺失。

3.2.1.3 模式2：依赖Annotated

在LangChain中，可以依赖 Annotated 和文档字符串传递给工具 Schema。如下所示：

from langchain_core.tools import tool
from typing_extensions import Annotated

@tool
def add(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Add two integers."""
    return a + b

@tool
def multiply(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Multiply two integers."""
    return a * b

3.2.1.4 使用StructuredTool类提供的函数创建工具

class langchain_core.tools.structured.StructuredTool 类用来初始化工具，其中 from_function 类方法通过给定的函数来创建并返回一个工具。from_function 类方法定义如下：

classmethod from_function(
    func: Callable | None = None,
    coroutine: Callable[[...], Awaitable[Any]] | None = None,
    name: str | None = None,
    description: str | None = None,
    return_direct: bool = False,
    args_schema: type[BaseModel] | dict[str, Any] | None = None,
    infer_schema: bool = True,
    *,
    response_format: Literal['content', 'content_and_artifact'] = 'content',
    parse_docstring: bool = False,
    error_on_invalid_docstring: bool = False,
    **kwargs: Any,
) → StructuredTool

关键参数说明：

• func：要设置的工具函数
• coroutine：协程函数，要设置的异步工具函数
• name：工具名称。默认为函数名称。
• description：工具描述。默认为函数文档字符串。
• args_schema：工具输入参数的schema。默认为None。
• response_format：工具响应格式。默认为"content"。
  • 如果配置为 "content"，则工具的输出为ToolMessage的 content 属性。对于 HumanMessage、AIMessage 已经见过，分别表示用户消息和AI消息响应，对于 ToolMessage，它表示对应工具角色所发出的消息。
  • 如果配置为 "content_and_artifact"，则输出应是ToolMessage的 content 属性与 artifact 属性相对应的二元组。

示例1：常规用法

对于用该类方法创建的工具，同样函数名、类型提示和文档字符串也都是传递给工具 Schema 的一部分，不可缺失。

from langchain_core.tools import StructuredTool

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

calculator_tool = StructuredTool.from_function(func=multiply)
print(calculator_tool.invoke({"a": 2, "b": 3}))  # 输出：6

示例2：加入配置，依赖Pydantic类

同样的，让工具函数不提供描述、文档字符串等需要传递给工具 Schema 的内容，此时可以：

1. 使用 args_schema 参数，依赖Pydantic类定义并提供工具输入参数的 schema 属性。
2. 使用 description 参数，替代文档字符串中对于工具描述的 schema 属性。

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")

def multiply(a: int, b: int) -> int:
    return a * b

calculator_tool = StructuredTool.from_function(
    func=multiply,
    name="Calculator",
    description="两数相乘",
    args_schema=CalculatorInput,
)
print(calculator_tool.invoke({"a": 2, "b": 3}))    # 输出：6
print(calculator_tool.name)                        # 输出：Calculator
print(calculator_tool.description)                 # 输出：两数相乘
print(calculator_tool.args)                        # 输出：{'a': {'description': 'first number', 'title': 'A', 'type': 'integer'}, 'b': {'description': 'second number', 'title': 'B', 'type': 'integer'}}

示例3：加入 response_format 配置（理解 artifact）

如果希望我们的工具区分消息内容（content）和其他工件（artifact），让大模型读取 content，而一些用来构造 content 的原始数据保存下来，若后续有一些记录、分析的步骤，就可以派上用场了，这就是 artifact。artifact 通常需要使用字典 Dict 或列表 List 保存。

举个例子再来理解下。例如我们定义了一个搜索天气的 tool，若使用搜索引擎工具询问"今天的天气如何？"时：

• content 可能是：“根据最新搜索结果，今天北京晴，气温在25°C到32°C之间。建议穿短袖衣物。”
• artifact 可能是某搜索引擎API返回的完整JSON响应，其中包含多个搜索结果条目、每个条目的标题、链接、摘要、排名等元数据。如下所示：

# Artifact 的示例结构
{
  'results': [
    {
      'title': '北京天气预报 - 中国天气网',
      'link': 'https://weather.com.cn/...',
      'snippet': '北京今天白天晴，最高气温32°C，夜间晴，最低气温25°C...'
    },
    {
      'title': '北京实时天气 - Weather.com',
      'link': 'https://www.weather.com/...',
      'snippet': 'Bejing, China Weather. Mostly sunny. High 32C...'
    }
    # ... 更多结果
  ],
  'search_parameters': { ... },
  'search_information': { ... }
}

则对于以上原生数据，无论我们今后做日志记录、分析，或自定义后续的处理都很方便。例如存在以下场景：

1. 我们不仅仅想要一个总结性的答案，还想要具体的链接、来源或多个备选答案。
2. 工具的 content 输出不符合你的预期，我们想查看原始数据来理解问题出在哪里（是工具解析的问题，还是API本身返回的问题）。
3. 需要记录每次工具调用的完整原始响应，以满足数据分析的要求。

从这里就可以对比出只返回content无法做到这些事情。如何做到？我们需要在定义工具时指定 response_format="content_and_artifact" 参数，并确保我们返回一个元组 (content, artifact)，代码如下：

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
from typing import List, Tuple

class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")

def multiply(a: int, b: int) -> Tuple[str, List[int]]:
    nums = [a, b]
    content = f"{nums}相乘的结果是{a * b}"
    return content, nums

calculator_tool = StructuredTool.from_function(
    func=multiply,
    name="Calculator",
    description="两数相乘",
    args_schema=CalculatorInput,
    response_format="content_and_artifact"
)

上面的代码中，我们将需要相乘的数据当作原始数据，作为样例（说明：@tool 也支持 response_format 参数）。

如果我们直接使用工具参数调用工具，将只返回输出的 content 部分：

print(calculator_tool.invoke({"a": 2, "b": 3}))    # 输出：[2, 3]相乘的结果是6

若想要看到工具返回的元组，我们需要模拟大模型调用工具的姿势，如下所示。这将返回一个 ToolMessage：

print(calculator_tool.invoke(
    {
        "name": "Calculator",
        "args": {"a": 2, "b": 3},
        "id": "123",          # 必须，与工具调用关联的标识符，将工具调用请求与工具调用结果相关联。
        "type": "tool_call",  # 必须
    }
))

结果如下：

# ToolMessage
content='[2, 3]相乘的结果是6' name='Calculator' tool_call_id='123' artifact=[2, 3]

小结一下，由于LLM大多理解文本，所以工具的主要输出 content 必须是结构良好、简洁的文本，以便模型能够轻松理解和基于它进行推理、生成下一步的指令。在链（Chain）中，工具调用之后的其他组件或函数，可能需要工具的原始且结构化数据（即 artifact）来执行特定操作。这些数据可能是庞大的、且非文本的，这些数据不适合直接塞给模型。因此，artifact 其实是为了给链中后续的组件或函数使用的，不被大模型所直接使用！！

3.2.2 绑定工具

为了实际将这些工具绑定到聊天模型，可以使用聊天模型的 .bind_tools() 方法。如下所示：

from langchain_openai import ChatOpenAI
# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")
# ...
# 绑定工具，返回一个 Runnable 实例
tools = [add, multiply]
model_with_tools = model.bind_tools(tools)

bind_tools() 方法定义：

bind_tools(
    tools: Sequence[dict[str, Any] | type | Callable | BaseTool],
    *,
    tool_choice: dict | str | Literal['auto', 'none', 'required', 'any'] | bool
 | None = None,
    strict: bool | None = None,
    parallel_tool_calls: bool | None = None,
    **kwargs: Any,
) → Runnable[PromptValue | str | Sequence[BaseMessage | list[str] | 
tuple[str, str] | str | dict[str, Any]], BaseMessage]

请求参数：

• tools：绑定到此聊天模型的工具定义列表。支持的类型为：字典、pydantic.BaseModel类、Python函数和BaseTool（如 @tool 装饰器创建的类）。
• tool_choice（默认空）：要求模型调用哪个工具。可以设置为：
  • 形式为 '<tool_name>' 的str：调用指定名称的工具。
  • 'auto'：自动选择工具（包括无工具）。
  • 'none'：不调用工具。
  • 'any' 或 'required' 或 True：强制调用至少一个工具。
  • False 或 None：无效果，默认OpenAI的行为。
• strict（默认空）：
  • 如果为 True，则保证模型输出与工具定义中提供的 JSON Schema 完全匹配。输入也将根据提供的 Schema 进行验证。
  • 如果为 False，则不会验证输入，也不会验证模型输出。
  • 如果为 None，则不会将 strict 参数传递给模型。
• parallel_tool_calls：默认为 None，允许并行工具使用。设置为 False 以禁用并行工具。
• kwargs(Any)：任何附加参数都直接传递给 bind()。

返回值：

• 返回一个Runnable实例。
  • 该实例支持多种格式输入：
    • 原始提示PromptValue
    • 字符串："上海天气如何？"
    • 消息或消息列表：[HumanMessage(content="...")]
  • 该实例的输出：
    • 包含工具调用信息的 AIMessage

3.2.3 工具调用

通过 .bind_tools() 方法我们可知，它返回了一个Runnable实例，因此我们可以使用该Runnable实例，调用 .invoke() 方法，完成工具调用。示例如下：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from typing_extensions import Annotated

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

@tool
def add(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Add two integers."""
    return a + b

@tool
def multiply(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Multiply two integers."""
    return a * b

# 绑定工具
tools = [add, multiply]
model_with_tools = model.bind_tools(tools)

# 调用工具
result = model_with_tools.invoke("9乘6等于多少？")
print(result)

输出结果（AIMessage）：

content='' additional_kwargs={'tool_calls': [{'id': 
'call_mBnxNMY7vfAOrExEETcdEC6l', 'function': {'arguments': '{"a":9,"b":6}', 
'name': 'multiply'}, 'type': 'function'}], 'refusal': None} response_metadata=
{... 'model_name': 'gpt-4o-mini-2024-07-18', ...} 
tool_calls=[{'name': 'multiply', 'args': {'a': 9, 'b': 6}, 'id': 
'call_mBnxNMY7vfAOrExEETcdEC6l', 'type': 'tool_call'}] 
usage_metadata={'input_tokens': 86, 'output_tokens': 17, 'total_tokens': 103, ...}

输出说明：

• AIMessage：来自AI的消息。从聊天模型返回，作为对提示（输入）的响应。
  • content：消息的内容。
  • additional_kwargs：与消息关联的其他有效负载数据。对于来自AI的消息，可能包括模型提供程序编码的工具调用。
  • response_metadata：响应元数据。例如：响应标头、logprobs、令牌计数、模型名称。

从输出结果看来，AI给出的响应是进行工具的调用！！

工具调用的一个关键原则是，模型根据输入的相关性决定何时使用工具。模型并不总是需要调用工具。例如，给定一个不相关的输入，模型不会调用该工具：

result = model_with_tools.invoke("hello world!")
print(result)

输出结果（AIMessage）：

content='Hello! How can I assist you today?' additional_kwargs={'refusal': 
None} response_metadata={... 'finish_reason': 'stop', ...} 

3.2.4 强制模型调用工具

当然我们也可以让模型强制调用工具，那就需要在绑定工具时，设置 tool_choice="any"，表示强制调用至少一个工具。示例如下：

model_with_tools = model.bind_tools(tools, tool_choice="any")
result = model_with_tools.invoke("hello world!")
print(result)

输出结果（AIMessage）：

content='' additional_kwargs={'tool_calls': [{'id': 
'call_zM7F5zPKzzMbTs2C2VtJdAot', 'function': {'arguments': '{"a":1,"b":2}', 
'name': 'add'}, 'type': 'function'}], 'refusal': None} response_metadata=
{... 'finish_reason': 'tool_calls', ...} 
tool_calls=[{'name': 'add', 'args': {'a': 1, 'b': 2}, 'id': 
'call_zM7F5zPKzzMbTs2C2VtJdAot', 'type': 'tool_call'}] usage_metadata=
{'input_tokens': 82, 'output_tokens': 17, 'total_tokens': 99, ...}

3.2.5 工具属性

现在我们知道，输出结果是一个 AIMessage。但是，如果调用了工具，则 result 将具有一个 tool_calls 属性。此属性包括执行该工具所需的一切，包括工具名称和输入参数，示例如下：

model_with_tools = model.bind_tools(tools, tool_choice="any")
result = model_with_tools.invoke("9乘6等于多少？")
print(result.tool_calls)

输出结果：

[{'name': 'multiply', 'args': {'a': 9, 'b': 6}, 'id': 
'call_csFbMmYD4Dmz8yMja3ZWK1QW', 'type': 'tool_call'}]

3.2.6 将工具输出传递给聊天模型

到这里可以发现，我们仅仅只是成功调用了工具，但是聊天模型并没有给我们返回我们真正需要的答案。此时就需要：

1. 将工具输出传递给聊天模型，包括 HumanMessage、AIMessage（工具调用）、ToolMessage
2. 聊天模型根据以上消息输入，将最终结果 AIMessage 返回。

为什么要发ToolMessage呢？

之前我们讲过，聊天模型通常不是接受单个字符串作为输入，而是接受聊天消息（XxxMessage）列表，因此在这里我们需要将工具的返回，构造成 ToolMessage，再传输给聊天模型！！！

方便的是，如果我们使用 @tool 装饰器创建的工具，使用 tool.invoke(tool_calls)，将自动返回一个 ToolMessage。完整示例如下：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from typing_extensions import Annotated

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 定义工具
@tool
def add(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Add two integers."""
    return a + b

@tool
def multiply(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Multiply two integers."""
    return a * b

# 绑定工具
tools = [add, multiply]
model_with_tools = model.bind_tools(tools)

# 添加AIMessage到消息中去
messages = [
    HumanMessage("9乘6等于多少？5加3等于多少？")
]
ai_msg = model_with_tools.invoke(messages)
messages.append(ai_msg)

for tool_call in ai_msg.tool_calls:
    # 根据工具名选择对应工具函数（不区分大小写）
    selected_tool = {"add": add, "multiply": multiply}[tool_call["name"].lower()]
    # 执行工具调用，返回 ToolMessage
    tool_msg = selected_tool.invoke(tool_call)
    # 将 ToolMessage 加入消息
    messages.append(tool_msg)

print(messages)
result = model.invoke(messages)
print(result)

打印结果如下：

[HumanMessage(content='9乘6等于多少？5加3等于多少？', ...), 
AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_tZhIzjEQrLk0GquaKXb6JZMy', 'function': {'arguments': '{"a": 9, "b": 6}', 'name': 'multiply'}, 'type': 'function'}, {'id': 'call_fesQTvJGD7AiuRi1hnBJ67el', 'function': {'arguments': '{"a": 5, "b": 3}', 'name': 'add'}, 'type': 'function'}], 'refusal': None}, ..., tool_calls=[{'name': 'multiply', 'args': {'a': 9, 'b': 6}, 'id': 'call_tZhIzjEQrLk0GquaKXb6JZMy', 'type': 'tool_call'}, {'name': 'add', 'args': {'a': 5, 'b': 3}, 'id': 'call_fesQTvJGD7AiuRi1hnBJ67el', 'type': 'tool_call'}], ...), 
ToolMessage(content='54', name='multiply', tool_call_id='call_tZhIzjEQrLk0GquaKXb6JZMy'), 
ToolMessage(content='8', name='add', tool_call_id='call_fesQTvJGD7AiuRi1hnBJ67el')]

content='9乘6等于54，5加3等于8。' additional_kwargs={'refusal': None} 
response_metadata={... 'finish_reason': 'stop', ...} 

从流程与代码中可以看到，实际上我们调用了两次聊天模型：

• 第一次：仅将【HumanMessage】发送给聊天模型进行处理，结果返回了【包含工具调用的AIMessage】，并没有返回我们想要的结果。然后我们执行工具，得到 ToolMessage。
• 第二次：将【HumanMessage + AIMessage + ToolMessage】消息记录发送给聊天模型进行处理，结果返回了【包含结果的 AIMessage】。

3.2.7 LangChain提供的工具：TavilySearch

工具也不是全部都需要我们自己手搓，其实LangChain官方也已经给我们提供了很多现成的工具（Tool）和工具包（Toolkit）。

LangChain提供的工具有搜索、数据库、网页浏览器等相关的工具。LangChain中的工具实际上是继承了 BaseTool 与 BaseToolkit。我们可以从下载的关于LangChain包的Lib中查看，全局搜索 (BaseTool) 和 (BaseToolkit)，展示的全部是不同功能的现成的工具，我们可以直接使用它们！找到相关的类名称，去官网API Reference全局搜索，即可获取工具的详细说明与用法。

下面我们简单看一个搜索工具类。

TavilySearch

TavilySearch类可以支持我们进行搜索。Tavily是一个专门为AI设计的搜索引擎，专为智能体检索与推理需求量身打造的工具。

Tavily不仅提供了高度可编程的API接口，还具备显著优于传统搜索引擎的上下文相关性理解能力。能够以结构化、可解析的形式返回搜索结果，便于将检索到的信息直接用于后续的推理、生成或任务执行流程。

• Tavily官网：https://www.tavily.com/ ，需魔法使用。登录完成后，新建API Keys。
• 点击左侧API Playground，可以使用刚申请的API Keys，进行搜索测试。这会返回根据查询内容得到的多条搜索结果。
• 点击左侧Use Cases，可以试用提供好的案例，如聊天中可以支持搜索（Chat）。

下面我们在LangChain中接入该搜索工具，步骤如下：

1. 安装 langchain-tavily 包：

pip install -U langchain-tavily

2. 配置环境变量 TAVILY_API_KEY，值为我们申请的API Key。

3. 代码接入TavilySearch类，实现搜索功能：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_tavily import TavilySearch

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")
# 绑定工具
tool = TavilySearch(max_results=4)  # max_results 返回的最大搜索结果
model_with_tools = model.bind_tools([tool])

# 添加AIMessage到消息中去
messages = [
    HumanMessage("中国西安今天的天气怎么样？")
]
ai_msg = model_with_tools.invoke(messages)
messages.append(ai_msg)

for tool_call in ai_msg.tool_calls:
    # 执行工具调用，返回 ToolMessage
    tool_msg = tool.invoke(tool_call)
    # 将 ToolMessage 加入消息
    messages.append(tool_msg)

result = model_with_tools.invoke(messages)
print(result.content)

结果打印：

今天西安的天气情况如下：
- **天气**：晴天转多云
- **最高气温**：约31℃
- **最低气温**：约21℃
- 预计在晚上会有降雨。
有关更详细的信息，你可以查看[中国气象局的天气预报](https://weather.cma.cn/web/weather/V8870.html)或其他相关天气网站。

3.3 聊天模型——结构化输出

在LangChain中，聊天模型提供了额外的功能：结构化输出。一种使聊天模型以结构化格式（例如JSON）进行响应的技术。例如，可能希望将模型输出存储在数据库中，并确保输出符合数据库模式。这种需求激发了结构化输出的概念，其中可以指示模型使用特定的输出结构进行响应。

这样做的核心原因是：从"字符串"到"对象"的范式转换。想象一下，在没有这个功能之前，我们调用聊天模型得到的是一个 AIMessage，其内容是一个字符串。例如下述伪代码：

model = ChatOpenAI()
response = model.invoke("告诉我关于苹果公司的最新消息。")
print(response.content)
# 输出: "苹果公司于昨日发布了新款iPhone...其股价上涨了2%..."

这个字符串对人类很友好，但对程序不友好。如果我们想从这段文本中提取出"公司名"和"股价变化"并用在后续逻辑中，则需要编写复杂且容易出错的解析代码（例如，使用正则表达式）。

聊天模型的 with_structured_output 方法则允许我们预先定义一个期望的数据结构，并要求大模型必须按照这个结构返回信息。

3.3.1 with_structured_output()方法

要想使用结构化输出能力，LangChain提供了一种方法 .with_structured_output()，该方法需要先定义输出结构，然后执行通过 .with_structured_output() 得到的Runnable实例。步骤如下（伪代码）：

# 1. 定义输出结构
schema = {"foo": "bar"}
# 2. 绑定schema，其实是生成支持结构化返回的 Runnable 实例
model_with_structure = model.with_structured_output(schema)
# 3. 执行
structured_output = model_with_structure.invoke(user_input)

这是获得结构化输出的最简单、最可靠的方法。此方法将输出结构作为参数输入，返回一个类似model的Runnable。不同之处在于执行Runnable后的输出结果，输出的不是字符串或消息，而是输出与给定输出结构相对应的对象。

该输出结构可以指定为TypedDict类、JSON Schema或Pydantic类。如果使用TypedDict或JSON Schema，则Runnable将返回一个字典，如果使用Pydantic类，则将返回一个Pydantic对象。

with_structured_output() 方法定义：

with_structured_output(
    schema: dict[str, Any] | type[_BM] | type | None = None,
    *,
    method: Literal['function_calling', 'json_mode', 'json_schema'] = 'json_schema',
    include_raw: bool = False,
    strict: bool | None = None,
    **kwargs: Any,
) → Runnable[PromptValue | str | Sequence[BaseMessage | list[str] | 
tuple[str, str] | str | dict[str, Any]], dict | _BM]

请求参数：

• schema：表示输出结构。可以传入为：JSON、TypedDict、Pydantic、OpenAI函数/工具
• method：表示LLM的生成方法：
  • json_schema（默认）：表示使用OpenAI的结构化输出API。
  • function_calling：使用OpenAI的工具调用（以前称为函数调用）
  • json_mode：使用OpenAI的 JSON mode。请注意，如果使用 json_mode，则必须在模型调用中包含将输出格式化为所需 schema 的说明。
• include_raw：
  • 如果为 False（默认），则仅返回解析的结构化输出。如果在模型输出解析过程中发生错误，则会引发错误。
  • 如果为 True，则将返回原始模型响应（BaseMessage）和解析的模型响应。如果在输出解析过程中发生错误，它也会被捕获并返回。最终输出始终是带有键 “raw”、“parsed” 和 “parsing_error” 的字典。
• strict：
  • 如果为 True，保证模型输出与 schema 完全匹配。输入 schema 也将根据支持的schema进行验证。
  • 如果为 False，输入 schema 将不会被验证，模型输出也不会被验证。
  • 如果为 None（默认），则不会将 strict 参数传递给模型。
• tools：要绑定到聊天模型的工具列表。要求：method 为 'json_schema'、strict=True、include_raw=True。则生成的AIMessage将在 ‘raw’ 中包含工具调用。
• kwargs(Any)：任何附加参数都直接传递给 bind()。

返回值：返回一个Runnable实例。将来执行时：

• 如果 include_raw=False
  • 且 schema是Pydantic类，则Runnable会输出Pydantic对象。
  • 否则Runnable输出一个字典。
• 如果 include_raw=True，则Runnable输出一个带有键的字典：
  • 'raw'：BaseMessage
  • 'parsed'：如果出现解析错误，则为None，否则类型取决于如上所述的 schema。
  • 'parsing_error'：Optional[BaseException]

3.3.2 返回Pydantic对象

我们可以设置执行Runnable后的输出结果指定为 Pydantic 类，这将返回一个Pydantic对象。当收到模型的响应后，LangChain会提取出代表 Pydantic 参数的JSON对象，并用 Pydantic 模型对其进行解析和验证，将这个验证后的JSON转换为一个可用的 Pydantic 对象实例返回。如下所示：

from langchain_openai import ChatOpenAI
from typing import Optional
from pydantic import BaseModel, Field

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 定义输出结构：Pydantic 类
class Joke(BaseModel):
    """给用户讲一个笑话。"""
    setup: str = Field(description="这个笑话的开头")
    punchline: str = Field(description="这个笑话的妙语")
    rating: Optional[int] = Field(
        default=None, description="从1到10分，给这个笑话评分"
    )

structured_model = model.with_structured_output(Joke)
result = structured_model.invoke("给我讲一个关于唱歌的笑话")
print(result)

打印结果：

setup='为什么歌手总是喜欢在洗手间里唱歌？' punchline='因为那里有很好的回音和灵感！' 
rating=7

还支持嵌套输出：

from langchain_openai import ChatOpenAI
from typing import Optional, List
from pydantic import BaseModel, Field

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 定义输出结构：Pydantic 类
class Joke(BaseModel):
    """给用户讲一个笑话。"""
    setup: str = Field(description="这个笑话的开头")
    punchline: str = Field(description="这个笑话的妙语")
    rating: Optional[int] = Field(
        default=None, description="从1到10分，给这个笑话评分"
    )

class Data(BaseModel):
    """获取关于笑话的数据。"""
    jokes: List[Joke]

structured_model = model.with_structured_output(Data)
result = structured_model.invoke("分别讲一个关于唱歌和跳舞的笑话")
print(result)

打印结果：

jokes=[
    Joke(setup='为什么唱歌的人总是很快乐？', punchline="因为他们总是'音'乐满满！", rating=8), 
    Joke(setup='一个跳舞的牛走进俱乐部，为什么大家都不理它？', punchline="因为它总是'踏'错节拍！", rating=7)
]

3.3.3 返回TypedDict

先了解一下 TypedDict，它用于为字典对象提供精确的、结构化的类型提示。它允许我们指定一个字典中应该有哪些键，以及每个键对应的值的类型。

最清晰、最常用的定义方式，就是类似于定义一个类（Python 3.8+），如下所示：

from typing import TypedDict

class User(TypedDict):
    name: str
    age: int
    email: str
    is_active: bool = True  # 默认值

这有什么用呢？对于它非常重要的一个能力就是捕捉键名拼写错误与类型错误。如：

user1: User = {
    "name": "Bob",
    "age": 25,
    "email": "bob@example.com"
}
# 类型检查器会捕获这些错误
bad_user: User = {
    "name": "Dave",
    "age": "forty",  # 错误：应该是int
    "emial": "dave@example.com"  # 错误：拼写错误
}

因此，我们也可以设置执行Runnable后的输出结果指定为TypedDict类，这将返回一个字典，且输出后，会根据设定进行验证。以下是该方法的使用姿势：

from langchain_openai import ChatOpenAI
from typing import Optional
from typing_extensions import Annotated, TypedDict

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 定义输出结构：TypedDict
class Joke(TypedDict):
    """给用户讲一个笑话。"""
    setup: Annotated[str, ..., "这个笑话的开头"]
    punchline: Annotated[str, ..., "这个笑话的妙语"]
    rating: Annotated[Optional[int], None, "从1到10分，给这个笑话评分"]

structured_model = model.with_structured_output(Joke)
result = structured_model.invoke("给我讲一个关于唱歌的笑话")
print(result)

打印结果：

{'setup': '为什么歌手总是带一把伞？', 'punchline': '因为他们怕下雨时会错过音调！', 'rating': 7}

让我们加入 include_raw=True，再来看看效果：

structured_model = model.with_structured_output(Joke, include_raw=True)
result = structured_model.invoke("给我讲一个关于唱歌的笑话")
print(result)

打印结果：

{
    'raw': AIMessage(content='{"setup":"你知道为什么歌手总是喜欢在吃饭的时候唱歌吗？","punchline":"因为他们想要增加自己的\'调味\'！","rating":7}', additional_kwargs={'parsed': None, 'refusal': None}, response_metadata={... 'model_name': 'gpt-4o-mini-2024-07-18', ... 'finish_reason': 'stop', ...}, usage_metadata={...}), 
    'parsed': {'setup': '你知道为什么歌手总是喜欢在吃饭的时候唱歌吗？', 'punchline': '因为他们想要增加自己的\'调味\'！', 'rating': 7}, 
    'parsing_error': None
}

3.3.4 返回JSON

还可以让聊天模型直接返回JSON，只不过为了声明JSON，我们需要定义JSON Schema，如下所示：

from langchain_openai import ChatOpenAI

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

json_schema = {
    "title": "joke",
    "description": "给用户讲一个笑话。",
    "type": "object",
    "properties": {
        "setup": {
            "type": "string",
            "description": "这个笑话的开头",
        },
        "punchline": {
            "type": "string",
            "description": "这个笑话的妙语",
        },
        "rating": {
            "type": "integer",
            "description": "从1到10分，给这个笑话评分",
            "default": None,
        },
    },
    "required": ["setup", "punchline"],
}

structured_model = model.with_structured_output(json_schema)
result = structured_model.invoke("给我讲一个关于唱歌的笑话")
print(result)

打印结果：

{'setup': '为什么唱歌的人总是很开心？', 'punchline': '因为他们总是有很多音符可供选择！', 'rating': 7}

3.3.5 选择输出格式

创建具有联合类型属性的父模式，以使用Pydantic为例（其他同理），代码如下：

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import Optional, Union

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

class Joke(BaseModel):
    """给用户讲一个笑话。"""
    setup: str = Field(description="这个笑话的开头")
    punchline: str = Field(description="这个笑话的妙语")
    rating: Optional[int] = Field(
        default=None, description="从1到10分，给这个笑话评分"
    )

class ConversationalResponse(BaseModel):
    """以对话的方式回应。待人友善，乐于助人。"""
    response: str = Field(description="对用户查询的会话响应")

class FinalResponse(BaseModel):
    final_output: Union[Joke, ConversationalResponse]

structured_model = model.with_structured_output(FinalResponse)
result = structured_model.invoke("给我讲一个关于唱歌的笑话")
print(result)
result = structured_model.invoke("你好")
print(result)

打印结果：

final_output=Joke(setup='为什么歌手总是带着梯子？', punchline='因为他们想要达到更高的音调！', rating=7)
final_output=ConversationalResponse(response='你好！有什么我可以帮助你的吗？')

3.3.6 实用场景

场景1：作为信息提取器

from langchain_openai import ChatOpenAI
from typing import Optional
from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage, SystemMessage

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

class Person(BaseModel):
    """一个人的信息。"""
    # 注意:
    # 1. 每个字段都是 Optional "可选的" —— 允许 LLM 在不知道答案时输出 None。
    # 2. 每个字段都有一个 description "描述" —— LLM使用这个描述。
    # 有一个好的描述可以帮助提高提取结果。
    name: Optional[str] = Field(default=None, description="这个人的名字")
    hair_color: Optional[str] = Field(default=None, description="如果知道这个人头发的颜色")
    skin_color: Optional[str] = Field(default=None, description="如果知道这个人的肤色")
    height_in_meters: Optional[str] = Field(default=None, description="以米为单位的高度")

structured_model = model.with_structured_output(schema=Person)

messages = [
    SystemMessage(content="你是一个提取信息的专家，只从文本中提取相关信息。如果您不知道要提取的属性的值，属性值返回null"),
    HumanMessage(content="史密斯身高6英尺，金发。")
]
result = structured_model.invoke(messages)
print(result)

打印结果：

name='史密斯' hair_color='金发' skin_color=None height_in_meters='1.83'

场景2：使用"少样本提示"来增强信息提取能力

由于[少样本提示]能力还未讲解，这部分代码示例放在[少样本提示]部分看，可直接跳过去看（见4.3.4节）。

场景3：与工具结合使用

注意，使用聊天模型原生的工具搭配结构化输出并不好用！！！这里只是了解下其能力。更好的用法见LangGraph Agent能力。

方式1：使用 with_structured_output() 的 tools 参数

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 结构输出对象
class SearchResult(BaseModel):
    """结构化搜索结果。"""
    query: str = Field(description="搜索查询")
    findings: str = Field(description="调查结果摘要")

@tool
def web_search(query: str) -> str:
    """在网上搜索信息。
    Args:
        query: 搜索查询
    """
    return "西安今天多云转小雨，气温18-23度，东南风2级，空气质量良好"

structured_search_model = model.with_structured_output(
    SearchResult,
    tools=[web_search],
    strict=True,
    include_raw=True,
)
result = structured_search_model.invoke("搜索当前最新的西安的天气")
print(result)

输出结果：

{
    'raw': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_8CJNlQLUOPcujpPuGi8Ltnu7', 'function': {'arguments': '{"query":"西安天气预报"}', 'name': 'web_search', 'parsed_arguments': {'query': '西安天气预报'}}, 'type': 'function'}], 'parsed': None, 'refusal': None}, response_metadata={... 'finish_reason': 'tool_calls', ...}, tool_calls=[{'name': 'web_search', 'args': {'query': '西安天气预报'}, 'id': 'call_8CJNlQLUOPcujpPuGi8Ltnu7', 'type': 'tool_call'}], ...), 
    'parsed': None, 
    'parsing_error': None
}

可以看到，它并不能直接帮我们输出想要的SearchResult搜索结果，而是只返回了AIMessage（工具调用信息）。

实际上，with_structured_output 方法只是让模型知道有哪些工具可以调用，但是并不会自动执行工具。要获得工具执行后的结果并整合到最终的结构化输出中，我们需要手动处理工具调用，然后将工具返回的结果再次传递给模型。见方式2。

方式2：拆解能力，单独依次执行

当需要同时使用结构化输出和其他工具时，需要注意顺序，不要弄反：

1. 首先绑定工具
2. 其次添加结构化输出

代码演示：

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 结构输出对象
class SearchResult(BaseModel):
    """结构化搜索结果。"""
    query: str = Field(description="搜索查询")
    findings: str = Field(description="调查结果摘要")

@tool
def web_search(query: str) -> str:
    """在网上搜索信息。
    Args:
        query: 搜索查询
    """
    # 模拟搜索返回结果
    return "西安今天多云转小雨，气温18-23度，东南风2级，空气质量良好"

# 一次性绑定工具和结构化输出
model_with_tools = model.bind_tools([web_search]).with_structured_output(SearchResult)
# 执行查询
messages = [HumanMessage("搜索当前最新的西安的天气")]
result = model_with_tools.invoke(messages)
print(result)

打印结果：

query='西安的天气' findings='当前西安的天气情况：预计多云，气温在18°C到25°C之间，有轻微的风，空气湿度适中，无降水。'

从打印结果可以发现，它也没有去调用工具，而是聊天模型直接给出了结果。让我们改造下：

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 结构输出对象
class SearchResult(BaseModel):
    """结构化搜索结果。"""
    query: str = Field(description="搜索查询")
    findings: str = Field(description="调查结果摘要")

@tool
def web_search(query: str) -> str:
    """在网上搜索信息。
    Args:
        query: 搜索查询
    """
    return "西安今天多云转小雨，气温18-23度，东南风2级，空气质量良好"

# 手动将工具结果加入消息列表
model_with_search = model.bind_tools([web_search])
messages = [
    HumanMessage("搜索当前最新的西安的天气")
]
ai_msg = model_with_search.invoke(messages)
messages.append(ai_msg)
for tool_call in ai_msg.tool_calls:
    tool_msg = web_search.invoke(tool_call)
    messages.append(tool_msg)

structured_search_model = model_with_search.with_structured_output(SearchResult)
result = structured_search_model.invoke(messages)
print(result)

打印结果：

query='西安天气' findings='西安今天多云转小雨，气温18-23度，东南风2级，空气质量良好。'

这次的结果符合我们的预期。但依旧很麻烦，因为其实我们实际上调用了两次模型：

• 第一次调用模型时，模型返回了工具调用，然后我们执行工具，将工具结果加入消息列表。
• 第二次调用时，我们使用结构化输出SearchResult的模型，来处理整个消息列表（包括工具调用的结果），然后输出结构化结果。

有没有只手动调用一次（实际上允许调用两次模型）就能获取到结构化输出的结果呢？将来在学习LangGraph部分的Agent就可以做到这件事。

3.4 聊天模型——流式传输

流式处理对于使基于LLM的应用程序能够响应最终用户至关重要。其通过逐步显示输出，甚至在完整的响应准备就绪之前，流式传输可以显著改善用户体验。

我们之前直接使用invoke的调用方式属于非流式传输，看到的现象是聊天模型直接返回全量内容，若模型思考时间较长，则我们等待的时间就越长。如下所示：

# 伪代码
model = ChatOpenAI()
model.invoke("讲一个1000字的笑话")

我们等待了20s之后，返回了全量结果：

从前有一个小镇，...... ，毕竟，生活中总需要一些笑声来调剂。

从结果看来，我们等待的时间太长了，对于用户来说，太长的等待时间严重影响体验。而使用流式传输（如DeepSeek客户端默认的流式返回），效果会得到显著提升！那么对于LangChain的聊天模型来说，它同样支持流式返回。

3.4.1 stream()同步传输

在LangChain聊天模型中，可以使用其 .stream() 方法，来同步生成流式响应的效果。

聊天模型的 .stream() 方法返回一个迭代器，该迭代器在生成输出时同步产生输出消息块。可以使用 for 循环实时处理每个块。代码如下：

from langchain_openai import ChatOpenAI

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 流式输出
chunks = []
for chunk in model.stream("讲一个50字的笑话"):
    chunks.append(chunk)
    print(chunk.content, end="|", flush=True)

打印结果：

|有|一天|，|兔|子|和|乌|龟|比|赛|跑|步|。|兔|子|跑|得|飞|快|，|没|想到|中|途|睡|着|了|。|等|它|醒|来|，|发现|乌|龟|快|到|终|点|了|，|兔|子|急|了|："|怎么|可能|！"|乌|龟|笑|着|说|："|慢|就是|快|，我|在|享|受|风|景|！"|

通过调试，让我们来看下chunk是什么：我们得到了一个叫做 AIMessageChunk 的东西，它代表 AIMessage 的一部分，也就是消息块。

消息块还可以直接相加，来看效果：

print("\n")
print(chunks[0] + chunks[1] + chunks[2] + chunks[3] + chunks[4])

结果如下（AIMessageChunk）：

content='有一天，兔' additional_kwargs={} response_metadata={} id='run--e619cbc3-9ee9-4ae7-a73e-32edb166d401'

3.4.2 astream()异步传输

对于流式传输，通常我们可以选择异步调用。先来了解下异步相关知识。

3.4.2.1 异步相关概念

想象一个场景：你需要煮一壶水，同时还要给朋友发一条短信。我们分别用同步（传统）和异步两种方式来完成，以此对比并引入协程和事件循环的概念。

同步（阻塞）方式：这就像是一个"死心眼"的人，做事必须一件一件来：

import time

def boil_water():
    print("开始煮水...")
    time.sleep(5)  # 模拟阻塞等待5秒
    print("水开了！")

def send_message():
    print("开始发短信...")
    time.sleep(2)  # 模拟阻塞等待2秒
    print("短信发送成功！")

# 主程序
def main():
    boil_water()  # 先花5秒煮水，期间什么也不能做
    send_message() # 水开后再花2秒发短信

main()

总耗时：7秒

问题：在 boil_water 函数等待的5秒里，CPU完全空闲，但却不能去做 send_message 任务，效率低下。

异步方式：我们请出 asyncio、协程和事件循环。

什么是协程？

• 多进程通常利用的是多核CPU的优势，同时执行多个计算任务。每个进程有自己独立的内存管理，所以不同进程之间要进行数据通信比较麻烦。
• 多线程是在一个CPU上创建多个子任务，当某一个子任务休息的时候其他任务接着执行。多线程的控制是由Python自己控制的。线程存在数据同步问题，所以要有锁机制。
• 协程的实现是在一个线程内实现的，相当于流水线作业。由于线程切换的消耗比较大，所以对于并发编程，可以优先使用协程。

协程，作为一种轻量级的并发编程模型，可以被视为用户态的"轻量级线程"。与传统线程相比，协程的核心优势在于其调度完全由用户空间掌控，避免了操作系统内核的频繁介入，从而显著降低了上下文切换的开销。在诸如网络数据刷新、资源加载、用户界面更新、以及I/O读写等场景下，如果并发任务的计算量相对较小、对系统资源占用较低，则不必动用操作系统级别的线程。

协程的切换则由程序员和编程语言控制，程序员决定在何时暂停或恢复协程。协程是一个特殊的函数，它可以在执行过程中暂停，并在稍后恢复执行。它用 async def 定义，并在需要暂停的地方使用 await。

在我们的例子里，boil_water_async 和 send_message_async 就是两个协程。

import asyncio

# 定义协程
async def boil_water_async():
    print("开始煮水...")
    await asyncio.sleep(5)  # 关键！await 表示"等待这个操作完成，但期间让事件循环去做别的事"
    print("水开了！")

async def send_message_async():
    print("开始发短信...")
    await asyncio.sleep(2)  # 同样，等待2秒，但让出控制权
    print("短信发送成功！")

什么是事件循环？

事件循环是 asyncio（Python标准库中的模块，用于编写异步I/O操作的代码）的核心，你可以把它想象成一个总调度员或一个高效的待办事项(To-Do List)管理员。它的工作流程非常简单：

1. 它维护着一个任务列表（比如：煮水、发短信）。
2. 它不断地循环检查每个任务：
   • 如果任务处于"等待I/O"状态（比如等水开、等网络响应），就暂停它，立即去执行下一个已经"就绪"的任务。
   • 如果任务的等待时间到了或者I/O操作完成了，事件循环就恢复执行这个任务。

如何运行？

# 主程序（也是一个协程）
async def main():
    # 创建两个任务，并交给事件循环去调度
    task1 = asyncio.create_task(boil_water_async())
    task2 = asyncio.create_task(send_message_async())
    # 等待两个任务都完成
    await task1
    await task2

# 它负责创建事件循环，并将第一个协程（主程序）放入其中运行。
asyncio.run(main())

输出结果：

开始煮水...  # 任务1开始
开始发短信... # 任务1遇到await，立即让出控制权，事件循环马上启动任务2
（此时两个任务都在后台"等待"）
（等待约2秒后...）
短信发送成功！ # 任务2的等待时间先到，任务2完成
（继续等待约3秒后...）
水开了！      # 任务1的等待时间也到了，任务1完成

总耗时：5秒（因为两个任务的等待时间是并发的）

通过使用 asyncio，我们可以在单线程中同时处理多个任务。一个在单线程内调度和管理所有协程的核心机制，就是事件循环。它不停地检查哪些协程可以执行，哪些在等待。

总结一下：

• 协程是 asyncio 的核心概念之一。它是一个特殊的函数，可以在执行过程中暂停，并在稍后恢复执行。协程通过 async def 关键字定义，并通过 await 关键字暂停执行，等待异步操作完成。
• 要运行一个协程，可以使用 asyncio.run() 函数。它会创建一个事件循环，并运行指定的协程。事件循环是 asyncio 的核心组件，负责调度和执行协程。它不断地检查是否有任务需要执行，并在任务完成后调用相应的回调函数。

3.4.2.2 astream()使用

可以使用 .astream() 方法，来异步生成流式响应的效果，这专为非阻塞工作流程而设计。可以在异步代码中使用它来实现相同的实时流式处理行为。代码如下：

from langchain_openai import ChatOpenAI

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 异步调用
async def async_stream():
    print("=== 异步调用 ===")
    async for chunk in model.astream("讲一个50字的笑话"):
        print(chunk.content, end="|", flush=True)

import asyncio
asyncio.run(async_stream())

打印结果：

=== 异步调用 ===
|有|一天|，|一|只|鸭|子|走|进|药|店|，|问|："|你|们|有|口|红|吗|？"|药|剂|师|说|："|没|有|。"|鸭|子|失|望|地|摇|摇|头|，|转|身|离|开|。|第|二|天|，|鸭|子|又|来|："|你|们|有|口|红|吗|？"|药|剂|师|说|："|我|说|过|没有|！"|鸭|子|说|："|那|你|们|为什么|不|进|货|呢|？"|

3.4.3 使用StrOutputParser解析模型的输出

还记得最早我们讲过Runnable接口：

Runnable接口：聊天模型、输出解析器等组件，都实现了LangChain的Runnable接口，他们都是Runnable接口的实例。Runnable定义了一个标准接口，允许Runnable组件：

• Invoked（调用）：单个输入转换为输出。
• Batched（批处理）：多个输入被有效地转换为输出。
• Streamed（流式传输）：输出在生成时进行流式传输。
• Inspected（检查）：可以访问有关 Runnable 的输入、输出和配置的原理图信息。
• Composed（组合）：可以组合多个Runnable，以使用LCEL协同工作以创建复杂的管道。

可以看到，流式传输实际上并不算是聊天模型定义的能力，而是只要实现了Runnable接口的实例都具备的能力！！

但要注意，并非所有组件都必须实现流式处理：在某些情况下流式处理要么是不必要的，要么很困难，要么根本没有意义。例如，以后我们会讲解的 Retrievers检索器，就不提供任何流式处理。

那么再得出一个关于流式传输的结论：.stream() 和 .astream() 方法产生的块类型取决于正在流式传输的组件。例如，我们当前正在使用聊天模型的流式传输，返回的每个块都将是一个 AIMessageChunk。但是，对于其他组件，块类型可能不同。

接下来让我们使用LCEL构建一个简单的链，该链结合了模型和解析器，并验证流是否有效。不要忘了使用LCEL创建的链也实现了Runnable接口。

我们将使用 StrOutputParser 来解析模型的输出，它从 AIMessageChunk 中提取内容字段，为我们提供模型返回的令牌。代码如下：

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")
# 定义输出解析器
parser = StrOutputParser()
# 定义链
chain = model | parser

for chunk in chain.stream("写一段关于爱情的歌词，需要5句话"):
    print(chunk, end="|", flush=True)

打印结果：

|在|星空|下|许|下|心愿|，|  
|你的|笑|容|如|晨|光|温暖|，|  
|手|握|手|走|过|每|段|光阴|，|  
|无论|风雨|依|然|不|离|不|弃|，|  
|爱|是|永恒|，|心|与|心|相|连|。|

3.4.4 自定义流式输出解析器

上面我们演示了如何让聊天模型进行流式输出。若此时我们希望修改上一步的输出样式（一个字或两个字的输出），将输出改为一句话一句话的输出，同时保留流式处理功能。那么我们需要在链中使用生成器函数，即可完成自定义流式输出的能力。

还记得之前说过，聊天模型的 .stream() 方法返回的是一个迭代器，该迭代器在生成输出时同步产生输出消息块。那么我们将实现的这些生成器的签名应该是 Iterator[Input] -> Iterator[Output]。或者对于异步生成器：AsyncIterator[Input] -> AsyncIterator[Output]。

下面是句号分隔列表的自定义输出解析器的示例：

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from typing import Iterator, List

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")
# 定义输出解析器
parser = StrOutputParser()

# 定义生成器
def split_into_list(input: Iterator[str]) -> Iterator[List[str]]:
    buffer = ""
    for chunk in input:
        buffer += chunk
        while "。" in buffer:
            # 只要缓冲区中包含句号，就找到第一个句号的位置
            stop_index = buffer.index("。")
            # 将句号之前的内容（去除首尾空格）作为一个句子放入列表中并产出
            yield [buffer[:stop_index].strip()]
            # 更新缓冲区，保留句号之后的内容
            buffer = buffer[stop_index + 1 :]
    yield [buffer.strip()]

# 定义链
chain = model | parser | split_into_list

for chunk in chain.stream("写一份关于爱情的歌词，需要5句话，每句话用句号分割"):
    print(chunk, end="|", flush=True)

打印结果：

['在星空下许下承诺的誓言']|['你的笑容如同晨曦，温暖了我的心']|['无论时光如何流转，我愿与你携手共行']|['爱情是我们心中永恒的旋律']|['每一次相拥，都是一场甜蜜的重逢']|['']|

3.4.5 深度探索流式传输

3.4.5.1 SSE协议介绍

HTTP协议本身设计为无状态的请求-响应模式，严格来说，是无法做到服务器主动推送消息到客户端，但通过 Server-Sent Events（服务器发送事件，简称SSE） 技术可实现流式传输，允许服务器主动向浏览器推送数据流。

也就是说，服务器向客户端声明，接下来要发送的是流消息(streaming)，这时客户端不会关闭连接，会一直等待服务器发送过来的新的数据流。

SSE（Server-Sent Events）是一种基于HTTP的轻量级实时通信协议，浏览器可以通过内置的EventSource API接收并处理这些实时事件。

核心特点：

• 基于HTTP协议：复用标准HTTP/HTTPS协议，无需额外端口或协议，兼容性好且易于部署。
• 单向通信机制：SSE仅支持服务器向客户端的单向数据推送，客户端通过普通HTTP请求建立连接后，服务器可持续发送数据流，但客户端无法通过同一连接向服务器发送数据。
• 自动重连机制：支持断线重连，连接中断时，浏览器会自动尝试重新连接（支持retry字段指定重连间隔）。
• 自定义消息类型：客户端发起请求后，服务器保持连接开放，响应头设置 Content-Type: text/event-stream，标识为事件流格式，持续推送事件流。

数据格式：

服务端向浏览器发送SSE数据，需要设置必要的HTTP头信息：

Content-Type: text/event-stream;charset=utf-8
Connection: keep-alive

每一次发送的消息，由若干个message组成，每个message之间由 \n\n 分隔，每个message内部由若干行组成，每一行都是如下格式：

[field]: value\n

Field可以取值为：

• data[必需]：数据内容
• event[非必需]：表示自定义的事件类型，默认是message事件
• id[非必需]：数据标识符，相当于每一条数据的编号
• retry[非必需]：指定浏览器重新发起连接的时间间隔

除此之外，还可以有冒号 : 开头的行，表示注释。

数据示例：

event: foo\n
data: a foo event\n\n
data: an unnamed event\n\n
event: end\n
data: a bar event\n\n

3.4.5.2 LangChain流式传输流程分析

LangChain本身并不"创造"或"规定"一个底层的网络传输协议，而是依赖于其底层的大模型供应商（如OpenAI）和我们自身服务应用所使用的Web框架（如FastAPI）的协议。

因此对于LangChain的流式传输能力，本身是因为大模型供应商提供了流式传输能力，由LangChain进行调用后接收并处理成一个个的 AIMessageChunk。

（1）通过源码分析流程

接下来我们将会通过分析相关源码探索整个传输流程。整个过程我们以OpenAI举例，其他大模型方式类似，可自行探索。当我们向OpenAI发起流式请求，LangChain实际上会通过 BaseChatOpenAI 类中的 _stream() 方法发起调用。

下面来看下 _stream() 方法的关键流程性源码（class langchain_openai.chat_models.base.BaseChatOpenAI）：

def _stream(
    self,
    messages: list[BaseMessage],                   # 输入消息列表
    stop: Optional[list[str]] = None,              # 可选的停止词列表，用于指定生成终止条件
    run_manager: Optional[CallbackManagerForLLMRun] = None,   # 回调管理器，用于处理生成过程中的回调事件
    *,
    stream_usage: Optional[bool] = None,           # 是否流式返回使用量统计信息
    **kwargs: Any,                                 # 其他可选参数
) -> Iterator[ChatGenerationChunk]:
    # 1. 流式配置
    kwargs["stream"] = True   # 强制启用流式模式
    stream_usage = self._should_stream_usage(stream_usage, **kwargs)

    # 2. 请求构建
    payload = self._get_request_payload(messages, stop=stop, **kwargs)

    # 注意这里定义了AIMessageChunk传输块
    default_chunk_class: type[BaseMessageChunk] = AIMessageChunk
    base_generation_info = {}
    # 3. 发起调用
    # 3.1 指定 response_format 调用
    if "response_format" in payload:
        response_stream = self.root_client.beta.chat.completions.stream(**payload)
        context_manager = response_stream
    # 3.2 普通流式调用
    else:
        response = self.client.create(**payload)
        context_manager = response

    # 4. 响应处理
    try:
        with context_manager as response:
            for chunk in response:
                # 4.1 将 OpenAI 数据块转换为 AIMessageChunk 数据块
                generation_chunk = self._convert_chunk_to_generation_chunk(
                    chunk,
                    default_chunk_class,
                    base_generation_info if is_first_chunk else {},
                )
                if generation_chunk is None:
                    continue

                # 4.2 触发新token回调
                if run_manager:
                    run_manager.on_llm_new_token(
                        generation_chunk.text,
                        chunk=generation_chunk,
                        logprobs=logprobs,
                    )

                # 4.3 产出生成块
                yield generation_chunk

    # 5. 处理OpenAI API错误异常
    except openai.BadRequestError as e:
        _handle_openai_bad_request(e)

从上述流程看来，这就是流式逐块产生 AIMessageChunk 聊天消息的核心方法。那么接下来看三个问题：

1. 发起调用时，底层使用什么协议？
2. 如何支持流式传输？
3. 返回的块是什么格式，如何转换成 AIMessageChunk？

这三个问题都掌握后，整个流式传输的流程就都能理解了。

（2）LangChain请求OpenAI使用什么协议？

回答这个问题，需要看LangChain关于OpenAI的客户端是怎么定义的。让我们找到 class langchain_openai.chat_models._client_utils._SyncHttpxClientWrapper，如下所示：

import openai

class _SyncHttpxClientWrapper(openai.DefaultHttpxClient):
    """Borrowed from openai._base_client"""
    def __del__(self) -> None:
        if self.is_closed:
            return
        try:
            self.close()
        except Exception:  # noqa: S110
            pass

def _build_sync_httpx_client(
    base_url: Optional[str], timeout: Any
) -> _SyncHttpxClientWrapper:
    return _SyncHttpxClientWrapper(
        base_url=base_url
        or os.environ.get("OPENAI_BASE_URL")
        or "https://api.openai.com/v1",
        timeout=timeout,
    )

从上面的代码看来，LangChain使用了OpenAI的官方的OpenAI SDK for Python接入方式，继承了 openai._base_client 定义了一个HTTP客户端。因此在调用时，发起的是 HTTP调用。

（3）LangChain如何支持流式传输？

开始我们就说了，LangChain本身并不"创造"或"规定"一个底层的网络传输协议，而是依赖于其底层的大模型供应商（如OpenAI）的协议。

因此当我们发起请求时，会在请求中设置 stream=True（_stream() 源码中的第一步），表示OpenAI服务器将在生成Response时向客户端发出数据（server-sent events，SSE）。此时API会保持HTTP连接打开，并以特定格式发送数据。

例如我们向原生的GPT模型发起一次设置了 stream=True 的HTTP请求：“你好，我是张三。”。此时，我们会收到来自OpenAI的事件块（简化后的有效负载序列）：

data:{
  "id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 
1717500000, "model": "gpt-4o-mini",
  "choices": [{"index": 0, "delta": {"role": "assistant", "content": "你好"}, 
"finish_reason": null}]
}
data:{
  "id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 
1717500000, "model": "gpt-4o-mini",
  "choices": [{"index": 0, "delta": {"content": "，"}, "finish_reason": null}]
}
data:{
  "id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 
1717500000, "model": "gpt-4o-mini",
  "choices": [{"index": 0, "delta": {"content": "张"}, "finish_reason": null}]
}
data:{
  "id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 
1717500000, "model": "gpt-4o-mini",
  "choices": [{"index": 0, "delta": {"content": "三"}, "finish_reason": null}]
}
data:{
  "id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 
1717500000, "model": "gpt-4o-mini",
  "choices": [{"index": 0, "delta": {"content": "！"}, "finish_reason": null}]
}
data:{
  "id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 
1717500000, "model": "gpt-4o-mini",
  "choices": [{"index": 0, "delta": {"content": "请问"}, "finish_reason": null}]
}
....

看了上述示例，我们应该可以回答第二个问题。那就是在请求中设置 stream=True 开启OpenAI服务端返回数据块！LangChain通过 _stream() 方法步骤1、2完成这件事。

（4）OpenAI返回的块是什么格式，如何转换成 AIMessageChunk？

OpenAI返回的数据块格式我们已经看到了，将其转换为LangChain自定义的 AIMessageChunk 则是通过 _convert_chunk_to_generation_chunk() 方法完成的。关键代码如下：

def _convert_chunk_to_generation_chunk(
    self,
    chunk: dict,                    # 从API获取的原始数据块，在当前例子中就表示 OpenAI 数据块
    default_chunk_class: type,      # 要创建的消息块类型，在当前例子中传入的是 AIMessageChunk
    base_generation_info: Optional[dict],  # 基础生成信息
) -> Optional[ChatGenerationChunk]:

    # 1. 提取选择项数据
    choices = (
        chunk.get("choices", [])
        # from beta.chat.completions.stream
        or chunk.get("chunk", {}).get("choices", [])
    )

    # 2. 处理有效选择项
    choice = choices[0]
    if choice["delta"] is None:
        return None
    # 3. 转换增量数据为消息块，将API的delta格式转换为 AIMessageChunk
    message_chunk = _convert_delta_to_message_chunk(
        choice["delta"], default_chunk_class
    )

    # 4. 构建其他生成信息
    generation_info = {**base_generation_info} if base_generation_info else {}
    ...
    # 5. 返回生成块
    generation_chunk = ChatGenerationChunk(
        message=message_chunk, generation_info=generation_info or None
    )
    return generation_chunk

def _convert_delta_to_message_chunk(
    _dict: Mapping[str, Any], default_class: type[BaseMessageChunk]
) -> BaseMessageChunk:
    # 1. 提取OpenAI格式数据
    id_ = _dict.get("id")
    role = cast(str, _dict.get("role"))               # 获取角色
    content = cast(str, _dict.get("content") or "")   # 获取消息内容
    additional_kwargs: dict = {}
    if _dict.get("function_call"):
        function_call = dict(_dict["function_call"])
        if "name" in function_call and function_call["name"] is None:
            function_call["name"] = ""
        additional_kwargs["function_call"] = function_call
    tool_call_chunks = []
    if raw_tool_calls := _dict.get("tool_calls"):
        additional_kwargs["tool_calls"] = raw_tool_calls
        try:
            tool_call_chunks = [
                tool_call_chunk(
                    name=rtc["function"].get("name"),
                    args=rtc["function"].get("arguments"),
                    id=rtc.get("id"),
                    index=rtc["index"],
                )
                for rtc in raw_tool_calls
            ]
        except KeyError:
            pass
    # 2. 根据角色，构造消息
    if role == "user" or default_class == HumanMessageChunk:
        return HumanMessageChunk(content=content, id=id_)
    elif role == "assistant" or default_class == AIMessageChunk:
        return AIMessageChunk(
            content=content,
            additional_kwargs=additional_kwargs,
            id=id_,
            tool_call_chunks=tool_call_chunks,  # type: ignore[arg-type]
        )
    elif role in ("system", "developer") or default_class == SystemMessageChunk:
        if role == "developer":
            additional_kwargs = {"__openai_role__": "developer"}
        else:
            additional_kwargs = {}
        return SystemMessageChunk(
            content=content, id=id_, additional_kwargs=additional_kwargs
        )
    elif role == "function" or default_class == FunctionMessageChunk:
        return FunctionMessageChunk(content=content, name=_dict["name"], id=id_)
    elif role == "tool" or default_class == ToolMessageChunk:
        return ToolMessageChunk(
            content=content, tool_call_id=_dict["tool_call_id"], id=id_
        )
    elif role or default_class == ChatMessageChunk:
        return ChatMessageChunk(content=content, role=role, id=id_)
    else:
        return default_class(content=content, id=id_)  # type: ignore

到此我们就知道了LangChain流式传输的完整流程与底层协议。总结一下：

1. langchain-openai 包通过集成OpenAI Python SDK，提供了一个HTTP客户端。
2. 因此，支持LangChain向OpenAI的API发起调用请求。
3. 若希望发起流式传输请求，则需在请求中加入 stream=True，向OpenAI说明以SSE协议进行流式返回。
4. LangChain接收OpenAI的SSE格式的响应，并将其转换为LangChain自封装的消息格式，如 AIMessageChunk 消息。这样就可以以统一的方式处理来自不同模型提供商（OpenAI、Anthropic等）的流式响应。

3.5 使用LangSmith跟踪LLM应用

使用LangChain构建的许多应用程序，可能会包含多个步骤和多次的LLM调用。随着这些应用程序变得越来越复杂，作为开发者，我们能够检查链或代理内部到底发生了什么变得至关重要。最好的方法是使用 LangSmith。

LangSmith与框架无关，它可以与langchain和langgraph一起使用，也可以不使用。LangSmith是一个用于帮助我们构建生产级LLM应用程序的平台，它将密切监控和评估我们的应用。

LangSmith平台地址：https://smith.langchain.com/ （新用户需要注册）。

要想让LangSmith跟踪LLM应用，第一步申请LangSmith API Key，点击Settings，就会跳转到"API Keys"设置页面，若没有跳转，可以在左侧tab栏中找到进入。创建完成后，保存好你的API Key。

接下来配置两个环境变量：

LANGSMITH_TRACING="true"
LANGSMITH_API_KEY="你的 LangSmith API Key"

配置完成后，让我们任意执行代码：

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 结构输出对象
class SearchResult(BaseModel):
    """结构化搜索结果。"""
    query: str = Field(description="搜索查询")
    findings: str = Field(description="调查结果摘要")

@tool
def web_search(query: str) -> str:
    """在网上搜索信息。
    Args:
        query: 搜索查询
    """
    return "西安今天多云转小雨，气温18-23度，东南风2级，空气质量良好"

# 手动将工具结果加入消息列表
model_with_search = model.bind_tools([web_search])
messages = [
    HumanMessage("搜索当前最新的西安的天气")
]
ai_msg = model_with_search.invoke(messages)
messages.append(ai_msg)
for tool_call in ai_msg.tool_calls:
    tool_msg = web_search.invoke(tool_call)
    messages.append(tool_msg)

structured_search_model = model_with_search.with_structured_output(SearchResult)
result = structured_search_model.invoke(messages)
print(result)

查看LangSmith平台，这将在LangSmith的默认跟踪项目中生成调用的跟踪。点击最新一次的调用追踪，跟踪会以瀑布流形式展示调用的完整步骤，以及每个步骤的详细信息和耗时，让我们能够检查内部到底发生了什么！！

解释：

• RunnableSequence：可运行序列，就是我们之前讲过的链，即我们将 model_with_search.invoke() 的结果（构造成 ToolMessage），当作入参传递给 structured_search_model.invoke()。
  • 说明：本次调用是RunnableSequence。但不是每次都展示RunnableSequence，根据实际情况而定。
• ChatOpenAI：实际处理的第一步内容，调用聊天模型，生成结果。
• RunnableLambda：实际处理的第二步内容，表示将python可调用对象转换为 Runnable，其实就是将AI生成的结果转换成为结构化对象。

可以看到我们在使用LangSmith时没有代码介入，只需要配置下环境，就可以直接监控我们的应用。

四、核心组件（Components）

前置说明，从这一小节开始，我们将会学习大量的概念、术语，及其具体用法和对应的接口方法。但不用担心，它们本质上都是围绕大模型调用衍生出来的一系列功能扩展，因为大模型也不是随随便便就能被很好地调用起来。因此，为了让整个交互流程更规范、更高效，LangChain定义了一系列的组件使我们与大模型的交互更规范。

因此在学习过程中，有几点要提前强调：

1. 理解每个组件的作用和适用场景是最重要的，不需要死记硬背接口和参数；
2. 因为课程中会逐步教大家如何查阅官方文档，掌握自主查询的方法才是关键；
3. 但前期我们仍会详细讲解常用接口和参数，随着内容深入，将更侧重核心功能与查询思路的培养。重要的内容我们都会覆盖到，大家可以放心。

4.1 消息（Messages）

消息是聊天模型中的通信单位，用于表示聊天模型的输入和输出，以及可能与对话关联的任何其他上下文或元数据。

4.1.1 LLM消息结构

每条消息都有一个角色和内容，以及因LLM的不同而不同的附加元数据。

• 消息角色(Role)：用来区分对话中不同类型的消息，并帮助聊天模型了解如何响应给定的消息序列。

角色	描述
system（系统角色）	用于告诉聊天模型如何行为并提供额外的上下文。并非所有聊天模型提供商都支持。
user（用户角色）	表示用户与模型交互的输入，通常以文本或其他交互式输入的形式。
assistant（助理角色）	表示来自模型的响应，其中可以包括文本或调用工具的请求。
tool（工具角色）	用于在检索外部数据或将工具调用的结果传递回模型的消息。与支持工具调用的聊天模型一起使用。

• 消息内容(Content)：表示多模态数据（例如，图像、音频、视频）的消息文本或字典列表的内容。内容的具体格式可能因底层不同的LLM而异。目前，大多数模型都支持文本作为主要内容类型，对多模态数据的支持仍然有限。

• 消息其他元数据(Additional metadata)

元数据	描述
ID	消息标识符。
Name	名称允许区分具有相同角色的不同实体。并非所有型号都支持此功能！
Metadata	有关消息的其他信息，例如时间戳、令牌使用情况等。
ToolCalls	模型发出的一个或多个工具的调用请求

下面展示一个OpenAI的格式消息列表：

[
    {
        "role": "user",
        "content": "Hello, how are you?",
    },
    {
        "role": "assistant",
        "content": "I'm doing well, thank you for asking.",
    },
    {
        "role": "user",
        "content": "Can you tell me a joke?",
    }
]

LangChain接受下面的格式作为聊天模型的输入：

chat_model.invoke([
    {
        "role": "user",
        "content": "Hello, how are you?",
    },
    {
        "role": "assistant",
        "content": "I'm doing well, thank you for asking.",
    },
    {
        "role": "user",
        "content": "Can you tell me a joke?",
    }
])

4.1.2 LangChain消息

LangChain提供了一种统一的消息格式，可以跨聊天模型使用，允许用户使用不同的聊天模型，而无需担心每个模型提供商使用的消息格式的具体细节。例如：

openai_model = init_chat_model("gpt-4o-mini", model_provider="openai")
anthropic_model = init_chat_model("claude-3-5-sonnet-latest", model_provider="anthropic")
deepseek_model = init_chat_model("deepseek-chat", model_provider="deepseek")
google_genai_model = init_chat_model("gemini-2.5-flash", model_provider="google_genai")

这些模型提供商不同，但对于其输入和输出，统一使用LangChain的消息格式。LangChain消息格式主要分为五种，分别是：

消息类型	对应角色	描述
SystemMessage	对应system系统角色	用于启动AI模型的行为并提供额外的上下文，例如指示模型采用特定角色或设定对话的基调（例如，“你是一个后端开发的专家”）。
HumanMessage	对应user用户角色	人类消息表示用户与模型交互的输入。大多数聊天模型都希望用户输入采用文本形式。
AIMessage	对应assistant助理角色	这是来自模型的响应，其中可以包括文本或调用工具的请求。它还可能包括其他媒体类型，如图像、音频或视频——尽管这目前仍然不常见。
AIMessageChunk	对应assistant助理角色，用于流式响应	通常在生成聊天模型时流式传输响应，因此用户可以实时看到响应，而不是等待生成整个响应后再显示。
ToolMessage	对应tool工具角色	这表示一条角色为"tool"的消息，其中包含调用工具的结果。

这几个消息类型，我们已经全部见过！它们都是LangChain BaseMessage 的子类，全部是作为LangChain聊天模型的输入和输出！！

4.1.2.1 BaseMessage抽象消息类

class langchain_core.messages.base.BaseMessage 是作为LangChain聊天模型的输入和输出！！参数如下：

• content：消息的字符串内容。
• additional_kwargs：与消息关联的其他有效负载数据。对于来自AI的消息，可能包括模型提供程序编码的工具调用。
• response_metadata：响应元数据。例如：响应标头、logprobs、令牌计数、模型名称。
• type：消息的类型。必须是消息类型唯一的字符串。此字段的目的是在对消息进行反序列化时方便地识别消息类型。
• name：消息名称，为消息提供一个人类可读的名称。该字段的使用是可选的，是否使用它取决于模型实现。
• id：消息的可选唯一标识符。理想情况下，这应该由创建消息的提供者/模型提供。

内置方法：

• pretty_print() → None：打印消息的漂亮表示。
• pretty_repr(html: bool = False) → str：获得消息的漂亮表示。
  • 请求：是否将消息格式化为HTML。如果为True，则消息将使用HTML标记进行格式化。默认值为False。
  • 响应：这是消息的漂亮表示。
• text() → str：获取消息的文本内容。

4.1.2.2 对话模式

大多数对话都以设置对话上下文的系统消息开始。接下来是包含用户输入的用户消息，然后是包含模型响应的助手消息。

4.1.3 缓存历史消息

4.1.3.1 多轮对话

在与大型语言模型交互的过程中，我们常常体验到与智能助手进行连贯多轮对话的便利性。但目前我们的系统还不支持此功能，代码如下：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 第一次对话
result = model.invoke([HumanMessage(content="Hi! I'm Bob")])
result.pretty_print()

# 第二次对话
result = model.invoke([HumanMessage(content="What's my name?")])
result.pretty_print()

打印结果：

================================== Ai Message ==================================
Hi Bob! How can I assist you today?
================================== Ai Message ==================================
I'm sorry, but I don't have access to personal information about you unless 
you share it with me. What would you like me to call you?

可以发现，聊天模型并不认识我们，更别说支持更多轮的对话了~

稍作修改，让我们将AI回复给我们的响应跟着新的用户消息一起发给聊天模型试试。代码如下：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 记录消息
messages = [
    HumanMessage(content="Hi! I'm Bob"),
    AIMessage(content="Hello Bob! How can I assist you today?"),
    HumanMessage(content="What's my name?"),
]
model.invoke(messages).pretty_print()

打印结果：

================================== Ai Message ==================================
Your name is Bob! How can I help you today, Bob?

从结果可知，只要将历史消息，重新发送给聊天模型，那么就可以实现多轮对话的功能。

4.1.3.2 内存缓存

那么对于历史消息的管理就显得尤为重要。在LangChain老版本中，可以使用 RunnableWithMessageHistory 消息历史类来包装另一个Runnable并为其管理聊天消息历史记录。它将跟踪模型的输入和输出，并将其存储在某个数据存储中。未来的交互将加载这些消息，并将其作为输入的一部分传递给链。

代码如下：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

store = {}

# 接受一个 session_id 并返回一个消息历史对象。
# 这个 session_id 用于区分不同的对话，并应作为配置的一部分在调用新链时传入
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        # InMemoryChatMessageHistory() 将消息存储在内存列表中。
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 包装model，管理聊天消息历史记录
with_message_history = RunnableWithMessageHistory(model, get_session_history)

config = {"configurable": {"session_id": "1"}}

with_message_history.invoke(
    [HumanMessage(content="Hi! I'm Bob")],
    config=config,
).pretty_print()

with_message_history.invoke(
    [HumanMessage(content="What's my name?")],
    config=config,
).pretty_print()

class RunnableWithMessageHistory 类初始化参数说明：

• runnable：被包装Runnable实例，这里就是我们定义的聊天模型
• get_session_history：返回类型为 BaseChatMessageHistory 的函数，传入后作为回调函数。此函数接受一个 session_id 字符串类型，并返回相应的聊天消息历史记录实例。

类方法说明：

• .invoke() 方法：此方法与其他Runnable实例的 .invoke() 方法相同。只不过注意其config配置，需要配置成 config={"configurable": {"session_id": ""}}，让RunnableWithMessageHistory 可以读取到会话id。

最终打印结果：

================================== Ai Message ==================================
Hi Bob! How can I assist you today?
================================== Ai Message ==================================
Your name is Bob! How can I help you today?

记忆功能已经实现！

说明：从LangChain的v0.3版本开始，官方建议LangChain用户不要使用 RunnableWithMessageHistory，而是利用 LangGraph 持久性 来完成。

原因是它们的功能有限，不太适合现实世界的对话式AI应用程序。这些内存抽象缺乏对多用户、多对话场景的内置支持，而这对于实际的对话式人工智能系统至关重要。这些实现中的大多数已在LangChain 0.3.x中被正式弃用，取而代之的是 LangGraph 持久性。LangGraph 持久性非常灵活，可以支持比 RunnableWithMessageHistory 接口更广泛的用例。

因此，RunnableWithMessageHistory 这部分我们讲解的并不深入。在之前，对于生产环境，我们还需要使用聊天消息历史记录的持久化实现，例如 RedisChatMessageHistory()，而不是 InMemoryChatMessageHistory()，但现在也已不推荐新应用使用它们了。

4.1.4 管理历史消息

4.1.4.1 前置概念

（1）上下文窗口

管理历史消息，无非就是理解如何"管理"，“管理"无非也就是一些"CRUD”。那么在了解如何管理消息之前，需要先了解下多轮对话的核心概念：上下文窗口。上下文窗口可以理解为模型的"短期工作记忆区"，即LLM在一次处理请求时，所能查看和处理的最大Token数量，它包含了：

• 用户的输入
• 大模型的输出
• 有时还包括系统指令（SystemMessage）和对话历史。

不同大模型支持的上下文窗口大小不同，例如：

• OpenAI下GPT-5模型上下文窗口为400000（最大Token数量）
• GPT-4.1模型上下文窗口为1047576（最大Token数量）
• 其他模型上下文窗口可参考对应模型官网说明。

（2）Token

在自然语言处理（NLP）中，Token是文本的基本单位。它不是完全等同于一个单词或一个汉字，而是一个更细粒度的划分。

为什么用Token？计算机无法直接理解文字，它需要将文本转换为数字（向量）。Tokenization（令牌化）就是这个转换过程的第一步，将句子分解成模型可以理解和处理的碎片。

• 对于英文：1个Token ~= 4个字符或0.75个单词，1000个Tokens约等于750个英文单词。一个Token可以是一个单词（如 “apple”）、一个词根（如 “un” 在 “unlikely” 中）、或者一个标点符号（如 “.”）。例如，“ChatGPT is great!” 可能会被分成 ["Chat", "G", "PT", " is", " great", "!"] 这6个Token。
• 对于中文：1个汉字 ~= 1.5-2个Tokens，1000个Tokens大约相当于500-700个汉字。常见的词和字可能是一个Token，生僻字或复杂词可能会被拆分成多个。

举个例子，上下文窗口就像一个固定大小的工作台，再把Token比作一个积木零件，把大模型比作一个工匠。工匠需要拼出模型，必须把所需的零件（输入的Token）放在工作台上，一边拼装（生成回复），一边把拼好的部分（输出的Token）也放在工作台上。整个过程（输入+输出）中，工作台上的所有积木（Tokens）总数都不能超过工作台的最大容量（上下文窗口大小）。

如果最初的零件太多，占满了工作台，工匠就没有空间进行拼装了。这时你就需要减少零件（精简输入）。

4.1.4.2 消息裁剪

有了上下文窗口和Token的认知，再来看多轮对话的实现原理，其实就是：

• 输入 = 系统消息 + 对话历史 + 最新用户问题
• 对于模型来说，并不真正"记忆"，而是每次都将完整的上下文重新输入。

由于所有模型的上下文窗口大小都是有限的，这意味着作为输入的Token也是有限的。如果累积了很长的消息历史记录，则需要管理传递给模型的消息的长度。

trim_messages 可用于将聊天历史记录的大小减小为指定的令牌计数或指定的消息计数。

（1）基于输入Token数的修剪

先来看看不做任何输入限制的聊天：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, trim_messages

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 历史消息记录
messages = [
    SystemMessage(content="you're a good assistant"),
    HumanMessage(content="hi! I'm bob"),
    AIMessage(content="hi!"),
    HumanMessage(content="I like vanilla ice cream"),
    AIMessage(content="nice"),
    HumanMessage(content="whats 2 + 2"),
    AIMessage(content="4"),
    HumanMessage(content="thanks"),
    AIMessage(content="no problem!"),
    HumanMessage(content="having fun?"),
    AIMessage(content="yes!"),
    HumanMessage(content="What's my name?"),
]
print(model.invoke(messages))

打印结果：

content='Your name is Bob.' 
additional_kwargs={'refusal': None} 
response_metadata={
    'token_usage': {
        'completion_tokens': 5, 
        'prompt_tokens': 88, 
        'total_tokens': 93, 
        ...
    }, 
    'model_name': 'gpt-4o-mini-2024-07-18', 
    ...
    'finish_reason': 'stop', 
    ...
} 
usage_metadata={'input_tokens': 88, 'output_tokens': 5, 'total_tokens': 93, ...}

从打印结果看来，LLM还认识我们，且共输入了88 tokens。接下来让我们对消息进行裁剪，我们只希望将来输入时，最多输入65 tokens，超出的需要按照一定的"规则"进行裁剪，代码如下：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, trim_messages

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 历史消息记录
messages = [
    SystemMessage(content="you're a good assistant"),
    HumanMessage(content="hi! I'm bob"),
    AIMessage(content="hi!"),
    HumanMessage(content="I like vanilla ice cream"),
    AIMessage(content="nice"),
    HumanMessage(content="whats 2 + 2"),
    AIMessage(content="4"),
    HumanMessage(content="thanks"),
    AIMessage(content="no problem!"),
    HumanMessage(content="having fun?"),
    AIMessage(content="yes!"),
    HumanMessage(content="What's my name?"),
]

# 使用 trim_messages 减少发送给模型的消息数量
trimmer = trim_messages(
    max_tokens=65,          # 修剪消息的最大令牌数，根据你想要的谈话长度来调整
    strategy="last",        # 修剪策略："last"（默认）：保留最后的消息。"first"：保留最早的消息。
    token_counter=model,    # 传入一个函数或一个语言模型（因为语言模型有消息令牌计数方法）
    include_system=True,    # 如果想始终保留初始系统消息，可以指定 include_system=True
    allow_partial=False,    # 是否允许拆分消息的内容
    start_on="human",       # 如果需要确保我们的第一条消息（不包括系统消息）始终是特定类型，可以指定 start_on
)

chain = trimmer | model
print(chain.invoke(messages))

打印结果：

content='I don't know your name. Would you like to share it with me?' 
...
'token_usage': {'completion_tokens': 16, 'prompt_tokens': 60, 'total_tokens': 76, ...}
...

可以看见，此时我们的输入message已经被修剪了。被修剪了哪些消息呢？来看下：

trimmer = trim_messages(
    ...
)
# 打印经过裁剪器裁剪后的结果
print(trimmer.invoke(messages))

结果如下：

[
    SystemMessage(content="you're a good assistant", additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='whats 2 + 2', additional_kwargs={}, response_metadata={}), 
    AIMessage(content='4', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='thanks', additional_kwargs={}, response_metadata={}), 
    AIMessage(content='no problem!', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='having fun?', additional_kwargs={}, response_metadata={}), 
    AIMessage(content='yes!', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content="What's my name?", additional_kwargs={}, response_metadata={})
]

从结果来看，确实是按照我们给定的裁剪"规则"来完成的。修剪聊天记录后，生成的聊天记录（输入）应该有效，需遵循对话模式原则：

• 聊天记录以 HumanMessage 或 SystemMessage 开头，后跟 HumanMessage。这可以通过设置 start_on="human" 来实现。
• 聊天记录以 HumanMessage 或 ToolMessage 结尾。这可以通过设置 ends_on=("human", "tool") 来实现。
• ToolMessage 只能出现在涉及工具调用的 AIMessage 之后。
• 如果原始聊天历史记录中存在 SystemMessage，则新聊天历史记录应包括SystemMessage，因为 SystemMessage 包含对聊天模型的特殊说明。SystemMessage 总是历史记录中的第一条消息（如果存在）。这可以通过设置 include_system=True。

（2）基于消息数的修剪

除了基于token的修剪，还可以通过设置 token_counter=len 根据消息数修剪聊天记录。在这种情况下，max_tokens 将控制最大消息数。示例如下：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, trim_messages

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 历史消息记录
messages = [
    SystemMessage(content="you're a good assistant"),
    HumanMessage(content="hi! I'm bob"),
    AIMessage(content="hi!"),
    HumanMessage(content="I like vanilla ice cream"),
    AIMessage(content="nice"),
    HumanMessage(content="whats 2 + 2"),
    AIMessage(content="4"),
    HumanMessage(content="thanks"),
    AIMessage(content="no problem!"),
    HumanMessage(content="having fun?"),
    AIMessage(content="yes!"),
    HumanMessage(content="What's my name?"),
]

# 使用 trim_messages 减少发送给模型的消息数量
trimmer = trim_messages(
    max_tokens=11,          # 最大消息数
    strategy="last",        # 修剪策略："last"（默认）：保留最后的消息。可获取消息列表中的最后一个 max_tokens。"first"：保留最早的消息。
    token_counter=len,      # 根据消息数裁剪
    include_system=True,    # 如果想始终保留初始系统消息，可以指定 include_system=True
    allow_partial=False,    # 是否允许拆分消息的内容
    start_on="human",       # 如果需要确保我们的第一条消息（不包括系统消息）始终是特定类型，可以指定 start_on
)
print(trimmer.invoke(messages))

结果如下：

[
    SystemMessage(content="you're a good assistant", additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='I like vanilla ice cream', additional_kwargs={}, response_metadata={}), 
    AIMessage(content='nice', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='whats 2 + 2', additional_kwargs={}, response_metadata={}), 
    AIMessage(content='4', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='thanks', additional_kwargs={}, response_metadata={}), 
    AIMessage(content='no problem!', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='having fun?', additional_kwargs={}, response_metadata={}), 
    AIMessage(content='yes!', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content="What's my name?", additional_kwargs={}, response_metadata={})
]

4.1.4.3 消息过滤

在更复杂的场景下，我们可能会使用消息列表来跟踪状态，例如我们可能只想将这个完整消息列表的子集传递模型调用，而不是所有的历史记录。

filter_messages 方法则可以轻松地按类型、ID或名称过滤message。

下面演示相关过滤示例，首先准备消息列表：

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, filter_messages

# 历史消息记录
messages = [
    SystemMessage("你是一个聊天助手", id="1"),
    HumanMessage("示例输入", id="2"),
    AIMessage("示例输出", id="3"),
    HumanMessage("真实输入", id="4"),
    AIMessage("真实输出", id="5"),
]

• 按类型进行筛选：

print(filter_messages(messages, include_types="human"))
# 注意写法等价于：
# print(filter_messages(include_types="human").invoke(messages))

• 按类型+ID进行筛选：

print(filter_messages(messages, include_types=[HumanMessage, AIMessage], exclude_ids=["3"]))

4.1.4.4 消息合并

若我们的消息列表存在连续某种类型相同的消息，但实际上某些模型不支持传递相同类型的连续消息。因此对于这种情况，我们可以使用 merge_message_runs 方法轻松合并相同类型的连续消息。

示例如下：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, merge_message_runs

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 历史消息记录
messages = [
    SystemMessage("你是一个聊天助手。"),
    SystemMessage("你总是以笑话回应。"),
    HumanMessage("为什么要使用 LangChain?"),
    HumanMessage("为什么要使用 LangGraph?"),
    AIMessage("因为当你试图让你的代码更有条理时，LangGraph 会让你感到"节点"是个好主意！"),
    AIMessage("不过别担心，它不会"分散"你的注意力！"),
    HumanMessage("选择LangChain还是LangGraph?"),
]

merged = merge_message_runs(messages)
# 打印合并后的每个消息
print("\n".join([repr(x) for x in merged]))

合并结果：

SystemMessage(content='你是一个聊天助手。\n你总是以笑话回应。', additional_kwargs={}, response_metadata={})
HumanMessage(content='为什么要使用 LangChain?\n为什么要使用 LangGraph?', additional_kwargs={}, response_metadata={})
AIMessage(content='因为当你试图让你的代码更有条理时，LangGraph 会让你感到"节点"是个好主意！\n不过别担心，它不会"分散"你的注意力！', additional_kwargs={}, response_metadata={})
HumanMessage(content='选择LangChain还是LangGraph?', additional_kwargs={}, response_metadata={})

调用大模型：

## 方式一：
merged = merge_message_runs(messages)
model.invoke(messages).pretty_print()

## 方式二：
merger = merge_message_runs()
chain = merger | model
chain.invoke(messages).pretty_print()

打印结果：

================================== Ai Message ==================================
这就像选择汉堡还是热狗！如果你想要一个多层次的体验，选择LangChain；如果你想要一个清晰的链接，LangGraph就像是你的"支架"！让你的代码加个"料"！

4.2 提示词模板（PromptTemplate）

4.2.1 概念

提示词模板（PromptTemplate）是LangChain的核心抽象之一，它被广泛应用于构建大语言模型（LLM）应用的各个环节。

简单来说，只要是需要动态、批量、或有结构地向大语言模型【发送请求】的地方，几乎都会用到提示词模板。

一个简单的例子，假设我们想根据一个城市名询问LLM其历史，按照之前的做法，我们可以定义 HumanMessage("请介绍上海的历史")、HumanMessage("请介绍西安的历史") 消息等等。可以发现每次询问都会描写重复的消息内容：请介绍xxx的历史。

在LangChain中，针对这种情况，可以定义一个模板：

• 固定文本（模板）：“请介绍{city}的历史。”
• 输入变量：["city"]

定义好后，可以使用该模板：

• 当我们需要查询北京时，就将city变量赋值为"北京"。模板引擎会生成：“请介绍北京的历史。”
• 当我们需要查询上海时，就将city变量赋值为"上海"。模板引擎会生成：“请介绍上海的历史。”

由此可得：提示词模板就是一个可复用的提示词蓝图，它允许我们动态地生成提示词，而不是每次都手动编写完整的提示词。它类似于编程中的字符串格式化功能。你创建一个带有"占位符"的模板，然后在运行时，用具体的值（变量）填充这些占位符，从而生成一个最终发送给LLM的完整提示词。

提示词模板解决了以下几个核心问题：

1. 可复用性：只需定义一个模板，就可以用于无数个类似的查询。
2. 关注点分离：将提示词的结构和逻辑（工程）与具体的内容和数据分离开。提示工程师可以专注于优化模板，而应用程序则负责提供变量值。
3. 一致性：确保发送给LLM的提示词结构统一，这有助于获得更稳定、可预测的输出结果。
4. 可维护性：如果需要修改提示词的风格或结构，只需修改一个模板文件，而不用在代码的无数个地方进行修改。

4.2.2 用法

4.2.2.1 字符串模板

LangChain提供了 PromptTemplate 类来轻松实现这一功能。PromptTemplate 实现了标准的Runnable接口。示例如下：

from langchain_core.prompts import PromptTemplate

# 1. 定义模板
prompt_template = PromptTemplate.from_template("Translate the following into {language}")

# 2. 实例化模板
print(prompt_template.invoke({"language": "Chinese"}))

打印结果：

text='Translate the following into Chinese'

说明：class langchain_core.prompts.prompt.PromptTemplate 类，其参数如下：

• template：提示模板
• input_variables：需要其值作为提示输入的变量的名称列表。

内置方法：

• from_template()：从模板定义提示模板。方法返回了一个 PromptTemplate 实例。

因此除了上面示例中 PromptTemplate.from_template 定义提示模板的方式外，下面这种方法也可以直接初始化模板：

prompt_template = PromptTemplate(
    input_variables=["language"],
    template="Translate the following into {language}",
)

4.2.2.2 聊天消息模板

ChatPromptTemplate 模板：专为LangChain聊天模型设计。可以方便地构建包含 SystemMessage、HumanMessage、AIMessage 的消息模板。如下代码所示：

from langchain_core.prompts import ChatPromptTemplate

# 1. 设置模板
prompt_template = ChatPromptTemplate(
    [
        ("system", "Translate the following into {language}."),
        ("user", "{text}")
    ]
)
# 说明：
# 在 0.2.24 版本后可以直接使用ChatPromptTemplate()来初始化模板
# 在 0.2.24 版本前，需要使用 ChatPromptTemplate.from_messages()来初始化模板

# 2. 实例化模板，获取消息实例
messagesValue = prompt_template.invoke(
    {
        "language": "Chinese",
        "text": "what is your name?"
    }
)
messages = messagesValue.to_messages()
print(messages)

打印结果：

[
    SystemMessage(content='Translate the following into Chinese.', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='what is your name?', additional_kwargs={}, response_metadata={})
]

现在，我们可以将该结果发送给任何一个LLM来获取答案。如下所示：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 1. 设置模板
prompt_template = ChatPromptTemplate(
    [
        ("system", "Translate the following into {language}."),
        ("user", "{text}")
    ]
)

# 2. 实例化模板，获取消息实例
messagesValue = prompt_template.invoke(
    {
        "language": "Chinese",
        "text": "what is your name?"
    }
)
messages = messagesValue.to_messages()
print(messages)

# 3. 输出解析
parser = StrOutputParser()
chain = model | parser
print(chain.invoke(messages))

打印结果：

你的名字是什么？

由于 ChatPromptTemplate 同样也实现了标准的Runnable接口，因此我们还可以通过链来完成调用。如下所示：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 1. 设置模板
prompt_template = ChatPromptTemplate(
    [
        ("system", "Translate the following into {language}."),
        ("user", "{text}")
    ]
)

# 2. 定义输出解析器
parser = StrOutputParser()

# 3. 定义链
chain = prompt_template | model | parser

for token in chain.stream(
        {
            "language": "English", 
            "text": "你好，我叫斯蒂芬，很高兴认识你"
        }
):
    print(token, end="|")

打印结果：

|Hello|,| my| name| is| Stephen|,| nice| to| meet| you|.|

4.2.2.3 消息占位符

在上面的ChatPromptTemplate中，我们看到了如何格式化两条消息，每条消息都是一个字符串。但如果我们希望将消息插入特定位置怎么办？使用 MessagesPlaceholder。

MessagesPlaceholder 负责在特定位置添加消息列表。代码如下：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

prompt_template = ChatPromptTemplate([
    ("system", "你是一个聊天助手"),
    MessagesPlaceholder("msgs")   # 消息占位符
])

messages_to_pass = [
    HumanMessage(content="中国首都是哪里?"),
    AIMessage(content="中国首都是北京。"),
    HumanMessage(content="那法国呢?")
]

formatted_prompt = prompt_template.invoke({"msgs": messages_to_pass})
print(formatted_prompt)

打印结果：

messages=[
    SystemMessage(content='你是一个聊天助手', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='中国首都是哪里?', additional_kwargs={}, response_metadata={}), 
    AIMessage(content='中国首都是北京。', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='那法国呢?', additional_kwargs={}, response_metadata={})
]

在不显式使用 MessagesPlaceholder 类也可以完成该能力：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, AIMessage

prompt_template = ChatPromptTemplate(
    [
        ("system", "You are a helpful assistant"),
        ("placeholder", "{msgs}")
    ]
)

messages_to_pass = [
    HumanMessage(content="中国首都是哪里?"),
    AIMessage(content="中国首都是北京。"),
    HumanMessage(content="那法国呢?")
]

formatted_prompt = prompt_template.invoke({"msgs": messages_to_pass})
print(formatted_prompt)

4.2.3 使用LangChain Hub的提示词模板

LangChain Hub是一个用于上传、浏览、拉取和管理提示词(prompts)的地方。

随着LLM的发展，提示变得越来越重要。LangChain正在打造一个与像GitHub这样的传统平台类似的平台——GitHub长期以来一直是共享和协作代码的首选平台，于是推出了LangChain Hub平台。

LangChain Hub创建一个分享和发现Prompt的平台，使得开发者可以更容易地发现新用例和精炼提示。这一举措使提示工程师更容易合作，重复使用现有的提示，并对其进行微调以实现特定的结果，从而加速对话代理和其他基于语言的应用程序的开发和部署。早期的时候LangChain Hub有Prompt、Chain、Agent，现在只有Prompt。

LangChain Hub官网地址：https://smith.langchain.com/hub/ 。通过登录到Hub来探索所有现有提示。

目前收藏最高的提示词模板是：hardkothari/prompt-maker。我们就以它为示例，演示一下如何使用LangChain Hub上的提示。

Prompt Maker 模板是一个【提示生成器】，它可以自动化优化提示的过程，从而提高语言模型在各种应用中的质量和效果。

要想使用该能力，需要先申请并配置 LangSmith 环境变量：LANGSMITH_API_KEY="你的 LangSmith API Key"。接着，需要从hub拉取相应的提示，并使用，代码如下：

from langchain_openai import ChatOpenAI
from langsmith import Client

# 从 hub 拉取 "hardkothari/prompt-maker" 提示词模板。
client = Client()
prompt = client.pull_prompt("hardkothari/prompt-maker", include_model=True)

# 定义模型
model = ChatOpenAI(model="gpt-4o-mini")

# 定义链
chain = prompt | model

while True:
    task = input("\n你的任务是什么？（输入 quit 退出聊天）\n")
    if task == 'quit':
        break
    lazy_prompt = input("\n你当前的提示是什么？（输入 quit 退出聊天）\n")
    if lazy_prompt == 'quit':
        break
    print("\n Response:")
    chain.invoke({'lazy_prompt': lazy_prompt, 'task': task}).pretty_print()

运行代码：

你的任务是什么？（输入 quit 退出聊天）
写一个快排代码
你当前的提示是什么？（输入 quit 退出聊天）
开发专家，需中文回复
 Response:
================================== Ai Message ==================================
As a coding expert specialized in algorithms and data structures, please write 
a detailed implementation of the Quick Sort algorithm in Python.
### Instructions:
Your code should be well-structured, with clear comments explaining each step 
of the algorithm. Additionally, include an example of how to use the function 
to sort a list of integers and print the sorted result.
### Context:
The implementation should focus on efficiency and clarity, ideally spanning no 
more than 20 lines of code. Ensure that the prompt is in Chinese, and 
structure your response to be easily understandable for readers with a basic 
knowledge of programming.
Example:
```python
def quick_sort(arr):
    # 快速排序函数实现
    ...
    return sorted_arr
# 使用示例
unsorted_list = [34, 7, 23, 32, 5, 62]
print(quick_sort(unsorted_list))

Please provide the complete code and any additional notes that could aid in
understanding the Quick Sort algorithm. Thank you!

通过使用这个模板，可以大大减少手动调整提示所需的工作量，从而节省时间和资源。PromptMaker通过分析初始提示的结构和内容，然后应用一组预定规则或算法来优化提示，以提高响应质量、清晰度和相关性。这在提示的质量对模型的输出有很大影响的场景中特别有用，比如客户服务机器人、对话代理或数据分析任务。

<a id="4-3"></a>
### 4.3 少样本提示（few-shotting）
#### 4.3.1 概念
少样本提示是一种通过向LLM提供少量具体示例或样本，来教会它如何执行某项特定任务的技术。提高模型性能的最有效方法之一是给出一个【模型示例】指导大模型你想做什么、怎么做。下面用一个例子解释少样本提示的作用。

- 参考之前学习的内容，我们可以将其想象为**"零样本提示"**：直接给模型一道考题，不给任何例子。模型会凭借已有的知识直接回答。例如我们输入 `"What is 2 🦜 9?"`，答案可能是：

```python
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini")
model.invoke("What is 2 🦜 9?").pretty_print()

================================== Ai Message ==================================

表达式"2🦜9"不是一个标准的数学运算或方程。它是数字2和鹦鹉🦜的组合，后面跟着数字9。它没有特定的数学意义。

• 少样本提示则是在给出考题前，先给它看几道类似的、附有正确答案的例题。添加示例输入和预期输出的技术给到模型提示，让模型通过例题来理解任务应该怎么做。例如以下样例，模型通过样例后可以发现 🦜 与 ➕ 含义相似，便可以按照此规则得出 What is 2 🦜 9? 的结果是 11：

examples = [
    {"input": "2 🦜 2", "output": "4"},
    {"input": "2 🦜 3", "output": "5"},
]

这能解决什么问题？LLM虽然知识渊博，但有时我们需要它以非常特定的格式、风格或逻辑来回答问题。**提供正确的示例可以减少模型"胡说八道"或犯低级错误的概率，将其输出约束在你提供的范例范围内。**举个例子更能理解：

1. 强制要求模型以特定的格式（如JSON、XML、特定的列表样式）输出结果。样例可以当作格式样板。
2. 有些任务很难用文字指令清晰描述（例如：“请用莎士比亚的风格写作”）。提供几个例子比写长篇大论的指令更有效。
3. 对于需要多步推理的复杂任务，示例可以展示出思考链，引导模型遵循类似的推理路径。

4.3.2 实现少样本提示

实现少样本提示的第一步也是最重要的一步是提出一个好的示例数据集。好的示例应该在运行时相关、清晰、信息丰富，并提供模型尚不知道的信息。如我们给出的例子，就能很好的提示大模型 🦜 与 ➕ 含义相似：

examples = [
    {"input": "2 🦜 2", "output": "4"},
    {"input": "2 🦜 3", "output": "5"},
]

如何让大模型看懂这份示例呢？之前我们说过聊天模型读的是聊天消息。因此，接下来我们需要将示例集实例化成聊天模型可以读懂的聊天消息。对于LangChain就需要创建一个 FewShotChatMessagePromptTemplate 对象来实例化示例集。

FewShotChatMessagePromptTemplate 是一个提示词模板，专门用来将示例集实例化为聊天消息，用法如下所示：

from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate

example_prompt = ChatPromptTemplate(
    [
        ("human", "{input}"),
        ("ai", "{output}"),
    ]
)

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,   # ChatPromptTemplate，用于格式化单个示例
    examples=examples,               # 样本示例
)
print(few_shot_prompt.invoke({}).to_messages())

可以得到示例集转化的聊天消息列表，结果如下：

[
    HumanMessage(content='2 🦜 2', additional_kwargs={}, response_metadata={}), 
    AIMessage(content='4', additional_kwargs={}, response_metadata={}), 
    HumanMessage(content='2 🦜 3', additional_kwargs={}, response_metadata={}), 
    AIMessage(content='5', additional_kwargs={}, response_metadata={})
]

先来看看 class langchain_core.prompts.few_shot.FewShotChatMessagePromptTemplate，它也实现了标准的Runnable接口。

类初始化参数说明：

• examples：样本示例。
• example_prompt：ChatPromptTemplate，用于格式化单个示例

类方法说明：

• .invoke() 方法：此方法与其他Runnable实例的 .invoke() 方法类似。输入一个字典给它，返回完整的提示内容 PromptValue：
  • PromptValue 的 to_string() 方法可以将提示值作为【字符串】返回。
  • PromptValue 的 to_messages() 方法可以将提示作为【消息列表】返回。

最后，得到示例集消息列表后，就可以带上一发起请求：

final_prompt = ChatPromptTemplate(
    [
        ("system", "你是一个神奇的数学奇才。"),
        few_shot_prompt,
        ("human", "{input}"),
    ]
)

from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini")
chain = final_prompt | model
chain.invoke({"input": "What is 2 🦜 9?"}).pretty_print()

结果如下：

================================== Ai Message ==================================
11

计算成功！

4.3.3 使用案例

4.3.3.1 案例一：推理引导

我们希望输入：

《教父》和《星球大战》的导演来自同一个国家吗？

让聊天模型可以先分析再得出结论，而不是直接得出结论。分析过程需要展示出来，示例如下：

from langchain_core.prompts import PromptTemplate

# 创建字符串模板
example_prompt = PromptTemplate.from_template("Question: {question}\n{answer}")

# 创建示例集
examples = [
    {
        "question": "李白和杜甫，谁更长寿？",
        "answer": """
        是否需要后续问题：是的。
        后续问题：李白享年多少岁？
        中间答案：李白享年61岁。
        后续问题：杜甫享年多少岁？
        中间答案：杜甫享年58岁。
        所以最终答案是：李白
        """
    },
    {
        "question": "腾讯的创始人什么时候出生？",
        "answer": """
        是否需要后续问题：是的。
        后续问题：腾讯的创始人是谁？
        中间答案：腾讯由马化腾创立。
        后续问题：马化腾什么时候出生？
        中间答案：马化腾出生于1971年10月29日。
        所以最终答案是：1971年10月29日
        """,
    },
    {
        "question": "孙中山的外祖父是谁？",
        "answer": """
        是否需要后续问题：是的。
        后续问题：孙中山的母亲是谁？
        中间答案：孙中山的母亲是杨太夫人。
        后续问题：杨太夫人的父亲是谁？
        中间答案：杨太夫人的父亲是杨胜辉。
        所以最终答案是：杨胜辉
        """,
    },
    {
        "question": "电影《红高粱》和《霸王别姬》的导演来自同一个国家吗？",
        "answer": """
        是否需要后续问题：是的。
        后续问题：《红高粱》的导演是谁？
        中间答案：《红高粱》的导演是张艺谋。
        后续问题：张艺来自哪里？
        中间答案：中国。
        后续问题：《霸王别姬》的导演是谁？
        中间答案：《霸王别姬》的导演是陈凯歌。
        后续问题：陈凯歌来自哪里？
        中间答案：中国。
        所以最终答案是：是
        """,
    },
]

# 测试：实例化第一个示例
print(example_prompt.invoke(examples[0]).to_string())

结果如下：

Question: 李白和杜甫，谁更长寿？
        是否需要后续问题：是的。
        后续问题：李白享年多少岁？
        中间答案：李白享年61岁。
        后续问题：杜甫享年多少岁？
        中间答案：杜甫享年58岁。
        所以最终答案是：李白

接下来我们需要格式化完整的样本提示。此时可以创建一个 FewShotPromptTemplate 对象来初始化少样本提示模板。其接收少量示例和少量示例的格式化程序。如下所示：

from langchain_core.prompts import FewShotPromptTemplate

prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,  # PromptTemplate，用于格式化单个示例
    suffix="Question: {input}",     # 放在示例之后的提示模板字符串。
    input_variables=["input"],      # 变量的名称列表，这些变量的值需要作为提示词的输入。
)
print(
    prompt.invoke({"input": "《教父》和《星球大战》的导演来自同一个国家吗？"}).to_string()
)

来看下 class langchain_core.prompts.few_shot.FewShotPromptTemplate，它也实现了标准的Runnable接口。

类初始化参数说明：

• examples：样本示例。
• example_prompt：PromptTemplate，用于格式化单个示例
• prefix：放在示例前面的提示模板字符串。
• suffix：放在示例之后的提示模板字符串。
• input_variables：变量的名称列表，这些变量的值需要作为提示词的输入。

类方法说明：

• .invoke() 方法：此方法与其他Runnable实例的 .invoke() 方法类似。输入一个字典给提示符模板，返回完整的提示内容 PromptValue：
  • PromptValue 的 to_string() 方法可以将提示值作为【字符串】返回。
  • PromptValue 的 to_messages() 方法可以将提示作为【消息列表】返回。

接下来让我们调用聊天模型看看结果，完整代码如下：

from langchain_core.prompts import PromptTemplate
from langchain_core.prompts import FewShotPromptTemplate
from langchain_openai import ChatOpenAI

# 创建字符串模板
example_prompt = PromptTemplate.from_template("Question: {question}\n{answer}")

# 创建示例集
examples = [
    # ...同上四个示例，此处省略...
]

prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    suffix="Question: {input}",
    input_variables=["input"],
)

# 生成 [HumanMessage] 消息列表
prompt_messages = prompt.invoke({"input": "《教父》和《星球大战》的导演来自同一个国家吗？"}).to_messages()

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")
model.invoke(prompt_messages).pretty_print()

结果如下：

================================== Ai Message ==================================
是否需要后续问题：是的。  
后续问题：《教父》的导演是谁？  
中间答案：《教父》的导演是弗朗西斯·福特·科波拉。  
后续问题：弗朗西斯·福特·科波拉来自哪里？  
中间答案：弗朗西斯·福特·科波拉来自美国。  
后续问题：《星球大战》的导演是谁？  
中间答案：《星球大战》的导演是乔治·卢卡斯。  
后续问题：乔治·卢卡斯来自哪里？  
中间答案：乔治·卢卡斯也来自美国。  
所以最终答案是：是

通过为模型提供这样的示例，我们可以引导模型做出更好的响应。

4.3.3.2 案例二：使用示例数据增强LangChain信息提取能力

在我们学习【结构化输出】小节时，其中有一个使用场景是使用结构化输出进行信息提取。当时说明可以结合少样本提示来实现。于是在这里，我们来实现一个基于LangChain的结构化信息提取系统，专门从文本中提取人物相关信息。

例如我们希望，对于输入以下文本：

“篮球场上，身高两米的中锋王伟默契地将球传给一米七的后卫挚友李明，完成一记绝杀。这对老友用十年配合弥补了身高的差距。”

代码会提取出结构化数据，如下所示：

people=[
    Person(name="王伟", height_in_meters="2", skin_color=None, hair_color=None),
    Person(name="李明", height_in_meters="1.7", skin_color=None, hair_color=None)
]

实现如下：

第一步：定义结构化返回对象。

from typing import List, Optional
from pydantic import BaseModel, Field

# 定义结构化输出
class Person(BaseModel):
    """一个人的信息。"""
    name: Optional[str] = Field(default=None, description="这个人的名字")
    hair_color: Optional[str] = Field(default=None, description="如果知道这个人头发的颜色")
    skin_color: Optional[str] = Field(default=None, description="如果知道这个人的肤色")
    height_in_meters: Optional[str] = Field(default=None, description="以米为单位的高度")

class Data(BaseModel):
    """提取关于人的数据。"""
    people: List[Person]

第二步：定义两个关键示例，每个示例中包含【文本】和【希望输出】：

1. 无人物场景：文本中没有人物，期望返回空列表
2. 部分信息场景：只有名字，其他字段缺失

# 定义2个示例
examples = [
    (
        "海洋是广阔而蓝色的。它有两万多英尺深。",
        Data(people=[]),  # 没有人物信息的情况
    ),
    (
        "小强从中国远行到美国。",
        Data(people=[
            Person(name="小强", height_in_meters=None, skin_color=None, hair_color=None),
        ]),  # 部分信息缺失的情况
    ),
]

第三步：定义提示词模板。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, SystemMessage

# 定义提示词模板
prompt_template = ChatPromptTemplate(
    [
        SystemMessage(content="你是一个提取信息的专家，只从文本中提取相关信息。"
                              "如果您不知道要提取的属性的值，属性值返回null"),
        MessagesPlaceholder("example_messages"),
        ("user", "{new_message}"),
    ]
)

第四步：构造请求的消息列表。 处理逻辑：

• 遍历每个示例对：文本、期望输出
• 根据 是否有人员信息 生成自然语言响应：“检测到人” or “未检测到人”
• 使用 tool_example_to_messages 转换格式，将每个示例转换为模型可理解的消息格式

from langchain_core.utils.function_calling import tool_example_to_messages

# 样例消息列表
example_messages = []
# 遍历示例对，将每个示例构造消息
for txt, tool_call in examples:
    # 根据提取结果生成AI响应文本
    if tool_call.people:
        ai_response = "检测到人"
    else:
        ai_response = "未检测到人"
    # 将示例转换为模型可理解的消息格式
    example_messages.extend(
        tool_example_to_messages(
            txt, [tool_call], ai_response=ai_response
        ) # 方法返回 list[BaseMessage]
    )

tool_example_to_messages 方法说明：

• 此功能处于测试阶段。它正在积极开发中，因此API可能会发生变化。
• 此方法是一个工具方法，它可以将单个示例转换为聊天模型可以识别的消息列表。
• 参数说明：
  • input：输入字符串
  • tool_calls：list[BaseModel]，表示为Pydantic BaseModel的工具调用列表
  • ai_response：可选的。如果提供，将是最终AIMessage的内容。
• 返回值：list[BaseMessage]消息列表

测试一下：

# 实例化提示词
formatted_prompt = prompt_template.invoke({
    "example_messages": example_messages,
    "new_message": "篮球场上，身高两米的中锋王伟默契地将球传给一米七的后卫挚友李明，完成一记绝杀。"
                   "这对老友用十年配合弥补了身高的差距。"
})
# 打印
for message in formatted_prompt.to_messages():
    message.pretty_print()

消息已经被我们构造成功！可以看到，一个示例经过转换构建出了4条消息：HumanMessage、AiMessage(tool)、ToolMessage、AiMessage。

第五步：初始化模型并绑定结构化输出模式，最后构件链，并调用。

from langchain_openai import ChatOpenAI

# 定义结构化输出模型，自动解析为 Pydantic 对象
model = ChatOpenAI(model="gpt-4o-mini")
structured_model = model.with_structured_output(schema=Data)

chain = prompt_template | structured_model

result = chain.invoke({
    "example_messages": example_messages,
    "new_message": "篮球场上，身高两米的中锋王伟默契地将球传给一米七的后卫挚友李明，完成一记绝杀。"
                   "这对老友用十年配合弥补了身高的差距。"
})
print("\n")
print(result)

结果如下：

people=[
    Person(name='王伟', hair_color=None, skin_color=None, height_in_meters='2.0'), 
    Person(name='李明', hair_color=None, skin_color=None, height_in_meters='1.7')
]

完整代码：

from langchain_openai import ChatOpenAI
from typing import List, Optional
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.utils.function_calling import tool_example_to_messages

# 1. 定义结构化输出
class Person(BaseModel):
    """一个人的信息。"""
    name: Optional[str] = Field(default=None, description="这个人的名字")
    hair_color: Optional[str] = Field(default=None, description="如果知道这个人头发的颜色")
    skin_color: Optional[str] = Field(default=None, description="如果知道这个人的肤色")
    height_in_meters: Optional[str] = Field(default=None, description="以米为单位的高度")

class Data(BaseModel):
    """提取关于人的数据。"""
    people: List[Person]

# 2. 定义2个示例
examples = [
    (
        "海洋是广阔而蓝色的。它有两万多英尺深。",
        Data(people=[]),  # 没有人物信息的情况
    ),
    (
        "小强从中国远行到美国。",
        Data(people=[
            Person(name="小强", height_in_meters=None, skin_color=None, hair_color=None),
        ]),  # 部分信息缺失的情况
    ),
]

# 3. 定义提示词模板
prompt_template = ChatPromptTemplate(
    [
        SystemMessage(content="你是一个提取信息的专家，只从文本中提取相关信息。"
                              "如果您不知道要提取的属性的值，属性值返回null"),
        MessagesPlaceholder("example_messages"),
        ("user", "{new_message}"),
    ]
)

# 4. 获取样例消息列表
example_messages = []
# 遍历示例对，将每个示例构造成聊天消息
for txt, tool_call in examples:
    # 根据提取结果生成AI响应文本
    if tool_call.people:
        ai_response = "检测到人"
    else:
        ai_response = "未检测到人"
    # 将示例转换为模型可理解的消息格式
    example_messages.extend(
        tool_example_to_messages(
            txt, [tool_call], ai_response=ai_response
        ) # 方法返回 list[BaseMessage]
    )

# 5. 调用
# 定义结构化输出模型，自动解析为 Pydantic 对象
model = ChatOpenAI(model="gpt-4o-mini")
structured_model = model.with_structured_output(schema=Data)
# 定义链
chain = prompt_template | structured_model
result = chain.invoke({
    "example_messages": example_messages,
    "new_message": "篮球场上，身高两米的中锋王伟默契地将球传给一米七的后卫挚友李明，完成一记绝杀。"
                   "这对老友用十年配合弥补了身高的差距。"
})
print("\n")
print(result)

总结：该案例展示了LangChain在结构化信息提取中的核心能力：通过Pydantic定义输出模式、少样本学习引导模型行为、提示词模板动态组装上下文，以及链式调用简化流程。最终实现从非结构化文本中精准提取结构化数据。

4.3.4 示例选择器（Example selectors）

4.3.4.1 概念

一旦我们有了示例数据集，就需要考虑提示中应该有多少个示例。关键的权衡是，更多的示例通常会提高性能，但更大的提示会增加成本和延迟。超过某个阈值，太多示例可能会开始混淆模型。

找到正确数量的示例在很大程度上取决于模型、任务、示例的质量以及成本和延迟限制。有趣的是，模型越好，它需要精准的示例就越少。但其实，最佳的方法是使用不同数量的示例进行一些实验。

若此时我们有【大量】的示例数据集。对于大模型来说，就没必要全部使用与参考。我们需要有一种方法可以根据给定的输入，从数据集中选择示例。

在LangChain中，示例选择器就可以帮我们从一组【示例的集合】中根据具体策略选择正确的【示例子集】构建少样本提示。

选择策略有：

• Length：根据特定【长度】内可以容纳的数量选择示例。
• Similarity：使用输入和示例之间的【语义相似性】来决定选择哪些示例。
• MMR：使用输入和示例之间的【最大边际相关性】来决定要选择哪些示例。
• Ngram：使用输入和示例之间的【ngram重叠】来决定要选择哪些示例。

这些其实都是自然语言处理（NLP）里的相似性衡量问题。

4.3.4.2 按长度选择示例（Length）

当我们担心构造提示时，将超过上下文窗口长度，根据特定长度内可以容纳的数量选择示例。对于较长的输入，它将选择更少的示例来包含；而对于较短的输入，它将选择更多示例。

实现按长度选择示例的示例选择器是：class langchain_core.example_selectors.length_based.LengthBasedExampleSelector 类，其参数如下：

• example_prompt：PromptTemplate，用于格式化示例的提示模板。
• examples：模板所需的示例列表。
• max_length：提示的最大长度，超过该长度将剪切示例。
• get_text_length：测量提示长度的方法。默认为字数统计。

内置方法：

• add_example(example: dict[str, str])：将新示例添加到列表中。
  • 输入：一个字典，其中键作为输入变量，值作为其值。
• select_examples(input_variables: dict[str, str]) → list[dict]：根据输入长度选择要使用的示例。
  • 输入：一个字典，其中键作为输入变量，值作为其值。
  • 输出：要包含在提示中的示例列表。

接下来，演示一下如何使用长度示例选择器，让我们：

1. 给一个示例集，输入和输出互为反义词
2. 定义 PromptTemplate 字符串模板，包含输入和输出两个"占位符"
3. 定义 LengthBasedExampleSelector 长度示例选择器，设置初始示例集与最大长度
4. 定义一个 FewShotPromptTemplate 模板对象，用于实例化示例，将示例转化为聊天消息
5. 打印消息结果

代码如下：

from langchain_core.example_selectors import LengthBasedExampleSelector
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate

# 反义词示例集合
examples = [
    {"input": "happy", "output": "sad"},
    {"input": "tall", "output": "short"},
    {"input": "energetic", "output": "lethargic"},
    {"input": "sunny", "output": "gloomy"},
    {"input": "windy", "output": "calm"},
]

# 字符串模板
example_prompt = PromptTemplate(
    input_variables=["input", "output"],
    template="Input: {input}\nOutput: {output}",
)

# 长度示例选择器
example_selector = LengthBasedExampleSelector(
    examples=examples,
    example_prompt=example_prompt,
    # 格式化示例的最大长度。
    # 长度由下面的get_text_length函数测量。
    max_length=25,
    # 用于获取字符串长度的函数，用于确定包含哪些示例。
    # 如果没有指定，它是作为默认值提供的。
    # 该函数返回一个整数，表示字符串中由换行符或空格分隔的"单词"数量
    # get_text_length: Callable[[str], int] = lambda x: len(re.split("\n| ", x))
)

# 用于实例化示例的模板
dynamic_prompt = FewShotPromptTemplate(
    # 提供了一个ExampleSelector而不是examples。
    example_selector=example_selector,
    example_prompt=example_prompt,
    prefix="给出每个输入的反义词",
    suffix="Input: {adjective}\nOutput:",
    input_variables=["adjective"],
)

print(
    dynamic_prompt.invoke({"adjective": "big"}).to_messages()[0].content
)

结果如下：

给出每个输入的反义词
Input: happy
Output: sad
Input: tall
Output: short
Input: energetic
Output: lethargic
Input: sunny
Output: gloomy
Input: windy
Output: calm
Input: big
Output:

可以看到，示例选择器将所有的示例都选择了，因为还没有超过设置的示例最大长度，长度计算是根据字符串中由换行符或空格分隔的"单词"数量得到的（默认）。

接下来让我们加入长输入，验证长度选择器筛选示例。代码如下：

# 一个长的示例输入，所有示例超出最大长度
long_string = "非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常大，比其他任何东西都要大得多"
print(
    dynamic_prompt.invoke({"adjective": long_string}).to_messages()[0].content
)

结果如下：

给出每个输入的反义词
Input: happy
Output: sad
Input: tall
Output: short
Input: 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常 非常大，比其他任何东西都要大得多
Output:

4.3.4.3 按语义相似性选择示例（Similarity）

（1）概念

什么是语义相似？它是衡量文本在【含义上】的接近程度。例如下述两段文本：

text1 = "我喜欢猫"
text2 = "我讨厌狗"

这两段文本表面相似度低，但语义上都是表达对动物的态度。

再例如：

text1 = "苹果很甜"
text2 = "苹果市值创新高"

"苹果"可以指水果或公司，语义相似可以解决一词多义问题，因此这两段文本语义上不相似。

LangChain能根据输入和示例之间的语义相似性来决定选择哪些示例，它通过查找与输入具有最大余弦相似性的嵌入示例来实现这一点。

（2）实现

实现按语义相似性选择示例的示例选择器是：class langchain_core.example_selectors.semantic_similarity.SemanticSimilarityExampleSelector 类。

内置方法：

• from_examples()：根据示例集生成语义相似示例选择器
  • 输入：
    • examples：示例列表
    • embeddings：初始化的嵌入API接口，如 OpenAIEmbeddings()
    • vectorstore_cls：向量存储数据库接口类。
    • k：最终要选择的示例的数量。默认值为4。
  • 输出：语义相似性示例选择器
• add_example(example: dict[str, str])：将新示例添加到列表中。
• select_examples(input_variables: dict[str, str]) → list[dict]：根据输入选择要使用的示例。

接下来，演示一下如何使用语义相似示例选择器，让我们：

1. 给一个示例集，输入和输出互为反义词
2. 定义 PromptTemplate 字符串模板，包含输入和输出两个"占位符"
3. 定义 SemanticSimilarityExampleSelector 语义相似示例选择器，设置初始示例集与嵌入API接口
4. 定义一个 FewShotPromptTemplate 模板对象，用于实例化示例，将示例转化为聊天消息
5. 打印消息结果

注意：运行前先安装 langchain-chroma 库：pip install -U langchain-chroma

from langchain_chroma import Chroma
from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
from langchain_openai import OpenAIEmbeddings

# 反义词示例集合
examples = [
    {"input": "happy", "output": "sad"},
    {"input": "tall", "output": "short"},
    {"input": "energetic", "output": "lethargic"},
    {"input": "sunny", "output": "gloomy"},
    {"input": "windy", "output": "calm"},
]

# 字符串模板
example_prompt = PromptTemplate(
    input_variables=["input", "output"],
    template="Input: {input}\nOutput: {output}",
)

# 语义相似示例选择器
example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples,
    # 嵌入类用于生成用于度量语义相似度的嵌入。
    OpenAIEmbeddings(),
    # VectorStore类，用于存储嵌入并对其进行相似性搜索。
    Chroma,
    # 生成示例的数量。
    k=1,
)

# 用于实例化示例的模板
similar_prompt = FewShotPromptTemplate(
    # 提供了一个ExampleSelector而不是examples。
    example_selector=example_selector,
    example_prompt=example_prompt,
    prefix="给出每个输入的反义词",
    suffix="Input: {adjective}\nOutput:",
    input_variables=["adjective"],
)

print(
    similar_prompt.invoke({"adjective": "worried"}).to_messages()[0].content
)

结果如下：

给出每个输入的反义词
Input: happy
Output: sad
Input: worried
Output:

我们输入的内容和最后选择的示例都表示心情，在语义上最相似！

4.3.4.4 按最大边际相关性选择示例（MMR）

（1）概念

什么是最大边际相关性？它是一种重新排序算法，它使用语义相似性作为基础工具，从一个候选集中挑选出一组既能代表查询主题又彼此多样化的结果。

听起来好像和语义相似性类似，用一个例子看下两者的区别：

• 【语义相似性】就像面试官衡量每个应聘者与职位要求的匹配度。他会给每个应聘者打一个分数。
• 【最大边际相关性】就像团队经理（MMR算法）要组建一个团队。目标是选出一组"精华"结果，而不是一个单一结果：
  • 每个成员都要满足基本职位要求（满足相关性）。
  • 但经理不希望团队里全是只会一种技能的程序员。他需要前端、后端、算法、测试等不同特长的人，以确保团队能力全面、减少冗余（新颖性/多样性）。
  • 经理的策略是：先招一个最匹配的技术大牛（第一步），然后接下来招的人，既要技术达标，又要和已招的人技能互补（迭代过程）。

了解下使用最大边际相关性的场景，更能让我们理解其概念：

• 语义相似性使用场景：搜索引擎的基础排序、重复检测、聚类、语义搜索。
• MMR使用场景：
  • 推荐系统：推荐与用户兴趣相关但又不同类型的物品，避免"信息茧房"。
  • 文档摘要：从长文档中选择能代表主旨又包含不同信息的句子，避免摘要内容重复。
  • RAG(检索增强生成)：在从知识库检索完一堆相关文档后，使用MMR进行去重和多样化筛选，再交给LLM生成答案，能有效提升答案质量和减少幻觉。

（2）实现

了解了相关概念后，LangChain提供了按最大边际相关性选择示例的能力，该示例选择器是：class langchain_core.example_selectors.semantic_similarity.MaxMarginalRelevanceExampleSelector 类，内置方法：

• from_examples()：根据示例集生成MMR示例选择器
  • 输入：
    • examples：示例列表
    • embeddings：初始化的嵌入API接口，如 OpenAIEmbeddings()
    • vectorstore_cls：向量存储数据库接口类。
    • k：最终要选择的示例的数量。默认值为4。
  • 输出：MMR示例选择器
• add_example(example: dict[str, str])：将新示例添加到列表中。
• select_examples(input_variables: dict[str, str]) → list[dict]：根据输入选择要使用的示例。

接下来，演示一下如何使用MMR示例选择器，让我们：

1. 给一个示例集，输入和输出互为反义词
2. 定义 PromptTemplate 字符串模板，包含输入和输出两个"占位符"
3. 定义 MaxMarginalRelevanceExampleSelector MMR示例选择器，设置初始示例集与嵌入API接口
4. 定义一个 FewShotPromptTemplate 模板对象，用于实例化示例，将示例转化为聊天消息
5. 打印消息结果

代码如下：

from langchain_chroma import Chroma
from langchain_core.example_selectors import MaxMarginalRelevanceExampleSelector
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
from langchain_openai import OpenAIEmbeddings

# 反义词示例集合
examples = [
    {"input": "happy", "output": "sad"},
    {"input": "tall", "output": "short"},
    {"input": "energetic", "output": "lethargic"},
    {"input": "sunny", "output": "gloomy"},
    {"input": "windy", "output": "calm"},
]

# 字符串模板
example_prompt = PromptTemplate(
    input_variables=["input", "output"],
    template="Input: {input}\nOutput: {output}",
)

# MMR 示例选择器
example_selector = MaxMarginalRelevanceExampleSelector.from_examples(
    examples,
    OpenAIEmbeddings(),
    Chroma,
    k=2,
)

# 用于实例化示例的模板
similar_prompt = FewShotPromptTemplate(
    # 提供了一个ExampleSelector而不是examples。
    example_selector=example_selector,
    example_prompt=example_prompt,
    prefix="给出每个输入的反义词",
    suffix="Input: {adjective}\nOutput:",
    input_variables=["adjective"],
)

print(
    similar_prompt.invoke({"adjective": "worried"}).to_messages()[0].content
)

结果如下：

给出每个输入的反义词
Input: happy
Output: sad
Input: windy
Output: calm
Input: worried
Output:

4.3.4.5 通过ngram重叠选择示例（Ngram）

（1）概念

什么是【ngram】？ngram指一个文本序列中连续的n个词（word）或字符（character）。

什么是【ngram重叠】？通过计算它们之间共同拥有的ngram数量来一种衡量两段文本相似度的方法。

例如下述两段文本：

text1 = "苹果手机很好用"  (分词后: 苹果 手机 很 好用)
text2 = "这款手机很好用"  (分词后: 这款 手机 很 好用)

这两段文本单词重复度很高，连续三个词的相同的情况也存在，因此ngram重叠高。

再看个例子：

text1 = "苹果手机很好用"  (分词后: 苹果 手机 很 好用)
text2 = "iPhone非常不错"  (分词后: iPhone 非常 不错)

这两段文本在含义上非常相似，但它们的ngram重叠度为0。

因此，传统ngram重叠是一种表面形式的匹配。它只关心词是否完全一样，但对于同义词却无法处理。

什么是【语义ngram重叠】？不再比较词本身，而是比较词背后的语义向量（Embedding）。也就是说，它不是看两个词 [苹果] 和 [iPhone] 的字面是否相同，而是计算它们在语义空间中的向量是否相似。如果相似度超过某个阈值，就认为它们"重叠"了。还是看这个例子：

text1 = "苹果手机很好用"  (分词后: 苹果 手机 很 好用)
text2 = "iPhone非常不错"  (分词后: iPhone 非常 不错)

• 计算 苹果 和 iPhone 的向量相似度 → 得分0.95（很高，视为重叠）
• 计算 手机 和 iPhone 的向量相似度 → 得分0.88（很高，但可能不会同时计分，取决于算法设计，避免重复计算）
• 计算 很 和 非常 的向量相似度 → 得分0.90（很高，视为重叠）
• 计算 好用 和 不错 的向量相似度 → 得分0.82（很高，视为重叠）
• 最终，语义上的unigram重叠度可能为3或4（非常相似！）。

那么语义ngram重叠的使用场景是什么？语义ngram重叠常用于需要更精准语义评估的场景，例如剽窃检测，能够发现那些改换了词汇但保留了核心思想的"智能"剽窃。

（2）实现

LangChain实现按语义ngram重叠选择示例的示例选择器是：class langchain_community.example_selectors.ngram_overlap.NGramOverlapExampleSelector 类，其参数如下：

• example_prompt：PromptTemplate，用于格式化示例的提示模板。
• examples：模板所需的示例列表。
• threshold：算法停止的阈值。默认设置为-1.0。
  • 对于负阈值：按【重叠分数】对示例进行排序，但不排除任何示例。
  • 对于等于0.0的阈值：按【重叠分数】进行排序，并排除与输入没有ngram重叠的示例。
  • 对于大于1.0的阈值：排除所有示例，并返回一个空列表。

内置方法：

• add_example(example: dict[str, str])：将新示例添加到列表中。
• select_examples(input_variables: dict[str, str]) → list[dict]：返回根据输入得到的重叠分数排序的降序示例列表。

接下来，演示一下如何使用ngram重叠示例选择器，让我们：

1. 给一个示例集，设置相关文本
2. 定义 PromptTemplate 字符串模板，包含输入和输出两个"占位符"
3. 定义 NGramOverlapExampleSelector 示例选择器，设置初始示例集与阈值（-1），表示对示例进行排序，但不排除任何示例。
4. 定义一个 FewShotPromptTemplate 模板对象，用于实例化示例，将示例转化为聊天消息
5. 打印消息排序结果

注意：

• 运行前先安装 nltk 库：pip install nltk。它是自然语言处理工具包，在NLP领域中最常用的一个Python库。
• 运行前先安装 langchain_community 库：pip install -U langchain_community

from langchain_community.example_selectors import NGramOverlapExampleSelector
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate

# 翻译示例
examples = [
    {"input": "See Spot run.", "output": "看见Spot跑。"},
    {"input": "My dog barks.", "output": "我的狗叫。"},
    {"input": "Spot can run.", "output": "Spot可以跑。"},
]

# 字符串模板
example_prompt = PromptTemplate(
    input_variables=["input", "output"],
    template="Input: {input}\nOutput: {output}",
)

# NGram 示例选择器
example_selector = NGramOverlapExampleSelector(
    examples=examples,
    example_prompt=example_prompt,
    threshold=-1.0,
)

# 用于实例化示例的模板
dynamic_prompt = FewShotPromptTemplate(
    example_selector=example_selector,
    example_prompt=example_prompt,
    prefix="给出每个输入的中文翻译",
    suffix="Input: {sentence}\nOutput:",
    input_variables=["sentence"],
)

# 按照重叠分数排序
print(
    dynamic_prompt.invoke({"sentence": "Spot can run fast."}).to_messages()[0].content
)

结果如下：

给出每个输入的中文翻译
Input: Spot can run.
Output: Spot可以跑。
Input: See Spot run.
Output: 看见Spot跑。
Input: My dog barks.
Output: 我的狗叫。
Input: Spot can run fast.
Output:

可以看到结果集是按照ngram重叠分数进行了排序。

可以分别设置 threshold=0.0 和 1.0 进行测试，打印结果如下：

threshold=0.0（表示排除不相关的示例）
给出每个输入的中文翻译
Input: Spot can run.
Output: Spot可以跑。
Input: See Spot run.
Output: 看见Spot跑。
Input: Spot can run fast.
Output:

threshold=1.0（排除所有示例）
给出每个输入的中文翻译
Input: Spot can run fast.
Output:

4.4 输出解析器（Output parsers）

4.4.1 概念

负责获取模型的输出，并将输出转换为更结构化的格式。当使用LLM生成结构化数据或规范化聊天模型和LLM的输出时，这很有用。

大型语言模型（LLM）的输出本质上是非结构化的文本。但在构建应用程序时，我们通常希望得到结构化的、机器可读的数据，这样可以将其转换为更适合下游任务的格式，比如：

• JSON对象
• Python字典或列表
• 一个特定的Pydantic模型实例
• 一个简单的布尔值或字符串枚举

输出解析器的作用就是架起这座桥梁：它们将LLM的非结构化文本输出转换为结构化格式。这使得与LLM的交互从"模糊的文本对话"变成了"精确的数据API调用"，是构建可靠、高效LLM应用不可或缺的组件。

与with_structured_output()的区别：

它们都实现了相同的功能，但：

• 维度不同：输出解析器是LangChain中的一个功能性组件，with_structured_output() 是聊天模型的一个方法。
• 用法不同：输出解析器的工作流程可以是链式的，可以将Prompt、LLM和Parser像管道一样连接起来：chain = prompt | llm | parser。而 with_structured_output(schema) 却不行，只能手动调用，返回一个新的、具备了结构化输出能力的Runnable（可运行对象）。可以根据自己的使用场景选择，如果想使用链式就选择输出解析器。

4.4.2 解析文本输出

其实对于使用 StrOutputParser 输出解析器输出文本，我们已经使用过多次了。对于 StrOutputParser，它也实现了标准的Runnable接口。示例如下：

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini")

chain = model | StrOutputParser()

for chunk in chain.stream("写一首夏天的诗词，50字以内。"):
    print(chunk, end="|")

输出如下：

|炎|夏|骄|阳|照|，|绿|树|映|蓝|天|。|  
|蝉|鸣|声|声|烈|，|荷|塘|映|清|鲜|。|  
|微|风|拂|面|过|，|凉|意|透|心|间|。|  
|烦|忧|随|汗|去|，|畅|享|此|夏|欢|。|

若是不使用输出解析器，而是直接得到聊天模型返回的AIMessage，文本内容则需要从消息中的 content 字段获取。

4.4.3 解析结构化对象输出

要输出结构化对象，需要用到的输出解析器是 PydanticOutputParser。

class langchain_core.output_parsers.pydantic.PydanticOutputParser 类，其参数如下：

• pydantic_object：要解析的pydantic模型。

内置方法：

• invoke()：将单个输入转换为输出。
• get_format_instructions() → str：重要！！
  • 作用：生成一个指令字符串，这个字符串会被添加到发送给LLM的提示（Prompt）的末尾。
  • 目的：告诉LLM应该以什么样的格式返回它的响应。例如，“请将你的回复封装在XML标签中"或"请以JSON格式输出，包含’name’和’age’两个字段”。

代码示例如下：

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate
from typing import Optional
from pydantic import BaseModel, Field

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 定义输出结构：Pydantic 类
class Joke(BaseModel):
    """给用户讲一个笑话。"""
    setup: str = Field(description="这个笑话的开头")
    punchline: str = Field(description="这个笑话的妙语")
    rating: Optional[int] = Field(
        default=None, description="从1到10分，给这个笑话评分"
    )

# 设置解析器
parser = PydanticOutputParser(pydantic_object=Joke)

# 提示词模板
prompt = PromptTemplate(
    template="Answer the user query.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    # partial_variables：提示模板携带的部分变量的字典，无需在每次调用提示时都传入它们。
    # 类型为 Mapping[str, Any]，传入template携带的部分变量的字典。
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | model | parser

for chunk in chain.stream({"query": "给我讲一个关于唱歌的笑话"}):
    print(chunk, end="|")

结果如下：

setup='为什么歌手总是带着铅笔去演出？' punchline='' rating=None|
setup='为什么歌手总是带着铅笔去演出？' punchline='因为' rating=None|
setup='为什么歌手总是带着铅笔去演出？' punchline='因为他们' rating=None|
setup='为什么歌手总是带着铅笔去演出？' punchline='因为他们想' rating=None|
setup='为什么歌手总是带着铅笔去演出？' punchline='因为他们想要' rating=None|
setup='为什么歌手总是带着铅笔去演出？' punchline='因为他们想要不断' rating=None|
setup='为什么歌手总是带着铅笔去演出？' punchline='因为他们想要不断调整' rating=None|
setup='为什么歌手总是带着铅笔去演出？' punchline='因为他们想要不断调整音' rating=None|
setup='为什么歌手总是带着铅笔去演出？' punchline='因为他们想要不断调整音调' rating=None|
setup='为什么歌手总是带着铅笔去演出？' punchline='因为他们想要不断调整音调！' rating=None|

4.4.4 解析JSON输出

要输出JSON格式，需要用到的输出解析器是 JsonOutputParser。

class langchain_core.output_parsers.json.JsonOutputParser 类，其参数如下：

• pydantic_object：用于验证的Pydantic对象。如果为空，则不执行任何验证。

内置方法：

• invoke()：将单个输入转换为输出。
• get_format_instructions() → str：重要！！
  • 作用：生成一个指令字符串，这个字符串会被添加到发送给LLM的提示（Prompt）的末尾。
  • 目的：告诉LLM应该以什么样的格式返回它的响应。

代码示例如下，这将提示模型返回JSON：

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 设置解析器
parser = JsonOutputParser()

# 提示词模板
prompt = PromptTemplate(
    template="Answer the user query.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | model | parser
print(chain.invoke({"query": "给我讲一个关于唱歌的笑话"}))

结果如下：

{'joke': '为什么歌手总是带着梯子？\n因为他们想要在音乐会上达到更高的层次！'}

也可以使用带Pydantic情况下的 JsonOutputParser，如下所示：

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from typing import Optional
from pydantic import BaseModel, Field

# 定义大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 定义输出结构：Pydantic 类
class Joke(BaseModel):
    """给用户讲一个笑话。"""
    setup: str = Field(description="这个笑话的开头")
    punchline: str = Field(description="这个笑话的妙语")
    rating: Optional[int] = Field(
        default=None, description="从1到10分，给这个笑话评分"
    )

# 设置解析器
parser = JsonOutputParser(pydantic_object=Joke)

# 提示词模板
prompt = PromptTemplate(
    template="Answer the user query.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | model | parser
print(chain.invoke({"query": "给我讲一个关于唱歌的笑话"}))

结果如下：

{'setup': '为什么歌手从不在森林里唱歌？', 'punchline': '因为他们会被树木的'静'止住！', 'rating': None}

除了上面讲的文本、对象、JSON解析器，其实LangChain官方还提供了更多类型的解析器，如：

• XML解析器：XMLOutputParser
• Yaml解析器：YamlOutputParser
• CSV解析器：CommaSeparatedListOutputParser
• 枚举解析器：EnumOutputParser
• 日期解析器：DatetimeOutputParser

等等。除此之外，LangChain还支持我们自定义输出解析器，以将模型输出结构化为自定义格式。

4.5 文档加载器（Document loaders）

4.5.1 RAG介绍

4.5.1.1 RAG概念

从本小节开始，我们将重点放在RAG阶段（Retrieval-Augmented Generation，检索增强生成）。这是当前大语言模型应用的核心模式。RAG的流程相对复杂，为了更好的理解RAG，我们先用AI搜索来引出RAG。

• 对于【AI大模型】来说，它最擅长的是语义理解和文本总结，最不擅长的就是获取实时的信息。因为大模型的训练数据是有截止日期的！
• 对于【搜索引擎】来说，它最擅长的就是获取实时的信息，缺点是信息分散，每次都需要人为进行总结。
• 大模型与搜索引擎的结合，就是给AI配备了一个活字典，让AI可以随时进行查阅。

一个最简单的AI搜索工作流程：搜索引擎在这里充当知识库，结合我们的查询语句，大模型便可以从知识库中获取相应的查询结果。

有了以上流程的铺垫，接下来，正式进入RAG的学习。

首先，先来思考一个问题：搜索引擎可以帮我们解决实时数据的获取，但获取到的数据也是受限的。它只能获取到公开在网络中的数据，而无法获取到一些本地数据，或企业内部的私有数据等，此时该如何？

答案是使用 RAG（检索增强生成）技术！当用户向LLM提问时，系统首先在知识库（如公司内部文档）中进行语义搜索，找到最相关的内容，然后将这些内容和问题一起交给LLM来生成答案。与AI搜索类比，本质是知识库改变了，从搜索引擎线上搜索改为了本地或私有知识库中搜索。

4.5.1.2 RAG流程

RAG的流程分为【离线数据处理】和【在线检索】两个过程。

上面提到，RAG知识库可以是本地文档、公司内部文档等一些私有化数据。但这些私有数据或文档实际上并不能很好地被直接进行检索访问。因此需要将这些私有化数据构建成可以被检索的知识库，这就是离线数据处理要干的事情。经过离线数据后，知识则会按照某种格式以及排列方式存储在知识库中，等待被使用。

而在线检索则是我们依赖知识库查询，通过大模型生成结果的过程。

流程中的各个步骤：

• 文档加载(Document Loading)：加载多种不同来源加载文档。LangChain提供了100多种不同的文档加载器，包括PDF在内的非结构化的数据、SQL在内的结构化的数据，以及Python、Java之类的代码等。
• 文本分割(Splitting)：文本分割器把Documents切分为指定大小的块。
• 存储(Storage)：存储涉及到两个环节，分别是：
  • 将切分好的文档块进行嵌入（Embedding），即将文档块转换成向量的形式。
  • 将Embedding后的向量数据，存储到向量数据库中。
• 检索(Retrieval)：数据存入向量数据库后。当我们需要进行数据检索时，会通过某种检索算法找到与输入问题相似的文档块。
• 输出(Output)：把问题以及检索出来的文档块一起提交给LLM，LLM会通过问题和检索出来的提示一起来生成更加合理的答案。

我们现在已经知道了RAG的完整流程，但也仅是知道RAG是什么，至于流程中为什么要进行文档加载、文本分割、存储，我们还无从了解。因此后续的各个LangChain组件讲解时都会涉及每个步骤被设计出来的原因。

4.5.1.3 RAG示例

讲这么多，不如一见。接下来就来演示一个RAG系统。在这个示例中，我们

• 提供了：《租房项目Q&A》文档
• 希望：通过聊天方式，提问关于项目的任何问题，最终得到答案。
• 要求：最多只用三句话回答，要简明扼要。

代码如下（不用理解代码含义，只需要看结果即可）：

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_redis import RedisConfig, RedisVectorStore
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# 定义聊天模型
model = ChatOpenAI(model="gpt-4o-mini")
# 定义嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# 配置 Redis 客户端
redis_url = "redis://192.168.100.238:6379"
config = RedisConfig(
    index_name="qa",
    redis_url=redis_url,
    metadata_schema=[
        {"name": "category", "type": "tag"},
        {"name": "num", "type": "numeric"},
    ],
)
# 定义 Redis 向量存储
vector_store = RedisVectorStore(embeddings, config=config)
# 生成检索器
retriever = vector_store.as_retriever()
# 定义提示词模板
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "human",
            """你是负责回答问题的助手。使用以下检索到的上下文片段来回答问题。如果你不知道答案，就说你不知道。最多只用三句话，回答要简明扼要。
Question: {question}
Context: {context}
Answer:""",
        ),
    ]
)

# 将文档转换为字符串
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 定义链
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 循环输入问题
while True:
    # 获取用户输入
    question = input("\n请输入您的问题（输入'退出'或'quit'结束程序）: ").strip()
    # 检查是否退出
    if question.lower() in ["退出", "quit"]:
        print("程序已结束，再见！")
        break
    # 检查输入是否为空
    if not question:
        print("问题不能为空，请重新输入。")
        continue
    # 执行链，流式输出
    print("回答: ", end="", flush=True)
    chunks = []
    for chunk in rag_chain.stream(question):
        chunks.append(chunk)
        print(chunk, end="", flush=True)
    print()  # 换行

运行代码：

请输入您的问题（输入'退出'或'quit'结束程序）: 介绍一下这个项目
回答: 这个项目是一个基于脚手架的微服务在线租房系统，旨在模仿贝壳、安居客等流行应用，具备真实的交互体验和架构设计。通过结合理论和实践，我在这个项目中加深了对 Java 编程语言的理解。这个项目的灵感来源于我在大学时和同学合租的经历。
请输入您的问题（输入'退出'或'quit'结束程序）: 项目设计难点有哪些？
回答: 项目设计的难点主要包括非技术方面的需求把控和技术方面的明确项目范围与目标。缺乏产品经理的介入使得设计过程中需自行绘制原型图并拆分功能，这增加了设计的复杂性。同时，测试和质量保证也是项目中的重要挑战，需要合理安排时间以确保项目质量。
请输入您的问题（输入'退出'或'quit'结束程序）: 介绍数据存储的相关设计
回答: 数据存储的设计中，使用MySQL作为关系数据库管理系统，并结合MyBatis简化SQL操作。为了更好地应对高并发场景，设计了Redis缓存方案来优化内存利用和减轻数据库压力。引入OSS对象存储是为了实现无限扩展性，相对于本地存储和MySQL，它具备更高的性能和成本效益。
请输入您的问题（输入'退出'或'quit'结束程序）: 详细介绍 Redis 与 Mysql 数据一致性方案
回答: Redis与MySQL的数据一致性方案通常采用"双写一致性"模式，通过Cache-Aside方法实现。在这一方案中，应用程序在读取数据时先查询Redis缓存，如果不存在再查询MySQL数据库，并将结果缓存到Redis中；在更新数据时，需同时更新MySQL和Redis，以确保两者状态一致。此方法也包括设置合理的缓存过期时间和使用布隆过滤器，以解决缓存穿透等问题。

我们提供的文档越详细，生成的结果越符合预期。

4.5.2 Document文档类

要想实现RAG，首先就需要从源中获取数据，即加载数据或文档。这是通过LangChain的文档加载器完成的。LangChain文档加载器可以将各种数据源加载成一系列的文档对象 Document。

class langchain_core.documents.base.Document 用于存储一段文本和相关元数据的类，我们可以直接定义LangChain文档列表，如下所示：

from langchain_core.documents import Document

documents = [
    # 单个Document对象通常表示较大文档的一个块
    Document(
        # 内容字符串
        page_content="狗是很好的伴侣，以忠诚和友好而闻名。",
        # 元数据字典
        # 元数据属性可以捕获有关文档源、与其他文档的关系以及其他信息的信息。
        metadata={"source": "mammal-pets-doc"},
    ),
    Document(
        page_content="猫是独立的宠物，经常享受自己的空间。",
        metadata={"source": "mammal-pets-doc"},
    ),
]

这里我们定义了一个 documents 文档列表，其内包含了两个 Document 文档对象。通常，单个Document对象表示较大文档的一个块/页。每个 Document 对象，包含了以下参数：

• id：可选的文档标识符。理想情况下，这应该在整个文档集合中是唯一的，并格式化为UUID，但不会强制执行。
• page_content：字符串文本
• metadata：与内容关联的任意元数据。类型为 dict [Optional]

4.5.3 加载PDF文档

将本地的PDF文档加载到LangChain中，其实就是将PDF文档转换为一个个 Document 对象。这时就需要我们使用 PyPDFLoader 文档加载器完成这一功能。

class langchain_community.document_loaders.pdf.PyPDFLoader 类，有以下关键函数：

• init() 初始化函数，入参 file_path，表示要加载的PDF文件的路径。
• load() → list[Document]：将数据加载到文档对象中。返回文档对象列表。

现在，让我们加载一个本地PDF文档看下效果。

from langchain_community.document_loaders import PyPDFLoader

file_path = "../Docs/PDF/脚手架级微服务租房平台Q&A.pdf"
loader = PyPDFLoader(file_path)
# 将 PDF 文件的每一页转换为一个独立的 Document 对象，并存储在列表 docs 中。
docs = loader.load()

接下来，来看看 Document 重要的一些参数：

print(f"问：PDF 文件的总页数为：\n{len(docs)}\n")
print(f"问：第一页文本内容的前200个字符是：\n{docs[0].page_content[:200]}\n")
print(f"问：第一页元数据：\n{docs[0].metadata}")

打印结果如下：

问：PDF 文件的总页数为：
 32 
问：第一页文本内容的前200个字符是：
 脚手架级微服务租房平台
通用问题
1. 为什么做这个项目？
• 回答1：（出于兴趣爱好开发）
大学期间，我和同学在外合租过一段时间，使用了一些租房平台，于是我有个想法，自己能不能开发一个租房平台，可以让我将理论知识与实践相结合。我希望通过实际项目来加深对Java编程语言和相关技术的理解。于是我便查找了一些资料，看了一些开源项目，进行了一些改进。
• 回答2：（开源项目的解释）
这个
问：第一页元数据：
 {'producer': 'pdfcpu v0.8.1 dev', 'creator': 'Chromium', 'creationdate': 
'2025-08-28T17:52:34+08:00', 'moddate': '2025-08-28T17:52:34+08:00', 'source': 
'../PDF/脚手架级微服务租房平台Q&A.pdf', 'total_pages': 32, 'page': 0, 
'page_label': '1'}

现在许多LLM支持对多模态输入（例如图像）进行推理。在某些应用程序中，例如对具有复杂布局、图表或扫描的PDF进行问答，可以跳过PDF解析，直接将PDF页面转换为图像并将其直接传递给模型可能是更准确的。

4.5.4 加载Markdown文件

将本地的 Markdown 文档加载到LangChain中，需要我们使用 UnstructuredMarkdownLoader 文档加载器完成这一功能。

class langchain_community.document_loaders.markdown.UnstructuredMarkdownLoader 类，有以下关键函数：

• init() 初始化函数，所需参数：
  • file_path：表示要加载的 Markdown 文件的路径。
  • mode：加载文件时要使用的模式。可以是single或elements。默认为single。
    • single：文档将作为单个 Document 对象返回
    • elements：会将文档拆分为Title和NarrativeText等不同类型的元素。
• load() → list[Document]：将数据加载到文档对象中。返回文档对象列表。

LangChain实现的 UnstructuredMarkdownLoader 需要依赖Unstructured包。因此在使用前我们需要先安装它：

pip install "unstructured[md]" nltk

现在，让我们使用 single 模式加载一个本地 Markdown 文档看下效果。

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_core.documents import Document

markdown_path = "../Docs/Markdown/脚手架级微服务租房平台Q&A.md"
# single 模式，加载后，默认只有一个 Document 对象
loader = UnstructuredMarkdownLoader(markdown_path)
data = loader.load()

assert len(data) == 1
assert isinstance(data[0], Document)
print(data[0].page_content[:200])
print(data[0].metadata)

打印结果如下：

通用问题
为什么做这个项目？
回答1
{'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}

接下来再看下 elements 模式下加载本地 Markdown 文档的效果：

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_core.documents import Document

markdown_path = "../Docs/Markdown/脚手架级微服务租房平台Q&A.md"
loader = UnstructuredMarkdownLoader(markdown_path, mode="elements")
data = loader.load()

print(f"问：文档个数为：\n{len(data)}\n")
print("问：前三个文档数据：")
for document in data[:3]:
    print(f"{document}\n")

打印结果如下：

问：文档个数为：
441
问：前三个文档数据：
page_content='通用问题' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md', 'category_depth': 0, 'languages': ['zho'], 'file_directory': '../Docs/Markdown', 'filename': '脚手架级微服务租房平台Q&A.md', 'filetype': 'text/markdown', 'last_modified': '2025-08-29T10:56:36', 'category': 'Title', 'element_id': '3a0670f9bfd58576e430ef11def41593'}
page_content='为什么做这个项目？' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md', 'category_depth': 2, 'emphasized_text_contents': ['为什么做这个项目？'], 'emphasized_text_tags': ['b'], 'languages': ['zho'], 'file_directory': '../Docs/Markdown', 'filename': '脚手架级微服务租房平台Q&A.md', 'filetype': 'text/markdown', 'last_modified': '2025-08-29T10:56:36', 'parent_id': '3a0670f9bfd58576e430ef11def41593', 'category': 'Title', 'element_id': 'fcb08b2a85942455eecebb9467ffca4c'}
page_content='回答1：（出于兴趣爱好开发）' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md', 'emphasized_text_contents': ['回答1：（出于兴趣爱好开发）'], 'emphasized_text_tags': ['b'], 'languages': ['zho'], 'file_directory': '../Docs/Markdown', 'filename': '脚手架级微服务租房平台Q&A.md', 'filetype': 'text/markdown', 'last_modified': '2025-08-29T10:56:36', 'parent_id': 'fcb08b2a85942455eecebb9467ffca4c', 'category': 'UncategorizedText', 'element_id': 'a6fc0b5a457d21234bf1c4a6ae0a18db'}

可以看见文档被分成了441个。根据什么规则拆分呢？答案是根据类型拆分。

在元数据中，有一个表示类型的字段：category。这些类型都是现代文档解析库（如Unstructured.io）中用于分类Markdown（MD）文档或从其他格式（如PDF, Word, HTML）解析后转换为Markdown的常见元素类型。

让我们把当前文档包含的所有类型拿出来看看：

print(set(document.metadata["category"] for document in data))

结果如下：

{'Image', 'Title', 'ListItem', 'Table', 'NarrativeText', 'UncategorizedText'}

含义如下：

• Image：图像。使用 ![alt text](image-url) 语法插入的图片。
• Title：标题。这里包含了一级、二级、三级等标题。
• ListItem：列表项。以 -、*、+ 开头的无序列表项，或以 1.、2. 等开头的有序列表项。
• Table：表格。使用 | 和 - 语法创建的表格。
• NarrativeText：叙述性文本。一个或多个连续的段落。
• UncategorizedText：未分类文本。通常是：表格中的脚注或注释、图片下方的简短说明、项目符号中非常简短的词组、文档页眉/页脚中的日期或页码等元数据。

这里的标题都是用Title表示，那么如何区分到底是几级标题呢？

首先，每个文档对象的元数据中都包含一个 element_id 用来识别其唯一性，除此之外，还有一个 parent_id 元素，用来表示其归属于那个文档对象下，可以用来表示层级关系。如示例中的前三个文档对象，就有明显的层级关系：

page_content='通用问题' metadata={
    'category': 'Title', 
    'element_id': '3a0670f9bfd58576e430ef11def41593'
}
page_content='为什么做这个项目？' metadata={
    'parent_id': '3a0670f9bfd58576e430ef11def41593', 
    'category': 'Title', 
    'element_id': 'fcb08b2a85942455eecebb9467ffca4c'
}
page_content='回答1：（出于兴趣爱好开发）' metadata={
    'parent_id': 'fcb08b2a85942455eecebb9467ffca4c', 
    'category': 'UncategorizedText', 
    'element_id': 'a6fc0b5a457d21234bf1c4a6ae0a18db'
}

经过这种设计，就可以还原出整个MD文档的内容了。

对于LangChain来说，能加载的文档类型远不止这些，它还能加载网页、一些云提供商文件、社交媒体平台文档等。

4.6 文本分割器（Text splitters）

4.6.1 概念

我们已经知道可以通过文档加载器完成各种数据源的加载，将其转换为文档对象 Document。那么接下来要做的就是文档拆分。

文档拆分通常是将大文本分解为更小的、易于管理的块。这对于索引数据并将其传递到模型中都很有用。因为，大块更难搜索并且不适合模型的有限上下文窗口。拆分可以提高搜索结果的粒度，从而可以更精确地将查询与相关文档部分进行匹配。

LangChain的文本分割器便能将大型文档分解为更小的块。

4.6.2 根据文档长度与文档语义拆分

我们可以直接根据文档的长度拆分文档，是最简单且有效的方法。可确保每个块不超过指定的大小限制。对于长度拆分，其实也分为两种：基于字符长度拆分和基于Token长度拆分。

4.6.2.1 基于字符长度拆分

根据给定的字符序列进行拆分，拆分的块长度则按字符数来衡量。

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import CharacterTextSplitter

markdown_path = "../Docs/Markdown/脚手架级微服务租房平台Q&A.md"
# single 模式加载后，默认只有一个 Document 对象
loader = UnstructuredMarkdownLoader(markdown_path)
data = loader.load()

# 文本分割器
text_splitter = CharacterTextSplitter(
    separator="\n\n",          # 选择分隔符：它有一个默认的分隔符优先级列表，通常是：["\n\n", "\n", " ", ""]。它会按顺序尝试这些分隔符
    chunk_size=100,            # 设定目标：目标块大小
    chunk_overlap=20,          # 设定目标：块之间的重叠大小
    length_function=len,       # 使用测量长度的函数
    is_separator_regex=False,  # 分隔符是正则表达式吗
)

# 分割文档，返回被分割的文档列表
texts = text_splitter.split_documents(data)

# 打印前10个被分割出来的文档
for document in texts[:10]:
    print("*" * 30)
    print(f"{document}\n")

打印结果：

Created a chunk of size 128, which is longer than the specified 100
Created a chunk of size 133, which is longer than the specified 100
Created a chunk of size 108, which is longer than the specified 100
Created a chunk of size 104, which is longer than the specified 100
...（大量类似日志）
******************************
page_content='通用问题
为什么做这个项目？
回答1：（出于兴趣爱好开发）' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
******************************
page_content='大学期间，我和同学在外合租过一段时间，使用了一些租房平台，于是我有个想法，自己能不能开发一个租房平台，可以让我将理论知识与实践相结合。我希望通过实际项目来加深对 Java 编程语言和相关技术的理解。于是我便查找了一些资料，看了一些开源项目，进行了一些改进。' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
******************************
page_content='回答2：（开源项目的解释）' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
...（后续分割块）

Created a chunk of size 128, which is longer than the specified 100，这表示分割时，超出了我们设定的 chunk_size=100 目标块大小。在这里要说明，这是一个在使用LangChain的文本分割器时非常常见的问题。看到这个信息，不要担心，这不是错误，而是预期的行为。

原因是为了保持语义的完整性！当文本分割器用尽所有指定的分隔符都无法将一段文本分割到你的目标大小 chunk_size 以下时，它会选择保留整个文本块，而不是强行将其截断为无意义的片段，因此我们会看到这个提示信息。因此我们可以看到，被分割出来的段落，基本上都是语义完整的一段话。

那么分割逻辑到底是什么，可以支持保持语义完整性？

• 尝试分割：
  • 首先，它尝试用 separator（我们设置的是 "\n\n" 双换行，这通常代表段落之间）来分割文本。如果分割后的任何一个段落仍然大于 chunk_size，它会继续下一步。
• 如果仍然有单个单词或字符串的长度超过了100，分割器就陷入了两难境地：
  • 选项A：强行把这个长字符串在任意位置（比如第100个字符处）截断。但这会破坏单词、URL或数据的完整性，导致生成无意义的片段（例如，把"Christopher"截断成"Christop"和"her"），严重影响后续的嵌入或语言模型处理效果。
  • 选项B：保留这个完整的、超长的字符串作为一个块，并记录一条信息告知用户。

那么很明显，这里它选择了B选项。因此，我们看到了 Created a chunk of size 128, which is longer than the specified 100 这条日志信息。

如何应对和处理？增大 chunk_size：如果我们的大部分块都超长，可能是 chunk_size 设置得太小了。尝试适当增大它。

因此，根据打印的日志，可以发现，被拆分的文本大都徘徊的100~200之间，因此，可以将 chunk_size 设置为200：

text_splitter = CharacterTextSplitter(
    separator="\n\n",
    chunk_size=200,
    chunk_overlap=20,
    length_function=len,
    is_separator_regex=False,
)

运行后，可以看到超出的分割已经很少了。

4.6.2.2 基于Token长度拆分

之前我们讲过，LLM大模型实际上并不是直接接收字符串，而是需要先做token切分编码。这里我们可以借助【tiktoken分词器】来进行token的切分编码。

先看个例子：给定一个文本字符串 "my name is LiHua!"，使用tiktoken分词器进行切分编码，会得到什么结果：

import tiktoken

# 定义cl100k_base编码方式的分词器
enc = tiktoken.get_encoding("cl100k_base")
# 进行切分编码
enc_output = enc.encode("my name is LiHua!")
# 打印结果
print(f"编码后的token：{str(enc_output)}")
for token in enc_output:
    print(f"将token: {str(token)} 变成文本: {str(enc.decode_single_token_bytes(token))}")

解释：cl100k_base 是 tiktoken 分词器中的一种编码方式。gpt-4、gpt-3.5-turbo 等都采用这种切分编码方式。

可以看到采用切分编码 cl100k_base，拆解后的文本字符串为 ["my", "name", "is", " Li", "H", "ua", "!"]。token编码表示为 [2465, 836, 374, 14851, 39, 4381, 0]。

以上跟大家介绍了基于Token拆分文本的基本方式，主要就是了解设定了某种编码格式的tiktoken分词器可以进行文本拆分。那么接下来，我们就可以使用根据 cl100k_base 编码方式的 tiktoken 分词器来拆分文档。这对于OpenAI模型来说，会更准确。

在LangChain中，我们可以使用 CharacterTextSplitter 分割器的 .from_tiktoken_encoder() 方法来定义根据 tiktoken 分词器拆分文本的分割器，代码如下所示：

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import CharacterTextSplitter

markdown_path = "../Docs/Markdown/脚手架级微服务租房平台Q&A.md"
# single 模式加载后，默认只有一个 Document 对象
loader = UnstructuredMarkdownLoader(markdown_path)
data = loader.load()

# 生成分割器
text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
    encoding_name="cl100k_base", chunk_size=200, chunk_overlap=50
)

# 分割文档
texts = text_splitter.split_documents(data)

# 打印前10个被分割出来的文档
for document in texts[:10]:
    print("*" * 30)
    print(f"{document}\n")

结果如下：

Created a chunk of size 916, which is longer than the specified 200
Created a chunk of size 916, which is longer than the specified 200
Created a chunk of size 260, which is longer than the specified 200
******************************
page_content='通用问题
为什么做这个项目？
回答1：（出于兴趣爱好开发）
大学期间，我和同学在外合租过一段时间，使用了一些租房平台，于是我有个想法，自己能不能开发一个租房平台，可以让我将理论知识与实践相结合。我希望通过实际项目来加深对 Java 编程语言和相关技术的理解。于是我便查找了一些资料，看了一些开源项目，进行了一些改进。
回答2：（开源项目的解释）' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
******************************
page_content='回答2：（开源项目的解释）
这个项目其实是在 github 上找到的一个开源项目，主要是可以支持一些常规的聊天项目，我对于聊天如何实现的比较感兴趣。顺便也想锻炼一下自己工程代码能力，在网上就找到了这个开源项目和项目的一些比较完善的文档和介绍，再加上找了一个业务场景：租房。所以就确定了这个项目。
回答3：（学校课设的解释）' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
...（后续分割块）

同样，为了保持语义的完整性，文本分割器无法将一段文本分割到你的目标大小 chunk_size 以下，它会选择保留整个文本块。

4.6.2.3 硬性约束长度拆分

如果我们就想要求任何块都不能超过指定大小，可以使用 RecursiveCharacterTextSplitter 类或 RecursiveCharacterTextSplitter.from_tiktoken_encoder 方法，它会严格遵守对块大小的硬约束。下面展示一下用法：

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

markdown_path = "../Docs/Markdown/脚手架级微服务租房平台Q&A.md"
# single 模式加载后，默认只有一个 Document 对象
loader = UnstructuredMarkdownLoader(markdown_path)
data = loader.load()

# 生成分割器
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    encoding_name="cl100k_base",
    chunk_size=100,
    chunk_overlap=0,
)

# 分割文档
texts = text_splitter.split_documents(data)

# 打印前10个被分割出来的文档
for document in texts[:10]:
    print("*" * 30)
    print(f"{document}\n")

结果如下，严格不超出指定块大小：

******************************
page_content='通用问题
为什么做这个项目？
回答1：（出于兴趣爱好开发）' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
******************************
page_content='大学期间，我和同学在外合租过一段时间，使用了一些租房平台，于是我有个想法，自己能不能开发一个租房平台，可以让我将理论知识与实践相结合。我希望通过实际项目来加深对 Java' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
******************************
page_content='编程语言和相关技术的理解。于是我便查找了一些资料，看了一些开源项目，进行了一些改进。' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
******************************
page_content='回答2：（开源项目的解释）' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
******************************
page_content='这个项目其实是在 github' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
******************************
page_content='上找到的一个开源项目，主要是可以支持一些常规的聊天项目，我对于聊天如何实现的比较感兴趣。顺便也想锻炼一下自己工程代码能力，在网上就找到了这个开源项目和项目的一些比较完善的' metadata={'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}
...（后续分割块）

但这样，其实是剥夺了一些保证语义完整性的能力，可以看到某些含义相似的内容，被强制分开。

除此之外，还要再说明的是，文本内容是中文时，使用默认分隔符列表 ["\n\n", "\n", " ", ""] 拆分文本可能会导致一个词组被拆分成两个字，导致语义失效。若要将词组放在一起，可以覆盖分隔符列表以包含其他标点符号，例如中文的逗号 ，、句号 。 或其他中文符号，如下所示：

text_splitter = RecursiveCharacterTextSplitter(
    separators=[
        "\n\n",
        "\n",
        " ",
        "。",
        ",",
        "",
    ],
    # Existing args
)

这样在分割时将递归用 separators 来尝试分割文本：

• 首先，它尝试用 "\n\n"（双换行，通常代表段落）来分割，如果分割后的任何一个段落仍然大于 chunk_size，它会继续下一步。
• 接着，它尝试用 "\n"（单换行，通常代表行之间）来分割那些仍然过大的段落。
• 然后，它尝试用 " "（空格，单词之间）来分割。
• …

4.6.3 特殊文档结构拆分

若对于代码等特殊文本，可以尝试使用 Language 提供的不同的分割器（如 PythonCodeTextSplitter、HTMLHeaderTextSplitter 等）效果会更好，它会理解代码的语法结构。

这里了解下常见的拆分原则即可：

• Markdown：根据标头拆分（例如，#、##、###）
• HTML：使用标签拆分
• JSON：按对象或数组元素拆分
• Code代码：按函数、类或逻辑块拆分

这里我们以Python代码举例，其他的使用姿势可以参考官网接口，实际上用法与我们上面讲解的类似。

from langchain_text_splitters import PythonCodeTextSplitter

# 字符串文档
PYTHON_CODE = """
def hello_world():
    print("Hello, World!")

def hello_python():
    print("Hello, Python!")
"""

python_splitter = PythonCodeTextSplitter(chunk_size=50, chunk_overlap=0)
python_docs = python_splitter.create_documents([PYTHON_CODE])

for document in python_docs[:2]:
    print("*" * 30)
    print(f"{document}\n")

结果如下：

******************************
page_content='def hello_world():
    print("Hello, World!")'
******************************
page_content='def hello_python():
    print("Hello, Python!")'

4.7 文本向量（Embeddings）与向量存储

4.7.1 嵌入与嵌入模型（Embedding and Embedding Models）

计算机天生擅长处理数字，但不理解文字、图片的含义。嵌入（Embedding）的核心思想就是将人类世界的符号（如单词、句子、产品、用户、图片）转换为计算机能够理解的数值形式（即向量，本质上是一个数字列表），并且要求这种转换能够保留原始符号的语义和关系。

我们可以把它想象成一个翻译过程，把人类语言"翻译"成计算机的"数学语言"。

说明：我们之前一直用的大语言模型是生成式模型。它理解输入并生成新的文本（回答问题、写文章）。它内部实际上也使用嵌入技术来理解输入，但最终目标是"创造"。

而嵌入模型（Embedding Models）是表示型模型。它的目标不是生成文本，而是为输入的文本创建一个最佳的、富含语义的数值表示（向量）。如OpenAI的 “text-embedding-3-large” 嵌入模型；Google的 “gemini-embedding-001” 嵌入模型；阿里的 “Qwen3-Embedding-8B” 嵌入模型等。

4.7.1.1 什么是向量？

首先我们要知道，嵌入的结果就是一个向量，它本质上是一个数字列表（一维数组）。例如：[0.023, 0.487, -0.129, ..., 0.325]。对于向量来说，有两个关键概念需要了解：

向量维度

嵌入结果得到的列表长度是固定的，称为向量的"维度"。例如，OpenAI的 text-embedding-ada-002 模型会生成一个1536维的向量，text-embedding-3-large 模型会生成一个3072维的向量。

维度越高，通常能捕捉更细微的语义信息，但也需要更多的计算和存储资源。

向量空间

想象一个无限延伸的、拥有无数个维度的宇宙，这个宇宙就是一个向量空间。这有点抽象，可以想象一下：

• 在三维世界里，一个点可以用 (x, y, z) 坐标表示，例如 (2, 5, -1)。
• 在机器学习的高维向量空间中，一个点可能是 (0.1, 0.7, -0.2, 0.4, ..., 0.02)，一个有几百或几千个数字的坐标。

在这个空间里，每个点（即每个向量）都能代表一个概念。例如在嵌入模型中，一个点可以代表一个单词、一句话、一张图片、一个用户、一部电影等。

到这里，向量空间的威力就能体现出来：我们可以用数学来度量语义。可以通过计算两个向量之间的"距离"或"相似度"来实现这一点。

• 欧氏距离（Euclidean Distance）：就是我们高中几何学的两点之间的直线距离。距离越短，相似度越高。
• 余弦相似度（Cosine Similarity）：它忽略向量的绝对长度（大小），只关注两个向量在方向上的差异。在文本和语义的世界里，“方向"代表"含义”，而"长度"往往只代表"文本的长度"或"词汇的多少"。换句话说，余弦相似度关注的是"你们是否指向同一个方向" / “你们是否代表同一个含义”。

因此，在捕捉语义上的相似性上，余弦相似度是更常用的度量方式。

我们又能反推出，由于使用向量来绘制向量空间，而向量是有维度的，维度越高，则更能捕捉极其细微和复杂的语义差别（比如"高兴"和"喜悦"的区别）。

这能干什么？这能解决一个传统数据库（如MySQL）不擅长的问题：基于内容的相似性搜索，而不是基于精确匹配的查询。

4.7.1.2 嵌入模型应用场景

对于嵌入模型，实际上在示例选择器部分，我们已经使用过。当时使用的场景就是可以根据语义相似性完成示例的筛选。

根据嵌入的特性，由此延伸出了许多嵌入模型在AI应用的使用场景：

• 语义搜索（Semantic Search）：传统搜索依赖关键词匹配（搜"苹果"，只能找到包含"苹果"这个词的文档）。语义搜索则能将查询（如"一种红色的水果"）和文档库中的所有文档都转换为向量。然后计算查询向量与所有文档向量的相似度，返回最相似的文档。这样即使文档里没有"红色"和"水果"这些词，但只要它是关于"苹果"的，就能被找到。

借助嵌入模型进行文档搜索的过程：

1. 为多文档生成其各自的向量，
2. 为搜索查询语句生成向量，
3. 衡量查询向量与每个文档向量之间的相似性，得到相似度最高的文档。

• 检索增强生成（Retrieval-Augmented Generation, RAG）：这是当前大语言模型应用的核心模式。当用户向LLM提问时，系统首先使用嵌入模型在知识库（如公司内部文档）中进行语义搜索，找到最相关的内容，然后将这些内容和问题一起交给LLM来生成答案。这极大地提高了答案的准确性和时效性。
• 推荐系统（Recommendation Systems）：将用户（根据其历史行为、偏好）和物品（商品、电影、新闻）都转换为向量。喜欢相似物品的用户，其向量会接近；相似的物品，其向量也会接近。通过计算用户和物品向量的相似度，就可以进行精准推荐。
• 异常检测（Anomaly Detection）：正常数据的向量通常会聚集在一起。如果一个新数据的向量远离大多数向量的聚集区，它就可能是一个异常点（如垃圾邮件、欺诈交易）。

4.7.2 Embeddings嵌入模型类

在LangChain中，有很多的嵌入模型提供方，使用不同的模型提供方，需要安装为其各自包，例如：

• OpenAI：pip install -U langchain-openai
• Ollama：pip install -U langchain-ollama
• Google Gemini：pip install -U langchain-google-genai

4.7.2.1 定义嵌入模型

在这里我们选择OpenAI来进行后续操作。定义OpenAI下的嵌入模型使用：class langchain_openai.embeddings.base.OpenAIEmbeddings。

定义OpenAIEmbeddings嵌入模型类与定义聊天模型类似，如下所示：

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
  model="text-embedding-3-large",  # 是 OpenAI 2024年发布的最新嵌入模型，生成3072维的高质量向量
)

在LangChain框架中基础Embeddings类（OpenAIEmbeddings 继承了它）设计了两个核心方法来处理文本嵌入：

• .embed_documents()：用于处理文档 Documents。它的输入是多个文本。例如要将一个知识库里的所有段落都转换成向量后存入数据库，就会使用这个方法。
  • 它返回一个【二维列表】List[List[float]]。外层列表的每个元素对应一个输入文档，内层列表则是该文档的向量表示。
• .embed_query()：用于处理查询 Query。它的输入是单个文本（一个字符串，str）。例如，当用户提出一个问题时，需要将这个问题转换成向量，以便在数据库中搜索相似的文档段落，就会使用这个方法。
  • 它返回一个【一维列表】，里面是浮点数（List[float]），代表单个查询文本的向量。

其实分别对应文档与查询的向量生成。之所以设计成两个方法，是因为某些嵌入模型提供商（如OpenAI、Cohere等）会针对"被搜索的文档"和"搜索查询本身"采用不同的优化策略和模型。即使底层是同一个模型，也可能对两者进行不同的预处理（例如添加不同的指令前缀），以获得更好的搜索效果。

4.7.2.2 嵌入文档列表

embed_documents 的语义是"索引"。它的目的是预处理大量文本，为它们创建向量表示，以便后续被搜索。这一般是一个离线、批量处理的过程。代码如下：

from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import CharacterTextSplitter

markdown_path = "../Docs/Markdown/脚手架级微服务租房平台Q&A.md"
# single 模式加载后，默认只有一个 Document 对象
loader = UnstructuredMarkdownLoader(markdown_path)
data = loader.load()

# 生成分割器
text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
    encoding_name="cl100k_base", chunk_size=200, chunk_overlap=50
)
# 分割文档
documents = text_splitter.split_documents(data)

# 定义嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

# 嵌入文档列表，生成向量列表
# 注意这里需要提取文档内容为字符串列表，才能传递给嵌入模型
texts = [doc.page_content for doc in documents]
documents_vector = embeddings.embed_documents(texts)

print(f"文档数量为：{len(documents)}，生成了{len(documents_vector)}个向量的列表")
print(f"第一个文档向量维度：{len(documents_vector[0])}")
print(f"第二个文档向量维度：{len(documents_vector[1])}")

结果如下：

Created a chunk of size 916, which is longer than the specified 200
Created a chunk of size 916, which is longer than the specified 200
Created a chunk of size 260, which is longer than the specified 200
文档数量为：96，生成了96个向量的列表
第一个文档向量维度：3072
第二个文档向量维度：3072

可以看到 text-embedding-3-large 嵌入模型，生成的便是3072维的高质量向量。共存在96个文档，因此生成了96个向量的列表。

4.7.2.3 嵌入单个查询

embed_query 的语义是"搜索"。它的目的是在用户发起请求时，实时地将一个问题或指令转换为向量，用于在已索引的文档向量中进行检索。这是一个在线、实时、按需处理的过程。

为单个查询生成向量的代码如下：

from langchain_openai import OpenAIEmbeddings

# 定义嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

# 嵌入单个查询
query_vector = embeddings.embed_query("项目中遇到了哪些挑战？如何解决？")
print(f"向量维度：{len(query_vector)}")
print(f"向量前五个数值为：{query_vector[:5]}")

结果如下：

向量维度：3072
向量前五个数值为：[-0.01306734886020422, -0.004718094132840633, 
-0.005446741823107004, -0.014186487533152103, 0.004673811607062817]

4.7.3 向量存储（Vector Stores）

4.7.3.1 向量数据库介绍

在LangChain中，实际并不需要我们直接手动调用嵌入模型去生成向量，然后手动去比较向量。在RAG知识地图中，存在一个Vector Stores向量存储。

实际上，向量是被管理在专门的向量存储介质中，如向量数据库。向量存储的核心任务是解决一个传统数据库（如MySQL）不擅长的问题：基于内容的相似性搜索（Similarity Search），而不是基于精确匹配的查询。

想象一下，一篇文档若转换成一个有1536个浮点数的向量。一百万篇文档就是1536MB（约1.5GB）的纯向量数据。这只是一个起点，现实中的数据集可能轻松达到千万甚至亿级。如何高效地存储和管理这些向量？

向量数据库则提供了专门用于高效存储、管理和检索高维向量的能力。其核心就是"高效地组织和检索这些数据"。

常见的向量数据库核心机制如下：

• 专门的索引：这是向量数据库的灵魂。它们不会使用暴力搜索，而是会预先为所有向量构建一种特殊的索引结构。常见的方法有近似最近邻（ANN）搜索：为了追求极致的速度，它愿意牺牲一点点精度。它不会保证找到绝对最相似的向量（即最近邻），但能以极高的概率找到非常相似的向量。通过聚类、分层、压缩等算法技术，将搜索范围从"整个数据库"缩小到"几个最可能的候选集"。

  这就像在一个大图书馆里找书，你不是从A到Z遍历所有书架，而是先根据分类（文学、历史、科学）找到大概区域，再在这个区域内仔细找，这样快得多。

• 向量相似度计算优化：向量数据库底层使用高度优化的库来进行向量运算。如FAISS向量数据库，它是Facebook AI研究院开发的一种高效的相似性搜索和聚类的库。它能够快速处理大规模数据，并且支持在高维空间中进行相似性搜索。

  这些库充分利用了CPU的SIMD指令集和GPU的并行计算能力，让大规模的向量计算速度极快。SIMD指令集指"单指令多数据流"技术，是一种采用一个控制器来控制多个处理器，本质上非常类似一个向量处理器，可对控制器上的一组数据（又称"数据向量"）同时分别执行相同的操作从而实现空间上的并行。简单来说，就是一个指令能够同时处理多个数据。

• 数据管理功能：现代的向量数据库不仅有专门的索引，还提供了完整的数据管理功能：

  • CRUD操作：支持增删改查，可以动态地更新向量数据。
  • 元数据过滤：这是非常关键的功能。除了向量本身，每篇文档还有元数据（Metadata），比如创建时间、作者、类别等。向量数据库允许你先用元数据过滤（“找出xxxx年以后的、属于科技类别的文档”），再在这个缩小的范围内进行向量相似度搜索，极大地提升了准确性和效率。
  • 可扩展性与持久化：它们可以轻松地分布式部署，处理海量数据；同时保证数据持久化，不像纯内存方案一样断电丢失。
  • 集成方便：提供友好的API（如gRPC, RESTful），使得像LangChain这样的框架可以轻松地与之集成，开发者无需关心底层细节。

因此，我们可以利用专门的向量数据库（如Chroma, Weaviate, Pinecone, Qdrant, Milvus等），通过构建索引、优化计算流程、并提供丰富的过滤和管理功能，从而实现对海量高维向量数据进行快速、可扩展的【相似性检索】的一套完整技术方案。

LangChain框架则通过与这些向量数据库集成，让开发者无需手动处理向量生成、存储和比较的复杂性，只需关注业务逻辑本身，极大地提高了开发效率和应用性能。

下面我们将演示如何使用LangChain提供的内存级存储，与集成的向量数据库。

4.7.3.2 内存存储（InMemoryVectorStore）

我们将使用LangChain的 InMemoryVectorStore 来实现向量的内存存储。

（1）基本操作

• 初始化：LangChain中的大多数向量在初始化向量存储时接受嵌入模型作为参数。如下所示：

from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore

# 定义嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# 内存存储初始化
vector_store = InMemoryVectorStore(embedding=embeddings)

• 添加文档：我们可以使用 add_documents 方法，向内存存储中去添加文档。要注意的是，该方法会为添加的文档编排索引，索引列表随着该方法返回。

这也就是在前文中，我们一直在提的：当我们想对某文本进行【数据检索】时的两个步骤：

• 编制索引：用于从源中摄取数据并为其编制索引。
• 检索和生成：接受用户查询并从索引中检索相关数据，然后将其传递给模型。

对于第一步，我们终于要完成了。如下所示：

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import CharacterTextSplitter

# 生成分割器
text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
    encoding_name="cl100k_base", chunk_size=200, chunk_overlap=50
)
# 加载文档
data = UnstructuredMarkdownLoader("../Docs/Markdown/脚手架级微服务租房平台Q&A.md").load()
# 分割文档
documents = text_splitter.split_documents(data)

# 添加文档
ids = vector_store.add_documents(documents=documents)
print(f"共编排了{len(ids)}个文档索引")
print(f"前3个文档的索引是：{ids[:3]}")

结果输出：

共编排了96个文档索引
前3个文档的索引是：['3b22073d-8a8f-4f74-b99a-09d6343e7602', '3c2e05ea-1681-4d5e-a31c-f01f2e0a0ac8', 'dcc3da40-b1bf-4ae3-9fd4-1f758b47af0f']

• 获取文档：使用 get_by_ids 方法，通过索引列表获取对应的文档列表。如下所示：

doc_3 = vector_store.get_by_ids(ids[:3])
print(f"{[doc.page_content for doc in doc_3]}")

• 删除文档：使用 delete 方法，删除传入索引列表对应的文档列表；若不传入索引列表，则认为全量删除。如下所示：

vector_store.delete(ids=ids[:3])

测试一下删除是否成功：

doc_3 = vector_store.get_by_ids(ids[:3])
print(f"{[doc.page_content for doc in doc_3]}")

最终输出：[]，可以看到确实删除成功。

（2）向量搜索

如果我们传入一个查询，向量存储将嵌入该查询，在所有嵌入的文档中执行相似性搜索，并返回最相似的文档。

这体现了两个重要的概念：首先，需要有一种方法来衡量查询与任何嵌入文档之间的相似性。其次，需要有一种算法能够高效地在所有嵌入文档中执行这种相似性搜索。

对于【相似性】这一点，之前已经讲过，再来回顾下：

• 欧氏距离（Euclidean Distance）：就是我们高中几何学的两点之间的直线距离。距离越短，相似度越高。
• 余弦相似度（Cosine Similarity）：它忽略向量的绝对长度（大小），只关注两个向量在方向上的差异。在文本和语义的世界里，“方向"代表"含义”，而"长度"往往只代表"文本的长度"或"词汇的多少"。换句话说，余弦相似度关注的是"你们是否指向同一个方向" / “你们是否代表同一个含义”。

因此，在捕捉语义上的相似性上，余弦相似度是更常用的度量方式。

对于【相似性搜索】则可以通过向量存储提供的搜索方法实现。

• 相似性搜索：想要获取根据相似性搜索的结果，即嵌入单个查询，并查找相似的文档，并将它们作为文档列表返回。这可以使用 similarity_search 方法来实现。说明：InMemoryVectorStore 是根据【余弦相似度】来捕捉语义的。

search_docs = vector_store.similarity_search(query="数据库表怎么设计的？", k=2)
for doc in search_docs:
    print("*" * 30)
    print(doc.page_content)

方法参数解释如下：

• query：输入的查询str。
• k：要返回的文档数。默认为4。

打印结果如下：

******************************
提供两种方案有以下好处： 
降低门槛：单机版让用户快速体验核心功能，避免初期陷入复杂部署。 
灵活演进：用户可从单机版过渡到集群版，逐步适应生产需求。 
业务专属问题 
为什么将房源表拆分为水平分表（如按状态分表）？具体如何实现？ 
原因： 
单表数据量过大（如房源状态频繁更新）会导致查询性能下降； 
不同状态（上架/已租/下架）的查询频率不同，拆分可减少锁争用； 
******************************
数据库：MySQL + MyBatis
MySQL 和 MyBatis 是数据存储和访问的一个常见做法。MySQL 是一个开源的关系数据库管理系统，它可以免费使用。MyBatis 提供了一种相对简单的方法来执行 SQL 语句，不需要编写复杂的 ORM 映射文件。
缓存：Redis

可以看见，搜索出来的内容，是较为符合我们的预取。

有的同学可能有疑问了：搜索出来的结果，为什么不直接是问题的答案，而只是相似的文档呢？

那么恭喜你已经具备了构建智能应用基础逻辑了。此时我们可以将"找到的最相关内容"和我们"提出的问题"一起交给LLM来生成答案，这便能极大地提高答案的准确性和时效性。以上流程，叫做检索增强生成（Retrieval-Augmented Generation, RAG），这是当前大语言模型应用的核心模式。

• 元数据过滤：回顾一下，每个 Document 对象，都包含了以下参数：
  • id：可选的文档标识符。理想情况下，这应该在整个文档集合中是唯一的，并格式化为UUID，但不会强制执行。
  • page_content：字符串文本
  • metadata：与内容关联的任意元数据。类型为 dict [Optional]

虽然向量数据库实现了搜索算法，来有效地搜索所有嵌入的文档以找到最相似的文档。但现实场景中，我们还希望通过先根据元数据进行过滤，来帮助缩小搜索范围。例如从特定来源或日期范围检索文档。

from langchain_core.documents import Document

def _filter_function(doc: Document) -> bool:
    return doc.metadata.get("source") == "hahaha"

search_docs = vector_store.similarity_search(
    query="数据库表怎么设计的？", 
    k=2, 
    filter=_filter_function
)
for doc in search_docs:
    print("*" * 30)
    print(doc.page_content)

这次，我们给搜索方法加入了 filter 参数，它接收一个bool值，表示我们可以根据条件选择是否过滤某些文档。因此我们定义了一个 _filter_function 过滤函数，可以根据文档元数据先过滤出文档，再去进行搜索。

示例中，我们给出的 source=="hahaha" 是无效的，因此会将所有的文档都过滤掉，导致示例代码运行后结果为空！将 source 换成正确的来源后，再次运行则结果正确。

4.7.3.3 Redis向量存储

我们还可以使用Redis来存储向量。大多数开发者都熟悉Redis，因为它速度快、拥有庞大的客户端库生态系统，并且多年来已被众多大型企业采用。从本质上讲，Redis是一种键值型的NoSQL数据库，除了传统用例之外，Redis还提供了诸如搜索和查询功能等额外能力，允许用户在Redis内创建二级索引结构。这使得Redis能够以缓存的速度充当向量数据库。

（1）基本概念

• 理解RediSearch：RediSearch是Redis官方提供的一款高性能【搜索】与【全文索引】引擎模块。它基于Redis构建，使用户能够直接在Redis数据库中执行复杂的【搜索】和【分词查询】，无需额外引入外部搜索引擎。RediSearch特别适用于轻量级、响应速度要求较高的分词搜索场景。

  RediSearch提供了内置的分词功能，既避免LIKE的局限性（无法支持分词查询。例如：数据为"Apple iPhone 17 Pro Max 256GB星宇橙色"，搜"苹果17橙色"无法匹配），又无需部署ES这种大型组件，集成简单、性能出色，非常适合中等规模的全文检索需求。

• 理解Index：Index（索引）是RediSearch模块里的概念，用于定义一个查询目录。Index是一个独立的数据结构，它建立在多个Redis Keys（Hash类型）之上，这专门为了极速执行文本搜索、过滤和聚合而设计。它本身不存储数据，而是存储了指向其他Redis Keys的指针，和这些Keys中特定字段的索引信息。

• 理解Index Fields：Index Fields（索引字段）是创建索引时，明确指定的那些需要被索引的字段。它们定义了索引的"结构"或"蓝图"，告诉RediSearch：“请针对这些字段的内容，以其特定的方式为我构建快速搜索的能力。”

  可以把它想象成在一本书后面制作索引页（比如人名索引、主题索引）。我们不会把书中的每一个字都做到索引里，而是只选择那些重要的关键词（字段），并记录下它们出现的页码（文档ID）。这里的"关键词"就是Index Fields。

  在RediSearch中，索引字段是有特定的类型：

  • TAG：精确匹配的分类/标签，可以用来多值分类、过滤、分组。
  • NUMERIC：整数/浮点数，可以进行数值范围查询，排序和统计。
  • TEXT：全文搜索的字符串，支持分词、词干化、模糊匹配。
  • GEO：经纬度坐标，用来查询地理空间，计算距离。

• 理解metadata schema：schema我们讲过，就是描述数据结构的声明格式。metadata schema 则用来描述元数据的结构声明。

  这里的元数据是指我们将来要嵌入文档的元数据。因为对于文档元数据来说，它在存入Redis后，就被定义成了索引字段。

  对于文档元数据来说，里面存放的就是一些文档属性值，如 source 表示文档来源。我们还可以手动加入其他元数据，这需要设置每个字段的声明：name 表示字段名，type 表示字段类型。

metadata_schema=[
    {"name": "category", "type": "tag"},
    {"name": "num", "type": "numeric"},
]

（2）环境设置

使用Redis来存储向量，首先需要将相关环境配置好。

• 第一步：启动Redis服务端：使用Docker启动Redis实例。

这里要说明：对于 Redis 版本 >= 8.0，使用：

docker run -d -p 6379:6379 -it redis:latest

对于 Redis 版本 < 8.0，使用：

docker run -d -p 6379:6379 redis/redis-stack:latest

使用 docker ps 查看是否启动成功。

• 第二步：安装Redis客户端包，以便将来定义客户端，以及运行搜索和查询命令。

由于我们使用Python进行开发，则选择 redis-py 库完成客户端定义。要安装redis-py，只需：

pip install redis

• 第三步：在LangChain中想要使用 Redis 向量库，需要安装 langchain-redis 包

pip install -qU langchain-redis

• 第四步：定义Redis连接URL，客户端连接 Redis 时需要使用。

Redis连接URL的基本结构是：

[protocol]://[auth]@[host]:[port]/[database]

这部分根据自己Redis服务情况而定。例如："redis://localhost:6379"

• 第五步：测试连接（Ping）

import redis

redis_url = "redis://localhost:6379"
# 定义Redis客户端
redis_client = redis.from_url(redis_url)
# Ping
print(redis_client.ping()) 

若输出 True，则表示连接测试成功。

（3）基本操作

• 初始化：LangChain中使用 RedisVectorStore 初始化 Redis 向量存储。由于Redis需要相关配置，如连接URL等，因此LangChain提供了 RedisConfig 配置类供我们使用。如下所示：

from langchain_openai import OpenAIEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore

# 定义嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# 配置 Redis 客户端
redis_url = "redis://192.168.100.238:6379"
config = RedisConfig(
    index_name="qa",
    redis_url=redis_url,
    metadata_schema=[
        {"name": "category", "type": "tag"},
        {"name": "num", "type": "numeric"},
    ],
)
# Redis 存储初始化
vector_store = RedisVectorStore(embeddings, config=config)

class langchain_redis.vectorstores.RedisVectorStore 类，其初始化参数如下：

• embeddings：用于此存储的Embeddings实例。
• config：可选的RedisConfig对象。

class langchain_redis.config.RedisConfig 配置类，其关键配置参数如下：

• index_name：Redis中索引的名称。默认为生成的ULID唯一标识符。
• key_prefix：Redis Key的前缀。如果未设置，则默认为index_name。
• redis_url：Redis实例的URL。默认为"redis://localhost:6379"。
• metadata_schema：元数据字段的schema。设置该字段对于将来的元数据过滤有帮助。

我们可以根据 Index Name 查询其下的所有的 Index Fields，这需要安装 redisvl（pip install -U redisvl），使用 rvl 命令行工具来检查索引。

rvl index info -i qa --host 192.168.100.238 --port 6379

可以看到 Index Fields 中就包括了：

• text：文档文本。
• embedding：向量，类型为数组。
• _index_name：索引名。
• metadata schema：文档元数据。

其中文档元数据的每个属性还都被当作单独的 Index Fields。

• 添加文档：我们可以使用 add_documents 方法，向向量库中去添加文档。这次我们可以给被分割的文档添加相关的元数据。如下所示：

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import CharacterTextSplitter

# 生成分割器
text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
    encoding_name="cl100k_base", chunk_size=200, chunk_overlap=50
)
# 加载文档
data = UnstructuredMarkdownLoader("../Docs/Markdown/脚手架级微服务租房平台Q&A.md", category="QA").load()
# 分割文档
documents = text_splitter.split_documents(data)
# 为文档添加元数据
for i, doc in enumerate(documents, start=1):
    doc.metadata["category"] = "QA"
    doc.metadata["num"] = i

ids = vector_store.add_documents(documents=documents)
print(f"共编排了{len(ids)}个文档索引")
print(f"前3个文档的索引是：{ids[:3]}")

结果输出：

共编排了96个文档索引
前3个文档的索引是：['qa::01K4Q0A3DSQVZBRFKJD5MS25HG', 
'qa::01K4Q0A3DSQVZBRFKJD5MS25HH', 'qa::01K4Q0A3DSQVZBRFKJD5MS25HJ']

在Redis中的存储如下所示，下面的 embedding 字段被转换为二进制存储，因此展示了乱码。除此之外，还存放了我们给文档设置的元数据信息。

• 获取文档：使用 get_by_ids 方法，通过索引列表获取对应的文档列表。如下所示：

ids = [":01K4Q0A3DSQVZBRFKJD5MS25HJ", ":01K4Q0A3DSQVZBRFKJD5MS25HK", 
":01K4Q0A3DSQVZBRFKJD5MS25HM"]
doc_3 = vector_store.get_by_ids(ids)
print(f"{[doc.page_content for doc in doc_3]}")

• 删除文档：使用 delete 方法，删除传入索引列表对应的文档列表。如下所示：

vector_store.delete([":01K4Q0A3DSQVZBRFKJD5MS25HJ"])

或者：

# 删除指定内容
vector_store.index.drop_keys(["qa::01K4Q0A3DSQVZBRFKJD5MS25HJ"])
# 全量删除，删除索引
vector_store.index.delete(drop=True)

（4）向量搜索

• 相似性搜索：想要获取根据相似性搜索的结果，即嵌入单个查询，并查找相似的文档，并将它们作为文档列表返回。这可以使用 similarity_search 方法来实现。代码如下：

search_docs = vector_store.similarity_search(query="数据库表怎么设计的？", k=2)
for doc in search_docs:
    print("*" * 30)
    print(doc.page_content)

方法参数解释如下：

• query：输入的查询str。
• k：要返回的文档数。默认为4。

除了上面的 similarity_search 方法，其实还提供了：

• 根据向量搜索方法：similarity_search_by_vector
• 根据查询搜索方法，并返回相似分值：similarity_search_with_score
• 根据向量搜索方法，并返回相似分值：similarity_search_with_score_by_vector

例如，我们希望【根据查询语句搜索，并返回相似分值】：

scored_results = vector_store.similarity_search_with_score(query="数据库表怎么设计的？", k=4)
for doc, score in scored_results:
    print("*" * 30)
    print(f"Content: {doc.page_content[:100]}...")
    print(f"Metadata: {doc.metadata}")
    print(f"Score: {score}")

结果如下：

******************************
Content: 提供两种方案有以下好处： 
降低门槛：单机版让用户快速体验核心功能，避免初期陷入复杂部署。 
灵活演进：用户可从单机版过渡到集群版，逐步适应生产需求。 
业务专属问题 
为什么将房源表拆分为水平分表（... 
Metadata: {'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md', 'category': 'QA', 'num': 79}
Score: 0.580324351788
******************************
Content: 数据库：MySQL + MyBatis
MySQL 和 MyBatis 是数据存储和访问的一个常见的做法。MySQL 是一个开源的关系数据库管理系
统，它可以免费使...
Metadata: {'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md', 'category': 'QA', 'num': 23}
Score: 0.582783222198

注意：分数越低表示相似度越高。

• 元数据过滤：对于上述列举的方法，在他们搜索之前，都可以根据元数据先进行过滤。对于 RedisVectorStore 来说，需要使用Redis过滤表达式进行筛选。示例如下：

from redisvl.query.filter import Tag

# 过滤表达式
filter_condition = Tag("source") == "hahahaha"
scored_results = vector_store.similarity_search_with_score(
    query="数据库表怎么设计的？", 
    k=2, 
    filter=filter_condition
)
for doc, score in scored_results:
    print("*" * 30)
    print(f"Content: {doc.page_content[:100]}...")
    print(f"Metadata: {doc.metadata}")
    print(f"Score: {score}")

示例中，我们给出的 source=="hahahaha" 是无效的，因此会将所有的文档都过滤掉，导致示例代码运行后结果为空！将 source 换成正确的来源后，再次运行则结果正确。

对于Redis过滤表达式，也可以使用 & 和 | 运算符组合。如我们现根据元数据中的 category 与 num 进行过滤，如下所示：

from redisvl.query.filter import Tag, Num

# 过滤表达式
category_is_qa = Tag("category") == "qa"
num_is_under_50 = Num("num") < 50
filter_condition = category_is_qa & num_is_under_50

scored_results = vector_store.similarity_search_with_score(
    query="数据库表怎么设计的？",
    k=2,
    filter=filter_condition
)
for doc, score in scored_results:
    print("*" * 30)
    print(f"Content: {doc.page_content[:100]}...")
    print(f"Metadata: {doc.metadata}")
    print(f"Score: {score}")

• 最大边际相关性搜索：回顾一下什么是最大边际相关性？它是一种重新排序算法，它使用语义相似性作为基础工具，从一个候选集中挑选出一组既能代表查询主题又彼此多样化的结果。

  • 【语义相似性】就像面试官衡量每个应聘者与职位要求的匹配度。他会给每个应聘者打一个分数。
  • 【最大边际相关性】就像团队经理（MMR算法）要组建一个团队。目标是选出一组"精华"结果，而不是一个单一结果。

  MMR使用场景：

  • 推荐系统：推荐与用户兴趣相关但又不同类型的物品，避免"信息茧房"。
  • 文档摘要：从长文档中选择能代表主旨又包含不同信息的句子，避免摘要内容重复。
  • RAG(检索增强生成)：在从知识库检索完一堆相关文档后，使用MMR进行去重和多样化筛选，再交给LLM生成答案，能有效提升答案质量和减少幻觉。

使用MMR搜索，需要用到 max_marginal_relevance_search 方法，代码如下：

from redisvl.query.filter import Tag, Num

# 过滤表达式
category_is_qa = Tag("category") == "qa"
num_is_under_50 = Num("num") < 50
filter_condition = category_is_qa & num_is_under_50

mmr_results = vector_store.max_marginal_relevance_search(
    query="数据库表怎么设计的？",
    k=2,
    fetch_k=10,
    filter=filter_condition
)
for doc in mmr_results:
    print("*" * 30)
    print(f"Content: {doc.page_content[:100]}...")
    print(f"Metadata: {doc.metadata}")

我们需要注意一个参数 fetch_k，fetch_k 是MMR算法第一步中，从向量库中初步获取的候选文档数量。

为了理解它，我们首先要明白 MMR搜索是一个两阶段过程：

1. 初步获取：系统首先根据查询的纯向量相似度，从庞大的向量库中找出最相似的 fetch_k 个文档。这一步的目标是"广撒网"，先找到一个足够大的相关文档池。
2. 重新排序与筛选：然后，MMR算法会在这个较小的候选池（大小为 fetch_k）中运行。它不再只考虑与查询的相似度，还会考虑候选文档之间的多样性。它会从这 fetch_k 个文档中，挑选出既与查询相关，彼此之间又不太相似的 k 个文档作为最终结果。

因此，其目的就是在保证相关性的前提下，提升结果的多样性。打印结果如下：

******************************
Content: 数据库：MySQL + MyBatis 
MySQL 和 MyBatis 是数据存储和访问的一个常见的做法。MySQL 是一个开源的关系数据库管理系
统，它可以免费使... 
Metadata: {'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md', 'category': 'QA', 'num': 23}
******************************
Content: 生产环境搭建
develop合并master
流水线部署
线上验证
你是如何优化项目性能的？你有哪些性能调优的经验？
房源搜索的筛选+排序用设计模式优化
热点数据存放缓存的优化
没有性能...
Metadata: {'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md', 'category': 'QA', 'num': 40}

4.7.3.4 Pinecone向量存储

（1）Pinecone介绍

Pinecone是为机器学习应用量身打造的生产级向量数据库服务，适用于高维向量数据的高效存储、索引与查询。它屏蔽了基础设施管理，提供无缝扩展、实时数据写入和强大安全保障，让开发者和数据科学家能够以极低运维成本，快速构建高效的相似度搜索、推荐系统和AI应用。

Pinecone是一个全托管的向量数据库平台，即负责所有后端维护、扩展、更新和监控，让用户专注于应用开发，无需担心数据库管理。

Pinecone地址：https://www.pinecone.io/ （魔法上网）

（2）环境设置

• 首次使用需注册新用户，选择个人免费版。
• 继续创建账户相关信息，或者直接右上角skip跳过。
• 注册成功会生成一个默认的API Key。注意保存好你的key。
• 也可以创建新的API key。
• 设置 PINECONE_API_KEY，将Key添加进环境变量。
• 更新包：

pip install -qU pinecone langchain-pinecone

（3）基本操作

• 初始化：LangChain中使用 PineconeVectorStore 类初始化 Pinecone 向量库。我们需要：
  1. 创建索引。
  2. 使用索引来初始化 PineconeVectorStore。

如下所示：

from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from pinecone import Pinecone, ServerlessSpec

# 建立索引
pc = Pinecone()
index_name = "qa"  
if not pc.has_index(index_name):
    pc.create_index(
        name=index_name,         # 索引名称
        dimension=3072,          # 尺寸，表示向量维度，需要和嵌入模型维度一致
        metric="cosine",         # 度量方式，cosine 表示余弦相似度
        spec=ServerlessSpec(
            cloud="aws",         # 亚马逊云
            region="us-east-1"   # 区域
        ),
    )

# 定义嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# 获取索引
index = pc.Index(index_name)
# 定义 Pinecone 向量存储
vector_store = PineconeVectorStore(embedding=embeddings, index=index)

PineconeVectorStore 类，其初始化参数如下：

• embeddings：用于此存储的Embeddings实例。
• index：Pinecone索引

如果我们是第一次创建索引，在Pinecone控制台则可以看见被创建的索引。

• 添加文档：我们可以使用 add_documents 方法，向向量库中去添加文档。如下所示：

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import CharacterTextSplitter

# 生成分割器
text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
    encoding_name="cl100k_base", chunk_size=200, chunk_overlap=50
)
# 加载文档
data = UnstructuredMarkdownLoader("../Docs/Markdown/脚手架级微服务租房平台Q&A.md", category="QA").load()
# 分割文档
documents = text_splitter.split_documents(data)
# 为文档添加元数据
for i, doc in enumerate(documents, start=1):
    doc.metadata["category"] = "QA"
    doc.metadata["num"] = i

ids = vector_store.add_documents(documents=documents)
print(f"共编排了{len(ids)}个文档索引")
print(f"前3个文档的索引是：{ids[:3]}")

结果如下：

共编排了96个文档索引
前3个文档的索引是：['eee6f5f6-696b-4088-94d0-d0511cc92623', '36b8f351-35e5-4096-
81ea-b29701a9b4c8', '8c133041-e2a4-4040-8b11-98b4d408a752']

控制台同步更新数据。

• 删除文档：

# 全量删除
vector_store.delete(delete_all=True)
# 删除指定id的文档列表
delete_ids = []
vector_store.delete(ids=delete_ids)

（4）向量搜索

想要获取根据相似性搜索的结果，即嵌入单个查询，并查找相似的文档，并将它们作为文档列表返回。这可以使用 similarity_search 方法来实现。代码如下：

search_docs = vector_store.similarity_search(
    query="数据库表怎么设计的？",
    k=2,
    filter={"category": "QA"},
)
for doc in search_docs:
    print("*" * 30)
    print(f"Content: {doc.page_content[:100]}...")
    print(f"Metadata: {doc.metadata}")

结果打印如下：

******************************
Content: 提供两种方案有以下好处： 
降低门槛：单机版让用户快速体验核心功能，避免初期陷入复杂部署。 
灵活演进：用户可从单机版过渡到集群版，逐步适应生产需求。 
业务专属问题 
为什么将房源表拆分为水平分表（... 
Metadata: {'category': 'QA', 'num': 79.0, 'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'} 
******************************
Content: 数据库：MySQL + MyBatis
MySQL 和 MyBatis 是数据存储和访问的一个常见的做法。MySQL 是一个开源的关系数据库管理系
统，它可以免费使用...
Metadata: {'category': 'QA', 'num': 23.0, 'source': '../Docs/Markdown/脚手架级微服务租房平台Q&A.md'}

4.8 检索器（Retrievers）与RAG实战

4.8.1 相关概念

4.8.1.1 检索系统

检索系统（Information Retrieval System, IR System）是一个为了满足用户信息需求，从大规模、非结构化的数据集合中，自动、高效地查找、排序并返回相关信息的计算机系统。

它的核心任务是：在正确的时间，以正确的方式，将正确的信息传递给正确的人。最常见的例子就是：搜索引擎（如Google、百度）。

随着大型语言模型的流行，检索系统已成为人工智能应用（例如RAG）的重要组成部分。且存在多种【不同类型】的检索系统，包括：

• 关系数据库：关系数据库是许多应用程序中使用的结构化数据存储的基本类型。数据存储在行（记录）和列（属性）中，可以通过SQL（结构化查询语言）进行高效的查询和操作。关系数据库擅长维护数据完整性、支持复杂查询以及处理不同数据实体之间的关系。
• 词法搜索索引：许多搜索引擎基于将查询中的单词与每个文档中的单词进行匹配，这种方法称为词法检索。即一个单词经常出现在用户的查询和特定文档中，那么这个文档可能是一个很好的匹配。这通常使用【倒排索引】实现。
• 向量数据库：向量存储不使用字频，而是使用【嵌入模型】将文档转换为高维向量表示。这允许使用余弦相似度等数学运算对嵌入向量进行有效的相似性搜索。

4.8.1.2 检索器

检索器是检索系统中的一个核心组件，它接收来自用户接口的查询（Query），检索出包含查询关键词的候选文档集合。

我们可以使用上面提到的任何检索系统实现方式创建检索器！如关系数据库、向量数据库等。由于其重要性和多样性，LangChain提供了一个统一的接口来与不同类型的检索系统进行交互。LangChain的检索器接口非常简单：

1. 输入：查询字符串
2. 输出：文档列表（标准化的LangChain文档对象Document）

例如，使用关系数据库的检索系统，检索器可以将问题转换为SQL语句，并执行查询，最后将查询结果响应用户。

4.8.2 使用向量数据库作为检索器

4.8.2.1 基本使用

向量存储是索引和检索非结构化数据的一种强大而有效的方法。可以通过调用向量数据库的 as_retriever 方法，将向量存储用作检索器。在这里我们使用Redis向量存储。

from langchain_openai import OpenAIEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore

# 定义嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# 配置 Redis 客户端
redis_url = "redis://192.168.100.238:6379"
config = RedisConfig(
    index_name="qa",
    redis_url=redis_url,
    metadata_schema=[
        {"name": "category", "type": "tag"},
        {"name": "num", "type": "numeric"},
    ],
)
# Redis 存储初始化
vector_store = RedisVectorStore(embeddings, config=config)
retriever = vector_store.as_retriever()

docs = retriever.invoke("数据库表怎么设计的？")
for doc in docs:
    print("*" * 30)
    print(doc.page_content[:30])

LangChain检索器是一个Runnable的对象，它是LangChain组件的标准接口。这意味着它有一些常用方法，包括 invoke，用于与其交互。默认情况下，向量存储检索器使用相似性搜索。

输出结果如下：

******************************
提供两种方案有以下好处： 
降低门槛：单机版让用户快速体验核 
******************************
数据库：MySQL + MyBatis
MySQL 和 M
******************************
不同状态（上架/已租/下架）的查询频率不同，拆分可减少锁争用 
******************************
分库分表：按user_id哈希分库，按月份分表（如chat_ 

as_retriever 方法也支持我们传递相关参数修改搜索结果，如：

• search_type：设置相似算法，包括："similarity"（默认）、"mmr" 或 "similarity_score_threshold"（相似性分数阈值）
• search_kwargs：
  • k：限制检索器返回的文档 k 数量。

retriever = vector_store.as_retriever(search_kwargs={"k": 2})

• fetch_k：要传递给MMR算法的文档量。

retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 2,
        "fetch_k": 10
    }
)

要注意的是，Retrievers 检索器虽然是Runnable对象，但其不提供任何流式处理，因为它本身通常是同步的、阻塞的操作。如下所示：

chunks = []
for chunk in retriever.stream("数据库表怎么设计的？"):
    chunks.append(chunk)
    print(chunk, end="|", flush=True)

调用可以发现，它是一次性返回，与 invoke 调用无异。

4.8.2.2 使用 @chain 创建"检索器"

除了使用 as_retriever 方法，我们还可以自行创建一个"检索器"。回想一下检索器的特点：

1. LangChain检索器是一个Runnable的对象
2. LangChain检索器输入为查询字符串，输出为文档列表（标准化的LangChain文档对象Document）

综上所述，我们可以：

from langchain_core.runnables import chain
from typing import List
from langchain_core.documents import Document

@chain
def retriever(query: str) -> List[Document]:
    return vector_store.similarity_search(query, k=2)

docs = retriever.invoke("数据库表怎么设计的？")
for doc in docs:
    print("*" * 30)
    print(doc.page_content[:30])

上面定义了一个函数，使用 @chain 修饰，该修饰可以使其成为Runnable函数，且满足检索器输入输出的要求。在函数中，我们依旧使用向量数据库的相似性搜索方法，这样灵活性也更高，想要进行元数据筛选也更方便。

注意，这并不是真正的检索器，检索器是一个Runnable对象，而我们定义的只是一个函数，具备其特点罢了。

4.8.3 RAG案例

RAG是当前大语言模型应用的核心模式。当用户向LLM提问时，系统首先使用嵌入模型在知识库中进行语义搜索，找到最相关的内容，然后将这些内容和问题一起交给LLM来生成答案。这极大地提高了答案的准确性和时效性。

由于LangChain检索器是一个Runnable的对象，我们便可以方便的使用链完成相关的调用。下面我们来完成一个最简单的RAG案例，将会完成：

1. 根据Query搜索最相关的4篇文档。
2. 将相关的文档转换为字符串，以便后续发送给聊天模型。
3. 将Query与文档字符串发送给聊天模型。
4. 聊天模型依据输出解析器格式输出内容。

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_redis import RedisConfig, RedisVectorStore
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# 定义聊天模型
model = ChatOpenAI(model="gpt-4o-mini")
# 定义嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# 配置 Redis 客户端
redis_url = "redis://192.168.100.238:6379"
config = RedisConfig(
    index_name="qa",
    redis_url=redis_url,
    metadata_schema=[
        {"name": "category", "type": "tag"},
        {"name": "num", "type": "numeric"},
    ],
)
# 定义 Redis 向量存储
vector_store = RedisVectorStore(embeddings, config=config)
# 生成检索器
retriever = vector_store.as_retriever()
# 定义提示词模板
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "human",
            """你是负责回答问题的助手。使用以下检索到的上下文片段来回答问题。如果你不知道答案，就说你不知道。最多只用三句话，回答要简明扼要。
Question: {question}
Context: {context}
Answer:""",
        ),
    ]
)

# 将文档转换为字符串
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 定义链
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 执行链，流式输出
for chunk in rag_chain.stream("数据库表怎么设计的？"):
    print(chunk, end="|", flush=True)

上述代码中，RunnablePassthrough 我们之前还没有见过，简单来说，RunnablePassthrough 是一个"伪"Runnable，它的主要作用是在链（Chain）中透明地传递输入数据，而不做任何修改。

当我们需要将原始输入和另一个处理过程的输出一起传递给下一个步骤时，就需要 RunnablePassthrough。就例如代码中，我们需要将【Query】与【通过检索出来的文档转换的字符串】同时发送给提示词模板。

最终打印结果如下：

|数据库|表|设计|可以|通过|水平|分|表|来|优化|性能|，|特别|是|当|单|表|数据|量|过|大|
|时|，可以|减少|锁|争|用|和|提升|查询|效率|。|实现|方式|是|将|状态|字段|独|立|存储|
|在|house|_status|表|中|，并|通过|house|_id|进行|关联|查询|。|使用|My|SQL|
和|My|Batis|进行|数据|存|储|与|访问|，同时|可以|利用|Redis|缓存|来|进|一步|提|高|性
能|。|

上面讲过检索器不提供任何流式处理。但这里可以使用流式输出。我们可以把像RAG这样的链的执行过程想象成两个阶段：

• 阶段一：准备阶段（同步、阻塞）：接收输入、检索文档、构建提示
• 阶段二：生成阶段（可流式）：调用LLM

流式输出的是最终答案，而非中间过程。

五、总结

1. LangChain的核心是标准化组件：通过Runnable统一接口（调用、批处理、流式传输、检查、组合），配合LCEL表达式（| 管道符）将模型、解析器、检索器等组件像管道一样自由编排成 RunnableSequence（链），从根本上解决了原生LLM的幻觉、提示词不规范、模型切换困难、输出非结构化、知识陈旧、无法连接外部系统六大痛点。
2. 聊天模型具备四大核心能力：
   • 定义模型：ChatOpenAI、init_chat_model工厂函数、ChatOllama支持API与本地部署多种接入方式；
   • 工具调用：通过 @tool、StructuredTool 创建工具，bind_tools 绑定工具，让LLM具备与外部世界交互的能力，工具输出以ToolMessage回传，content给模型阅读，artifact给链中后续组件使用；
   • 结构化输出：with_structured_output() 实现从"字符串"到"对象"的范式转换，支持Pydantic、TypedDict、JSON Schema三种输出结构；
   • 流式传输：stream()/astream() 逐块返回AIMessageChunk，底层依赖SSE协议（stream=True）+ OpenAI SDK HTTP客户端，源码层面通过 _stream() 与 _convert_chunk_to_generation_chunk() 完成转换。
3. 八大核心组件支撑RAG全流程：消息（Messages）统一通信格式并支持裁剪/过滤/合并管理历史；提示词模板实现提示词的复用与工程化；少样本提示用示例引导模型行为；输出解析器将非结构化输出转为结构化数据；文档加载器将各类数据源转为Document；文本分割器按长度/Token/语义拆分文档；嵌入模型与向量存储（内存/Redis/Pinecone）实现语义相似性检索；检索器以统一接口（输入查询字符串、输出文档列表）串联检索系统。
4. RAG是当前LLM应用的核心模式：离线阶段完成文档加载→文本分割→嵌入→向量入库；在线阶段通过检索器召回相关文档，与问题一起交给LLM生成答案，显著提升答案的准确性与时效性。

掌握LangChain的关键在于理解每个组件的作用和适用场景，而非死记硬背接口和参数——所有组件都是围绕"让大模型被更好地调用"这一目标衍生出来的功能扩展，组合它们，你就能构建出从简单对话到企业级知识库问答的完整AI应用。