一篇代码能跑通、原理讲清楚、注释写到位的硬核实战文章写在前面你有没有遇到过这种情况在电商平台问客服“帮我找一款拍照好、续航长的手机”结果客服机器人给你推荐了一堆完全不相关的商品或者你问“苹果手机有哪些型号支持5G”它只给你返回一个“苹果手机”的链接传统的客服机器人大多基于关键词匹配或向量检索。关键词匹配只能识别你话里预设好的词换个说法就失效向量检索虽然能理解语义但它不擅长处理关系——比如“苹果手机”和“iPhone”是什么关系“拍照好”这个描述跟哪些商品有关今天我要带你实现的是一个基于知识图谱的智能电商客服系统。它不是简单地匹配关键词而是先把电商数据商品表、分类表、品牌表和商品描述文本里的特征标签比如“拍照旗舰”“续航持久”整理成一个图数据库然后让大模型理解你的问题自动去图里查询最后用自然语言回答你。这套方案代码完整、注释详细、可以直接运行。我保证你跟着一步步做不仅能跑起来还能真正理解里面的每一处设计。一、整体架构先看图再说话在动手写代码之前我们先搞明白整个系统长什么样。这里面有两个关键模块知识图谱存储电商的结构化数据商品、分类、品牌、属性以及从描述文本里抽出来的标签。实体抽取模型NER从商品描述里自动提取标签比如“顶级拍照旗舰” → “拍照旗舰”。下面我们按顺序从环境配置开始一步步搭建。二、环境准备把地基打好2.1 为什么用 Conda不同项目依赖的 Python 版本、库版本可能冲突。Conda 能创建隔离的环境每个项目一套独立的环境互不干扰。# 创建名为 graph 的环境指定 Python 3.12 conda create -n graph python3.12 # 激活环境 conda activate graph激活后命令行前面会出现 (graph)表示你现在就在这个隔离环境里。2.2 安装 PyTorch根据你的显卡PyTorch 是深度学习框架我们用来训练和运行 NER 模型。如果你的电脑有 NVIDIA 显卡CUDA安装 GPU 版本会快很多如果没有装 CPU 版本也能跑。# 有 CUDA 12.x 的显卡比如 RTX 30/40 系列 pip3 install torch --index-url https://download.pytorch.org/whl/cu128 # 如果没有独立显卡或者不想折腾 CUDA装 CPU 版本 pip3 install torch --index-url https://download.pytorch.org/whl/cpu小提示不确定自己有没有 CUDA可以在命令行输入 nvidia-smi如果能显示显卡信息就有。2.3 安装所有依赖库下面这条命令会安装本项目用到的所有第三方库。我把每个库的用途都写在注释里了pip install \ pymysql \ # 连接 MySQL 数据库 neo4j \ # 连接 Neo4j 图数据库 transformers \ # Hugging Face 的预训练模型库BERT等 accelerate \ # 加速模型训练自动混合精度等 datasets \ # 方便地加载和处理数据集 tensorboard \ # 可视化训练过程看损失曲线 fastapi \ # 写 API 接口的 Web 框架 uvicorn \ # 运行 FastAPI 的服务器 langchain \ # 大模型应用开发框架 langchain-deepspeak \ # 连接 DeepSeek 大模型也可以用其他模型 python-dotenv \ # 加载 .env 配置文件存放 API Key evaluate \ # 评估模型指标精确率、召回率等 langchain_huggingface \# LangChain 对 Hugging Face 的集成 langchain_neo4j \ # LangChain 对 Neo4j 的集成 sentence-transformers # 生成文本向量用于混合检索2.4 数据库准备MySQL存放原始电商数据项目里的模拟数据是一个叫 gmall.sql 的脚本。你需要先创建一个数据库CREATE DATABASE gmall CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;然后用数据库管理工具比如 Navicat、DataGrip或者命令行把这个脚本导入进去。导入后你会看到这些表base_category1/2/3商品的三级分类比如“手机”是一级“智能手机”是二级“拍照手机”是三级base_attr_info平台属性比如“运行内存”“屏幕尺寸”spu_info标准产品单元比如“Apple iPhone 16 Pro”sku_info库存量单位比如“iPhone 16 Pro 黑色 256G”base_trademark品牌比如“Apple”数据库脚本下载https://pan.baidu.com/s/1SbovpOMRD6AOhLukrF447A?pwdsgmsNeo4j图数据库去 Neo4j 官网 下载 Desktop 版本免费创建一个新数据库设置用户名和密码默认都是 neo4j。后面代码里的配置文件会用到这些信息。三、实体抽取模型让机器读懂商品描述3.1 这个模型要解决什么问题电商的商品描述里有很多特征词比如“顶级拍照旗舰”“高性价比”“续航持久”。这些词不在 SKU 表里但用户搜索时经常用。我们的目标是自动从描述文本里把这些标签抽出来然后作为节点放进知识图谱。举个例子原始描述“顶级拍照旗舰影像效果超乎想象。”模型输出[拍照旗舰]原始描述“2018秋冬季新款韩版平底高帮鞋女休闲二棉鞋加绒运动厚底高邦鞋潮”模型输出[2018秋冬季新款, 韩版, 加绒, 厚底]这种任务叫命名实体识别NER。我们用的是 BERT 中文预训练模型在标注好的数据上微调。3.2 数据标注用 LabelStudio 打标签你不需要自己从零标数据。我提供了标注好的 JSON 文件。数据集下载https://pan.baidu.com/s/1WW9NapxTAC6UpqYz_-qDxw?pwd1im2但为了让你理解格式我解释一下{ text: 麦德龙德国进口双心多维叶黄素护眼营养软胶囊30粒x3盒眼干涩, label: [ { start: 3, // 实体开始位置从0数 end: 7, // 实体结束位置不包含7 text: 德国进口, labels: [TAG] }, { start: 14, end: 16, text: 护眼, labels: [TAG] } ] }start 和 end 是字符位置。比如“麦德龙”三个字占位置 0,1,2那么“德国进口”就从位置 3 开始到位置 7 结束不包含7实际占 3,4,5,6。3.3 数据预处理把标注转成 BIO 格式BERT 模型不能直接吃 start/end我们需要把它转成BIO 标签序列。B实体的第一个字I实体的后续字O非实体例如“德国进口”四个字对应的标签是 B, I, I, I。代码 process.py 负责这个转换# graph/src/models/ner/process.py from datasets import load_dataset from transformers import AutoTokenizer from configuration import config # 配置文件里面定义了数据路径、模型名称等 def process(): # ------------------- 1. 加载原始 JSON 数据 ------------------- # load_dataset 是 Hugging Face 提供的万能数据加载器支持 json、csv、parquet 等格式 dataset load_dataset(json, data_filesstr(config.DATA_DIR / ner / raw / data.json))[train] # 去掉我们用不到的字段只保留 text 和 label dataset dataset.remove_columns([id, annotator, annotation_id, created_at, updated_at, lead_time]) # ------------------- 2. 划分训练集、验证集、测试集 ------------------- # 先分出 80% 作为训练集剩下 20% 再平分给验证集和测试集各10% dataset_dict dataset.train_test_split(train_size0.8) dataset_dict[test], dataset_dict[valid] dataset_dict[test].train_test_split(test_size0.5).values() # ------------------- 3. 加载分词器和标签映射 ------------------- tokenizer AutoTokenizer.from_pretrained(config.NER_MODEL) # config.NER_MODEL bert-base-chinese id2label [B, I, O] # 数字 - 标签 label2id {label: id for id, label in enumerate(id2label)} # 标签 - 数字 # ------------------- 4. 定义转换函数核心 ------------------- def map_func(example): # 把字符串拆成单个字符列表因为中文适合按字切分 tokens list(example[text]) # 先用 tokenizer 把字符转成模型输入的 id同时生成 attention_mask 等 # is_split_into_wordsTrue 表示输入已经是 token 列表不需要再分词 inputs tokenizer(tokens, truncationTrue, is_split_into_wordsTrue) # 初始化所有位置的标签为 O数字2因为 O 对应 id2 labels [label2id[O]] * len(tokens) # 遍历标注的实体把对应位置的标签改为 B 和 I for entity in example[label]: start entity[start] end entity[end] # 第一个字是 B后面都是 I labels[start:end] [label2id[B]] [label2id[I]] * (end - start - 1) # BERT 会在句子开头加 [CLS] token结尾加 [SEP] token # 对应位置的标签我们设为 -100训练时会自动忽略 labels [-100] labels [-100] inputs[labels] labels return inputs # 对三个数据集都执行 map_func并删除原来的 text 和 label 列 dataset_dict dataset_dict.map(map_func, batchedFalse, remove_columns[text, label]) # 保存处理后的数据到磁盘格式是 Hugging Face 的 Dataset 格式方便下次直接加载 dataset_dict.save_to_disk(config.DATA_DIR / ner / processed) if __name__ __main__: process()通俗解释这个过程就像把一份标注了“哪个词到哪个词是标签”的文档变成每个字旁边写一个字母B/I/O的新文档方便模型学习。3.4 模型训练让 BERT 学会识别标签训练脚本 train.py 负责加载预处理好的数据用 BERT 进行微调。# graph/src/models/ner/train.py import evaluate from datasets import load_from_disk from transformers import ( AutoModelForTokenClassification, Trainer, TrainingArguments, DataCollatorForTokenClassification, AutoTokenizer ) from configuration import config # ------------------- 1. 加载模型 ------------------- # AutoModelForTokenClassification 会自动为 BERT 加上一个分类头输出维度 标签数3 model AutoModelForTokenClassification.from_pretrained( config.NER_MODEL, # bert-base-chinese num_labelslen(config.LABELS), # config.LABELS [B,I,O] id2labelid2label, label2idlabel2id ) # ------------------- 2. 加载数据集 ------------------- train_dataset load_from_disk(config.DATA_DIR / ner / processed / train) valid_dataset load_from_disk(config.DATA_DIR / ner / processed / valid) # ------------------- 3. 评估指标 ------------------- # seqeval 是专门用于序列标注NER、分词的评估库 seqeval evaluate.load(seqeval) def compute_metrics(prediction): 输入模型预测的 logits 和真实标签 输出精确率、召回率、F1 等 logits prediction.predictions preds logits.argmax(axis-1) # 取概率最大的类别索引 labels prediction.label_ids # 把数字标签转回 B,I,O 字符串同时过滤掉 -100填充位 true_predictions [ [id2label[p] for (p, l) in zip(pred, label) if l ! -100] for pred, label in zip(preds, labels) ] true_labels [ [id2label[l] for (p, l) in zip(pred, label) if l ! -100] for pred, label in zip(preds, labels) ] return seqeval.compute(predictionstrue_predictions, referencestrue_labels) # ------------------- 4. 训练参数 ------------------- training_args TrainingArguments( output_dirstr(config.CHECKPOINT_DIR / ner), # 模型保存目录 per_device_train_batch_size2, # 每块 GPU 的训练批次大小小数据集可以设小点 logging_steps20, # 每20步打印一次日志 num_train_epochs10, # 训练10轮 save_steps20, # 每20步保存一次检查点 save_total_limit3, # 最多保留3个检查点 eval_strategysteps, # 按步数进行评估 eval_steps20, # 每20步评估一次 load_best_model_at_endTrue, # 训练结束后加载验证集上最好的模型 metric_for_best_modeleval_overall_f1, # 用 F1 分数判断好坏 greater_is_betterTrue ) # ------------------- 5. 创建 Trainer 并开始训练 ------------------- trainer Trainer( modelmodel, argstraining_args, data_collatorDataCollatorForTokenClassification(tokenizertokenizer, paddingTrue), train_datasettrain_dataset, eval_datasetvalid_dataset, compute_metricscompute_metrics ) trainer.train() trainer.save_model(config.CHECKPOINT_DIR / ner / best_model)通俗解释训练就像让一个聪明的学生BERT做很多道“给字贴标签”的练习题每做一批题就看看答案错了就调整一下思路。10轮下来它就学会了。3.5 模型预测从文本里抽实体训练好后我们需要一个方便的类来调用模型从任意文本中提取实体标签。# graph/src/models/ner/predict.py import torch from transformers import AutoModelForTokenClassification, AutoTokenizer from configuration import config class Predictor: def __init__(self, model, tokenizer, device): self.model model.to(device) self.model.eval() # 切换到评估模式关闭 dropout 等 self.tokenizer tokenizer self.device device def predict(self, inputs: str | list, batch_size8): 返回 BIO 标签序列 # 为了方便统一处理成列表 is_str isinstance(inputs, str) if is_str: inputs [inputs] predictions [] for i in range(0, len(inputs), batch_size): batch inputs[i:ibatch_size] # 把每个句子拆成字列表 batch [list(text) for text in batch] batch_inputs self.tokenizer(batch, return_tensorspt, paddingTrue, truncationTrue, is_split_into_wordsTrue) batch_inputs {k: v.to(self.device) for k, v in batch_inputs.items()} with torch.no_grad(): # 不计算梯度加快推理 outputs self.model(**batch_inputs) logits outputs.logits batch_preds torch.argmax(logits, dim-1).tolist() # 去掉 [CLS] 和 [SEP] 对应的预测 for text, pred in zip(batch, batch_preds): pred pred[1:1len(text)] # 第一个是 [CLS]最后一个是 [SEP]都去掉 pred [self.model.config.id2label[id] for id in pred] predictions.append(pred) if is_str: return predictions[0] return predictions def extract(self, inputs: str | list): 从文本中提取实体列表连续 B-I 序列 is_str isinstance(inputs, str) if is_str: inputs [inputs] # 先预测 BIO 标签 results self.predict(inputs) all_entities [] for text, pred in zip(inputs, results): entities self._bio_to_entities(text, pred) all_entities.append(entities) return all_entities[0] if is_str else all_entities def _bio_to_entities(self, tokens, labels): 把 BIO 序列转成实体列表。 规则遇到 B 开始新实体遇到 I 且前面有实体则追加遇到 O 结束当前实体。 entities [] current_entity for token, label in zip(tokens, labels): if label B: if current_entity: entities.append(current_entity) current_entity token elif label I: if current_entity: # 只有前面有 B 才接续 current_entity token # 如果前面没有 B这个 I 就丢弃 else: # O if current_entity: entities.append(current_entity) current_entity if current_entity: entities.append(current_entity) return entities # 使用示例 if __name__ __main__: model AutoModelForTokenClassification.from_pretrained(str(config.CHECKPOINT_DIR / ner / best_model)) tokenizer AutoTokenizer.from_pretrained(str(config.CHECKPOINT_DIR / ner / best_model)) device torch.device(cuda if torch.cuda.is_available() else cpu) predictor Predictor(model, tokenizer, device) text 2018秋冬季新款韩版平底高帮鞋女休闲二棉鞋加绒运动厚底高邦鞋潮 entities predictor.extract(text) print(entities) # 输出: [2018秋冬季新款, 韩版, 加绒, 厚底]四、知识图谱构建把数据变成一张大网4.1 为什么要用图数据库传统数据库MySQL存的是表格要查“苹果手机有哪些型号”需要做多次 JOIN连接表很慢且不直观。图数据库里数据是节点和关系比如节点(品牌:Apple)、(SPU:iPhone 16 Pro)、(分类:手机)关系(iPhone 16 Pro)-[:Belong_to]-(Apple)、(iPhone 16 Pro)-[:Has_Tag]-(拍照旗舰)查询时沿着关系走几步就能拿到结果非常自然。4.2 同步结构化数据把 MySQL 的表搬到 Neo4j我们需要把 MySQL 里的分类、属性、SPU、SKU、品牌都转换成 Neo4j 的节点和关系。首先写两个工具类MysqlReader 负责从 MySQL 读数据Neo4jWriter 负责往 Neo4j 写数据。# graph/src/datasync/utils.py import pymysql from neo4j import GraphDatabase from pymysql.cursors import DictCursor from configuration import config class MysqlReader: def __init__(self): # 建立 MySQL 连接 self.connection pymysql.connect(**config.MYSQL_CONFIG) # DictCursor 让查询结果返回字典字段名作为 key self.cursor self.connection.cursor(cursorDictCursor) def read_data(self, sql): self.cursor.execute(sql) return self.cursor.fetchall() # 返回列表每个元素是一个 dict def close(self): self.cursor.close() self.connection.close() class Neo4jWriter: def __init__(self): self.driver GraphDatabase.driver( uriconfig.NEO4J_CONFIG[uri], auth(config.NEO4J_CONFIG[user], config.NEO4J_CONFIG[password]) ) def write_nodes(self, label, batch_data, batch_size20): 批量创建节点使用 MERGE 避免重复 for i in range(0, len(batch_data), batch_size): batch batch_data[i:ibatch_size] # UNWIND 把列表展开成多行每行执行 MERGE cypher f UNWIND $batch AS row MERGE (n:{label} {{id: row.id, name: row.name}}) self.driver.execute_query(cypher, parameters_{batch: batch}) def write_relationships(self, start_label, end_label, relationships, rel_type, batch_size20): 批量创建关系 for i in range(0, len(relationships), batch_size): batch relationships[i:ibatch_size] cypher f UNWIND $batch AS row MATCH (start:{start_label} {{id: row.start_id}}) MATCH (end:{end_label} {{id: row.end_id}}) MERGE (start)-[:{rel_type}]-(end) self.driver.execute_query(cypher, parameters_{batch: batch})然后对每一张表写一个同步方法。以分类为例# graph/src/datasync/table_sync.py class TableSynchronizer: def __init__(self): self.mysql_reader MysqlReader() self.neo4j_writer Neo4jWriter() # 同步一级分类 def sync_base_category1(self): sql SELECT id, name FROM base_category1 data self.mysql_reader.read_data(sql) self.neo4j_writer.write_nodes(labelCategory1, batch_datadata) # 同步二级分类 def sync_base_category2(self): sql SELECT id, name FROM base_category2 data self.mysql_reader.read_data(sql) self.neo4j_writer.write_nodes(labelCategory2, batch_datadata) # 同步三级分类 def sync_base_category3(self): sql SELECT id, name FROM base_category3 data self.mysql_reader.read_data(sql) self.neo4j_writer.write_nodes(labelCategory3, batch_datadata) # 同步一级分类和二级分类之间的 Belong 关系 def sync_category1_category2(self): sql SELECT c2.id AS start_id, c2.category1_id AS end_id FROM base_category2 c2 rels self.mysql_reader.read_data(sql) self.neo4j_writer.write_relationships( start_labelCategory2, end_labelCategory1, relationshipsrels, rel_typeBelong ) # ... 其他同步方法类似属性、SPU、SKU、品牌等最后在 if __name__ __main__ 里按顺序调用所有同步方法。执行后Neo4j 里就会出现完整的商品结构图谱。4.3 同步非结构化数据把商品描述变成标签节点这是项目的一大亮点。我们利用上面训练好的 NER 模型从 spu_info 表的 description 字段里提取标签然后把每个标签作为一个 Tag 节点创建到图谱中并建立 (SPU)-[:Have]-(Tag) 的关系。# graph/src/datasync/text_sync.py import torch from transformers import AutoModelForTokenClassification, AutoTokenizer from configuration import config from datasync.utils import Neo4jWriter, MysqlReader from models.ner.predict import Predictor class TextSynchronizer: def __init__(self): self.neo4j_writer Neo4jWriter() self.mysql_reader MysqlReader() self.extractor self._init_extractor() def _init_extractor(self): model AutoModelForTokenClassification.from_pretrained(str(config.CHECKPOINT_DIR / ner / best_model)) tokenizer AutoTokenizer.from_pretrained(str(config.CHECKPOINT_DIR / ner / best_model)) device torch.device(cuda if torch.cuda.is_available() else cpu) return Predictor(model, tokenizer, device) def sync_spu_desc(self): # 1. 从 MySQL 读取所有 SPU 的 id 和 description sql SELECT id, description FROM spu_info spus self.mysql_reader.read_data(sql) ids [spu[id] for spu in spus] descs [spu[description] for spu in spus] # 2. 批量调用 NER 模型提取标签每个描述可能提取出多个标签 all_entities self.extractor.extract(descs) # 返回 list of list # 3. 准备节点和关系数据 nodes [] # 将要创建的 Tag 节点 rels [] # 将要创建的 Have 关系 for spu_id, entities in zip(ids, all_entities): for idx, entity in enumerate(entities): node_id f{spu_id}-{idx} # 组合 ID 确保唯一性 nodes.append({id: node_id, name: entity}) rels.append({start_id: spu_id, end_id: node_id}) # 4. 写入 Neo4j self.neo4j_writer.write_nodes(Tag, nodes) self.neo4j_writer.write_relationships(SPU, Tag, rels, Have) if __name__ __main__: synchronizer TextSynchronizer() synchronizer.sync_spu_desc()执行后Neo4j 里就会出现大量 Tag 节点比如“拍照旗舰”“续航持久”“高性价比”并且每个商品都通过 Have 关系连接到了对应的标签。五、智能客服让大模型替你写查询现在图谱已经建好了里面有商品、分类、品牌、标签还有各种关系。用户问“推荐一款拍照好的手机”我们怎么让系统理解并查询思路是让大模型LLM根据用户问题自动写出图查询语言Cypher。然后再用混合检索把模糊的实体名称比如“拍照好”对齐到图谱里准确的标签比如“拍照旗舰”最后执行查询把结果交给大模型生成回答。5.1 创建混合检索索引实体对齐需要同时用到全文检索精确匹配和向量检索语义相似。我们提前为 SPU、品牌、分类等实体创建两种索引。# graph/src/web/utils.py from langchain_huggingface import HuggingFaceEmbeddings from langchain_neo4j import Neo4jGraph def create_full_text_index(graph, index_name, label, property_name): 创建全文索引 cypher f CREATE FULLTEXT INDEX {index_name} FOR (n:{label}) ON EACH [n.{property_name}] graph.query(cypher) def create_embedding_index(graph, index_name, label, property_name, embedding_model, dim512): 创建向量索引先生成 embedding 存到节点属性再建索引 # 1. 查询所有需要 embedding 的节点 query fMATCH (n:{label}) WHERE n.{property_name} IS NOT NULL RETURN id(n) AS node_id, n.{property_name} AS text nodes graph.query(query) # 2. 分批生成 embedding 并更新节点 batch_size 100 for i in range(0, len(nodes), batch_size): batch nodes[i:ibatch_size] texts [record[text] for record in batch] embeddings embedding_model.embed_documents(texts) # 调用模型生成向量 for record, emb in zip(batch, embeddings): update_query fMATCH (n) WHERE id(n)$node_id SET n.embedding$embedding graph.query(update_query, {node_id: record[node_id], embedding: emb}) # 3. 创建向量索引HNSW 算法 cypher_index f CREATE VECTOR INDEX {index_name} FOR (n:{label}) ON n.embedding OPTIONS {{indexConfig: {{ vector.dimensions: {dim}, vector.similarity_function: cosine }}}} graph.query(cypher_index) # 主函数为所有实体类型创建索引 if __name__ __main__: graph Neo4jGraph(urlconfig.NEO4J_CONFIG[uri], usernameconfig.NEO4J_CONFIG[user], passwordconfig.NEO4J_CONFIG[password]) # 创建全文索引 for label in [SPU, BaseTrademark, Category3, Category2, Category1]: create_full_text_index(graph, f{label.lower()}_fulltext, label, name) # 创建向量索引 embedding_model HuggingFaceEmbeddings( model_nameBAAI/bge-small-zh-v1.5, encode_kwargs{normalize_embeddings: True} ) for label in [SPU, BaseTrademark, Category3, Category2, Category1]: create_embedding_index(graph, f{label.lower()}_vector, label, name, embedding_model, dim512)5.2 智能客服服务类核心类 ChatService 整合了 LLM、Neo4j 连接、向量存储和四个步骤。# graph/src/web/service.py from langchain_core.output_parsers import JsonOutputParser, StrOutputParser from langchain_core.prompts import PromptTemplate from langchain_deepseek import ChatDeepSeek from langchain_huggingface import HuggingFaceEmbeddings from langchain_neo4j import Neo4jGraph, Neo4jVector from neo4j_graphgraph.types import SearchType from configuration import config class ChatService: def __init__(self): # 大模型使用 DeepSeek API也可以换成其他 self.llm ChatDeepSeek(modeldeepseek-chat, temperature0, api_keyconfig.DEEPSEEK_API_KEY) # Neo4j 图连接 self.graph Neo4jGraph( urlconfig.NEO4J_CONFIG[uri], usernameconfig.NEO4J_CONFIG[user], passwordconfig.NEO4J_CONFIG[password] ) # Embedding 模型 self.embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-small-zh-v1.5, encode_kwargs{normalize_embeddings: True} ) # 为每个实体类型创建向量存储支持混合检索 self.vector_stores { SPU: Neo4jVector.from_existing_index( self.embeddings, index_namespu_vector, search_typeSearchType.HYBRID ), BaseTrademark: Neo4jVector.from_existing_index( self.embeddings, index_nametrademark_vector, search_typeSearchType.HYBRID ), Category3: Neo4jVector.from_existing_index( self.embeddings, index_namecategory3_vector, search_typeSearchType.HYBRID ), # ... 其他类型 } self.json_parser JsonOutputParser() self.str_parser StrOutputParser() def _generate_cypher(self, question: str, schema_info: str): 步骤1LLM 生成 Cypher 查询和待对齐实体 prompt PromptTemplate( input_variables[question, schema_info], template 你是一个专业的 Neo4j Cypher 查询生成器。根据用户问题生成 Cypher 查询语句。 用户问题: {question} 知识图谱结构节点类型、关系类型、属性: {schema_info} 要求: 1. 使用参数化查询参数名用 param_0, param_1, ... 2. 识别出需要对齐的实体比如用户说的“拍照好”可能对应 Tag 节点中的“拍照旗舰” 输出格式必须是严格的 JSON例如 {{ cypher_query: MATCH (spu:SPU)-[:Have]-(tag:Tag) WHERE tag.name $param_0 RETURN spu.name, entities_to_align: [ {{param_name: param_0, entity: 拍照好, label: Tag}} ] }} ) chain prompt | self.llm | self.json_parser return chain.invoke({question: question, schema_info: schema_info}) def _entity_align(self, entities_to_align): 步骤2用混合检索把模糊实体对齐到图谱里的标准名称 aligned [] for item in entities_to_align: label item[label] entity_text item[entity] if label in self.vector_stores: # 混合检索先全文匹配再向量相似度取最佳结果 results self.vector_stores[label].similarity_search(entity_text, k1) if results: # 把实体替换成检索到的标准名称 item[entity] results[0].page_content aligned.append(item) return aligned def _execute_cypher(self, cypher: str, params: dict): 步骤3执行 Cypher 查询 return self.graph.query(cypher, paramsparams) def _generate_answer(self, question: str, query_result: list): 步骤4把查询结果转成自然语言回答 prompt PromptTemplate( input_variables[question, query_result], template 你是一个电商智能客服。根据用户问题和数据库查询结果生成简洁、准确的回答。 用户问题: {question} 查询结果: {query_result} 要求回答要自然、友好如果查询结果为空请如实告知。 ) chain prompt | self.llm | self.str_parser return chain.invoke({question: question, query_result: query_result}) def chat(self, question: str) - str: 主流程 # 获取图谱结构信息自动从 Neo4j 提取 schema self.graph.get_schema # LangChain 提供的方法 # 1. 生成 Cypher cypher_info self._generate_cypher(question, schema) cypher_query cypher_info[cypher_query] entities cypher_info[entities_to_align] # 2. 实体对齐 aligned_entities self._entity_align(entities) params {item[param_name]: item[entity] for item in aligned_entities} # 3. 执行查询 result self._execute_cypher(cypher_query, params) # 4. 生成回答 answer self._generate_answer(question, result) return answer5.3 提供 Web 接口最后我们用 FastAPI 把服务包装成 HTTP 接口并提供一个简单的聊天页面。# graph/src/web/schemas.py from pydantic import BaseModel class Question(BaseModel): message: str class Answer(BaseModel): message: str# graph/src/web/app.py import uvicorn from fastapi import FastAPI from starlette.responses import RedirectResponse from starlette.staticfiles import StaticFiles from web.schemas import Question, Answer from web.service import ChatService from configuration import config app FastAPI() app.mount(/static, StaticFiles(directorystr(config.WEB_STATIC_DIR)), namestatic) service ChatService() app.get(/) def root(): return RedirectResponse(/static/index.html) app.post(/api/chat) def chat(question: Question) - Answer: answer service.chat(question.message) return Answer(messageanswer) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)启动服务后访问 http://localhost:8000你会看到一个聊天界面。输入问题后台就会按四个步骤处理并返回答案。六、总结与思考6.1 我们做了什么环境搭建Conda PyTorch 各种库一条命令全搞定。实体抽取模型从商品描述里自动提取标签NER把非结构化数据变成可查询的图节点。知识图谱构建把 MySQL 里的结构化数据和 NER 提取的标签全部同步到 Neo4j 图数据库。混合检索索引为实体创建全文索引和向量索引让模糊匹配成为可能。智能客服服务用大模型自动生成 Cypher 查询混合检索对齐实体执行查询再生成自然答案。Web 界面FastAPI 静态页面开箱即用。6.2 这套方案好在哪里结构化与非结构化数据融合既有 SKU 表的精确属性又有描述文本里的特征标签查询更全面。关系推理能力图数据库天然支持多跳查询比如“苹果手机有哪些支持5G的型号”可以沿着品牌→SPU→属性走两步。大模型辅助查询不用写死规则LLM 自动理解意图并生成 Cypher适应各种问法。混合检索对齐用户说“拍照好”系统能匹配到“拍照旗舰”因为向量相似度高用户说“苹果”系统能精确匹配“Apple”品牌因为全文检索也生效。6.3 可以怎么改进多轮对话记忆目前每次对话是独立的。可以引入对话历史让模型记住用户之前问过什么。主动推荐根据用户常问的标签比如“高性价比”在用户没问的时候主动推荐相关商品。性能优化embedding 生成和 Cypher 查询都可以加缓存减少重复计算。更多模态商品图片也能用多模态模型提取特征标签进一步丰富图谱。6.4 写在最后这篇文章从零开始完整实现了电商智能客服系统。你可能注意到整个项目没有用到特别复杂或者闭源的技术——BERT 微调 NER 模型、Neo4j 图数据库、LangChain 调用大模型都是当前非常成熟且开源的方案。希望这篇教程能帮你建立起知识图谱 大模型的技术直觉。如果你在实践过程中遇到任何问题或者有新的想法欢迎交流讨论。代码即文档动手出真知。现在就去敲键盘吧完整代码下载包含数据集https://pan.baidu.com/s/1Cn5CxCUZhdCvjnkxirrV9w?pwdvhqc