Azure AI Search大文档向量化：无重复索引的Chunk处理咨询

分块后是否需要重组文本块？ #

不需要重组。向量搜索的核心是通过文本块的嵌入向量匹配用户查询，每个块都是独立的检索单元。但必须给每个块添加原文档ID、块序号、所属章节等元数据，后续可通过这些元数据关联相关块，返回结果时也能还原完整上下文逻辑。

避免同一文档多次出现在搜索结果中 #

可通过两种方式处理：

• 结果聚合：搜索时使用$groupby参数按原文档ID分组，仅返回每个文档中匹配度最高的块，或聚合所有相关块内容后返回完整文档信息。
• 索引设计优化：给索引添加document_id、document_title等原文档级字段，展示搜索结果时，对相同document_id的结果进行合并，只显示一次文档条目，同时在详情里列出所有匹配的文本块。

无Azure存储的大文档分块索引示例（Python） #

以下是手动分块、生成嵌入向量并批量导入Azure AI Search的实现代码：

1. 依赖安装 #

pip install azure-search-documents azure-identity openai langchain

2. 核心代码 #

import os
from azure.search.documents import SearchClient
from azure.identity import DefaultAzureCredential
from openai import OpenAI
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 配置参数
SEARCH_SERVICE_ENDPOINT = os.getenv("AZURE_SEARCH_ENDPOINT")
SEARCH_INDEX_NAME = "large-docs-vector-index"
OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT")
OPENAI_API_KEY = os.getenv("AZURE_OPENAI_API_KEY")
OPENAI_EMBEDDING_MODEL = "text-embedding-ada-002"

# 初始化客户端
search_client = SearchClient(
    endpoint=SEARCH_SERVICE_ENDPOINT,
    index_name=SEARCH_INDEX_NAME,
    credential=DefaultAzureCredential()
)
openai_client = OpenAI(
    api_key=OPENAI_API_KEY,
    base_url=f"{OPENAI_ENDPOINT}/openai/deployments/{OPENAI_EMBEDDING_MODEL}/extensions"
)

# 大文档分块函数
def split_large_document(text, document_id, document_title):
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        length_function=len,
        separators=["\n\n", "\n", " ", ""]
    )
    chunks = text_splitter.split_text(text)
    chunk_docs = []
    for idx, chunk in enumerate(chunks):
        # 生成嵌入向量
        embedding = openai_client.embeddings.create(input=chunk, model=OPENAI_EMBEDDING_MODEL).data[0].embedding
        chunk_docs.append({
            "id": f"{document_id}-chunk-{idx}",
            "document_id": document_id,
            "document_title": document_title,
            "chunk_content": chunk,
            "chunk_index": idx,
            "content_vector": embedding
        })
    return chunk_docs

# 处理单篇大文档示例
with open("large_document.txt", "r", encoding="utf-8") as f:
    doc_text = f.read()

# 生成分块文档并导入索引
chunk_documents = split_large_document(doc_text, "doc-001", "XX行业白皮书")
search_client.upload_documents(documents=chunk_documents)

# 搜索时合并同一文档结果
def search_and_merge_results(query):
    # 生成查询向量
    query_embedding = openai_client.embeddings.create(input=query, model=OPENAI_EMBEDDING_MODEL).data[0].embedding
    # 搜索匹配块
    results = search_client.search(
        search_text=None,
        vector=query_embedding,
        top=10,
        select=["document_id", "document_title", "chunk_content", "chunk_index"]
    )
    # 按文档ID合并结果
    merged_docs = {}
    for result in results:
        doc_id = result["document_id"]
        if doc_id not in merged_docs:
            merged_docs[doc_id] = {
                "title": result["document_title"],
                "chunks": []
            }
        merged_docs[doc_id]["chunks"].append({
            "index": result["chunk_index"],
            "content": result["chunk_content"]
        })
    # 按块序号排序
    for doc in merged_docs.values():
        doc["chunks"].sort(key=lambda x: x["index"])
    return merged_docs

# 测试搜索
query = "XX技术在行业中的应用场景"
merged_results = search_and_merge_results(query)
for doc_id, doc_info in merged_results.items():
    print(f"文档ID: {doc_id}")
    print(f"标题: {doc_info['title']}")
    print("匹配内容:")
    for chunk in doc_info["chunks"]:
        print(f"块{chunk['index']}: {chunk['content']}")

关键说明 #

• 分块采用RecursiveCharacterTextSplitter按语义分割，保留上下文重叠，提升检索准确性。
• 每个索引条目携带原文档ID、块序号等元数据，为后续结果合并提供依据。
• 搜索时通过代码层面合并同一文档的匹配块，避免重复展示文档条目。

内容的提问来源于stack exchange，提问作者Ray Suelzer