Structured Ingestion

from graphrag_sdk import Column, Entity, GraphRAG, Link, Ontology, TableMapping

ONTOLOGY = Ontology(
    entities=[Entity(label="Person"), Entity(label="Organization")],
    tables=[
        TableMapping(
            source="employees.csv",
            label="Person",
            key="employee_id",
            name="full_name",
            properties={"age": Column("age", "INTEGER")},
            links=[Link("WORKS_AT", to="Organization", by="org_id")],
        ),
    ],
)

async with GraphRAG(connection=..., llm=..., embedder=..., ontology=ONTOLOGY) as rag:
    await rag.ingest("data/board_review.pdf")   # prose
    await rag.ingest("data/employees.csv")      # a table
    summary = await rag.finalize()

// every entity, with where its properties came from
MATCH (e:__Entity__)
RETURN e.name, [k IN keys(e) WHERE k CONTAINS '__'] AS signed, coalesce(e.description, '') <> '' AS from_prose
ORDER BY e.name

// a person, the row that produced her, and the page that mentions her
MATCH (p:Person {name: 'Maya Ellison'})-[:MENTIONED_IN]->(c:Chunk)<-[:PART_OF]-(d:Document)
RETURN d.id, c.kind, c.record_key

// the edges a CSV Link created, with the signed value they carry
MATCH (p:Person)-[r:RELATES]->(o:Organization)
WHERE r.rel_type = 'WORKS_AT'
RETURN p.name, p.employees__age, o.name, o.organizations__country

from graphrag_sdk import Column, TableMapping

ORGS = TableMapping(
    source="orgs.csv",       # the file this maps, matched on its basename
    label="Organization",    # what each record becomes
    key="org_id",            # the column identifying it
    name="org_name",         # the column holding its display name
    properties={
        "hq_country": "hq_country",
        "employee_count": Column("employee_count", "INTEGER"),
    },
)

from graphrag_sdk import Link

EMPLOYEES = TableMapping(
    source="employees.csv",
    label="Person",
    key="employee_id",
    name="full_name",
    properties={
        "age": Column("age", "INTEGER"),
        "title": Column("job_title"),
    },
    links=[Link("WORKS_AT", to="Organization", by="org_id")],
)

ONTOLOGY = Ontology(
    entities=[Entity(label="Person"), Entity(label="Organization")],
    tables=[ORGS, EMPLOYEES],
)

Link("WORKS_AT", to="Organization", by="org_id",
     name="org_name",                                  # names the target
     properties={"since": Column("start_date", "DATE")})  # goes on the edge

properties={"hq_country": Column("HQ Country")}

result = await rag.ingest("employees.csv")
result.rows_in_source   # 4  — what the file held
result.rows_skipped     # 1  — what had no key
result.records          # 3  — what was written

result = await rag.ingest("employees.csv")   # nothing declared for it

No table mapping is declared for employees.csv, so it was proposed by the model:
label 'Person', name 'full_name', key 'employee_id', links WORKS_AT->Organization by
org_id. finalize() reports it under proposed_mappings. Declare a
TableMapping(source='employees.csv', ...) in the ontology to replace it, or
drop_table('employees.csv') to remove it.

summary = await rag.finalize()
summary.proposed_mappings   # ["employees.csv"]

await rag.drop_table("employees.csv")

MappingError: readings.csv has no column that is unique and complete, so no row
can be given a stable identity. Declare a mapping naming the key, or add an id
column to the export.

from graphrag_sdk import TextLoader

await rag.ingest("tickets.csv", loader=TextLoader())

MappingError: mapping does not fit employees.csv:
  node 'Person': property 'age' reads missing column 'age'
  node 'Organization': key column 'org_id' is not in the source

ontology = await rag.get_ontology()
# Person: employees__employee_id:STRING, employees__age:INTEGER,
#         employees__title:STRING, employees__start_date:DATE
# WORKS_AT: [('Person', 'Organization')]

await rag.ingest("employees.csv")   # first write
# ... the export is regenerated, with one row corrected, one added, one gone ...
result = await rag.ingest("employees.csv")   # re-sync
result.entities_deleted   # 1 — the row that disappeared took its entity with it

employees.csv   age              →  Person.employees__age
organizations.csv   country      →  Organization.organizations__country
employees.csv   WORKS_AT.since   →  employees__since   (on the edge)

rows = await rag.query(
    "MATCH (p:Person) RETURN avg(p.employees__age) AS mean_age"
)

summary = await rag.finalize()
summary.property_conflicts
# ["Person.grade — supplied by finance.csv, hr.csv; 1 of 2 entities hold different values"]

'/exports/2026-02/hr.csv' and the already-registered '/exports/2026-01/hr.csv'
both reduce to the property signature 'hr'. […] The signature is the file's
basename, so give one of them a distinct filename (for example hr_2026_02.csv).

employees.csv row  →  Person  name: "Alice Smith"   id: alice_smith__person
board_note.txt     →  Person  name: "Alice Smith"   id: alice_smith__person
                                                        ^^^^^^^^^^^^^^^^^^^ the same node

summary = await rag.finalize()
summary.probable_duplicates   # names that look like one thing and did not merge

await rag.ingest("employees.csv")                                   # Priya Raman, 39
await rag.ingest(text=MARKET_NOTE, document_id="market_note.txt")   # "Ms. Raman said…"
summary = await rag.finalize()
summary.resolved_duplicates   # ["Person 'Ms. Raman' -> 'Priya Raman'"]

summary.resolved_duplicates   # ["Person 'Ms. Raman' -> 'Priya Raman'"]   merged
summary.rejected_duplicates   # ["Contract 'transmission agreement' | 'Nordic transmission agreement'"]
summary.probable_duplicates   # what is left: flagged by a name rule, judged by nobody

await rag.query("""
    MATCH (e:__Entity__)-[:MENTIONED_IN]->(:Chunk)<-[:PART_OF]-(d:Document)
    WITH e, collect(DISTINCT d.id) AS sources
    WHERE size(sources) > 1
    RETURN e.name, sources ORDER BY e.name
""")

await rag.ingest("employees.csv")
await rag.ingest(text=BOARD_NOTE, document_id="board_note.txt")
await rag.finalize()

result = await rag.finalize()
result.unmerged_name_collisions   # {"Apple": ["organization", "product"]}

rag = GraphRAG(
    connection=ConnectionConfig(host="localhost", graph_name="my_graph"),
    llm=llm,
    embedder=embedder,
    ontology=ONTOLOGY,
    enable_cypher=True,
)

answer = await rag.completion("What is the average age of employees at Acme Corp?")

rows = await rag.query(
    "MATCH (p:Person)-[r:RELATES]->(o:Organization) "
    "WHERE r.rel_type = 'WORKS_AT' AND o.name = $company "
    "RETURN avg(p.employees__age) AS mean_age",
    {"company": "Acme Corp"},
)

Q: Who works at the company that reported the revenue miss, and how old are they?
A: Alice Smith, who is 34, and Bob Jones, who is 45, work at Acme Corp.

from graphrag_sdk import DocumentInfo, RecordBatch, RecordLoaderStrategy

class ParquetRecordLoader(RecordLoaderStrategy):
    async def load_records(self, source: str, ctx) -> RecordBatch:
        return RecordBatch(
            open_records=lambda: iter(read_parquet(source)),
            columns=["a", "b"],
            document_info=DocumentInfo(uid=source, path=source),
        )

await rag.ingest("exports/readings.csv", record_loader=ParquetRecordLoader())