阶段 0：部署与项目骨架（对应课程 day01）

当前状态：从零开始 → Docker 环境就绪 + 项目目录结构 + 3 个基础设施节点

知识标记总览：

知识点	出现次数	本阶段扮演的角色
LangGraph StateGraph	🟢 第 2 次（Ch18→Ch19）	导入流程编排，核心骨架
BaseNode 抽象类	🔴 第 1 次（全新设计模式）	所有节点的统一基类（含日志/追踪/异常处理）
ImportGraphState	🟢 第 2 次（Ch18 State→Ch19）	8 个节点共享的全局状态
EntryNode 入口节点	🔴 第 1 次（本项目特有）	文件类型判断 + 路由
LLM 客户端配置	🟢 第 2 次（Ch16→Ch19）	dataclass 的配置管理模式
Docker Compose	🔴 第 1 次（正式引入）	6 个服务的全链路编排

人/AI 协作标记：🤖 = AI 可生成 / 👤 = 手动必须 / 🤝 = 协作完成

0.1 阶段起始状态：空项目

这是第 19 章学习的第 0 天——你面前是一个空目录，需要决定技术栈和环境。

shopkeeper_brain/                          ← 空目录
└── README.md                              ← 唯一的文件

决策 1：为什么用 LangGraph 而非 LCEL

对比项	LCEL（Ch16 方式）	LangGraph（本项目方式）
表达方式	`\|` 线性管道	StateGraph + Node + Edge
状态管理	无，Node 之间无共享 State	TypedDict 定义的全局 State
节点间通信	仅通过返回值串联	共享同一个 State 字典
错误恢复	失败从头重跑	Checkpointer 从中断恢复
任务追踪	无	BaseNode 内置 task_id + 日志 + 进度推送

结论：导入流程涉及 8 个节点的状态共享和递进式数据处理，StateGraph 是比 LCEL 更自然的抽象。

0.2 环境部署 👤 手动（约 30 min）

服务清单

服务	版本	端口	用途	部署方式
Milvus	2.5+	19530	向量数据库	Docker
Neo4J	5+	7474 / 7687	图数据库	Docker
MinIO	latest	9000 / 9001	对象存储（图片）	Docker
Redis	7+	6379	缓存	Docker
MongoDB	7+	27017	对话历史	Docker
MySQL	8+	3306	元数据（掌柜问数用）	Docker

全量 docker-compose.yml 🤖 AI 生成

yaml

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
      MYSQL_DATABASE: meta
    volumes: ["./data/mysql:/var/lib/mysql"]

部署验证：启动后依次确认各服务端口可访问。Milvus 通过 http://localhost:19530（返回空即正常），Neo4J 通过 http://localhost:7474（浏览器访问）。

0.3 项目骨架搭建（增量）

阶段结束时的目录树 🟢 新增高亮

shopkeeper_brain/
├── .env                                ← 🔴 新增：环境变量
├── knowledge/
│   ├── __init__.py
│   ├── api/
│   │   ├── __init__.py
│   │   └── import_router.py            ← 🔴 新增：导入 API 路由
│   ├── core/
│   │   ├── __init__.py
│   │   ├── deps.py                     ← 🔴 新增：依赖注入
│   │   └── paths.py                    ← 🔴 新增：路径管理
│   ├── processor/
│   │   ├── __init__.py
│   │   └── import_process/
│   │       ├── __init__.py
│   │       ├── base.py                 ← 🔴 新增：BaseNode 基类 ← ★核心
│   │       ├── state.py                ← 🔴 新增：ImportGraphState ← ★核心
│   │       ├── config.py               ← 🔴 新增：配置管理
│   │       ├── exceptions.py           ← 🔴 新增：异常体系
│   │       ├── main_graph.py           ← 🔴 新增：LangGraph 图定义
│   │       └── nodes/
│   │           ├── __init__.py
│   │           └── entry_node.py       ← 🔴 新增：入口节点 ← ★核心
│   └── utils/
│       ├── __init__.py
│       └── task_util.py                ← 🔴 新增：任务追踪工具
└── requirements.txt                    ← 🔴 新增

本阶段共新增 15 个文件，未来阶段在此基础上增量添加。

0.4 核心代码骨架 🤖 AI 辅助

① BaseNode：所有节点的统一基类

python

from abc import ABC, abstractmethod
from knowledge.processor.import_process.config import ImportConfig, get_config
from knowledge.processor.import_process.exceptions import ImportProcessError
from knowledge.utils.task_util import add_running_task, add_done_task

class BaseNode(ABC):
    """所有导入节点的基类——提供统一的 __call__ 入口 + 日志 + 任务追踪"""
    
    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) -> dict:
        """LangGraph 调用的统一入口"""
        task_id = state.get("task_id", "")
        try:
            self.logger.info(f"--- {self.name} 开始 ---")
            if task_id: add_running_task(task_id, self.name)
            result = self.process(state)
            self.logger.info(f"--- {self.name} 完成 ---")
            if task_id: add_done_task(task_id, self.name)
            return result
        except Exception as e:
            raise ImportProcessError(str(e), node_name=self.name, cause=e)

    @abstractmethod
    def process(self, state): ...

🔑 设计决策：为什么不直接写函数作为 Node，而要封装成类？
每个节点需要自己的 logger（self.logger），函数做不到
每个节点需要自己的 config（self.config），函数做不到
每个节点需要统一的异常处理和进度追踪（__call__ 里的 try-except）
继承比组合更自然——8 个导入节点共享同一套模板方法

② ImportGraphState：8 个节点共享的全局状态

python

from typing import TypedDict, List

class ImportGraphState(TypedDict, total=False):
    """导入流程的全局状态——随着节点增加字段会逐步扩展"""
    
    # 任务标识
    task_id: str
    
    # 控制标志
    is_md_read_enabled: bool   # 是否启用 MD 读取
    is_pdf_read_enabled: bool  # 是否启用 PDF 读取
    
    # 路径信息
    import_file_path: str      # 导入文件路径
    file_dir: str              # 文件所在目录
    pdf_path: str              # PDF 文件路径
    md_path: str              # 转换后的 Markdown 路径
    file_title: str           # 文件标题
    
    # 处理中间数据（后续阶段会扩充）
    md_content: str           # Markdown 文档内容
    chunks: List              # 文档切片列表（day04-05 加入）

🔑 设计决策：为什么用 total=False？
字段是逐步填充的——entry_node 只填路径信息，后续节点只填自己的部分
使用 total=False 后，字段可以不存在，节点获取时自己加判空

③ EntryNode：导入流水线的入口

python

class EntryNode(BaseNode):
    """判断上传的文件是 .pdf 还是 .md，路由到不同处理链路"""
    
    name = "entry"

    def process(self, state: ImportGraphState) -> ImportGraphState:
        file_dir = state.get('file_dir')
        import_file_path = state.get('import_file_path')
        
        # 1. 校验路径
        if not file_dir or not import_file_path:
            raise ValidationError("文件路径不存在", self.name)
        
        # 2. 判断后缀
        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)
        
        # 3. 提取标题
        state['file_title'] = Path(import_file_path).stem
        return state

④ main_graph：LangGraph 图定义

python

from langgraph.graph import StateGraph
from knowledge.processor.import_process.nodes.entry_node import EntryNode

def import_router(state: ImportGraphState) -> str:
    """条件边：判断走 PDF 还是 MD 链路"""
    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():
    """构建导入流水线的 StateGraph"""
    builder = StateGraph(ImportGraphState)
    
    # 添加节点（本阶段只有入口节点）
    builder.add_node("entry_node", EntryNode())
    # 后续阶段逐步添加：PdfToMdNode, MarkDownImageNode, DocumentSplitNode...
    
    # 添加条件边
    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()

⑤ .env 配置模板 🤖 AI 辅助填充

env

# LLM 配置
OPENAI_API_BASE=https://api.openai.com/v1
OPENAI_API_KEY=sk-xxxx
MODEL=gpt-4o-mini
VL_MODEL=qwen-vl-plus          # 视觉模型（VLM 图片摘要用）
ITEM_MODEL=gpt-4o-mini         # 商品名识别专用模型

# Milvus 配置
MILVUS_URL=http://localhost:19530
CHUNKS_COLLECTION=chunks       # 文档切片集合
ITEM_NAME_COLLECTION=item_name # 商品名称集合

# Neo4J 配置
NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=password
NEO4J_DATABASE=neo4j

# MinIO 配置
MINIO_ENDPOINT=localhost:9000
MINIO_ACCESS_KEY=minioadmin
MINIO_SECRET_KEY=minioadmin
MINIO_BUCKET_NAME=shopkeeper

# 向量配置
EMBEDDING_DIM=1024

为什么不用 config.py 而用 .env？ 本项目有 6 个服务，每个服务的连接信息在不同环境（开发/生产）下不同。.env + dataclass 的分离模式比硬编码更灵活。

0.5 阶段结束时的演进架构图

本阶段已搭建的节点数：1/8（entry_node） 本阶段新增的文件：15 个（项目骨架 + 基础设施） 下一阶段将新增：3 个节点（PDF 解析 → 图片处理 → 文档切分）

📂 对应的原始代码快照：day01_项目介绍&环境部署/3_code/初始化项目/shopkeer_brain/ 注意：day01 的目录名是 shopkeer_brain（少了一个 p），不是最终版的 shopkeeper_brain——这是课程实录的原始状态

验证方法：

python

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"})
print(result)  # 应该看到 is_pdf_read_enabled=True, file_title="test"

阶段 0：部署与项目骨架（对应课程 day01） ​

0.1 阶段起始状态：空项目 ​

决策 1：为什么用 LangGraph 而非 LCEL ​

0.2 环境部署 👤 手动（约 30 min） ​

服务清单 ​

全量 docker-compose.yml 🤖 AI 生成 ​

0.3 项目骨架搭建（增量） ​

阶段结束时的目录树 🟢 新增高亮 ​

0.4 核心代码骨架 🤖 AI 辅助 ​

① BaseNode：所有节点的统一基类 ​

② ImportGraphState：8 个节点共享的全局状态 ​

③ EntryNode：导入流水线的入口 ​

④ main_graph：LangGraph 图定义 ​

⑤ .env 配置模板 🤖 AI 辅助填充 ​

0.5 阶段结束时的演进架构图 ​

阶段 0：部署与项目骨架（对应课程 day01）

0.1 阶段起始状态：空项目

决策 1：为什么用 LangGraph 而非 LCEL

0.2 环境部署 👤 手动（约 30 min）

服务清单

全量 docker-compose.yml 🤖 AI 生成

0.3 项目骨架搭建（增量）

阶段结束时的目录树 🟢 新增高亮

0.4 核心代码骨架 🤖 AI 辅助

① BaseNode：所有节点的统一基类

② ImportGraphState：8 个节点共享的全局状态

③ EntryNode：导入流水线的入口

④ main_graph：LangGraph 图定义

⑤ .env 配置模板 🤖 AI 辅助填充

0.5 阶段结束时的演进架构图