AgentDepthAgent 工程学习指南
输入概念或问题，检索正文和待编写目录。
第一次学习？从 L0 总览开始 →
知识总览 / L5 知识目录

采集、清洗与文档解析

更新 2026-10-05
我的学习 →

正在读取学习记录…

数据摄取是把原始资料转成可检索的文档记录。采集负责取得资料，解析负责识别正文、标题等结构，清洗负责去掉重复空白等噪声。入库后还要能回答“这段文字来自哪份资料、哪次更新”。

用退款资料观察入库

退款政策会重复采集，也会从“三至五个工作日”更新为“二至四个工作日”。我们保留来源地址形成稳定文档编号；内容摘要用来区分重复采集与实际变更。

固定输入包含四次采集。source是逻辑来源地址，raw是取得的原文字；清洗后的text才进入正文。保存记录时增加digest内容摘要和version版本号。store是一份保存在内存中的字典，主程序重新运行会从空库开始，便于重复观察同一流程。

保存完整程序并运行

先在 PowerShell 创建并进入学习目录：

Bash
mkdir "F:\学习\l5-ingestion" -Force
cd "F:\学习\l5-ingestion"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建学习目录，例如 F:\学习\l5-ingestion。将下面代码完整保存为 ingestion_demo.py；代码包含固定资料和运行入口，保存后就能执行。

先在 PowerShell 创建并进入学习目录：

Bash
mkdir "F:\学习\l5-ingestion" -Force
cd "F:\学习\l5-ingestion"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

Python
"""清洗退款资料，以稳定文档编号实现重复采集与更新。"""
import hashlib
import re

RAW = [
    ("policy://A/refund", "退款  通常三至五个工作日到账。"),
    ("policy://A/refund", "退款 通常三至五个工作日到账。"),
    ("policy://A/refund", "退款 通常二至四个工作日到账。"),
    ("policy://A/empty", "   "),
]

def clean(text):
    # 合并空白，保留业务期限、标点与正文顺序。
    return re.sub(r"\s+", " ", text).strip()

def ingest(store, source, raw):
    text = clean(raw)
    if not text:
        return "跳过空正文", None
    # 来源标识决定文档身份；内容变化时文档编号仍保持一致。
    doc_id = hashlib.sha256(source.encode("utf-8")).hexdigest()[:12]
    digest = hashlib.sha256(text.encode("utf-8")).hexdigest()
    previous = store.get(doc_id)
    if previous and previous["digest"] == digest:
        return "重复资料", doc_id
    version = previous["version"] + 1 if previous else 1
    store[doc_id] = {"source": source, "text": text,
                     "digest": digest, "version": version}
    return ("更新" if previous else "新增"), doc_id

def main():
    store = {}
    for source, raw in RAW:
        action, doc_id = ingest(store, source, raw)
        print(f"{action}：{doc_id or '无文档编号'}")
    for doc_id, doc in store.items():
        print(f"库内 {doc_id} v{doc['version']} [{doc['source']}]：{doc['text']}")

if __name__ == "__main__":
    main()

在保存文件的文件夹地址栏输入 powershell 并回车，再执行：

macOS：先在 Python 篇的学习文件夹中打开终端，执行 source .venv/bin/activate。再输入 cd ，把当前例子的文件夹拖入终端并回车，运行下面的命令。

Bash
cd "F:\学习\l5-ingestion"
ls
python -X utf8 ingestion_demo.py

使用其他目录时，把第一行换成自己的路径。dir 应能列出 ingestion_demo.py。终端输出如下：

TXT
新增：25fa200c0aa0
重复资料：25fa200c0aa0
更新：25fa200c0aa0
跳过空正文：无文档编号
库内 25fa200c0aa0 v2 [policy://A/refund]：退款 通常二至四个工作日到账。

第一次采集创建v1；第二次只有空白差异，清洗后的正文一致，版本保持不变；第三次到账期限改变，同一编号更新为v2。空白文档没有进入库。最后库里只有一份退款政策，正文是最新采集的二至四个工作日。

从来源身份到内容版本

clean先合并空白，再去首尾空白。ingest把来源转换为固定编号，把清洗后的正文转换为内容摘要。摘要是用哈希函数得到的内容指纹：相同文字得到相同值，用来比较内容是否变化。来源身份与内容指纹承担不同任务，因此改正文不会生成第二份逻辑文档。

演示使用12位编号便于阅读；大规模入库应保留足够长度或数据库唯一约束，处理哈希碰撞。两个地址可能指向同一资料，要先做来源地址规范化或维护业务文档ID；两份资料正文相同但权限不同，也不能仅按内容删除其中一份。

真实PDF解析还需要处理页码、页眉、表格与扫描件。扫描页需要OCR，即从图片识别文字；表格应把列标题和数据行保留在一起，避免退款天数失去适用渠道。清洗规则应谨慎删除页眉，不能误删“生鲜除外”这样的条件。

增量采集应保留更新时间、解析器版本、来源版本和失败原因。此程序的更新顺序由固定输入决定；并发采集需要版本比较与事务，防止迟到的旧版本覆盖新版本。删除事件还要传给索引层，下一篇的分块记录也需要继承这些来源信息。

policy://A/refund 固定身份；第一次：空白清洗 → 新增 v1；第二次：内容摘要相同 → 保持 v1；第三次：期限改变 → 原编号更新 v2；空正文 → 跳过入库
来源决定文档身份，正文摘要决定重复或更新。

改一处输入，核对机制变化

只把RAW第三项的“二至四”改为“三至五”，运行 python -X utf8 ingestion_demo.py。第三次采集变为重复资料，最后仍是v1，编号保持25fa200c0aa0。恢复二至四后重跑，第三次才更新为v2。

运行配套检查

把下面测试保存为 test_ingestion_demo.py，与主程序放在同一目录。5项测试检查上面的实际处理结果，也覆盖重复采集、内容更新和空正文。

查看测试代码：test_ingestion_demo.py
Python
"""核对采集、清洗与文档解析的正常结果与具体边界。"""
import unittest
import ingestion_demo as demo


class DemoTests(unittest.TestCase):
    def test_whitespace_cleaning(self):
        # 空白清洗后保留完整业务文字。
        self.assertEqual(demo.clean("  退款\n 到账  "), "退款 到账")

    def test_repeated_content_does_not_increment_version(self):
        # 同来源相同正文保持原版本。
        store = {}
        _, first = demo.ingest(store, "s", "退款  到账")
        action, second = demo.ingest(store, "s", "退款 到账")
        self.assertEqual((action, first), ("重复资料", second))
        self.assertEqual(store[first]["version"], 1)

    def test_changed_content_replaces_same_document(self):
        # 期限更新替换同一文档，版本递增。
        store = {}
        _, key = demo.ingest(store, "s", "三至五日")
        action, updated = demo.ingest(store, "s", "二至四日")
        self.assertEqual((action, updated), ("更新", key))
        self.assertEqual((len(store), store[key]["version"]), (1, 2))

    def test_empty_document_never_enters_store(self):
        # 空正文不创建任何库记录。
        store = {}
        self.assertEqual(demo.ingest(store, "s", "  ")[0], "跳过空正文")
        self.assertEqual(store, {})

    def test_distinct_sources_keep_separate_identity(self):
        # 不同来源的相同文字仍保留各自身份。
        store = {}
        demo.ingest(store, "a", "同一正文")
        demo.ingest(store, "b", "同一正文")
        self.assertEqual(len(store), 2)


if __name__ == "__main__":
    unittest.main()
Bash
python -X utf8 -m unittest -v test_ingestion_demo.py

测试名称后应显示 ok，末尾是 Ran 5 tests 和 OK。

运行时怎样定位问题

看到的情况检查与操作
找不到 ingestion_demo.py运行 dir，切换到保存主程序的目录
重复资料变成多个编号确认source是否多了追踪参数，先规范化来源地址
更新后仍检索到旧期限向索引同步新版本并删除旧块，核对更新任务
解析后表格条件丢失保留列标题与行关系，先核对解析正文再分块

资料记录稳定后，分块需要继续保留文档身份、版本与权限。

参考阅读：LlamaIndex 源码与检索流程。阅读时可以沿文档、节点、索引和检索入口对照这里的数据流。

课后练习展开习题收起习题

4 道练习 · 客观题规则评分 · 其余题目待批改

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“采集、清洗与文档解析”，哪种做法更稳妥？

对应知识点：L5-ingestion

选择一个答案
练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L5-ingestion

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L5-ingestion

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L5-ingestion

下一篇：分块、元数据与权限继承。

图片查看器
100%

文章配图 · 按 Esc 关闭

← 返回本层总览继续学习下一篇 →