-
EAP – Document AI Parser
EAP – Document AI Parser # ── STEP 1: Parse files via DocAI ───────────────────────── parsed_prefix = f"{tenant_id}/{connector_id}/parsed" logger.info("📄 Parsing Files With DocAI") t0 = perf_counter() parse_result = await parse_files( input_bucket=bucket, input_prefix=base_path, engine="docai", output_prefix_base=parsed_prefix, ) timings["parsing"] = perf_counter() – t0 logger.info( f"Parsing complete [{_fmt(timings['parsing'])}]: " f"processed={parse_result['processed']} | " f"success={parse_result['succeeded']} | " f"failed={parse_result['failed']}" ) async def parse_files( input_bucket: str, engine: str = "mistral", input_prefix: str | None = None, output_bucket: str | None = None, output_prefix_base: str | None = "parsed", ) -> dict: """ Parse files from an input GCS bucket and upload parsed outputs. Files are discovered by listing blobs
-
EAP – RAG Pipeline Design
EAP – RAG Pipeline Design 1. A document event arrives └─ Via Google Pub/Sub, or HTTP POST `/invoke` 2. `main.py` starts the pipeline └─ Reads tenant, connector, bucket, file path, and event type └─ Ignores duplicate events already in progress 3. Parse documents └─ Reads files from Google Cloud Storage └─ Uses Document AI by default └─ Extracts text, tables, layout, and image details └─ Saves parsed output under: `{tenant}/{connector}/parsed` 4. Chunk documents └─ Splits parsed content into small meaningful sections └─ Keeps headings/metadata where possible └─ Saves chunks under: `{tenant}/{connector}/chunked` 5. Index chunks └─ Creates/uses a tenant collection: `col-{tenant_id}`
-
EAP – Knowledge Base To RAG Workflow
EAP – Knowledge Base To RAG As A Service USER │ │ 1. Create Connector ▼ ┌─────────────────────────┐ │ FastAPI API Service │ │ │ │ Validate Request │ │ Create Integration │ │ Create Job │ │ Save to Firestore │ └───────────┬─────────────┘ │ 2. Save Metadata │ ▼ ┌─────────────────────────┐ │ Firestore │ │─────────────────────────│ │ integrations │ │ jobs │ │ tenants │ │ file_items │ │ errors │ └───────────┬─────────────┘ │ 3. Publish Event │ ▼ ┌────────────────────────┐ │ GCP Pub/Sub │ └───────────┬────────────┘ │ API Returns Success │ Immediately (2 sec) │ │ ▼ ┌────────────────────────┐ │ Worker Service │ │────────────────────────│ │ Listen
-
EAP – Connector Config Details
EAP – Connector Config Details
-
EAP – Firestore Different Collection
EAP – Firestore Collections
-
EAP – GCP Firestore Storage
EAP – GCP Firestore Storage Example: Import Firestore: from google.cloud import firestore Create Firestore Class: class FirestoreAdapter(DatabaseAdapter): """Firestore database adapter implementation.""" COLLECTION_INTEGRATIONS = "integrations" COLLECTION_JOBS = "jobs" COLLECTION_FILE_ITEMS = "file_items" COLLECTION_ERRORS = "errors" COLLECTION_TENANTS = "tenants" def __init__(self): """Initialize Firestore client.""" self.db: Optional[firestore.AsyncClient] = None self.project_id = settings.gcp_project_id self.database_id = settings.FIRESTORE_DATABASE_ID Establish Firestore Connection: async def connect(self) -> None: """Establish Firestore connection.""" try: self.db = firestore.AsyncClient( project=self.project_id, database=self.database_id, credentials=get_gcp_credentials(), ) logger.info(f"Connected to Firestore: {self.project_id}/{self.database_id}") except Exception as e: # Log only the exception type — the message may contain GCP project # details or auth tokens that should not appear
-
EAP – Connectors API
EAP – Connector API Details """ Connectors Endpoint — API Service Thin router layer: handles CRUD mechanics (DB, Pub/Sub, HTTP responses) and delegates ALL connector-specific logic to the appropriate orchestrator. Routes: GET /v1/{tenant_id}/connectors — list connectors GET /v1/{tenant_id}/connectors/{connector_id} — get connector detail POST /v1/{tenant_id}/connectors — create connector + publish worker event PATCH /v1/{tenant_id}/connectors/{connector_id} — update connector fields DELETE /v1/{tenant_id}/connectors/{connector_id} — hard-delete connector + all data To add a new connector type, create app/connectors/<source>.py implementing ConnectorOrchestrator and register it in app/connectors/registry.py. This file never needs to change. """ Get Connector Details: # – – ———————————————————————– # GET /v1/{tenant_id}/connectors/{connector_id} # – –
-
FAST API – include_router() ?
What Is include_router() ? from app.api.v1.router import router as api_v1_router app.include_router(api_v1_router) """ API v1 Router Single place where ALL routes are mounted. Adding a connector never touches this file — only endpoints/connectors.py and the schema layer change. """ from fastapi import APIRouter from app.api.v1.endpoints import connectors, health, jobs, sync, retriever router = APIRouter() router.include_router(health.router, prefix="/v1", tags=["Health"]) router.include_router(jobs.router, prefix="/v1", tags=["Jobs"]) router.include_router(connectors.router, prefix="/v1", tags=["Connectors"]) router.include_router(sync.router, prefix="/v1", tags=["Sync"]) router.include_router(retriever.router, prefix="/v1", tags=["Retrieval"])
-
FAST API – What Is add_middleware() ?
What Is add_middleware() ? add_middleware() What Is CORSMiddleware ? What Does CORSMiddleware Adds In The Response ? What Is Function Based Middleware & Class Based Middleware ?
-
FAST API – What Is FastAPI Class ?
What Is FastAPI Class ?
