-
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} # – –
-
Enterprice Agentic Platform
No posts found.
-
Data Science Use Case – Enterprise AI Assistant
Enterprise AI Platform
