1. 为什么要把向量数据库和图数据库放在一起用1.1 从一个真实需求说起去年下半年我接手了一个内部知识库的改造项目需求方给的原话是“我们想要一个能理解语义、还能顺着关系往下查的智能问答系统。”这句话听起来简单但拆开来看它其实包含了两个完全不同维度的检索需求。第一个维度是语义相似性检索。用户问“设备过热怎么处理”系统需要找到那些讲“温度异常”“散热方案”“热管理策略”的文档哪怕这些文档里根本没有出现“过热”这两个字。这是向量数据库的强项——把文本转成高维向量通过余弦相似度或内积计算找到语义上最接近的内容。第二个维度是关联关系推理。用户问“A设备故障会影响哪些下游产线”系统需要沿着“A设备→所属工序→下游工序→关联产线”这条关系链一步步推下去。这是图数据库的强项——节点和边构成的网络结构天然适合做多跳查询和路径分析。单独用向量数据库你能找到语义相近的文档片段但没法回答“这个故障会传导到哪些环节”这种需要关系推理的问题。单独用图数据库你能精确地沿着预定义的关系查询但没法处理“用户用自然语言描述了一个模糊需求”这种场景。所以这套方案的核心思路就是用向量数据库做“模糊入口”用图数据库做“精确推理”中间用大模型做“翻译官”和“调度员”。1.2 这套架构适合谁参考如果你正在做以下几类事情这套方案可以直接参考企业知识库需要支持自然语言问答同时要求答案能追溯到文档来源和关联实体运维监控系统需要根据告警描述自动定位故障根因并推荐处理方案电商场景需要根据用户模糊描述推荐商品同时展示商品之间的替代、互补关系任何需要“先模糊匹配、再精确推理”的检索增强生成应用技术栈方面我选的是Milvus向量库 Neo4j图库 一个大模型API。选型理由后面会详细说但核心原则是向量库要支持高性能近似最近邻搜索图库要支持灵活的Cypher查询大模型要能稳定输出结构化结果。2. 整体架构设计与核心思路拆解2.1 三层架构的职责划分整个系统我把它拆成三层每层职责非常明确第一层语义理解与路由层。用户输入的自然语言先经过大模型做两件事——一是提取关键实体和意图二是判断这个查询应该走哪条路径。比如“设备过热怎么处理”偏向语义检索“A设备故障影响哪些产线”偏向图查询“B产线的设备维护记录里有没有提到温度异常”则需要两者结合。第二层检索与推理层。向量数据库负责接收文本向量返回Top-K相似片段图数据库负责接收实体和关系模式返回关联子图。这一层的关键是结果融合——向量检索返回的是文本片段和相似度分数图查询返回的是节点、边和属性两者需要对齐到同一个语义空间。第三层生成与溯源层。大模型拿到融合后的上下文生成最终答案同时标注每个结论的来源——是来自某个文档片段还是来自某条图路径。这一步对可信度至关重要。2.2 为什么选Milvus而不是其他向量库向量数据库这两年冒出来很多选择我最终选Milvus有几个实际考量索引类型的灵活性。Milvus支持IVF_FLAT、IVF_SQ8、HNSW、DiskANN等多种索引。对于知识库场景文档量级在百万级左右HNSW在召回率和延迟之间平衡得最好。我实测下来100万条768维向量HNSW索引在16核64G的机器上Top-10查询平均延迟在8ms左右完全够用。标量字段过滤。知识库场景经常需要“只在某个部门文档里搜”或者“只搜最近半年的内容”Milvus支持向量检索和标量过滤同时进行不需要先全量检索再过滤效率高很多。分区设计。我按文档类型做了分区——技术文档、运维记录、故障案例各一个分区。查询时可以指定分区减少搜索范围。这个设计在文档量增长到千万级时优势会非常明显。2.3 为什么选Neo4j而不是其他图库图数据库的选择相对少一些Neo4j的优势在于Cypher查询语言成熟。表达多跳关系非常直观比如MATCH (a:Device)-[:BELONGS_TO]-(p:Process)-[:DOWNSTREAM*1..3]-(d:Process) RETURN d就能查出设备所属工序的下游三层工序。这种查询用其他方式写会非常痛苦。社区版免费且功能够用。对于中小规模知识图谱社区版完全能支撑。企业版主要是集群和高可用初期用不上。与Python生态集成好。官方驱动neo4j-driver很稳定和LangChain、LlamaIndex这些框架的集成也很成熟。2.4 大模型在架构中的三个角色大模型在这套系统里不是简单的内容生成器它承担了三个关键角色角色一实体与意图抽取。用户输入“A设备过热会不会影响B产线”大模型需要抽取出实体“A设备”“B产线”关系“影响”以及隐含的查询意图“故障传导分析”。角色二查询语句生成。根据抽取结果大模型生成对应的Cypher查询语句和向量检索的查询文本。这一步需要给大模型提供图数据库的Schema信息否则它不知道有哪些节点类型和关系类型可用。角色三结果融合与答案生成。把向量检索的文本片段和图查询的结构化结果一起塞进Prompt让大模型生成自然语言答案并要求它标注每个结论的来源。3. 核心细节解析与实操要点3.1 数据准备从原始文档到向量和图谱这一步是整个项目最耗时的环节我大概花了60%的时间在这上面。原始数据是各种格式的文档——PDF、Word、Markdown、Confluence导出的HTML。处理流程分两条线向量化线路文档先做分块我用的策略是按语义段落分块而不是固定长度切分。具体做法是先用规则把文档拆成段落然后相邻段落如果语义相似度高就合并直到接近512个token。这样每个块是一个完整的语义单元检索时不会出现“半句话”的情况。分块后用嵌入模型转成向量我选的是768维的模型维度适中存储和检索成本都可控。图谱构建线路这一步需要从文档中抽取实体和关系。我的做法是先用大模型做三元组抽取——给大模型一段文本让它输出“实体1-关系-实体2”的列表。然后对抽取结果做实体对齐把“A设备”“设备A”“A号机”统一到同一个节点。最后用Neo4j的Cypher语句批量导入。注意三元组抽取的质量直接决定图谱的可用性。我踩过的坑是初期没有做实体对齐导致同一个设备在图上出现了三个节点查询时经常漏掉关系。后来加了一个基于编辑距离和语义相似度的对齐步骤准确率提升很明显。3.2 向量索引参数调优Milvus的HNSW索引有几个关键参数参数含义我的取值调整逻辑M每个节点的最大连接数32值越大召回率越高但内存占用越大32在百万级数据上是平衡点efConstruction构建时的候选集大小256影响索引构建速度和质量256构建时间可接受ef查询时的候选集大小128运行时参数越大召回越高但延迟增加128实测召回率95%以上这些参数不是拍脑袋定的我做了对比实验M从16到64ef从64到256组合测试了召回率和延迟。最终选的是在召回率95%时延迟最低的组合。3.3 图Schema设计的关键决策图Schema设计我改了三个版本才稳定下来。第一版太细把文档的每个段落都做成节点结果图太大查询慢。第二版太粗只做了文档级节点关系推理能力弱。第三版是现在的方案节点类型文档、章节、实体设备、工序、产线、故障类型、处理方案关系类型文档包含章节、章节提及实体、实体属于工序、工序属于产线、故障影响设备、处理方案适用于故障这个粒度的好处是既能做文档级的语义检索又能做实体级的关系推理还能通过“章节提及实体”这个关系把两者关联起来。3.4 大模型Prompt设计要点Prompt设计我迭代了七八版核心经验是给Schema不给数据。告诉大模型有哪些节点类型和关系类型但不要把整个图谱塞进去。Schema信息控制在500token以内。要求输出结构化。让大模型输出JSON格式包含entities、relations、query_type、cypher、vector_query这几个字段。这样后续处理不需要再做解析。Few-shot示例要精准。我放了三个示例——纯向量查询、纯图查询、混合查询各一个。示例的质量比数量重要三个精准示例比十个模糊示例效果好得多。4. 实操过程与核心环节实现4.1 环境搭建与依赖安装基础环境是Python 3.10主要依赖pip install pymilvus2.3.0 pip install neo4j5.14.0 pip install openai1.3.0 pip install langchain0.0.340 pip install sentence-transformers2.2.2Milvus和Neo4j我用Docker部署Milvus用standalone模式Neo4j用社区版。硬件配置是16核64G存储用SSD。这个配置支撑百万级向量和十万级节点没问题。4.2 文档向量化完整流程from sentence_transformers import SentenceTransformer from pymilvus import Collection, CollectionSchema, FieldSchema, DataType # 加载嵌入模型 model SentenceTransformer(paraphrase-multilingual-mpnet-base-v2) # 定义Collection Schema fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim768), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length2000), FieldSchema(namedoc_id, dtypeDataType.VARCHAR, max_length100), FieldSchema(namechapter, dtypeDataType.VARCHAR, max_length200), ] schema CollectionSchema(fields, description知识库文档向量) collection Collection(knowledge_vectors, schema) # 创建HNSW索引 index_params { metric_type: COSINE, index_type: HNSW, params: {M: 32, efConstruction: 256} } collection.create_index(embedding, index_params) # 批量插入 def insert_documents(chunks): embeddings model.encode([c[text] for c in chunks]) entities [ [c[text] for c in chunks], [c[doc_id] for c in chunks], [c[chapter] for c in chunks], embeddings.tolist() ] collection.insert(entities) collection.flush()这段代码的关键点是metric_type选COSINE。知识库场景文本长度差异大余弦相似度对长度不敏感比内积更稳定。4.3 图谱构建与Cypher导入图谱构建分两步先抽取三元组再导入Neo4j。from neo4j import GraphDatabase driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, password)) def create_entity(tx, name, entity_type, properties): query f MERGE (n:{entity_type} {{name: $name}}) SET n $properties RETURN n tx.run(query, namename, propertiesproperties) def create_relation(tx, from_name, from_type, rel_type, to_name, to_type): query f MATCH (a:{from_type} {{name: $from_name}}) MATCH (b:{to_type} {{name: $to_name}}) MERGE (a)-[:{rel_type}]-(b) tx.run(query, from_namefrom_name, to_nameto_name) with driver.session() as session: for triple in triples: session.execute_write(create_entity, triple[head], triple[head_type], {}) session.execute_write(create_entity, triple[tail], triple[tail_type], {}) session.execute_write(create_relation, triple[head], triple[head_type], triple[relation], triple[tail], triple[tail_type])注意MERGE而不是CREATE这样重复实体不会创建多个节点。但MERGE在并发写入时可能有性能问题大批量导入建议用LOAD CSV或者apoc.periodic.iterate。4.4 查询路由与混合检索实现这是整个系统的核心逻辑。用户输入后先经过大模型做意图识别和查询生成import json from openai import OpenAI client OpenAI(api_keyyour-key, base_urlyour-endpoint) ROUTER_PROMPT 你是一个查询路由器。根据用户问题判断查询类型并生成相应语句。 图谱Schema - 节点Device(设备), Process(工序), Line(产线), Fault(故障), Solution(处理方案), Document(文档), Chapter(章节) - 关系BELONGS_TO(属于), DOWNSTREAM(下游), AFFECTS(影响), SOLVES(解决), CONTAINS(包含), MENTIONS(提及) 输出JSON格式 { query_type: vector | graph | hybrid, entities: [{name: ..., type: ...}], cypher: ... (如果是graph或hybrid), vector_query: ... (如果是vector或hybrid) } def route_query(user_input): response client.chat.completions.create( modelgpt-4, messages[ {role: system, content: ROUTER_PROMPT}, {role: user, content: user_input} ], response_format{type: json_object} ) return json.loads(response.choices[0].message.content)拿到路由结果后分别执行向量检索和图查询def vector_search(query_text, top_k5): query_vector model.encode([query_text])[0].tolist() search_params {metric_type: COSINE, params: {ef: 128}} results collection.search( data[query_vector], anns_fieldembedding, paramsearch_params, limittop_k, output_fields[text, doc_id, chapter] ) return [{text: r.entity.get(text), score: r.score, doc_id: r.entity.get(doc_id)} for r in results[0]] def graph_search(cypher_query): with driver.session() as session: result session.run(cypher_query) return [dict(record) for record in result]4.5 结果融合与答案生成融合策略我试过两种串行融合和并行融合。串行是先向量检索把结果作为图查询的输入并行是同时执行然后合并。我最终选了并行因为延迟更低而且两个维度的结果可以互相补充。def generate_answer(user_input, vector_results, graph_results): context 向量检索结果\n for i, r in enumerate(vector_results): context f[文档{i1}] {r[text][:500]}\n context \n图谱查询结果\n for r in graph_results: context f{r}\n prompt f 基于以下上下文回答用户问题。要求 1. 每个结论标注来源文档编号或图路径 2. 如果上下文不足以回答明确说明 3. 不要编造上下文中没有的信息 上下文 {context} 用户问题{user_input} response client.chat.completions.create( modelgpt-4, messages[{role: user, content: prompt}] ) return response.choices[0].message.content5. 常见问题与排查技巧实录5.1 向量检索召回率低怎么办这是最常见的问题。我排查下来主要有三个原因分块策略不合理。如果按固定长度切分一个完整的语义单元可能被切成两半检索时匹配到的是半句话。解决方法是改用语义分块确保每个块是一个完整的意思表达。嵌入模型不匹配。中文场景用英文模型效果会差很多。我试过几个模型最终选了多语言版本中文语义相似度明显更好。索引参数太保守。ef值设太小会导致候选集不够召回率下降。建议从128起步逐步调大观察效果。5.2 图查询返回空结果怎么排查图查询空结果通常不是查询语句写错了而是数据本身的问题。我的排查顺序是先查节点是否存在MATCH (n:Device {name: A设备}) RETURN n再查关系是否存在MATCH (a:Device {name: A设备})-[r]-(b) RETURN type(r), b.name最后查多跳路径逐步增加跳数看在哪一跳断掉常见原因是实体对齐没做好图上存在多个相似但不完全相同的节点名。我写了一个简单的对齐脚本用编辑距离小于2且类型相同的节点做合并。5.3 大模型生成的Cypher语句执行报错大模型生成的Cypher经常有语法问题比如关系类型写错、属性名不存在。我的解决方案是在Prompt里给Schema明确告诉它有哪些节点类型、关系类型和属性名。加一层校验。执行前先用EXPLAIN检查语法不通过就返回给大模型重新生成。限制重试次数。最多重试两次两次都失败就降级到纯向量检索保证系统可用性。5.4 混合检索的结果冲突怎么处理有时候向量检索说“A设备需要更换散热模块”图查询说“A设备属于产线B产线B的维护记录显示散热正常”。这种冲突需要大模型做判断。我的做法是在Prompt里明确要求“如果不同来源的信息存在冲突请分别列出并说明可能的原因不要强行统一。”5.5 性能瓶颈与优化经验系统上线后遇到的性能问题主要有两个向量检索延迟随数据量增长。百万级时延迟还可以接受但到五百万级时明显变慢。解决方案是分区分片按文档类型分区查询时指定分区减少搜索范围。图查询多跳性能下降。三跳以内的查询很快四跳以上明显变慢。解决方案是预计算部分路径把常用的多跳关系物化成直接关系。比如“设备→工序→产线”这条两跳路径如果查询频繁可以直接建一条“设备→产线”的关系。问题现象可能原因排查方法解决方案向量召回低分块不合理检查检索结果是否语义完整改语义分块图查询空实体未对齐查节点是否存在实体对齐合并Cypher报错Schema不明确看错误信息Prompt加Schema校验结果冲突多源信息不一致对比不同来源Prompt要求分别列出延迟增长数据量增大监控各阶段耗时分区预计算路径5.6 几个我踩过的坑坑一嵌入模型换了但没重建索引。换模型后向量空间变了旧索引完全不能用。必须全量重建没有捷径。坑二图数据库事务太大。一次性导入十万条三元组事务超时。后来改成每批一千条用apoc.periodic.iterate分批提交。坑三大模型输出不稳定。同样的输入有时候输出JSON有时候输出Markdown。解决方案是用response_format{type: json_object}强制JSON输出同时在Prompt里强调“只输出JSON不要其他内容”。坑四忘记处理并发。多个用户同时查询时Milvus连接池不够用。后来加了连接池配置每个查询独立获取连接用完释放。这套系统我前后调了大概两个月从最初的原型到稳定运行中间经历了多次重构。最深的体会是向量库和图库的协同不是简单的叠加而是需要在数据层面做好对齐在查询层面做好路由在结果层面做好融合。任何一个环节偷懒最终效果都会打折扣。