大模型 RAG 文档切分进阶实战

import re

# text文本处理
def clean_text(text):
    # 去除多余空白
    text = re.sub(r'\s+', ' ', text)

    # 修复断开的单词
    text = re.sub(r'(\w+)-\s+(\w+)', r'\1\2', text)

    # 标准化引号
    text = text.replace('"', '"').replace('"', '"')
    return text.strip()

# html文本处理
def clean_html(text):
    return re.sub(r'<.*?>', '', text)  # 移除HTML标签

# markdown文本处理
def clean_markdown(text):
    text = re.sub(r'\*\*(.*?)\*\*', r'\1', text)  # 移除粗体标记但保留文字
    text = re.sub(r'\[(.*?)\]\(.*?\)', r'\1', text)  # 移除链接标记但保留文字
    return text

text = "Hello World! This is a **bold** text with a [link](http://example.com)."
print(clean_markdown(text))

# 安装pandas
# pip install pandas

import pandas as pd

def clean_table(df):
    if type(df) != pd.DataFrame:
        df = pd.DataFrame(df)

    # 去除完全空白的行和列
    df = df.dropna(how='all').dropna(axis=1, how='all')

    # 填充NaN值
    df = df.fillna('')

    # 删除完全重复的行 (不要使用 inplace=True，它会返回 None)
    df = df.drop_duplicates()

    # 标准化列名
    df.columns = [str(col).strip() for col in df.columns]

    return df

# 创建一个包含常见数据问题的示例数据集
data = {
    '姓名': ['小明', '小红', '小刚', '小丽', '小强', '小红'],
    '年龄': [18, 19, 17, 20, 250, 19],  # 250是一个异常值
    '语文成绩': [85, 92, 78, None, 88, 92],  # 包含一个缺失值
    '数学成绩': [90, None, 85, 92, 87, None],  # 包含缺失值
    '是否及格': ['是', '是', '是', '是', '是', '是']
}

df = pd.DataFrame(data)
print("原始数据：")
print(df)
print("=" * 60)

df_deal = clean_table(df)
print("处理后数据：")
print(df_deal)

#安装依赖包
# pip install cv2 matplotlib
import cv2
from matplotlib import pyplot as plt

# 显示图像的函数
def show_image(img, title="Image", cmap=None):
    plt.figure(figsize=(4, 2))
    if cmap:
        plt.imshow(img, cmap=cmap)
    else:
        plt.imshow(img)
    plt.title(title)
    plt.axis('off')
    plt.show()

def clean_image(img_path):
    """
    图像降噪 - 去除扫描文档中的噪声点
    """
    # 读取图像
    img = cv2.imread(img_path)
    img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)

    # 转换为灰度图
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

    print("原始图像:")
    show_image(img_rgb, "Original Image")

    # 方法1: 高斯模糊去噪
    gaussian_denoised = cv2.GaussianBlur(gray, (5, 5), 0)

    # 方法2: 中值滤波去噪 (对椒盐噪声效果好)
    median_denoised = cv2.medianBlur(gray, 3)

    # 方法3: 双边滤波 (保持边缘)
    bilateral_denoised = cv2.bilateralFilter(gray, 9, 75, 75)

    print("高斯模糊去噪:")
    show_image(gaussian_denoised, "Gaussian Denoised", cmap='gray')

    print("中值滤波去噪:")
    show_image(median_denoised, "Median Denoised", cmap='gray')

    print("双边滤波去噪:")
    show_image(bilateral_denoised, "Bilateral Denoised", cmap='gray')

    return gray, gaussian_denoised, median_denoised, bilateral_denoised

# 使用示例
gray, gaussian, median, bilateral = clean_image("手写公式.png")

import re

def clean_code(code_text, remove_comments=False):
    """
    增强版代码清洗
    """
    if not code_text:
        return ""

    cleaned_lines = []
    in_multiline_comment = False

    for line in code_text.split('\n'):
        # 移除行尾空白
        clean_line = line.rstrip()

        # 可选：移除单行注释
        if remove_comments:
            if not in_multiline_comment:
                # 检查是否进入多行注释
                if '"""' in clean_line or "'''" in clean_line:
                    in_multiline_comment = not in_multiline_comment
                    # 简单处理：直接跳过含有多行注释符号的行
                    continue
                # 移除单行注释（# 后面的内容）
                clean_line = re.sub(r'#.*$', '', clean_line)
            else:
                # 在多行注释中，跳过这行
                if '"""' in clean_line or "'''" in clean_line:
                    in_multiline_comment = False
                continue

        # 如果行不为空，或者我们保留空行（这里保留一个空行）
        if clean_line or (cleaned_lines and not cleaned_lines[-1]):
            cleaned_lines.append(clean_line)

    # 重新组合并确保首尾没有空行
    result = '\n'.join(cleaned_lines).strip()

    # 确保以换行符结束（可选）
    if result and not result.endswith('\n'):
        result += '\n'

    return result

# 测试示例
dirty_code = """
def example_function():
    # 这是一个示例函数
    data = [1, 2, 3]

    for item in data:
        print(item)



    return True

"""
print(clean_code(dirty_code))

def clean_mixed_content(content_type, content):
        """
        统一清洗入口
        content_type: 'text', 'table', 'image', 'code'
        """
        try:
            if content_type == 'text':
                text = clean_text(content)
                text = clean_html(text)
                text = clean_markdown(text)
                return text

            elif content_type == 'table':
                return clean_table(content)

            elif content_type == 'image':
                return clean_image(content)

            elif content_type == 'code':
                return clean_code(content)

            else:
                print(f"未知内容类型: {content_type}")
                return content

        except Exception as e:
            print(f"清洗 {content_type} 时出错: {e}")
            return content

def batch_clean(documents):
    """
    批量清洗文档
    documents: 列表，每个元素是 (content_type, content) 元组
    """
    results = []
    for i, (doc_type, content) in enumerate(documents):
        print(f"清洗文档 {i+1}: {doc_type}")
        cleaned = clean_mixed_content(doc_type, content)
        results.append((doc_type, cleaned))

    return results

# 准备测试数据
test_documents = [
    # 文本测试数据
    ('text', '  这是一段  有很多多余  空格的文本！！！  \n\n还有换行符...  '),

    # 表格测试数据
    ('table', [
        ['  Name  ', ' Age ', ' City   '],
        [' Alice ', None, 'New York'],
        ['  Bob  ', ' 25 ', '  ']
    ]),

    # 图片测试数据（文件路径）
    #('image', '手写公式.png'),

    # 代码测试数据
    ('code', 'def hello():\n    print("world")  \n\n\n    return True  ')
]

# 执行批量清洗
print("开始清洗混合文档...")
cleaned_docs = batch_clean(test_documents)

# 显示清洗结果
print("\n=== 清洗结果 ===")
for doc_type, content in cleaned_docs:
    print(f"\n类型: {doc_type}")
    print(f"内容: ")
    print(content)
    print("-" * 50)

from llama_index.core import Document
# 导入SentenceSplitter句子分割器
from llama_index.core.node_parser import SentenceSplitter

# 创建Document并设置元数据
doc = Document(
    text="这是一份关于RAG技术的文档...",
    metadata={
        "file_name": "rag_guide.pdf",
        "category": "技术文档",
        "author": "AI研究团队",
        "created_date": "2023-11-15"
    }
)

# 从Document创建Node时，元数据会自动传播
splitter = SentenceSplitter()
nodes = splitter.get_nodes_from_documents([doc])

# 每个node都会继承doc的metadata
nodes[0].metadata

from llama_index.core.schema import TextNode

# Node的基本结构
node = TextNode(
    text="这是切分后的文本内容",           # 文本内容
    metadata={                            # 元数据
        "file_name": "document.pdf",
        "page_number": 1,
        "chunk_id": 0
    },
    id_="node_id_123",                   # 唯一标识符
    embeddings=[]                        # 文本嵌入向量(可选)
)

from llama_index.core.schema import NodeRelationship
import json

# Node的基本结构
node0 = TextNode(
    text="这是切分后的文本内容0",           # 文本内容
    metadata={                            # 元数据
        "file_name": "document.pdf",
        "page_number": 1,
        "chunk_id": 0
    },
    id_="node_id_0",                   # 唯一标识符
)

node1 = TextNode(
    text="这是切分后的文本内容1",           # 文本内容
    metadata={                            # 元数据
        "file_name": "document.pdf",
        "page_number": 1,
        "chunk_id": 1
    },
    id_="node_id_1",                   # 唯一标识符
)

node2 = TextNode(
    text="这是切分后的文本内容2",           # 文本内容
    metadata={                            # 元数据
        "file_name": "document.pdf",
        "page_number": 1,
        "chunk_id": 2
    },
    id_="node_id_2",                   # 唯一标识符
)

# 顺序关系
node1.relationships[NodeRelationship.SOURCE] = node0.id_
node1.relationships[NodeRelationship.NEXT] = node2.id_
node2.relationships[NodeRelationship.PREVIOUS] = node1.id_

# 层次化关系
node2.relationships[NodeRelationship.SOURCE] = node0.id_
node1.relationships[NodeRelationship.CHILD] = node2.id_
node2.relationships[NodeRelationship.PARENT] = node1.id_

# 打印节点关系
print(node1.relationships)
print(node2.relationships)

from llama_index.core.node_parser import TokenTextSplitter

# 初始化 TokenTextSplitter
token_splitter = TokenTextSplitter(
    chunk_size=512,        # 每 chunk 目标 token 数（可调）
    chunk_overlap=64,      # 重叠 token 数（可调）
    separator=" "          # 分隔符（一般用空格）
)

# 将 documents 转成 nodes（LlamaIndex 内部 node/fragment）
#nodes_from_tokens = token_splitter.get_nodes_from_documents(documents)
#print(nodes_from_tokens[1].get_content())

# 测试文本
test_text = """在检索增强生成（RAG）系统中，文档切分与 Node 转换作为连接原始数据与语言模型的关键预处理环节，直接决定了系统的检索精度、生成质量及整体性能。行业实践数据表明，90% 的 RAG 效果问题源于元数据与分块策略不当，而通过优化分块策略可使检索准确率提升 30 - 50%，语义分块较固定分块的准确率优势可达 27%。这一技术环节的重要性体现在：分块过大易引入冗余噪音，增加语言模型理解负担；分块过小或切分不当则可能破坏语义连贯性，导致完整知识点被拆分；未能适配文档结构的机械分块方式还会忽视标题、列表等结构化信息，影响信息提取完整性。
LlamaIndex 作为连接自定义数据与大语言模型（LLMs）的核心框架，通过将文档（如 PDF、文本文件）分解为包含文本内容、向量嵌入和元数据的 Node 组件，构建了结构化文档管理的技术范式。其核心抽象在于将原始文档转换为语义连贯的 Node 集合，向量存储仅保留 Node 内容的嵌入向量与文本信息，这一机制简化了索引构建流程并提升了检索相关性。文档切分与 Node 转换的质量不仅影响向量检索的效率，更决定了上下文增强（Context Augmentation）这一 RAG 核心能力的实现效果。
本文聚焦文档切分与 Node 转换的技术实践，结合 LlamaIndex 框架的实现机制，系统调研分块策略设计、元数据管理及 Node 组件化等关键技术点。通过分析行业最佳实践与典型案例，旨在为 RAG 系统开发者提供可落地的优化方案，解决分块噪音、语义断裂、结构信息丢失等核心痛点，为构建高性能检索增强生成应用奠定技术基础。"""

# 按照文本来进行切分
nodes_from_tokens = token_splitter.split_text(test_text)
# 打印切分后的文本数量
print( "切分后的文本数量: " + str(len(nodes_from_tokens)))
print("===============================================")
# 打印第一个文本
print("第一个文本:")
print(nodes_from_tokens[0])
print("===============================================")
# 打印第二个文本
print("第二个文本:")
print(nodes_from_tokens[1])
print("===============================================")

from llama_index.core.node_parser import SentenceSplitter
# 初始化 SentenceSplitter
sentence_splitter = SentenceSplitter(
    chunk_size=512,      # 这里 chunk_size 表示 token 近似或字符近似，视版本调整
    chunk_overlap=64
)
# 将 documents 转成 nodes
# nodes_from_sentences = sentence_splitter.get_nodes_from_documents(documents)
# print(nodes_from_sentences[0].get_content())

# 测试文本
test_text = """在检索增强生成（RAG）系统中，文档切分与 Node 转换作为连接原始数据与语言模型的关键预处理环节，直接决定了系统的检索精度、生成质量及整体性能。行业实践数据表明，90% 的 RAG 效果问题源于元数据与分块策略不当，而通过优化分块策略可使检索准确率提升 30 - 50%，语义分块较固定分块的准确率优势可达 27%。这一技术环节的重要性体现在：分块过大易引入冗余噪音，增加语言模型理解负担；分块过小或切分不当则可能破坏语义连贯性，导致完整知识点被拆分；未能适配文档结构的机械分块方式还会忽视标题、列表等结构化信息，影响信息提取完整性。
LlamaIndex 作为连接自定义数据与大语言模型（LLMs）的核心框架，通过将文档（如 PDF、文本文件）分解为包含文本内容、向量嵌入和元数据的 Node 组件，构建了结构化文档管理的技术范式。其核心抽象在于将原始文档转换为语义连贯的 Node 集合，向量存储仅保留 Node 内容的嵌入向量与文本信息，这一机制简化了索引构建流程并提升了检索相关性。文档切分与 Node 转换的质量不仅影响向量检索的效率，更决定了上下文增强（Context Augmentation）这一 RAG 核心能力的实现效果。
本文聚焦文档切分与 Node 转换的技术实践，结合 LlamaIndex 框架的实现机制，系统调研分块策略设计、元数据管理及 Node 组件化等关键技术点。通过分析行业最佳实践与典型案例，旨在为 RAG 系统开发者提供可落地的优化方案，解决分块噪音、语义断裂、结构信息丢失等核心痛点，为构建高性能检索增强生成应用奠定技术基础。"""


# 按照文本来进行切分
nodes_from_sentences = sentence_splitter.split_text(test_text)
print( "切分后的文本数量: " + str(len(nodes_from_sentences)))
print("===============================================")
print("第一个文本:")
print(nodes_from_sentences[0])
print("===============================================")
print("第二个文本:")
print(nodes_from_sentences[1])
print("===============================================")

from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.schema import Document

# SentenceSplitter 测试递归切分
# 示例文本：一个长句子和一个短句子
text = "这是一个非常长的句子，它包含了多个逗号分隔的部分，和一些短的文本，因此整个句子的长度会很容易超过我们设定的块大小限制。短期。"
document = Document(text=text)

# 初始化SentenceSplitter，设定块大小非常小以触发递归分割
splitter = SentenceSplitter(chunk_size=10, chunk_overlap=0,separator="，。！？!?.\n¡¿")
nodes = splitter.get_nodes_from_documents([document])

# 打印分割结果
print("=" * 60)
for i, node in enumerate(nodes):
    print(f"块 {i+1}: {node.text}")

print("=" * 60)
print("打印Node信息：")
print(nodes[0].__dict__)

# TextNode节点信息
{
	'id_': '5c0c3c5f-614e-41fb-b26b-eec080d8096e',
	'embedding': None,
	'metadata': {},
	'excluded_embed_metadata_keys': [],
	'excluded_llm_metadata_keys': [],
	'relationships': { < NodeRelationship.SOURCE: '1' > :RelatedNodeInfo(node_id = 'c21e5d07-d7d6-43ed-a54c-523ec7046359', node_type =  , metadata = {},
		hash = '5941844fa9c40295c3fe1ee08abb5952ab4590d1951fb2150f644e3f19a18159'),
		 :RelatedNodeInfo(node_id = '7d757fbc-4f2a-48aa-852a-8be7692b0572', node_type =  , metadata = {},
		hash = 'e611f789a713d9540163feb55ff3c55fb0f80f43142631e1ce9c642ec49944ff')
	},
	'metadata_template': '{key}: {value}',
	'metadata_separator': '\n',
	'text': '这是一个非常长的句',
	'mimetype': 'text/plain',
	'start_char_idx': 0,
	'end_char_idx': 9,
	'metadata_seperator': '\n',
	'text_template': '{metadata_str}\n\n{content}'
}

#安装依赖
#!pip install tree_sitter tree_sitter_language_pack

from llama_index.core.node_parser import CodeSplitter
from llama_index.core.schema import Document

# 示例Python代码：一个简单的函数和类
sample_code = '''
def calculate_fibonacci(n):
    """计算斐波那契数列的第n项"""
    if n <= 1:
        return n
    else:
        return calculate_fibonacci(n-1) + calculate_fibonacci(n-2)

class MathOperations:
    """一个简单的数学操作类"""

    def __init__(self):
        self.version = "1.0"

    def factorial(self, n):
        """计算阶乘"""
        if n == 0:
            return 1
        result = 1
        for i in range(1, n+1):
            result *= i
        return result

    def is_prime(self, n):
        """判断是否为质数"""
        if n < 2:
            return False
        for i in range(2, int(n**0.5)+1):
            if n % i == 0:
                return False
        return True
'''

# 创建文档对象
# document = Document(text=sample_code)
#
# 初始化CodeSplitter，指定Python语言
code_splitter = CodeSplitter(
    language="python",    # 指定编程语言
    chunk_lines=10,       # 每块大约行数
    chunk_lines_overlap=2, # 块之间重叠行数
    max_chars=600        # 每块最大字符数
)

# 执行切分
# nodes = code_splitter.get_nodes_from_documents([document])
nodes = code_splitter.split_text(sample_code)

print(f"原始代码字符数: {len(sample_code)}")
print(f"切分后的节点数量: {len(nodes)}")
print("\n" + "="*50 + "\n")

# 显示切分结果
for i, node in enumerate(nodes):
    print(f"节点 {i+1} (字符数: {len(node)}):")
    print("-" * 30)
    print(node)
    print("\n" + "="*50 + "\n")

# 文本类型是markdown的
from llama_index.core.node_parser import MarkdownNodeParser
from llama_index.core.readers import SimpleDirectoryReader

markdown_docs = SimpleDirectoryReader(input_files=["扩展调参.md"]).load_data()
# markdown_docs = [Document(text="# 主标题\n\n这是第一段。\n\n## 子标题\n\n这是第二段。")]

# 创建Markdown解析器
parser = MarkdownNodeParser()

# 从Markdown文件创建节点
nodes = parser.get_nodes_from_documents(markdown_docs)

# 显示切分结果
for i, node in enumerate(nodes):
    print(f"节点 {i+1} (字符数: {len(node.text)}):")
    print("-" * 30)
    print(node.text)
    print("\n" + "="*50 + "\n")

from llama_index.core.node_parser import JSONNodeParser

json = """
{
    "id_": "0a1eee9a-635a-4391-8b74-75bf3c648f0e",
    "embedding": null,
    "metadata": {
        "document_id": "FULadzkWmovlfkxSgLPcE4oWnPf"
    },
    "excluded_embed_metadata_keys": [],
    "excluded_llm_metadata_keys": [],
    "mimetype": "text/plain"
}
"""

# 创建JSON解析器
parser = JSONNodeParser()

json_docs = [Document(text=json)]

# 从JSON文件创建节点
nodes = parser.get_nodes_from_documents(json_docs)

# 显示切分结果
for i, node in enumerate(nodes):
    print(f"节点 {i+1} (字符数: {len(node.text)}):")
    print("-" * 30)
    print(node.text)
    print("\n" + "="*50 + "\n")

# requirements (示例)
# pip install llama-index

from llama_index.core import Document
from llama_index.embeddings.openai import OpenAIEmbedding

import os
from llama_index.core.settings import Settings
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()

# 设置为全局默认Embedding模型
Settings.embed_model = OpenAIEmbedding(
    model="text-embedding-3-small",
    api_key=os.getenv("OPENAI_API_KEY"),
    api_base=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)

from llama_index.core.node_parser import SemanticSplitterNodeParser
from typing import List

# 1) 中文拆句器（更鲁棒）
def split_chinese_sentences(text: str) -> List[str]:
    """
    将中文文本按常见句末标点拆分，尽量保留标点并去除多余空白。
    处理省略号、连续标点和英文句点等。
    """
    if not text:
        return []
    # 将长省略号统一处理
    text = text.replace("……", "…")
    # 在常见句末标点后断句（保留标点）
    pieces = re.split(r'(?<=[。！？…\?\!\.])\s*', text)
    # 清理空白与空串
    sentences = [p.strip() for p in pieces if p and p.strip()]
    return sentences

long_chinese_text = (
    "本季度公司财务表现良好，营收增长15%，净利润同比提升10%。"
    "在产品方面，我们完成了新一代搜索引擎的内测，搜索精度和召回率都有明显提升。"
    "同时，基础设施团队迁移到新的集群架构，缩短了部署时间并降低了成本。"
    "关于市场推广，最近在北京与上海分别举办了两场线下用户交流会，"
    "收集到了大量用户反馈，尤其是对移动端体验的改进建议。"
    "另一方面，我们正在探索与第三方数据提供商的合作，"
    "以期在广告定向和推荐系统上获得更准确的信号。"
    "此外，法律合规团队提醒需关注新的隐私合规要求，"
    "包括数据最小化和用户可解释性方面的合规文档准备。"
    "最后，团队在招聘方面也有所动作，已开放多个后端与算法岗位。"
)

# 创建Document
doc = Document(text=long_chinese_text, metadata={"doc_id": "示例文档1"})

# SemanticChunker 会使用嵌入相似度来决定是否将句子合并到同一 chunk 中。
splitter = SemanticSplitterNodeParser(
    buffer_size=2,                        # buffer_size 用于把多少句子为一组做相似度计算。
    breakpoint_percentile_threshold=80,   # 相似度阈值，低于该阈值的相似度会被视为断点。
    embed_model=Settings.embed_model,     # embed_model 表示用于计算嵌入的模型。
    sentence_splitter=split_chinese_sentences, # 自定义中文拆句器
    include_metadata=True,                # 是否包含 metadata
    include_prev_next_rel=True,           # 是否包含上一句与下一句的关系
)

# also baseline splitter
# nodes_from_semantic = splitter.get_nodes_from_documents(documents)
nodes_from_semantic = splitter.get_nodes_from_documents([doc])

print("=== 切割后（chunks） ===")
for idx, node in enumerate(nodes_from_semantic):
    print(f"--- chunk {idx} ---")
    # node.text: chunk 的主文本（通常是若干句子合并）
    print("chunk.text:", node.text)
    # 如果 include_metadata=True，会有一些 metadata（例如原始句子索引、chunk_type等）
    print("metadata keys:", list(node.metadata.keys()))
    # 如果有 prev/next 关系或 chunk 索引，可一并查看
    # 例如：node.metadata.get('chunk_index'), node.metadata.get('chunk_type')
    print("metadata (sample):", {k: node.metadata.get(k) for k in ['chunk_index','chunk_type'] if k in node.metadata})
    print()

from llama_index.core.node_parser import SentenceWindowNodeParser

# 使用一个简短的文本进行测试
text = "I love programming. Python is my most favorite language. I love LLMs. I love LlamaIndex."

node_parser = SentenceWindowNodeParser.from_defaults(
    # how many sentences on either side to capture
    window_size=1,
    # the metadata key that holds the window of surrounding sentences
    window_metadata_key="window",
    # the metadata key that holds the original sentence
    original_text_metadata_key="original_sentence",
    include_metadata=True,
    #sentence_splitter=custom_splitter  # 传入自定义的分割器
)

nodes = node_parser.get_nodes_from_documents([Document(text=text)])

# 打印所有节点的文本
print("所有节点文本:")
for i, node in enumerate(nodes):
    print(f"Node {i}: {node.text}")

# 查看第二个节点周围的窗口文本
print("\n第二个节点周围的窗口文本:")
print(nodes[1].metadata["window"])
print("=" * 80)
print(nodes[1].relationships)

# SentenceWindowNodeParser切分后的relationShip结构
{
< NodeRelationship.SOURCE: '1' > :
        RelatedNodeInfo(
            node_id = 'f445ded2-368b-4e2a-9769-70049028bdda',
            node_type =  ,
            metadata = {},
	        hash = '23f0fdc1f3f7d315c4602ce57a4fd967dde771b7105a3a62a1d602e8a8fabba8'),
 :
        RelatedNodeInfo(
            node_id = '4049b298-bd47-48f9-814e-7239aab8f02d',
            node_type =  ,
            metadata = {
		        'window': 'I love programming.  Python is my most favorite language. ',
		        'original_sentence': 'I love programming. '
	    },
	        hash = '2177457ce21b218a610394927ed308ab4ef27723b2d7d3cd2d1e3d1365cacdb7'),
 :
        RelatedNodeInfo(
            node_id = '6dab22c2-3e5b-4391-b511-e0425cadbb1e',
            node_type =  ,
            metadata = {
                'window': 'Python is my most favorite language.  I love LLMs.  I love LlamaIndex.',
                'original_sentence': 'I love LLMs. '
	    },
	    hash = '750a235a276c28bca95e4305d511e7e76d8b3e52b67eafc688a8c1957fe5d02e')
}

import re
from typing import List
from llama_index.core import Document  # or from llama_index.schema import Document depending on version
from llama_index.core.node_parser import SentenceWindowNodeParser
from llama_index.core.postprocessor import MetadataReplacementPostProcessor

# 1) 中文拆句器（示例）
def split_chinese_sentences(text: str) -> List[str]:
    """
    将中文文本按常见中文句末标点拆为句子（保留标点）。
    返回去除空串和首尾空白的句子列表。
    """
    # 以中文/英文句末标点为分界（正则使用前后肯定断言保留标点）
    pieces = re.split(r'(?<=[。！？\?!.])', text)
    # 清理空白与空串
    sentences = [p.strip() for p in pieces if p and p.strip()]
    return sentences

# 2) 创建 SentenceWindowNodeParser
node_parser = SentenceWindowNodeParser(
    sentence_splitter=split_chinese_sentences,  # 自定义中文拆句器
    window_size=1,                              # 左右各1句作为window
    window_metadata_key="window",               # metadata 中存储窗口的 key
    original_text_metadata_key="original_text", # 可选：保存原始文本
    include_metadata=True,                      # 是否包含 metadata
    include_prev_next_rel=True,                 # 是否包含上一句与下一句的关系
)

# 3) 示例中文文本与节点生成（调试）
# text = (
#     "这是第一句。这里是第二句，它比较长，会测试拆分。"
#     "第三句来了！第四句？第五句，继续测试。"
# )
text = "本报告中的信息均来源于我们认为可靠的已公开资料，本公司对这些信息的真实性、准确性及完整性不作任何保证。 本报告中的信息、意见等均仅供客户参考，该等信息、意见并未考虑到获取本报告人员的具体投资目的、财务状况以 及特定需求，在任何时候均不构成对任何人的个人推荐。客户应当对本报告中的信息和意见进行独立评估，并应同时 思量各自的投资目的、财务状况以及特定需求，必要时就法律、商业、财务、税收等方面咨询专家的意见。客户应自 主作出投资决策并自行承担投资风险。本公司特别提示，本公司不会与任何客户以任何形式分享证券投资收益或分担 证券投资损失，任何形式的分享证券投资收益或者分担证券投资损失的书面或口头承诺均为无效。市场有风险，投资 须谨慎。对依据或者使用本报告所造成的一切后果，本公司和关联人员均不承担任何法律责任。"
doc = Document(text=text, metadata={"doc_id": "示例文档1"})
nodes = node_parser.get_nodes_from_documents([doc])

# 打印每个节点及其 metadata（调试用）
for i, node in enumerate(nodes[:2]):
    print("---- NODE", i, "----")
    print("node.text:", repr(node.text))  # node.text 应为单句
    print("=" * 80)
    #print("relationships:", node.relationships)
    # 存在的 metadata key
    #print("metadata keys:", node.metadata.keys())
    # window 存放左右句（字符串或列表，取决于实现；通常是字符串）
    print("window metadata:", node.metadata.get("window"))
    # original_text 为可选，存放原始文本
    print("original_text:", node.metadata.get("original_text"))
    print()

# 4) 在检索/查询时把 window 替换回去
# 用于 QueryEngine 的 node_postprocessor 示例：把 node.text 替换为 metadata['window']（如果存在）
postproc = MetadataReplacementPostProcessor(target_metadata_key="window")

# 当你用 index.as_query_engine(..., node_postprocessors=[postproc]) 时，
# 被检索到的 node.text 会被 postproc 替换为 metadata['window']（即包含左右句的文本）
# 之后传给 LLM 的就是带上下文的窗口文本，而不是单个句子

# 使用SentenceWindowNodeParser切分后的relationShip结构
{
< NodeRelationship.SOURCE: '1' > :
    RelatedNodeInfo(
        node_id = 'e7fa8003-7f50-46da-ae8f-890f7a95094b',
        node_type =  ,
        metadata = {
		    'doc_id': '示例文档1'
	    },
	    hash = '86a8e13bcb62a07f7b89cb0ffb677eeb071f755a78d34a5bcf76bc1e7c01d369'),
 :
    RelatedNodeInfo(
        node_id = '3ab2f3a4-57c3-4ba3-a885-91c670b46af9',
        node_type =  ,
        metadata = {
            'doc_id': '示例文档1',
            'window': '本报告中的信息均来源于我们认为可靠的已公开资料，本公司对这些信息的真实性、准确性及完整性不作任何保证。 本报告中的信息、意见等均仅供客户参考，该等信息、意见并未考虑到获取本报告人员的具体投资目的、财务状况以 及特定需求，在任何时候均不构成对任何人的个人推荐。',
            'original_text': '本报告中的信息均来源于我们认为可靠的已公开资料，本公司对这些信息的真实性、准确性及完整性不作任何保证。'
        },
	    hash = '711fb2cc917c527b4c754584b97d3421b3579ccdc2f325b7515f6c338ea119d3'),
 :
    RelatedNodeInfo(
        node_id = '613e77ed-d739-4085-83f7-b0d2641d3fc6',
        node_type =  ,
        metadata = {
            'window': '本报告中的信息、意见等均仅供客户参考，该等信息、意见并未考虑到获取本报告人员的具体投资目的、财务状况以 及特定需求，在任何时候均不构成对任何人的个人推荐。 客户应当对本报告中的信息和意见进行独立评估，并应同时 思量各自的投资目的、财务状况以及特定需求，必要时就法律、商业、财务、税收等方面咨询专家的意见。 客户应自 主作出投资决策并自行承担投资风险。',
            'original_text': '客户应当对本报告中的信息和意见进行独立评估，并应同时 思量各自的投资目的、财务状况以及特定需求，必要时就法律、商业、财务、税收等方面咨询专家的意见。'
        },
	    hash = '05631a6d45a08706f49acb3776579a3697803a57abcc304f87b71e19ce7b100d')
}

# -*- coding: utf-8 -*-
import re
import json
from typing import List
from llama_index.core import Document
from llama_index.core.node_parser.relational.hierarchical import HierarchicalNodeParser

# 构造示例中文较长文档
text = (
    "第一章：公司发展背景。公司成立于2005年，最初是一家小型软件外包公司。"
    "随着云计算与大数据的兴起，公司在2010年转型为云服务提供商，"
    "并在2015年完成了A轮融资，融资金额达数千万美元。"
    "第二章：产品与服务。公司主要提供数据分析平台、实时流处理系统和人工智能模型服务。"
    "其中，数据分析平台支持海量日志处理；流处理系统可实现毫秒级别延迟。"
    "第三章：市场与竞争。国内外竞争者众多，我们面临来自大型互联网公司的压力，"
    "但我们的优势在于垂直行业深耕与定制化服务。"
    "第四章：未来展望。我们计划在2026年前进入国际市场，并开展亚太地区的业务。"
)

print("=== 切分前（原文） ===")
print(text)
print("\n")

# 创建 HierarchicalNodeParser
node_parser = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[300, 120],   # 例：300字符/120字符级别
    chunk_overlap=30,         # 重叠区域大小
    include_metadata=True,      # 是否包含metadata
    include_prev_next_rel=True  # 是否包含前后关系
)

# 注意：默认内部会为每个级别创建 SentenceSplitter 用于拆分
nodes = node_parser.get_nodes_from_documents([Document(text=text, metadata={"doc_id":"示例文档"})])

print("=== 切分后（nodes） ===")
for idx, node in enumerate(nodes):
    print(f"--- node {idx} ---")
    print("relationships:", node.relationships)
    print("=" * 60)
    print("text:", repr(node.text))
    print("metadata keys:", list(node.metadata.keys()))
    # 可显示 chunk size 或 parent id
    print("metadata sample:", {k: node.metadata.get(k) for k in ["chunk_size","chunk_level","doc_id"] if k in node.metadata})
    print()

# 使用HierarchicalNodeParser切分后的relationShip结构
{
< NodeRelationship.SOURCE: '1' > :
    RelatedNodeInfo(
        node_id = 'c63cd21b-eeef-4019-8763-f25d29fe95c5',
        node_type =  ,
        metadata = {
		'doc_id': '示例文档'
	},
	hash = 'ad3c5540f4eb39f00473b948c146058f99683b7254f39c05a1a3ecb596e41f19'),
 :
    [RelatedNodeInfo(
        node_id = '0997e9aa-cc6a-4d7c-bf22-839c8a235ced',
        node_type =  ,
        metadata = {
		'doc_id': '示例文档'
	},
	hash = 'fc71df1d905d3f30b1a57044e9ae52e45476d0297ab2718145ad17ef7099d964'),
    RelatedNodeInfo(
        node_id = '599a0923-2651-49bf-981f-ec4304ed24ef',
        node_type =  ,
        metadata = {
		'doc_id': '示例文档'
	},
	hash = '558df4aaa18d1d121269ead16ca2671c2db4174d41b3bc9bbc3705c0b5088938'),
    RelatedNodeInfo(
        node_id = '6425b6e7-0edc-4169-9a24-07d5cc6b998a',
        node_type =  ,
        metadata = {
		'doc_id': '示例文档'
	},
	hash = '78b577f816806b9a9a932de2333ee5f672318b3ca4396e63692649e24c23f9e8')]
}

from unstructured.partition.pdf import partition_pdf

# 使用 partition_pdf 函数解析 PDF 文档 为elements类型
elements = partition_pdf(
    filename="甬兴证券-AI行业点评报告：海外科技巨头持续发力AI，龙头公司中报业绩亮眼.pdf",
    strategy="hi_res",  # 使用高精度模式
    extract_images_in_pdf=False,
)

# 导入unstructured的chunk_by_title切分方法
from unstructured.chunking.title import chunk_by_title

# 文档切分
chunked_elements = chunk_by_title(
    elements,                       # 读取的元素列表
    max_characters=800,             # 每个chunk的最大字符数
    combine_text_under_n_chars=150, # 小于该字符数的文本块会合并
)

print(f"✓ 解析出 {len(elements)} 个元素")
print(f"✓ 切分成 {len(chunked_elements)} 个chunks")

import json
from llama_index.core.schema import Document

llamaindex_documents = []

for chunk in chunked_elements:
    # 提取metadata并清理不需要的字段
    metadata = chunk.metadata.to_dict()

    # 增强metadata: 添加元素类型信息
    metadata['element_type'] = type(chunk).__name__

    # 如果有原始元素,可以提取更多信息
    if hasattr(chunk.metadata, 'orig_elements') and chunk.metadata.orig_elements:
        # 提取所有原始元素的类型
        metadata['orig_element_types'] = [type(e).__name__ for e in chunk.metadata.orig_elements]
        # 标记是否包含Title
        metadata['contains_title'] = any(
            type(e).__name__ == 'Title' for e in chunk.metadata.orig_elements
        )

    # 移除一些序列化时可能有问题的字段
    metadata.pop('languages', None)
    metadata.pop('orig_elements', None)  # 太大,不适合存储在向量数据库

    # 创建Document对象
    doc = Document(
        text=chunk.text,
        metadata=metadata,
    )
    llamaindex_documents.append(doc)

print("\n" + "=" * 80)
print("转换为LlamaIndex Documents")
print("=" * 80)
for i, doc in enumerate(llamaindex_documents[:3]):
    print(f"\n--- Document {i+1} ---")
    print(f"文本长度: {len(doc.text)} 字符")
    print(f"文本预览: {doc.text[:80]}...")
    print(f"Metadata keys: {list(doc.metadata.keys())}")
    print(f"Metadata: {json.dumps(doc.metadata, ensure_ascii=False, indent=2)}")

from llama_index.core.schema import TextNode

# 直接转换为Node格式
nodes = []
for i, chunk in enumerate(chunked_elements):
    # 处理metadata
    metadata = chunk.metadata.to_dict()

    # 如果有原始元素,可以提取更多信息
    if hasattr(chunk.metadata, 'orig_elements') and chunk.metadata.orig_elements:
        # 提取所有原始元素的类型
        metadata['orig_element_types'] = [type(e).__name__ for e in chunk.metadata.orig_elements]
        # 标记是否包含Title
        metadata['contains_title'] = any(
            type(e).__name__ == 'Title' for e in chunk.metadata.orig_elements
        )

    # 移除languages
    metadata.pop('languages', None)
    # 移除orig_elements
    metadata.pop('orig_elements', None)

    # 创建TextNode
    node = TextNode(
        text=chunk.text,
        metadata=metadata,
        id_=f"chunk_{i}",
    )
    nodes.append(node)

print(f"✓ 创建 {len(nodes)} 个TextNodes")
print("=" * 80)
print(nodes[0].text)
print("=" * 80)
print(nodes[0].metadata)

from llama_index.readers.file.unstructured import UnstructuredReader
from unstructured.partition.auto import partition
from llama_index.core import Document
from pathlib import Path

def smart_load(file_path):
    """
    智能文档加载器：根据文件类型选择最佳解析策略

    Args:
        file_path: 文件路径

    Returns:
        解析后的Document对象列表
    """
    file_path = Path(file_path)
    file_ext = file_path.suffix.lower()

    # 定义复杂文件类型（需要高精度解析）
    complex_types = {
        '.pdf',     # PDF文档（可能包含表格、图像、复杂布局）
        '.png', '.jpg', '.jpeg', '.gif', '.bmp', '.tiff',  # 图片文件（需要OCR）
        '.docx', '.doc',  # Word文档（可能包含复杂格式）
        '.pptx', '.ppt',  # PowerPoint（复杂布局）
        '.xlsx', '.xls'   # Excel（表格结构）
    }

    # 简单文件类型（可以用Reader直接处理）
    simple_types = {
        '.txt', '.md', '.csv', '.html', '.xml', '.json'
    }

    if file_ext in complex_types:
        # 复杂文件使用底层解析，获得更好的结构识别
        print(f"检测到复杂文件类型 {file_ext}，使用partition高精度解析")
        try:
            elements = partition(
                filename=str(file_path),
                # 使用hi_res模式进行高精度解析
                strategy="hi_res",
                # 支持中文、英文
                languages=["eng", "chi_sim"],
                # 推断表格结构
                infer_table_structure=True
            )
            # 将解析元素转换为Document对象
            return [Document(text=e.text, metadata={
                "source": str(file_path),
                "element_type": type(e).__name__,
                "file_type": file_ext
            }) for e in elements if e.text.strip()]  # 过滤空文本
        except Exception as e:
            print(f"高精度解析失败，回退到Reader: {e}")
            # 回退到Reader
            reader = UnstructuredReader()
            return reader.load_data(file=file_path)

    else:
        # 简单文件或未知类型优先使用Reader
        print(f"检测到简单文件类型 {file_ext}，使用Reader解析")
        try:
            # 直接使用Reader进行简单解析
            reader = UnstructuredReader()
            # 加载解析后的文档，返回 Document 对象列表
            docs = reader.load_data(file=file_path)
            return docs
        except Exception as e:
            print(f"Reader解析失败，回退到partition: {e}")
            # 回退到底层解析
            elements = partition(filename=str(file_path), strategy="auto")
            return [Document(text=e.text, metadata={"source": str(file_path)}) for e in elements]

documents = smart_load("甬兴证券-AI行业点评报告：海外科技巨头持续发力AI，龙头公司中报业绩亮眼.pdf")

documents[8].text

from llama_index.core.node_parser import (
    SentenceSplitter,
    SemanticSplitterNodeParser,
    HierarchicalNodeParser
)
# from llama_index.embeddings.openai import OpenAIEmbedding
#
# # 创建嵌入模型
# embed_model = OpenAIEmbedding(model="text-embedding-ada-002")

# 创建层次化解析器
hierarchical_parser = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[512, 256, 128]  # 父段落、子段落、孙段落
)

# 创建语义分割器
semantic_splitter = SemanticSplitterNodeParser(
    buffer_size=1,                      # 用于把多少句子为一组做相似度计算。
    breakpoint_percentile_threshold=90, # 相似度阈值，低于该阈值的相似度会被视为断点。
    embed_model=Settings.embed_model             # 嵌入模型
)

# 创建句子分割器作为后备
sentence_splitter = SentenceSplitter(
    chunk_size=500,
    chunk_overlap=50
)

# 混合策略：先使用层次化解析器，再对过大的块使用语义分割器
def hybrid_chunking(documents):
    # 第一阶段：层次化解析
    nodes = hierarchical_parser.get_nodes_from_documents(documents)

    # 第二阶段：对过大的节点进行语义分割
    final_nodes = []
    for node in nodes:
        if len(node.text) > 500:  # 对过大的节点进行二次分割
            sub_nodes = semantic_splitter.get_nodes_from_documents([node])
            final_nodes.extend(sub_nodes)
        else:
            final_nodes.append(node)

    return final_nodes

# 应用混合策略
nodes = hybrid_chunking(documents)
nodes[25].text

from llama_index.core.node_parser import SentenceSplitter
from llama_index.core import Document
from typing import List

# ============= 步骤1: 预处理unstructured elements =============
def create_documents_with_title_context(elements) -> List[Document]:
    """
    将unstructured elements转换为Documents,并将Title信息注入metadata
    """
    documents = []
    current_title_hierarchy = {
        "h1": "",
        "h2": "",
        "h3": "",
    }

    accumulated_text = []
    accumulated_metadata = {}

    for elem in elements:
        elem_type = type(elem).__name__
        elem_metadata = elem.metadata.to_dict()

        # 如果是Title,更新层级信息
        if elem_type == "Title":
            # 如果有累积的文本,先创建Document
            if accumulated_text:
                doc = Document(
                    text="\n\n".join(accumulated_text),
                    metadata=accumulated_metadata.copy()
                )
                documents.append(doc)
                accumulated_text = []

            # 简单的层级判断(可根据实际情况改进)
            title_text = elem.text
            if len(title_text) < 20:  # 短标题可能是低层级
                current_title_hierarchy["h3"] = title_text
            elif len(title_text) < 40:
                current_title_hierarchy["h2"] = title_text
                current_title_hierarchy["h3"] = ""
            else:
                current_title_hierarchy["h1"] = title_text
                current_title_hierarchy["h2"] = ""
                current_title_hierarchy["h3"] = ""

            # 初始化新section的metadata
            accumulated_metadata = {
                **elem_metadata,
                "section_title": title_text,
                "title_h1": current_title_hierarchy["h1"],
                "title_h2": current_title_hierarchy["h2"],
                "title_h3": current_title_hierarchy["h3"],
            }

            # Title本身也加入文本
            accumulated_text.append(f"# {title_text}")

        else:
            # 非Title元素,累积到当前section
            if not accumulated_metadata:
                # 如果还没有metadata(文档开头没有Title的情况)
                accumulated_metadata = {
                    **elem_metadata,
                    "section_title": "前言",
                }

            accumulated_text.append(elem.text)

            # 更新metadata(保留最新的page_number等信息)
            accumulated_metadata.update({
                k: v for k, v in elem_metadata.items()
                if k in ['page_number', 'filename']
            })

    # 处理最后累积的文本
    if accumulated_text:
        doc = Document(
            text="\n\n".join(accumulated_text),
            metadata=accumulated_metadata
        )
        documents.append(doc)

    return documents

# ============= 步骤2: 应用自定义处理 =============
enriched_documents = create_documents_with_title_context(elements)

print("=" * 80)
print("Metadata增强的Documents")
print("=" * 80)
for i, doc in enumerate(enriched_documents[:3]):
    print(f"\n--- Document {i+1} ---")
    print(f"文本长度: {len(doc.text)} 字符")
    print(f"文本预览:\n{doc.text[:150]}...")
    print(f"\nMetadata:")
    for key, value in doc.metadata.items():
        print(f"  {key}: {value}")

# ============= 步骤3: 使用SentenceSplitter进一步切分 =============
"""
如果Documents还是太大,可以进一步切分
关键: metadata会自动继承到所有child nodes
"""

node_parser = SentenceSplitter(
    chunk_size=512,
    chunk_overlap=50,
    separator=" ",
)

nodes = node_parser.get_nodes_from_documents(enriched_documents)

print("\n" + "=" * 80)
print("进一步切分后的Nodes (metadata已继承)")
print("=" * 80)
print(f"总节点数: {len(nodes)}")
for i, node in enumerate(nodes[:3]):
    print(f"\n--- Node {i+1} ---")
    print(f"Node ID: {node.node_id}")
    print(f"文本预览: {node.text[:80]}...")
    print(f"继承的Metadata: {json.dumps(node.metadata, ensure_ascii=False, indent=2)}")

from llama_index.core.settings import Settings
import os
from llama_index.llms.openai import OpenAI          # 导入OpenAI LLM类
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()

# 设置为全局默认Embedding模型
Settings.embed_model = OpenAIEmbedding(
    model="text-embedding-3-small",
    api_key=os.getenv("OPENAI_API_KEY"),
    api_base=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)

# 设置为全局默认 LLM
Settings.llm = OpenAI(
    model="gpt-3.5-turbo",
    api_key=os.getenv("OPENAI_API_KEY"),
    api_base=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)

# ============= 步骤4: 使用Metadata Extractor进一步增强 =============
"""
可选: 使用LLM自动提取更多metadata
"""
# 导入metadata提取器
from llama_index.core.extractors import (
    TitleExtractor,
    SummaryExtractor,
    KeywordExtractor,
    QuestionsAnsweredExtractor,
)
# 导入IngestionPipeline管道
from llama_index.core.ingestion import IngestionPipeline

# 定义metadata提取器
extractors = [
    TitleExtractor(nodes=5),  # 从前5个节点提取标题
    KeywordExtractor(keywords=10),  # 提取10个关键词
    SummaryExtractor(summaries=["self"]),  # 生成摘要
    QuestionsAnsweredExtractor(questions=3),  # 生成3个问题
]

# 构建处理管道
pipeline = IngestionPipeline(
    transformations=[
        node_parser,  # 先切分
        *extractors,  # 再提取metadata
    ]
)

# 执行
enhanced_nodes = pipeline.run(documents=enriched_documents)

# 查看增强后的metadata
for node in enhanced_nodes[:2]:
    print(f"\\nNode: {node.text[:50]}...")
    print(f"Metadata: {node.metadata}")