Building Custom Enrichment Agent Workflows with Download, Enrich, and Publish Steps

The GoogleCloudPlatform/knowledge-catalog repository provides a modular three-stage pipeline—download, enrich, and publish—that extracts BigQuery metadata, generates AI-powered documentation, and registers enriched assets back to Knowledge Catalog.

The GoogleCloudPlatform/knowledge-catalog repository enables data teams to automate metadata documentation through a flexible enrichment framework. By building custom enrichment agent workflows, you can orchestrate the extraction of raw schema data, augment it with LLM-generated descriptions, and publish structured assets to your organization's catalog. This pattern leverages the toolbox/enrichment library to standardize data handling while allowing complete customization of each processing stage.

Three-Stage Pipeline Architecture

The enrichment workflow follows a strict separation of concerns across three independent stages. Each stage reads from and writes to a local fileset/ directory, creating a persistent pipeline that can be rerun, debugged, or modified without losing intermediate state.

• Download: Retrieves source metadata from BigQuery (tables, schemas) and stores normalized JSON/YAML snapshots locally.
• Enrich: Invokes the LLM-driven documentation agent to generate human-readable markdown and augmented schema tags.
• Publish: Pushes enriched artifacts to Knowledge Catalog via the Catalog API, creating OKF-compliant bundles.

This modular design means you can swap the download source (e.g., from Pub/Sub instead of BigQuery), replace the default OpenAI-based agent with a proprietary LLM, or publish results to a static site instead of the catalog—all without rewriting the core orchestration logic.

Download Stage: Extracting Source Metadata

The download step is implemented in samples/enrichment/src/enrichment/download.py. This script uses the BigQuery client to enumerate tables, fetch schema details, and persist raw snapshots to your local filesystem.

The core logic relies on enrichment.metadata.snapshot, which provides utilities for fetching and serializing table definitions:

# samples/enrichment/src/enrichment/download.py

from enrichment.metadata import snapshot

def main(project_id, dataset_id, output_dir):
    """Pull BigQuery table definitions and save to fileset."""
    tables = snapshot.fetch_bigquery_tables(project_id, dataset_id)
    snapshot.save_snapshots(tables, output_dir)

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--project", required=True)
    parser.add_argument("--dataset", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    main(args.project, args.dataset, args.output)

Running this module creates a fileset/ folder containing raw metadata snapshots. These JSON files serve as the immutable input for subsequent enrichment stages, ensuring the LLM processes consistent schema representations regardless of upstream changes.

Enrich Stage: LLM-Driven Documentation Generation

Once raw snapshots exist locally, the enrich stage invokes the documentation agent to generate markdown documentation. The entry point is samples/enrichment/src/enrichment/enrich.py, which imports enrichment.documentation.agent and feeds it the downloaded snapshots.

The agent processes each snapshot and returns enriched content, including table descriptions, column-level documentation, and inferred relationships:

# samples/enrichment/src/enrichment/enrich.py

from enrichment.documentation import agent
from enrichment.util import markdown

def main(input_dir, output_dir):
    """Generate enriched markdown from raw snapshots."""
    snapshots = markdown.load_snapshots(input_dir)
    enriched = agent.enrich_snapshots(snapshots)
    markdown.save_enriched(enriched, output_dir)

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    main(args.input, args.output)

Output files appear under fileset/tables/<table>.md, containing AI-generated documentation that conforms to the repository's markdown utilities (enrichment.util.markdown). This normalized format ensures the publish stage can parse enriched content regardless of which LLM or prompt template generated it.

Publish Stage: Catalog Registration and OKF Bundles

The final stage pushes enriched artifacts to Knowledge Catalog using samples/enrichment/src/enrichment/publish.py. This script reads the markdown files produced in the enrich stage and constructs Open Knowledge Format (OKF) bundles before calling the Catalog API.

The implementation uses enrichment.metadata.catalog to handle authentication, bundle assembly, and API communication:

# samples/enrichment/src/enrichment/publish.py

from enrichment.metadata import catalog
from enrichment.util import markdown

def main(input_dir, catalog_project):
    """Publish enriched docs to Knowledge Catalog."""
    docs = markdown.load_enriched(input_dir)
    for doc in docs:
        catalog.publish_document(doc, catalog_project)

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--catalog-project", required=True)
    args = parser.parse_args()
    main(args.input, args.catalog_project)

According to the OKF specification defined in okf/SPEC.md, each published document becomes a versioned asset visible in the Knowledge Catalog UI, complete with the AI-generated descriptions and schema annotations from the enrich stage.

Extending the Workflow

Because each stage is a standard Python module invoked via python3 -m enrichment.<stage>, you can replace components without disrupting the pipeline.

Custom Downloaders

Implement a new module conforming to the download() signature in samples/enrichment/src/enrichment/custom_download.py. Use enrichment.metadata.snapshot utilities to ensure your output matches the expected JSON schema, then invoke with python3 -m enrichment.custom_download --project=....

Alternative Enrichment Agents

Swap enrichment.documentation.agent for a bespoke prompt template or different model endpoint. The agent's input format (a snapshot JSON) remains constant, so custom implementations only need to return markdown-compatible strings that enrichment.util.markdown can serialize.

Additional Publish Targets

Modify publish.py to call alternative APIs besides Knowledge Catalog. Because the enrich stage outputs standard markdown, you can route enriched files to static site generators, data wikis, or internal documentation portals by replacing the catalog.publish_document call with your own HTTP client or file writer.

Summary

• The GoogleCloudPlatform/knowledge-catalog repository implements a three-stage pipeline (download, enrich, publish) via independent Python modules in samples/enrichment/src/enrichment/.
• Download uses snapshot.fetch_bigquery_tables to create local JSON snapshots in a fileset/ directory.
• Enrich leverages agent.enrich_snapshots to generate markdown documentation using LLM prompts.
• Publish utilizes catalog.publish_document to create OKF bundles and register assets with the Knowledge Catalog API.
• Each stage can be run independently, extended with custom logic, or replaced entirely while maintaining compatibility with the shared toolbox/enrichment utilities.

Frequently Asked Questions

What is the Open Knowledge Format (OKF) used in the publish stage?

OKF is the Open Knowledge Format specification defined in okf/SPEC.md that standardizes how enriched metadata is packaged and registered in Knowledge Catalog. It ensures that table schemas, generated documentation, and lineage information are serialized consistently for the Catalog API to consume.

Can I use a different LLM provider for the enrich stage?

Yes. The enrich stage in samples/enrichment/src/enrichment/enrich.py imports enrichment.documentation.agent as a pluggable component. You can replace this with any Python module that accepts snapshot JSON and returns markdown strings, allowing integration with OpenAI, Anthropic, Vertex AI, or self-hosted models without changing the pipeline structure.

How do I run individual stages without executing the full pipeline?

Each stage is a standalone Python module with its own CLI. Run python3 -m enrichment.download --project=..., python3 -m enrichment.enrich --input=..., or python3 -m enrichment.publish --catalog-project=... independently. The fileset/ directory serves as the persistent interface between stages.

What dependencies are required to run the enrichment scripts?

The scripts require the Python packages listed in samples/enrichment/requirements.txt, which include the BigQuery client, the Knowledge Catalog API client, and the shared toolbox/enrichment utilities. Install them with pip install -r samples/enrichment/requirements.txt before executing any stage.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}