03 Chroma_向量化：Qwen模型的丝滑接入

💡 一句话核心概念

Embedding 是把"文字的意思"翻译成"一串数字"，Chroma 负责存和搜，Qwen 负责翻译。你不需要懂向量数学——你只需要知道怎么把 Qwen 接进来。

---

🧩 关键实操

1. 安装 Qwen 相关依赖

uv add openai  # Qwen 的 API 兼容 OpenAI 格式，省掉额外 SDK

对，你没看错——连 dashscope 都不用装。阿里云 DashScope 完全兼容 OpenAI 的 /v1/embeddings 接口，直接用 openai 库调用。

2. 封装一个 Qwen Embedding 函数

# 03_qwen_embedding.py
from openai import OpenAI
from chromadb import Client
from chromadb.api.types import EmbeddingFunction, Embeddings
import os

# ===== 第一步：封装 Qwen Embedding =====
class QwenEmbedding(EmbeddingFunction):
    """
    把 Qwen 的 Embedding API 包装成 Chroma 认识的格式。
    原理简单：Chroma 调你 → 你调 DashScope → 返回向量列表 → Chroma 存起来
    """

    def __init__(self, api_key: str | None = None, model: str = "text-embedding-v4"):
        # DashScope 兼容 OpenAI 格式的 base_url
        self.client = OpenAI(
            api_key=api_key or os.getenv("DASHSCOPE_API_KEY"),
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        )
        self.model = model

    def __call__(self, input: list[str]) -> Embeddings:
        """
        入参：文档列表 ["文本1", "文本2", ...]
        出参：向量列表 [[0.1, 0.2, ...], [0.3, 0.4, ...]]
        ↓ 为什么是 list[str] → Embeddings？
        因为 Chroma 批量调用你一次，传入一批文本，你返回一批向量——减少网络开销。
        """
        if not input:
            return []

        response = self.client.embeddings.create(
            model=self.model,
            input=input,
            # text-embedding-v4 最多一次传 25 条，生产环境注意分片
        )

        # 按输入顺序排列（API 保证顺序一致，但显式 sort 一下不亏）
        return [item.embedding for item in response.data]


# ===== 第二步：用自定义 Embedding 创建 Collection =====
embed_fn = QwenEmbedding(
    api_key=os.getenv("DASHSCOPE_API_KEY"),  # 你的 DashScope Key
)

client = Client()
collection = client.create_collection(
    name="qwen_powered_kb",
    embedding_function=embed_fn,  # ← 关键！告诉 Chroma 用 Qwen 做向量化
    metadata={"hnsw:space": "cosine"},
)

# ===== 第三步：添加文档——Chroma 会自动调 QwenEmbedding =====
collection.add(
    documents=[
        "Transformer 架构的核心是 Self-Attention 机制。",
        "RAG 是 Retrieval-Augmented Generation 的缩写，结合了检索和生成。",
        "Chroma 是一个开源的向量数据库，专为 AI 应用设计。",
    ],
    ids=["transformer", "rag", "chroma"],
)

print(f"✅ 集合已有 {collection.count()} 条文档（由 Qwen text-embedding-v4 向量化）")

# ===== 第四步：查询 =====
results = collection.query(
    query_texts=["什么是自注意力机制？"],
    n_results=2,
)

for i, (doc_id, doc, dist) in enumerate(zip(
    results["ids"][0], results["documents"][0], results["distances"][0]
)):
    print(f"  #{i+1} [{doc_id}] 距离={dist:.4f} → {doc}")

# 运行前确保环境变量有 DASHSCOPE_API_KEY
# Windows PowerShell:
$env:DASHSCOPE_API_KEY="sk-your-key-here"

# 运行
uv run python 03_qwen_embedding.py

3. 使用本地 Qwen 模型（不想花钱调 API？）

# 03_qwen_local.py —— 用 sentence-transformers 加载本地 Qwen 模型
# 先安装：uv add sentence-transformers

from chromadb import Client
from chromadb.utils.embedding_functions import SentenceTransformerEmbeddingFunction

# Chroma 内置了 SentenceTransformer 的适配器
# Qwen 有开源 Embedding 模型，比如 qwen2.5-7b-instruct 也能做 embedding
# 但推荐用专门的 embedding 模型，小而快
embed_fn = SentenceTransformerEmbeddingFunction(
    model_name="Qwen/Qwen2.5-0.5B-Instruct",  # 最小的 Qwen，足够做 demo
    # 注意：这是个 instruct 模型不是专门的 embedding 模型
    # 生产环境建议用 Qwen/Qwen2.5-7B 或阿里云的 text-embedding-v4 API
)

client = Client()
collection = client.create_collection(
    name="local_qwen_kb",
    embedding_function=embed_fn,
)

# 后面的操作完全一样...
print("✅ 本地 Qwen Embedding 已就绪")

⚠️ 本地跑 embedding 模型需要 GPU 或耐心。0.5B 模型在 CPU 上大概每秒处理 5-10 条短文本，做 demo 够用。生产环境还是老老实实调 API。

---

🚧 避坑指南

坑	现象	解法
API Key 没设	openai.AuthenticationError	检查环境变量 DASHSCOPE_API_KEY，或在代码里显式传入
Embedding 维度不一致	写入时报 dimension mismatch	text-embedding-v4 是 1024 维，如果你之前用默认的 all-MiniLM-L6-v2（384 维），维度不匹配。要么重建集合，要么始终用同一个 embedding 函数
本地模型太大	下载半天，内存爆了	做 demo 用 0.5B 模型，生产用 API。别跟自己电脑过不去

---

🎤 Chroma 面试题与通关答案

Q1：Chroma 的 EmbeddingFunction 接口设计为什么是 __call__(input: list[str]) -> Embeddings 而不是单条处理？

考点拆解： 接口设计模式、批量处理优化思维。

通关答案：

设计原因：「批量处理」是向量数据库的生命线。

1. 网络开销： 一次 API 调用的延迟 ≈ 50-200ms，如果一条一条传，1000 条文档要 50-200 秒；一次传 25 条，只需要 40 次调用 ≈ 2-8 秒——40 倍差距。
2. GPU 利用率： Embedding 模型在 GPU 上跑时，batch inference 能充分利用并行计算。单条进入 GPU，90% 的算力都在摸鱼。
3. 接口契约： Chroma 保证传给你的是 batch，你的职责是一次性返回对应数量的向量。这个契约让 Chroma 可以内部做并发控制和重试，不需要你的 EmbeddingFunction 操心。

源码视角：

# chromadb/api/types.py 中的定义
EmbeddingFunction = Callable[[list[str]], Embeddings]
# Embeddings = list[list[float]]

就是个简单的 Callable，Python 的 duck typing 设计哲学：不搞抽象类继承，只要你的对象能 __call__ 就行。

一句话总结： 批量接口设计不是 Chroma 多事——是向量计算本身要求批量，单条处理等于让 GPU 等你打字。

---

Q2：Qwen 的 text-embedding-v4 和开源的 Qwen2.5-7B 做 Embedding 有什么区别？怎么选？

考点拆解： API vs 本地部署的工程权衡，面试官想看你有没有生产经验。

通关答案：

维度	text-embedding-v4 (API)	Qwen2.5-7B (本地)
维度	1024 维	3584 维（LLM 的 hidden_size）
速度	快（云端 GPU，免运维）	取决于你的硬件
成本	按 token 计费（~0.0005 元/1k tokens）	一次性硬件投入
质量	专门为 Embedding 训练，MTEB 榜单 SOTA	是个聊天模型，Embedding 是副业
适用场景	生产环境、快速迭代	离线环境、数据安全要求高

💀 致命坑点： 不要拿 LLM（如 Qwen2.5-7B-Instruct）直接做 Embedding！LLM 是 causal decoder，取 last hidden state 做 embedding 效果远不如专门的 embedding 模型。阿里云把 embedding 模型单独训练了一个版本（text-embedding-v4），专门优化了语义表示的区分度。

最佳实践：

开发/测试 → 本地小模型（Qwen2.5-0.5B）快速验证
生产环境 → text-embedding-v4 API（稳定、高效、免运维）
涉密场景 → 私有化部署专门的 Embedding 模型

一句话总结： 能调 API 就别自己部署——你省下来的时间比那点 API 费用值钱一万倍。

---

Q3：如果用 Qwen 做 RAG 的 Embedding 和生成，两个环节的模型怎么搭配？

考点拆解： RAG 架构中的模型选型与一致性原则。

通关答案：

核心原则：Embedding 和生成是两个独立环节，不需要同一家族的模型。

[用户问题]
    ↓
[Qwen text-embedding-v4] ← 把问题向量化
    ↓
[Chroma 检索] ← 找到相关文档
    ↓
[拼接 Prompt: 文档 + 问题]
    ↓
[Qwen2.5-7B-Instruct / qwen-max] ← 用检索到的上下文生成答案

为什么可以混搭？

Embedding 模型只负责"理解语义，找到相关段落"，生成模型只负责"看了段落，写出回答"。两个环节之间传递的是自然语言文本，不是向量——没有"向量空间必须一致"的限制。

推荐组合：

# Embedding：便宜、快、专门优化
embedding_model = "text-embedding-v4"  # 1024 维，MTEB 高分

# 生成：聪明、上下文窗口大
generation_model = "qwen3.6-max-preview"  # 1M 上下文，够你把整个知识库塞进去

# 或者省钱组合：
generation_model = "qwen2.5-7b-instruct"  # 本地跑，零 API 费用

一句话总结： Embedding 用专门模型（text-embedding-v4），生成用大模型（qwen-max）——各司其职，专业的事交给专业的模型。

---