前言

Function Calling解决了大模型与Python工具之间的结构化通信问题，但模型仍然需要先找到与用户需求相关的商品。只使用WHERE product_name LIKE '%关键词%'可以处理“查询足球”这样的明确问题，却难以理解“我准备跑马拉松，需要透气并且能支撑足弓的装备”这类自然语言需求。

本文基于电商客服AI Agent的真实代码，使用text-embedding-v3、JSON向量索引、余弦相似度和中文关键词匹配实现轻量混合RAG。语义检索负责理解需求，关键词检索负责保留商品名和品牌等精确信号，两种分数融合后返回Top-K候选商品。

文章还会重点解释一个容易被忽略的边界：RAG结果只是候选，不是价格和库存的权威证据。候选商品必须继续通过SQLite按商品ID验证，才能进入最终回答。

---

---

项目效果与流程图

商品RAG检索流程。用户问题先转换成查询向量，再与商品索引进行混合评分，最终返回候选商品ID。

混合检索分数构成。存在向量索引时，语义分数占65%，关键词分数占35%。

商品向量索引结构。每条记录保存商品ID、检索文档和1024维向量。

检索与验证的边界。RAG负责召回候选，SQLite负责确认商品事实。

一、为什么商品查询需要RAG

传统名称查询通常是：

SELECT *
FROM products
WHERE product_name LIKE '%足球%';

它适合用户明确说出标准商品名，但存在三个限制：

• 用户可能只描述用途，不说商品名。
• 同一种需求可以有多种自然语言表达。
• 商品描述、规格和使用场景中的信息没有被充分利用。

例如用户提问：

我想参加长距离跑步，需要透气并且能支撑足弓的产品。

数据库中的标准名称是“跑步鞋”。如果只匹配用户原句中的商品名称，查询可能没有结果。Embedding可以把用户需求和商品描述转换到相同的向量空间，使语义相近的文本获得较高相似度。

二、RAG的三个阶段

RAG是Retrieval-Augmented Generation，即检索增强生成。

Retrieval：检索相关商品
↓
Augmentation：把检索结果加入模型上下文
↓
Generation：模型基于上下文生成回答

本项目进一步增加SQL验证：

用户问题
→ 混合RAG召回候选商品ID
→ SQLite按ID查询权威记录
→ 模型基于验证结果回答

因此，项目中的RAG并不直接承担最终事实输出。

三、RAG相关目录结构

function_calling_rag_agent/
│
├── config.py
├── llm_client.py
├── database.py
│
├── data/
│   ├── products.db
│   └── product_embeddings.json
│
├── rag/
│   ├── __init__.py
│   ├── retriever.py
│   └── build_index.py
│
├── tools/
│   └── product_tools.py
│
└── tests/
    └── test_retriever.py

文件职责如下：

文件	职责
config.py	保存数据库、索引路径和Embedding配置
llm_client.py	调用百炼Embedding接口
retriever.py	构建索引、计算相似度和混合评分
build_index.py	执行真实向量索引构建
product_tools.py	将检索器包装为Function Calling工具
test_retriever.py	使用假向量离线验证检索流程

四、Embedding配置

项目配置：

from dataclasses import dataclass
from pathlib import Path


PROJECT_ROOT = Path(__file__).resolve().parent
DATABASE_PATH = PROJECT_ROOT / "data" / "products.db"
INDEX_PATH = PROJECT_ROOT / "data" / "product_embeddings.json"


@dataclass(frozen=True)
class Settings:
    base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1"
    chat_model: str = "qwen-plus"
    embedding_model: str = "text-embedding-v3"
    embedding_dimensions: int = 1024
    max_iterations: int = 10
    top_k: int = 3


SETTINGS = Settings()

embedding_model

text-embedding-v3

负责把文本转换成向量。

embedding_dimensions

1024

每段商品文本会转换为长度1024的浮点数列表。向量维度不是商品数量，也不是词语数量，而是模型用于表达文本语义特征的数值空间。

top_k

默认返回3个候选商品。候选太少可能漏掉相关结果，太多则会增加后续SQL查询和模型上下文内容。

五、调用Embedding接口

def embed_texts(client: OpenAI, texts: list[str]) -> list[list[float]]:
    response = client.embeddings.create(
        model=SETTINGS.embedding_model,
        input=texts,
        dimensions=SETTINGS.embedding_dimensions,
    )
    return [item.embedding for item in response.data]

输入

texts: list[str]

可以一次传入多条商品文档，减少逐条请求造成的网络往返。

返回值

list[list[float]]

外层列表对应输入文本，内层列表是每条文本的1024维向量。

调试提示

建议在返回前观察：

len(texts)
len(response.data)
len(response.data[0].embedding)

正常情况下，输入数量与返回数量一致，单条向量长度为1024。不要打印全部向量，否则控制台会出现大量浮点数。

六、商品如何转换为检索文档

product_document()把SQLite的一条商品记录拼成自然语言：

def product_document(product: dict) -> str:
    return (
        f"商品ID：{product['product_id']}；"
        f"商品名称：{product['product_name']}；"
        f"品牌：{product['brand']}；"
        f"描述：{product['description']}；"
        f"规格：{product['specifications']}；"
        f"适用场景：{product['usage']}"
    )

价格和库存没有放进检索文档。原因是价格和库存属于可能变化的业务事实，最终回答必须重新查询SQLite，而不是依赖旧向量索引中的文本。

以跑步鞋为例，文档类似：

商品ID：004；商品名称：跑步鞋；品牌：阿迪达斯；
描述：适合长距离跑步，舒适透气，提供良好的足弓支撑；
规格：多种尺码，透气网布；适用场景：长跑、日常训练

文档字段越贴近用户真实表达，语义检索越容易召回相关商品。

七、从SQLite读取商品

def load_products() -> list[dict]:
    with connect() as connection:
        rows = connection.execute(
            "SELECT * FROM products ORDER BY product_id"
        ).fetchall()
    return rows_to_dicts(rows)

返回结构：

[
    {
        "product_id": "004",
        "product_name": "跑步鞋",
        "description": "适合长距离跑步...",
        "specifications": "多种尺码，透气网布",
        "usage": "长跑、日常训练",
        "brand": "阿迪达斯",
        "price": 500.0,
        "stock_quantity": 20
    }
]

检索器读取全部商品用于构建轻量索引。当前只有10条数据，这种方式足够直观；商品数量很大时，应考虑增量索引和专业检索服务。

八、构建商品向量索引

def save_embedding_index(embedder, index_path=INDEX_PATH) -> int:
    products = load_products()
    documents = [product_document(product) for product in products]
    vectors = embedder(documents)

    payload = [
        {
            "product_id": product["product_id"],
            "document": document,
            "embedding": vector,
        }
        for product, document, vector in zip(
            products,
            documents,
            vectors,
            strict=True
        )
    ]

    index_path.parent.mkdir(parents=True, exist_ok=True)
    index_path.write_text(
        json.dumps(payload, ensure_ascii=False),
        encoding="utf-8"
    )
    return len(payload)

zip(..., strict=True)要求商品、文档和向量数量完全一致。如果Embedding接口少返回一条，程序会及时报错，避免商品ID与错误向量对应。

索引最终保存为JSON：

[
  {
    "product_id": "004",
    "document": "商品ID：004；商品名称：跑步鞋...",
    "embedding": [0.012, -0.038, 0.104]
  }
]

示例只展示少量数值，真实向量包含1024个浮点数。

九、运行索引构建脚本

rag/build_index.py：

from llm_client import create_client, embed_texts
from rag.retriever import save_embedding_index


if __name__ == "__main__":
    client = create_client()
    count = save_embedding_index(
        lambda texts: embed_texts(client, texts)
    )
    print(f"向量索引构建完成，共写入 {count} 条商品记录。")

运行：

python rag/build_index.py

项目实际构建结果为：

向量索引构建完成，共写入10条商品记录。

修改商品名称、描述、规格、品牌或用途后，需要重新构建索引。只修改价格或库存时不需要，因为这两个字段不在检索文档中。

十、余弦相似度

def _cosine_similarity(left: list[float], right: list[float]) -> float:
    numerator = sum(
        a * b for a, b in zip(left, right, strict=True)
    )
    left_norm = math.sqrt(sum(value * value for value in left))
    right_norm = math.sqrt(sum(value * value for value in right))

    if left_norm == 0 or right_norm == 0:
        return 0.0

    return numerator / (left_norm * right_norm)

公式为：

cosine(A, B) = A·B / (||A|| × ||B||)

通俗理解：它比较两个向量的方向是否接近。用户问题与某个商品文档语义越接近，余弦相似度通常越高。

当前项目使用Python标准库计算，没有依赖NumPy。数据量只有10条时性能足够；大规模商品库不适合逐条Python计算。

十一、中文关键词评分

def _lexical_score(query: str, document: str) -> float:
    normalized_query = re.sub(r"\s+", "", query.lower())
    normalized_document = re.sub(r"\s+", "", document.lower())

    if not normalized_query:
        return 0.0

    tokens = set(normalized_query)
    tokens.update(
        normalized_query[index:index + 2]
        for index in range(len(normalized_query) - 1)
    )

    matches = sum(
        1 for token in tokens
        if token and token in normalized_document
    )
    exact_bonus = 2 if normalized_query in normalized_document else 0

    return min(
        1.0,
        (matches + exact_bonus) / max(len(tokens), 1)
    )

这段代码生成两类Token：

• 单个字符，例如“跑”“步”“鞋”。
• 连续两个字符，例如“跑步”“步鞋”。

二元词组比单字更能保留中文局部语义。若完整查询直接出现在商品文档中，再增加精确匹配奖励。

这不是专业中文分词器，而是适合小型教学项目的轻量实现。

十二、混合检索核心

class HybridProductRetriever:
    def __init__(self, embedder=None, index_path=INDEX_PATH):
        self.embedder = embedder
        self.index_path = index_path

    def search(self, query: str, top_k: int = 3) -> list[dict]:
        products = load_products()
        semantic_scores = {}

        if self.embedder and self.index_path.exists():
            index = json.loads(
                self.index_path.read_text(encoding="utf-8")
            )
            query_vector = self.embedder([query])[0]
            semantic_scores = {
                item["product_id"]: _cosine_similarity(
                    query_vector,
                    item["embedding"]
                )
                for item in index
            }

        results = []
        for product in products:
            document = product_document(product)
            lexical = _lexical_score(query, document)
            semantic = semantic_scores.get(product["product_id"], 0.0)

            combined = (
                0.65 * semantic + 0.35 * lexical
                if semantic_scores
                else lexical
            )

            if combined > 0:
                results.append(
                    {
                        "product_id": product["product_id"],
                        "product_name": product["product_name"],
                        "retrieval_score": round(combined, 4),
                        "matched_document": document,
                        "is_verified_fact": False,
                    }
                )

        results.sort(
            key=lambda item: item["retrieval_score"],
            reverse=True
        )
        return results[:max(1, min(top_k, 10))]

有向量索引时

combined = 0.65 × semantic + 0.35 × lexical

语义相似度权重更高，关键词分数用于保留精确名称和品牌信号。

没有向量索引时

combined = lexical

检索器自动降级为关键词模式，基本查询仍然可用，但自然语言需求理解能力会下降。

top_k限制

results[:max(1, min(top_k, 10))]

保证至少返回1条、最多返回10条，防止模型传入异常值导致大量候选进入上下文。

十三、为什么is_verified_fact是False

候选结果包含：

"is_verified_fact": False

它明确提醒模型和程序：

这条记录只是检索候选
不等于价格、库存、品牌已经验证

search_products工具还会返回警告：

{
    "warning": (
        "候选结果不是权威事实，"
        "必须用get_product_details进行SQL验证。"
    )
}

RAG负责找到相关商品，下一步必须根据product_id查询SQLite。

十四、将检索器包装为工具

def search_products(
    retriever: HybridProductRetriever,
    query: str,
    top_k: int = 3
) -> dict:
    candidates = retriever.search(query=query, top_k=top_k)

    return {
        "status": (
            "candidates_found" if candidates else "not_found"
        ),
        "source": "hybrid_rag_index",
        "warning": (
            "候选结果不是权威事实，"
            "必须用get_product_details进行SQL验证。"
        ),
        "candidates": candidates,
    }

Function Calling模型只接触这个工具返回的JSON，不需要了解余弦相似度和索引文件的内部实现。

十五、离线假向量测试

真实Embedding接口需要网络和API。测试使用确定性假向量：

def fake_embedder(texts: list[str]) -> list[list[float]]:
    vectors = []
    for text in texts:
        vectors.append(
            [
                float("长跑" in text or "跑步" in text),
                float("足球" in text),
                float("瑜伽" in text),
            ]
        )
    return vectors

测试代码：

def test_semantic_retrieval_returns_running_shoes(self):
    with tempfile.TemporaryDirectory() as directory:
        index_path = Path(directory) / "index.json"
        save_embedding_index(fake_embedder, index_path)

        retriever = HybridProductRetriever(
            fake_embedder,
            index_path
        )
        results = retriever.search(
            "我想长跑，需要透气和足弓支撑",
            top_k=1
        )

        self.assertEqual(results[0]["product_id"], "004")
        self.assertFalse(results[0]["is_verified_fact"])

该测试验证：

• 自然语言需求能够召回跑步鞋。
• Top-1商品ID为004。
• 检索结果仍被标记为未验证事实。

它不证明真实Embedding模型的准确率，也没有计算召回率等指标。

十六、完整运行步骤

1. 进入新版项目目录。

cd "D:\Jupyter_Projects\PythonProject\大模型学习\AI Agent\function_calling_rag_agent"

2. 初始化数据库。

python scripts/initialize_database.py

3. 构建真实向量索引。

python rag/build_index.py

4. 运行离线检索测试。

python -m unittest tests.test_retriever -v

5. 启动完整客服Agent。

python main.py

6. 输入语义需求。

我想进行长距离跑步，需要透气并且能支撑足弓的商品。

十七、断点调试建议

推荐断点：

1. documents = [...]：查看商品检索文档。
2. vectors = embedder(documents)：检查向量数量。
3. query_vector = self.embedder([query])[0]：检查查询向量。
4. _cosine_similarity(...)：比较单个候选分数。
5. combined = ...：观察两种分数融合。
6. results.sort(...)：查看排序前后结果。

关键变量：

query
documents
query_vector
semantic_scores
lexical
semantic
combined
results

不要在调试器中展开全部1024维向量，优先观察长度、前5个数值和最终相似度。

十八、常见问题与解决方法

1. 找不到product_embeddings.json

原因：尚未运行索引构建脚本，或索引路径错误。

解决：运行python rag/build_index.py，并检查INDEX_PATH。

2. 没有索引时程序是否完全不可用

原因：误以为RAG只能依赖向量。

解决：当前代码会降级到关键词评分，但语义召回能力会下降。

3. Embedding接口连接失败

原因：网络权限、接口地址或系统环境变量存在问题。

解决：检查网络和DASHSCOPE_API_KEY，不要打印密钥。

4. 向量维度不是1024

原因：模型配置、dimensions参数或旧索引不一致。

解决：确认配置后重新构建完整索引。

5. 商品数量和向量数量不一致

原因：Embedding返回不完整或构建过程中数据变化。

解决：zip(strict=True)会报错，应重新构建而不是忽略差异。

6. 精确商品名排序不靠前

原因：语义分数占比较高，关键词特征不足。

解决：检查lexical分数，按真实测试调整权重，不要凭感觉修改。

7. 语义需求召回错误商品

原因：商品描述过短、字段缺失或需求表达模糊。

解决：完善描述、规格和适用场景，并建立带标准答案的评测集合。

8. top_k传入0或很大数值

原因：模型参数可能异常。

解决：当前代码将结果限制在1～10条，Schema也定义相同范围。

9. JSON索引文件过大

原因：商品数量和向量维度增加。

解决：当前方案适合小数据教学；大规模场景需要专业向量存储。

10. 修改价格后是否需要重建索引

原因：不清楚哪些字段进入检索文档。

解决：价格和库存未写入检索文档，无需重建；描述、规格等变化需要重建。

11. RAG结果能否直接回答库存

原因：混淆候选召回和事实验证。

解决：不能。必须调用get_product_details查询SQLite。

12. 测试通过是否代表真实模型准确

原因：离线测试使用了人为设计的三维假向量。

解决：它只验证程序流程；真实效果需要单独构建查询集和召回指标。

十九、性能与工程优化

1. 商品数据量增加时改为增量索引。
2. 批量调用Embedding，减少网络往返。
3. 缓存高频查询向量。
4. 为索引增加版本和生成时间元数据。
5. 商品更新后只重建受影响记录。
6. 建立真实查询—商品ID评测集合。
7. 使用Recall@K、MRR等检索指标评估权重。
8. 大规模数据迁移到专业向量数据库。
9. 对索引文件进行完整性校验。
10. 把检索耗时和候选分数写入调试日志。

这些属于后续工程化方向，当前项目没有实现相应性能指标或专业向量数据库。

二十、总结

本文完成了一个轻量商品混合RAG检索器。项目将商品名称、品牌、描述、规格和适用场景组成检索文档，使用text-embedding-v3生成1024维向量，并将向量保存到JSON索引。

查询时，用户问题同样转换为向量，程序使用余弦相似度得到语义分数，同时通过中文字符和二元词组计算关键词分数。存在向量索引时，最终分数由65%语义分数和35%关键词分数组成；索引不可用时则降级为关键词检索。

最重要的设计边界是：RAG只返回候选商品ID，并明确标记is_verified_fact=False。价格、库存、品牌和优惠必须继续通过SQLite验证。下一篇将分析SQL复核、Decimal价格计算、EvidenceLedger和最终回答拦截如何共同降低模型幻觉。