企业官网建设流程全解析

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从16到64，ef从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 pymilvus==2.3.0 pip install neo4j==5.14.0 pip install openai==1.3.0 pip install langchain==0.0.340 pip install sentence-transformers==2.2.2

Milvus和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(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768), FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=2000), FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=100), FieldSchema(name="chapter", dtype=DataType.VARCHAR, max_length=200), ] 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, name=name, properties=properties) 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_name=from_name, to_name=to_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_key="your-key", base_url="your-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( model="gpt-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_k=5): query_vector = model.encode([query_text])[0].tolist() search_params = {"metric_type": "COSINE", "params": {"ef": 128}} results = collection.search( data=[query_vector], anns_field="embedding", param=search_params, limit=top_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"[文档{i+1}] {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( model="gpt-4", messages=[{"role": "user", "content": prompt}] ) return response.choices[0].message.content

5. 常见问题与排查技巧实录

5.1 向量检索召回率低怎么办

这是最常见的问题。我排查下来主要有三个原因：

分块策略不合理。如果按固定长度切分，一个完整的语义单元可能被切成两半，检索时匹配到的是半句话。解决方法是改用语义分块，确保每个块是一个完整的意思表达。

嵌入模型不匹配。中文场景用英文模型效果会差很多。我试过几个模型，最终选了多语言版本，中文语义相似度明显更好。

索引参数太保守。ef值设太小会导致候选集不够，召回率下降。建议从128起步，逐步调大观察效果。

5.2 图查询返回空结果怎么排查

图查询空结果通常不是查询语句写错了，而是数据本身的问题。我的排查顺序是：

1. 先查节点是否存在：MATCH (n:Device {name: 'A设备'}) RETURN n
2. 再查关系是否存在：MATCH (a:Device {name: 'A设备'})-[r]->(b) RETURN type(r), b.name
3. 最后查多跳路径：逐步增加跳数，看在哪一跳断掉

常见原因是实体对齐没做好，图上存在多个相似但不完全相同的节点名。我写了一个简单的对齐脚本，用编辑距离小于2且类型相同的节点做合并。

5.3 大模型生成的Cypher语句执行报错

大模型生成的Cypher经常有语法问题，比如关系类型写错、属性名不存在。我的解决方案是：

在Prompt里给Schema，明确告诉它有哪些节点类型、关系类型和属性名。

加一层校验。执行前先用EXPLAIN检查语法，不通过就返回给大模型重新生成。

限制重试次数。最多重试两次，两次都失败就降级到纯向量检索，保证系统可用性。

5.4 混合检索的结果冲突怎么处理

有时候向量检索说“A设备需要更换散热模块”，图查询说“A设备属于产线B，产线B的维护记录显示散热正常”。这种冲突需要大模型做判断。我的做法是在Prompt里明确要求：“如果不同来源的信息存在冲突，请分别列出并说明可能的原因，不要强行统一。”

5.5 性能瓶颈与优化经验

系统上线后遇到的性能问题主要有两个：

向量检索延迟随数据量增长。百万级时延迟还可以接受，但到五百万级时明显变慢。解决方案是分区+分片，按文档类型分区，查询时指定分区，减少搜索范围。

图查询多跳性能下降。三跳以内的查询很快，四跳以上明显变慢。解决方案是预计算部分路径，把常用的多跳关系物化成直接关系。比如“设备→工序→产线”这条两跳路径，如果查询频繁，可以直接建一条“设备→产线”的关系。

5.6 几个我踩过的坑

坑一：嵌入模型换了但没重建索引。换模型后向量空间变了，旧索引完全不能用。必须全量重建，没有捷径。

坑二：图数据库事务太大。一次性导入十万条三元组，事务超时。后来改成每批一千条，用apoc.periodic.iterate分批提交。

坑三：大模型输出不稳定。同样的输入，有时候输出JSON有时候输出Markdown。解决方案是用response_format={"type": "json_object"}强制JSON输出，同时在Prompt里强调“只输出JSON，不要其他内容”。

坑四：忘记处理并发。多个用户同时查询时，Milvus连接池不够用。后来加了连接池配置，每个查询独立获取连接，用完释放。

这套系统我前后调了大概两个月，从最初的原型到稳定运行，中间经历了多次重构。最深的体会是：向量库和图库的协同不是简单的叠加，而是需要在数据层面做好对齐，在查询层面做好路由，在结果层面做好融合。任何一个环节偷懒，最终效果都会打折扣。