Dagster for Enterprise AI: Architecture & Integration

from dagster import asset, asset_check, AssetCheckResult, MaterializeResult, Definitions
import pandas as pd

@asset
def raw_enterprise_documents() -> pd.DataFrame:
  """Extract raw document catalog from relational database."""
  data = [{"id": 1, "body": "Enterprise compliance document 2026..."}, {"id": 2, "body": "Financial audit memo..."}]
  return pd.DataFrame(data)

@asset
def chunked_text_passages(raw_enterprise_documents: pd.DataFrame) -> pd.DataFrame:
  """Transform raw documents into semantic passages."""
  passages = []
  for _, row in raw_enterprise_documents.iterrows():
      passages.append({"doc_id": row["id"], "chunk": row["body"][:30]})
  return pd.DataFrame(passages)

@asset_check(asset=chunked_text_passages)
def check_no_empty_chunks(chunked_text_passages: pd.DataFrame):
  """Quality check ensuring no empty passages are emitted."""
  empty_count = (chunked_text_passages["chunk"].str.len() == 0).sum()
  return AssetCheckResult(passed=bool(empty_count == 0), metadata={"empty_chunks": int(empty_count)})

@asset
def vector_knowledge_base(chunked_text_passages: pd.DataFrame) -> MaterializeResult:
  """Materialize vector embeddings inside enterprise vector database."""
  total_vectors = len(chunked_text_passages)
  # Upsert logic to vector database
  return MaterializeResult(
      metadata={
          "vector_count": total_vectors,
          "embedding_model": "text-embedding-3-small",
          "target_store": "qdrant-prod"
      }
  )

defs = Definitions(
  assets=[raw_enterprise_documents, chunked_text_passages, vector_knowledge_base],
  asset_checks=[check_no_empty_chunks]
)