from langchain_text_splitters importCharacterTextSplitter
sample_text =("LangChain was created by Harrison Chase in 2022. It provides a framework for developing applications ""powered by language models. The library is known for its modularity and ease of use. ""One of its key components is the TextSplitter class, which helps in document chunking.")
text_splitter =CharacterTextSplitter(    separator =" ",# 按空格分割    chunk_size=100,# 增大块大小    chunk_overlap=20,# 调整重叠比例    length_function=len,)
docs = text_splitter.create_documents([sample_text])for i, doc in enumerate(docs):print(f"--- Chunk {i+1} ---")print(doc.page_content)

from langchain_text_splitters importRecursiveCharacterTextSplitter
# 使用与上文相同的 sample_texttext_splitter =RecursiveCharacterTextSplitter(    chunk_size=100,    chunk_overlap=20,# 默认分隔符为 ["\n\n", "\n", " ", ""])
docs = text_splitter.create_documents([sample_text])for i, doc in enumerate(docs):print(f"--- Chunk {i+1} ---")print(doc.page_content)

import nltktry:    nltk.data.find('tokenizers/punkt')except nltk.downloader.DownloadError:    nltk.download('punkt')
from nltk.tokenize import sent_tokenize
def chunk_by_sentences(text, max_chars=500, overlap_sentences=1):    sentences = sent_tokenize(text)    chunks =[]    current_chunk =""for i, sentence in enumerate(sentences):if len(current_chunk)+ len(sentence)<= max_chars:            current_chunk +=" "+ sentenceelse:            chunks.append(current_chunk.strip())# 创建重叠            start_index = max(0, i - overlap_sentences)            current_chunk =" ".join(sentences[start_index:i+1])if current_chunk:        chunks.append(current_chunk.strip())return chunks
long_text ="This is the first sentence. This is the second sentence, which is a bit longer. Now we have a third one. The fourth sentence follows. Finally, the fifth sentence concludes this paragraph."chunks = chunk_by_sentences(long_text, max_chars=100)for i, chunk in enumerate(chunks):print(f"--- Chunk {i+1} ---")print(chunk)

from langchain_text_splitters importMarkdownHeaderTextSplitter
markdown_document ="""# Chapter 1: The Beginning
## Section 1.1: The Old WorldThisis the story of a time long past.
## Section 1.2: A New HopeA new hero emerges.
# Chapter 2: The Journey
## Section 2.1: The Call to AdventureThe hero receives a mysterious call."""
headers_to_split_on =[("#","Header 1"),("##","Header 2"),]
markdown_splitter =MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)md_header_splits = markdown_splitter.split_text(markdown_document)
for split in md_header_splits:print(f"Metadata: {split.metadata}")print(split.page_content)print("-"*20)

dialogue =["Alice: Hi, I'm having trouble with my order.","Bot: I can help with that. What's your order number?","Alice: It's 12345.","Alice: I haven't received any shipping updates.","Bot: Let me check... It seems your order was shipped yesterday.","Alice: Oh, great! Thank you.",]
def chunk_dialogue(dialogue_lines, max_turns_per_chunk=3):    chunks =[]for i in range(0, len(dialogue_lines), max_turns_per_chunk):        chunk ="\n".join(dialogue_lines[i:i + max_turns_per_chunk])        chunks.append(chunk)return chunks
chunks = chunk_dialogue(dialogue)for i, chunk in enumerate(chunks):print(f"--- Chunk {i+1} ---")print(chunk)

import osfrom langchain_experimental.text_splitter importSemanticChunkerfrom langchain_huggingface importHuggingFaceEmbeddings
os.environ["TOKENIZERS_PARALLELISM"]="false"embeddings =HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
# 创建 SemanticChunker 实例# LangChain 的 SemanticChunker 默认使用 percentile 阈值# 可以尝试不同的 breakpoint_threshold_type: "percentile", "standard_deviation", "interquartile", "gradient"text_splitter =SemanticChunker(    embeddings,    breakpoint_threshold_type="percentile",# 使用百分位作为阈值类型    breakpoint_threshold_amount=70# 设置阈值为80)print("SemanticChunker configured.")print("-"*50)
long_text =("The Wright brothers, Orville and Wilbur, were two American aviation pioneers ""generally credited with inventing, building, and flying the world's first successful motor-operated airplane. ""They made the first controlled, sustained flight of a powered, heavier-than-air aircraft on December 17, 1903. ""In the following years, they continued to develop their aircraft. ""Switching topics completely, let's talk about cooking. ""A good pizza starts with a perfect dough, which needs yeast, flour, water, and salt. ""The sauce is typically tomato-based, seasoned with herbs like oregano and basil. ""Toppings can vary from simple mozzarella to a wide range of meats and vegetables. ""Finally, let's consider the solar system. ""It is a gravitationally bound system of the Sun and the objects that orbit it. ""The largest objects are the eight planets, in order from the Sun: Mercury, Venus, Earth, Mars, Jupiter, Saturn, Uranus, and Neptune.")
docs = text_splitter.create_documents([long_text])
for i, doc in enumerate(docs):print(f"--- Chunk {i+1} ---")print(doc.page_content)print()

import numpy as npimport refrom sklearn.feature_extraction.text importCountVectorizerfrom sklearn.decomposition importLatentDirichletAllocationimport nltkfrom nltk.corpus import stopwords
try:    stopwords.words('english')exceptLookupError:    nltk.download('stopwords')
def lda_topic_chunking(text: str, n_topics:int=3)-> list[str]:"""基于LDA主题模型的分块函数。
:param text:需要分块的原始文本。:param n_topics:期望从文本中发现的主题数量。:return:文本块列表。"""# 1. 文本预处理 和 重新定义“文档”单元# 将文本按段落分割，一个段落作为一个“文档”    paragraphs =[p.strip()for p in text.split('\n\n')if p.strip()]if len(paragraphs)<=1:return[text]# 如果只有一个段落，无需分割
# 简单的文本清洗：移除特殊字符，转为小写    cleaned_paragraphs =[re.sub(r'[^a-zA-Z\s]','', p).lower()for p in paragraphs]
# 2. 词袋模型 + 去停用词# min_df=1 因为在单文档上下文中，一个词只在一个段落出现也是有意义的    vectorizer =CountVectorizer(min_df=1, stop_words=stopwords.words('english'))    X = vectorizer.fit_transform(cleaned_paragraphs)
# 如果词汇表为空，则无法进行LDA，直接返回段落if X.shape[1]==0:return paragraphs
# 3. LDA 主题建模# n_components 是正确的参数名    lda =LatentDirichletAllocation(n_components=n_topics, random_state=42)    lda.fit(X)
# 4. 计算每个段落的主导主题    topic_dist = lda.transform(X)# np.argmax 返回每行（每个段落）最大值的索引（即主导主题的编号）    dominant_topics = np.argmax(topic_dist, axis=1)
# 5. 实现正确的分块逻辑：在主题变化时切分    chunks =[]    current_chunk_paragraphs =[]# 第一个段落的主题作为起始主题    current_topic = dominant_topics[0]
for i, paragraph in enumerate(paragraphs):if dominant_topics[i]== current_topic:# 如果主题相同，则加入当前块            current_chunk_paragraphs.append(paragraph)else:# 如果主题变化，则保存上一个块，并开始一个新块            chunks.append("\n\n".join(current_chunk_paragraphs))            current_chunk_paragraphs =[paragraph]            current_topic = dominant_topics[i]
# 添加最后一个块    chunks.append("\n\n".join(current_chunk_paragraphs))
return chunks
document ="""..............................."""
final_chunks = lda_topic_chunking(document, n_topics=3)
print(f"文档被分成了 {len(final_chunks)} 个块：")print("="*80)for i, chunk in enumerate(final_chunks):print(f"--- 块 {i+1} ---")print(chunk)print("-"*80)

from langchain.embeddings importOpenAIEmbeddingsfrom langchain_text_splitters importRecursiveCharacterTextSplitterfrom langchain.retrievers importParentDocumentRetrieverfrom langchain_community.document_loaders importTextLoaderfrom langchain_chroma importChromafrom langchain.storage importInMemoryStore# from langchain_core.documents import Document # 假设 Document 已被导入
# docs = [Document(page_content="......")] # 假设这是你的文档# embeddings = OpenAIEmbeddings()# vectorstore = Chroma(embedding_function=embeddings, collection_name="split_parents")# store = InMemoryStore()
# parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)# child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
# retriever = ParentDocumentRetriever(#     vectorstore=vectorstore,#     docstore=store,#     child_splitter=child_splitter,#     parent_splitter=parent_splitter,# )
# retriever.add_documents(docs)# sub_docs = vectorstore.similarity_search("query")# retrieved_docs = retriever.get_relevant_documents("query")# print(retrieved_docs[0].page_content)

import textwrapfrom langchain_openai importChatOpenAIfrom langchain.prompts importPromptTemplatefrom langchain_core.output_parsers importPydanticOutputParserfrom pydantic importBaseModel,Fieldfrom typing importList
classKnowledgeChunk(BaseModel):    chunk_title: str =Field(description="这个知识块的简洁明了的标题")    chunk_text: str =Field(description="从原文中提取并重组的、自包含的文本内容")    representative_question: str =Field(description="一个可以被这个块内容直接回答的典型问题")
classChunkList(BaseModel):    chunks:List[KnowledgeChunk]
parser =PydanticOutputParser(pydantic_object=ChunkList)
prompt_template ="""【角色】:你是一位顶尖的科学文档分析师，你的任务是将复杂的科学文本段落，分解成一组核心的、自包含的“知识块(KnowledgeChunks)”。【核心任务】:阅读用户提供的文本段落，识别其中包含的独立的核心概念。【规则】:1.**自包含性**:每个“知识块”必须是“自包含的(self-contained)”。2.**概念单一性**:每个“知识块”应该只围绕一个核心概念。3.**提取并重组**:从原文中提取与该核心概念相关的所有句子，并将它们组合成一个通顺、连贯的段落。4.**遵循格式**:严格按照下面的JSON格式指令来构建你的输出。
{format_instructions}
【待处理文本】:{paragraph_text}"""prompt =PromptTemplate(template=prompt_template,    input_variables=["paragraph_text"],    partial_variables={"format_instructions": parser.get_format_instructions()},)
# 以下 model 定义需要根据实际情况修改# model = ChatOpenAI(model="Qwen3-235B-A22B", base_url="http://10.1.18.99:8089/v1",api_key="sk-",temperature=0.0).bind(#     response_format={"type": "json_object"}# )# chain = prompt | model | parser
def agentic_chunker(paragraph_text: str)->List[KnowledgeChunk]:try:# result: ChunkList = chain.invoke({"paragraph_text": paragraph_text})# return result.chunks# 模拟返回，因为无法执行 chain.invokeprint("模拟 agentic_chunker 调用")return[]exceptExceptionas e:return[]
document ="""水循环，也称为水文循环，描述了水在地球表面、之上和之下的连续运动。这个循环至关重要，因为它确保了水对所有生命形式的可用性。循环的第一阶段是蒸发，这是水从海洋、湖泊和河流等表面转化为水蒸气并上升到大气中的过程，植物的蒸腾作用也对此有贡献。当温暖、潮湿的空气上升并冷却时，会发生第二阶段：凝结。在这个阶段，水蒸气变回微小的液态水滴，形成云。随着这些水滴碰撞并增长，它们最终变得足够重，以降水的形式落回地球，这是第三阶段，形式可以是雨、雪、雨夹雪或冰雹。最后，一旦水到达地面，它可能以多种方式移动，构成了第四个阶段：汇集。一些水会作为地表径流流入河流、湖泊和海洋。其他水则会渗入地下，成为地下水，最终也可能返回地表或海洋，从而重新开始整个循环。"""paragraphs = document.strip().split('\n\n')all_chunks =[]
for i, para in enumerate(paragraphs):print(f"--- 正在处理第 {i+1}/{len(paragraphs)} 段 ---")    chunks_from_para = agentic_chunker(para)# 调用新函数if chunks_from_para:        all_chunks.extend(chunks_from_para)print(f"成功从该段落中提取了 {len(chunks_from_para)} 个知识块。")
ifnot all_chunks:print("未能生成任何知识块。")else:for i, chunk in enumerate(all_chunks):print(f"【知识块 {i+1}】")print(f"  - 标题: {chunk.chunk_title}")print(f"  - 代表性问题: {chunk.representative_question}")print(f"  - 文本内容:")        wrapped_text = textwrap.fill(chunk.chunk_text, width=78, initial_indent='    ', subsequent_indent='    ')print(wrapped_text)print("-"*80)

from langchain_text_splitters importMarkdownHeaderTextSplitter,RecursiveCharacterTextSplitterfrom langchain_core.documents importDocument
markdown_document ="""# 第一章：公司简介
本公司成立于2017年，致力于推动人工智能技术的创新与应用。我们的使命是通过先进的AI解决方案，为各行各业赋能，创造更大的价值。我们拥有一支由顶尖科学家和工程师组成的团队，专注于深度学习、自然语言处理和计算机视觉等前沿领域。
## 1.1 发展历程
公司自创立以来，经历了快速的发展。从最初的几人团队，到如今拥有数百名员工的规模，我们始终坚持技术驱动、客户至上的原则。
# 第二章：核心技术
本章将详细介绍我们的核心技术。我们的技术框架基于先进的分布式计算理念，确保了高可用性和可扩展性。系统的核心是一个自主研发的深度学习引擎，它能够处理海量数据并进行高效的模型训练。这个引擎支持多种神经网络结构，包括卷积神经网络（CNNs）用于图像识别，以及循环神经网络（RNNs）和Transformer模型用于自然语言理解。我们特别优化了Transformer架构，提出了一种名为“注意力压缩”的新机制，该机制在保持模型性能的同时，显著减少了计算资源的需求。这一创新使得我们能够在边缘设备上部署复杂的AI模型，为物联网（IoT）应用场景提供了强大的支持。不仅如此，我们还构建了一套完整的数据处理流水线，从数据采集、清洗、标注到最终的模型评估，全程自动化，极大地提升了研发效率。这套流水线处理的数据量已达到PB级别，每日处理的请求数超过十亿次。为了保障数据安全，我们采用了端到-端加密和联邦学习等多种先进技术，确保客户数据的隐私和安全。我们相信，强大的技术实力是公司发展的基石，也是我们服务客户的信心所在。我们不断探索技术的边界，致力于将最新的科研成果转化为可靠、易用的产品和服务，帮助客户在激烈的市场竞争中保持领先地位。
## 2.1 技术原理
我们的技术原理融合了统计学、机器学习和运筹学等多个学科的知识。
# 第三章：未来展望
展望未来，我们将继续加大在人工智能领域的投入，探索通用人工智能（AGI）的可能性。
"""
# 定义混合分块函数def hybrid_chunking_optimized(    markdown_document: str,    coarse_chunk_threshold:int=400,# 定义一个粗粒度块的大小阈值    fine_chunk_size:int=100,# 定义精细切分的目标大小    fine_chunk_overlap:int=20# 定义精细切分的重叠大小)-> list[Document]:"""使用结构化+递归的混合策略，并确保元数据在二次切分中得以保留。
:param markdown_document:完整的Markdown格式文档。:param coarse_chunk_threshold:粗粒度块的长度阈值，超过则进行二次切分。:param fine_chunk_size:二次切分时，每个小块的目标长度。:param fine_chunk_overlap:二次切分时，小块间的重叠长度。:return:Document对象的列表，包含最终的所有分块。"""# 宏观的结构化分块 (按H1和H2标题)    headers_to_split_on =[("#","Header 1"),("##","Header 2")]    markdown_splitter =MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)    coarse_chunks = markdown_splitter.split_text(markdown_document)
# 初始化精细分块器    fine_splitter =RecursiveCharacterTextSplitter(        chunk_size=fine_chunk_size,        chunk_overlap=fine_chunk_overlap)
# 迭代处理，对过大的块进行二次切分    final_chunks =[]for chunk in coarse_chunks:# 检查块的文本长度是否超过阈值if len(chunk.page_content)> coarse_chunk_threshold:print(f"--- 块过大 (长度: {len(chunk.page_content)}), 正在进行二次切分... ---")print(f"    原始元数据: {chunk.metadata}")
            finer_chunks = fine_splitter.split_documents([chunk])            final_chunks.extend(finer_chunks)else:            final_chunks.append(chunk)
return final_chunks
final_chunks = hybrid_chunking_optimized(markdown_document)
for i, chunk in enumerate(final_chunks):print(f"--- 最终块 {i+1} (长度: {len(chunk.page_content)}) ---")print(f"元数据: {chunk.metadata}")print("文本内容:")print(chunk.page_content)print("-"*80)