owlapy.agen_kg.graph_extractor

Attributes

logger

Classes

`GraphExtractorMeta`	Metaclass that resolves conflicts between dspy.Module and ABC.
`GraphExtractor`	Base class for all graph extractors.

Module Contents

owlapy.agen_kg.graph_extractor.logger

class owlapy.agen_kg.graph_extractor.GraphExtractorMeta[source]

Bases: type(dspy.Module), type(ABC)

Metaclass that resolves conflicts between dspy.Module and ABC.

class owlapy.agen_kg.graph_extractor.GraphExtractor(enable_logging=False, use_incremental_merging=True)[source]

Bases: dspy.Module, abc.ABC

Base class for all graph extractors. Provides common functionality for entity clustering, coherence checking, text chunking for large documents, and utility methods shared across all extractor types.

fact_checking_instructions = None

triple_with_literal_extraction_instructions = None

literal_extraction_instructions = None

type_assertion_instructions = None

type_generation_instructions = None

triple_extraction_instructions = None

entity_extraction_instructions = None

logging = False

use_incremental_merging = True

plan_decomposer

entity_deduplicator

coherence_checker

type_clusterer

relation_clusterer

entity_deduplicator_with_summary

type_clusterer_with_summary

relation_clusterer_with_summary

text_summarizer

chunk_summarizer

entity_merger

triple_merger

type_merger

batch_rdfs_comment_generator

text_loader

text_chunker

auto_chunk_threshold = 4000

summarization_threshold = 8000

max_summary_length = 3000

static snake_case(text)[source]

static format_type_name(text)[source]

Format a type name to start with capital letter and rest lowercase. E.g., ‘PERSON’ -> ‘Person’, ‘person’ -> ‘Person’, ‘PERSON_TYPE’ -> ‘Person_type’

Parameters:: text – The type name to format.
Returns:: Formatted type name with capital first letter and lowercase rest.

static format_rdfs_label(label: str, is_property: bool = False) → str[source]

Convert a raw entity identifier into a clean rdfs:label suitable for RDF annotation.

This method normalizes the input string by removing special characters, standardizing whitespace, and applying consistent casing rules. For non-property labels, it produces a title-cased form with a capitalized first character and lowercase remainder. For properties, it produces a fully lowercase form to preserve relational readability conventions.

After normalization, underscores are replaced with spaces to produce a human-readable label.

Examples

‘PERSON’ -> ‘Person’
‘person_type’ -> ‘Person type’
‘PERSON TYPE’ -> ‘Person type’
is_property=True: ‘has_name’ -> ‘has name’

Parameters:

label – Raw entity identifier string.
is_property – If True, formats the label as a property name (fully lowercase). If False, formats it as a class/individual-style label.

Returns:

label string with: - consistent casing rules applied - underscores replaced with spaces - improved human readability

Return type:

A cleaned rdfs

plan_decompose(query: str)[source]

load_text(source: str | pathlib.Path, file_type: str | None = None) → str[source]

Load text from various sources (files or raw text).

Supports multiple file formats: - Plain text: .txt - PDF: .pdf - Word documents: .docx, .doc - Rich Text: .rtf - HTML: .html, .htm - Raw text strings

Parameters:

source – File path (str or Path) or raw text string.
file_type – Optional file extension (e.g., ‘.pdf’, ‘.txt’). If not provided, will be auto-detected from the source.

Returns:

Extracted text content as a string.

Raises:

ValueError – If the file type is not supported or content cannot be extracted.
FileNotFoundError – If the specified file does not exist.

Examples

# Load from file text = extractor.load_text(‘document.pdf’) text = extractor.load_text(‘/path/to/file.docx’)

# Load from raw text text = extractor.load_text(‘This is raw text input’)

# Specify file type explicitly text = extractor.load_text(‘file.txt’, file_type=’.txt’)

get_supported_formats() → list[source]

Get list of supported file formats.

Returns:: List of supported file extensions (e.g., [‘.txt’, ‘.pdf’, ‘.docx’, …])

configure_chunking(chunk_size: int = None, overlap: int = None, strategy: str = None, auto_chunk_threshold: int = None, summarization_threshold: int = None, max_summary_length: int = None)[source]

Configure text chunking settings for handling large documents.

This method allows fine-tuning of how large texts are split into manageable pieces for LLM processing.

Parameters:

chunk_size – Maximum characters per chunk (default: 3000, ~750 tokens).
overlap – Characters to overlap between chunks (default: 200).
strategy – Chunking strategy - “sentence”, “paragraph”, or “fixed”.
auto_chunk_threshold – Character threshold for automatic chunking (default: 4000). Texts larger than this will be automatically chunked.
summarization_threshold – Character threshold for using summarization in clustering (default: 8000). When text exceeds this, summaries are generated to provide context for clustering operations.
max_summary_length – Maximum length of summaries used for clustering context (default: 3000, ~750 tokens).

Example

# Configure for a model with smaller context window extractor.configure_chunking(chunk_size=2000, overlap=150, strategy=”sentence”)

# Configure for larger context window (e.g., GPT-4-turbo) extractor.configure_chunking(chunk_size=8000, overlap=500)

# Configure summarization thresholds extractor.configure_chunking(summarization_threshold=10000, max_summary_length=4000)

configure_chunking_for_model(max_context_tokens: int, prompt_overhead_tokens: int = 1500)[source]

Automatically configure chunking based on model specifications.

This is a convenience method that calculates optimal chunk size based on your model’s context window.

Parameters:

max_context_tokens – Maximum context window of your model (e.g., 4096 for GPT-3.5).
prompt_overhead_tokens – Estimated tokens used by prompts/few-shot examples.

Example

# For GPT-3.5-turbo (4K context) extractor.configure_chunking_for_model(4096, 1000)

# For GPT-4 (8K context) extractor.configure_chunking_for_model(8192, 1500)

# For GPT-4-turbo (128K context) extractor.configure_chunking_for_model(128000, 2000)

should_chunk_text(text: str) → bool[source]

Determine if text should be chunked based on its size.

Parameters:: text – The text to check.
Returns:: True if text exceeds the auto_chunk_threshold.

chunk_text(text: str) → List[str][source]

Split text into chunks for processing.

Parameters:: text – The text to chunk.
Returns:: List of text chunks.

get_chunking_info(text: str) → dict[source]

Get information about how a text would be chunked.

Parameters:: text – The text to analyze.
Returns:: Dictionary with chunking information including number of chunks, chunk sizes, etc.

should_use_summarization(text: str) → bool[source]

Determine if text is large enough to require summarization for clustering.

Parameters:: text – The text to check.
Returns:: True if text exceeds the summarization_threshold.

summarize_chunk(chunk: str) → str[source]

Generate a summary of a text chunk that preserves key entities and relationships.

Parameters:: chunk – Text chunk to summarize.
Returns:: Summary of the chunk.

summarize_chunks(chunks: List[str]) → List[str][source]

Generate summaries for multiple text chunks.

Parameters:: chunks – List of text chunks.
Returns:: List of chunk summaries.

create_combined_summary(chunk_summaries: List[str]) → str[source]

Combine multiple chunk summaries into a unified summary for clustering context.

Parameters:: chunk_summaries – List of summaries from individual chunks.
Returns:: A combined summary suitable for clustering operations.

get_clustering_context(text: str, chunks: List[str] = None) → str[source]

Get appropriate context for clustering operations based on text size.

For small texts, returns the text directly (possibly truncated). For large texts, generates a summary to use as clustering context.

Parameters:

text – The full text.
chunks – Optional pre-computed chunks of the text.

Returns:

Context string suitable for clustering operations.

clear_summary_cache()[source]: Clear the chunk summary cache.

configure_cache(max_cache_size: int = None)[source]

Configure summary cache settings.

Parameters:: max_cache_size – Maximum number of summaries to cache (default: 1000). Set to 0 to disable caching.

Example

# Increase cache size for large batch processing extractor.configure_cache(max_cache_size=5000)

# Disable caching to minimize memory usage extractor.configure_cache(max_cache_size=0)

filter_entities(entities: List[str], text: str, use_summary: bool = None) → list[source]

Deduplicate entities by removing redundant near-duplicates.

For large texts, automatically uses summarization to provide context that fits within token limits.

Parameters:

entities – List of extracted entities.
text – Original text context (or pre-computed summary).
use_summary – Whether to use summary-based deduplication. - None (default): Auto-detect based on text size. - True: Force summary-based deduplication. - False: Use direct text (may fail for large texts).

Returns:

Dictionary mapping original entity names to their canonical names.

cluster_types(types: List[str], text: str, use_summary: bool = None) → dict[source]

Cluster entity types to identify and merge duplicates.

For large texts, automatically uses summarization to provide context that fits within token limits.

Parameters:

types – List of extracted entity types.
text – Original text context (or pre-computed summary).
use_summary – Whether to use summary-based clustering. - None (default): Auto-detect based on text size. - True: Force summary-based clustering. - False: Use direct text (may fail for large texts).

Returns:

Dictionary mapping original type names to their canonical names.

cluster_relations(relations: List[str], text: str, use_summary: bool = None) → dict[source]

Cluster relations to identify and merge duplicates.

For large texts, automatically uses summarization to provide context that fits within token limits.

Parameters:

relations – List of extracted relations.
text – Original text context (or pre-computed summary).
use_summary – Whether to use summary-based clustering. - None (default): Auto-detect based on text size. - True: Force summary-based clustering. - False: Use direct text (may fail for large texts).

Returns:

Dictionary mapping original relation names to their canonical names.

check_coherence(triples: List[tuple], text: str, task_instructions: str = None, batch_size: int = 50, threshold: int = 3, use_summary: bool = None) → List[tuple][source]

Check coherence of extracted triples and filter out low-quality ones.

For large texts, automatically uses summarization to provide context that fits within token limits.

Parameters:

triples – List of extracted triples.
text – Original text context (or pre-computed summary).
task_instructions – Optional instructions to guide coherence checking.
batch_size – Number of triples to check per batch.
threshold – Minimum coherence score (1-5) to keep a triple.
use_summary – Whether to use summary for coherence checking. - None (default): Auto-detect based on text size. - True: Force summary-based checking. - False: Use direct text (may fail for large texts).

Returns:

List of coherent triples that passed the threshold.

get_corresponding_literal(literal_value: str) → owlapy.owl_literal.OWLLiteral[source]: Convert a string literal value to an OWLLiteral with the appropriate datatype.

abstractmethod generate_ontology(text: str | pathlib.Path, ontology_type: str = 'open', **kwargs) → owlapy.owl_ontology.Ontology[source]

Generate an ontology from the given text or file. Must be implemented by subclasses.

Parameters:

text – Input text or file path to extract ontology from. Supports files: .txt, .pdf, .docx, .doc, .rtf, .html, .htm
ontology_type (str) – The ontology type to use. Options are: 1. ‘domain’: Focused on a specific domain (e.g., healthcare, finance), 2. ‘cross-domain’: Spans multiple related domains, 3. ‘enterprise’: Tailored for organizational knowledge representation, 4. ‘open’: General-purpose ontology covering a wide range of topics, similar to Wikidata.
**kwargs – Additional arguments specific to the extractor type.

Returns:

Generated Ontology object.

get_rdfs_label_axiom(entity_iri: owlapy.iri.IRI, label: str) → owlapy.owl_axiom.OWLAnnotationAssertionAxiom[source]

Create an RDFS label annotation assertion axiom for an entity.

Utilizes the standard RDFS label IRI (http://www.w3.org/2000/01/rdf-schema#label) to assign a human-readable name to the entity.

Parameters:

entity_iri (IRI) – The IRI of the entity to receive the label.
label (str) – The text for the rdfs:label.

Returns:

An annotation assertion axiom for rdfs:label.

Return type:

OWLAnnotationAssertionAxiom

generate_batch_rdfs_comment_axioms(entities_meta: list[tuple[owlapy.iri.IRI, Literal['class', 'property', 'individual']]], context: str) → list[owlapy.owl_axiom.OWLAnnotationAssertionAxiom][source]

Generate RDFS comment annotation assertion axioms for multiple OWL entities.

This method sends a batch of (entity IRI, entity type) pairs to an LLM, which returns (IRI, comment) pairs. Each generated comment is wrapped into an rdfs:comment OWL annotation assertion axiom.

Only comments for entities whose IRIs are present in entities_meta are converted into axioms. Returned IRIs are validated against the input entity IRIs after converting them to strings.

Parameters:

entities_meta –

List of tuples:

(
entity_iri (IRI), entity_type (str): ‘class’ | ‘property’ | ‘individual’

)
context – Shared ontology/domain context used to ground generated comments.

Returns:

One annotation assertion axiom per valid entity that received a comment. Returns an empty list if the LLM call for this batch fails, so a single failed batch does not abort the rest of the pipeline (e.g. already-generated entities/triples/labels).

Return type:

List[OWLAnnotationAssertionAxiom]