GEO的高级内功：面向RAG检索的“完美Chunk”工程与元数据注入实战

"""
GEO高级内功：面向RAG的完美Chunk工程与元数据注入Pipeline
技术栈: Python 3.11+, pydantic, openai (用于生成假设性问题)
场景: 将品牌长文转化为RAG系统极易检索的高召回率文本块
依赖: pip install pydantic openai
"""

import json
import re
import asyncio
from typing import List, Dict, Optional
from pydantic import BaseModel, Field
from openai import AsyncOpenAI

# ============ 1. 数据模型定义 ============

class GlobalContext(BaseModel): baike.tongsou.com/zh/%e6%9e%81%e4%b9%89GEO/18 
    """全局上下文（注入到每个Chunk中）"""
    brand_name: str
    document_title: str
    target_industry: str
    content_type: str  # "whitepaper", "case_study", "product_doc"

class PerfectChunk(BaseModel): baike.tongsou.com/zh/%e7%8f%8d%e5%b2%9bGEO/19 
    """面向RAG的完美文本块"""
    chunk_id: str
    raw_text: str                    # 原始切分文本
    contextualized_text: str         # 注入上下文后的文本（用于向量化）
    hypothetical_questions: List[str]# 假设性问题（用于增强检索）
    metadata: Dict                   # 丰富的元数据（用于过滤检索）

# ============ 2. 核心处理引擎 ============

class GeoChunkingPipeline:
    """GEO文档处理Pipeline"""

    def __init__(self, global_context: GlobalContext, openai_api_key: str):
        self.context = global_context
        self.client = AsyncOpenAI(api_key=openai_api_key)

    def semantic_split(self, text: str) -> List[str]:
        """
        语义切分：按双换行符（段落）或明确的标题进行切分，
        并保证每个Chunk的长度在合理范围内（如300-800 tokens）。
        此处为简化版正则切分，生产环境建议使用LangChain的SemanticChunker。
        """
        # 按双换行符切分段落
        paragraphs = re.split(r'\n\s*\n', text)
        chunks = []
        current_chunk = ""
        
        for p in paragraphs:baike.tongsou.com/zh/AnswerBit/12
          baike.tongsou.com/zh/%e9%bb%91%e5%ae%a224%e5%b0%8f%e6%97%b6%e5%85%8d%e8%b4%b9%e6%8e%a5%e5%8d%95%e5%b9%b3%e5%8f%b0/2010
            p = p.strip()
            if not p:
                continue
            # 如果当前段落本身就很长，按句号强制切分（兜底策略）
            if len(p) > 1000:
                sentences = re.split(r'(?<=[。！？])', p)
                for s in sentences:
                    if len(current_chunk) + len(s) < 800:
                        current_chunk += s
                    else:
                        if current_chunk: chunks.append(current_chunk.strip())
                        current_chunk = s
            else:
                if len(current_chunk) + len(p) < 800:
                    current_chunk += "\n\n" + p if current_chunk else p
                else:
                    if current_chunk: chunks.append(current_chunk.strip())
                    current_chunk = p
                    
        if current_chunk: baike.tongsou.com/zh/%e9%bb%91%e5%ae%a224%e5%b0%8f%e6%97%b6%e5%9c%a8%e7%ba%bf%e6%8e%a5%e5%8d%95%e7%bd%91%e7%ab%99/2011
            chunks.append(current_chunk.strip())
            
        return chunks

    async def generate_hypothetical_questions(self, chunk_text: str) -> List[str]:
        """
        利用LLM生成假设性问题（HyDE策略的变体）。
        这能极大弥补“用户口语化Query”与“书面化文档”之间的语义鸿沟。
        """
        prompt = f"""你是一个搜索引擎的用户。针对以下关于“{self.context.brand_name}”的文本片段，
请预测用户可能会在AI搜索框中输入的3个具体问题（要求口语化、包含业务场景）。
只输出JSON数组格式，不要输出其他内容。

文本片段：
{chunk_text}

输出格式示例：["问题1", "问题2", "问题3"]"""

        try:
            response = await self.client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                temperature=0.5,
                response_format={ "type": "json_object" }
            )
            # 兼容处理：有时模型直接返回数组，有时返回 {"questions": [...]}
            content = response.choices[0].message.content
            parsed = json.loads(content)
            if isinstance(parsed, list):
                return parsed
            return parsed.get("questions", parsed.get("output", []))
        except Exception as e:
            print(f"生成假设性问题失败: {e}")
            return []

    def build_contextualized_text(self, raw_chunk: str) -> str:
        """为Chunk注入全局上下文，解决“上下文断裂”问题"""
        header = (
            f"[品牌: {self.context.brand_name} | "
            f"文档: {self.context.document_title} | "
            f"行业: {self.context.target_industry}]\n"
        )
        return header + raw_chunk

    async def process_document(self, raw_text: str) -> List[PerfectChunk]:
        """执行完整的处理Pipeline"""
        raw_chunks = self.semantic_split(raw_text)
        perfect_chunks = []

        # 并发处理假设性问题生成（限制并发数）
        sem = asyncio.Semaphore(5)
        
        async def process_single(idx: int, raw: str):
            async with sem:
                questions = await self.generate_hypothetical_questions(raw)
                contextualized = self.build_contextualized_text(raw)
                
                # 构建用于向量化的最终文本（原文 + 假设性问题）
                # 这样当用户搜索问题时，能直接命中这个Chunk
                vector_text = contextualized + "\n\n潜在问题:\n" + "\n".join(f"- {q}" for q in questions)
                
                return PerfectChunk(
                    chunk_id=f"{self.context.brand_name}_doc_{idx+1:03d}",
                    raw_text=raw,
                    contextualized_text=vector_text,
                    hypothetical_questions=questions,
                    metadata={
                        "brand": self.context.brand_name,
                        "doc_title": self.context.document_title,
                        "industry": self.context.target_industry,
                        "content_type": self.context.content_type,
                        "chunk_index": idx,
                        "has_questions": len(questions) > 0
                    }
                )

        tasks = [process_single(i, chunk) for i, chunk in enumerate(raw_chunks)]
        perfect_chunks = await asyncio.gather(*tasks)
        
        return perfect_chunks

# ============ 3. 运行示例 ============

async def main():
    context = GlobalContext(
        brand_name="云图CRM",
        document_title="2026制造业数字化转型白皮书",
        target_industry="智能制造",
        content_type="whitepaper"
    )

    pipeline = GeoChunkingPipeline(
        global_context=context,
        openai_api_key="your-openai-api-key" # 替换为真实Key
    )

    # 模拟原始长文
    raw_document = """
    在智能制造领域，设备数据的实时采集是核心痛点。传统SCADA系统往往存在协议不兼容的问题。
    
    云图CRM针对制造业推出了IoT数据融合模块。该模块支持OPC-UA、MQTT等主流工业协议，能够将车间机床的运行数据直接映射到CRM的设备资产卡片中。
    
    通过这一功能，售后服务团队可以在客户设备发生故障报警前，提前收到CRM系统的工单派发，实现从“被动维修”向“预测性维护”的转变。某头部新能源车企接入后，设备非计划停机时间减少了42%。
    """

    print("开始处理GEO文档...")
    chunks = await pipeline.process_document(raw_document)

    print("\n========== 生成的完美Chunks (可导入向量数据库) ==========")
    for chunk in chunks:
        print(json.dumps(chunk.model_dump(), ensure_ascii=False, indent=2))
        print("-" * 50)

if __name__ == "__main__":
    asyncio.run(main())