阶段二：导入流程 — AI 处理链路（对应课程 day06~07）

当前状态：文档处理链路已就绪（6 节点）→ 新增 3 个 AI 处理节点

知识标记总览：

知识点	出现次数	扮演的角色
LLM 客户端封装（缓存 + 重试）	🟢 第 2 次（Ch16→Ch19）	商品名识别 + 批量调用
bge-m3 嵌入模型	🔵 第 3 次（Ch16→Ch17→Ch19）	稠密+稀疏双向量嵌入
Milvus 集合操作	🔵 第 3 次（Ch16→Ch17→Ch19）	创建 Schema / Index / 插入
Milvus 标量字段索引坑	🔴 第 1 次（本项目的坑）	V2.5→V2.6 迁移时的兼容问题

2.1 阶段起始状态：6 个灰色节点

2.2 本阶段新增文件

shopkeeper_brain/knowledge/processor/import_process/nodes/
├── item_name_recognition_node.py    ← 🔴 新增：商品名识别
├── bge_embedding_chunks_node.py     ← 🔴 新增：批量嵌入
├── import_milvus_node.py            ← 🔴 新增：Milvus 入库

shopkeeper_brain/knowledge/utils/
├── bge_m3_embedding_util.py         ← 🔴 新增：bge-m3 工具类
├── milvus_util.py                   ← 🔴 新增：Milvus 工具类
├── llm_client_util.py               ← 🔴 新增：LLM 客户端

2.3 核心代码骨架

① LLM 客户端工具类（🟢 第 2 次，Ch16→Ch19）

python

# → knowledge/utils/llm_client_util.py
import json, time
from openai import OpenAI

class LLMClient:
    """带缓存和重试的 LLM 客户端"""
    
    def __init__(self, config):
        self.client = OpenAI(
            base_url=config.openai_api_base,
            api_key=config.openai_api_key
        )
        self.cache = {}  # 简单内存缓存
    
    def chat(self, messages, model=None, cache_key=None):
        """带缓存的 LLM 调用"""
        if cache_key and cache_key in self.cache:
            return self.cache[cache_key]
        
        for attempt in range(3):
            try:
                resp = self.client.chat.completions.create(
                    model=model or self.config.default_model,
                    messages=messages
                )
                result = resp.choices[0].message.content
                if cache_key:
                    self.cache[cache_key] = result
                return result
            except Exception as e:
                if attempt == 2: raise
                time.sleep(2 ** attempt)  # 指数退避

② ItemNameRecognitionNode：商品名识别（🟢 LLM + 🔵 Milvus 第 3 次）

python

class ItemNameRecognitionNode(BaseNode):
    """从文档内容中提取商品名称并存入 Milvus"""
    
    name = "item_name_rec"

    def process(self, state: ImportGraphState) -> ImportGraphState:
        chunks = state.get('chunks', [])
        llm_client = LLMClient(self.config)
        
        # 1. 取前 N 个 chunks 作为 LLM 上下文
        context = "\n".join(
            c['content'][:self.config.item_name_chunk_size] 
            for c in chunks[:self.config.item_name_chunk_k]
        )
        
        # 2. LLM 提取商品名
        prompt = f"从以下文档中提取商品/产品名称，以 JSON 列表返回：\n{context}"
        response = llm_client.chat([
            {"role": "system", "content": "你是一个商品名提取器，只输出 JSON 数组。"},
            {"role": "user", "content": prompt}
        ], cache_key=f"item_name_{state['file_title']}")
        item_names = json.loads(response)
        
        # 3. 计算商品名向量并存入 Milvus
        embed_util = BGEM3EmbeddingUtil(self.config)
        milvus_util = MilvusUtil(self.config)
        
        for name in item_names:
            vectors = embed_util.encode(name)  # 稠密+稀疏
            milvus_util.insert(
                collection="item_name",
                data=[{"text": name, **vectors}]
            )
        
        state['item_name'] = item_names
        return state

③ BGEM3EmbeddingUtil：bge-m3 嵌入工具类（🔵 第 3 次）

python

# → knowledge/utils/bge_m3_embedding_util.py
class BGEM3EmbeddingUtil:
    """bge-m3 嵌入工具——同时输出稠密+稀疏向量"""
    
    def __init__(self, config):
        from FlagEmbedding import BGEM3FlagModel
        self.model = BGEM3FlagModel("BAAI/bge-m3")
        self.dim = config.embedding_dim
    
    def encode(self, text: str) -> dict:
        result = self.model.encode([text], return_dense=True, return_sparse=True)
        return {
            "dense_vector": result["dense_vecs"][0].tolist(),
            "sparse_vector": result["lexical_weights"][0]
        }
    
    def encode_batch(self, texts: list) -> list:
        """批量嵌入（低显存场景下逐条执行）"""
        return [self.encode(t) for t in texts]

④ BgeEmbeddingChunksNode：批量嵌入节点

python

class BgeEmbeddingChunksNode(BaseNode):
    """对所有切片批量计算 bge-m3 嵌入向量"""
    
    name = "bge_embedding"

    def process(self, state: ImportGraphState) -> ImportGraphState:
        chunks = state.get('chunks', [])
        embed_util = BGEM3EmbeddingUtil(self.config)
        
        for chunk in chunks:
            vectors = embed_util.encode(chunk['content'])
            chunk['dense_vector'] = vectors['dense_vector']
            chunk['sparse_vector'] = vectors['sparse_vector']
        
        return state

⑤ ImportMilvusNode：Milvus 入库节点（🔵 第 3 次 + 🔴 标量索引坑）

python

class ImportMilvusNode(BaseNode):
    """将嵌入后的 chunks 存入 Milvus"""
    
    name = "import_milvus"

    def process(self, state: ImportGraphState) -> ImportGraphState:
        chunks = state.get('chunks', [])
        
        # 1. 创建集合（含 schema + 索引）
        milvus_util = MilvusUtil(self.config)
        collection_name = self.config.chunks_collection
        
        schema = self._build_schema()
        index_params = self._build_index()
        milvus_util.create_collection(collection_name, schema, index_params)
        
        # 2. 批量插入
        entities = []
        for chunk in chunks:
            entities.append({
                "text": chunk['content'],
                "title_path": chunk['title_path'],
                "vector": chunk['dense_vector'],
                "sparse_vector": chunk['sparse_vector'],
                "metadata": chunk['metadata']
            })
        
        milvus_util.insert(collection_name, entities)
        return state

    def _build_schema(self):
        from pymilvus import DataType
        schema = MilvusClient.create_schema(auto_id=True)
        schema.add_field("id", DataType.INT64, is_primary=True)
        schema.add_field("vector", DataType.FLOAT_VECTOR, dim=self.config.embedding_dim)
        schema.add_field("sparse_vector", DataType.SPARSE_FLOAT_VECTOR)
        schema.add_field("text", DataType.VARCHAR, max_length=5000)
        schema.add_field("title_path", DataType.VARCHAR, max_length=500)
        schema.add_field("metadata", DataType.JSON)
        return schema

⚠️ 🔴 标量字段索引坑点（Milvus V2.5 和 V2.6 的差异）：
V2.5：不需要对 VARCHAR 和 JSON 字段显式创建索引，自动映射
V2.6：对 VARCHAR 字段执行 LIKE / == 过滤时，需要手动创建 Inverted 类型索引，否则报错
解决方案：如果在 V2.6 上遇到标量过滤失败，添加 index_params.add_index("title_path", index_type="INVERTED")

⑥ MilvusUtil 工具类

python

# → knowledge/utils/milvus_util.py
from pymilvus import MilvusClient

class MilvusUtil:
    def __init__(self, config):
        self.client = MilvusClient(uri=config.milvus_url)
    
    def create_collection(self, name, schema, index_params):
        if self.client.has_collection(name):
            self.client.drop_collection(name)
        self.client.create_collection(name, schema=schema, index_params=index_params)
    
    def insert(self, collection, data):
        return self.client.insert(collection, data)
    
    def search(self, collection, vector, limit=5):
        return self.client.search(
            collection_name=collection,
            data=[vector],
            anns_field="vector",
            limit=limit,
            output_fields=["text", "title_path"]
        )

2.4 设计决策

决策	选项	理由
商品名识别用 LLM 而非规则	LLM / 正则	文档主题多样，规则无法覆盖所有命名模式
嵌入模型选 bge-m3 而非 text-embedding-3	bge-m3 / OpenAI	bge-m3 同时输出稠密+稀疏向量，支持混合检索
批量嵌入逐条而非批量	逐条 / 批量	逐条执行避免显存 OOM（低配置环境友好）
Milvus 集合 auto_id=True	auto_id / 手动 id	利用 Milvus 自增 ID，避免业务层维护 ID 映射

2.5 阶段结束时的演进架构图

本阶段已搭建的节点数：7/8（至此导入流程只差知识图谱） 本阶段新增的文件：6 个（3 node + 3 util） 新增的知识：🟢 LLM 客户端封装 / 🔵 bge-m3 第 3 次 / 🔵 Milvus 第 3 次 / 🔴 标量索引坑 下一阶段：Neo4J + GraphRAG（🔴 全新知识，重点展开）

📂 对应的原始代码快照：day06/ → day07/ 之间的逐日增量

阶段二：导入流程 — AI 处理链路（对应课程 day06~07） ​

2.1 阶段起始状态：6 个灰色节点 ​

2.2 本阶段新增文件 ​

2.3 核心代码骨架 ​

① LLM 客户端工具类（🟢 第 2 次，Ch16→Ch19） ​

② ItemNameRecognitionNode：商品名识别（🟢 LLM + 🔵 Milvus 第 3 次） ​

③ BGEM3EmbeddingUtil：bge-m3 嵌入工具类（🔵 第 3 次） ​

④ BgeEmbeddingChunksNode：批量嵌入节点 ​

⑤ ImportMilvusNode：Milvus 入库节点（🔵 第 3 次 + 🔴 标量索引坑） ​

⑥ MilvusUtil 工具类 ​

2.4 设计决策 ​

2.5 阶段结束时的演进架构图 ​

阶段二：导入流程 — AI 处理链路（对应课程 day06~07）

2.1 阶段起始状态：6 个灰色节点

2.2 本阶段新增文件

2.3 核心代码骨架

① LLM 客户端工具类（🟢 第 2 次，Ch16→Ch19）

② ItemNameRecognitionNode：商品名识别（🟢 LLM + 🔵 Milvus 第 3 次）

③ BGEM3EmbeddingUtil：bge-m3 嵌入工具类（🔵 第 3 次）

④ BgeEmbeddingChunksNode：批量嵌入节点

⑤ ImportMilvusNode：Milvus 入库节点（🔵 第 3 次 + 🔴 标量索引坑）

⑥ MilvusUtil 工具类

2.4 设计决策

2.5 阶段结束时的演进架构图