Metadata

Bases: ServiceClient

Source code in blue/metadata.py

class MetaData(ServiceClient):

    def __init__(self, properties=None):
        self.name = "metadata"
        super().__init__(self.name, properties=properties)
        self._init_metadata_properties()

    ###### initialization
    def _init_metadata_properties(self):

        self.properties['openai.api'] = 'ChatCompletion'
        self.properties['openai.model'] = "gpt-4o"
        self.properties['input_json'] = "[{\"role\": \"user\"}]"
        self.properties['input_context'] = "$[0]"
        self.properties['input_context_field'] = "content"
        self.properties['input_field'] = "messages"
        self.properties['input_template'] = "${input}"
        self.properties['output_path'] = '$.choices[0].message.content'
        self.properties['openai.stream'] = False
        self.properties['openai.max_tokens'] = 300

        # prefix for service specific properties
        self.properties['service_prefix'] = 'openai'
        # service_url, set as default
        self.properties["service_url"] = PROPERTIES["services.openai.service_url"]

        # transformations
        self.properties['output_transformations'] = [{"transformation": "replace", "from": "```", "to": ""}, {"transformation": "replace", "from": "json", "to": ""}]
        self.properties['output_strip'] = True

        self.properties['enable_entity_description_generation'] = True
        self.properties['enable_attribute_description_generation'] = True

        # Description aggregation from children
        self.properties['aggregation_prompt'] = AGGREGATION_PROMPT
        self.properties['enable_database_description_generation'] = True
        self.properties['enable_collection_description_generation'] = True

    def build_entity_description_prompt(self, entity_obj, attributes):
        """
        Build a prompt for generating an entity description using an LLM.

        Constructs a structured text prompt containing entity metadata and attribute
        information, suitable for guiding an LLM to produce a JSON-formatted
        description of the entity and its attributes.

        Parameters:
            entity_obj (dict): A dictionary representing the entity, from
                the data registry.
            attributes (list[dict]): A list of attribute definitions.

        Returns:
            str: A formatted multi-line string prompt, instructing the LLM to produce
            a JSON object with:
                - "table_description": Human-readable description of the entity.
                - "attributes": Mapping of attribute names to their descriptions.

        """

        # Extract basic info
        name = entity_obj.get("name", "Unknown")
        scope = entity_obj.get("scope", "Unknown")
        etype = entity_obj.get("type", "Unknown")

        attr_lines = []

        for attr in attributes:
            attr_properties = attr.get("properties", {})
            attr_properties_info = attr_properties.get("info", {})
            attr_type = attr_properties_info.get("type", "unknown")

            attr_name = attr.get("name")
            attr_stats = attr_properties.get("stats", {})

            sample_values = attr_stats.get("sample_values", [])

            attr_lines.append(f"- {attr_name} ({attr_type}), samples: {', '.join(map(str, sample_values[:3]))}")

        # Build the final prompt
        prompt = f"""
        You are given a database entity definition with its attributes and metadata.
        Your task is to generate a structured JSON output with:
        1. A concise human-readable description of what this table/entity represents.
        2. Concise descriptions of each attribute.

        Entity Name: {name}
        Scope: {scope}
        Type: {etype}

        Attributes:
        {chr(10).join(attr_lines)}

        Output JSON format (do not include extra commentary, only valid JSON):

        {{
        "table_description": "string",
        "attributes": {{
            "attr_name": "description of attribute",
            ...
        }}
        }}
        """
        return prompt

    def enrich_entity(self, entity, attributes):
        """
        Generate an enriched description for an entity using its attributes.

        Builds a prompt from the entity and its attributes, then calls the
        external LLM API to produce the enriched description.

        Parameters:
            entity (dict): The entity metadata to enrich.
            attributes (dict): Attribute data associated with the entity.

        Returns:
            str: Enriched description text generated by the API.
        """
        entity_prompt = self.build_entity_description_prompt(entity, attributes)
        return self.execute_api_call(entity_prompt, properties=self.properties, additional_data={})

    def collect_source_metadata(self, data_registry, source, recursive=False, rebuild=False):
        """
        Collect and optionally recursively enrich metadata for a data source.

        If recursive is True, iterates through all databases under the source
        and collects/enriches their metadata.

        Parameters:
            data_registry (DataRegistry): Registry instance for metadata access/storage.
            source (str): Identifier of the data source.
            recursive (bool, optional): Whether to include child databases. Defaults to False.
            rebuild (bool, optional): Whether to regenerate existing descriptions. Defaults to False.

        Returns:
            None
        """
        if recursive:
            databases = data_registry.get_source_databases(source)
            for database in databases:
                self.collect_source_database_metadata(data_registry, source, database, recursive=recursive, rebuild=rebuild)
        return

    def collect_source_database_metadata(self, data_registry, source, database, recursive=False, rebuild=False):
        """
        Collect and enrich metadata for a database within a data source.

        This method checks whether the database already has a description. If not,
        it uses available metadata and collection descriptions to generate an
        enriched description (via `enrich_database_description`) and stores it
        back into the data registry. Optionally, it can also recurse into
        collections to collect their metadata.

        Parameters:
            data_registry (DataRegistry): The registry object that manages sources,
                databases, collections, and metadata.
            source (str): Identifier for the data source.
            database (str): Name of the database to collect metadata for.
            recursive (bool, optional): If True, also collect metadata for all
                collections within the database. Defaults to False.
            rebuild (bool, optional): If True, forces metadata to be rebuilt or
                refreshed even if it already exists. Defaults to False.

        Returns:
            None

        """
        collections = data_registry.get_source_database_collections(source, database)
        collection_descriptions = {}

        if self.properties.get('enable_database_description_generation', True):
            current_description = data_registry.get_source_database_description(source, database)
            if not current_description or current_description.strip() == "":

                database_metadata = data_registry.get_source_database_property(source, database, "metadata")

                if not database_metadata:
                    database_metadata = {"name": database, "type": "database"}

                for collection in collections:
                    collection_name = collection.get("name")
                    collection_desc = collection.get("description")
                    collection_descriptions[collection_name] = collection_desc

                database_desc = self.enrich_database_description(database, collection_descriptions, database_metadata)

                data_registry.set_source_database_description(source, database, database_desc, rebuild=rebuild)

        if recursive:
            for collection in collections:
                self.collect_source_database_collection_metadata(data_registry, source, database, collection, recursive=recursive, rebuild=rebuild)

        return

    def collect_source_database_collection_metadata(self, data_registry, source, database, collection, recursive=False, rebuild=False):
        """
        Collect and enrich metadata for a specific collection and its entities within a database.

        For each entity in the collection, generates enriched table and attribute descriptions
        using the LLM-based enrichment process, and stores them in the data registry if missing.
        Optionally, also generates a collection-level description.

        Parameters:
            data_registry (DataRegistry): Registry instance for accessing and storing metadata.
            source (str): Identifier of the data source.
            database (str): Name of the database containing the collection.
            collection (str): Name of the collection to process.
            recursive (bool, optional): Whether to process nested collections or entities. Defaults to False.
            rebuild (bool, optional): Whether to regenerate existing descriptions. Defaults to False.

        Returns:
            None
        """

        entities = data_registry.get_source_database_collection_entities(source, database, collection)

        entity_descriptions = {}
        for entity in entities:
            entity_name = entity.get("name")

            attributes = data_registry.get_source_database_collection_entity_attributes(source, database, collection, entity_name)

            entity_attribute_description = self.enrich_entity(entity, attributes)

            try:
                parsed = json_utils.safe_json_parse(entity_attribute_description)
                if not parsed:
                    logging.warning(f"Entity {entity} returned invalid or empty JSON.")
                    continue
            except json.JSONDecodeError:
                logging.warning("LLM did not return valid JSON. Skipping entity enrichment.")
                parsed = {}

            table_desc = parsed.get("table_description", "")
            attribute_descs = parsed.get("attributes", {})
            entity_descriptions[entity_name] = table_desc

            if self.properties.get('enable_entity_description_generation', True):
                current_description = data_registry.get_source_database_collection_entity_description(source, database, collection, entity_name)

                if not current_description or current_description.strip() == "":
                    data_registry.set_source_database_collection_entity_description(source, database, collection, entity_name, table_desc, rebuild=rebuild)

            if self.properties.get('enable_attribute_description_generation', True):
                for attr, desc in attribute_descs.items():
                    current_description = data_registry.get_source_database_collection_entity_attribute_description(source, database, collection, entity_name, attr)
                    if not current_description or current_description.strip() == "":
                        data_registry.set_source_database_collection_entity_attribute_description(source, database, collection, entity_name, attr, desc, rebuild=rebuild)

        if self.properties.get('enable_collection_description_generation', True):
            current_description = data_registry.get_source_database_collection_description(source, database, collection)
            if not current_description or current_description.strip() == "":

                collection_metadata = data_registry.get_source_database_collection_property(source, database, collection, "metadata")

                if not collection_metadata:
                    collection_metadata = {"name": collection, "type": "collection"}

                collection_desc = self.enrich_collection_description(database, entity_descriptions, collection_metadata)

                data_registry.set_source_database_collection_description(source, database, collection, collection_desc, rebuild=rebuild)

    ###### Aggregation
    def build_collection_description_prompt(self, collection_name, entity_descriptions, collection_metadata):
        """
        Build a prompt string for generating or enriching a collection description.

        Constructs a formatted text prompt using entity-level descriptions and
        metadata, suitable for passing to an LLM or enrichment API.

        Parameters:
            collection_name (str): The name of the collection.
            entity_descriptions (dict): Mapping of entity names to their descriptions.
            collection_metadata (dict or str): Additional metadata for the collection.

        Returns:
            str: A formatted prompt string for collection-level description enrichment.
        """
        child_descriptions = [f"{name}: {desc}" for name, desc in entity_descriptions.items() if desc]
        if not child_descriptions:
            child_descriptions = ["No entity descriptions available"]

        return self.properties['aggregation_prompt'].format(
            child_type='entity',
            parent_type='collection',
            child_descriptions='\n'.join(child_descriptions),
            parent_metadata=f"Collection name: {collection_name}\nMetadata: {collection_metadata}",
        )

    def build_database_description_prompt(self, database_name, collection_descriptions, database_metadata):
        """
        Build a prompt string for generating or enriching a database description.

        Constructs a formatted text prompt using collection-level descriptions and
        metadata, suitable for passing to an LLM or enrichment API.

        Parameters:
            database_name (str): The name of the database.
            collection_descriptions (dict): Mapping of collection names to their descriptions.
            database_metadata (dict or str): Additional metadata for the database.

        Returns:
            str: A formatted prompt string for database-level description enrichment.
        """

        child_descriptions = [f"{name}: {desc}" for name, desc in collection_descriptions.items() if desc]
        if not child_descriptions:
            child_descriptions = ["No collection descriptions available"]

        return self.properties['aggregation_prompt'].format(
            child_type='collection', parent_type='database', child_descriptions='\n'.join(child_descriptions), parent_metadata=f"Database name: {database_name}\nMetadata: {database_metadata}"
        )

    def enrich_collection_description(self, collection_name, entity_descriptions, collection_metadata):
        """
        Enrich a collection description using LLM.

        Builds a prompt from the provided entity descriptions and metadata, then
        executes an LLM call to generate or refine the collection-level description.

        Parameters:
            collection_name (str): The name of the collection.
            entity_descriptions (dict): Mapping of entity names to their descriptions.
            collection_metadata (dict or str): Additional metadata for the collection.

        Returns:
            Any: The enriched collection description, as returned by the LLM.
        """
        prompt = self.build_collection_description_prompt(collection_name, entity_descriptions, collection_metadata)
        return self.execute_api_call(prompt, properties=self.properties, additional_data={})

    def enrich_database_description(self, database_name, collection_descriptions, database_metadata):
        """
        Enrich a database description using LLM.

        Builds a prompt from the provided collection descriptions and metadata, then
        executes an API call to generate or refine the database-level description.

        Parameters:
            database_name (str): The name of the database.
            collection_descriptions (dict): Mapping of collection names to their descriptions.
            database_metadata (dict or str): Additional metadata for the database.

        Returns:
            Any: The enriched database description.
        """

        prompt = self.build_database_description_prompt(database_name, collection_descriptions, database_metadata)
        return self.execute_api_call(prompt, properties=self.properties, additional_data={})

build_collection_description_prompt(collection_name, entity_descriptions, collection_metadata)

Build a prompt string for generating or enriching a collection description.

Constructs a formatted text prompt using entity-level descriptions and metadata, suitable for passing to an LLM or enrichment API.

Parameters:

Name	Type	Description	Default
collection_name	str	The name of the collection.	required
entity_descriptions	dict	Mapping of entity names to their descriptions.	required
collection_metadata	dict or str	Additional metadata for the collection.	required

Returns:

Name	Type	Description
str		A formatted prompt string for collection-level description enrichment.

Source code in blue/metadata.py

def build_collection_description_prompt(self, collection_name, entity_descriptions, collection_metadata):
    """
    Build a prompt string for generating or enriching a collection description.

    Constructs a formatted text prompt using entity-level descriptions and
    metadata, suitable for passing to an LLM or enrichment API.

    Parameters:
        collection_name (str): The name of the collection.
        entity_descriptions (dict): Mapping of entity names to their descriptions.
        collection_metadata (dict or str): Additional metadata for the collection.

    Returns:
        str: A formatted prompt string for collection-level description enrichment.
    """
    child_descriptions = [f"{name}: {desc}" for name, desc in entity_descriptions.items() if desc]
    if not child_descriptions:
        child_descriptions = ["No entity descriptions available"]

    return self.properties['aggregation_prompt'].format(
        child_type='entity',
        parent_type='collection',
        child_descriptions='\n'.join(child_descriptions),
        parent_metadata=f"Collection name: {collection_name}\nMetadata: {collection_metadata}",
    )

build_database_description_prompt(database_name, collection_descriptions, database_metadata)

Build a prompt string for generating or enriching a database description.

Constructs a formatted text prompt using collection-level descriptions and metadata, suitable for passing to an LLM or enrichment API.

Parameters:

Name	Type	Description	Default
database_name	str	The name of the database.	required
collection_descriptions	dict	Mapping of collection names to their descriptions.	required
database_metadata	dict or str	Additional metadata for the database.	required

Returns:

Name	Type	Description
str		A formatted prompt string for database-level description enrichment.

Source code in blue/metadata.py

def build_database_description_prompt(self, database_name, collection_descriptions, database_metadata):
    """
    Build a prompt string for generating or enriching a database description.

    Constructs a formatted text prompt using collection-level descriptions and
    metadata, suitable for passing to an LLM or enrichment API.

    Parameters:
        database_name (str): The name of the database.
        collection_descriptions (dict): Mapping of collection names to their descriptions.
        database_metadata (dict or str): Additional metadata for the database.

    Returns:
        str: A formatted prompt string for database-level description enrichment.
    """

    child_descriptions = [f"{name}: {desc}" for name, desc in collection_descriptions.items() if desc]
    if not child_descriptions:
        child_descriptions = ["No collection descriptions available"]

    return self.properties['aggregation_prompt'].format(
        child_type='collection', parent_type='database', child_descriptions='\n'.join(child_descriptions), parent_metadata=f"Database name: {database_name}\nMetadata: {database_metadata}"
    )

build_entity_description_prompt(entity_obj, attributes)

Build a prompt for generating an entity description using an LLM.

Constructs a structured text prompt containing entity metadata and attribute information, suitable for guiding an LLM to produce a JSON-formatted description of the entity and its attributes.

Parameters:

Name	Type	Description	Default
entity_obj	dict	A dictionary representing the entity, from the data registry.	required
attributes	list[dict]	A list of attribute definitions.	required

Returns:

Name	Type	Description
str		A formatted multi-line string prompt, instructing the LLM to produce
		a JSON object with: - "table_description": Human-readable description of the entity. - "attributes": Mapping of attribute names to their descriptions.

Source code in blue/metadata.py

def build_entity_description_prompt(self, entity_obj, attributes):
    """
    Build a prompt for generating an entity description using an LLM.

    Constructs a structured text prompt containing entity metadata and attribute
    information, suitable for guiding an LLM to produce a JSON-formatted
    description of the entity and its attributes.

    Parameters:
        entity_obj (dict): A dictionary representing the entity, from
            the data registry.
        attributes (list[dict]): A list of attribute definitions.

    Returns:
        str: A formatted multi-line string prompt, instructing the LLM to produce
        a JSON object with:
            - "table_description": Human-readable description of the entity.
            - "attributes": Mapping of attribute names to their descriptions.

    """

    # Extract basic info
    name = entity_obj.get("name", "Unknown")
    scope = entity_obj.get("scope", "Unknown")
    etype = entity_obj.get("type", "Unknown")

    attr_lines = []

    for attr in attributes:
        attr_properties = attr.get("properties", {})
        attr_properties_info = attr_properties.get("info", {})
        attr_type = attr_properties_info.get("type", "unknown")

        attr_name = attr.get("name")
        attr_stats = attr_properties.get("stats", {})

        sample_values = attr_stats.get("sample_values", [])

        attr_lines.append(f"- {attr_name} ({attr_type}), samples: {', '.join(map(str, sample_values[:3]))}")

    # Build the final prompt
    prompt = f"""
    You are given a database entity definition with its attributes and metadata.
    Your task is to generate a structured JSON output with:
    1. A concise human-readable description of what this table/entity represents.
    2. Concise descriptions of each attribute.

    Entity Name: {name}
    Scope: {scope}
    Type: {etype}

    Attributes:
    {chr(10).join(attr_lines)}

    Output JSON format (do not include extra commentary, only valid JSON):

    {{
    "table_description": "string",
    "attributes": {{
        "attr_name": "description of attribute",
        ...
    }}
    }}
    """
    return prompt

collect_source_database_collection_metadata(data_registry, source, database, collection, recursive=False, rebuild=False)

Collect and enrich metadata for a specific collection and its entities within a database.

For each entity in the collection, generates enriched table and attribute descriptions using the LLM-based enrichment process, and stores them in the data registry if missing. Optionally, also generates a collection-level description.

Parameters:

Name	Type	Description	Default
data_registry	DataRegistry	Registry instance for accessing and storing metadata.	required
source	str	Identifier of the data source.	required
database	str	Name of the database containing the collection.	required
collection	str	Name of the collection to process.	required
recursive	bool	Whether to process nested collections or entities. Defaults to False.	False
rebuild	bool	Whether to regenerate existing descriptions. Defaults to False.	False

Returns:

Type	Description
	None

Source code in blue/metadata.py

def collect_source_database_collection_metadata(self, data_registry, source, database, collection, recursive=False, rebuild=False):
    """
    Collect and enrich metadata for a specific collection and its entities within a database.

    For each entity in the collection, generates enriched table and attribute descriptions
    using the LLM-based enrichment process, and stores them in the data registry if missing.
    Optionally, also generates a collection-level description.

    Parameters:
        data_registry (DataRegistry): Registry instance for accessing and storing metadata.
        source (str): Identifier of the data source.
        database (str): Name of the database containing the collection.
        collection (str): Name of the collection to process.
        recursive (bool, optional): Whether to process nested collections or entities. Defaults to False.
        rebuild (bool, optional): Whether to regenerate existing descriptions. Defaults to False.

    Returns:
        None
    """

    entities = data_registry.get_source_database_collection_entities(source, database, collection)

    entity_descriptions = {}
    for entity in entities:
        entity_name = entity.get("name")

        attributes = data_registry.get_source_database_collection_entity_attributes(source, database, collection, entity_name)

        entity_attribute_description = self.enrich_entity(entity, attributes)

        try:
            parsed = json_utils.safe_json_parse(entity_attribute_description)
            if not parsed:
                logging.warning(f"Entity {entity} returned invalid or empty JSON.")
                continue
        except json.JSONDecodeError:
            logging.warning("LLM did not return valid JSON. Skipping entity enrichment.")
            parsed = {}

        table_desc = parsed.get("table_description", "")
        attribute_descs = parsed.get("attributes", {})
        entity_descriptions[entity_name] = table_desc

        if self.properties.get('enable_entity_description_generation', True):
            current_description = data_registry.get_source_database_collection_entity_description(source, database, collection, entity_name)

            if not current_description or current_description.strip() == "":
                data_registry.set_source_database_collection_entity_description(source, database, collection, entity_name, table_desc, rebuild=rebuild)

        if self.properties.get('enable_attribute_description_generation', True):
            for attr, desc in attribute_descs.items():
                current_description = data_registry.get_source_database_collection_entity_attribute_description(source, database, collection, entity_name, attr)
                if not current_description or current_description.strip() == "":
                    data_registry.set_source_database_collection_entity_attribute_description(source, database, collection, entity_name, attr, desc, rebuild=rebuild)

    if self.properties.get('enable_collection_description_generation', True):
        current_description = data_registry.get_source_database_collection_description(source, database, collection)
        if not current_description or current_description.strip() == "":

            collection_metadata = data_registry.get_source_database_collection_property(source, database, collection, "metadata")

            if not collection_metadata:
                collection_metadata = {"name": collection, "type": "collection"}

            collection_desc = self.enrich_collection_description(database, entity_descriptions, collection_metadata)

            data_registry.set_source_database_collection_description(source, database, collection, collection_desc, rebuild=rebuild)

collect_source_database_metadata(data_registry, source, database, recursive=False, rebuild=False)

Collect and enrich metadata for a database within a data source.

This method checks whether the database already has a description. If not, it uses available metadata and collection descriptions to generate an enriched description (via enrich_database_description) and stores it back into the data registry. Optionally, it can also recurse into collections to collect their metadata.

Parameters:

Name	Type	Description	Default
data_registry	DataRegistry	The registry object that manages sources, databases, collections, and metadata.	required
source	str	Identifier for the data source.	required
database	str	Name of the database to collect metadata for.	required
recursive	bool	If True, also collect metadata for all collections within the database. Defaults to False.	False
rebuild	bool	If True, forces metadata to be rebuilt or refreshed even if it already exists. Defaults to False.	False

Returns:

Type	Description
	None

Source code in blue/metadata.py

def collect_source_database_metadata(self, data_registry, source, database, recursive=False, rebuild=False):
    """
    Collect and enrich metadata for a database within a data source.

    This method checks whether the database already has a description. If not,
    it uses available metadata and collection descriptions to generate an
    enriched description (via `enrich_database_description`) and stores it
    back into the data registry. Optionally, it can also recurse into
    collections to collect their metadata.

    Parameters:
        data_registry (DataRegistry): The registry object that manages sources,
            databases, collections, and metadata.
        source (str): Identifier for the data source.
        database (str): Name of the database to collect metadata for.
        recursive (bool, optional): If True, also collect metadata for all
            collections within the database. Defaults to False.
        rebuild (bool, optional): If True, forces metadata to be rebuilt or
            refreshed even if it already exists. Defaults to False.

    Returns:
        None

    """
    collections = data_registry.get_source_database_collections(source, database)
    collection_descriptions = {}

    if self.properties.get('enable_database_description_generation', True):
        current_description = data_registry.get_source_database_description(source, database)
        if not current_description or current_description.strip() == "":

            database_metadata = data_registry.get_source_database_property(source, database, "metadata")

            if not database_metadata:
                database_metadata = {"name": database, "type": "database"}

            for collection in collections:
                collection_name = collection.get("name")
                collection_desc = collection.get("description")
                collection_descriptions[collection_name] = collection_desc

            database_desc = self.enrich_database_description(database, collection_descriptions, database_metadata)

            data_registry.set_source_database_description(source, database, database_desc, rebuild=rebuild)

    if recursive:
        for collection in collections:
            self.collect_source_database_collection_metadata(data_registry, source, database, collection, recursive=recursive, rebuild=rebuild)

    return

collect_source_metadata(data_registry, source, recursive=False, rebuild=False)

Collect and optionally recursively enrich metadata for a data source.

If recursive is True, iterates through all databases under the source and collects/enriches their metadata.

Parameters:

Name	Type	Description	Default
data_registry	DataRegistry	Registry instance for metadata access/storage.	required
source	str	Identifier of the data source.	required
recursive	bool	Whether to include child databases. Defaults to False.	False
rebuild	bool	Whether to regenerate existing descriptions. Defaults to False.	False

Returns:

Type	Description
	None

Source code in blue/metadata.py

def collect_source_metadata(self, data_registry, source, recursive=False, rebuild=False):
    """
    Collect and optionally recursively enrich metadata for a data source.

    If recursive is True, iterates through all databases under the source
    and collects/enriches their metadata.

    Parameters:
        data_registry (DataRegistry): Registry instance for metadata access/storage.
        source (str): Identifier of the data source.
        recursive (bool, optional): Whether to include child databases. Defaults to False.
        rebuild (bool, optional): Whether to regenerate existing descriptions. Defaults to False.

    Returns:
        None
    """
    if recursive:
        databases = data_registry.get_source_databases(source)
        for database in databases:
            self.collect_source_database_metadata(data_registry, source, database, recursive=recursive, rebuild=rebuild)
    return

enrich_collection_description(collection_name, entity_descriptions, collection_metadata)

Enrich a collection description using LLM.

Builds a prompt from the provided entity descriptions and metadata, then executes an LLM call to generate or refine the collection-level description.

Parameters:

Name	Type	Description	Default
collection_name	str	The name of the collection.	required
entity_descriptions	dict	Mapping of entity names to their descriptions.	required
collection_metadata	dict or str	Additional metadata for the collection.	required

Returns:

Name	Type	Description
Any		The enriched collection description, as returned by the LLM.

Source code in blue/metadata.py

def enrich_collection_description(self, collection_name, entity_descriptions, collection_metadata):
    """
    Enrich a collection description using LLM.

    Builds a prompt from the provided entity descriptions and metadata, then
    executes an LLM call to generate or refine the collection-level description.

    Parameters:
        collection_name (str): The name of the collection.
        entity_descriptions (dict): Mapping of entity names to their descriptions.
        collection_metadata (dict or str): Additional metadata for the collection.

    Returns:
        Any: The enriched collection description, as returned by the LLM.
    """
    prompt = self.build_collection_description_prompt(collection_name, entity_descriptions, collection_metadata)
    return self.execute_api_call(prompt, properties=self.properties, additional_data={})

enrich_database_description(database_name, collection_descriptions, database_metadata)

Enrich a database description using LLM.

Builds a prompt from the provided collection descriptions and metadata, then executes an API call to generate or refine the database-level description.

Parameters:

Name	Type	Description	Default
database_name	str	The name of the database.	required
collection_descriptions	dict	Mapping of collection names to their descriptions.	required
database_metadata	dict or str	Additional metadata for the database.	required

Returns:

Name	Type	Description
Any		The enriched database description.

Source code in blue/metadata.py

def enrich_database_description(self, database_name, collection_descriptions, database_metadata):
    """
    Enrich a database description using LLM.

    Builds a prompt from the provided collection descriptions and metadata, then
    executes an API call to generate or refine the database-level description.

    Parameters:
        database_name (str): The name of the database.
        collection_descriptions (dict): Mapping of collection names to their descriptions.
        database_metadata (dict or str): Additional metadata for the database.

    Returns:
        Any: The enriched database description.
    """

    prompt = self.build_database_description_prompt(database_name, collection_descriptions, database_metadata)
    return self.execute_api_call(prompt, properties=self.properties, additional_data={})

enrich_entity(entity, attributes)

Generate an enriched description for an entity using its attributes.

Builds a prompt from the entity and its attributes, then calls the external LLM API to produce the enriched description.

Parameters:

Name	Type	Description	Default
entity	dict	The entity metadata to enrich.	required
attributes	dict	Attribute data associated with the entity.	required

Returns:

Name	Type	Description
str		Enriched description text generated by the API.

Source code in blue/metadata.py

def enrich_entity(self, entity, attributes):
    """
    Generate an enriched description for an entity using its attributes.

    Builds a prompt from the entity and its attributes, then calls the
    external LLM API to produce the enriched description.

    Parameters:
        entity (dict): The entity metadata to enrich.
        attributes (dict): Attribute data associated with the entity.

    Returns:
        str: Enriched description text generated by the API.
    """
    entity_prompt = self.build_entity_description_prompt(entity, attributes)
    return self.execute_api_call(entity_prompt, properties=self.properties, additional_data={})

Last update: 2025-10-09