Graph Operations

Build Graph

Graph building consumes usage on the Early Access plan ( Plans & Billing ).

The build_graph function constructs a knowledge graph from a list of files, optionally applying an ontology.

Synchronous

build_graph

def build_graph(
    file_paths: List[str],
    ontology_path: Optional[str] = None,
    refresh_graph: bool = False,
    metadata: Optional[Dict[str, Any]] = None,
) -> List[KnowledgeGraph]:

Builds a knowledge graph from one or more files.

Arguments

NameTypeDescriptionDefault
file_pathsList[str]A list of local file paths to include in the graph.-
ontology_pathOptional[str]The local path to an ontology file to apply.None
refresh_graphboolIf True, forces a re-processing of all files.False
metadataOptional[Dict[str, Any]]A dictionary of metadata to associate with the graph.None

Returns

TypeDescription
List[KnowledgeGraph]A list of KnowledgeGraph objects, one for each input file.

Example

import perseus_client

file_paths = ["./path/to/file1.txt", "./path/to/file2.txt"]

try:
    graphs = perseus_client.build_graph(file_paths, refresh_graph=True)
    print(f"Successfully built {len(graphs)} graphs.")
except Exception as e:
    print(f"An error occurred: {e}")

Asynchronous

build_graph_async

async def build_graph_async(
    file_paths: List[str],
    ontology_path: Optional[str] = None,
    refresh_graph: bool = False,
    metadata: Optional[Dict[str, Any]] = None,
) -> List[KnowledgeGraph]:

Asynchronously builds a knowledge graph from one or more files.

Arguments

NameTypeDescriptionDefault
file_pathsList[str]A list of local file paths to include in the graph.-
ontology_pathOptional[str]The local path to an ontology file to apply.None
refresh_graphboolIf True, forces a re-processing of all files.False
metadataOptional[Dict[str, Any]]A dictionary of metadata to associate with the graph.None

Returns

TypeDescription
List[KnowledgeGraph]A list of KnowledgeGraph objects, one for each input file.

Example

import asyncio
import perseus_client

file_paths = ["./path/to/file1.txt", "./path/to/file2.txt"]

async def main():
    try:
        graphs = await perseus_client.build_graph_async(file_paths, refresh_graph=True)
        print(f"Successfully built {len(graphs)} graphs.")
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        await perseus_client.close_async()

if __name__ == "__main__":
    asyncio.run(main())