最近想从文档知识库里提取一些文本片段，送给大模型，整理出想要的内容，并减少幻觉，这就是常说的检索增强生成技术（RAG：Retrieval-Augmented Generation）。
一条可行的技术路线就是使用LangChain，带着这个想法开始学习LangChain，没想到LangChain官网的第一个实例就是语义搜索，链接在这里：https://docs.langchain.com/oss/python/langchain/knowledge-base
一步一步完成了这个例子，一个简陋的基于PDF的问答程序就出来了。当然，这里还没有做界面。
大模型API
个人进行大模型的本地部署有一定难度，还对显卡有要求，openai有免费的API接口，可惜国内的大多数用户访问不了，这里花11.66元买了aliyun的三个月的100万token试用，足够学习过程的各种折腾。
订单完成之后，在阿里云百炼平台的密钥管理里可以创建API Key，类似这样的“sk-2026***0204”一个字符串。
为了不在源代码里暴露这个secret key，我在windows里设置了系统环境变量 DASHSCOPE_API_KEY。
设置了环境变量之后，记得重启vscode开发环境。否则，你可能会看到下面的错误信息：
ValueError: Sync client is not available. This happens when an async callable was provided for the API key. Use async methods (ainvoke, astream) instead, or provide a string or sync callable for the API key.
大模型的Hello World
安装langchain
安装langchain的相关软件包，使用pip安装即可，最好一次性把相关内容先装上，后面早晚会用到。
pip install langchain langchain-openai langchain-core langchain-community langchain-chroma
使用pip show langchain可以查看langchain的版本，我使用的是V1.2.7。
openai的Hello World写法
阿里官网上提供的第一个程序：
import os
from openai import OpenAI
client = OpenAI(
# 若没有配置环境变量，请用阿里云百炼API Key将下行替换为: api_key="sk-xxx",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
completion = client.chat.completions.create(
model="qwen-plus", # 模型列表: https://help.aliyun.com/model-studio/getting-started/models
messages=[
{'role': 'system', 'content': '你是AI大模型助手'},
{'role': 'user', 'content': '你是谁？'}
]
)
print(completion.choices[0].message.content)
如果一切顺利，会得到类似下面的输出：
你好！我是通义千问（Qwen），阿里巴巴集团旗下的超大规模语言模型。我能够回答问题、创作文字，比如写故事、写公文、写邮件、写剧本、逻辑推理、编程等等，还能表达观点，玩游戏等。如果你有任何问题或需要帮助，欢迎随时告诉我！😊
上面的程序并没有使用langchain，只使用了openai的类库。
langchain的Hello World
如果只是简单的问答，使用langchain的最简练的代码是：
import os
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="qwen-plus",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
response = llm.invoke('你是谁？')
print(response.content)
带系统提示词的Hello World
可以加上系统提示词，代码里多了两个类HumanMessage, SystemMessage：
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
llm = ChatOpenAI(
model="qwen-plus",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
messages = [
SystemMessage(content="你是AI大模型助手"),
HumanMessage(content="你是谁？")
]
response = llm.invoke(messages)
print(response.content)
批量对话
可以一次问大模型多个问题。
import os
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="qwen-plus",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
results = llm.batch([
"LangChain里最重要的几个概念是什么？",
"给一份10倍速学会LangChain的大纲",
])
for idx, res in enumerate(results):
print(f"\n批量调用第{idx+1}条：", res.content)
print('=' * 50)
读入PDF
先要安装pypdf包。
pip install langchain-community pypdf
langchain里有一个Document类，实现了对文档的抽象，主要有三个属性：
page_content：文档内容，字符串;metadata：文档元数据，字典。例如：title:文档标题，creationdate:建立时间。id：（可选）文档的字符串标识符。
最近正在读Dan Koe的一篇文章《How to fix your entire life in 1 day》，保存成PDF，19页，正好试着用langchain分析一下。
from langchain_community.document_loaders import PyPDFLoader
file_path = "how-to-fix-your-entire-life.pdf"
loader = PyPDFLoader(file_path)
docs = loader.load()
print("PDF的页数：", len(docs))
print(f"{docs[0].page_content[:200]}\n") #第一页里的前200个字符
print(docs[0].metadata) #第一页的元数据
print(docs[1].metadata) #第二页的元数据
文本分隔
PDF文档的内容一般都比较长，直接送给大模型肯定不行，一个常用的办法就是把文本分块处理，一个块称为chunk。LangChain里已经提供了TextSplitter类，拆分文档非常方便。
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 每个块(chunk)有1000个字符，其中200个字符重复
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, chunk_overlap=200, add_start_index=True
)
all_splits = text_splitter.split_documents(docs)
print("chunks: ", len(all_splits))
print(all_splits[0].metadata)
print(all_splits[1].metadata)
print(all_splits[-1].metadata)
原始PDF有19页，每1000个字符一个块，处理后得到44个块。为了防止断章取义，使用了重叠字符chunk_overlap。
每个块里多了一个元数据start_index。例如，最后一个块的元数据：
{
'producer': 'Skia/PDF m144',
'creator': 'Mozilla/5.0...',
'creationdate': '2026-02-02T01:35:09+00:00',
'moddate': '2026-02-04T09:20:45+08:00',
'title': 'How to fix your entire life in 1 day - by DAN KOE',
'source': 'how-to-fix-your-entire-life.pdf',
'total_pages': 19,
'page': 18,
'page_label': '19',
'start_index': 803
}
向量嵌入
学过深度学习、大模型的基础知识后，应该知道，任何一串文本，可以映射为一个多维向量（元素都是浮点数）。比如GPT-3是768维的向量，阿里的text-embedding-v3支持1024维向量，text-embedding-v4则可以用2048维向量来表达。
将手里的文档全部嵌入之后，后面就可以进行相似度的检索，你输入的任何一段文字，都可以与向量库里的内容进行匹配查询，挑几条最相关的内容，再送给大模型进行推理分析，从而得到更准确的结果。
向量嵌入这一节的内容踩了几个坑，Langchain官方教程里的下面两行无法通过：
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
阿里百炼给的官方文本嵌入步骤是：
import os
from openai import OpenAI
client = OpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
# completion 指文本补全响应的结果
# text-embedding-v3 / v4 模型：
# 输入为字符串：最长支持 8,192 Token。
# 输入为字符串列表或文件：最多支持 10 条（行），每条（行）最长支持 8,192 Token。
# text-embedding-v1 / v2 模型：
# 输入为字符串：最长支持 2,048 Token。
# 输入为字符串列表或文件：最多支持 25 条（行），每条（行）最长支持 2,048 Token。
completion = client.embeddings.create(
model="text-embedding-v4",
input=[ # 可以是一个字符串，也支持多个字符串的数组。
"每天吃一个鸡蛋可以补充优质蛋白质，但胆固醇偏高的人应注意摄入量。",
"长期不吃早餐可能会影响血糖稳定，也容易导致午餐时暴饮暴食。",
"多喝水有助于促进新陈代谢，但短时间内大量饮水也可能增加身体负担。",
"适量运动有助于控制体重和增强心肺功能，但运动强度应循序渐进。",
],
# 指定的向量维度，默认值为1024，必须为以下值之一：
# 2048、1536（仅适用于text-embedding-v4）、
# 1024、768、512、256、128 或 64。
dimensions=1024,
)
#print(completion.model_dump_json())
vec1 = completion.data[0].embedding
vec2 = completion.data[1].embedding
print(f"嵌入向量维度: {len(vec1)}") # 应该是1024
print(f"向量样例: {vec1[:5]}") # 前5个值
print(f"向量样例: {vec2[:5]}") # 前5个值
注意这里使用了OpenAI，而不是ChatOpenAI，顺利的话，输出如下：
嵌入向量维度: 1024
向量样例: [-0.010881742462515831, -0.042937248945236206, -0.015206357464194298, -0.012966824695467949, 0.04257218539714813]
向量样例: [-0.019730322062969208, -0.059398338198661804, -0.012864702381193638, -0.0652344822883606, 0.0032735744025558233]
通过多次询问大模型，终于给出了一个正确的阿里百炼Embeddings适配器。可以正常调用embed_query函数。
import os
from openai import OpenAI
from langchain_core.embeddings import Embeddings
# 阿里百炼 Embeddings 适配器
class DashScopeEmbeddings(Embeddings):
def __init__(
self,
api_key: str,
model: str = "text-embedding-v4",
dimensions: int = 1024,
):
self.client = OpenAI(
api_key=api_key,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
self.model = model
self.dimensions = dimensions
def embed_documents(self, texts: list[str]) -> list[list[float]]:
response = self.client.embeddings.create(
model=self.model,
input=texts,
dimensions=self.dimensions,
)
return [item.embedding for item in response.data]
def embed_query(self, text: str) -> list[float]:
response = self.client.embeddings.create(
model=self.model,
input=[text],
dimensions=self.dimensions,
)
return response.data[0].embedding
embeddings = DashScopeEmbeddings(
api_key=os.getenv("DASHSCOPE_API_KEY"),
model="text-embedding-v4",
dimensions=1024,
)
vec1 = embeddings.embed_query("每天吃一个鸡蛋可以补充优质蛋白质，但胆固醇偏高的人应注意摄入量。")
vec2 = embeddings.embed_query("长期不吃早餐可能会影响血糖稳定，也容易导致午餐时暴饮暴食。")
print(f"嵌入向量维度: {len(vec1)}") # 应该是1024
print(f"向量样例: {vec1[:5]}") # 前5个值
print(f"向量样例: {vec2[:5]}") # 前5个值
向量存储
内存存储
大量文本被向量化之后，有多种存储方案。最简单的是内存存储，不过缺少持久化。还可以放在MongoDB和PostgreSQL数据库中。
适配的Embeddings可以使用InMemoryVectorStore内存存储。
from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
docs = [Document(page_content=input,)
for input in [
"每天吃一个鸡蛋可以补充优质蛋白质，但胆固醇偏高的人应注意摄入量。",
"长期不吃早餐可能会影响血糖稳定，也容易导致午餐时暴饮暴食。",
"多喝水有助于促进新陈代谢，但短时间内大量饮水也可能增加身体负担。",
"适量运动有助于控制体重和增强心肺功能，但运动强度应循序渐进。",
]]
vector_store = InMemoryVectorStore(embeddings)
ids = vector_store.add_documents(documents=docs)
print(ids)
chroma持久存储
专门做向量数据库存储的方案当前有很多，有FAISS（Meta公司的），Milvus，还有一个轻量级、开箱即用的适合小型项目的Chroma。增加几条语句，就可以将这些向量保存到Chroma里，内部采用Sqlite进行管理，非常方便。
import os
from openai import OpenAI
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_chroma import Chroma
# 阿里百炼 Embeddings 适配器
class DashScopeEmbeddings(Embeddings):
def __init__(
self,
api_key: str,
model: str = "text-embedding-v4",
dimensions: int = 1024,
):
self.client = OpenAI(
api_key=api_key,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
self.model = model
self.dimensions = dimensions
def embed_documents(self, texts: list[str]) -> list[list[float]]:
response = self.client.embeddings.create(
model=self.model,
input=texts,
dimensions=self.dimensions,
)
return [item.embedding for item in response.data]
def embed_query(self, text: str) -> list[float]:
response = self.client.embeddings.create(
model=self.model,
input=[text],
dimensions=self.dimensions,
)
return response.data[0].embedding
docs = [Document(page_content=input,)
for input in [
"每天吃一个鸡蛋可以补充优质蛋白质，但胆固醇偏高的人应注意摄入量。",
"长期不吃早餐可能会影响血糖稳定，也容易导致午餐时暴饮暴食。",
"多喝水有助于促进新陈代谢，但短时间内大量饮水也可能增加身体负担。",
"适量运动有助于控制体重和增强心肺功能，但运动强度应循序渐进。",
]]
def main():
embeddings = DashScopeEmbeddings(
api_key=os.getenv("DASHSCOPE_API_KEY"),
model="text-embedding-v4",
dimensions=1024,
)
vector_store = Chroma(
collection_name="daily_health_knowledge",
embedding_function=embeddings,
persist_directory="./health_chroma_db",
)
if vector_store._collection.count() == 0:
print("首次运行，写入中文健康常识向量库...\n")
vector_store.add_documents(docs)
else:
print("向量库已存在，直接查询...\n")
# 相似度查询示例
queries = [
"长期不吃早餐可能会影响？",
"不吃早餐有什么影响？",
"鸡蛋每天能不能多吃？",
"运动对身体有什么好处？",
"南京的天气怎么样？", # 明显无关查询
]
# < 0.3 高度相关
# 0.3 ~ 0.6 可能相关
# > 0.6 基本不相关（应视为“没命中”）
for query in queries:
print(f"Query：{query}")
results = vector_store.similarity_search_with_score(query, k=1)
for doc, score in results:
print(f'{score:.2f}, {doc.page_content}')
print("-" * 50)
if __name__ == "__main__":
main()
上面的程序会建立一个文件夹 health_chroma_db/，里面有sqlite3数据库，还有一些不知道用处的文件。我的查询结果：
Query：长期不吃早餐可能会影响？
0.43, 长期不吃早餐可能会影响血糖稳定，也容易导致午餐时暴饮暴食。
--------------------------------------------------
Query：不吃早餐有什么影响？
0.51, 长期不吃早餐可能会影响血糖稳定，也容易导致午餐时暴饮暴食。
--------------------------------------------------
Query：鸡蛋每天能不能多吃？
0.46, 每天吃一个鸡蛋可以补充优质蛋白质，但胆固醇偏高的人应注意摄入量。
--------------------------------------------------
Query：运动对身体有什么好处？
0.63, 适量运动有助于控制体重和增强心肺功能，但运动强度应循序渐进。
--------------------------------------------------
Query：南京的天气怎么样？
1.39, 多喝水有助于促进新陈代谢，但短时间内大量饮水也可能增加身体负担。
可以看出，比较相近的查询，返回的分数值应该小于0.7，而明显不相关的查询，会返回一个大小1.0的值，比如上面在一堆健康知识里询问南京的天气，分值是1.39。
基于PDF的问答程序
现在，可以把前面的逻辑拼在一起，完成一个基于PDF的问答程序，大概步骤：
1）读入PDF
2）分块。这里每块使用5000个字符，提高问题的命中率。
3）嵌入chroma向量库。这里发现一个问题，百炼一次只能嵌入10行文本，这里的PDF有19页，要分批处理。
4）将3条相似度最高的回答拼在一起，送给LLM大模型，加一点简单的提示词，就可以得到一个相对不错的回答。
import os
from openai import OpenAI
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_chroma import Chroma
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 阿里百炼 Embeddings 适配器
class DashScopeEmbeddings(Embeddings):
def __init__(
self,
api_key: str,
model: str = "text-embedding-v4",
dimensions: int = 1024,
):
self.client = OpenAI(
api_key=api_key,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
self.model = model
self.dimensions = dimensions
def embed_documents(self, texts: list[str]) -> list[list[float]]:
response = self.client.embeddings.create(
model=self.model,
input=texts,
dimensions=self.dimensions,
)
return [item.embedding for item in response.data]
def embed_query(self, text: str) -> list[float]:
response = self.client.embeddings.create(
model=self.model,
input=[text],
dimensions=self.dimensions,
)
return response.data[0].embedding
# 读入PDF
file_path = "how-to-fix-your-entire-life.pdf"
loader = PyPDFLoader(file_path)
docs = loader.load()
print("PDF的页数：", len(docs))
# 文本分块
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=5000, chunk_overlap=500, add_start_index=True
)
all_splits = text_splitter.split_documents(docs)
print("chunks: ", len(all_splits))
# 嵌入向量库
embeddings = DashScopeEmbeddings(
api_key=os.getenv("DASHSCOPE_API_KEY"),
model="text-embedding-v4",
dimensions=1024,
)
vector_store = Chroma(
collection_name="how_fix_life",
embedding_function=embeddings,
persist_directory="./how_fix_life_chroma_db",
)
if vector_store._collection.count() == 0:
print("首次运行，写入Chroma向量库...\n")
batch_size = 10 #阿里云百炼的文本嵌入，一次最多10行
for i in range(0, len(all_splits), batch_size):
batch = all_splits[i:i+batch_size]
vector_store.add_documents(batch)
else:
print("向量库已存在，直接查询...\n")
# 查询
llm = ChatOpenAI(
model="qwen-plus",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
queries = [
"human 3.0 model",
"anti-vision",
"take your life as video game",
"如何彻底改变你的人生", #原文是英文，这里直接使用中文
]
for query in queries:
print(f"Query：{query}")
results = vector_store.similarity_search_with_score(query, k=3) #取三近邻
for doc, score in results:
print('-'*10, f'{score:.2f}', '-'*10)
print(f'{doc.page_content[:100]} ... ')
# 把最相似的三个片段拼在一起，送给大模型
combined_text = " ".join([doc.page_content for doc, score in results])
print(f"问题查询：{query}")
response = llm.invoke(f'请使用中文回答这个问题：{query}，控制在500字左右，要严格基于下面文字素材： {combined_text}')
print(response.content)
print("=" * 50)
学习Langchain这几天，用了不到10万token。
