风雪围城

数据准备

数据采集

Chunk 分解

按照固定大小或者文档结构进行分块，是比较粗暴的分块方式。成本低、速度快，但也容易出现信息撕裂，也并不符合人类本身的分类方式。

更自然的，是通过语义进行分块。对文本的每个句子向量化，然后通过语义相似度进行分块。两个句子相似度高于某个阈值，则认为它们属于同一个主题或者 Chunk。

让我们将这个过程，更加拟人化，引入 Agentic Chunking。Agentic Chunking 通过 LLM 做推理和智能决策，去理解结构和语义，基于意义分界而非物理分解来重构内容。
将文本内容发给 LLM，让 LLM 将内容分解成一个个命题，即，最小单位的知识点。这些知识点将失去上下文耦合，各自独立、完整。这将为后续的主题构建奠定基础。

propositionalization_prompt = """
将以下内容分解为独立完整的命题，确保每个命题脱离上下文也能理解。

规则：
1. 将所有代词（他、她、它、这、那）替换为具体指代对象
2. 将复合句拆分为简单句
3. 确保每个命题是自包含的

内容：
{毛泽东是中国共产党的创始人之一。
他在1949年宣布中华人民共和国成立。
他领导了新民主主义革命。
这场革命彻底改变了中国的命运。}
"""

命题化后，LLM 的输出为：
["毛泽东是中国共产党的创始人之一。",
"毛泽东在1949年宣布中华人民共和国成立。",
"毛泽东领导了新民主主义革命。",
"新民主主义革命彻底改变了中国的命运。"]

通过对原子知识的聚合，可以得到一个主题：

# 逐个处理，动态聚合
chunks = {}

for proposition in propositions:
    # 关键：每次处理一个命题时，通过"匹配"来决定聚合
    
    if not chunks:
        # 第一个命题：创建第一个块
        chunks["chunk_0"] = {
            "title": generate_title(proposition),  # 基于这一个命题
            "propositions": [proposition]
        }
    else:
        # 后续命题：判断应该加入哪个已有块
        matched_chunk = find_best_match(proposition, chunks)
        
        if matched_chunk:
            chunks[matched_chunk].propositions.append(proposition)
        else:
            # 创建新块
            new_chunk_id = f"chunk_{len(chunks)}"
            chunks[new_chunk_id] = {
                "title": generate_title(proposition),
                "propositions": [proposition]
            }

在主题匹配逻辑中，可以按照相似度，甚至直接让 LLM 决策：

def generate_initial_topic(proposition):
    """
    基于首个命题生成主题标题和摘要
    """
    
    prompt = f"""
    基于以下命题，生成主题标题和摘要：
    
    命题：{proposition}
    
    要求：
    - 标题：5-10个字，提取核心实体或概念
    - 摘要：一句话说明主题范围
    
    示例：
    命题："毛泽东出生于湖南湘潭"
    {{
        "title": "毛泽东生平",
        "summary": "关于毛泽东的个人经历"
    }}
    
    返回JSON格式。
    """
    
    response = llm.invoke(prompt)
    return json.loads(response)


def find_best_match_with_llm(proposition, chunks):
    """让 LLM 判断命题应该归入哪个块"""
    
    # 如果没有任何块，返回None（需要创建新块）
    if not chunks:
        return None
    
    # 构建现有块的概要
    chunk_outline = "\n".join([
        f"{chunk_id}: {chunk['title']} - {chunk['summary']}"
        for chunk_id, chunk in chunks.items()
    ])
    
    prompt = f"""
    现有的主题块：
    {chunk_outline}
    
    新命题：{proposition}
    
    问题：这个命题应该归入哪个主题块？
    
    规则：
    1. 如果命题与某个块的主题明确相关，返回该块ID
    2. 如果都不相关，返回 "NEW_CHUNK"
    
    只返回块ID或NEW_CHUNK，不要解释。
    """
    
    response = llm.invoke(prompt)
    
    if response.strip() == "NEW_CHUNK":
        return None
    return response.strip()


# 使用示例
class AgenticChunker:
    def __init__(self):
        self.chunks = {}
        self.llm = ChatOpenAI(model='gpt-4o-mini')
    
    def add_proposition(self, proposition):
        """处理单个命题"""
        
        # 步骤1：判断归属
        matched_chunk_id = find_best_match_with_llm(proposition, self.chunks)
        
        if matched_chunk_id is None:
            # 步骤2：需要创建新块时，生成主题
            chunk_id = f"chunk_{len(self.chunks)}"
            topic = generate_initial_topic(proposition)  # ← 这里生成主题
            
            self.chunks[chunk_id] = {
                "title": topic["title"],
                "summary": topic["summary"],
                "propositions": [proposition]
            }
            print(f"创建新块 [{chunk_id}]: {topic['title']}")
        else:
            # 步骤3：添加到现有块
            self.chunks[matched_chunk_id]["propositions"].append(proposition)
            print(f"添加到 [{matched_chunk_id}]: {self.chunks[matched_chunk_id]['title']}")


# 运行示例
chunker = AgenticChunker()

propositions = [
    "毛泽东出生于湖南湘潭。",           # → 创建块: "毛泽东生平"
    "邓小平推动了改革开放。",           # → 创建块: "邓小平与改革"
    "毛泽东参与创建了中国共产党。",     # → 归入: "毛泽东生平"
    "毛泽东是著名诗人。"                # → 归入: "毛泽东生平"
]

for prop in propositions:
    chunker.add_proposition(prop)

基于语义相似度、通过阈值分类是个经济的手段。直接通过 LLM 决策非常消耗 Token。可以根据实际业务场景需求决定，或者使用小模型来做。

元数据管理

即使，我们完成了 Chunk，在做信息关键检索的时候，结果也未必是精准的。数据块可能比较大，噪声比较多。提炼出 Chunk 中的关键信息作为元数据，在匹配的时候让元数据参与进来，可以提升检索的准确性。

就像是去图书馆找书，我们需要根据书的分类，判断它在哪一层楼，再根据指示，找到它在哪个区域…分类、位置、标签就是这本书的元数据。

一个 Chunk 的元信息，通常应包含：标题、摘要、关键词、分类、标签、来源、作者、发布时间、这个 Chunk 的假设性问题等，都有助于我们对信息做更加精准的定位，也有助于我们构建一个更加全面的关系图谱。

class ChunkMetadata(BaseModel):
    """文档块的结构化元数据"""
    
    summary: str = Field(
        description="块的简洁摘要，1-2句话概括核心内容"
    )
    
    title: str = Field(
        description="块的主题标题，5-10个字"
    )
    
    keywords: List[str] = Field(
        description="关键词列表，包含5-7个核心主题、实体或概念",
        min_items=3,
        max_items=7
    )
    
    hypothetical_questions: List[str] = Field(
        description="这个块能回答的3-5个潜在问题",
        min_items=3,
        max_items=5
    )
    
    entities: Optional[List[str]] = Field(
        description="提取的命名实体（人物、组织、地点等）",
        default=None
    )
    
    table_summary: Optional[str] = Field(
        description="如果块包含表格，用自然语言总结表格的关键信息",
        default=None
    )
    
    chunk_type: Optional[str] = Field(
        description="块的类型：定义/事实/流程/案例/对比等",
        default="事实"
    )

    class Config:
        json_schema_extra = {
            "example": {
                "summary": "关于毛泽东早年生平和参与创建中国共产党的历史记录",
                "title": "毛泽东生平与建党",
                "keywords": ["毛泽东", "湖南湘潭", "中国共产党", "1921年", "创始人"],
                "hypothetical_questions": [
                    "毛泽东出生在哪里？",
                    "中国共产党是什么时候创建的？",
                    "毛泽东在建党过程中的角色是什么？"
                ],
                "entities": ["毛泽东", "湖南湘潭", "中国共产党"],
                "table_summary": None,
                "chunk_type": "事实"
            }
        }

元数据可以帮助我们加强信息密度，提高信噪比。建立在元数据基础上，构建 embedding，增强信息表达能力。

def create_embedding_text(chunk: Dict) -> str:
    """
    增强版：创建更丰富的嵌入文本
    包含更多元数据以提高检索准确性
    """
    
    parts = []
    
    # 1. 主题标题（权重最高）
    if 'title' in chunk:
        parts.append(f"主题: {chunk['title']}")
    
    # 2. 摘要
    parts.append(f"摘要: {chunk['summary']}")
    
    # 3. 关键词
    if chunk.get('keywords'):
        parts.append(f"关键词: {', '.join(chunk['keywords'])}")
    
    # 4. 实体（如果有）
    if chunk.get('entities'):
        parts.append(f"核心实体: {', '.join(chunk['entities'])}")
    
    # 5. 潜在问题（提高问答检索效果）
    if chunk.get('hypothetical_questions'):
        questions = ' | '.join(chunk['hypothetical_questions'][:3])
        parts.append(f"可回答问题: {questions}")
    
    # 6. 实际内容（截断到1000字符）
    content = chunk.get('content', '')
    if isinstance(content, list):  # 如果是命题列表
        content = ' '.join(content)
    parts.append(f"内容: {content[:1000]}")
    
    # 7. 块类型（如果有）
    if chunk.get('chunk_type'):
        parts.append(f"类型: {chunk['chunk_type']}")
    
    return '\n'.join(parts)

数据向量化

语义的相似性，最终要通过量化实现。

def generate_embeddings(texts: List[str], openai_client) -> List[List[float]]:
    """批量生成嵌入向量"""
    
    print(f"\n生成 {len(texts)} 个嵌入向量...")
    
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",  # 1536维
        input=texts
    )
    
    vectors = [item.embedding for item in response.data]
    
    print(f"✓ 生成完成，向量维度: {len(vectors[0])}")
    
    return vectors

最终，数据被表达为一个具有固定维度的空间向量。维度越高，向量空间越大，向量之间的距离或者余弦相似度通常会越准确，当然，准确性也依赖于模型本身的质量。更高维的向量，通常也意味着更大的计算量和存储空间需求。

内容被输入到向量模型中，被转化为向量，从不规则数据转移到规则数据空间。在多维向量空间内，通过距离或者余弦相似度，可以计算出语义的相似性。

# 这里简化为 3 维空间向量
sentences = {
    "A": "毛泽东出生于湖南",
    "B": "毛泽东生于湖南省",      # 与A几乎相同
    "C": "邓小平推动改革开放"      # 完全不同
}

# 假设这是嵌入模型生成的向量（简化为3维）
vectors = {
    "A": np.array([0.8, 0.6, 0.1]),   # 毛泽东出生于湖南
    "B": np.array([0.75, 0.65, 0.15]), # 毛泽东生于湖南省（非常相似）
    "C": np.array([0.2, 0.1, 0.9])    # 邓小平推动改革开放（完全不同）
}

def cosine_similarity(vec1, vec2):
    """计算余弦相似度"""
    # 点积
    dot_product = np.dot(vec1, vec2)
    
    # 向量的模
    norm_vec1 = norm(vec1)
    norm_vec2 = norm(vec2)
    
    # 余弦相似度
    similarity = dot_product / (norm_vec1 * norm_vec2)
    
    return similarity

至于语义的相似性如何和向量空间的距离或者余弦相似度对应起来，这就是向量模型需要解决的问题。

存储介质

文档数据库

存储文档内容

关系型数据库

元数据

向量数据库 Qdrant

存储向量数据

图形数据库 Neo4j

存储图谱数据

数据检索

检索的关键，在于找到最相关的内容，而非最相似的内容，相似性可能会带来更高概率的相关性。为了提高这个概率，需要不同存储介质上的内容协同。

在不同存储介质中的协同，可以提高检索的精准度。但是，没有银弹，只有平衡。高复杂度的协同，可能会带来更高的维护成本和更低的性能，提升的业务价值未必能抵消成本。