Extract and resolve entities

This guide shows how to extract entities and relations from chunks, and how to resolve duplicate mentions into groups.

Extract with the default schema

import asyncio
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    ConsoleSpanExporter,
    SimpleSpanProcessor,
)

from agrag.common.data_models.graph_schema import GENERIC
from agrag.ingestion.extract import BAMLExtractor, ExtractionLLMSettings

# One tracer for the whole application. Every component below that takes
# `tracer=` writes into this trace. Leave it out and no span is recorded.
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
tracer = provider.get_tracer("my-app")


async def main():
    # Point at your OpenAI-compatible endpoint via environment variables:
    #   LLM_BASE_URL, LLM_MODEL_ID, and optionally LLM_API_KEY
    settings = ExtractionLLMSettings.from_openai_compatible_env()
    extractor = BAMLExtractor(settings=settings, tracer=tracer)
    result = await extractor.extract(chunk, GENERIC)
    print(result.entities)


asyncio.run(main())

BAMLExtractor needs the llm extra:

pip install "agentic-graphrag[llm]"

Give the model the section heading

When a chunk has a heading path, BAMLExtractor sends it as a separate section line above the text block, for example Guide > Setup. The prompt says that the line is context only. Character offsets still index the chunk text, because the heading is not part of it. Set include_heading_path=False to turn this off:

extractor = BAMLExtractor(settings=settings, include_heading_path=False)

Only BAMLExtractor uses the heading path. The local and escalating extractors read the chunk text only. See Configure chunking for the embedding side of this setting.

Run locally first, escalating to an LLM

import asyncio
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    ConsoleSpanExporter,
    SimpleSpanProcessor,
)

from agrag.common.data_models.graph_schema import GENERIC
from agrag.ingestion.extract import (
    BAMLExtractor,
    EscalatingExtractor,
    ExtractionLLMSettings,
    GlinerExtractor,
)

# A second component of the same application writes into the same trace.
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
tracer = provider.get_tracer("my-app")


async def main():
    settings = ExtractionLLMSettings.from_openai_compatible_env()
    extractor = EscalatingExtractor(
        primary=GlinerExtractor(tracer=tracer),
        escalate_to=BAMLExtractor(settings=settings, tracer=tracer),
        tracer=tracer,
    )
    result = await extractor.extract(chunk, GENERIC)
    print(result.entities)


asyncio.run(main())

GlinerExtractor needs the extract extra:

pip install "agentic-graphrag[extract]"

Configure the LLM provider

Set EXTRACTION_LLM_CLIENTS (or a .env file) to point at any supported provider — Anthropic, OpenAI, Azure OpenAI, AWS Bedrock, Google AI, Vertex AI, or a self-hosted OpenAI-compatible endpoint via openai-generic.

Alternatively, use ExtractionLLMSettings.from_openai_compatible_env() to read from LLM_BASE_URL, LLM_API_KEY, and LLM_MODEL_ID environment variables.

Resolution runs inside every write

You never call resolution directly. graph.add() resolves every batch automatically: exact-name identity groups mentions without evidence, a fuzzy fast path merges near-identical names, embedding similarity decides the rest, and a capped LLM tier reviews only the ambiguous band. Raw entity nodes are permanent. Confirmed non-exact matches persist as MATCHES edges with a derived resolved node, never by deleting a node.

result = await graph.add(text="Ada Lovelace wrote the first algorithm.")
print(result.resolution.exact_match_hits)
print(result.resolution.ambiguous_count)

Configure resolution through the settings you pass when you open the graph: the schema, the embedder, and the extractor. Inspect every run through the resolution stats on the returned AddResult.

Consolidate the whole graph

consolidate() runs the same resolution as a full pass over every persisted entity of each type. Dry-run first, then apply:

report = await graph.consolidate()
print([(m.entity_a_id, m.entity_b_id) for m in report.would_match])
print(report.ambiguous_count)

applied = await graph.consolidate(apply=True)

A dry run writes nothing. Applying writes MATCHES edges and derived resolved nodes, and reports exactly the matches it wrote. LLM review calls stay bounded no matter how large the graph grows.

Re-evaluate specific entities

reevaluate() re-runs the same resolution for entities you name, when new evidence lands or a past decision looks wrong. It adds newly confirmed matches, deactivates edges the current evidence no longer supports, and leaves every entity outside the list untouched:

report = await graph.reevaluate([entity_id_a, entity_id_b])
print(report.matches_added)
print(report.matches_removed)

There is no automatic trigger: a caller that learns something new calls reevaluate() explicitly.

Resolver.resolve() also takes an optional neighbors_by_index map of per-entity neighboring-relationship strings, which batched LLMVerify verification sends to the model as extra context. Graph.add() and Graph.consolidate() build one automatically from a batch's extracted relations and each persisted entity's stored graph edges.

Define your own schema

from agrag.common.data_models.graph_schema import (
    EntityType,
    GraphSchema,
    RelationType,
)

schema = GraphSchema(
    name="clinical",
    version="1",
    entities=[EntityType(label="Drug", description="A medication.")],
    relations=[],
)

Next steps

• See the API reference for every Extractor, Comparator, and schema field.