Graph-RAG Moduler `graph_rag` — Enterprise API Reference

┌────────────────────────────────────────────────────────────────┐
│                          GraphRAG                               │
│                                                                  │
│  ┌─────────────────────┐      ┌──────────────────────────────┐  │
│  │    VectorDatabase   │      │      KnowledgeGraph           │  │
│  │  (FAISS + embedder) │      │  nodes · edges · adjacency    │  │
│  └────────┬────────────┘      └──────────────┬───────────────┘  │
│           │ vector similarity search          │ BFS expansion    │
│           │                                  │                  │
│           └─────────────────┬────────────────┘                  │
│                             │                                    │
│                     _expand_graph_context                        │
│                             │                                    │
│                     _build_results → LLM prompt → answer        │
└────────────────────────────────────────────────────────────────┘

add_document_with_relations()
    │
    ├── GraphNode per entity   →  KnowledgeGraph.add_node()
    ├── DocumentChunk per entity  →  VectorDatabase.add()
    └── GraphEdge per relation  →  KnowledgeGraph.add_edge()

retrieve_with_context(query)
    │
    ├── VectorDatabase.search()       → seed (chunk, score) pairs
    ├── _chunk_to_node mapping        → seed node scores
    ├── _expand_graph_context()       → BFS neighbour expansion
    └── _build_results()              → ranked result dicts

query(query)
    │
    ├── retrieve_with_context()
    ├── _build_graph_context()        → rich LLM context string
    ├── _build_prompt()               → language-aware prompt
    └── llm.generate(prompt)          → final answer string

from fennec_community.rag.types.graph_rag import GraphRAG, GraphNode, GraphEdge, KnowledgeGraph, ConfigGraphRAG

# 1. Configure
config = ConfigGraphRAG(k=5, context_depth=2, embedder_name="all-MiniLM-L6-v2")

# 2. Initialise (requires a VectorDatabase instance)
rag = GraphRAG(vector_db=my_vector_db, llm=my_llm, config=config)

# 3. Ingest a document
rag.add_document_with_relations(
    content="Paris is the capital of France. The Eiffel Tower is in Paris.",
    entities=[
        {"id": "Paris",        "type": "city",     "text": "Paris"},
        {"id": "France",       "type": "country",  "text": "France"},
        {"id": "EiffelTower",  "type": "landmark", "text": "Eiffel Tower"},
    ],
    relations=[
        {"source": "Paris",       "target": "France",      "type": "capital_of"},
        {"source": "EiffelTower", "target": "Paris",       "type": "located_in"},
    ],
    doc_id="doc_001",
)

# 4. Query
answer = rag.query("Where is the Eiffel Tower?", language="en")
print(answer)

# 5. Async usage
answer = await rag.aquery("Where is the Eiffel Tower?", language="en")

from fennec_community.rag.types.graph_rag import ConfigGraphRAG

ConfigGraphRAG(
    # ── Cache
    l1_size:              int   = 50,
    l2_size:              int   = 50,
    l3_size:              int   = 50,
    cache_ttl:            int   = 300,

    # ── Retrieval
    k:                    int   = 5,
    context_depth:        int   = 2,
    max_depth:            int   = 2,

    # ── Embedding
    embedder_name:        str   = "paraphrase-multilingual-MiniLM-L12-v2",
    embedding_dim:        Optional[int] = None,
    batch_size:           int   = 32,
    normalize_embeddings: bool  = True,

    # ── FAISS
    use_gpu:              bool  = False,
    faiss_nlist:          Optional[int] = None,
    faiss_nprobe:         int   = 10,
    rebuild_threshold:    int   = 100,

    # ── Search
    enable_hybrid_search: bool  = False,
    hybrid_alpha:         float = 0.5,

    # ── Performance
    max_cache_embeddings: int   = 1000,
    parallel_processing:  bool  = True,

    # ── Logging
    log_level:            str   = "INFO",
)

config = ConfigGraphRAG(
    k=10,
    context_depth=3,
    embedder_name="all-MiniLM-L6-v2",
    use_gpu=True,
    enable_hybrid_search=True,
    hybrid_alpha=0.7,
    log_level="DEBUG",
)

@classmethod
ConfigGraphRAG.from_env() -> ConfigGraphRAG

export GRAPHRAG_K=10
export GRAPHRAG_EMBEDDER=all-mpnet-base-v2
export GRAPHRAG_USE_GPU=true

config = ConfigGraphRAG.from_env()
rag = GraphRAG(vector_db=vdb, config=config)

config.to_dict() -> dict

import json
config = ConfigGraphRAG(k=8)
print(json.dumps(config.to_dict(), indent=2))
# {
#   "l1_size": 50,
#   "k": 8,
#   "embedder_name": "paraphrase-multilingual-MiniLM-L12-v2",
#   ...
# }

from fennec_community.rag.types.graph_rag import GraphNode

GraphNode(
    id:        str,
    content:   str,
    node_type: str,
    metadata:  Dict[str, Any] = field(default_factory=dict),
    embedding: Optional[List[float]] = None,
)

node = GraphNode(
    id="Paris",
    content="Paris is the capital and most populous city of France.",
    node_type="city",
    metadata={"country": "France", "population": 2_161_000},
)

node.has_embedding -> bool

if not node.has_embedding:
    print(f"Node '{node.id}' needs to be embedded before indexing.")

node.get_content_hash() -> str

old_hash = node.get_content_hash()
node.content = "Paris is the capital of France and the largest city."
new_hash = node.get_content_hash()

if old_hash != new_hash:
    print("Content changed — re-embed and re-index this node.")

node.to_dict(include_embedding: bool = False) -> Dict[str, Any]

# Lightweight (no embedding)
data = node.to_dict()
# {"id": "Paris", "content": "...", "node_type": "city", "metadata": {...}}

# Full (with embedding for export)
data = node.to_dict(include_embedding=True)

@classmethod
GraphNode.from_dict(data: Dict[str, Any]) -> GraphNode

import json

with open("nodes.json") as f:
    raw_nodes = json.load(f)

nodes = [GraphNode.from_dict(d) for d in raw_nodes]

node.update_metadata(key: str, value: Any) -> None

node.update_metadata("verified", True)
node.update_metadata("last_updated", "2026-05-10")

node.merge_metadata(metadata: Dict[str, Any]) -> None

node.merge_metadata({
    "source": "Wikipedia",
    "confidence": 0.95,
    "language": "en",
})

from fennec_community.rag.types.graph_rag import GraphEdge

GraphEdge(
    source:        str,
    target:        str,
    relation:      str,
    weight:        float = 1.0,
    metadata:      Dict[str, Any] = field(default_factory=dict),
    bidirectional: bool = False,
)

edge = GraphEdge(
    source="EiffelTower",
    target="Paris",
    relation="located_in",
    weight=1.0,
    bidirectional=False,
)

edge.edge_id -> str

edge = GraphEdge("Paris", "France", "capital_of")
print(edge.edge_id)  # "Paris-capital_of->France"

edge.reverse() -> GraphEdge

forward = GraphEdge("Paris", "France", "capital_of", weight=1.0)
backward = forward.reverse()
print(backward)
# GraphEdge(France -> Paris [inverse_capital_of] w=1.0)

edge.to_dict() -> Dict[str, Any]

data = edge.to_dict()
# {"source": "Paris", "target": "France", "relation": "capital_of",
#  "weight": 1.0, "metadata": {}, "bidirectional": False}

@classmethod
GraphEdge.from_dict(data: Dict[str, Any]) -> GraphEdge

import json

with open("edges.json") as f:
    raw_edges = json.load(f)

edges = [GraphEdge.from_dict(d) for d in raw_edges]

edge.update_weight(new_weight: float) -> None

edge.update_weight(2.5)  # Strengthen this relationship
edge.update_weight(0.1)  # Weaken this relationship

from fennec_community.rag.types.graph_rag import KnowledgeGraph

KnowledgeGraph(config: Optional[ConfigGraphRAG] = None)

kg = KnowledgeGraph()

# Or with custom config:
config = ConfigGraphRAG(max_depth=3)
kg = KnowledgeGraph(config=config)

kg.add_node(node: GraphNode) -> bool

node = GraphNode(id="Paris", content="Capital of France", node_type="city")
is_new = kg.add_node(node)
print("New node?" , is_new)  # True
kg.add_node(node)             # Returns False (update)

kg.remove_node(node_id: str) -> bool

removed = kg.remove_node("Paris")
# All edges sourced from or targeting "Paris" are also deleted

kg.get_node(node_id: str) -> Optional[GraphNode]

node = kg.get_node("Paris")
if node:
    print(node.content)

kg.add_edge(edge: GraphEdge) -> bool

edge = GraphEdge(source="EiffelTower", target="Paris", relation="located_in")
success = kg.add_edge(edge)
# Returns False if "EiffelTower" or "Paris" nodes don't exist yet

kg.get_edge(source: str, target: str, relation: str) -> Optional[GraphEdge]

edge = kg.get_edge("EiffelTower", "Paris", "located_in")
if edge:
    print(f"Weight: {edge.weight}")

kg.get_neighbors(
    node_id:          str,
    max_depth:        int  = None,
    include_incoming: bool = True,
) -> Set[str]

# Get all nodes within 2 hops of "Paris" (outgoing and incoming)
neighbors = kg.get_neighbors("Paris", max_depth=2, include_incoming=True)
print(neighbors)  # {"France", "EiffelTower", "EU", ...}

# Direct outgoing neighbours only
direct = kg.get_neighbors("Paris", max_depth=1, include_incoming=False)

kg.find_path(
    start:      str,
    end:        str,
    max_length: Optional[int] = None,
) -> Optional[List[str]]

path = kg.find_path("EiffelTower", "France")
if path:
    print(" → ".join(path))
    # EiffelTower → Paris → France
else:
    print("No path found")

kg.find_all_paths(
    start:      str,
    end:        str,
    max_length: int = 5,
) -> List[List[str]]

paths = kg.find_all_paths("EiffelTower", "Europe", max_length=4)
for path in paths:
    print(" → ".join(path))
# EiffelTower → Paris → France → Europe
# EiffelTower → Paris → EU → Europe

kg.get_subgraph(node_ids: Set[str]) -> KnowledgeGraph

europe_nodes = {"Paris", "France", "Berlin", "Germany"}
europe_graph = kg.get_subgraph(europe_nodes)
print(europe_graph)
# KnowledgeGraph(nodes=4, edges=3)

kg.get_node_degree(node_id: str) -> Dict[str, int]

degree = kg.get_node_degree("Paris")
# {"in_degree": 2, "out_degree": 1, "total_degree": 3}

if degree["total_degree"] == 0:
    print("Isolated node — no connections.")

kg.get_connected_components() -> List[Set[str]]

components = kg.get_connected_components()
print(f"Graph has {len(components)} connected components")
for i, comp in enumerate(components):
    print(f"  Component {i+1}: {comp}")

kg.get_stats() -> Dict[str, Any]

stats = kg.get_stats()
print(f"Nodes: {stats['num_nodes']}  Edges: {stats['num_edges']}")
print(f"Density: {stats['density']:.4f}  Components: {stats['num_components']}")

kg.validate_integrity() -> Dict[str, List[str]]

issues = kg.validate_integrity()

if any(issues.values()):
    print("⚠️ Integrity issues found:")
    for issue_type, items in issues.items():
        if items:
            print(f"  {issue_type}: {items}")
else:
    print("✅ Graph integrity OK")

from fennec_community.rag.types.graph_rag import GraphRAG

GraphRAG(
    vector_db: Any,
    llm:       Optional[Any] = None,
    config:    Optional[ConfigGraphRAG] = None,
)

from fennec_community.rag.types.graph_rag import GraphRAG, ConfigGraphRAG

config = ConfigGraphRAG(k=10, context_depth=2, embedder_name="all-MiniLM-L6-v2")
rag = GraphRAG(vector_db=my_vdb, llm=my_llm, config=config)
print(rag)
# GraphRAG(nodes=0, edges=0, llm=MyLLM)

rag.add_document_with_relations(
    content:   str,
    entities:  List[Dict[str, Any]],
    relations: List[Dict[str, Any]],
    doc_id:    str,
    metadata:  Optional[Dict] = None,
) -> Dict[str, Any]

result = rag.add_document_with_relations(
    content="""
        The Eiffel Tower is a wrought-iron lattice tower on the Champ de Mars in Paris, France.
        It was named after the engineer Gustave Eiffel.
    """,
    entities=[
        {"id": "EiffelTower", "text": "Eiffel Tower",  "type": "landmark"},
        {"id": "Paris",       "text": "Paris",          "type": "city"},
        {"id": "France",      "text": "France",         "type": "country"},
        {"id": "GEiffel",     "text": "Gustave Eiffel", "type": "person"},
    ],
    relations=[
        {"source": "EiffelTower", "target": "Paris",    "type": "located_in"},
        {"source": "Paris",       "target": "France",   "type": "capital_of"},
        {"source": "EiffelTower", "target": "GEiffel",  "type": "named_after"},
    ],
    doc_id="doc_eiffel_001",
    metadata={"source": "Wikipedia", "language": "en"},
)

print(result)
# {"nodes_added": 4, "edges_added": 3, "chunks_created": 4}

rag.retrieve_with_context(
    query:          str,
    k:              int   = None,
    context_depth:  int   = None,
    min_similarity: float = 0.0,
    combine_scores: bool  = True,
) -> List[Dict[str, Any]]

results = rag.retrieve_with_context(
    query="Where is the Eiffel Tower?",
    k=5,
    context_depth=2,
    min_similarity=0.3,
)

for r in results:
    print(f"[{r['type']}] {r['id']} — score: {r['score']:.3f}")
    print(f"  Content: {r['content']}")
    print(f"  Neighbors: {[n['id'] for n in r['neighbors']]}")

rag.semantic_search(
    query:              str,
    top_k:              int  = 5,
    include_graph_info: bool = True,
) -> List[Dict[str, Any]]

results = rag.semantic_search("Eiffel Tower engineer", top_k=3)
for r in results:
    print(f"Score {r['score']:.3f}: {r['text']}")
    if "graph_node" in r:
        print(f"  → Linked node: {r['graph_node']['id']} ({r['graph_node']['type']})")
        print(f"  → Neighbours: {r['graph_node']['neighbors_count']}")

rag.get_node_context(
    node_id:        str,
    max_depth:      int  = 2,
    include_chunks: bool = True,
) -> Dict[str, Any]

{
    "node": {
        "id":       str,
        "content":  str,
        "type":     str,
        "metadata": dict,
    },
    "neighbors_by_depth": {
        1: [{"id": str, "content": str, "type": str}, ...],
        2: [{"id": str, "content": str, "type": str}, ...],
        # ... up to max_depth
    },
    "related_chunks": [           # only when include_chunks=True
        {"text": str, "metadata": dict},
        ...
    ],
}

ctx = rag.get_node_context("Paris", max_depth=2, include_chunks=True)

print("Node:", ctx["node"])
print("Direct neighbours:", ctx["neighbors_by_depth"][1])
print("2-hop neighbours:", ctx["neighbors_by_depth"][2])
print("Associated chunks:", len(ctx["related_chunks"]))

rag.query(
    query:           str,
    k:               int  = None,
    language:        str  = "ar",
    include_sources: bool = False,
    **llm_kwargs,
) -> str

# Arabic (default)
answer = rag.query("أين يقع برج إيفل؟", language="ar")
print(answer)

# English with sources
answer = rag.query(
    "Where is the Eiffel Tower?",
    k=5,
    language="en",
    include_sources=True,
    temperature=0.1,     # forwarded to llm.generate
)
print(answer)
# "The Eiffel Tower is located in Paris, France..."
# 📚 Source :
# • [landmark] EiffelTower (score: 0.92)
# • [city] Paris (score: 0.87)

rag.generate(query: str, k: int = None, **kwargs) -> str

# Used by FederatedRAG when this GraphRAG is registered as a source:
answer = rag.generate("What is the Eiffel Tower?")

# Equivalent to:
answer = rag.query("What is the Eiffel Tower?")

rag.get_statistics() -> Dict[str, Any]

stats = rag.get_statistics()
print("Graph:", stats["graph"])
print("Vector DB:", stats["vector_db"])
print("Mappings:", stats["mappings"])

# {"graph": {"num_nodes": 150, "num_edges": 320, "density": 0.014, ...},
#  "vector_db": {"num_vectors": 150, "dim": 384, ...},
#  "mappings": {"chunks_to_nodes": 150, "nodes_with_chunks": 150}}

rag.save(path: str) -> None

<path>/
├── vector_db/          ← VectorDatabase.save() output
├── graph.json          ← nodes[] and edges[] arrays
└── mappings.json       ← chunk_to_node, node_to_chunks, doc_contents

rag.save("./saved_graphrag/v1")

rag.load(path: str) -> None

rag2 = GraphRAG(vector_db=new_vdb, llm=my_llm)
rag2.load("./saved_graphrag/v1")
answer = rag2.query("Where is the Eiffel Tower?", language="en")

async def aquery(query: str, k: int = None, **kwargs) -> str

import asyncio

async def main():
    answer = await rag.aquery(
        "Where is the Eiffel Tower?",
        k=5,
        language="en",
        include_sources=True,
    )
    print(answer)

asyncio.run(main())

async def agenerate(query: str, k: int = None, **kwargs) -> str

# Called automatically by FederatedRAG when GraphRAG is a registered source
answer = await rag.agenerate("What is the Eiffel Tower?")

async def aretrieve(
    query:  str,
    top_k:  int = None,
    k:      int = None,
    **kwargs,
) -> List[Dict[str, Any]]

results = await rag.aretrieve("Paris landmarks", top_k=5)
for r in results:
    print(r["id"], r["score"])

neighbor_score = parent_score × 0.5^(current_depth + 1)

<save_path>/
├── vector_db/
│   └── (VectorDatabase internal files — FAISS index, chunk list, etc.)
├── graph.json
│   ├── nodes: [{"id": ..., "content": ..., "node_type": ..., "metadata": ...}, ...]
│   └── edges: [{"source": ..., "target": ..., "relation": ..., "weight": ..., "bidirectional": ...}, ...]
└── mappings.json
    ├── chunk_to_node:      {"entity_Paris": "Paris", ...}
    ├── node_to_chunks:     {"Paris": ["entity_Paris"], ...}
    ├── node_to_chunks_set: {"Paris": ["entity_Paris"], ...}
    └── doc_contents:       {"doc_001": "Full text of the document...", ...}

from fennec_community.rag.types.graph_rag import GraphRAG, ConfigGraphRAG

config = ConfigGraphRAG(k=5, context_depth=2, log_level="INFO")
rag = GraphRAG(vector_db=my_vdb, llm=my_llm, config=config)

# Ingest
rag.add_document_with_relations(
    content="Paris is the capital of France. The Eiffel Tower is a famous landmark in Paris.",
    entities=[
        {"id": "Paris",       "text": "Paris",        "type": "city"},
        {"id": "France",      "text": "France",        "type": "country"},
        {"id": "EiffelTower", "text": "Eiffel Tower",  "type": "landmark"},
    ],
    relations=[
        {"source": "Paris",       "target": "France",      "type": "capital_of"},
        {"source": "EiffelTower", "target": "Paris",       "type": "located_in"},
    ],
    doc_id="geo_001",
)

# Query in English
answer = rag.query("What landmarks are in Paris?", language="en", include_sources=True)
print(answer)

from fennec_community.rag.types.graph_rag import KnowledgeGraph, GraphNode, GraphEdge

kg = KnowledgeGraph()

# Add nodes
kg.add_node(GraphNode("A", "Node A content", "concept"))
kg.add_node(GraphNode("B", "Node B content", "concept"))
kg.add_node(GraphNode("C", "Node C content", "concept"))

# Add edges
kg.add_edge(GraphEdge("A", "B", "relates_to"))
kg.add_edge(GraphEdge("B", "C", "leads_to"))

# Traverse
print(kg.get_neighbors("A", max_depth=2))     # {"B", "C"}
print(kg.find_path("A", "C"))                  # ["A", "B", "C"]
print(kg.find_all_paths("A", "C", max_length=5))

# Analytics
print(kg.get_stats())
print(kg.get_node_degree("B"))  # {"in_degree": 1, "out_degree": 1, "total_degree": 2}
print(kg.validate_integrity())  # {"orphan_edges": [], "duplicate_edges": [], "self_loops": []}

from fastapi import FastAPI
from fennec_community.rag.types.graph_rag import GraphRAG, ConfigGraphRAG

app = FastAPI()
config = ConfigGraphRAG.from_env()
rag = GraphRAG(vector_db=my_vdb, llm=my_llm, config=config)

@app.get("/query")
async def answer_query(q: str, lang: str = "en"):
    answer = await rag.aquery(q, language=lang, include_sources=True)
    return {"answer": answer}

@app.get("/retrieve")
async def retrieve(q: str, top_k: int = 5):
    results = await rag.aretrieve(q, top_k=top_k)
    return {"results": results}

@app.get("/node/{node_id}")
async def node_context(node_id: str, depth: int = 2):
    ctx = rag.get_node_context(node_id, max_depth=depth)
    return ctx

from fennec_community.rag.types.graph_rag import GraphRAG, ConfigGraphRAG

# Build and populate
rag = GraphRAG(vector_db=my_vdb, llm=my_llm)
rag.add_document_with_relations(...)

# Save
rag.save("./my_graphrag_state")

# Restore in a new process
rag2 = GraphRAG(vector_db=fresh_vdb, llm=my_llm)
rag2.load("./my_graphrag_state")

answer = rag2.query("Where is the Eiffel Tower?", language="en")
print(answer)

from fennec_community.rag.types.graph_rag import GraphRAG
from fennec_community.rag.types.federated_rag import FederatedRAG, AggregationMethod

graph_rag_1 = GraphRAG(vector_db=vdb1, llm=my_llm)
graph_rag_2 = GraphRAG(vector_db=vdb2, llm=my_llm)

# Register both GraphRAG instances as federated sources
fed = FederatedRAG(aggregation_method=AggregationMethod.RANKING)
fed.add_source("knowledge_base_1", graph_rag_1, weight=2.0, timeout=10.0)
fed.add_source("knowledge_base_2", graph_rag_2, weight=1.0, timeout=8.0)

# FederatedRAG calls graph_rag.aretrieve() + graph_rag.agenerate() automatically
result = await fed.query_async("What is the Eiffel Tower?")
print(result["answer"])

from fennec_community.rag.types.graph_rag import GraphNode

# Create
node = GraphNode(
    id="GEiffel",
    content="Gustave Eiffel was a French civil engineer.",
    node_type="person",
    metadata={"nationality": "French", "born": 1832},
)

# Inspect
print(node.has_embedding)         # False
print(node.get_content_hash())    # MD5 hash string

# Serialise
data = node.to_dict(include_embedding=False)

# Deserialise
node2 = GraphNode.from_dict(data)
assert node == node2  # True (equality based on ID)

# Update
node.update_metadata("died", 1923)
node.merge_metadata({"awards": ["Legion of Honour"], "profession": "engineer"})