flowchart TB
    subgraph Client["客户端层"]
        WebApp[Web 应用]
        APIGW[API Gateway]
        OtherServices[其他微服务]
    end

    subgraph Middleware["中间件层（按执行顺序）"]
        CORS[1. CORS 中间件<br/>跨域请求处理]
        OTEL[2. OpenTelemetry<br/>分布式追踪]
        RateLimit[3. 限流中间件<br/>令牌桶算法<br/>Redis实现]
        Idempotency[4. 幂等性中间件<br/>响应缓存<br/>120秒TTL]
    end

    subgraph Router["路由层"]
        KGAPI[知识图谱 API<br/>/api/v1/kg/*<br/>提取/查询/统计]
        CommAPI[社区检测 API<br/>/api/v1/community/*<br/>Leiden/Louvain]
        DisambAPI[实体消歧 API<br/>/api/v1/knowledge/disambiguation/*<br/>相似度/合并]
        AdminAPI[管理 API<br/>/api/v1/admin/*<br/>清理/补偿]
        GraphRAGAPI[GraphRAG API<br/>/api/v1/graphrag/*<br/>图谱增强检索]
    end

    subgraph Service["服务层"]
        KGService[KnowledgeGraphService<br/>核心编排服务]
        EntityExtractor[EntityExtractor<br/>SpaCy NER<br/>+Fallback正则]
        RelationExtractor[RelationExtractor<br/>依存句法分析<br/>SVO三元组]
        CommunityService[CommunityDetectionService<br/>社区检测<br/>图投影管理]
        DisambService[EntityDisambiguationService<br/>实体消歧<br/>相似度计算]
        CleanupService[CleanupService<br/>定期清理<br/>孤立数据/过期缓存]
        GraphRAGService[GraphRAGService<br/>混合检索<br/>图谱增强]
    end

    subgraph Infrastructure["基础设施层"]
        Neo4jClient[Neo4j Client<br/>异步连接池<br/>Size: 50]
        KafkaProducer[Kafka Producer<br/>异步发送<br/>GZIP压缩]
        EventComp[EventCompensation<br/>失败重试<br/>指数退避]
        RedisClient[Redis Client<br/>限流/幂等/缓存<br/>连接池: 50]
    end

    subgraph Storage["存储层"]
        Neo4j[(Neo4j 5.16<br/>图数据库<br/>GDS插件)]
        Redis[(Redis<br/>缓存/队列<br/>分布式锁)]
        Kafka[(Kafka 3.x<br/>事件流<br/>Topic: knowledge.events)]
    end

    subgraph External["外部消费者"]
        RAGEngine[RAG Engine<br/>消费实体/关系事件]
        Analytics[Analytics Service<br/>消费社区事件]
        Notification[Notification Service<br/>消费图谱构建事件]
    end

    Client --> CORS
    CORS --> OTEL
    OTEL --> RateLimit
    RateLimit --> Idempotency
    Idempotency --> Router

    Router --> Service
    Service --> Infrastructure
    Infrastructure --> Storage

    KGService --> EntityExtractor
    KGService --> RelationExtractor
    KGService --> Neo4jClient
    KGService --> KafkaProducer

    KafkaProducer --> EventComp
    EventComp --> RedisClient
    CleanupService --> Neo4jClient
    CleanupService --> RedisClient
    CommunityService --> Neo4jClient
    DisambService --> Neo4jClient
    GraphRAGService --> Neo4jClient

    Kafka -.事件订阅.-> RAGEngine
    Kafka -.事件订阅.-> Analytics
    Kafka -.事件订阅.-> Notification

    style Client fill:#e1f5ff
    style Middleware fill:#fff3e0
    style Router fill:#f3e5f5
    style Service fill:#e8f5e9
    style Infrastructure fill:#fce4ec
    style Storage fill:#fff9c4
    style External fill:#f1f8e9

客户端请求 → 中间件（限流/幂等） → 路由 → 服务层 → Neo4j → 响应
延迟：P50=100ms, P95=200ms, P99=500ms

服务层 → Kafka Producer → Kafka Topic → 下游消费者
吞吐量：1000 events/s, 丢失率 < 0.01%

Kafka发送失败 → EventCompensation → Redis队列 → 定期重试（指数退避）
成功率：99.9%（3次重试后）

CleanupService（每24h） → Neo4j/Redis → 删除孤立数据
清理量：平均每次删除 < 0.1% 数据

Neo4j连接池: 50（可扩展至100）
Redis连接池: 50
单实例支持并发: 200 QPS

默认配置: 60 req/min, burst=10
可按租户/用户/IP分别限流
防护效果: 过载下P99延迟 < 1s（无限流时 > 10s）

重试策略: 1min, 2min, 4min（指数退避）
最大重试: 3次
最终一致性: > 99.9%

{
  "neo4j": "healthy",
  "redis": "healthy",
  "kafka": "healthy",
  "spacy_model": "loaded"
}

sequenceDiagram
    autonumber
    participant C as 客户端
    participant CORS as CORS中间件
    participant OTEL as OpenTelemetry
    participant Rate as 限流中间件
    participant Idem as 幂等性中间件
    participant R as 知识图谱路由<br/>/api/v1/kg/extract
    participant KG as KnowledgeGraphService
    participant EE as EntityExtractor<br/>(SpaCy NER)
    participant RE as RelationExtractor<br/>(依存句法)
    participant Neo4j as Neo4j Client<br/>(连接池)
    participant Kafka as Kafka Producer
    participant EventComp as EventCompensation
    participant Redis as Redis

    C->>CORS: POST /api/v1/kg/extract<br/>{"text": "Apple was founded...", "source": "wiki"}
    Note over CORS: 验证CORS<br/>< 1ms
    CORS->>OTEL: 通过
    Note over OTEL: 注入TraceID<br/>创建Span<br/>< 5ms
    OTEL->>Rate: 通过

    Rate->>Redis: Lua脚本：令牌桶检查
    Note over Redis: 检查剩余令牌<br/>< 10ms
    Redis-->>Rate: allowed=1
    Rate->>Idem: 通过

    Idem->>Redis: GET idempotency:{key}
    Note over Redis: 检查幂等键<br/>< 5ms
    alt 缓存命中（重复请求）
        Redis-->>Idem: 缓存响应
        Idem-->>C: 200 OK (缓存)
        Note over C: 重复请求直接返回<br/>成本降低 30%
    else 缓存未命中（新请求）
        Redis-->>Idem: null
        Idem->>R: 通过

        R->>KG: extract_and_store(text, source)
        Note over KG: 开始知识图谱构建<br/>预计耗时: 100-500ms

        par 实体提取
            KG->>EE: extract_entities(text)
            Note over EE: SpaCy NER模型<br/>识别实体类型<br/>PERSON, ORG, GPE等<br/>耗时: 50-100ms
            EE-->>KG: entities[4]<br/>[{text:"Apple", label:"ORG"},<br/>{text:"Steve Jobs", label:"PERSON"}, ...]
        end

        par 关系提取
            KG->>RE: extract_relations(text, entities)
            Note over RE: 依存句法分析<br/>提取SVO三元组<br/>耗时: 50-100ms
            RE-->>KG: relations[3]<br/>[{subject:"Apple", predicate:"founded",<br/>object:"Steve Jobs"}, ...]
        end

        loop 遍历实体 (4个实体)
            KG->>Neo4j: create_node(label, properties)
            Note over Neo4j: 创建节点<br/>MERGE/CREATE<br/>耗时: 10-20ms/节点
            Neo4j-->>KG: node_id="4:abc:123"

            KG->>Kafka: publish_entity_created(entity_id, tenant_id, entity_data)
            alt Kafka 发送成功
                Kafka-->>KG: 成功
                Note over Kafka: 发送至topic: knowledge.events<br/>吞吐量: 1000 events/s
            else Kafka 发送失败
                Kafka->>EventComp: record_failed_event(event_type, payload, error)
                EventComp->>Redis: RPUSH failed_events
                Note over Redis: 记录失败事件<br/>等待补偿重试<br/>重试策略: 1min, 2min, 4min
                Redis-->>EventComp: 记录成功
            end
        end

        loop 遍历关系 (3个关系)
            alt 主语和宾语节点都存在
                KG->>Neo4j: create_relationship(from_id, to_id, rel_type, props)
                Note over Neo4j: 创建关系边<br/>MATCH + CREATE<br/>耗时: 15-30ms/关系
                Neo4j-->>KG: success=true

                KG->>Kafka: publish_relation_created(relation_id, source_id, target_id, type)
            else 节点不存在
                Note over KG: 跳过关系创建<br/>避免孤立边
            end
        end

        KG->>Kafka: publish_graph_built(graph_id, entity_count, relation_count)
        Note over Kafka: 图谱构建完成事件<br/>下游服务（RAG Engine）订阅

        KG-->>R: {success: true, entities_extracted: 4,<br/>entities_stored: 4, relations_stored: 3}
        R-->>Idem: 200 OK

        Idem->>Redis: SETEX idempotency:{key} 120 <response>
        Note over Redis: 缓存响应120秒<br/>幂等性保证
        Idem-->>C: {"success": true, "entities_extracted": 4, ...}
    end

    Note over C,Redis: 总耗时分析：<br/>中间件: 20ms, 实体提取: 80ms, 关系提取: 80ms,<br/>Neo4j存储: 100ms, Kafka发送: 20ms<br/>总计: P95=200ms

# app/graph/entity_extractor.py:51-79
def extract_entities(self, text: str) -> List[Dict[str, Any]]:
    doc = self.nlp(text)  # SpaCy NER处理
    entities = []
    for ent in doc.ents:
        entities.append({
            "text": ent.text,
            "label": ent.label_,  # PERSON, ORG, GPE等
            "start": ent.start_char,
            "end": ent.end_char,
            "confidence": 1.0
        })
    return entities

# app/graph/neo4j_client.py:107-133
async def create_node(self, label: str, properties: Dict[str, Any]) -> Optional[str]:
    query = f"CREATE (n:{label} $props) RETURN elementId(n) as id"
    result = await self.execute_query(query, {"props": properties})
    return result[0].get("id") if result else None

# app/infrastructure/kafka_producer.py:75-121
async def publish_event(self, event_type: str, payload: Dict, metadata: Dict):
    event = {
        "event_id": str(uuid4()),
        "event_type": event_type,
        "timestamp": datetime.utcnow().isoformat(),
        "payload": payload,
        "metadata": metadata
    }
    try:
        await asyncio.to_thread(  # 异步发送，不阻塞主线程
            self.producer.produce,
            self.topic_knowledge_events,
            value=json.dumps(event).encode("utf-8"),
            callback=self._delivery_report
        )
    except Exception as e:
        # 记录到补偿服务，后续重试
        await self.compensation_service.record_failed_event(...)

UNWIND $entities AS entity
CREATE (n) SET n = entity.properties
RETURN elementId(n) AS id

CREATE INDEX entity_text IF NOT EXISTS FOR (n:Entity) ON (n.text);

客户端
  → CORS中间件
  → OpenTelemetry中间件
  → 限流中间件 (RateLimiterMiddleware)
  → 幂等性中间件 (IdempotencyMiddleware)
  → 路由处理器 (knowledge_graph.extract_and_store)
  → KnowledgeGraphService.extract_and_store()
    ├─→ EntityExtractor.extract_entities()
    │   └─→ SpaCy NER模型处理
    ├─→ RelationExtractor.extract_relations()
    │   └─→ SpaCy依存句法分析
    ├─→ Neo4jClient.create_node() (循环: N个实体)
    │   └─→ AsyncGraphDatabase.session().run()
    ├─→ KafkaProducer.publish_entity_created() (循环: N个实体)
    │   └─→ confluent_kafka.Producer.produce()
    │       └─→ [失败] EventCompensation.record_failed_event()
    ├─→ Neo4jClient.create_relationship() (循环: M个关系)
    └─→ KafkaProducer.publish_graph_built()
  → 幂等性中间件缓存响应
  → 返回客户端

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant Router as 路由层<br/>extract_and_store()
    participant KGService as KnowledgeGraphService
    participant EntityExt as EntityExtractor
    participant RelationExt as RelationExtractor
    participant Neo4j as Neo4jClient
    participant Kafka as KafkaProducer

    Client->>Router: POST /api/v1/kg/extract<br/>{text, source}
    Router->>KGService: extract_and_store(text, source)

    Note over KGService: 步骤1: 实体提取
    KGService->>EntityExt: extract_entities(text)
    activate EntityExt
    EntityExt->>EntityExt: self.nlp(text)
    Note over EntityExt: SpaCy NER处理<br/>识别PERSON, ORG, GPE等
    EntityExt-->>KGService: entities[]
    deactivate EntityExt

    Note over KGService: 步骤2: 关系提取
    KGService->>RelationExt: extract_relations(text, entities)
    activate RelationExt
    RelationExt->>RelationExt: 依存句法分析
    Note over RelationExt: 提取SVO三元组
    RelationExt-->>KGService: relations[]
    deactivate RelationExt

    Note over KGService: 步骤3: 存储实体
    loop 遍历每个实体
        KGService->>Neo4j: create_node(label, properties)
        activate Neo4j
        Neo4j->>Neo4j: CREATE (n:Label $props)
        Neo4j-->>KGService: node_id
        deactivate Neo4j

        KGService->>Kafka: publish_entity_created(entity_id, ...)
        Note over Kafka: 异步发送<br/>不阻塞主流程
    end

    Note over KGService: 步骤4: 存储关系
    loop 遍历每个关系
        KGService->>Neo4j: create_relationship(from_id, to_id, rel_type)
        Neo4j-->>KGService: success

        KGService->>Kafka: publish_relation_created(...)
    end

    KGService->>Kafka: publish_graph_built(...)
    KGService-->>Router: {success, entities_extracted, ...}
    Router-->>Client: 200 OK

# app/routers/knowledge_graph.py:45-72
@router.post("/extract", summary="提取实体和关系并存储到图谱")
async def extract_and_store(request: ExtractRequest):
    """
    从文本提取实体和关系，并存储到知识图谱

    性能指标: P95=200ms, P99=500ms
    准确率: 实体提取 > 85%, 关系提取 > 70%
    """
    try:
        kg_service = get_kg_service()
        result = await kg_service.extract_and_store(
            text=request.text,
            source=request.source
        )

        if not result.get("success"):
            raise HTTPException(
                status_code=500,
                detail=result.get("error", "Extraction failed")
            )

        return result
    except Exception as e:
        logger.error(f"Extract and store failed: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail=str(e))

# app/graph/knowledge_graph_service.py:27-138
async def extract_and_store(self, text: str, source: Optional[str] = None):
    """
    核心编排逻辑

    关键优化点:
    1. 实体提取和关系提取可以并行（当前顺序执行）
    2. 实体存储可以批量化（当前逐个存储）
    3. Kafka发送异步化（已实现）
    """
    try:
        # 1. 提取实体（耗时: 50-100ms）
        entities = self.entity_extractor.extract_entities(text)
        logger.info(f"Extracted {len(entities)} entities")

        # 2. 提取关系（耗时: 50-100ms）
        relations = self.relation_extractor.extract_relations(text, entities)
        logger.info(f"Extracted {len(relations)} relations")

        # 3. 存储实体到Neo4j（耗时: 10-20ms/实体）
        entity_ids = {}
        for entity in entities:
            node_properties = {
                "text": entity["text"],
                "label": entity["label"],
                "confidence": entity.get("confidence", 1.0),
            }
            if source:
                node_properties["source"] = source

            node_id = await self.neo4j_client.create_node(
                label=entity["label"],
                properties=node_properties
            )

            if node_id:
                entity_ids[entity["text"]] = node_id

                # 发布实体创建事件（异步，不阻塞）
                if self.kafka_producer:
                    try:
                        await self.kafka_producer.publish_entity_created(
                            entity_id=node_id,
                            tenant_id=node_properties.get("tenant_id", "default"),
                            entity_data={
                                "name": entity["text"],
                                "type": entity["label"],
                                "description": "",
                            }
                        )
                    except Exception as e:
                        logger.warning(f"Failed to publish entity event: {e}")

        # 4. 存储关系（耗时: 15-30ms/关系）
        stored_relations = 0
        for relation in relations:
            subject_id = entity_ids.get(relation["subject"])
            object_id = entity_ids.get(relation["object"])

            if subject_id and object_id:
                rel_type = relation["predicate"].upper().replace(" ", "_")
                success = await self.neo4j_client.create_relationship(
                    from_id=subject_id,
                    to_id=object_id,
                    rel_type=rel_type,
                    properties={"confidence": relation.get("confidence", 0.8)},
                )
                if success:
                    stored_relations += 1

                    # 发布关系创建事件
                    if self.kafka_producer:
                        try:
                            await self.kafka_producer.publish_relation_created(
                                relation_id=f"{subject_id}_{rel_type}_{object_id}",
                                tenant_id="default",
                                source_id=subject_id,
                                target_id=object_id,
                                relation_type=rel_type,
                            )
                        except Exception as e:
                            logger.warning(f"Failed to publish relation event: {e}")

        result = {
            "success": True,
            "entities_extracted": len(entities),
            "entities_stored": len(entity_ids),
            "relations_extracted": len(relations),
            "relations_stored": stored_relations,
        }

        # 发布图谱构建完成事件
        if self.kafka_producer and (len(entity_ids) > 0 or stored_relations > 0):
            try:
                await self.kafka_producer.publish_graph_built(
                    graph_id=f"graph_{hash(text[:100])}",
                    tenant_id="default",
                    entity_count=len(entity_ids),
                    relation_count=stored_relations,
                    metadata={"source": source or "unknown"}
                )
            except Exception as e:
                logger.warning(f"Failed to publish graph built event: {e}")

        return result

    except Exception as e:
        logger.error(f"Extract and store failed: {e}", exc_info=True)
        return {"success": False, "error": str(e)}

客户端
  → [中间件层]
  → 路由处理器 (knowledge_graph.query_entity)
  → KnowledgeGraphService.query_entity()
    ├─→ Neo4jClient.execute_query("MATCH (n) WHERE n.text = $text ...")
    │   └─→ 查询实体节点
    └─→ Neo4jClient.find_relationships(node_id)
        └─→ 查询所有关系
  → 返回客户端

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant Router as 路由层<br/>query_entity()
    participant KGService as KnowledgeGraphService
    participant Neo4j as Neo4jClient

    Client->>Router: POST /api/v1/kg/query/entity<br/>{entity: "Apple"}
    Router->>KGService: query_entity(entity_text="Apple")

    Note over KGService: 步骤1: 查询实体节点
    KGService->>Neo4j: execute_query(MATCH WHERE n.text=$text)
    activate Neo4j
    Neo4j->>Neo4j: 索引查找<br/>（如有索引: 20ms, 否则: 100ms）
    Neo4j-->>KGService: node {id, labels, properties}
    deactivate Neo4j

    alt 实体不存在
        KGService-->>Router: None
        Router-->>Client: 404 Not Found
    else 实体存在
        Note over KGService: 步骤2: 查询所有关系
        KGService->>Neo4j: find_relationships(node_id)
        activate Neo4j
        Neo4j->>Neo4j: MATCH (a)-[r]->(b)<br/>WHERE elementId(a)=$id
        Neo4j-->>KGService: relations[]
        deactivate Neo4j

        KGService-->>Router: {id, labels, properties, relations}
        Router-->>Client: 200 OK
    end

    Note over Client,Neo4j: 总耗时：<br/>有索引: 30-50ms<br/>无索引: 120-150ms

# app/graph/knowledge_graph_service.py:140-180
async def query_entity(self, entity_text: str) -> Optional[Dict[str, Any]]:
    """
    查询实体详细信息

    性能优化:
    1. 需要为text字段创建索引（查询延迟从100ms降至20ms）
    2. 关系数量较多时可分页返回
    """
    try:
        # 步骤1: 查询实体节点
        nodes = await self.neo4j_client.execute_query(
            """
            MATCH (n)
            WHERE n.text = $text
            RETURN n, elementId(n) as id, labels(n) as labels
            LIMIT 1
            """,
            {"text": entity_text},
        )

        if not nodes:
            return None

        node = nodes[0]
        node_id = node["id"]

        # 步骤2: 查询所有关系（性能关键点）
        relations = await self.neo4j_client.find_relationships(node_id)

        return {
            "id": node_id,
            "labels": node["labels"],
            "properties": dict(node["n"]),
            "relations": relations,  # 可能包含大量关系
        }
    except Exception as e:
        logger.error(f"Query entity failed: {e}")
        return None

客户端
  → [中间件层]
  → 路由处理器 (knowledge_graph.query_path)
  → KnowledgeGraphService.query_path()
    └─→ Neo4jClient.execute_query("MATCH path = (start)-[*1..depth]-(end)")
  → 返回客户端

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant Router as 路由层<br/>query_path()
    participant KGService as KnowledgeGraphService
    participant Neo4j as Neo4jClient

    Client->>Router: POST /api/v1/kg/query/path<br/>{start: "Apple", end: "Steve Jobs", max_depth: 3}
    Router->>KGService: query_path(start, end, max_depth=3)

    KGService->>Neo4j: execute_query(MATCH path=...)
    activate Neo4j
    Note over Neo4j: 图遍历算法<br/>深度优先搜索<br/>复杂度: O(b^d)
    Neo4j->>Neo4j: 查找所有路径<br/>最多返回10条
    Neo4j-->>KGService: paths[] (nodes, relations)
    deactivate Neo4j

    KGService-->>Router: {paths: [...], count: N}
    Router-->>Client: 200 OK

    Note over Client,Neo4j: 性能警告：<br/>max_depth > 3 时<br/>查询可能超过10s

# app/graph/knowledge_graph_service.py:182-220
async def query_path(self, start_entity: str, end_entity: str, max_depth: int = 3):
    """
    查询实体之间的路径

    性能警告:
    - max_depth <= 3: 通常 < 1s
    - max_depth = 4: 可能 5-10s
    - max_depth >= 5: 可能 > 30s (建议禁止)

    优化建议:
    1. 限制max_depth最大为3
    2. 使用Neo4j GDS的shortestPath算法（更快）
    3. 结果缓存（路径通常较稳定）
    """
    try:
        result = await self.neo4j_client.execute_query(
            f"""
            MATCH path = (start)-[*1..{max_depth}]-(end)
            WHERE start.text = $start AND end.text = $end
            RETURN [node in nodes(path) | {{text: node.text, labels: labels(node)}}] as nodes,
                   [rel in relationships(path) | type(rel)] as relations
            LIMIT 10
            """,
            {"start": start_entity, "end": end_entity},
        )

        paths = []
        for record in result:
            paths.append({
                "nodes": record["nodes"],
                "relations": record["relations"],
            })

        return paths
    except Exception as e:
        logger.error(f"Query path failed: {e}")
        return []

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant Router as 路由层
    participant KGService as KnowledgeGraphService
    participant Neo4j as Neo4jClient

    Client->>Router: GET /api/v1/kg/statistics
    Router->>KGService: get_statistics()

    par 并行查询
        KGService->>Neo4j: MATCH (n) RETURN count(n)
        Note over Neo4j: 节点总数查询<br/>10-20ms
        Neo4j-->>KGService: node_count
    and
        KGService->>Neo4j: MATCH ()-[r]->() RETURN count(r)
        Note over Neo4j: 关系总数查询<br/>10-20ms
        Neo4j-->>KGService: rel_count
    and
        KGService->>Neo4j: MATCH (n) RETURN labels(n)[0], count(*)
        Note over Neo4j: 标签统计查询<br/>20-50ms
        Neo4j-->>KGService: label_stats[]
    end

    KGService-->>Router: {total_nodes, total_relationships, label_statistics}
    Router-->>Client: 200 OK

    Note over Client,Neo4j: 总耗时: 50-100ms<br/>可缓存1小时，提升性能90%

algo/knowledge-service/
├── main.py                                # 应用入口
├── requirements.txt                       # 依赖清单
├── pytest.ini                            # 测试配置
├── Makefile                              # 自动化脚本
├── app/
│   ├── __init__.py
│   ├── core/                             # 核心配置
│   │   ├── config.py                     # 配置管理
│   │   ├── logging_config.py             # 日志配置
│   │   └── observability.py              # OpenTelemetry 集成
│   ├── graph/                            # 知识图谱核心
│   │   ├── entity_extractor.py           # 实体提取
│   │   ├── relation_extractor.py         # 关系提取
│   │   ├── knowledge_graph_service.py    # 核心服务
│   │   └── neo4j_client.py               # Neo4j 客户端
│   ├── infrastructure/                   # 基础设施
│   │   ├── kafka_producer.py             # Kafka 生产者
│   │   └── event_compensation.py         # 事件补偿
│   ├── middleware/                       # 中间件
│   │   ├── rate_limiter.py               # 限流中间件
│   │   └── idempotency.py                # 幂等性中间件
│   ├── routers/                          # API 路由
│   │   ├── knowledge_graph.py            # 知识图谱 API
│   │   ├── community.py                  # 社区检测 API
│   │   ├── disambiguation.py             # 实体消歧 API
│   │   └── admin.py                      # 管理 API
│   └── services/                         # 业务服务
│       ├── cleanup_service.py            # 清理服务
│       ├── community_detection_service.py # 社区检测
│       ├── entity_disambiguation_service.py # 实体消歧
│       ├── embedding_service.py          # 向量嵌入
│       ├── entity_extraction_service.py  # 实体提取服务
│       └── entity_similarity_merger.py   # 实体合并
└── tests/                                # 测试用例
    ├── test_neo4j_client.py
    ├── test_kafka_producer.py
    └── conftest.py

class Settings(BaseSettings):
    # 服务基本信息
    APP_NAME: str = "Knowledge Service"
    VERSION: str = "1.0.0"
    HOST: str = "0.0.0.0"
    PORT: int = 8006
    ENVIRONMENT: str = "development"

    # Neo4j 配置
    NEO4J_URI: str
    NEO4J_USER: str
    NEO4J_PASSWORD: str
    NEO4J_MAX_CONNECTION_LIFETIME: int = 3600
    NEO4J_MAX_CONNECTION_POOL_SIZE: int = 50
    NEO4J_CONNECTION_TIMEOUT: int = 30
    NEO4J_MAX_TRANSACTION_RETRY_TIME: int = 30

    # Redis 配置
    REDIS_URL: str
    REDIS_MAX_CONNECTIONS: int = 50
    REDIS_SOCKET_TIMEOUT: int = 5
    REDIS_SOCKET_CONNECT_TIMEOUT: int = 5

    # Kafka 配置
    KAFKA_BOOTSTRAP_SERVERS: str
    KAFKA_TOPIC_KNOWLEDGE_EVENTS: str = "knowledge.events"
    KAFKA_ACKS: str = "all"
    KAFKA_RETRIES: int = 3
    KAFKA_LINGER_MS: int = 5
    KAFKA_COMPRESSION_TYPE: str = "gzip"

    # SpaCy 模型
    SPACY_MODEL: str = "en_core_web_sm"

    # 限流配置
    RATE_LIMIT_ENABLED: bool = True
    RATE_LIMIT_REQUESTS_PER_MINUTE: int = 60
    RATE_LIMIT_BURST: int = 10

    # 清理任务配置
    CLEANUP_INTERVAL_HOURS: int = 24
    CLEANUP_ORPHAN_DAYS_THRESHOLD: int = 30
    CLEANUP_COMMUNITY_DAYS_THRESHOLD: int = 7

    # OpenTelemetry 配置
    OTEL_ENABLED: bool = True
    OTEL_EXPORTER_OTLP_ENDPOINT: str = "http://localhost:4317"
    OTEL_SERVICE_NAME: str = "knowledge-service"

# 配置加载流程
class Settings(BaseSettings):
    class Config:
        env_file = ".env"        # 从 .env 文件加载
        case_sensitive = True    # 环境变量区分大小写

# 全局配置实例
settings = Settings()

from app.core.config import settings

# 访问配置
neo4j_uri = settings.NEO4J_URI
rate_limit = settings.RATE_LIMIT_REQUESTS_PER_MINUTE

# 服务配置
ENVIRONMENT=development
LOG_LEVEL=DEBUG

# Neo4j (本地开发)
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=password

# Redis (本地开发)
REDIS_URL=redis://localhost:6379/0

# Kafka (可选，开发时可关闭)
KAFKA_BOOTSTRAP_SERVERS=localhost:9092

# 限流（开发时可放宽）
RATE_LIMIT_REQUESTS_PER_MINUTE=1000
RATE_LIMIT_BURST=100

# 服务配置
ENVIRONMENT=production
LOG_LEVEL=INFO

# Neo4j (集群)
NEO4J_URI=bolt+s://neo4j.prod.example.com:7687
NEO4J_USER=${NEO4J_USER_SECRET}
NEO4J_PASSWORD=${NEO4J_PASSWORD_SECRET}
NEO4J_MAX_CONNECTION_POOL_SIZE=100

# Redis (集群)
REDIS_URL=redis://redis.prod.example.com:6379/0
REDIS_MAX_CONNECTIONS=100

# Kafka (集群)
KAFKA_BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092
KAFKA_ACKS=all
KAFKA_COMPRESSION_TYPE=zstd

# 限流（严格）
RATE_LIMIT_REQUESTS_PER_MINUTE=60
RATE_LIMIT_BURST=10

# 清理（更频繁）
CLEANUP_INTERVAL_HOURS=12
CLEANUP_ORPHAN_DAYS_THRESHOLD=7

# OpenTelemetry (Jaeger)
OTEL_ENABLED=true
OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger-collector:4317

class EntityExtractor:
    def __init__(self, model: str = "en_core_web_sm"):
        # 初始化 SpaCy 模型
        self.model_name = model
        self.nlp = spacy.load(model)

    def extract_entities(self, text: str) -> List[Dict[str, Any]]:
        # 提取实体
        pass

    def _fallback_extract(self, text: str) -> List[Dict[str, Any]]:
        # 后备提取方法（基于正则）
        pass

    def extract_with_context(self, text: str, context_window: int = 50) -> List[Dict[str, Any]]:
        # 提取实体并包含上下文
        pass

    def group_by_label(self, entities: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]:
        # 按标签分组实体
        pass

def extract_entities(self, text: str) -> List[Dict[str, Any]]

[
    {
        "text": str,        # 实体文本
        "label": str,       # 实体类型（PERSON/ORG/GPE/DATE等）
        "start": int,       # 起始位置
        "end": int,         # 结束位置
        "confidence": float # 置信度（SpaCy 默认 1.0）
    },
    ...
]

def extract_entities(self, text: str) -> List[Dict[str, Any]]:
    if not self.nlp:
        return self._fallback_extract(text)

    try:
        # 使用 SpaCy 处理文本
        doc = self.nlp(text)

        # 遍历识别的实体
        entities = []
        for ent in doc.ents:
            entities.append({
                "text": ent.text,
                "label": ent.label_,
                "start": ent.start_char,
                "end": ent.end_char,
                "confidence": 1.0,  # SpaCy 不提供置信度
            })

        return entities
    except Exception as e:
        # 处理失败，降级到 fallback
        return self._fallback_extract(text)

def _fallback_extract(self, text: str) -> List[Dict[str, Any]]:
    entities = []

    # 提取大写开头的词组（可能是人名或地名）
    # 匹配模式：连续的大写字母开头的单词
    capitalized_pattern = r'\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\b'
    matches = re.finditer(capitalized_pattern, text)

    for match in matches:
        entities.append({
            "text": match.group(),
            "label": "UNKNOWN",  # 无法确定类型
            "start": match.start(),
            "end": match.end(),
            "confidence": 0.5,    # 低置信度
        })

    return entities

def extract_with_context(self, text: str, context_window: int = 50) -> List[Dict[str, Any]]:
    entities = self.extract_entities(text)

    # 为每个实体添加上下文
    for entity in entities:
        start = max(0, entity["start"] - context_window)
        end = min(len(text), entity["end"] + context_window)
        entity["context"] = text[start:end]

    return entities

"Apple was founded by Steve Jobs in 1976. The company is based in Cupertino."

[
    {
        "text": "Apple",
        "label": "ORG",
        "start": 0,
        "end": 5,
        "confidence": 1.0,
        "context": "Apple was founded by Steve Jobs in 1976"  # 前后各 50 字符
    },
    {
        "text": "Steve Jobs",
        "label": "PERSON",
        "start": 21,
        "end": 31,
        "confidence": 1.0,
        "context": "Apple was founded by Steve Jobs in 1976. The company"
    },
    ...
]

sequenceDiagram
    participant KG as KnowledgeGraphService
    participant EE as EntityExtractor
    participant Spacy as SpaCy NLP

    KG->>EE: extract_entities(text)

    alt SpaCy 模型可用
        EE->>Spacy: nlp(text)
        Spacy->>Spacy: 分词、词性标注、NER
        Spacy-->>EE: doc.ents[]
        EE->>EE: 构造实体列表
        EE-->>KG: entities[]
    else SpaCy 不可用或失败
        EE->>EE: _fallback_extract(text)
        EE->>EE: 正则匹配大写词组
        EE-->>KG: entities[] (低置信度)
    end

# 全局实例
_entity_extractor: Optional[EntityExtractor] = None

def get_entity_extractor() -> EntityExtractor:
    """获取实体提取器实例（单例）"""
    global _entity_extractor

    if _entity_extractor is None:
        _entity_extractor = EntityExtractor()

    return _entity_extractor

class RelationExtractor:
    def __init__(self, model: str = "en_core_web_sm"):
        self.model_name = model
        self.nlp = spacy.load(model)

    def extract_relations(self, text: str, entities: Optional[List[Dict[str, Any]]] = None) -> List[Dict[str, Any]]:
        # 提取关系
        pass

    def _expand_entity(self, token) -> str:
        # 扩展实体（包含修饰词）
        pass

    def _fallback_extract(self, text: str, entities: Optional[List[Dict[str, Any]]]) -> List[Dict[str, Any]]:
        # 后备提取方法（基于正则）
        pass

    def extract_triplets(self, text: str) -> List[Tuple[str, str, str]]:
        # 提取三元组
        pass

    def filter_by_confidence(self, relations: List[Dict[str, Any]], min_confidence: float = 0.5) -> List[Dict[str, Any]]:
        # 按置信度过滤
        pass

def extract_relations(self, text: str, entities: Optional[List[Dict[str, Any]]] = None) -> List[Dict[str, Any]]

[
    {
        "subject": str,      # 主语（头实体）
        "predicate": str,    # 谓词（关系类型）
        "object": str,       # 宾语（尾实体）
        "confidence": float  # 置信度
    },
    ...
]

def extract_relations(self, text: str, entities: Optional[List[Dict[str, Any]]] = None) -> List[Dict[str, Any]]:
    if not self.nlp:
        return self._fallback_extract(text, entities)

    try:
        doc = self.nlp(text)
        relations = []

        # 遍历每个句子
        for sent in doc.sents:
            # 遍历每个词
            for token in sent:
                # 查找动词作为关系谓词
                if token.pos_ == "VERB":
                    subject = None
                    obj = None

                    # 查找主语
                    for child in token.children:
                        if child.dep_ in ("nsubj", "nsubjpass"):
                            subject = child
                            break

                    # 查找宾语
                    for child in token.children:
                        if child.dep_ in ("dobj", "pobj", "attr"):
                            obj = child
                            break

                    # 如果找到主语和宾语，创建关系
                    if subject and obj:
                        relations.append({
                            "subject": self._expand_entity(subject),
                            "predicate": token.text,
                            "object": self._expand_entity(obj),
                            "confidence": 0.8,  # 基于规则的置信度
                        })

        return relations
    except Exception as e:
        return self._fallback_extract(text, entities)

def _expand_entity(self, token) -> str:
    # 获取 token 的所有子树节点
    subtree = list(token.subtree)

    # 按位置排序
    subtree.sort(key=lambda t: t.i)

    # 合并文本
    return " ".join(t.text for t in subtree)

输入句子: "The CEO of Apple founded the company in 1976."
原始 token: "CEO"
扩展结果: "The CEO of Apple"

原始 token: "company"
扩展结果: "the company"

def _fallback_extract(self, text: str, entities: Optional[List[Dict[str, Any]]] = None) -> List[Dict[str, Any]]:
    relations = []

    # 简单的 SVO 模式
    patterns = [
        r'(\w+(?:\s+\w+)*)\s+(is|are|was|were|has|have|owns|creates|manages)\s+(\w+(?:\s+\w+)*)',
        r'(\w+(?:\s+\w+)*)\s+(works\s+for|belongs\s+to|located\s+in)\s+(\w+(?:\s+\w+)*)',
    ]

    for pattern in patterns:
        matches = re.finditer(pattern, text, re.IGNORECASE)
        for match in matches:
            relations.append({
                "subject": match.group(1).strip(),
                "predicate": match.group(2).strip(),
                "object": match.group(3).strip(),
                "confidence": 0.5,  # 低置信度
            })

    return relations

def extract_triplets(self, text: str) -> List[Tuple[str, str, str]]:
    relations = self.extract_relations(text)

    triplets = [
        (rel["subject"], rel["predicate"], rel["object"])
        for rel in relations
    ]

    return triplets