flowchart TB
    subgraph External["外部系统"]
        KnowledgeService["Knowledge Service<br/>文档上传"]
        RetrievalService["Retrieval Service<br/>向量检索"]
    end

    subgraph KafkaCluster["Kafka 集群"]
        Topic["document.events<br/>文档事件流"]
    end

    subgraph IndexingService["Indexing Service (本服务)"]
        subgraph API["API 层"]
            HealthAPI["/health, /readiness"]
            StatsAPI["/stats"]
            IncrementalAPI["/incremental/*"]
        end

        subgraph Core["核心处理层"]
            Consumer["Kafka Consumer<br/>事件消费"]
            IncrementalIndexer["IncrementalIndexer<br/>增量索引协调器"]
            VersionManager["VersionManager<br/>版本管理"]
            DocumentProcessor["DocumentProcessor<br/>文档处理协调器"]
        end

        subgraph Modules["功能模块"]
            ParserFactory["ParserFactory<br/>解析器工厂"]
            Parsers["Parsers<br/>PDF/Word/Excel/..."]
            Chunker["DocumentChunker<br/>语义分块"]
            Embedder["BGE_M3_Embedder<br/>向量化"]
            GraphBuilder["GraphBuilder<br/>图谱构建"]
        end
    end

    subgraph Adapters["适配器层"]
        MinIOClient["MinIOClient<br/>文档下载"]
        VectorStoreClient["VectorStoreClient<br/>向量存储适配器"]
        Neo4jClient["Neo4jClient<br/>图数据库适配器"]
    end

    subgraph StorageAdapter["Vector-Store-Adapter"]
        AdapterAPI["适配器 API<br/>/collections/*/insert"]
        MilvusImpl["Milvus 实现"]
        PgVectorImpl["PgVector 实现"]
    end

    subgraph Storage["存储层"]
        MinIO["MinIO<br/>对象存储"]
        Redis["Redis<br/>版本缓存"]
        Milvus["Milvus<br/>向量数据库"]
        Neo4j["Neo4j<br/>知识图谱"]
    end

    %% 外部系统连接
    KnowledgeService -->|上传文档| MinIO
    KnowledgeService -->|发送事件| Topic
    RetrievalService -->|检索向量| Milvus

    %% Kafka 事件流
    Topic --> Consumer

    %% API 调用
    External -.->|HTTP| HealthAPI
    External -.->|HTTP| IncrementalAPI

    %% 核心流程
    Consumer --> IncrementalIndexer
    IncrementalAPI --> IncrementalIndexer

    IncrementalIndexer --> VersionManager
    IncrementalIndexer --> DocumentProcessor

    VersionManager --> Redis

    DocumentProcessor --> ParserFactory
    ParserFactory --> Parsers
    DocumentProcessor --> Chunker
    DocumentProcessor --> Embedder
    DocumentProcessor --> GraphBuilder

    %% 适配器调用
    DocumentProcessor --> MinIOClient
    DocumentProcessor --> VectorStoreClient
    GraphBuilder --> Neo4jClient

    MinIOClient --> MinIO
    VectorStoreClient --> AdapterAPI
    Neo4jClient --> Neo4j

    %% 适配器内部
    AdapterAPI --> MilvusImpl
    AdapterAPI --> PgVectorImpl
    MilvusImpl --> Milvus

    %% 样式
    style External fill:#e1f5fe
    style KafkaCluster fill:#fff3e0
    style IndexingService fill:#f3e5f5
    style API fill:#e8f5e9
    style Core fill:#fff9c4
    style Modules fill:#fce4ec
    style Adapters fill:#e0f2f1
    style StorageAdapter fill:#f1f8e9
    style Storage fill:#ede7f6

    %% 关键路径标注
    linkStyle 2,3,5,6,7,8,9,10 stroke:#f00,stroke-width:2px

flowchart TB
    subgraph External["外部触发层"]
        Upload["Knowledge Service<br/>文档上传"]
        Kafka["Kafka<br/>document.created事件"]
    end

    subgraph IndexingSvc["Indexing Service"]
        Consumer["Kafka Consumer<br/>任务消费"]
        Processor["DocumentProcessor<br/>处理协调"]
    end

    subgraph Parsers["解析器"]
        ParserFactory["ParserFactory<br/>解析器工厂"]
        PDFParser["PDF Parser"]
        DOCXParser["DOCX Parser"]
        TXTParser["TXT Parser"]
        MarkdownParser["Markdown Parser"]
    end

    subgraph Processing["处理层"]
        Chunker["DocumentChunker<br/>语义分块"]
        Embedder["BGE-M3 Embedder<br/>向量化"]
        GraphBuilder["GraphBuilder<br/>图谱构建"]
    end

    subgraph Storage["存储"]
        MinIO["MinIO<br/>文档下载"]
        Milvus["Milvus<br/>向量存储"]
        Neo4j["Neo4j<br/>知识图谱"]
        PostgreSQL["PostgreSQL<br/>状态回写"]
    end

    Upload --> Kafka
    Kafka --> Consumer
    Consumer --> Processor

    Processor --> MinIO
    MinIO --> ParserFactory

    ParserFactory --> PDFParser
    ParserFactory --> DOCXParser
    ParserFactory --> TXTParser
    ParserFactory --> MarkdownParser

    ParserFactory --> Chunker
    Chunker --> Embedder
    Embedder --> Milvus

    Chunker --> GraphBuilder
    GraphBuilder --> Neo4j

    Processor --> PostgreSQL

    style External fill:#e3f2fd
    style IndexingSvc fill:#fff3e0
    style Parsers fill:#f3e5f5
    style Processing fill:#e8f5e9
    style Storage fill:#e0f2f1

flowchart LR
    subgraph API["API层"]
        Health["/health<br/>健康检查"]
        Readiness["/readiness<br/>就绪检查"]
        Stats["/stats<br/>统计信息"]
        Trigger["/trigger<br/>手动触发"]
    end

    subgraph Infrastructure["基础设施层"]
        KafkaConsumer["KafkaConsumer<br/>消息消费"]
        MinIOClient["MinIOClient<br/>对象存储"]
        Neo4jClient["Neo4jClient<br/>图数据库"]
        VectorStoreClient["VectorStoreClient<br/>向量数据库"]
    end

    subgraph Core["核心业务层"]
        DocumentProcessor["DocumentProcessor<br/>文档处理协调器"]

        subgraph Parsers2["解析模块"]
            ParserFactory2["ParserFactory"]
            BaseParser["BaseParser接口"]
            PDFParser2["PDFParser"]
            WordParser["WordParser"]
            ExcelParser["ExcelParser"]
            HTMLParser["HTMLParser"]
        end

        subgraph Processing2["处理模块"]
            DocumentChunker2["DocumentChunker<br/>固定分块"]
            AdaptiveChunker["AdaptiveChunker<br/>自适应分块"]
            BGE_M3_Embedder2["BGE_M3_Embedder<br/>向量化"]
            CachedEmbedder["CachedEmbedder<br/>缓存向量化"]
        end

        subgraph GraphModule["图谱模块"]
            EntityExtractor["EntityExtractor<br/>实体提取"]
            GraphBuilder2["GraphBuilder<br/>图谱构建"]
        end
    end

    API --> DocumentProcessor
    KafkaConsumer --> DocumentProcessor

    DocumentProcessor --> MinIOClient
    DocumentProcessor --> ParserFactory2
    DocumentProcessor --> DocumentChunker2
    DocumentProcessor --> BGE_M3_Embedder2
    DocumentProcessor --> VectorStoreClient
    DocumentProcessor --> GraphBuilder2

    ParserFactory2 -.实例化.-> BaseParser
    BaseParser -.实现.-> PDFParser2
    BaseParser -.实现.-> WordParser
    BaseParser -.实现.-> ExcelParser
    BaseParser -.实现.-> HTMLParser

    DocumentChunker2 -.继承.-> AdaptiveChunker
    BGE_M3_Embedder2 -.继承.-> CachedEmbedder

    GraphBuilder2 --> EntityExtractor
    GraphBuilder2 --> Neo4jClient

    style API fill:#e3f2fd
    style Infrastructure fill:#fff3e0
    style Core fill:#e8f5e9
    style Parsers2 fill:#f3e5f5
    style Processing2 fill:#fce4ec
    style GraphModule fill:#e0f2f1

graph TB
    subgraph "主流程"
        A[main.py] --> B[FastAPI App]
        B --> C[DocumentEventConsumer]
        B --> D[DocumentProcessor]
    end

    subgraph "DocumentProcessor依赖"
        D --> E[ParserFactory]
        D --> F[DocumentChunker]
        D --> G[BGE_M3_Embedder]
        D --> H[GraphBuilder]
        D --> I[MinIOClient]
        D --> J[VectorStoreClient]
        D --> K[Neo4jClient]
    end

    subgraph "Parser层级"
        E --> L[BaseParser接口]
        L --> M[PDFParser]
        L --> N[WordParser]
        L --> O[ExcelParser]
        L --> P[HTMLParser]
        L --> Q[MarkdownParser]
        L --> R[TextParser]
        L --> S[PPTParser]
    end

    subgraph "Chunker层级"
        F --> T[RecursiveCharacterTextSplitter<br/>LangChain]
        F --> U[AdaptiveChunker]
    end

    subgraph "Embedder层级"
        G --> V[SentenceTransformer<br/>BAAI/bge-m3]
        G --> W[CachedEmbedder]
    end

    subgraph "GraphBuilder依赖"
        H --> X[EntityExtractor]
        H --> K
    end

    subgraph "外部库"
        M --> Y[pdfplumber]
        M --> Z[PyPDF2]
        N --> AA[python-docx]
        O --> AB[openpyxl]
        P --> AC[BeautifulSoup4]
        S --> AD[python-pptx]
    end

    C -.消费事件.-> D

    style A fill:#e3f2fd
    style B fill:#e3f2fd
    style D fill:#fff3e0
    style H fill:#e8f5e9
    style E fill:#f3e5f5
    style F fill:#fce4ec
    style G fill:#e1bee7

classDiagram
    class DocumentProcessor {
        -ParserFactory parser_factory
        -DocumentChunker chunker
        -BGE_M3_Embedder embedder
        -VectorStoreClient vector_store_client
        -MinIOClient minio_client
        -Neo4jClient neo4j_client
        -GraphBuilder graph_builder
        -dict stats
        +process_document(document_id, tenant_id, file_path)
        -_download_document()
        -_parse_document()
        -_chunk_document()
        -_vectorize_chunks()
        -_store_vectors()
        -_build_graph()
        +get_stats()
    }

    class ParserFactory {
        -dict parsers
        +get_parser(file_path)
        +register_parser(extension, parser)
    }

    class BaseParser {
        <<interface>>
        +parse(file_data)
        +extract_metadata(file_data)
        +clean_text(text)
    }

    class PDFParser {
        +parse(file_data)
        -_parse_with_pdfplumber()
        -_parse_with_pypdf2()
        +extract_metadata()
    }

    class DocumentChunker {
        -int chunk_size
        -int chunk_overlap
        -RecursiveCharacterTextSplitter splitter
        +chunk(text, document_id)
        -_generate_chunk_id()
        -_estimate_tokens()
    }

    class AdaptiveChunker {
        -dict strategies
        +chunk(text, document_id, doc_type)
    }

    class BGE_M3_Embedder {
        -string model_name
        -int dimension
        -SentenceTransformer model
        +embed_query(text)
        +embed_batch(texts)
    }

    class GraphBuilder {
        -Neo4jClient neo4j_client
        +extract(text)
        +build(entities, relationships, doc_id)
    }

    class Chunk {
        +string id
        +string content
        +int index
        +int tokens
        +dict metadata
    }

    class ProcessingResult {
        +string document_id
        +string status
        +int chunks_count
        +float duration
        +string error
        +is_success()
    }

    DocumentProcessor "1" *-- "1" ParserFactory
    DocumentProcessor "1" *-- "1" DocumentChunker
    DocumentProcessor "1" *-- "1" BGE_M3_Embedder
    DocumentProcessor "1" *-- "1" GraphBuilder
    DocumentProcessor ..> ProcessingResult : returns

    ParserFactory ..> BaseParser : creates
    BaseParser <|.. PDFParser : implements

    DocumentChunker <|-- AdaptiveChunker : extends
    DocumentChunker ..> Chunk : creates

# app/infrastructure/kafka_consumer.py
class DocumentEventConsumer:
    async def start(self):
        """启动Kafka消费者"""
        self.consumer = Consumer(self.config)
        self.consumer.subscribe([self.topic])  # document.events
        self._running = True
        await self._consume_loop()

# main.py (行 56-89)
@asynccontextmanager
async def lifespan(app: FastAPI):
    kafka_consumer = DocumentEventConsumer()
    document_processor = DocumentProcessor()

    # 异步启动Kafka消费者
    asyncio.create_task(kafka_consumer.start())

    yield

    # 关闭资源
    await kafka_consumer.stop()
    await document_processor.cleanup()

# app/infrastructure/kafka_consumer.py (行 94-120)
async def _consume_loop(self):
    while self._running:
        msg = self.consumer.poll(timeout=1.0)
        if msg and not msg.error():
            await self._handle_message(msg)

# app/infrastructure/kafka_consumer.py (行 121-153)
async def _handle_message(self, msg):
    event = json.loads(msg.value().decode("utf-8"))
    event_type = event.get("event_type")  # document.uploaded
    payload = event.get("payload", {})

    # 分发到注册的处理器
    handlers = self.handlers.get(event_type, [])
    for handler in handlers:
        await handler(payload)

sequenceDiagram
    autonumber
    participant Main as main.py
    participant App as FastAPI App
    participant Consumer as DocumentEventConsumer
    participant Kafka as Kafka Broker
    participant Handler as document_uploaded_handler
    participant Processor as DocumentProcessor

    Main->>App: uvicorn.run(app)
    App->>App: lifespan startup
    App->>Consumer: __init__()
    App->>Processor: __init__()

    App->>Consumer: asyncio.create_task(start())
    Consumer->>Kafka: subscribe("document.events")

    loop 消费循环
        Consumer->>Kafka: poll(timeout=1.0)
        Kafka-->>Consumer: message

        Consumer->>Consumer: _handle_message(msg)
        Consumer->>Consumer: json.loads(msg.value())
        Consumer->>Consumer: 提取event_type和payload

        Consumer->>Handler: await handler(payload)
        Handler->>Processor: process_document(doc_id, tenant_id, file_path)
        Processor-->>Handler: ProcessingResult
        Handler-->>Consumer: success
    end

    Note over Consumer,Kafka: 轮询间隔1秒，最大轮询间隔5分钟<br/>自动提交偏移量（5秒间隔）

sequenceDiagram
    autonumber
    participant Processor as DocumentProcessor
    participant MinIO as MinIOClient
    participant Parser as ParserFactory
    participant PDF as PDFParser
    participant Chunker as DocumentChunker
    participant Embedder as BGE_M3_Embedder
    participant Vector as VectorStoreClient
    participant Graph as GraphBuilder
    participant Neo4j as Neo4jClient

    Processor->>Processor: process_document(doc_id, tenant_id, file_path)
    Processor->>Processor: start_time = time.time()

    rect rgb(230, 240, 255)
        Note over Processor,MinIO: 步骤1：下载文档
        Processor->>MinIO: _download_document(file_path)
        MinIO->>MinIO: download_file(file_path)
        MinIO-->>Processor: file_data (bytes)
    end

    rect rgb(255, 240, 230)
        Note over Processor,PDF: 步骤2：解析文档
        Processor->>Parser: _parse_document(file_data, file_path)
        Parser->>Parser: get_parser(file_path)
        Parser->>PDF: parse(file_data)
        PDF->>PDF: _parse_with_pdfplumber(file_data)
        alt pdfplumber成功且文本>100字符
            PDF-->>Parser: text
        else 降级到PyPDF2
            PDF->>PDF: _parse_with_pypdf2(file_data)
            PDF-->>Parser: text
        end
        Parser->>Parser: clean_text(text)
        Parser-->>Processor: text (纯文本)
    end

    rect rgb(230, 255, 230)
        Note over Processor,Chunker: 步骤3：语义分块
        Processor->>Chunker: _chunk_document(text, doc_id)
        Chunker->>Chunker: splitter.split_text(text)
        loop 遍历每个chunk
            Chunker->>Chunker: _generate_chunk_id(doc_id, index, text)
            Chunker->>Chunker: _estimate_tokens(text)
        end
        Chunker-->>Processor: chunks (List[Dict])
    end

    rect rgb(255, 230, 255)
        Note over Processor,Embedder: 步骤4：批量向量化
        Processor->>Embedder: _vectorize_chunks(chunks)
        Embedder->>Embedder: embed_batch(texts)
        loop 分批处理 (batch_size=32)
            Embedder->>Embedder: model.encode(batch)
        end
        Embedder-->>Processor: embeddings (List[1024-dim])
    end

    rect rgb(230, 255, 255)
        Note over Processor,Vector: 步骤5：存储向量
        Processor->>Vector: _store_vectors(chunks, embeddings, doc_id, tenant_id)
        Vector->>Vector: 准备数据 (chunk_id, content, embedding, metadata)
        Vector->>Vector: insert_batch(data)
        Vector-->>Processor: success
    end

    par 步骤6：异步构建图谱（不阻塞）
        Processor->>Graph: asyncio.create_task(_build_graph(...))
        Graph->>Graph: extract(text)
        Graph->>Graph: _extract_entities(text)
        Graph->>Graph: _extract_relationships(text, entities)
        Graph->>Neo4j: batch_create_nodes(entities)
        Graph->>Neo4j: batch_create_relationships(relationships)
        Neo4j-->>Graph: success
    end

    Processor->>Processor: 更新统计信息
    Processor->>Processor: duration = time.time() - start_time
    Processor-->>Processor: return {status, doc_id, chunks_count, duration}

# app/core/document_processor.py (行 111-114)
async def _download_document(self, file_path: str) -> bytes:
    logger.info(f"Downloading document: {file_path}")
    return await self.minio_client.download_file(file_path)

# app/core/document_processor.py (行 116-128)
async def _parse_document(self, file_data: bytes, file_path: str) -> str:
    logger.info(f"Parsing document: {file_path}")
    parser = self.parser_factory.get_parser(file_path)
    text = await parser.parse(file_data)
    logger.info(f"Parsed document, text length: {len(text)} chars")
    return text

# app/core/parsers/pdf_parser.py (行 18-33)
async def parse(self, file_data: bytes, **kwargs) -> str:
    try:
        # 主解析器：pdfplumber
        text = await self._parse_with_pdfplumber(file_data)

        if not text or len(text) < 100:
            # 降级到PyPDF2
            logger.warning("pdfplumber failed, trying PyPDF2")
            text = await self._parse_with_pypdf2(file_data)

        return self.clean_text(text)
    except Exception as e:
        logger.error(f"Failed to parse PDF: {e}")
        raise

# app/core/document_processor.py (行 130-138)
async def _chunk_document(self, text: str, document_id: str) -> List[Dict]:
    logger.info(f"Chunking document: {document_id}")
    chunks = await self.chunker.chunk(text, document_id)
    logger.info(f"Created {len(chunks)} chunks")
    return chunks

# app/core/chunker.py (行 40-77)
async def chunk(self, text: str, document_id: str) -> List[Dict]:
    if not text or not text.strip():
        return []

    # LangChain分块
    chunks = self.splitter.split_text(text)

    # 构建结果
    result = []
    for i, chunk_text in enumerate(chunks):
        chunk_id = self._generate_chunk_id(document_id, i, chunk_text)
        result.append({
            "id": chunk_id,
            "content": chunk_text,
            "index": i,
            "tokens": self._estimate_tokens(chunk_text),
            "metadata": {
                "document_id": document_id,
                "chunk_index": i,
                "total_chunks": len(chunks),
            },
        })
    return result

# app/core/document_processor.py (行 140-152)
async def _vectorize_chunks(self, chunks: List[Dict]) -> List[List[float]]:
    logger.info(f"Vectorizing {len(chunks)} chunks")
    texts = [chunk["content"] for chunk in chunks]
    embeddings = await self.embedder.embed_batch(texts)
    logger.info(f"Generated {len(embeddings)} embeddings")
    return embeddings

# app/core/embedder.py (行 57-84)
async def embed_batch(self, texts: List[str]) -> List[List[float]]:
    if not texts:
        return []

    # 过滤空文本
    valid_texts = [text if text and text.strip() else " " for text in texts]

    # 批量生成向量
    embeddings = self.model.encode(
        valid_texts,
        batch_size=self.batch_size,  # 32
        show_progress_bar=len(valid_texts) > 100,
        convert_to_numpy=True,
    )

    return embeddings.tolist()

# app/core/document_processor.py (行 154-175)
async def _store_vectors(self, chunks: List[Dict], embeddings: List[List[float]],
                       document_id: str, tenant_id: str = None):
    logger.info(f"Storing {len(embeddings)} vectors to Milvus")

    # 准备数据
    data = []
    for chunk, embedding in zip(chunks, embeddings):
        data.append({
            "chunk_id": chunk["id"],
            "document_id": document_id,
            "tenant_id": tenant_id or "default",
            "content": chunk["content"],
            "embedding": embedding,
            "metadata": chunk.get("metadata", {}),
        })

    # 批量插入
    await self.vector_store_client.insert_batch(data)

# app/core/document_processor.py (行 77-80)
asyncio.create_task(
    self._build_graph(text, document_id, tenant_id)
)

# app/core/document_processor.py (行 176-196)
async def _build_graph(self, text: str, document_id: str, tenant_id: str = None):
    try:
        entities, relationships = await self.graph_builder.extract(text)
        await self.graph_builder.build(
            entities=entities,
            relationships=relationships,
            document_id=document_id,
            tenant_id=tenant_id,
        )
    except Exception as e:
        logger.error(f"Error building graph: {e}", exc_info=True)

sequenceDiagram
    autonumber
    participant Processor as DocumentProcessor
    participant Factory as ParserFactory
    participant PDF as PDFParser
    participant Plumber as pdfplumber
    participant PyPDF as PyPDF2

    Processor->>Factory: get_parser(file_path)
    Factory->>Factory: 提取扩展名: .pdf
    Factory->>PDF: __init__()
    Factory-->>Processor: PDFParser实例

    Processor->>PDF: parse(file_data)

    rect rgb(230, 255, 230)
        Note over PDF,Plumber: 主解析器：pdfplumber
        PDF->>Plumber: _parse_with_pdfplumber(file_data)
        Plumber->>Plumber: pdfplumber.open(BytesIO(file_data))

        loop 遍历每一页
            Plumber->>Plumber: page.extract_text()
            Plumber->>Plumber: text_parts.append(text)
        end

        Plumber->>Plumber: "\n\n".join(text_parts)
        Plumber-->>PDF: text

        PDF->>PDF: 检查文本长度

        alt 文本长度 >= 100字符
            Note over PDF: pdfplumber成功
        else 文本过短或为空
            rect rgb(255, 230, 230)
                Note over PDF,PyPDF: 降级解析器：PyPDF2
                PDF->>PyPDF: _parse_with_pypdf2(file_data)
                PyPDF->>PyPDF: PyPDF2.PdfReader(BytesIO(file_data))

                loop 遍历每一页
                    PyPDF->>PyPDF: page.extract_text()
                    PyPDF->>PyPDF: text_parts.append(text)
                end

                PyPDF->>PyPDF: "\n\n".join(text_parts)
                PyPDF-->>PDF: text
            end
        end
    end

    PDF->>PDF: clean_text(text)
    PDF->>PDF: 去除多余空格和特殊字符
    PDF-->>Processor: 清理后的text

    Note over PDF,PyPDF: 降级策略：pdfplumber失败或文本<100字符时切换
    Note over PDF,PyPDF: 准确率：pdfplumber 95%+，PyPDF2 85%+
    Note over PDF,PyPDF: 性能：10页PDF 2-3秒，100页PDF 15-20秒

sequenceDiagram
    autonumber
    participant Processor as DocumentProcessor
    participant Chunker as DocumentChunker
    participant Splitter as RecursiveCharacterTextSplitter

    Processor->>Chunker: chunk(text, document_id)

    Chunker->>Chunker: 检查文本是否为空
    alt 文本为空
        Chunker-->>Processor: []
    end

    rect rgb(230, 240, 255)
        Note over Chunker,Splitter: LangChain递归分块
        Chunker->>Splitter: split_text(text)

        Splitter->>Splitter: 尝试分隔符: "\n\n"
        alt 分块成功且长度<=chunk_size
            Note over Splitter: 段落级分块
        else 继续尝试下一个分隔符
            Splitter->>Splitter: 尝试分隔符: "\n"
            alt 分块成功
                Note over Splitter: 行级分块
            else 继续尝试
                Splitter->>Splitter: 尝试句号: "。", "."
                alt 分块成功
                    Note over Splitter: 句子级分块
                else 最终兜底
                    Splitter->>Splitter: 字符级硬切分
                end
            end
        end

        Splitter->>Splitter: 应用chunk_overlap (50字符)
        Splitter-->>Chunker: chunks (List[str])
    end

    rect rgb(255, 240, 230)
        Note over Chunker: 构建Chunk结构
        loop 遍历每个chunk_text
            Chunker->>Chunker: _generate_chunk_id(doc_id, index, text)
            Chunker->>Chunker: MD5哈希前8位

            Chunker->>Chunker: _estimate_tokens(text)
            Chunker->>Chunker: 中文字符 × 1.5 + 英文单词 × 1.3

            Chunker->>Chunker: 构建chunk对象
            Note over Chunker: {id, content, index, tokens, metadata}
        end
    end

    Chunker-->>Processor: chunks (List[Dict])

    Note over Chunker,Splitter: 分块策略：优先语义边界，兜底字符切分
    Note over Chunker,Splitter: chunk_size=500, overlap=50 (10%重叠)
    Note over Chunker,Splitter: 10页PDF(5000字) → 10-15 chunks

sequenceDiagram
    autonumber
    participant Processor as DocumentProcessor
    participant Embedder as BGE_M3_Embedder
    participant Model as SentenceTransformer
    participant GPU as CUDA Device

    Processor->>Embedder: embed_batch(texts)

    Embedder->>Embedder: 过滤空文本
    loop 遍历texts
        alt text为空或空白
            Embedder->>Embedder: 替换为单空格 " "
        end
    end

    rect rgb(230, 240, 255)
        Note over Embedder,GPU: 批量向量化（batch_size=32）
        loop 分批处理
            Embedder->>Embedder: 提取batch (32条)

            Embedder->>Model: encode(batch, batch_size=32)
            Model->>Model: 文本tokenize
            Model->>Model: 截断或padding到max_length (512)

            Model->>GPU: 前向推理
            GPU->>GPU: BERT模型推理
            GPU->>GPU: Pooling (CLS或Mean)
            GPU->>GPU: L2归一化
            GPU-->>Model: embeddings (32 × 1024)

            Model-->>Embedder: batch_embeddings
            Embedder->>Embedder: all_embeddings.extend(batch)
        end
    end

    Embedder->>Embedder: embeddings.tolist()
    Embedder-->>Processor: List[List[float]]

    Note over Embedder,GPU: 批量处理：32条/批，相比单条提升10倍+
    Note over Embedder,GPU: 性能：50 chunks约3-5秒(GPU)，30-50秒(CPU)
    Note over Embedder,GPU: 显存占用：2-3GB (batch_size=32)

sequenceDiagram
    autonumber
    participant Processor as DocumentProcessor
    participant Client as VectorStoreClient
    participant Milvus as Milvus Server

    Processor->>Client: insert_batch(data)

    Client->>Client: 准备插入数据
    loop 遍历data
        Client->>Client: 构建记录
        Note over Client: {chunk_id, doc_id, tenant_id,<br/>content, embedding, metadata}
    end

    Client->>Client: 检查batch大小
    alt batch_size > 10000
        Client->>Client: 拆分为多个批次
    end

    rect rgb(230, 240, 255)
        Note over Client,Milvus: Milvus批量插入
        loop 每个批次
            Client->>Milvus: collection.insert(entities)
            Milvus->>Milvus: 根据tenant_id选择分区
            Milvus->>Milvus: 写入向量数据
            Milvus->>Milvus: 更新主键索引
            Milvus->>Milvus: 触发HNSW索引增量更新
            Milvus-->>Client: insert_result

            Client->>Client: 检查insert_result.err_count
            alt err_count > 0
                Client->>Client: 记录错误
                Note over Client: 主键冲突或数据格式错误
            end
        end
    end

    Client->>Milvus: collection.flush()
    Note over Client,Milvus: 确保数据持久化到磁盘

    Client-->>Processor: success

    Note over Client,Milvus: 批量插入：单批次最大10000条
    Note over Client,Milvus: 性能：50向量约1-2秒，1000向量约10-15秒
    Note over Client,Milvus: 索引更新：异步增量更新，不阻塞插入

sequenceDiagram
    autonumber
    participant Processor as DocumentProcessor
    participant Client as VectorStoreClient<br/>(algo/common)
    participant Adapter as vector-store-adapter<br/>HTTP Service
    participant Milvus as Milvus Server

    Processor->>Client: insert_batch(data_list)

    rect rgb(230, 240, 255)
        Note over Client,Adapter: 步骤1-2: HTTP调用适配器服务
        Client->>Client: 准备请求JSON
        Client->>Client: {backend: "milvus", data: [...]}

        Client->>Adapter: POST /collections/{name}/insert
        Note over Adapter: 接收请求并验证
    end

    rect rgb(255, 240, 230)
        Note over Adapter,Milvus: 步骤3-6: 适配器转换为Milvus调用
        Adapter->>Adapter: 选择后端实现(Milvus/PgVector)
        Adapter->>Adapter: 转换数据格式

        Adapter->>Milvus: collection.insert(entities)
        Note over Milvus: 批量写入向量数据<br/>更新HNSW索引

        Milvus-->>Adapter: insert_result
        Adapter->>Adapter: 检查错误
    end

    rect rgb(230, 255, 230)
        Note over Adapter,Client: 步骤7-8: 返回结果
        Adapter-->>Client: HTTP 200 + JSON result
        Client->>Client: 解析响应
        Client-->>Processor: success
    end

    Note over Client,Milvus: 适配器模式优势:<br/>1. 解耦服务与存储<br/>2. 支持多后端切换<br/>3. 统一错误处理<br/>4. 集中式监控

# algo/common/vector_store_client.py
class VectorStoreClient:
    def __init__(self, base_url: str = "http://vector-store-adapter:8003"):
        self.base_url = base_url
        self.backend = "milvus"  # 或 "pgvector"
        self.client = httpx.AsyncClient(base_url=base_url, timeout=30.0)

    async def insert_batch(self, data_list: List[Dict]):
        """批量插入向量"""
        response = await self.client.post(
            f"/collections/{self.collection_name}/insert",
            json={"backend": self.backend, "data": data_list}
        )
        response.raise_for_status()
        return response.json()

    async def search(self, query_vector, top_k=10, tenant_id=None):
        """向量检索"""
        response = await self.client.post(
            f"/collections/{self.collection_name}/search",
            json={
                "backend": self.backend,
                "query_vector": query_vector,
                "top_k": top_k,
                "tenant_id": tenant_id
            }
        )
        return response.json()["results"]

{
    "document_id": "doc_123",
    "tenant_id": "tenant_abc",
    "user_id": "user_456",
    "file_path": "tenant_abc/2025-01-27/doc.pdf",
    "collection_id": "coll_789",
    "created_at": "2025-01-27T10:00:00Z"
}

async def process_document(self, document_id: str, tenant_id: str,
                          user_id: str, file_path: str) -> Dict:
    """
    处理文档（完整流程）
    """
    start_time = time.time()

    try:
        # 1. 下载文档
        file_data = await self._download_document(file_path)
        # MinIO下载，返回bytes

        # 2. 解析文档
        text = await self._parse_document(file_data, file_path)
        # 根据扩展名选择解析器（PDF/DOCX/TXT等）
        # 提取纯文本内容

        # 3. 分块
        chunks = await self._chunk_document(text, document_id)
        # RecursiveCharacterTextSplitter语义分块
        # 生成chunk_id、估算tokens

        # 4. 向量化
        embeddings = await self._vectorize_chunks(chunks)
        # BGE-M3批量向量化
        # 返回List[List[float]]，每个1024维向量

        # 5. 存储到Milvus
        await self._store_vectors(chunks, embeddings, document_id, tenant_id)
        # 批量插入Milvus
        # 字段：chunk_id, doc_id, tenant_id, content, embedding, metadata

        # 6. 构建知识图谱（异步，不阻塞）
        asyncio.create_task(
            self._build_graph(text, document_id, tenant_id)
        )
        # 提取实体和关系
        # 存储到Neo4j

        duration = time.time() - start_time

        return {
            "status": "success",
            "document_id": document_id,
            "chunks_count": len(chunks),
            "duration": duration,
        }

    except Exception as e:
        # 记录错误日志
        # 更新文档状态为FAILED
        return {
            "status": "failed",
            "document_id": document_id,
            "error": str(e),
        }

async def _parse_document(self, file_data: bytes, file_path: str) -> str:
    """
    解析文档，提取文本内容
    """
    # 1. 根据文件扩展名选择解析器
    parser = self.parser_factory.get_parser(file_path)
    # 支持：.pdf, .docx, .txt, .md, .html, .xlsx, .pptx

    # 2. 解析文档
    text = await parser.parse(file_data)
    # 每个解析器实现BaseParser接口

    return text

# PDF解析器示例
class PDFParser(BaseParser):
    async def parse(self, file_data: bytes) -> str:
        """解析PDF文档"""
        try:
            # 优先使用pdfplumber（更准确）
            text = await self._parse_with_pdfplumber(file_data)

            if not text or len(text) < 100:
                # 降级使用PyPDF2
                text = await self._parse_with_pypdf2(file_data)

            return self.clean_text(text)
        except Exception as e:
            # 记录错误并抛出
            raise

    async def _parse_with_pdfplumber(self, file_data: bytes) -> str:
        """使用pdfplumber解析"""
        text_parts = []

        with pdfplumber.open(io.BytesIO(file_data)) as pdf:
            for page in pdf.pages:
                text = page.extract_text()
                if text:
                    text_parts.append(text)

        return "\n\n".join(text_parts)

async def _chunk_document(self, text: str, document_id: str) -> List[Dict]:
    """
    分块文档，保证语义完整性
    """
    if not text or not text.strip():
        return []

    # 1. 使用LangChain RecursiveCharacterTextSplitter
    chunks = self.splitter.split_text(text)
    # separators：按段落、句子、标点分割
    # chunk_size：500字符
    # chunk_overlap：50字符（10%重叠）

    # 2. 构建chunk结构
    result = []
    for i, chunk_text in enumerate(chunks):
        chunk_id = self._generate_chunk_id(document_id, i, chunk_text)

        result.append({
            "id": chunk_id,
            "content": chunk_text,
            "index": i,
            "tokens": self._estimate_tokens(chunk_text),
            "metadata": {
                "document_id": document_id,
                "chunk_index": i,
                "total_chunks": len(chunks),
            },
        })

    return result

def _generate_chunk_id(self, document_id: str, index: int, text: str) -> str:
    """
    生成chunk ID，保证唯一性

    格式：{document_id}_{index}_{text_hash}
    text_hash取MD5前8位，避免内容变化导致ID冲突
    """
    text_hash = hashlib.md5(text.encode()).hexdigest()[:8]
    return f"{document_id}_{index}_{text_hash}"

def _estimate_tokens(self, text: str) -> int:
    """
    估算Token数量

    策略：
    - 中文字符：1字符 ≈ 1.5 tokens
    - 英文单词：1单词 ≈ 1.3 tokens

    更准确的方法是使用tiktoken，但有性能开销
    """
    chinese_chars = sum(1 for char in text if "\u4e00" <= char <= "\u9fff")
    english_words = len([word for word in text.split() if any(c.isalpha() for c in word)])

    estimated_tokens = int(chinese_chars * 1.5 + english_words * 1.3)
    return estimated_tokens

async def _vectorize_chunks(self, chunks: List[Dict]) -> List[List[float]]:
    """
    批量向量化chunk，提升吞吐量
    """
    # 1. 提取文本
    texts = [chunk["content"] for chunk in chunks]

    # 2. 批量向量化（batch_size=32）
    embeddings = await self.embedder.embed_batch(texts)
    # BGE-M3模型
    # 输入：List[str]
    # 输出：List[List[float]]，每个1024维

    return embeddings

# BGE-M3 Embedder实现
class BGE_M3_Embedder:
    async def embed_batch(self, texts: List[str]) -> List[List[float]]:
        """批量向量化"""
        # 分批处理，避免内存溢出
        batch_size = 32
        all_embeddings = []

        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]

            # SentenceTransformer模型推理
            embeddings = self.model.encode(
                batch,
                batch_size=batch_size,
                show_progress_bar=False,
                normalize_embeddings=True,  # L2归一化
            )

            all_embeddings.extend(embeddings.tolist())

        return all_embeddings

async def _store_vectors(self, chunks: List[Dict], embeddings: List[List[float]],
                       document_id: str, tenant_id: str):
    """
    批量存储向量到Milvus
    """
    # 1. 准备数据
    data = []
    for chunk, embedding in zip(chunks, embeddings):
        data.append({
            "chunk_id": chunk["id"],
            "document_id": document_id,
            "tenant_id": tenant_id or "default",
            "content": chunk["content"],
            "embedding": embedding,  # 1024维向量
            "metadata": chunk.get("metadata", {}),
        })

    # 2. 批量插入Milvus
    await self.vector_store_client.insert_batch(data)
    # collection: documents
    # 索引: HNSW (M=16, efConstruction=200)
    # 分区: 按tenant_id

sequenceDiagram
    autonumber
    participant KS as Knowledge Service
    participant Kafka
    participant Consumer as Kafka Consumer
    participant Indexer as IncrementalIndexer
    participant VersionMgr as VersionManager
    participant Redis
    participant Processor as DocumentProcessor
    participant MinIO
    participant Parser as PDFParser
    participant Chunker
    participant Embedder as BGE-M3
    participant VSAdapter as vector-store-adapter
    participant Milvus
    participant Neo4j

    rect rgb(225, 245, 254)
        Note over KS,Kafka: 步骤1-3: 文档上传与事件发送
        KS->>MinIO: 上传文档
        MinIO-->>KS: file_path
        KS->>Kafka: 发送 document.uploaded 事件
    end

    rect rgb(255, 243, 224)
        Note over Kafka,Indexer: 步骤4-6: Kafka 消费与分发
        Kafka->>Consumer: 轮询消息
        Consumer->>Consumer: 解析 event_type 和 payload
        Consumer->>Indexer: 调用增量索引器
    end

    rect rgb(200, 230, 201)
        Note over Indexer,Redis: 步骤7-11: 增量索引变更检测
        Indexer->>Indexer: calculate_content_hash(content)
        Indexer->>VersionMgr: detect_changes(doc_id, hashes)
        VersionMgr->>Redis: GET doc_version:{tenant}:{doc_id}
        Redis-->>VersionMgr: current_version
        VersionMgr-->>Indexer: VersionChange(needs_reindex)
    end

    alt 文档未变更（跳过索引）
        rect rgb(200, 230, 201)
            Note over Indexer,Consumer: 快速返回，节省 99% 时间
            Indexer-->>Consumer: {status: "skipped", elapsed: 50ms}
            Consumer->>Kafka: 提交 offset
        end

    else 文档变更或新文档（执行索引）
        rect rgb(255, 224, 178)
            Note over Indexer,Processor: 步骤12-20: 完整文档处理流程
            Indexer->>Processor: process_document(doc_id, ...)

            Processor->>MinIO: download_file(file_path)
            MinIO-->>Processor: file_data (bytes)

            Processor->>Parser: parse(file_data)
            Parser->>Parser: pdfplumber → PyPDF2 降级
            Parser-->>Processor: text (纯文本)

            Processor->>Chunker: chunk(text, doc_id)
            Chunker->>Chunker: RecursiveTextSplitter<br/>chunk_size=500, overlap=50
            Chunker-->>Processor: chunks (List[Dict])

            Processor->>Embedder: embed_batch(texts)
            Embedder->>Embedder: SentenceTransformer.encode()<br/>batch_size=32
            Embedder-->>Processor: embeddings (1024-dim)
        end

        rect rgb(255, 249, 196)
            Note over Processor,Milvus: 步骤21-26: 向量存储（适配器模式）
            Processor->>VSAdapter: VectorStoreClient.insert_batch(data)
            VSAdapter->>VSAdapter: POST /collections/{name}/insert
            VSAdapter->>VSAdapter: 选择后端: Milvus
            VSAdapter->>Milvus: collection.insert(entities)
            Milvus-->>VSAdapter: insert_result
            VSAdapter-->>Processor: HTTP 200 + success
        end

        par 步骤27-30: 异步构建知识图谱（不阻塞）
            Processor->>Processor: asyncio.create_task()
            Processor->>Processor: GraphBuilder.extract(text)
            Processor->>Neo4j: batch_create_nodes(entities)
            Neo4j-->>Processor: success
        end

        rect rgb(200, 230, 201)
            Note over Indexer,Redis: 步骤31-33: 保存新版本
            Processor-->>Indexer: {status, chunks_count, duration}
            Indexer->>VersionMgr: save_version(doc_id, hashes, chunk_count)
            VersionMgr->>Redis: SET doc_version + LPUSH history
            Redis-->>VersionMgr: ok
        end

        Indexer-->>Consumer: {status: "indexed", elapsed: 12s}
        Consumer->>Kafka: 提交 offset
        Consumer->>Kafka: 发送 document.indexed 事件（可选）
    end

    Note over KS,Neo4j: 性能对比：<br/>未变更文档: 50ms（跳过）<br/>新文档/变更: 10-15秒（完整索引）<br/>增量索引节省 60-80% 成本

sequenceDiagram
    autonumber
    participant KS as Knowledge Service
    participant Kafka
    participant Consumer as Kafka Consumer
    participant Processor as DocumentProcessor
    participant MinIO
    participant Parser as PDFParser
    participant Chunker as DocumentChunker
    participant Embedder as BGE-M3 Embedder
    participant Milvus
    participant GraphBuilder
    participant Neo4j
    participant DB as PostgreSQL

    KS->>MinIO: 上传文档
    MinIO-->>KS: file_path

    KS->>Kafka: 发送document.created事件
    Note over Kafka: topic: document.events<br/>partition: tenant_id哈希

    Kafka->>Consumer: 消费事件
    Consumer->>Processor: process_document(doc_id, tenant_id, file_path)

    Processor->>MinIO: download_file(file_path)
    MinIO-->>Processor: file_data (bytes)

    Processor->>Parser: parse(file_data)
    Parser->>Parser: pdfplumber.open(BytesIO)
    Parser->>Parser: extract_text()逐页提取
    Parser-->>Processor: text (纯文本)

    Processor->>Chunker: chunk(text, doc_id)
    Chunker->>Chunker: RecursiveCharacterTextSplitter<br/>chunk_size=500, overlap=50
    Chunker->>Chunker: generate_chunk_id()<br/>estimate_tokens()
    Chunker-->>Processor: chunks (List[Dict])

    Processor->>Embedder: embed_batch(texts)
    Embedder->>Embedder: SentenceTransformer.encode()<br/>batch_size=32, normalize=True
    Embedder-->>Processor: embeddings (List[1024-dim vectors])

    Processor->>Milvus: insert_batch(chunks + embeddings)
    Note over Milvus: collection: documents<br/>index: HNSW<br/>partition: tenant_id
    Milvus-->>Processor: success

    par 异步构建图谱（不阻塞）
        Processor->>GraphBuilder: extract(text)
        GraphBuilder->>GraphBuilder: NER实体识别
        GraphBuilder->>GraphBuilder: 关系抽取
        GraphBuilder-->>Processor: entities + relationships

        Processor->>GraphBuilder: build(entities, rels, doc_id)
        GraphBuilder->>Neo4j: CREATE (Entity)节点
        GraphBuilder->>Neo4j: CREATE ()-[RELATED_TO]->()边
        Neo4j-->>GraphBuilder: success
    end

    Processor->>DB: UPDATE documents SET status='COMPLETED'
    DB-->>Processor: ok

    Processor-->>Consumer: ProcessingResult(success, chunks_count, duration)
    Consumer->>Kafka: 发送document.indexed事件（可选）

flowchart TB
    subgraph API["API 层"]
        CheckAPI["/incremental/check<br/>检查是否需要索引"]
        ReindexAPI["/incremental/reindex<br/>强制重新索引"]
    end

    subgraph IncrementalModule["增量索引模块"]
        IncrementalIndexer["IncrementalIndexer<br/>增量索引器"]
        VersionManager["VersionManager<br/>版本管理器"]
    end

    subgraph Storage["存储层"]
        Redis["Redis<br/>版本信息存储"]
        VectorStore["Milvus<br/>向量数据"]
    end

    subgraph Processing["处理层"]
        DocProcessor["DocumentProcessor<br/>文档处理器"]
    end

    CheckAPI --> IncrementalIndexer
    ReindexAPI --> IncrementalIndexer

    IncrementalIndexer --> VersionManager
    IncrementalIndexer --> DocProcessor

    VersionManager --> Redis
    DocProcessor --> VectorStore

    style API fill:#e3f2fd
    style IncrementalModule fill:#fff3e0
    style Storage fill:#e8f5e9
    style Processing fill:#f3e5f5

class VersionManager:
    def calculate_content_hash(self, content: str) -> str:
        """计算内容哈希（SHA256）"""
        return hashlib.sha256(content.encode('utf-8')).hexdigest()

    async def detect_changes(
        self,
        document_id: str,
        new_content_hash: str,
        new_metadata_hash: str,
        tenant_id: Optional[str] = None
    ) -> VersionChange:
        """检测文档变更"""
        current_version = await self.get_current_version(document_id, tenant_id)

        if not current_version:
            return VersionChange(
                change_type="created",
                needs_reindex=True
            )

        content_changed = new_content_hash != current_version.content_hash
        metadata_changed = new_metadata_hash != current_version.metadata_hash

        if not content_changed and not metadata_changed:
            return VersionChange(
                change_type="unchanged",
                needs_reindex=False
            )

        return VersionChange(
            change_type="updated",
            content_changed=content_changed,
            metadata_changed=metadata_changed,
            needs_reindex=content_changed  # 仅内容变更需要重新索引
        )