LangChain核心组件-数据连接

jwang2026-01-302026-01-30

在LLM应用程序中，用户特定的数据不在大模型中，可能在外部系统或文档中，要使用这些数据进行增强，则需要数据连接
数据连接组件包括：文档加载器、文档切分、文本嵌入、向量存储、检索器与大模型生成

文档加载器

文档类型：文本、word、excel、pdf、视频等文件
文档加载器：将各种文档类型数据源加载数据，并将数据统一转换为“文档Document”,Document对象包含文本及其相关元数据

load方法

用于从指定的数据源读取数据，并将其转换成一个或多个文档，不仅仅局限于文本数据，还可以是网页、视频字幕等

如何加载网页(例)

我们为每个网页恢复一个 Document，其内容表示为“扁平化”字符串
其它加载器查看（https://www.langchain.com.cn/docs/how_to/）

简单快速 解析，我们为每个网页恢复一个 Document，其内容表示为“扁平化”字符串；
%pip install -qU langchain-community beautifulsoup4
高级 解析，我们为每个页面恢复多个 Document 对象，允许识别和遍历部分、链接、表格和其他结构。
%pip install -qU langchain-unstructured

简单快速的文本提取
它将返回一个 Document 对象的列表——每个页面一个——包含页面文本的单个字符串。在底层，它使用 beautifulsoup4 Python 库。
LangChain 文档加载器实现了 lazy_load 及其异步变体 alazy_load，返回 Document 对象的迭代器

import bs4
from langchain_community.document_loaders import WebBaseLoader

page_url = "https://python.langchain.com/docs/how_to/chatbots_memory/"

loader = WebBaseLoader(web_paths=[page_url])
docs = []
async for doc in loader.alazy_load():
    docs.append(doc)

assert len(docs) == 1
doc = docs[0]

==> 
USER_AGENT environment variable not set, consider setting it to identify your requests.

print(f"{doc.metadata}\n")
print(doc.page_content[:500].strip())


==> 
{'source': 'https://python.langchain.com/docs/how_to/chatbots_memory/', 'title': 'How to add memory to chatbots | \uf8ffü¶úÔ∏è\uf8ffüîó LangChain', 'description': 'A key feature of chatbots is their ability to use content of previous conversation turns as context. This state management can take several forms, including:', 'language': 'en'}

How to add memory to chatbots | ü¶úÔ∏èüîó LangChain


Skip to main contentShare your thoughts on AI agents. Take the 3-min survey.IntegrationsAPI ReferenceMoreContributingPeopleLangSmithLangGraphLangChain HubLangChain JS/TSv0.3v0.3v0.2v0.

这基本上是页面 HTML 中文本的转储。它可能包含多余的信息，如标题和导航栏。如果您熟悉预期的 HTML，您可以通过 BeautifulSoup 指定所需的 <div> 类和其他参数。下面我们仅解析文章的主体文本：

loader = WebBaseLoader(
    web_paths=[page_url],
    bs_kwargs={
        "parse_only": bs4.SoupStrainer(class_="theme-doc-markdown markdown"),
    },
    bs_get_text_kwargs={"separator": " | ", "strip": True},
)

docs = []
async for doc in loader.alazy_load():
    docs.append(doc)

assert len(docs) == 1
doc = docs[0]

print(f"{doc.metadata}\n")
print(doc.page_content[:500])


==> 
{'source': 'https://python.langchain.com/docs/how_to/chatbots_memory/'}

How to add memory to chatbots | A key feature of chatbots is their ability to use content of previous conversation turns as context. This state management can take several forms, including: 


对页面内容进行向量搜索
一旦我们将页面内容加载到LangChain Document 对象中，我们可以以通常的方式对其进行索引（例如，用于RAG应用）。下面我们使用OpenAI 嵌入，尽管任何LangChain嵌入模型都可以。
%pip install -qU langchain-openai

import getpass
import os

if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass.getpass("OpenAI API Key:")

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

vector_store = InMemoryVectorStore.from_documents(setup_docs, OpenAIEmbeddings())
retrieved_docs = vector_store.similarity_search("Install Tavily", k=2)
for doc in retrieved_docs:
    print(f'Page {doc.metadata["url"]}: {doc.page_content[:300]}\n')

文本分割器

如何分割JSON数据-RecursiveJsonSplitter

这个 JSON 分割器在允许控制块大小的同时分割 JSON 数据。它深度优先遍历 JSON 数据并构建较小的 JSON 块。它尝试保持嵌套的 JSON 对象完整，但如果需要，会将其分割，以保持块在 min_chunk_size 和 max_chunk_size 之间。

如果值不是嵌套的 JSON，而是一个非常大的字符串，则该字符串不会被分割。如果您需要对块大小进行严格限制，可以考虑将其与递归文本分割器组合使用。还有一个可选的预处理步骤，可以通过先将列表转换为 JSON（字典）然后再进行分割来分割列表。

文本如何分割：JSON 值。
块大小如何测量：按字符数。

%pip install -qU langchain-text-splitters


# 加载数据
import json

import requests

# This is a large nested json object and will be loaded as a python dict
json_data = requests.get("https://api.smith.langchain.com/openapi.json").json()

# 指定 max_chunk_size 来限制块大小
from langchain_text_splitters import RecursiveJsonSplitter
splitter = RecursiveJsonSplitter(max_chunk_size=300)
# 获取JSON块
json_chunks = splitter.split_json(json_data=json_data)
#获取 LangChain 文档 对象
docs = splitter.create_documents(texts=[json_data])
#或者使用 .split_text 直接获取字符串内容
texts = splitter.split_text(json_data=json_data)

#默认情况下，json分割器不会分割列表
#指定 convert_lists=True 来预处理json，将列表内容转换为 index:item 形式的字典，作为 key:val 对
texts = splitter.split_text(json_data=json_data, convert_lists=True)
#列表已被转换为字典，但即使分成多个块，仍保留所有所需的上下文信息：
print(texts[1])

如何按字符递归分割文本-RecursiveCharacterTextSplitter

这个文本分割器是推荐用于通用文本的。它通过字符列表进行参数化。它尝试按顺序在这些字符上进行分割，直到块的大小足够小。默认列表是 [‘\n\n’, ‘\n’, ‘ ‘, ‘’]。这会尽量保持所有段落（然后是句子，再然后是单词）在一起，因为这些通常被认为是语义上最相关的文本片段

文本是如何分割的：按字符列表。
块大小是如何测量的：按字符数。

%pip install -qU langchain-text-splitters
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Load example document
with open("state_of_the_union.txt") as f:
    state_of_the_union = f.read()

text_splitter = RecursiveCharacterTextSplitter(
    # Set a really small chunk size, just to show.
    chunk_size=100,
    chunk_overlap=20,
    length_function=len,
    is_separator_regex=False,
)
texts = text_splitter.create_documents([state_of_the_union])
print(texts[0])
print(texts[1])
text_splitter.split_text(state_of_the_union)[:2]

RecursiveCharacterTextSplitter 设置的参数：
  chunk_size: 每个块的最大大小，大小由 length_function 决定。
  chunk_overlap: 块之间的目标重叠。重叠的块有助于减轻在块之间划分上下文时信息的丢失。
  length_function: 确定块大小的函数。
  is_separator_regex: 分隔符列表（默认为 ['\n\n', '\n', ' ', '']）是否应被解释为正则表达式。

嵌入模型包装器和向量数据库

文本嵌入模型Embedding Model

将文本作为输入并返回词向量

Enbedding类

LangChain提供，它为多种文本嵌入模型（OpenAi、Cohcre等）提供了统一的接口，通过该类实例化的嵌入模型包装器，可以将文档转换为向量数据，同时将搜索的问题也转换为向量数据，可通过计算搜索问题和文档在向量空间中的距离，寻找在向量空间中最相似的文本
实例化的Emdedding类被称为嵌入模型包装器，与model I/O模块的LLM模式包装器和聊天吗模型包装器Chat Model 为三大模型包装器

向量数据库

文本向量化后，需要保存供后面步骤使用，则需要向量数据库

LangChain中使用

向量存储库种类很多，比如chroma、FAISS、Pinecone。。。，LangChain提供了一个标准接口，通过向量存储库的示例上调用as_retriever方法就可以得到一个基于向量存储库的检索器

检索器

向量存储库的包装器，包装了一个统一接口，无论底层的向量存储库是什么，都可以使用同样的方式对其进行查询，可以轻松切换向量存储库
检索组件是LangChain解决”数据孤岛”问题的核心技术，通过RAG模式让LLM具备访问私有知识的能力。


文档加载 → 文本分割 → 向量化 → 存储 → 检索 → 增强生成


from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA

# 1. 文档加载
loader = WebBaseLoader("https://example.com/article")
documents = loader.load()

# 2. 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)

# 3. 向量化存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# 4. 检索增强生成
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever
)

# 使用
answer = qa_chain.run("文章的主要观点是什么？")