RAGExporter

Cuts an inventory into text chunks with metadata, ready to embed for retrieval-augmented generation.

Last changed v0.6.0

A language model answering questions about your cloud needs the right slice of the inventory in its context window, not the whole inventory-map.json. RAGExporter writes three kinds of chunk, each aimed at a different kind of question, as one JSON object per line. Embed the content field, keep metadata for filtering, and you have a retrieval corpus. Nothing in the exporter calls a model or an embedding API.

Chunk kinds

Kind	chunk_id	One chunk per	Good for
entity	entity::<asset id>	asset	"tell me about orders-db"
community	community::<n>	cluster of two or more tightly connected nodes (Louvain)	"what sits around this subnet?", blast-radius style questions
relation_group	relation_group::<GROUP>	relation group present (NETWORK, IAM, SECURITY and so on)	"list every rule open to the internet", "show the IAM grants"

Every chunk has the same five keys: chunk_id, chunk_type, content (the text to embed), metadata (a flat-ish dict for filtering) and relations (a list of dicts).

Kind	metadata keys	relations
entity	asset_type, provider, region, account_id, arn, is_internet_exposed, relation_types, severity_max, compliance_frameworks, neighbour_count, finding_count	every inferred relation of the asset's edges, with predicate, direction (outgoing or incoming), object, object_id, evidence
community	community_id, member_count, asset_types, internal_edges, external_edges, internet_exposed_count, finding_count, severity_max, compliance_frameworks, risk_score	empty
relation_group	relation_group, total_relations, relation_type_counts, unique_subjects, unique_objects	the triples that fit into the chunk's text, with predicate, object, evidence

severity_max is the worst severity among the matched findings, or "NONE". compliance_frameworks lists the frameworks of every matched finding, including findings cut from the chunk's text. A community's risk_score is 0.3 per member plus 0.5 per finding plus 2.0 per internet-exposed member, capped at 10.

The relations are the ones the ontology uses: infer_relations() on each edge (so a security group rule from 0.0.0.0/0 on port 443 becomes INGRESS_ALLOWED, INTERNET_REACHABLE and ONLY_HTTPS), plus the asset-level relations from metadata (ENCRYPTED_BY_KMS, VPC containment from vpc_id, tag ownership, key rotation, EC2 security group membership). Relation-group chunks carry both; entity chunks carry the edge relations only.

Findings attach to assets through AssetIndex, so a finding whose resource_id is an ARN or a unique name still counts toward the right entity and community.

Output files

export_all() writes two files into output_dir (created if missing) and returns their paths as {"chunks": Path, "index": Path}. Which chunk kinds go in depends on chunk_strategy: hybrid (the default) writes all three, entity, community or relation_group writes that kind only, and the index counts the other kinds as 0.

File	Contents
rag_chunks.jsonl	one chunk per line: entity chunks first, then community, then relation-group
rag_metadata_index.json	total_chunks, entity_chunks, community_chunks, relation_group_chunks, chunk_ids (in file order) and chunk_types

flowchart LR
  A["assets, edges"] --> E["export_entity_chunks"]
  G["graph"] --> L["Louvain communities"] --> C["export_community_chunks"]
  A --> R["export_relation_chunks"]
  F["findings"] --> E
  F --> C
  E --> J["rag_chunks.jsonl"]
  C --> J
  R --> J
  J --> I["rag_metadata_index.json"]

What export_all() does

cloudg run writes these files twice. Phase 2c exports right after the graph is built, when the only findings are the reachability ones. After the scanners finish, a silent second pass rewrites both files with every finding. If that second pass fails, the files from phase 2c stay, with only reachability findings in them. --no-rag-export or rag.enabled: false turns the export off.

Examples

Export and reshape for a vector store

The script needs nothing beyond cloudg. It exports the chunks, then turns every chunk into an {id, text, metadata} record with plain-text content and scalar metadata, which is the shape most vector stores accept.

rag_demo.py

"""Export RAG chunks for a small estate and reshape them for a vector store."""

import json
from pathlib import Path

from cloudg.graph.builder import GraphBuilder
from cloudg.graph.rag_export import RAGExporter
from cloudg.schema.models import (
    AssetType,
    CloudAsset,
    CloudProvider,
    EdgeType,
    Finding,
    NetworkEdge,
    Severity,
)

ACCOUNT = "123456789012"
REGION = "eu-west-1"

def asset(asset_id: str, asset_type: AssetType, arn: str, **extra) -> CloudAsset:
    return CloudAsset(
        id=asset_id,
        name=asset_id,
        asset_type=asset_type,
        provider=CloudProvider.AWS,
        region=REGION,
        account_id=ACCOUNT,
        arn=arn,
        **extra,
    )

assets = [
    asset("sg-web", AssetType.SECURITY_GROUP, f"arn:aws:ec2:{REGION}:{ACCOUNT}:security-group/sg-web"),
    asset("web-1", AssetType.EC2, f"arn:aws:ec2:{REGION}:{ACCOUNT}:instance/i-0web1",
          is_internet_exposed=True),
    asset("web-2", AssetType.EC2, f"arn:aws:ec2:{REGION}:{ACCOUNT}:instance/i-0web2"),
    asset("app-role", AssetType.IAM_ROLE, f"arn:aws:iam::{ACCOUNT}:role/app-role"),
    asset("orders-db", AssetType.RDS_INSTANCE, f"arn:aws:rds:{REGION}:{ACCOUNT}:db:orders-db"),
]
edges = [
    NetworkEdge(source_id="0.0.0.0/0", target_id="sg-web", edge_type=EdgeType.SECURITY_GROUP_RULE,
                port_range="443", protocol="TCP", cidr="0.0.0.0/0", direction="ingress"),
    NetworkEdge(source_id="web-1", target_id="sg-web", edge_type=EdgeType.ATTACHED_TO),
    NetworkEdge(source_id="web-2", target_id="sg-web", edge_type=EdgeType.ATTACHED_TO),
    NetworkEdge(source_id="web-1", target_id="app-role", edge_type=EdgeType.ASSUMES_ROLE),
    NetworkEdge(source_id="app-role", target_id="orders-db", edge_type=EdgeType.GRANTS_ACCESS),
]
findings = [
    Finding(
        resource_id="web-1",
        severity=Severity.HIGH,
        title="Instance metadata service v1 enabled",
        description="IMDSv2 is not enforced.",
        source_tool="prowler",
        compliance_frameworks=["CIS"],
    )
]

graph = GraphBuilder().build(assets, edges)
paths = RAGExporter().export_all(assets, edges, graph, findings, output_dir="rag-out")
print({kind: path.name for kind, path in paths.items()})

index = json.loads(paths["index"].read_text())
print({k: v for k, v in index.items() if k.endswith("chunks")})

chunks = [json.loads(line) for line in paths["chunks"].read_text().splitlines()]

# The content uses a few symbols; plain words embed and display more reliably
MARKERS = {
    "\N{WARNING SIGN}": "WARNING:",
    "\N{RIGHTWARDS ARROW}": "->",
    "\N{LEFTWARDS ARROW}": "<-",
    "\N{BULLET}": "-",
}

def clean(text: str) -> str:
    for symbol, word in MARKERS.items():
        text = text.replace(symbol, word)
    return text

def scalar(value):
    """Most vector stores accept only str, int, float and bool metadata."""
    if isinstance(value, (str, int, float, bool)):
        return value
    if isinstance(value, list):
        return ",".join(map(str, value))
    return json.dumps(value, sort_keys=True)

records = [
    {
        "id": c["chunk_id"],
        "text": clean(c["content"]),
        "metadata": {"chunk_type": c["chunk_type"],
                     **{k: scalar(v) for k, v in c["metadata"].items()}},
    }
    for c in chunks
]
out = Path("rag-out") / "vector-records.jsonl"
out.write_text("".join(json.dumps(r) + "\n" for r in records))
print(len(records), "records ->", out)

web_1 = next(r for r in records if r["id"] == "entity::web-1")
print(web_1["text"])
print(web_1["metadata"])

# Pre-filter on metadata before any similarity search
urgent = [r["id"] for r in records if r["metadata"].get("severity_max") in ("CRITICAL", "HIGH")]
print("high or critical:", urgent)

$ python rag_demo.py
{'chunks': 'rag_chunks.jsonl', 'index': 'rag_metadata_index.json'}
{'total_chunks': 11, 'entity_chunks': 5, 'community_chunks': 2, 'relation_group_chunks': 4}
11 records -> rag-out/vector-records.jsonl
Resource: web-1
Type: EC2
Provider: AWS
Region: eu-west-1
ARN: arn:aws:ec2:eu-west-1:123456789012:instance/i-0web1
Account: 123456789012
WARNING: INTERNET EXPOSED

Relations (2):
  -> PROTECTED_BY_SG: sg-web
  -> RUNS_ON: app-role

Findings (1):
  [HIGH] Instance metadata service v1 enabled
{'chunk_type': 'entity', 'asset_type': 'EC2', 'provider': 'AWS', 'region': 'eu-west-1', 'account_id': '123456789012', 'is_internet_exposed': True, 'relation_types': 'PROTECTED_BY_SG,RUNS_ON', 'severity_max': 'HIGH', 'compliance_frameworks': 'CIS', 'neighbour_count': 2, 'finding_count': 1, 'arn': 'arn:aws:ec2:eu-west-1:123456789012:instance/i-0web1'}
high or critical: ['entity::web-1', 'community::1']

The community number can differ on your run (see the notes). Each line of vector-records.jsonl looks like this, wrapped here for reading:

rag-out/vector-records.jsonl

{"id": "community::1",
 "text": "Community 1 (3 resources):\n  - web-1 (EC2)\n  - app-role (IAM_ROLE)\n  - orders-db (RDS_INSTANCE)\n\nAsset types: {'EC2': 1, 'IAM_ROLE': 1, 'RDS_INSTANCE': 1}\nInternal edges: 2, External edges: 1\nWARNING: 1 internet-exposed resources\nFindings: 1 (max severity: HIGH)",
 "metadata": {"chunk_type": "community", "community_id": 1, "member_count": 3,
              "asset_types": "{\"EC2\": 1, \"IAM_ROLE\": 1, \"RDS_INSTANCE\": 1}",
              "internal_edges": 2, "external_edges": 1, "internet_exposed_count": 1,
              "finding_count": 1, "severity_max": "HIGH", "compliance_frameworks": "CIS",
              "risk_score": 3.4}}

From here, loading into a store is a few lines in that store's own client: pass id, text (or your embedding of it) and metadata to its add or upsert call. Use the chunk_type and severity_max fields as filters so a question about one asset retrieves entity chunks first.

Only one kind of chunk

The three export_* methods return RAGChunk objects without writing anything. Call one directly when you only want, say, the relation groups:

relation_chunks.py

from cloudg.graph.rag_export import RAGExporter
from cloudg.inventory import InventoryResult

result = InventoryResult.load("./reports")
assets_by_id = {a.id: a for a in result.assets}
for chunk in RAGExporter().export_relation_chunks(result.edges, assets_by_id):
    data = chunk.to_dict()
    print(data["chunk_id"], data["metadata"]["total_relations"])

Notes

• RAGExporter(max_chunk_tokens=2000, chunk_strategy="hybrid") takes the two rag keys of config.yaml, and cloudg run and CloudGEngine.analyze() pass them in. An unknown strategy or a max_chunk_tokens below 1 raises ValueError.
• max_chunk_tokens is a size budget for each chunk's content, counted as about 4 characters per token. The lists in a chunk (an entity's relations and findings, a community's members, a relation group's triples) are cut when the next line would not fit, and a last line such as ... and 12 more says how many were left out. When an entity chunk has both relations and findings, the two lists take turns, so a long one cannot push the other out. The fixed lines (name, type, ARN, tags) are always kept, so a chunk can still run over the budget by a little.
• Community numbers come from python-louvain's best_partition without a fixed seed, so community::3 in one run is not the same cluster as community::3 in the next. Key on members, not on the id. Without python-louvain installed, the exporter logs a warning and uses connected components instead.
• Communities are found on the undirected version of the graph, and nodes left alone in a community (singletons) produce no chunk. Placeholder nodes such as 0.0.0.0/0 take part in clustering like any other node.
• Entity chunks are built from the NetworkEdge list, so two rules on the same pair appear separately there. Community chunks read the graph, where GraphBuilder has merged them, and take is_internet_exposed from the graph's node attributes.
• content uses four non-ASCII symbols: a warning sign (U+26A0) before exposure lines, arrows (U+2192, U+2190) before outgoing and incoming relations, and a bullet (U+2022) before community members. rag_chunks.jsonl stores them as \u escapes, and they come back as the symbols when parsed. The example swaps them for words before embedding.
• chunk_types in the index comes from a set, so its order varies between runs.

Reference

RAGExporter

RAGExporter(max_chunk_tokens: int = 2000, chunk_strategy: str = 'hybrid')

Generates RAG-ready chunks from cloud infrastructure data.

Three strategies, each producing different chunk types for retrieval:

1. Entity-centric: one chunk per asset with all direct relations, findings, and compliance status. Best for "tell me about resource X".

2. Community-detection: Louvain-clustered groups of tightly-connected resources. Best for "what's the blast radius of this subnet?".

3. Relation-group: all triples of a semantic group (Network, IAM, etc.). Best for "show me all IAM access rules" or "list all network paths".

Parameters:

Name	Type	Description	Default
max_chunk_tokens	int	Approximate content size limit of one chunk (rag.max_chunk_tokens), counted as CHARS_PER_TOKEN characters per token. Lists that do not fit are cut with an ... and N more line; the fixed lines of a chunk (an asset's name, type, ARN, tags) are always kept, so one chunk can still go over the limit.	2000
chunk_strategy	str	Chunk kinds export_all writes (rag.chunk_strategy): entity, community, relation_group, or hybrid for all three.	'hybrid'

Raises:

Type	Description
ValueError	max_chunk_tokens is below 1 or chunk_strategy is not one of the four names.

Source code in cloudg/graph/rag_export.py

def __init__(self, max_chunk_tokens: int = 2000, chunk_strategy: str = "hybrid") -> None:
    """
    Args:
        max_chunk_tokens: Approximate content size limit of one chunk
            (``rag.max_chunk_tokens``), counted as
            :data:`CHARS_PER_TOKEN` characters per token. Lists that
            do not fit are cut with an ``... and N more`` line; the
            fixed lines of a chunk (an asset's name, type, ARN, tags)
            are always kept, so one chunk can still go over the limit.
        chunk_strategy: Chunk kinds :meth:`export_all` writes
            (``rag.chunk_strategy``): ``entity``, ``community``,
            ``relation_group``, or ``hybrid`` for all three.

    Raises:
        ValueError: ``max_chunk_tokens`` is below 1 or
            ``chunk_strategy`` is not one of the four names.
    """
    if max_chunk_tokens < 1:
        raise ValueError(f"max_chunk_tokens must be at least 1, got {max_chunk_tokens}")
    if chunk_strategy not in CHUNK_STRATEGIES:
        raise ValueError(
            f"Unknown chunk_strategy {chunk_strategy!r}; expected one of "
            + ", ".join(CHUNK_STRATEGIES)
        )
    self._max_tokens = max_chunk_tokens
    self._strategy = chunk_strategy

export_all

export_all(
    assets: list[CloudAsset],
    edges: list[NetworkEdge],
    graph: DiGraph,
    findings: list[Finding] | None = None,
    output_dir: str | Path = "./reports",
) -> dict[str, Path]

Export the chunk kinds of the chunk strategy to files.

With the default hybrid strategy that is all three kinds; any other strategy writes one kind and counts the others as 0 in rag_metadata_index.json.

Returns:

Type	Description
dict[str, Path]	Dict mapping chunk type → output file path.

Source code in cloudg/graph/rag_export.py

def export_all(
    self,
    assets: list[CloudAsset],
    edges: list[NetworkEdge],
    graph: nx.DiGraph,
    findings: list[Finding] | None = None,
    output_dir: str | Path = "./reports",
) -> dict[str, Path]:
    """Export the chunk kinds of the chunk strategy to files.

    With the default ``hybrid`` strategy that is all three kinds; any
    other strategy writes one kind and counts the others as 0 in
    rag_metadata_index.json.

    Returns:
        Dict mapping chunk type → output file path.
    """
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)

    assets_by_id = {a.id: a for a in assets}
    kinds = (
        {"entity", "community", "relation_group"}
        if self._strategy == "hybrid"
        else {self._strategy}
    )
    entity_chunks = (
        self.export_entity_chunks(assets, edges, findings) if "entity" in kinds else []
    )
    community_chunks = (
        self.export_community_chunks(graph, assets_by_id, findings)
        if "community" in kinds
        else []
    )
    relation_chunks = (
        self.export_relation_chunks(edges, assets_by_id) if "relation_group" in kinds else []
    )
    all_chunks = [*entity_chunks, *community_chunks, *relation_chunks]
    paths = _write_export(
        out,
        all_chunks,
        {
            "entity_chunks": len(entity_chunks),
            "community_chunks": len(community_chunks),
            "relation_group_chunks": len(relation_chunks),
        },
    )

    logger.info(
        "RAG export complete: %d chunks (%d entity, %d community, %d relation) -> %s",
        len(all_chunks),
        len(entity_chunks),
        len(community_chunks),
        len(relation_chunks),
        out,
    )
    return paths

export_entity_chunks

export_entity_chunks(
    assets: list[CloudAsset],
    edges: list[NetworkEdge],
    findings: list[Finding] | None = None,
) -> list[RAGChunk]

Generate one chunk per cloud asset with 1-hop neighbourhood.

Source code in cloudg/graph/rag_export.py

def export_entity_chunks(
    self,
    assets: list[CloudAsset],
    edges: list[NetworkEdge],
    findings: list[Finding] | None = None,
) -> list[RAGChunk]:
    """Generate one chunk per cloud asset with 1-hop neighbourhood."""
    assets_by_id = {a.id: a for a in assets}
    findings_by_resource = _index_findings(findings, AssetIndex(assets_by_id.values()))

    # Build adjacency from edges
    outgoing: dict[str, list[NetworkEdge]] = defaultdict(list)
    incoming: dict[str, list[NetworkEdge]] = defaultdict(list)
    for edge in edges:
        outgoing[edge.source_id].append(edge)
        incoming[edge.target_id].append(edge)

    chunks = [
        self._build_entity_chunk(
            asset,
            _entity_relations(asset, assets_by_id, outgoing, incoming),
            findings_by_resource.get(asset.id, []),
        )
        for asset in assets
    ]

    logger.info("Generated %d entity chunks", len(chunks))
    return chunks

export_community_chunks

export_community_chunks(
    graph: DiGraph,
    assets_by_id: dict[str, CloudAsset] | None = None,
    findings: list[Finding] | None = None,
) -> list[RAGChunk]

Generate one chunk per detected community of resources.

Source code in cloudg/graph/rag_export.py

def export_community_chunks(
    self,
    graph: nx.DiGraph,
    assets_by_id: dict[str, CloudAsset] | None = None,
    findings: list[Finding] | None = None,
) -> list[RAGChunk]:
    """Generate one chunk per detected community of resources."""
    if graph.number_of_nodes() == 0:
        return []

    partition = _detect_communities(graph)

    # Group nodes by community
    communities: dict[int, list[str]] = defaultdict(list)
    for node_id, comm_id in partition.items():
        communities[comm_id].append(node_id)

    findings_by_resource = _index_findings(findings, _graph_index(graph, assets_by_id))

    chunks = [
        self._build_community_chunk(graph, comm_id, members, findings_by_resource)
        for comm_id, members in communities.items()
        if len(members) >= 2  # Skip singleton communities
    ]

    logger.info("Generated %d community chunks", len(chunks))
    return chunks

export_relation_chunks

export_relation_chunks(
    edges: list[NetworkEdge], assets_by_id: dict[str, CloudAsset]
) -> list[RAGChunk]

Generate one chunk per semantic relation group.

Source code in cloudg/graph/rag_export.py

def export_relation_chunks(
    self,
    edges: list[NetworkEdge],
    assets_by_id: dict[str, CloudAsset],
) -> list[RAGChunk]:
    """Generate one chunk per semantic relation group."""
    # Group edges by inferred relation group
    group_triples = _group_relation_triples(edges, assets_by_id)

    chunks = [
        self._build_relation_group_chunk(group, triples)
        for group, triples in group_triples.items()
    ]

    logger.info("Generated %d relation-group chunks", len(chunks))
    return chunks

Related

• CloudOntology

  The same relations as RDF.

• GraphBuilder

  Builds the graph the community chunks are cut from.

• Graph, ontology and RAG

  The analysis phase end to end.

• cloudg run

  The --rag-export switch.

• Output files

  Where rag_chunks.jsonl sits among the other outputs.