1
2
3
4
5
6
7
8
9

from langchain_community.document_loaders.csv_loader import CSVLoader

loader = CSVLoader(file_path="data.csv")
# 全量加载（适用于小数据）
docs = loader.load() 

# 惰性流式加载（适用于大数据，生成器模式）
for doc in loader.lazy_load():
    process_document(doc)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Author         : nanzet
# Description    : 演示自定义继承 BaseLoader 接口，实现带 Frontmatter 元数据提取和流式加载的 Markdown 文档解析器
# requirements   : pip install langchain-core pyyaml

import os
from pathlib import Path
from typing import Iterator

import yaml
from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document


class LocalMarkdownLoader(BaseLoader):
    """自定义的本地 Markdown 加载器，支持提取 Frontmatter 作为元数据。"""

    def __init__(self, file_path: str):
        self.file_path = file_path

    def lazy_load(self) -> Iterator[Document]:
        """
        核心逻辑：使用生成器 (yield) 实现 lazy_load。
        在处理成千上万个本地文件时，这能确保内存占用保持在极低水平。
        """
        path = Path(self.file_path)
        if not path.exists():
            raise FileNotFoundError(f"文件未找到: {self.file_path}")

        with open(path, "r", encoding="utf-8") as f:
            content = f.read()

        # 初始化元数据，至少保留数据溯源信息
        metadata = {"source": str(path)}
        page_content = content

        # 简单的 Frontmatter (--- yaml ---) 解析逻辑
        if content.startswith("---"):
            parts = content.split("---", 2)
            if len(parts) >= 3:
                try:
                    # 将解析出的标签、日期等属性合并到 metadata 中
                    frontmatter = yaml.safe_load(parts[1])
                    if isinstance(frontmatter, dict):
                        metadata.update(frontmatter)
                    page_content = parts[2].strip()
                except Exception as e:
                    print(f"[解析警告] Frontmatter 提取失败: {e}")

        # 产出标准化的 Document 对象给 LangChain 下游使用
        yield Document(page_content=page_content, metadata=metadata)


def create_mock_markdown(file_path: str):
    """辅助方法：生成一个带有 Frontmatter 的虚拟 Markdown 文件用于测试"""
    mock_content = """---
title: "M3 Mac 使用 UTM 安装 Kali Linux 实战"
date: "2026-05-29"
category: "运维与安全"
tags: ["Mac", "UTM", "Kali"]
---
# 弃坑 Parallels！丝滑安装全记录

本文将详细介绍如何在 M3 芯片的 Mac 上使用 UTM 虚拟机安装 Kali Linux，并提供完整的黑屏修复方案...
"""
    with open(file_path, "w", encoding="utf-8") as f:
        f.write(mock_content)


def main():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    test_file = os.path.join(current_dir, "test_markdown.md")  # 确保测试文件在当前目录下创建和访问

    try:
        # 1. 准备测试数据
        create_mock_markdown(test_file)

        # 2. 实例化自定义加载器
        loader = LocalMarkdownLoader(test_file)

        # 3. 生产级推荐做法：流式加载 (Lazy Load)
        print("--- 开始流式加载 (适合大批量文件导入向量库) ---")
        for doc in loader.lazy_load():
            print("抽取到的元数据 (Metadata):")
            for key, value in doc.metadata.items():
                print(f"  - {key}: {value}")
            print(f"\n清洗后的正文片段:\n{doc.page_content[:60]}...")

            # 在真实工程中，这里会接入下游逻辑：
            # chunked_docs = text_splitter.split_documents([doc])
            # vector_store.add_documents(chunked_docs)

    finally:
        # 清理测试现场
        if os.path.exists(test_file):
            os.remove(test_file)


if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

--- 开始流式加载 (适合大批量文件导入向量库) ---
抽取到的元数据 (Metadata):
  - source: /.../test_markdown.md
  - title: M3 Mac 使用 UTM 安装 Kali Linux 实战
  - date: 2026-05-29
  - category: 运维与安全
  - tags: ['Mac', 'UTM', 'Kali']

清洗后的正文片段:
# 弃坑 Parallels！丝滑安装全记录

本文将详细介绍如何在 M3 芯片的 Mac 上使用 UTM 虚拟机安装 ...