大模型项目：生成器与ChromaDB向量数据库
·
一. 生成器（LLM调用）
大模型（LLM）是 RAG 流程中的决策层/生成器，负责将检索到的碎片化信息转化为逻辑严密的语言回复。
生成器核心任务：在检索阶段获得的文档片段或信息的基础上，生成自然语言的回答、摘要或相关文本。简单说就是：把用户输入的问题与检索到的外部资源进行合并，然后传递给大语言模型进行最终的生成。
1. 非流式输出 — invoke
一次性返回完整的回复内容，适合不需要实时展示"打字效果"的场景。
from langchain_openai import ChatOpenAI
import os
# 使用ChatOpenAI加载千问的模型
chatLLM = ChatOpenAI(
# 通过环境变量中读取api-key
api_key=os.getenv("DASHSCOPE_API_KEY"),
# 指定基础服务地址 --- 阿里公司提供的模型访问地址
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
# 指定模型名称
model="qwen3.7-max-preview",
)
# 输入问题
question = input("请输入问题：\n")
# 消息列表
messages = [
# 系统角色描述 --- 可以不要
{"role": "system", "content": "You are a helpful assistant."},
# 用户消息 --- 用户问题
{"role": "user", "content": question}
]
# invoke方法：调用LLM生成回复、非流式输出结果
response = chatLLM.invoke(messages)
print("输出结果：\n")
# 打印输出结果
print(response.content)
关键点：
ChatOpenAI兼容 OpenAI 接口格式，可接入阿里千问等三方模型invoke()接收消息列表，返回完整的AIMessage对象- 通过
response.content获取文本内容 base_url指向阿里 DashScope 的兼容接口
os.getenv("DASHSCOPE_API_KEY") 详解：
如何将DASHSCOPE_API_KEY写入系统底层的环境变量？
以管理员身份运行cmd，执行：
setx DASHSCOPE_API_KEY "你的key"
这一步执行完后，要退出当前cmd窗口并且重启IDE，之后打开IDE环境的终端，执行：
echo %DASHSCOPE_API_KEY%
可以验证key是否写入。
若要重置key值，直接将key的位置写成空值：
setx DASHSCOPE_API_KEY ""
然后关注一下读取值所用的方法：
os.getenv(key, default=None)
os.getenv是 Python 内置模块os的方法，用于从操作系统环境变量中读取指定的值"DASHSCOPE_API_KEY"是环境变量的名称，对应阿里云 DashScope 服务的 API 密钥- 为什么用环境变量而不直接写死 API Key？
- 安全：API Key 是敏感凭据，写死在代码中会导致泄漏风险（尤其是在 Git 等版本控制系统中！！！一旦透露如同钱包外放到世界窗口）
- 灵活：不同环境（开发 / 测试 / 生产）可以配置不同的 Key，通过写到底层直接调用DASHSCOPE_API_KEY，无需修改代码
- 规范：遵循 12-Factor App 原则——配置与代码分离
使用方式：在终端中通过
export DASHSCOPE_API_KEY="sk-xxxx"（Linux/Mac）或set DASHSCOPE_API_KEY=sk-xxxx（Windows cmd）设置环境变量后，即可运行代码。
2. 流式输出 — stream
逐 token 返回内容，实现"打字机"效果，用户体验更好。
from langchain_openai import ChatOpenAI
import os
# 使用ChatOpenAI加载千问的模型
chatLLM = ChatOpenAI(
# 通过环境变量中读取api-key
api_key=os.getenv("DASHSCOPE_API_KEY"),
# 指定基础服务地址 --- 阿里公司提供的模型访问地址
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
# 指定模型名称
model="qwen3.7-max-preview",
# 流式输出
streaming=True,
)
# 输入问题
question = input("请输入问题：\n")
# 消息列表
messages = [
# 系统角色描述 --- 可以不要
{"role": "system", "content": "请用一句话回复用户的问题"},
# 用户消息 --- 用户问题
{"role": "user", "content": question}
]
# stream方法：调用LLM生成回复、流式输出结果
for chunk in chatLLM.stream(messages):
print(chunk.content, end="", flush=True)
关键点：
- 初始化时设置
streaming=True - 使用
stream()替代invoke()，返回迭代器 flush=True确保逐字刷新缓冲区- 可设置
system角色约束回复格式（如"请用一句话回复"）
chatLLM.stream(messages) 详解：
stream() 是 ChatOpenAI 提供的流式调用方法，与一次性返回的 invoke() 对应：
| 对比 | invoke() | stream() |
|---|---|---|
| 返回方式 | 等模型生成完整回复后一次性返回 | 边生成边返回，逐 token 推送 |
| 返回值类型 | AIMessage 对象 | Iterator[StreamingChunk]（迭代器） |
| 用户体验 | 需要等待，有"卡住"的感觉 | 实时显示，如真人打字 |
| 适用场景 | 后台处理、无需交互的批处理 | 对话机器人、实时问答 |
stream()返回一个生成器/迭代器，每轮for循环从迭代器中取出一个chunk- 每个
chunk是一个StreamingChunk对象，通过chunk.content获取该片段的文本增量 - 第一个 chunk 可能包含角色信息（如
role: "assistant"），后续 chunk 只包含content - 底层原理：LLM 生成文本时是逐个 token 预测的（如 “我” → “是” → “一” → “个” → …），
stream()就是每预测出一个 token 就立即返回，不等全部生成完
flush=True 详解：
print(chunk.content, end="", flush=True)
end=""：取消 print 默认的换行符，让后续内容在同一行继续输出flush=True：强制刷新输出缓冲区- 为什么需要 flush？
- Python 的
print()函数默认有输出缓冲机制——内容先写入内存缓冲区，等缓冲区满或遇到换行符才真正输出到终端 - 不加
flush=True时，多个print(end="")的片段可能累积在缓冲区，导致用户看到的是"一段一段跳出来"而非"一个字一个字流畅打出" flush=True告诉 Python 立即将缓冲区内容推送到终端，每个 token 到达后毫秒级显示
- Python 的
- 类比：就像水龙头 — 不开 flush 相当于先接满一桶水再倒出来（卡顿），开了 flush 相当于打开水龙头让水持续流出（流畅）
3. 带上下文的回复测试
RAG 的核心思想：将检索到的外部知识与用户问题拼接后送入 LLM，让模型基于参考资料回答，而非凭空生成。
from langchain_openai import ChatOpenAI
import os
# 使用ChatOpenAI加载千问的模型
chatLLM = ChatOpenAI(
# 通过环境变量中读取api-key
api_key=os.getenv("DASHSCOPE_API_KEY"),
# 指定基础服务地址 --- 阿里公司提供的模型访问地址
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
# 指定模型名称
model="qwen3.7-max-preview",
# 流式输出
streaming=True,
)
# 输入问题
question = input("请输入问题：\n")
# 模拟上下文 --- 后期来源于向量数据库中查询出来的结果
context = "cc是一名华清远见成都中心的AI讲师，他擅长用Python开发AI模型。"
# 拼接问题和上下文 --- 后期使用提示词来完成
content = "请根据上下文信息：\n" + context + "\n回答用户问题：\n" + question
# 消息列表
messages = [
# 系统角色描述 --- 可以不要
{"role": "system", "content": "请用一句话回复用户的问题"},
# 用户消息 --- 用户问题
{"role": "user", "content": content},
]
response = chatLLM.invoke(messages)
print(response.content)
关键点：
context模拟从向量数据库检索到的 top-k 相关文档（后期替换为真实检索结果）- 通过字符串拼接将上下文和问题组合（生产环境中使用 Prompt Template 结构化构建）
- 生成器工作流程：用户问题 → 向量数据库检索 top-k → 获取上下文 → 拼接送入 LLM → 输出回答
🤖 Prompt Engineering（提示工程）要点：
从 07.RAG核心组件.md 引入——生成器的 Prompt 设计是 RAG 效果的关键：
| 要素 | 说明 | 示例 |
|---|---|---|
| 角色定义 | 告诉模型以什么身份回答问题 | "你是一名专业的IT技术支持" |
| 约束条件 | 限制模型不要随意发挥 | "只能根据给定的参考资料回答，不要编造。" |
| 结构化输出 | 要求特定格式返回 | "请以Markdown格式回复" |
| 上下文引用 | 让模型明确信息来源 | "请基于以下资料回答用户的问题：\n{context}" |
优秀的 Prompt 能大幅减少"幻觉"（Hallucination），即模型编造不存在的信息。
二. ChromaDB 向量数据库
向量数据库（Vector Database） 是专门用于存储和检索高维向量数据的数据库系统。在 RAG 流程中，它的作用是：
- 存储文档的向量表示（embedding）
- 在用户提问时，快速找到与查询语义最接近的文档
为什么不能用传统数据库做检索？
| 功能项 | 关系型数据库（MySQL） | 向量数据库（ChromaDB） |
|---|---|---|
| 持久化 | 支持本地存储 | 支持本地存储 |
| 增删改查 | 支持 | 支持（侧重查） |
| 相似度计算 | ❌ 不支持 | ✅ 余弦相似度 / 内积 / 欧氏距离 |
| 数据存储 | 表/字段 | 集合/文档 + 向量 |
| 索引 | B+树索引 | 向量索引（HNSW、IVF等） |
| 数据类型 | 多样的结构化数据 | 字符和数值（向量） |
用户问题"我喜欢吃水果"如果用 MySQL 的
LIKE或=去匹配数据库，根本匹配不上"小明喜欢吃苹果"。而向量数据库把文本转为向量后，可以通过余弦相似度计算两句话的语义接近程度。
1. 客户端/服务器模式
生产环境中将 ChromaDB 作为独立服务运行，客户端通过 HTTP 连接。
"""
启动命令：
chroma run --path 数据存储路径 --host 0.0.0.0 --port 9000
--path：数据存储路径
--host：服务器地址
--port：服务器端口
"""
import chromadb
# 连接服务器 --- 操作任何数据库的第一步就是获取连接对象
client = chromadb.HttpClient(
host="localhost",
port=9000,
)
print(client)
关键点：
- 先启动服务端：
chroma run --path ./rag/chroma_data --host 0.0.0.0 --port 9000 - 客户端通过
HttpClient连接 - 适用于多机部署或需要持久化运行的场景
| 启动参数 | 说明 | 示例 |
|---|---|---|
--path | 最重要的参数，指定数据持久化路径 | ./rag/chroma_data |
--host | 服务器监听地址 | 0.0.0.0（允许所有IP访问） |
--port | 服务器端口 | 9000 |
--workers | 工作进程数（多核CPU提升并发） | 4 |
2. 本地持久化模式
最常用方案：数据存储在本地的指定目录，重启后自动加载。这是个人项目和单机应用的核心方式。
"""
必须掌握的方案
过程：设置一个文件夹用于存放数据
"""
import chromadb
"""
PersistentClient 参数：
1、path：数据库存储路径
"""
vector = chromadb.PersistentClient(
path="./chroma",
)
print(vector)
关键点：
PersistentClient(path="./chroma")在当前目录创建chroma文件夹存储数据- 必须掌握的方案，适合个人项目或单机应用
- 程序重启后数据不会丢失，自动加载之前的索引
| 连接模式 | 类 | 特点 |
|---|---|---|
| 内存模式 | EphemeralClient() | 数据仅存内存，关闭即消失，适合测试 |
| 本地持久化 | PersistentClient(path=...) | 数据存本地磁盘，适合个人项目 |
| 客户端/服务器 | HttpClient(host=..., port=...) | 连接独立服务，适合生产部署 |
3. 集合操作 — 查询集合
集合（Collection） 是 ChromaDB 存储数据的基本容器，类似关系数据库中的表。不同的项目数据应该存储在不同的集合中。
"""
集合：用于存储数据的容器，数据本质上就是保存在集合中
不同的项目的数据内容存储在不同的集合中
我们操作的内容就是集合
"""
import chromadb
# 获取连接对象
vector = chromadb.PersistentClient(path="./chroma")
# 查看数据库中所有的集合
lists = vector.list_collections()
print(f"所有的集合：{lists}")
# 判断某个集合是否存在
try:
exist = vector.get_collection(name="test01")
print("集合存在")
except Exception as e:
print(f"集合不存在：{e}")
关键点：
list_collections()→ 列出当前客户端中所有集合，返回集合对象列表，每个对象有.name属性get_collection(name=...)→ 获取指定名称的集合，不存在则抛异常try/except是判断集合是否存在的常用方式
4. 集合操作 — 创建集合
创建集合是使用向量数据库的第一步，需要指定集合名称和向量化模型（Embedding 模型）。
"""
创建集合，create_collection重要参数：
1、集合名称
2、向量化模型
"""
import chromadb
from chromadb.utils import embedding_functions
# 获取连接对象
vector = chromadb.PersistentClient("./chroma")
# 设置创建的集合名称
collection_name = "test01"
# 设置向量化模型 --- 目前只能选择这种方式，不考虑其他的加载方式
embedding_model = embedding_functions.SentenceTransformerEmbeddingFunction(
# 模型名称
model_name="thenlper/gte-small-zh",
# 运行设备
device="cuda",
# 设置缓存路径
cache_folder=r"G:\models\gte-small-zh"
)
# 创建集合
result = vector.create_collection(
# 集合名称
name=collection_name,
# 向量化模型
embedding_function=embedding_model
)
print(result)
print(type(result))
关键点：
create_collection核心参数：name：集合名称（必填，类似数据库表名，须唯一）embedding_function：向量化模型（创建时指定后，后续增删查操作无需再传该模型）
SentenceTransformerEmbeddingFunction封装了 HuggingFace 的 Sentence Transformer 模型- 模型
thenlper/gte-small-zh是针对中文优化的轻量级文本嵌入模型
| 参数 | 说明 | 示例 |
|---|---|---|
name | 集合名称（类似表名） | "test01" |
embedding_function | 向量化模型函数 | SentenceTransformerEmbeddingFunction(...) |
metadata | 集合元数据（可选） | {"hnsw:space": "cosine"} |
get_or_create | 存在则返回不报错（可选） | True |
📐 相似度度量详解
向量数据库能"理解"语义，靠的是数学上的向量相似度计算。最常用的有三种：
① 余弦相似度（Cosine Similarity） — RAG 最常用
值越大（越接近 1）表示两个文本语义越相似。
cos(θ) = (A·B) / (||A|| × ||B||)
= Σ(Ai × Bi) / √Σ(Ai²) × √Σ(Bi²)
② 欧氏距离（Euclidean Distance）
值越小表示两个向量越接近。ChromaDB 中一般用平方欧氏距离。
d = Σ(Ai - Bi)²
③ 内积（Dot Product）
值越大表示相关性越高。
直观理解："国王"和"男人"的向量在空间中离得近（余弦相似度高），"国王"和"狗"离得远（余弦相似度低）。
🔄 向量数据库完整工作流程（重点）
这是 RAG 中最核心的流程，请务必理解：
阶段一：数据入库（离线操作）
原始文档（txt/pdf/word等）
│
▼
加载器（Loader）→ 读取为 Document 对象
│
▼
文本分割器（Splitter）→ 按段落/长度切分为小片段
│
▼
Embedding 模型（如 gte-small-zh）→ 每个片段转为向量
│
▼
存入 ChromaDB 集合（向量 + 原文 + 元数据一起存储）
- 嵌入模型（Embedding Model）：将文本（如"华清远见成立于2004年"）转换为高维向量（如
[0.123, -0.456, 0.789, ...]），这个过程叫"向量化" - 入库是离线操作，即在开发 RAG 项目之前就应该构建好知识库
- 知识库的构建效果直接关系到后期检索的效果
阶段二：在线检索（用户提问时）
用户问题："介绍下华清远见"
│
▼
用同一个 Embedding 模型将问题转为向量
│
▼
在 ChromaDB 集合中计算问题向量与所有文档向量的余弦相似度
│
▼
返回相似度最高的 top-k 条文档片段（含原文内容和距离分数）
│
▼
将检索结果作为上下文 + 用户问题 → 送入 LLM 生成回复
- ⚠️ 关键原则：入库用的 Embedding 模型和检索时的 Embedding 模型必须是同一个，否则向量空间不一致，相似度计算毫无意义
- 索引优化：ChromaDB 内部使用 HNSW（分层小世界图）等索引算法加速检索，而非暴力遍历所有向量，实现毫秒级响应
检索增强技术
除了基础向量检索，还有以下高级技术可以显著提升 RAG 效果：
| 技术 | 作用 | 说明 |
|---|---|---|
| Query Rewrite（查询重写） | 让 LLM 将用户口语化问题转为更适合检索的关键词 | 用户说"那个做培训的机构怎么样" → 改写为"华清远见教育机构评价" |
| HyDE（假设性文档嵌入） | 先让 LLM 生成一个假设性答案，用该答案去检索 | 伪答案往往比原始问题更贴近目标文档的语义 |
| Rerank（重排序） | 🔥 最强提质手段 | 检索出 100 条后，用精排模型（如 BGE-Reranker）筛选出最相关的 Top-5，大幅提升准确率 |
检索器（Retriever） 的核心目标是提高"召回率"（该找到的都找到）和"准确率"（找到的都是相关的）。
总结 — Day2 知识串联
RAG 完整流程
┌─────────────────────────────────────────────────────┐
│ 离线阶段 │
│ 文档 → 加载器 → 文本分割 → Embedding → ChromaDB │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 在线阶段 │
│ │
│ 用户问题 │
│ │ │
│ ▼ │
│ 问题向量化（同一个 Embedding 模型） │
│ │ │
│ ▼ │
│ ChromaDB 相似度检索（余弦相似度 + HNSW 索引） │
│ │ │
│ ▼ │
│ 返回 top-k 相关上下文 │
│ │ │
│ ▼ │
│ Prompt 拼接（角色定义 + 约束条件 + 上下文 + 问题） │
│ │ │
│ ▼ │
│ LLM 生成回复 │
│ ├── invoke() 非流式 → 一次性返回完整内容 │
│ └── stream() 流式 → 逐 token 显示（打字效果） │
└─────────────────────────────────────────────────────┘
- 生成器负责"说话"：把检索到的碎片信息组织成完整的语言回复
- 向量数据库负责"记忆"：存储文档的向量表示，快速找到相关上下文
- 二者结合就是 RAG 的核心公式：检索（Retrieval）→ 增强（Augmented）→ 生成（Generation）
最终思考：今天模拟上下文用的是在代码中硬编码的字符串（
context = "cc是一名..."），明天学完完整的 ChromaDB 增删查操作后，就可以用collection.query(query_texts=[question])从向量数据库中真正检索出上下文，实现完整的 RAG 闭环！
更多推荐
6
0- 0
已为社区贡献10条内容
所有评论(0)