curate_gpt package

Subpackages

Submodules

curate_gpt.cli module

Command line interface for curate-gpt.

Module contents

CurateGPT: A framework semi-assisted curation of knowledge bases.

Architecture

  • store: json object stores that allow for embedding based search

  • wrappers: wraps external APIs and data sources for ingest

  • extract: extraction of json objects from LLMs

  • agents: agents that chain together search and generate components

  • formatters: formats data objects for presentation to humans and machine agents

  • app: streamlit application

class curate_gpt.BasicExtractor(schema_proxy=None, model_name='gpt-4o', api_key=None, raise_error_if_unparsable=False, serialization_format='json')

Bases: Extractor

Extractor that is purely example driven.

deserialize(text, format=None, **kwargs)

Deserialize text into an annotated object

Parameters:

text (str)

Return type:

AnnotatedObject

Returns:

deserialize_yaml(text, multiple=False)
Return type:

AnnotatedObject

extract(text, target_class, examples=None, background_text=None, rules=None, min_examples=1, **kwargs)

Schema-guided extraction

Parameters:

  • text (str)

  • kwargs

Return type:

AnnotatedObject

Returns:

model_config = {'protected_namespaces': ()}
model_name: str = 'gpt-4o'
serialization_format: str = 'json'
serialize(ao)
Return type:

str

class curate_gpt.ChromaDBAdapter(path=None, schema_proxy=None, collection=None, _field_names_by_collection=None, client=None, id_field='id', text_lookup='text', id_to_object=<factory>, **_kwargs)

Bases: DBAdapter

An Adapter that wraps a ChromaDB client

client: ClientAPI = None
collection_metadata(collection_name=None, include_derived=False, **kwargs)

Get the metadata for a collection.

Parameters:

collection_name (Optional[str])

Return type:

Optional[CollectionMetadata]

Returns:

Parameters

collections()

Return the names of all collections in the database.

Return type:

Iterator[str]

Returns:

default_max_document_length: ClassVar[int] = 6000
default_model = 'all-MiniLM-L6-v2'
Return type:

Iterator[Tuple[DuckDBSearchResult, Dict, float, Optional[Dict]]]

dump_then_load(collection=None, target=None)

Dump a collection to a file, then load it into another database.

Parameters:

  • collection (Optional[str])

  • target (Optional[DBAdapter])

Returns:

fetch_all_objects_memory_safe(collection=None, batch_size=100, **kwargs)

Fetch all objects from a collection, in batches to avoid memory overload.

Return type:

Iterator[Union[YAMLRoot, BaseModel, Dict, DuckDBSearchResult]]

find(where=None, projection=None, collection=None, **kwargs)

Query the database.

>>> from curate_gpt.store import get_store
>>> store = get_store("chromadb", "db")
>>> objs = list(store.find({"name": "NeuronOfTheForebrain"}, collection="ont_cl"))
Parameters:

  • collection (Optional[str])

  • where (Union[str, YAMLRoot, BaseModel, Dict, DuckDBSearchResult, None])

  • projection (Union[str, List[str], None])

  • kwargs

Return type:

Iterator[Tuple[DuckDBSearchResult, Dict, float, Optional[Dict]]]

Returns:

id_field: str = 'id'
id_to_object: Mapping[str, Union[YAMLRoot, BaseModel, Dict, DuckDBSearchResult]]
insert(objs, **kwargs)

Insert an object or list of objects into the store.

>>> from curate_gpt.store import get_store
>>> store = get_store("in_memory")
>>> store.insert([{"name": "John", "age": 42}], collection="people")
Parameters:
Returns:

list_collection_names()

List all collections in the database.

Return type:

List[str]

Returns:

lookup(id, collection=None, **kwargs)

Lookup an object by its ID.

Parameters:

  • id (str)

  • collection (Optional[str])

Return type:

Union[YAMLRoot, BaseModel, Dict, DuckDBSearchResult]

Returns:

matches(obj, **kwargs)

Query the database for matches to an object.

Parameters:
Return type:

Iterator[Tuple[DuckDBSearchResult, Dict, float, Optional[Dict]]]

Returns:

name: ClassVar[str] = 'chromadb'
peek(collection=None, limit=5, offset=0, **kwargs)

Peek at first N objects in a collection.

Parameters:

  • collection (Optional[str])

  • limit

Return type:

Iterator[Union[YAMLRoot, BaseModel, Dict, DuckDBSearchResult]]

Returns:

remove_collection(collection=None, exists_ok=False, **kwargs)

Remove a collection from the database.

Parameters:

  • collection (Optional[str])

  • exists_ok

Returns:

reset()

Reset/delete the database.

search(text, **kwargs)

Query the database for a text string.

>>> from curate_gpt.store import get_store
>>> store = get_store("chromadb", "db")
>>> for obj, distance, info in store.search("forebrain neurons", collection="ont_cl"):
...     obj_id = obj["id"]
...     # print at precision of 2 decimal places
...     print(f"{obj_id} {distance:.2f}")

...
NeuronOfTheForebrain 0.28
...
Parameters:

  • text (str)

  • collection

  • where

  • kwargs

Return type:

Iterator[Tuple[DuckDBSearchResult, Dict, float, Optional[Dict]]]

Returns:

tuple of object, distance, metadata

set_collection_metadata(collection_name, metadata, **kwargs)

Set the metadata for a collection.

Parameters:
Returns:

text_lookup: Union[str, Callable, None] = 'text'
update(objs, **kwargs)

Update an object or list of objects in the store.

Parameters:
Returns:

update_collection_metadata(collection_name, **kwargs)

Update the metadata for a collection.

Parameters:

  • collection_name (str)

  • kwargs

Return type:

CollectionMetadata

Returns:

upsert(objs, **kwargs)

Update an object or list of objects in the store.

Parameters:
Returns:

class curate_gpt.DBAdapter(path=None, schema_proxy=None, collection=None, _field_names_by_collection=None)

Bases: ABC

Base class for stores.

This base class provides a common interface for a wide variety of document or object stores. The interface is intended to closely mimic the kind of interface found for document stores such as mongoDB or vector databases such as ChromaDB, but the intention is that can be used for SQL databases, SPARQL endpoints, or even file systems.

The store allows for storage and retrieval of objects which are arbitrary dictionary objects, equivalient to a JSON object.

Objects are partitioned into collections, which maps to the equivalent concept in MongoDB and ChromaDB.

>>> from curate_gpt.store import get_store
>>> store = get_store("in_memory")
>>> store.insert({"name": "John", "age": 42}, collection="people")

If you are used to working with MongoDB and ChromaDB APIs directly, one difference is that here we do not provide a separate Collection object, everything is handled through the store object. You can optionally bind a store object to a collection, which effectively gives you a collection object:

>>> from curate_gpt.store import get_store
>>> store = get_store("in_memory")
>>> store.set_collection("people")
>>> store.insert({"name": "John", "age": 42})

TODO: decide if this is the final interface

collection: Optional[str] = None

Default collection

abstract collection_metadata(collection_name=None, include_derived=False, **kwargs)

Get the metadata for a collection.

Parameters:

  • collection_name (Optional[str])

  • include_derived – Include derived metadata, e.g. counts

Return type:

Optional[CollectionMetadata]

Returns:

create_view(view_name, collection, expression, **kwargs)

Create a view in the database.

Todo:

param view:

return:

delete(id, collection=None, **kwargs)

Delete an object by its ID.

Parameters:

  • id (str)

  • collection (Optional[str])

Returns:

dump(collection=None, to_file=None, metadata_to_file=None, format=None, include=None, **kwargs)

Dump the database to a file.

Parameters:

  • collection (Optional[str])

  • kwargs

Returns:

dump_then_load(collection=None, target=None)

Dump a collection to a file, then load it into another database.

Parameters:

  • collection (Optional[str])

  • target (Optional[DBAdapter])

Returns:

abstract fetch_all_objects_memory_safe(collection=None, batch_size=100, **kwargs)

Fetch all objects from a collection, in batches to avoid memory overload.

Return type:

Iterator[Union[YAMLRoot, BaseModel, Dict, DuckDBSearchResult]]

field_names(collection=None)

Return the names of all top level fields in the database for a collection.

Parameters:

collection (Optional[str])

Return type:

List[str]

Returns:

find(where=None, projection=None, collection=None, **kwargs)

Query the database.

>>> from curate_gpt.store import get_store
>>> store = get_store("chromadb", "db")
>>> objs = list(store.find({"name": "NeuronOfTheForebrain"}, collection="ont_cl"))
Parameters:

  • collection (Optional[str])

  • where (Union[str, YAMLRoot, BaseModel, Dict, DuckDBSearchResult, None])

  • projection (Union[str, List[str], None])

  • kwargs

Return type:

Iterator[Tuple[DuckDBSearchResult, Dict, float, Optional[Dict]]]

Returns:

identifier_field(collection=None)
Return type:

str

abstract insert(objs, collection=None, **kwargs)

Insert an object or list of objects into the store.

>>> from curate_gpt.store import get_store
>>> store = get_store("in_memory")
>>> store.insert([{"name": "John", "age": 42}], collection="people")
Parameters:
Returns:

label_field(collection=None)
Return type:

str

abstract list_collection_names()

List all collections in the database.

Return type:

List[str]

Returns:

names of collections

abstract lookup(id, collection=None, **kwargs)

Lookup an object by its ID.

Parameters:

  • id (str)

  • collection (Optional[str])

Return type:

Union[YAMLRoot, BaseModel, Dict, DuckDBSearchResult]

Returns:

lookup_multiple(ids, **kwargs)

Lookup an object by its ID.

Parameters:

  • id

  • collection

Return type:

Iterator[Union[YAMLRoot, BaseModel, Dict, DuckDBSearchResult]]

Returns:

abstract matches(obj, **kwargs)

Query the database for matches to an object.

Parameters:
Return type:

Iterator[Tuple[DuckDBSearchResult, Dict, float, Optional[Dict]]]

Returns:

name: ClassVar[str] = 'base'
path: str = None

Path to a location where the database is stored or disk or the network.

abstract peek(collection=None, limit=5, **kwargs)

Peek at first N objects in a collection.

Parameters:

  • collection (Optional[str])

  • limit

Return type:

Iterator[Union[YAMLRoot, BaseModel, Dict, DuckDBSearchResult]]

Returns:

remove_collection(collection=None, exists_ok=False, **kwargs)

Remove a collection from the database.

Parameters:

collection (Optional[str])

Returns:

schema_proxy: Optional[SchemaProxy] = None

Schema manager

abstract search(text, where=None, collection=None, **kwargs)

Query the database for a text string.

>>> from curate_gpt.store import get_store
>>> store = get_store("chromadb", "db")
>>> for obj, distance, info in store.search("forebrain neurons", collection="ont_cl"):
...     obj_id = obj["id"]
...     # print at precision of 2 decimal places
...     print(f"{obj_id} {distance:.2f}")

...
NeuronOfTheForebrain 0.28
...
Parameters:

  • text (str)

  • collection (Optional[str])

  • where (Union[str, YAMLRoot, BaseModel, Dict, DuckDBSearchResult, None])

  • kwargs

Return type:

Iterator[Tuple[DuckDBSearchResult, Dict, float, Optional[Dict]]]

Returns:

tuple of object, distance, metadata

set_collection(collection)

Set the current collection.

If this is set, then all subsequent operations will be performed on this collection, unless overridden.

This allows the following

>>> from curate_gpt.store import get_store
>>> store = get_store("in_memory")
>>> store.set_collection("people")
>>> store.insert([{"name": "John", "age": 42}])

to be written in place of

>>> from curate_gpt.store import get_store
>>> store = get_store("in_memory")
>>> store.insert([{"name": "John", "age": 42}], collection="people")
Parameters:

collection (str)

Returns:

set_collection_metadata(collection_name, metadata, **kwargs)

Set the metadata for a collection.

>>> from curate_gpt.store import get_store
>>> from curate_gpt.store import CollectionMetadata
>>> store = get_store("in_memory")
>>> cm = CollectionMetadata(name="People", description="People in the database")
>>> store.set_collection_metadata("people", cm)
Parameters:

collection_name (Optional[str])

Returns:

update(objs, collection=None, **kwargs)

Update an object or list of objects in the store.

Parameters:
Returns:

update_collection_metadata(collection_name, **kwargs)

Update the metadata for a collection.

Parameters:

  • collection_name (str)

  • kwargs

Return type:

CollectionMetadata

Returns:

upsert(objs, collection=None, **kwargs)

Upsert an object or list of objects in the store.

Parameters:
Returns:

class curate_gpt.Extractor(schema_proxy=None, model_name=None, api_key=None, raise_error_if_unparsable=False)

Bases: ABC

api_key: str = None
deserialize(text, **kwargs)

Deserialize text into an annotated object

Parameters:

text (str)

Return type:

AnnotatedObject

Returns:

abstract extract(text, target_class, examples=None, **kwargs)

Schema-guided extraction

Parameters:

  • text (str)

  • kwargs

Return type:

AnnotatedObject

Returns:

property model

Get the model

Parameters:

model_name

Returns:

model_name: str = None
property pydantic_root_model: BaseModel
raise_error_if_unparsable: bool = False
schema_proxy: SchemaProxy = None
property schemaview: SchemaView