Prefect for Enterprise AI: Architecture & Integration

import asyncio
from prefect import flow, task
from prefect.artifacts import create_markdown_artifact

@task(retries=3, retry_delay_seconds=5)
def fetch_unprocessed_documents(limit: int = 500):
  print(f"Fetching up to {limit} unprocessed documents...")
  return [{"id": i, "content": f"Sample document payload text {i}"} for i in range(limit)]

@task
async def generate_vector_embeddings_async(documents: list):
  print(f"Generating embeddings asynchronously for {len(documents)} items...")
  await asyncio.sleep(0.5)  # Simulating async vector embedding call
  return [{"id": d["id"], "vector": [0.015] * 1536} for d in documents]

@task
def upsert_vectors_to_database(embedded_docs: list):
  print(f"Upserting {len(embedded_docs)} vectors to enterprise database...")
  return len(embedded_docs)

@flow(name="enterprise_rag_prefect_ingestion", log_prints=True)
async def rag_ingestion_pipeline(document_batch_size: int = 1000):
  raw_docs = fetch_unprocessed_documents(limit=document_batch_size)
  embeddings = await generate_vector_embeddings_async(raw_docs)
  total_inserted = upsert_vectors_to_database(embeddings)
  
  # Create Prefect UI summary artifact
  create_markdown_artifact(
      key="ingestion-summary",
      markdown=f"### RAG Ingestion Complete\n- **Total Processed:** {total_inserted}\n- **Status:** Success"
  )

if __name__ == "__main__":
  asyncio.run(rag_ingestion_pipeline(500))