Conduit

• 

  Home Page

• 

  System Overview

• 

  Incidents

• 

  Live Reasoning

• 

  Traces

• 

  Evaluation

• 

  Connectors

Inspiration

Every data team has lived through the same nightmare: you wake up to a Slack flood at 2 AM because three pipelines silently broke overnight, dashboards are stale, and the on-call engineer is manually clicking through connector logs trying to figure out if it's a schema change, a rate limit, or rotated credentials. Mean time to resolution stretches to hours — not because the fix is hard, but because finding the problem is. I wanted to build something that closed that loop entirely. Not an alert that tells you something broke. An agent that wakes up, finds what broke, fixes what it can, and only pages a human when it genuinely cannot proceed on its own. The name Conduit comes from that idea — data should flow. When it stops, something should restore the flow automatically.

What It Does

Conduit is a 24/7 autonomous data pipeline incident response agent. It monitors Fivetran connectors continuously, classifies failure types (schema drift, rate limiting, auth expiry, network errors), takes remediation actions where possible, verifies the fix actually worked, and escalates to humans with full context when it cannot resolve the issue autonomously. Every decision the agent makes is scored across four dimensions: ( \text{Overall Score} = \frac{\text{Diagnosis} + \text{Action} + \text{Reasoning} + \text{Escalation}}{4} ) Traced in Arize Phoenix with full span-level observability Scored by an LLM-as-a-Judge evaluator on four dimensions: diagnosis accuracy, action appropriateness, reasoning clarity, and escalation judgment Stored so the agent can read its own past failures and improve in future runs The result is an agent that gets measurably better over time — a self-improving incident response system.

How I Built It

Agent layer: Google ADK (the code-first Agent Builder path) with Gemini 2.5 Flash as the reasoning engine. The agent has six Fivetran tools — list_connectors, get_connector_details, get_connector_logs, trigger_resync, pause_connector, and resume_connector — and a multi-step system prompt that enforces verify-after-fix behavior. Observability: Arize Phoenix with openinference-instrumentation-google-adk auto-instrumentation. Every Gemini call, tool invocation, and evaluator run is captured as a span in Phoenix Cloud with zero manual tracing code. Self-improvement loop: After each run, a second Gemini call acts as an LLM judge. The improvement dynamic follows: $$S_{t+1} > S_t \iff \text{lessons}(S_t) \in \text{context}(t+1)$$ Low-scoring runs are extracted and injected into the agent's system prompt on subsequent runs as explicit lessons. The agent reads its own failure history before diagnosing the next incident. Backend: FastAPI with Server-Sent Events for live reasoning stream delivery to the frontend. Frontend: Next.js dashboard — Overview, Live Reasoning, Evaluations, Traces, Connectors — using hash-based routing so the page reloads to the correct section. Design language draws from Grafana and Linear: dark, data-dense, operational. Data: Fivetran REST API integration for real connector data, with a stateful mock layer simulating schema drift, rate limiting, and auth expiry scenarios for reliable demo reproducibility.

Challenges I Faced

The authentication stack was genuinely hard. Fivetran's base64 API credentials contained non-ASCII characters that broke Python's requests library HTTP headers silently. The fix — encoding the key and secret directly using base64.b64encode(credentials.encode('utf-8')).decode('ascii') — took most of a day to diagnose. Vertex AI model access took 24 hours to propagate. New GCP projects don't have Gemini model access immediately after billing is linked. Every model name returned 404 for the first day, forcing a temporary fallback to AI Studio keys that burned through free credits before the switch was complete. The evaluator JSON parsing was brittle. Gemini's evaluation responses occasionally contained literal newlines inside JSON string values, breaking json.loads() with "Unterminated string" errors. The fix required extracting JSON by brace boundaries and stripping control characters with a regex pass before parsing. The always-on auto-monitor was too aggressive. An early version ran every ( t = 300 ) seconds indefinitely, burning API credits with no one watching. The architecture was rethought to be frontend-controlled — monitoring starts when the user opens the dashboard and stops when they close the tab via beforeunload cleanup. Building solo in under 4 weeks as a student meant hard scope decisions daily. The self-improvement loop, Phoenix MCP integration, and deployment had to be sequenced carefully to avoid the Fivetran 14-day trial expiring before the core integration was recorded as evidence.

What I Learned

The most important thing I learned is the difference between an agent that triggers actions and an agent that closes the loop. Adding a verification step — calling list_connectors again after remediation to confirm the connector moved to syncing state — transformed Conduit from automation into genuine autonomous behavior. LLM-as-a-Judge is a surprisingly effective self-improvement mechanism. Watching the agent score ( 3/5 ) on a contradictory response, read that evaluation as a lesson, then score ( 5/5 ) on the same connector the next run made the entire project feel real. The recovery time improvement speaks for itself: $$\text{MTTR}{\text{before}} \approx 4\ \text{hours} \quad \longrightarrow \quad \text{MTTR}{\text{after}} \approx 35\ \text{seconds}$$ Always build the mock data layer on day two while real API responses are fresh in your mind. Do not wait until the trial expires.

What's Next

Real-time webhook integration so Fivetran pushes failure events to Conduit rather than polling Multi-destination support (Snowflake, BigQuery, Redshift) with destination-specific validation queries Slack/Teams escalation with one-click human approval workflows Historical pattern recognition using Phoenix trace embeddings to predict failures before they occur

Built With

• cloud-run
• fastapi
• gcp-billing-/-iam-partner-integration:-fivetran-rest-api-(connector-management
• github-(version-control
• languages:-python
• log-retrieval)-via-direct-api-integration-and-the-fivetran-mcp-server-for-agent-tool-connectivity-observability:-arize-phoenix-cloud
• next.js
• openinference-instrumentation-(openinference-instrumentation-google-adk)
• opentelemetry-sdk-other-tools:-python-dotenv
• requests
• secret-manager
• sync-orchestration
• tailwind-css-ai-/-llm:-gemini-2.5-flash-via-google-generative-ai-sdk-google-cloud-products:-vertex-ai
• typescript-frameworks:-google-adk-(agent-development-kit)
• uvicorn
• vercel-(frontend-hosting)