00 快速复现清单：掌柜智库 — 知识图谱 RAG 全栈系统

🏃 本文档目标：复制粘贴 + 回车 = 跑通完整项目 ❌ 不讲原理——原理在 01~07 篇 💬 卡住了？ 把错误信息复制给 AI："按清单第 X 步卡住了，错误信息：..."

0. 前置检查

bash

# 确认以下工具已安装
docker --version           # Docker ≥ 24
python --version           # Python ≥ 3.12
git --version              # Git ≥ 2.40

# 确认以下端口未被占用
# 19530(Milvus) 7474+7687(Neo4J) 9000+9001(MinIO) 6379(Redis) 27017(MongoDB) 3306(MySQL)
# ✅ 前置检查完成

1. 阶段一：环境部署（≈15 min）

Step 1.1 创建项目根目录

bash

mkdir shopkeeper_brain && cd shopkeeper_brain

Step 1.2 创建 docker-compose.yml

yaml

# → docker-compose.yml
version: '3.8'
services:
  milvus:
    image: milvusdb/milvus:latest
    ports: ["19530:19530"]
    volumes: ["./data/milvus:/var/lib/milvus"]
    environment:
      ETCD_EMBEDDED: "true"
  neo4j:
    image: neo4j:5
    ports: ["7474:7474", "7687:7687"]
    environment:
      NEO4J_AUTH: neo4j/password
  minio:
    image: minio/minio
    ports: ["9000:9000", "9001:9001"]
    command: server /data --console-address :9001
    volumes: ["./data/minio:/data"]
  redis:
    image: redis:7
    ports: ["6379:6379"]
  mongodb:
    image: mongo:7
    ports: ["27017:27017"]
    volumes: ["./data/mongo:/data/db"]
  mysql:
    image: mysql:8
    ports: ["3306:3306"]
    environment:
      MYSQL_ROOT_PASSWORD: root
    volumes: ["./data/mysql:/var/lib/mysql"]

bash

# 启动所有服务
docker compose up -d

# 验证各服务
curl -s http://localhost:19530 > /dev/null && echo "✅ Milvus OK" || echo "❌ Milvus"
curl -s http://localhost:7474 > /dev/null && echo "✅ Neo4J OK" || echo "❌ Neo4J"
curl -s http://localhost:9001 > /dev/null && echo "✅ MinIO OK" || echo "❌ MinIO"
curl -s http://localhost:6379 > /dev/null && echo "✅ Redis OK" || echo "❌ Redis"
# ✅ 所有服务启动成功

2. 阶段二：项目骨架（≈10 min）

Step 2.1 创建目录结构

bash

mkdir -p knowledge/{api,core,processor/import_process/nodes,prompts/{upload,query},schema,services,utils,test/{import,query},temp_data}
mkdir -p data/{milvus,minio,mongo,mysql}
touch knowledge/__init__.py
touch knowledge/api/__init__.py
touch knowledge/core/__init__.py
touch knowledge/processor/__init__.py
touch knowledge/processor/import_process/__init__.py
touch knowledge/processor/import_process/nodes/__init__.py
touch knowledge/utils/__init__.py
touch knowledge/schema/__init__.py
touch knowledge/services/__init__.py
# ✅ 目录结构创建完成

Step 2.2 创建 BaseNode 基类

python

# → knowledge/processor/import_process/base.py
from abc import ABC, abstractmethod
import logging
from knowledge.processor.import_process.config import ImportConfig, get_config
from knowledge.processor.import_process.exceptions import ImportProcessError

class BaseNode(ABC):
    name: str = "base_node"
    def __init__(self, config: ImportConfig = None):
        self.config = config or get_config()
        self.logger = logging.getLogger(f"import.{self.name}")
    def __call__(self, state):
        try:
            self.logger.info(f"--- {self.name} 开始 ---")
            result = self.process(state)
            self.logger.info(f"--- {self.name} 完成 ---")
            return result
        except Exception as e:
            raise ImportProcessError(str(e), node_name=self.name, cause=e)
    @abstractmethod
    def process(self, state): pass

bash

# ✅ BaseNode 创建完成

Step 2.3 创建配置管理

python

# → knowledge/processor/import_process/config.py
from dataclasses import dataclass, field
import os
from dotenv import load_dotenv
load_dotenv()

@dataclass
class ImportConfig:
    openai_api_base: str = field(default_factory=lambda: os.getenv("OPENAI_API_BASE", ""))
    openai_api_key: str = field(default_factory=lambda: os.getenv("OPENAI_API_KEY", ""))
    milvus_url: str = field(default_factory=lambda: os.getenv("MILVUS_URL", ""))
    neo4j_uri: str = field(default_factory=lambda: os.getenv("NEO4J_URI", ""))
    neo4j_username: str = field(default_factory=lambda: os.getenv("NEO4J_USERNAME", "neo4j"))
    neo4j_password: str = field(default_factory=lambda: os.getenv("NEO4J_PASSWORD", "password"))
    minio_endpoint: str = field(default_factory=lambda: os.getenv("MINIO_ENDPOINT", ""))
    minio_access_key: str = field(default_factory=lambda: os.getenv("MINIO_ACCESS_KEY", ""))
    minio_secret_key: str = field(default_factory=lambda: os.getenv("MINIO_SECRET_KEY", ""))
    minio_bucket: str = field(default_factory=lambda: os.getenv("MINIO_BUCKET_NAME", ""))
    embedding_dim: int = field(default_factory=lambda: int(os.getenv("EMBEDDING_DIM", "1024")))

_config = None
def get_config() -> ImportConfig:
    global _config
    if _config is None: _config = ImportConfig()
    return _config

bash

# ✅ config 创建完成

Step 2.4 创建异常体系

python

# → knowledge/processor/import_process/exceptions.py
class ImportProcessError(Exception):
    def __init__(self, message="", node_name="", cause=None):
        self.node_name = node_name
        self.cause = cause
        super().__init__(f"[{node_name}] {message}" if node_name else message)

class ValidationError(ImportProcessError):
    pass

bash

# ✅ exceptions 创建完成

Step 2.5 创建 State 定义

python

# → knowledge/processor/import_process/state.py
from typing import TypedDict, List

class ImportGraphState(TypedDict, total=False):
    task_id: str
    import_file_path: str
    file_dir: str
    file_title: str
    is_pdf_read_enabled: bool
    is_md_read_enabled: bool
    pdf_path: str
    md_path: str
    md_content: str
    chunks: List

bash

# ✅ state 创建完成

Step 2.6 创建 EntryNode

python

# → knowledge/processor/import_process/nodes/entry_node.py
from pathlib import Path
from knowledge.processor.import_process.base import BaseNode
from knowledge.processor.import_process.state import ImportGraphState
from knowledge.processor.import_process.exceptions import ValidationError

class EntryNode(BaseNode):
    name = "entry"
    def process(self, state: ImportGraphState) -> ImportGraphState:
        file_dir = state.get('file_dir')
        import_file_path = state.get('import_file_path')
        if not file_dir or not import_file_path:
            raise ValidationError("文件路径不存在", self.name)
        suffix = Path(import_file_path).suffix.lower()
        if suffix == '.pdf':
            state['is_pdf_read_enabled'] = True
            state['pdf_path'] = import_file_path
        elif suffix == '.md':
            state['is_md_read_enabled'] = True
            state['md_path'] = import_file_path
        else:
            raise ValidationError(f"不支持的文件类型: {suffix}", self.name)
        state['file_title'] = Path(import_file_path).stem
        return state

bash

# ✅ EntryNode 创建完成

Step 2.7 创建 main_graph

python

# → knowledge/processor/import_process/main_graph.py
from langgraph.graph import StateGraph
from langgraph.constants import END
from knowledge.processor.import_process.state import ImportGraphState
from knowledge.processor.import_process.nodes.entry_node import EntryNode

def import_router(state):
    if state.get('is_pdf_read_enabled'): return "pdf_to_md_node"
    elif state.get('is_md_read_enabled'): return "md_img_node"
    return END

def create_import_graph():
    builder = StateGraph(ImportGraphState)
    builder.add_node("entry_node", EntryNode())
    builder.add_conditional_edges("entry_node", import_router, {
        "md_img_node": "md_img_node", "pdf_to_md_node": "pdf_to_md_node", END: END
    })
    return builder.compile()

bash

# ✅ main_graph 创建完成

Step 2.8 创建 .env

env

# → .env
OPENAI_API_BASE=https://api.openai.com/v1
OPENAI_API_KEY=sk-your-key-here
MODEL=gpt-4o-mini
MILVUS_URL=http://localhost:19530
NEO4J_URI=bolt://localhost:7687
NEO4J_PASSWORD=password
MINIO_ENDPOINT=localhost:9000
MINIO_ACCESS_KEY=minioadmin
MINIO_SECRET_KEY=minioadmin
MINIO_BUCKET_NAME=shopkeeper
EMBEDDING_DIM=1024

bash

# ✅ .env 创建完成

Step 2.9 验证骨架

bash

# 安装依赖
pip install langgraph langchain-openai pydantic python-dotenv

# 测试 EntryNode
python -c "
from knowledge.processor.import_process.main_graph import create_import_graph
graph = create_import_graph()
result = graph.invoke({'file_dir': '/tmp', 'import_file_path': '/tmp/test.pdf'})
assert result.get('is_pdf_read_enabled') == True
assert result.get('file_title') == 'test'
print('✅ 骨架验证通过')
"

3. 阶段三：文档处理节点（≈15 min）—— day02~05

Step 3.1 PDF→MD 节点

python

# → knowledge/processor/import_process/nodes/pdf_to_md_node.py
# 使用 MinerU 将 PDF 转为 Markdown
# 前提：已安装 MinerU（pip install mineru）
# 详细配置见 MinerU 官方文档
class PdfToMdNode(BaseNode):
    name = "pdf_to_md"
    def process(self, state):
        pdf_path = state.get('pdf_path')
        # mineru CLI: mineru --pdf <pdf_path> --output <md_dir>
        # 将生成的 md 路径写入 state['md_path']
        return state

Step 3.2 图片处理节点

python

# → knowledge/processor/import_process/nodes/md_img_node.py
# 扫描 Markdown 中的图片引用
# 使用 VLM 模型为每张图片生成摘要
# 将图片上传到 MinIO
# 替换 MD 中的图片引用为 MinIO URL + 摘要
class MarkDownImageNode(BaseNode):
    name = "md_img"
    def process(self, state):
        # 1. 读取 MD 内容，提取所有图片路径
        # 2. 调用 VLM API 为每张图生成描述
        # 3. 上传图片到 MinIO
        # 4. 替换 MD 中的图片引用
        return state

Step 3.3 文档切分节点

python

# → knowledge/processor/import_process/nodes/document_split_node.py
# 按 Markdown 一级标题切分文档
# 短内容和相邻合并
class DocumentSplitNode(BaseNode):
    name = "document_split"
    def process(self, state):
        md_content = state.get('md_content', '')
        # 按标题切分（RegEx），每个 chunk 保留标题路径
        # 短 chunk（<500 字）与下一个合并
        state['chunks'] = chunks
        return state

bash

# ✅ 文档处理节点创建完成

4. 阶段四：AI 处理节点（≈15 min）—— day06~07

Step 4.1 商品名识别节点

python

# → knowledge/processor/import_process/nodes/item_name_recognition_node.py
# 调用 LLM 从文档内容中提取商品名称
# 将商品名向量化并存入 Milvus（item_name 集合）
class ItemNameRecognitionNode(BaseNode):
    name = "item_name_rec"
    def process(self, state):
        # 1. 选取前 N 个 chunks 作为 LLM 上下文
        # 2. LLM 提取商品名称列表
        # 3. bge-m3 计算每个商品名的稠密+稀疏向量
        # 4. 存入 Milvus item_name 集合
        return state

Step 4.2 批量嵌入节点

python

# → knowledge/processor/import_process/nodes/bge_embedding_chunks_node.py
# 对所有文档切片批量计算 bge-m3 嵌入向量
class BgeEmbeddingChunksNode(BaseNode):
    name = "bge_embedding"
    def process(self, state):
        chunks = state.get('chunks', [])
        # 使用 bge-m3 模型批量计算 embedding
        # 每 chunk 输出: {text, dense_vector, sparse_vector, metadata}
        return state

Step 4.3 Milvus 入库节点

python

# → knowledge/processor/import_process/nodes/import_milvus_node.py
# 将嵌入后的 chunks 存入 Milvus（chunks 集合）
class ImportMilvusNode(BaseNode):
    name = "import_milvus"
    def process(self, state):
        # 1. 创建 Milvus collection（含稠密+稀疏+标量字段，注意标量索引坑）
        # 2. 批量插入向量数据
        return state

bash

# ✅ AI 处理节点创建完成

5. 阶段五：知识图谱节点（≈10 min）—— day08~10

Step 5.1 创建 Neo4J 节点

python

# → knowledge/processor/import_process/nodes/kg_graph_node.py
# 调用 LLM 从文档中抽取实体和关系
# 写入 Neo4J 图数据库
class KnowledgeGraphNode(BaseNode):
    name = "kg_node"
    def process(self, state):
        chunks = state.get('chunks', [])
        # 1. LLM 抽取实体（商品、品牌、属性等）和关系
        # 2. 用 Cypher 写入 Neo4J
        # CREATE (n:Entity {name: '...', type: 'product'})
        # MATCH (a),(b) CREATE (a)-[r:RELATED_TO]->(b)
        return state

bash

# 验证知识图谱
# 浏览器访问 http://localhost:7474，执行：
# MATCH (n) RETURN n LIMIT 25
# ✅ 看到实体节点即成功

6. 阶段六：查询流程（≈20 min）—— day11~18

Step 6.1 查询流程 main_graph

python

# → knowledge/processor/query_process/main_graph.py
from langgraph.graph import StateGraph
from langgraph.constants import START, END

class QueryGraphState(TypedDict, total=False):
    query: str
    item_name: str
    vector_results: list
    kg_results: list
    mcp_results: list
    fused_results: list
    reranked_results: list
    answer: str

def create_query_graph():
    builder = StateGraph(QueryGraphState)
    # 节点：item_name_confirm → vector/kg/mcp 三路并行 → rrf → rerank → answer
    return builder.compile()

Step 6.2 三路检索验证命令

python

# 向量检索：Milvus 相似度搜索
# 知识图谱：Neo4J Cypher 查询
# MCP 检索：外部工具调用

Step 6.3 RRF + Reranker

python

# RRF: Reciprocal Rank Fusion
# score = sum(1/(k + rank_i)) for each document in each result list
def rrf_fuse(results: list[list[dict]], k: int = 60) -> list[dict]:
    scores = {}
    for rank_list in results:
        for rank, doc in enumerate(rank_list):
            doc_id = doc['id']
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

# Reranker: 使用 bge-reranker 对 RRF 结果重排序

Step 6.4 启动 API 服务

bash

# 启动 FastAPI 后端
uvicorn knowledge.api.import_router:app --host 0.0.0.0 --port 8000

# 验证 API
curl -X POST "http://localhost:8000/chat" -H "Content-Type: application/json" \
  -d '{"query": "万用表怎么使用？"}'
# ✅ 返回 JSON 格式的回答

7. 验证清单

编号	验证项	命令	预期结果
1	Docker 服务全部启动	`docker ps --format "{{.Names}}"`	看到 milvus/neo4j/minio/redis/mongo/mysql
2	骨架节点正常	`python -c "from knowledge.processor.import_process.main_graph import create_import_graph; print('OK')"`	输出 OK
3	入口节点路由	见 Step 2.9	is_pdf_read_enabled=True
4	Milvus 集合	通过 Attu 或 Python 客户端	能看到 chunks 和 item_name 集合
5	Neo4J 节点	`MATCH (n) RETURN count(n)`	count > 0
6	端到端导入	触发完整导入流程	PDF → Milvus + Neo4J 均有数据
7	端到端查询	`curl /chat -d '{"query":"万用表"}'`	返回含知识的回答

00 快速复现清单：掌柜智库 — 知识图谱 RAG 全栈系统 ​

0. 前置检查 ​

1. 阶段一：环境部署（≈15 min） ​

Step 1.1 创建项目根目录 ​

Step 1.2 创建 docker-compose.yml ​

2. 阶段二：项目骨架（≈10 min） ​

Step 2.1 创建目录结构 ​

Step 2.2 创建 BaseNode 基类 ​

Step 2.3 创建配置管理 ​

Step 2.4 创建异常体系 ​

Step 2.5 创建 State 定义 ​

Step 2.6 创建 EntryNode ​

Step 2.7 创建 main_graph ​

Step 2.8 创建 .env ​

Step 2.9 验证骨架 ​

3. 阶段三：文档处理节点（≈15 min）—— day02~05 ​

Step 3.1 PDF→MD 节点 ​

Step 3.2 图片处理节点 ​

Step 3.3 文档切分节点 ​

4. 阶段四：AI 处理节点（≈15 min）—— day06~07 ​

Step 4.1 商品名识别节点 ​

Step 4.2 批量嵌入节点 ​

Step 4.3 Milvus 入库节点 ​

5. 阶段五：知识图谱节点（≈10 min）—— day08~10 ​

Step 5.1 创建 Neo4J 节点 ​

6. 阶段六：查询流程（≈20 min）—— day11~18 ​

Step 6.1 查询流程 main_graph ​

Step 6.2 三路检索验证命令 ​

Step 6.3 RRF + Reranker ​

Step 6.4 启动 API 服务 ​

7. 验证清单 ​

00 快速复现清单：掌柜智库 — 知识图谱 RAG 全栈系统

0. 前置检查

1. 阶段一：环境部署（≈15 min）

Step 1.1 创建项目根目录

Step 1.2 创建 docker-compose.yml

2. 阶段二：项目骨架（≈10 min）

Step 2.1 创建目录结构

Step 2.2 创建 BaseNode 基类

Step 2.3 创建配置管理

Step 2.4 创建异常体系

Step 2.5 创建 State 定义

Step 2.6 创建 EntryNode

Step 2.7 创建 main_graph

Step 2.8 创建 .env

Step 2.9 验证骨架

3. 阶段三：文档处理节点（≈15 min）—— day02~05

Step 3.1 PDF→MD 节点

Step 3.2 图片处理节点

Step 3.3 文档切分节点

4. 阶段四：AI 处理节点（≈15 min）—— day06~07

Step 4.1 商品名识别节点

Step 4.2 批量嵌入节点

Step 4.3 Milvus 入库节点

5. 阶段五：知识图谱节点（≈10 min）—— day08~10

Step 5.1 创建 Neo4J 节点

6. 阶段六：查询流程（≈20 min）—— day11~18

Step 6.1 查询流程 main_graph

Step 6.2 三路检索验证命令

Step 6.3 RRF + Reranker

Step 6.4 启动 API 服务

7. 验证清单