北陌大叔1. 图RAG系统架构与环境配置
在构建智能烹饪助手系统时，我们采用了图RAG（Retrieval-Augmented Generation）架构，这种设计能够充分利用图数据库的关系推理能力和向量检索的语义理解优势。整个系统由Neo4j图数据库和Milvus向量数据库双引擎驱动，配合大语言模型实现智能问答功能。
1.1 核心业务节点设计
菜谱领域的图模型需要精准反映烹饪知识的结构化特征。我们设计了以下核心节点类型：
-
Recipe（菜谱）：作为中心节点，包含
name（菜名）、cooking_time（烹饪时长）、servings（人均份）等基础属性。例如"鱼香肉丝"这个节点会记录其需要20分钟烹饪时间，适合2人食用。 -
Ingredient（食材）：记录
name（名称）、unit（单位）、is_common（是否常见）等属性。比如"里脊肉"会被标记为常见食材，计量单位是"克"。 -
CookingStep（烹饪步骤）：通过
description（步骤描述）和step_order（顺序号）确保流程正确性。"将肉丝腌制10分钟"这样的步骤会带有明确的执行序号。 -
RecipeCategory（菜谱分类）：按
cuisine_type（菜系）和flavor（口味）进行多维分类。川菜、粤菜等菜系和辣/不辣等口味标签都通过这个节点实现。 -
DifficultyLevel（难度等级）：用
level_name（如入门、进阶、大师级）和level_value（1-5数字评分）量化操作难度。
1.2 层次化辅助节点
为了提升图模型的查询效率和管理便利性，我们补充了以下辅助节点：
-
Root根节点：作为整个图谱的单一入口点，通过
CONNECTS_TO关系链接所有实体类型节点。这种设计使得批量更新和维护变得简单，例如可以通过MATCH (r:Root)-[:CONNECTS_TO]->(n) DETACH DELETE n快速清空所有业务数据。 -
CookingMethod（烹饪方法）：记录
name（如炒、炖、蒸）和heat_level（火力要求）等属性。不同烹饪方法与菜谱的关联能支持"用炒的做法"这类条件查询。 -
CookingTool（烹饪工具）：包含
name（如炒锅、砂锅）和is_electric（是否电器）等特征。当用户询问"不用烤箱能做的甜点"时，系统可以通过这个节点快速过滤。
1.3 关系设计的核心原则
在图数据库中，关系（Relationship）和属性（Property）的设计直接影响查询效率和表达能力。我们遵循以下原则：
-
语义明确性：每个关系类型都采用业务场景中自然语言表述。例如：
- 用
REQUIRES替代模糊的RELATE_TO表示菜谱需要特定食材 - 用
HAS_STEP替代通用的HAS表示菜谱包含烹饪步骤
- 用
-
属性丰富化：在关系上附加关键业务属性。典型案例如：
cypher复制(recipe:Recipe)-[r:REQUIRES {amount: 200, unit: 'g'}]->(ing:Ingredient)这种设计可以直接在关系中存储"鱼香肉丝需要200克里脊肉"这样的精确信息，避免频繁的节点跳转查询。
-
索引优化：为高频查询字段创建索引。例如对菜谱的
name和cooking_time建立复合索引，加速"查找30分钟内能完成的菜谱"这类查询：cypher复制CREATE INDEX recipe_name_time IF NOT EXISTS FOR (r:Recipe) ON (r.name, r.cooking_time)
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 开发环境配置
2.1 创建虚拟环境
我们推荐使用conda管理Python环境，确保依赖隔离：
bash复制# 创建指定Python版本的环境
conda create -n graph-rag python=3.12.7
conda activate graph-rag
# 验证Python版本
python --version
注意：Python 3.12对类型注解和异步IO有更好支持，这对构建高性能RAG系统很重要。如果必须使用其他版本，需要测试关键库的兼容性。
2.2 安装核心依赖
项目依赖分为以下几类，通过requirements.txt统一管理：
- 图数据库驱动：
neo4j官方库提供Cypher查询接口 - 向量数据库SDK：
pymilvus支持Milvus 2.x的向量操作 - 数据处理工具：
pandas用于结构化数据处理 - 深度学习框架：
torch和transformers支撑Embedding模型
安装命令：
bash复制cd code/C9
pip install -r requirements.txt
典型requirements.txt内容示例：
code复制neo4j==5.18.0
pymilvus==2.5.11
pandas==2.2.1
torch==2.2.1
transformers==4.38.2
sentence-transformers==2.7.0
2.3 Neo4j数据库配置
2.3.1 Docker Compose部署
我们使用容器化部署保证环境一致性，docker-compose.yml关键配置如下：
yaml复制version: '3'
services:
neo4j:
image: neo4j:5.18.0-enterprise
ports:
- "7474:7474" # Browser界面
- "7687:7687" # Bolt协议端口
volumes:
- ./data:/data
- ./logs:/logs
environment:
NEO4J_AUTH: neo4j/all-in-rag
NEO4J_ACCEPT_LICENSE_AGREEMENT: yes
healthcheck:
test: ["CMD", "cypher-shell", "-u", "neo4j", "-p", "all-in-rag", "RETURN 1"]
启动命令：
bash复制cd data/C9
docker-compose up -d
# 验证服务状态
docker-compose ps
2.3.2 数据初始化
首次启动后需要通过Web界面(http://localhost:7474)或cypher-shell执行初始化：
-
创建约束确保数据唯一性：
cypher复制CREATE CONSTRAINT recipe_name_unique IF NOT EXISTS FOR (r:Recipe) REQUIRE r.name IS UNIQUE; CREATE CONSTRAINT ingredient_name_unique IF NOT EXISTS FOR (i:Ingredient) REQUIRE i.name IS UNIQUE; -
导入示例数据：
cypher复制LOAD CSV WITH HEADERS FROM 'file:///recipes.csv' AS row MERGE (r:Recipe {name: row.recipe_name}) SET r.cooking_time = toInteger(row.cooking_time), r.servings = toInteger(row.servings)
2.4 Milvus向量数据库配置
2.4.1 Standalone模式部署
对于开发环境，使用Standalone模式足够：
bash复制wget https://github.com/milvus-io/milvus/releases/download/v2.5.11/milvus-standalone-docker-compose.yml -O docker-compose.yml
docker-compose up -d
# 检查服务状态
docker-compose ps
2.4.2 集合Schema设计
菜谱向量的Schema需要包含图数据库的关联信息：
python复制from pymilvus import CollectionSchema, FieldSchema, DataType
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
FieldSchema(name="node_id", dtype=DataType.VARCHAR, max_length=64), # 对应Neo4j节点ID
FieldSchema(name="recipe_name", dtype=DataType.VARCHAR, max_length=128),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768)
]
schema = CollectionSchema(fields, description="Recipe Vector Collection")
2.5 环境变量配置
项目通过.env文件管理敏感配置：
ini复制# Neo4j配置
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=all-in-rag
NEO4J_DATABASE=neo4j
# Milvus配置
MILVUS_HOST=localhost
MILVUS_PORT=19530
# LLM API配置
MOONSHOT_API_KEY=your_api_key_here
安全提示：务必在.gitignore中添加.env，避免密钥泄露。生产环境应使用Vault等专业密钥管理工具。
3. 图数据建模与Neo4j集成
3.1 数据转换流程
原始菜谱数据通常以Markdown格式存储，我们需要通过LLM进行信息抽取：
-
实体识别：使用LLM从Markdown中提取菜谱、食材、步骤等实体
python复制def extract_entities(markdown_text): prompt = f"""从以下菜谱中提取实体： {markdown_text} 按JSON格式返回，包含recipe、ingredients、steps等字段""" response = llm.invoke(prompt) return json.loads(response) -
CSV生成：将实体转换为Neo4j可导入的格式
- nodes.csv包含所有节点的id、label和属性
- relationships.csv记录源节点、目标节点和关系类型
-
数据校验：检查食材用量单位是否合法、步骤序号是否连续等
3.2 图数据模型实现
3.2.1 核心关系设计
| 关系类型 | 关联实体 | 核心属性 | 示例 |
|---|---|---|---|
| REQUIRES | Recipe → Ingredient | amount, unit | 鱼香肉丝-REQUIRES[200g]->里脊肉 |
| CONTAINS_STEP | Recipe → CookingStep | step_order | 鱼香肉丝-CONTAINS_STEP[1]->"切肉丝" |
| BELONGS_TO_CATEGORY | Recipe → RecipeCategory | - | 鱼香肉丝-BELONGS_TO_CATEGORY->川菜 |
| HAS_DIFFICULTY_LEVEL | Recipe → DifficultyLevel | - | 鱼香肉丝-HAS_DIFFICULTY_LEVEL->入门 |
3.2.2 三层检索结构
我们设计了Root -> 实体类型 -> 实例节点的三层结构：
-
根节点：作为唯一入口
cypher复制CREATE (root:Root {name:'CookingRoot'}) -
类型节点：分类组织各类实体
cypher复制CREATE (rt:RecipeType {name:'Recipe'}) CREATE (it:IngredientType {name:'Ingredient'}) MERGE (root)-[:CONNECTS_TO]->(rt) MERGE (root)-[:CONNECTS_TO]->(it) -
实例连接：具体业务数据
cypher复制MATCH (rt:RecipeType) CREATE (rt)-[:HAS_INSTANCE]->(r:Recipe {name:'鱼香肉丝'})
这种设计支持高效的类型级查询，如"查找所有食材节点"：
cypher复制MATCH (:Root)-[:CONNECTS_TO]->(:IngredientType)-[:HAS_INSTANCE]->(i:Ingredient)
RETURN i
3.3 数据导入与索引
3.3.1 批量导入
使用neo4j-admin import工具实现高效批量导入：
bash复制neo4j-admin database import full \
--nodes=import/nodes.csv \
--relationships=import/relationships.csv \
--delimiter="," \
--array-delimiter="|"
3.3.2 Python封装
对于增量更新，我们封装了GraphDataPreparationModule类：
python复制class GraphDataPreparationModule:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def load_csv_data(self, node_path, rel_path):
with self.driver.session() as session:
session.run("""
LOAD CSV WITH HEADERS FROM $node_path AS row
CALL apoc.create.node([row.label], apoc.map.clean(row, ['label'], []))
YIELD node SET node.id = row.id
""", {"node_path": node_path})
session.run("""
LOAD CSV WITH HEADERS FROM $rel_path AS row
MATCH (s) WHERE s.id = row.source
MATCH (t) WHERE t.id = row.target
CALL apoc.create.relationship(s, row.type, apoc.map.clean(row, ['source','target','type'], []), t)
YIELD rel RETURN count(rel)
""", {"rel_path": rel_path})
3.4 图数据转RAG文档
3.4.1 文档生成策略
将图数据转换为LangChain可处理的Document对象：
python复制def graph_to_document(recipe_node):
# 获取关联数据
ingredients = get_related_ingredients(recipe_node)
steps = get_related_steps(recipe_node)
# 构建Markdown内容
content = f"""## {recipe_node['name']}
**烹饪时间**: {recipe_node['cooking_time']}分钟 | **难度**: {recipe_node['difficulty']}
### 食材清单
{ingredients.to_markdown()}
### 烹饪步骤
{steps.to_markdown()}
"""
return Document(
page_content=content,
metadata={
"node_id": recipe_node.id,
"type": "recipe"
}
)
3.4.2 分块策略
根据内容类型采用不同分块方式：
- 短文本：完整保留单个步骤或食材条目
- 长文本：按语义划分：
- 菜谱描述单独成块
- 食材列表每3-5项一组
- 烹饪步骤每2-3步一段
python复制from langchain.text_splitter import RecursiveCharacterTextSplitter
recipe_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\n## ", "\n### ", "\n\n", "\n", "。"]
)
4. Milvus索引构建
4.1 向量化流程
图RAG系统的向量索引构建分为五个阶段：
- 图数据提取：从Neo4j导出结构化数据
- 文档生成：转换为自然语言格式
- 智能分块：按语义切分文本
- 文本向量化：使用Embedding模型处理
- 索引落地：存入Milvus并建立索引
4.2 模块化实现
MilvusIndexConstructionModule核心方法：
python复制class MilvusIndexConstructionModule:
def __init__(self, host, port, collection_name):
self.conn = connections.connect(host=host, port=port)
self.collection = self._init_collection(collection_name)
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
def build_vector_index(self, chunks):
# 生成向量
texts = [chunk.page_content for chunk in chunks]
embeddings = self.embedding_model.encode(texts)
# 准备插入数据
entities = [
[i for i in range(len(chunks))], # 主键
[chunk.metadata.get("node_id", "") for chunk in chunks],
[chunk.metadata.get("recipe_name", "") for chunk in chunks],
embeddings.tolist()
]
# 插入数据
insert_result = self.collection.insert(entities)
# 创建IVF_FLAT索引
index_params = {
"metric_type": "L2",
"index_type": "IVF_FLAT",
"params": {"nlist": 128}
}
self.collection.create_index("embedding", index_params)
# 加载集合到内存
self.collection.load()
return True
4.3 混合检索策略
结合向量搜索和图遍历的优势：
-
初步检索：在Milvus中执行语义搜索
python复制def vector_search(query, top_k=3): query_embedding = model.encode(query) search_params = {"metric_type": "L2", "params": {"nprobe": 10}} results = collection.search( data=[query_embedding], anns_field="embedding", param=search_params, limit=top_k, output_fields=["node_id"] ) return [hit.entity.get("node_id") for hit in results[0]] -
图扩展：根据节点ID在Neo4j中获取关联信息
cypher复制MATCH (n)-[r]-(m) WHERE n.nodeId IN $node_ids RETURN n, r, m -
结果融合：综合相关度和图结构信息排序
5. 系统实现与效果展示
5.1 核心类设计
AdvancedGraphRAGSystem类架构：
mermaid复制classDiagram
class AdvancedGraphRAGSystem {
+config: GraphRAGConfig
+data_module: GraphDataPreparationModule
+index_module: MilvusIndexConstructionModule
+generation_module: GenerationIntegrationModule
+traditional_retrieval: HybridRetrievalModule
+graph_rag_retrieval: GraphRAGRetrieval
+query_router: IntelligentQueryRouter
+initialize_system()
+build_knowledge_base()
+ask_question_with_routing()
}
5.2 智能路由策略
查询路由器根据以下特征选择检索策略：
-
查询复杂度分析：
- 简单查询（如"红烧肉怎么做"）：纯向量检索
- 复杂查询（如"不用烤箱的30分钟西餐"）：图遍历+向量混合
-
关系密集度评估：
python复制def analyze_relationship_intensity(query): # 检查是否包含关系型关键词 rel_keywords = ["需要", "搭配", "替代", "不用"] return sum(1 for kw in rel_keywords if kw in query) / len(query) -
策略选择矩阵：
| 查询类型 | 推荐策略 | 原因 |
|---|---|---|
| 简单事实查询 | 传统向量检索 | 响应速度快 |
| 多条件过滤 | 图检索 | 支持属性过滤 |
| 关系推理 | 图RAG混合 | 需要多跳遍历 |
| 模糊语义 | 向量+图融合 | 兼顾语义和结构 |
5.3 效果展示
5.3.1 用户界面
Streamlit构建的交互界面包含：
- 系统状态面板：显示知识库统计信息
- 问答输入区：支持自然语言提问
- 路由解释模块：可视化检索策略选择过程
5.3.2 典型查询示例
-
简单查询："鱼香肉丝怎么做"
- 策略：传统向量检索
- 响应：直接返回最匹配的菜谱文档
-
复杂查询："推荐不需要炒锅的30分钟家常菜"
- 策略：图RAG混合
- 处理流程：
a. 在Milvus中查找"家常菜"相关菜谱
b. 通过Neo4j过滤出烹饪时间≤30分钟的
c. 排除需要炒锅的菜谱
d. 按关联食材常见度排序
-
关系查询："可以用什么替代里脊肉做鱼香肉丝"
- 策略：图遍历优先
- 处理流程：
a. 找到鱼香肉丝菜谱节点
b. 获取所有REQUIRES关系中的食材
c. 查找具有"可替代"关系的其他食材
d. 返回替代建议及用量调整方案
6. 性能优化与实践经验
6.1 图查询优化技巧
-
查询计划分析：使用
EXPLAIN检查Cypher执行计划cypher复制EXPLAIN MATCH (r:Recipe)-[:REQUIRES]->(i:Ingredient) WHERE i.name CONTAINS '牛肉' RETURN r.name -
APOC优化：利用APOC库的并行执行
cypher复制CALL apoc.cypher.parallel2( 'MATCH (r:Recipe) WHERE r.cooking_time < $time RETURN r', {time: 30}, 'r' ) YIELD value RETURN value.r -
缓存利用：对热点查询配置缓存
python复制@lru_cache(maxsize=100) def get_common_ingredients(): query = """ MATCH (i:Ingredient) WHERE i.is_common = true RETURN i.name ORDER BY i.name """ return session.run(query).data()
6.2 向量检索调优
-
索引类型选择：
- IVF_FLAT：平衡精度和速度，适合开发环境
- HNSW：高召回率，适合生产环境
-
搜索参数调整：
python复制search_params = { "metric_type": "IP", # 内积相似度 "params": { "ef": 32, # HNSW的搜索范围 "nprobe": 16 # IVF的搜索桶数量 } } -
批量处理：减少网络往返
python复制# 批量插入1000条数据比单条插入快10倍以上 collection.insert([entities]*1000)
6.3 踩坑记录
-
Neo4j连接泄漏：
- 现象：长时间运行后连接耗尽
- 解决：确保每次查询后关闭session
python复制def run_query(query): with driver.session() as session: return session.run(query).data() -
向量维度不匹配：
- 现象：Milvus报错"Dimension mismatch"
- 解决：统一使用模型原始维度(如all-MiniLM-L6-v2是384维)
-
中文分词问题：
- 现象：Embedding模型对专业厨艺术语处理不佳
- 解决：使用领域适配的模型或添加自定义词汇表
7. 扩展应用与未来改进
7.1 多模态扩展
-
菜谱图片向量化：使用CLIP模型处理菜品图片
python复制from PIL import Image import clip model, preprocess = clip.load("ViT-B/32") image = preprocess(Image.open("dish.jpg")).unsqueeze(0) image_features = model.encode_image(image) -
视频步骤分析：提取烹饪视频关键帧
7.2 实时学习机制
-
用户反馈学习：记录问题-答案对的修正记录
cypher复制CREATE (f:Feedback { query: "如何做溏心蛋", original_answer: "...", corrected_answer: "...", timestamp: datetime() }) -
图谱动态更新：定期同步用户新增内容
python复制def sync_user_content(): new_recipes = get_user_submissions() for recipe in new_recipes: create_recipe_node(recipe) create_relationships(recipe) update_milvus_index()
7.3 生产环境部署建议
-
高可用架构：
- Neo4j集群：3节点Causal Cluster
- Milvus集群：分片+查询节点分离
-
性能监控：
- Prometheus收集指标
- Grafana展示关键指标：
- 查询响应时间
- 系统资源占用
- 缓存命中率
-
灾备方案：
- 定期导出Neo4j数据快照
- Milvus集合定时备份到对象存储
这个图RAG系统通过深度整合图数据库的关系推理能力和向量数据库的语义理解优势，为烹饪领域提供了智能化的知识检索方案。在实际应用中，开发者需要根据具体场景调整数据模型和检索策略，持续优化系统性能。