ZerdaZerda

Graph RAG

On this page

[!WARNING] External dependency graph_rag.py imports from ....chunks import DocumentChunk — four package levels above this archive. This symbol is not defined anywhere in the uploaded files, so GraphRAG cannot be instantiated standalone without the parent "zerda" project supplying a chunks module. KnowledgeGraph, GraphNode, and GraphEdge have no such dependency and can be used independently.

Overview

GraphRAG combines a vector database with an in-memory KnowledgeGraph of entities (GraphNode) and relations (GraphEdge), expanding initial semantic search hits via BFS graph traversal to pull in connected context before generation.

ConfigGraphRAG

Python · 22 lines
@dataclass
class ConfigGraphRAG:
    l1_size: int = 50
    l2_size: int = 50
    l3_size: int = 50
    cache_ttl: int = 300
    k: int = 5
    context_depth: int = 2
    max_depth: int = 2
    embedder_name: str = "paraphrase-multilingual-MiniLM-L12-v2"
    embedding_dim: Optional[int] = None
    batch_size: int = 32
    normalize_embeddings: bool = True
    use_gpu: bool = False
    faiss_nlist: Optional[int] = None
    faiss_nprobe: int = 10
    rebuild_threshold: int = 100
    enable_hybrid_search: bool = False
    hybrid_alpha: float = 0.5
    max_cache_embeddings: int = 1000
    parallel_processing: bool = True
    log_level: str = "INFO"
FieldPurpose
l1_size/l2_size/l3_sizeMulti-level cache sizes.
kDefault number of top results to retrieve.
context_depthDepth of graph-context expansion around retrieved nodes.
max_depthMaximum BFS traversal depth for graph queries.
embedder_nameSentence-transformer model name.
embedding_dimAuto-detected if None.
batch_sizeBatch size for embedding generation.
use_gpuWhether FAISS operations use GPU.
faiss_nlist/faiss_nprobeIVF index tuning (auto-calculated nlist if None).
rebuild_thresholdRebuild the index after N additions.
cache_ttlCache time-to-live in seconds.
enable_hybrid_searchEnables combined semantic+keyword search.

from_env() (classmethod) builds a config from environment variables (GRAPHRAG_K, GRAPHRAG_EMBEDDER, GRAPHRAG_BATCH_SIZE, GRAPHRAG_USE_GPU, GRAPHRAG_LOG_LEVEL). to_dict() serializes the config.

GraphNode

Python · 7 lines
@dataclass
class GraphNode:
    id: str
    content: str
    node_type: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    embedding: Optional[List[float]] = None
MemberDescription
has_embedding (property)Whether an embedding vector is set.
get_content_hash()MD5 hash of content, for change detection.
to_dict(include_embedding=False) / from_dict(data)Serialization round-trip.
update_metadata(key, value) / merge_metadata(metadata)Runtime metadata edits.
Equality/hashBased solely on id.

GraphEdge

Python · 8 lines
@dataclass
class GraphEdge:
    source: str
    target: str
    relation: str
    weight: float = 1.0
    metadata: Dict[str, Any] = field(default_factory=dict)
    bidirectional: bool = False

Represents (source) --[relation]--> (target), e.g. (Paris) --[capital_of]--> (France).

MemberDescription
edge_id (property)Unique identifier generated from source/target/relation.
reverse()Returns a new GraphEdge with source/target swapped.
to_dict() / from_dict(data)Serialization round-trip.
update_weight(new_weight)Mutates the edge weight.
Equality/hashBased on (source, target, relation) / edge_id.

KnowledgeGraph

Python · 2 lines
class KnowledgeGraph:
    def __init__(self, config: Optional[ConfigGraphRAG] = None): ...

Defaults to ConfigGraphRAG() when no config is supplied — the docstring notes this specifically so standalone usage (without GraphRAG) still works.

MethodDescription
add_node(node) -> boolAdds a node; True if new, False if it updated an existing one.
remove_node(node_id) -> boolRemoves a node and all connected edges.
get_node(node_id) -> Optional[GraphNode]Direct lookup.
add_edge(edge) -> boolAdds an edge.
get_edge(source, target, relation) -> Optional[GraphEdge]Direct lookup.
get_neighbors(node_id, max_depth=None, include_incoming=True) -> Set[str]BFS neighbor discovery.
find_path(start, end, max_length=None) -> Optional[List[str]]Shortest path via BFS.
find_all_paths(start, end, max_length=5) -> List[List[str]]All paths via DFS, bounded by max_length.
get_subgraph(node_ids) -> KnowledgeGraphExtracts an induced subgraph.
get_node_degree(node_id) -> Dict[str, int]O(1) lookup — the source comments note this was fixed (labeled "FIX 7") to use pre-built _in_degree counters and adjacency-list length instead of scanning every adjacency list, which was previously O(nodes).
get_connected_components() -> List[Set[str]]Undirected connected components.
get_stats() -> Dict[str, Any]Node/edge counts and summary stats.
validate_integrity() -> Dict[str, List[str]]Structural consistency checks (e.g. dangling edges).

GraphRAG

Constructor

Python
def __init__(self, vector_db, llm: Optional[Any] = None, config: Optional[ConfigGraphRAG] = None)

vector_db is required; llm is any object exposing .generate(prompt).

Core methods

MethodPurpose
add_document_with_relations(content, entities, relations, doc_id, metadata=None) -> Dict[str, Any]Adds a document plus its extracted entities/relations to both the vector DB and the knowledge graph.
retrieve_with_context(query, k=None, context_depth=None, min_similarity=0.0, combine_scores=True) -> List[Dict[str, Any]]Semantic search, then _expand_graph_context performs BFS expansion from the initial hit nodes, optionally combining vector similarity with graph-derived scores.
get_node_context(node_id, max_depth=2, include_chunks=True) -> Dict[str, Any]Full context (neighbors, chunks) for one node.
semantic_search(query, top_k=5, include_graph_info=True) -> List[Dict[str, Any]]Pure vector search, optionally annotated with graph metadata.
query(query, k=None, language="ar", include_sources=False, **llm_kwargs) -> strMain interface: retrieve → build LLM context (_build_graph_context) → build prompt (_build_prompt) → generate.
generate(query, k=None, **kwargs) -> strAlias for query(), for interoperability with other RAG types in this package.
get_statistics() -> Dict[str, Any]Combined vector DB + graph stats.
save(path) / load(path)Persist/restore system state.

Async API

aquery, agenerate (delegates to aquery), aretrieve, plus __aenter__/__aexit__.

[!NOTE] Default language is Arabic query()'s language parameter defaults to "ar" — pass language="en" explicitly for English prompts.

Complete example

Python · 17 lines
from zerda.rag.types.graph_rag import GraphRAG, ConfigGraphRAG

config = ConfigGraphRAG(k=8, context_depth=2, max_depth=3)
graph_rag = GraphRAG(vector_db=my_vector_db, llm=my_llm, config=config)

graph_rag.add_document_with_relations(
    content="Paris is the capital of France. The Eiffel Tower is in Paris.",
    entities=[{"id": "paris", "content": "Paris", "node_type": "city"},
              {"id": "france", "content": "France", "node_type": "country"},
              {"id": "eiffel_tower", "content": "Eiffel Tower", "node_type": "landmark"}],
    relations=[{"source": "paris", "target": "france", "relation": "capital_of"},
               {"source": "eiffel_tower", "target": "paris", "relation": "located_in"}],
    doc_id="doc_1",
)

answer = graph_rag.query("What landmark is in the French capital?", language="en")
print(answer)

Best practices

  • Keep max_depth small (2–3); BFS expansion cost and noise both grow quickly with depth in dense graphs.
  • Use KnowledgeGraph directly (without GraphRAG) if you only need graph storage/traversal utilities and already have your own vector search.

Common mistakes

  • Instantiating GraphRAG without a working ....chunks.DocumentChunk import available in the parent project — this will raise an ImportError at module load time, not at call time.
  • Forgetting query() defaults to Arabic output — a common surprise when testing with English-only content.
  • Hybrid Search — combine with enable_hybrid_search in ConfigGraphRAG for keyword+semantic+graph retrieval.

Related pages