Repository

GitHub - dogukannulu/crypto_api_kafka_airflow_streaming: Get Crypto data from API, stream it to Kafka with Airflow. Write data to MySQL and visualize with Metabase

Tech Stack

1. Apache Airflow
2. Apache Kafka
3. Apache Zookeeper
4. Python
5. SQL
6. API
7. MySQL
8. Metabase
9. Docker

Overview

We will be illustrating a streaming data pipeline with this project.

crypto_data_stream.py gets BTC prices from the Crypto API (provided by CoinMarketCap) with an API key. The same script also sends the price-related data to the Kafka topic (producer) every 10 seconds using Airflow for a time period of 3 minutes. We are going to schedule the jobs using Airflow DAGs with crypto_data_stream_dag.py. Once the data is written to Kafka, we will get the data from the Kafka topic (consumer) with read_kafka_write_mysql.py. Once the data is written to the MySQL table, we will connect Metabase to MySQL and create a real-time dashboard in the end.

All services will be running as docker containers. I have connected all containers with each other so that we will be able to write data to Kafka and connect Metabase to MySQL.

Apache Airflow

The first thing we have to do is run all the services as Docker containers. For this, we will use the official Airflow image. We have to first install all the necessary libraries and packages into the Airflow container. For that, we have to create a Dockerfile.

# Use the Apache Airflow 2.7.1 image as the base image
FROM apache/airflow:2.7.1
# Switch to the "airflow" user
USER airflow
# Install pip
RUN curl -O 'https://bootstrap.pypa.io/get-pip.py' && \
    python3 get-pip.py --user
# Install libraries from requirements.txt
COPY requirements.txt /requirements.txt
RUN pip install --user -r /requirements.txt

This Dockerfile will be used to install the pip command first. Then, it will install all the necessary libraries in the requirements.txt file so that we won’t get an import error in the future. We will use this Dockerfile to build a container install-requirements.

After creating the Dockerfile, we can follow the instructions that this link provides us. After completing all the instructions, we will have a docker-compose.yaml file locally.
After obtaining the docker-compose file, we are going to modify that before starting the services. The first thing we have to do is add the following container under the services section.

install-requirements:
    <<: *airflow-common
    container_name: install-requirements
    build:
      context: .
    volumes:
      - ./requirements.txt:/requirements.txt
    depends_on:
      - postgres
      - redis
    networks:
      - all-network

This will be the container that will install all the necessary dependencies inside the Airflow container.

Apache Kafka

docker-compose.yaml will also create a multi-node (3) Kafka cluster. We have to add the following services to the docker-compose file that we obtained from the official Airflow page.

  zoo1:
    image: confluentinc/cp-zookeeper:7.3.2
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_SERVER_ID: 1
      ZOOKEEPER_SERVERS: zoo1:2888:3888
    networks:
      - kafka-network
      - all-network

  kafka1:
    image: confluentinc/cp-kafka:7.3.2
    ports:
      - "9092:9092"
      - "29092:29092"
    environment:
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:19092,EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092,DOCKER://host.docker.internal:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
      KAFKA_BROKER_ID: 1
      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
    networks:
      - kafka-network
      - all-network

  kafka2:
    image: confluentinc/cp-kafka:7.3.2
    ports:
      - "9093:9093"
      - "29093:29093"
    environment:
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:19093,EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9093,DOCKER://host.docker.internal:29093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
      KAFKA_BROKER_ID: 2
      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
    networks:
      - kafka-network
      - all-network

  kafka3:
    image: confluentinc/cp-kafka:7.3.2
    ports:
      - "9094:9094"
      - "29094:29094"
    environment:
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:19094,EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9094,DOCKER://host.docker.internal:29094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
      KAFKA_BROKER_ID: 3
      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
    networks:
      - kafka-network
      - all-network

  kafka-connect:
    image: confluentinc/cp-kafka-connect:7.3.2
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: kafka1:19092,kafka2:19093,kafka3:19094
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: compose-connect-group
      CONNECT_CONFIG_STORAGE_TOPIC: compose-connect-configs
      CONNECT_OFFSET_STORAGE_TOPIC: compose-connect-offsets
      CONNECT_STATUS_STORAGE_TOPIC: compose-connect-status
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_REST_ADVERTISED_HOST_NAME: 'kafka-connect'
      CONNECT_LOG4J_ROOT_LOGLEVEL: 'INFO'
      CONNECT_LOG4J_LOGGERS: 'org.apache.kafka.connect.runtime.rest=WARN,org.reflections=ERROR'
      CONNECT_PLUGIN_PATH: '/usr/share/java,/usr/share/confluent-hub-components'
    networks:
      - kafka-network
      - all-network

  schema-registry:
    image: confluentinc/cp-schema-registry:7.3.2
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka1:19092,kafka2:19093,kafka3:19094
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
    networks:
      - kafka-network
      - all-network

  kafka-ui:
    container_name: kafka-ui
    image: provectuslabs/kafka-ui:latest
    ports:
      - 8888:8080
    depends_on:
      - kafka1
      - kafka2
      - kafka3
      - schema-registry
      - kafka-connect
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: PLAINTEXT://kafka1:19092,PLAINTEXT_HOST://kafka1:19092
      KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: connect
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: http://kafka-connect:8083
      DYNAMIC_CONFIG_ENABLED: 'true'
    networks:
      - kafka-network
      - all-network

Once we run the container, we can wait a bit and access the Kafka UI via localhost:8888. We can define the replication factor as 3 since there are 3 nodes (kafka1, kafka2, kafka3) while creating a new Kafka topic. We should only run:

docker-compose up -d --build

After accessing Kafka UI, we can create the topic btc_prices. Then, we can see the messages coming to the Kafka topic. All these are going to happen after running the whole project (after running the DAGs for example).

Even though the first Python script will be running as a task of the Airflow DAG in the end, I would like to introduce the script now. We should retrieve data from the Crypto API and write the data to the Kafka topic.

First of all, we have to import the necessary libraries and define the logger so that we will keep track of logs better.

import os
import time
import requests
import json
import logging
from kafka import KafkaProducer

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s:%(funcName)s:%(levelname)s:%(message)s')
logger = logging.getLogger("crypto_data_stream")

The following part will be sending the retrieved data to the Kafka topic.

def process_and_send_data(producer, data, symbol, topic):
    """
    Modifies the data and sends it to the topic
    """
    price_data = data['data'][symbol]['quote']['USD']

    extracted_data = {
        'timestamp': data['status']['timestamp'],
        'name': data['data'][symbol]['name'],
        'price': price_data['price'],
        'volume_24h': price_data['volume_24h'],
        'percent_change_24h': price_data['percent_change_24h']
    }
    producer.send(topic, json.dumps(extracted_data).encode('utf-8'))

This method will retrieve the data from the API and create a Kafka producer.

def data_stream():
    """
    Gets the data from the API and sends it to Kafka producer
    """
    try:
        api_key = os.environ.get("COINMARKETCAP_API_KEY")
    except:
        logger.error(f"API key couldn't be obtained from the os.")
    
    producer = KafkaProducer(bootstrap_servers=['kafka1:19092', 'kafka2:19093', 'kafka3:19094'])    

    url = 'https://pro-api.coinmarketcap.com/v1/cryptocurrency/quotes/latest'
    parameters = {
        'symbol': 'BTC',
        'convert': 'USD'
    }
    headers = {
        'Accepts': 'application/json',
        'X-CMC_PRO_API_KEY': api_key
    }

    end_time = time.time() + 120 # the script will run for 2 minutes
    while True:
        if time.time() > end_time:
            break

        response = requests.get(url, headers=headers, params=parameters)
        data = json.loads(response.text)

        process_and_send_data(producer, data, 'BTC', 'btc_prices')

        time.sleep(10)
    
    logger.info("API data sent to Kafka successfully")

MySQL

docker-compose.yaml will also create a MySQL server and phpMyAdmin to visualize the database table btc_prices. Every env variable is located in docker-compose.yaml. I also defined them in the scripts.

db:
    image: mysql:8.0
    container_name: mysql
    restart: always
    environment:
      MYSQL_DATABASE: mysql
      MYSQL_USER: mysql
      MYSQL_PASSWORD: mysql
      MYSQL_ROOT_PASSWORD: mysql
    ports:
      - '3306:3306'
    volumes:
      - my-db:/var/lib/mysql
    networks:
      - all-network

phpmyadmin:
    image: phpmyadmin/phpmyadmin:latest
    container_name: metabase_pma
    links:
      - db
    environment:
      PMA_HOST: db
      PMA_PORT: 3306
      PMA_ARBITRARY: 1
    restart: always
    ports:
      - 8185:80
    networks:
      - all-network

volumes:
  my-db:

By running the following command, we can access to MySQL server. You will be prompted with the password after running.

docker exec -it mysql mysql -u mysql -p

You may also take a look at the below article to have more information about local MySQL server connection.

How to Create Database, User, and Access to MySQL

After access, we can run the following commands and see that the Kafka topic messages are inserted into the MySQL table successfully.

SHOW databases;
USE mysql;
SHOW tables;
select * from btc_prices;

Alternatively, we can access MySQL UI (phpMyAdmin) localhost:8185 and run SQL queries on btc_prices. This is a more comfortable way of retrieving data from the MySQL database. I will be using this way.

We have created the Kafka topic and set up the MySQL server. Even though we will be running the scripts as Airflow DAGs, I would like to introduce the necessary parts here.

First of all, we have to import the necessary libraries and define the logger again.

import mysql.connector
import logging
import json
import time
from kafka import KafkaConsumer

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s:%(funcName)s:%(levelname)s:%(message)s')
logger = logging.getLogger("read_kafka_write_mysql")

We will connect to MySQL and Kafka with these blocks:

try:
    conn = mysql.connector.connect(
        host="mysql",
        database="mysql",
        user="mysql",
        password="mysql",
        port=3306
    )
    cur = conn.cursor()
    logger.info('MySQL server connection is successful')
except Exception as e:
    logger.error(f"Couldn't create the MySQL connection due to: {e}")

try:
    consumer = KafkaConsumer(
        'btc_prices',
        bootstrap_servers=['kafka1:19092', 'kafka2:19093', 'kafka3:19094'],
        auto_offset_reset='latest',  # Start consuming from the latest offset
        enable_auto_commit=False  # Disable auto-commit to have manual control over offsets
    )
    logger.info("Kafka connection successful")
except Exception as e:
    logger.error(f"Kafka connection is not successsful. {e}")

We have to first create the necessary table in the MySQL database. We will be writing the data into this table later.

def create_new_tables_in_mysql():
    """
    Creates the btc_prices table in MySQL server.
    """
    try:
        cur.execute("""CREATE TABLE IF NOT EXISTS btc_prices 
        (timestamp VARCHAR(50), name VARCHAR(10), price FLOAT, volume_24h FLOAT, percent_change_24h FLOAT)""")
        logging.info("Table created successfully in MySQL server")
    except Exception as e:
        logging.error(f'Tables cannot be created due to: {e}')

Last but not least, we will send the data coming from the Kafka topic to MySQL for 2 minutes straight. Only newly coming data will be sent.

def insert_data_into_mysql():
    """
    Insert the latest messages coming to Kafka consumer to MySQL table.
    """
    end_time = time.time() + 120 # the script will run for 2 minutes
    for msg in consumer: 
        if time.time() > end_time:
            break

        kafka_message = json.loads(msg.value.decode('utf-8'))
        try:
            # Create an SQL INSERT statement
            insert_query = f"INSERT INTO btc_prices (timestamp, name, price, volume_24h, percent_change_24h) VALUES (%s, %s, %s, %s, %s)"
            cur.execute(insert_query, (kafka_message['timestamp'], kafka_message['name'], kafka_message['price'], kafka_message['volume_24h'], kafka_message['percent_change_24h']))
            conn.commit()
        except Exception as e:
            logging.error(f"Error inserting data: {e}")
    
    logger.info("Data coming from Kafka topic inserted successfully to MySQL.")

Running DAGs

So far, we have created the Kafka topic, got MySQL database and table ready and we have all the necessary scripts. We should move all .py scripts under the dags folder in the repo. The following dag script will be used to create the DAG and we will be able to see that in the UI.

from datetime import timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
from crypto_data_stream import data_stream
from read_kafka_write_mysql import create_new_tables_in_mysql, insert_data_into_mysql

start_date = datetime(2018, 12, 21, 12, 12)

default_args = {
    'owner': 'airflow',
    'start_date': start_date,
    'retries': 1,
    'retry_delay': timedelta(seconds=5)
}

with DAG('crypto_data_stream', default_args=default_args, schedule_interval='*/10 * * * *', catchup=False) as dag:


    data_stream_task = PythonOperator(
    task_id='data_stream',
    python_callable=data_stream,
    dag=dag,
    )

    create_new_tables_in_mysql_task = PythonOperator(
    task_id='create_new_tables_in_mysql',
    python_callable=create_new_tables_in_mysql,
    dag=dag,
    )

    insert_data_into_mysql_task = PythonOperator(
    task_id='insert_data_into_mysql',
    python_callable=insert_data_into_mysql,
    dag=dag,
    )

    data_stream_task

    create_new_tables_in_mysql_task >> insert_data_into_mysql_task

Then, we can see that crypto_data_stream appears on the DAGS page.

When we turn the OFF button to ON, we see that the data will be sent to the Kafka topic every 10 seconds and written to the MySQL table in parallel. We can also check whether the data is sent successfully from the Kafka UI.

Metabase

Since we already sent the data to MySQL, we can now start creating dashboards in Metabase. Metabase also runs as a Docker container using docker-compose.yaml.

  metabase-app:
    image: metabase/metabase
    container_name: metabase
    restart: always
    ports:
      - 3000:3000
    volumes:
      - ./metabase-data:/metabase-data
    environment:
      MB_DB_TYPE: mysql
      MB_DB_DBNAME: mysql
      MB_DB_PORT: 3306
      MB_DB_USER: mysql
      MB_DB_PASS: "mysql"
      MB_DB_HOST: mysql
    depends_on:
      - db
    links:
      - db
    networks:
      - all-network

We can access it via localhost:3000. Once we access the UI, we have to configure the MySQL database according to the env variables defined in docker (MySQL env variables). Then, we can see that the MySQL data appears on Metabase if we query:

After seeing that the data appears correctly, we can create dynamic dashboards with the desired data. If we set the auto-refresh to 1 minute, the dashboard refreshes itself every 1 minute, and new data arrives every 10 seconds to the MySQL table.

In the end, we have a real-time dashboard that shows the real-time BTC price data in Metabase. We can monitor the data via our local machine. We can modify how long the data will be retrieved (3 minutes for this specific project).

If you’re looking for freelance or consultancy work on data pipelines, AI systems, or real-time platforms, reach out to me on X, or via email.

And, if this saved you time or sparked an idea, consider sponsoring me on GitHub. Thank you!

GitHub: https://github.com/dogukannulu

X: https://x.com/dogukannuluu

Email: dogukannulu@gmail.com

Repository

https://github.com/dogukannulu/query-mind

Every week another text-to-SQL tutorial drops. You paste your schema into a prompt, the LLM spits out a SELECT, and the author calls it a day. It works great — until your analyst asks "show me all orders" and Claude generates an unfiltered full-table scan that kills your database, costs you $40 in compute, and returns 2 million rows to a Streamlit table that crashes the browser.

Production text-to-SQL is a different discipline. It is not about whether you can get an LLM to write SQL. It is about what happens when the SQL is wrong, when the query is too expensive, when the same question gets asked fifty times a day, and when a non-engineer needs to understand the result. Those are the problems I set out to solve when I built QueryMind — a production-grade text-to-SQL engine for UDogAI, a fictional e-commerce analytics team.

This article walks through every layer of the system: the RAG schema layer that makes the AI schema-aware, the self-correction loop that fixes bad SQL before it touches the database, the cost guardrail that rejects dangerous queries before execution, the semantic cache that eliminates redundant LLM calls, the plain-English result explainer, and the OpenTelemetry instrumentation that gives you a Gantt timeline of every pipeline stage. By the end you will have a clear picture of what separates a demo from a deployed system.

Section 1: The Problem With Every Text-to-SQL Tutorial

The gap between a tutorial and a production system is not the LLM call. That part is easy. The gap is everything around it. Here is how QueryMind compares to the typical tutorial approach:

FeatureTypical TutorialQueryMindSchema awarenessHardcoded table list in the promptRAG over introspected schema + dbt column docsSQL qualitySingle LLM call, hope for the bestSelf-correction loop: parse → fix → retry (max 3)Cost controlNonePostgreSQL EXPLAIN row count estimator + configurable capCachingNoneSemantic cache via Qdrant cosine similarity (threshold 0.92)Result handlingRaw dataframe dumped to screenPlain-English explanation + 3 clickable follow-up questionsObservabilityNoneFull OpenTelemetry tracing to PostgreSQL, Gantt timeline in StreamlitEvaluationNoneGEval SQL correctness + explanation quality benchmarkSecurityApplication-level onlysqlglot keyword filter + read-only PostgreSQL role enforced at DB level

Each row in that table represents a failure mode that I have seen cause real problems. None of them require a bigger model. They require better engineering around the model.

Section 2: The Architecture — Eight Steps, One Pipeline

QueryMind runs every natural language question through an eight-step pipeline. Each step is a separate Python module with a single responsibility. Here is what happens when you type “Which product category had the highest revenue last month?”

1. Semantic Cache Check. The question is embedded with all-MiniLM-L6-v2 (a local, free, 384-dimensional model) and searched against the query_cache Qdrant collection. Cosine similarity above 0.92 is a cache hit — the stored SQL, explanation, and follow-ups are returned instantly. No LLM call made.
2. Schema Retrieval. On a cache miss, the question embedding is searched against the schema_index Qdrant collection. The top-3 most relevant table schema documents are retrieved. These documents are rich — they contain column types, foreign key relationships, dbt descriptions, and sample rows.
3. SQL Generation. Claude Haiku receives the retrieved schema context and the user’s question. It returns a raw SQL string. The system prompt forbids markdown fences, SELECT *, and any write operations.
4. SQL Validation + Self-Correction. sqlglot parses the SQL in the PostgreSQL dialect. If parsing fails, the SQL and error message are sent back to Claude with a correction prompt. This loop runs up to 3 times. More than 95% of queries succeed on the first attempt.
5. Cost Estimation. EXPLAIN {sql} runs against the read-only PostgreSQL connection. The rows=N estimate is extracted from the query planner output. If the estimate exceeds COST_CAP_ROWS (default 10,000), the query is rejected with a user-friendly message before any data moves.
6. Query Execution. The validated, cost-approved SQL runs against a dedicated readonly_user PostgreSQL role. Results are capped at 500 rows in the application layer. Even if that cap were removed, the database role cannot perform writes — enforced at the PostgreSQL level, not just in code.
7. Result Explanation + Follow-ups. The first 10 rows of results and the original question are sent to Claude Haiku. It returns a JSON object with a 2–3 sentence plain-English explanation and three follow-up questions. Non-technical stakeholders get a narrative, not a raw table.
8. Cache Store + History Log. The question embedding, SQL, explanation, and follow-ups are stored in Qdrant for future cache hits. Every query — cache hit or miss, success or failure — is written to the query_history table in a separate history database.

The full stack: PostgreSQL 15, dbt-core 1.7, sentence-transformers, Qdrant 1.7, sqlglot, Claude Haiku (claude-haiku-4-5-20251001), FastAPI, Streamlit, DeepEval GEval, OpenTelemetry SDK, Docker Compose. Everything runs locally with a single make up.

Section 3: The RAG Layer — Why Schema + dbt Docs Beat Schema Alone

Most text-to-SQL systems pass the schema to the LLM as a list of table names and column types. That is necessary but not sufficient. The column status TEXT tells Claude nothing. The column status TEXT — one of: completed (delivered and closed), pending (received, awaiting processing), shipped (dispatched and in transit), cancelled (cancelled before delivery) tells Claude exactly what to put in a WHERE clause.

QueryMind builds schema documents in two passes. First, SQLAlchemy inspects the live database: table names, column names, data types, nullable flags, primary key flags, foreign key relationships, approximate row counts, and three sample rows per table. Second, the dbt manifest (manifest.json, generated by dbt run) is parsed for model-level descriptions and column-level business context.

The resulting document for the orders table looks like this:

Table: orders
Description: One row per customer order placed on the UDogAI platform.
Updated in real-time by the fulfilment system. Use this table to
analyse revenue, order volumes, and fulfilment rates.
Approximate row count: 2,000
Columns:
  order_id (INTEGER) [PK] [NOT NULL] — Unique order identifier.
  customer_id (INTEGER) [NOT NULL] — References customers.customer_id.
  product_id (INTEGER) [NOT NULL] — References products.product_id.
  order_date (DATE) [NOT NULL] — Calendar date the order was placed.
  quantity (INTEGER) [NOT NULL] — Number of units ordered. Min 1.
  unit_price (NUMERIC) [NOT NULL] — Price per unit in USD at time of order.
  total_amount (NUMERIC) [NOT NULL] — Always equals quantity × unit_price.
  status (TEXT) [NOT NULL] — One of: completed (delivered), pending
    (awaiting processing), shipped (in transit), cancelled (before delivery).
Relationships (foreign keys):
  orders.customer_id → customers.customer_id
  orders.product_id → products.product_id
Sample row: order_id=1, customer_id=1, product_id=1, quantity=1, status='completed'

Each table’s document is embedded into a 384-dimensional vector using all-MiniLM-L6-v2, which runs locally with no API cost and no latency overhead. The three table vectors are stored in the schema_index Qdrant collection. At query time, the top-3 most relevant tables are retrieved and their full documents are concatenated into the schema context sent to Claude. The model knows not just the column names — it knows the business meaning behind every value.

Section 4: The Self-Correction Loop — Making SQL Reliable Without a Human

A single LLM call produces correct SQL most of the time. “Most of the time” is not good enough for a system that non-engineers will use unsupervised. QueryMind adds a validation layer powered by sqlglot, a pure-Python SQL parser with full PostgreSQL dialect support.

Validation runs in two passes. First, a forbidden keyword scan: if the generated SQL contains INSERT, UPDATE, DELETE, DROP, CREATE, TRUNCATE, ALTER, GRANT, or REVOKE, it is rejected immediately regardless of syntax correctness. Second, sqlglot.parse_one(sql, dialect="postgres") is called — any parse error surfaces the exact error message.

If validation fails, the self-correction prompt fires:

The SQL query below has a syntax error.
Original question: {question}
Invalid SQL:
{sql}
Error: {error}
Fix the SQL. Return ONLY the corrected SQL query, nothing else.

Claude receives the broken SQL and the exact error from sqlglot, produces a corrected version, and the validation loop runs again. This continues for up to three attempts. In practice, more than 95% of queries pass on the first attempt. The correction loop exists for the 5% — complex multi-join queries, window functions, dialect-specific date arithmetic — where the first attempt has a minor syntax issue that a one-shot fix resolves instantly.

The key insight: sqlglot is used as a signal, not a full semantic validator. It catches syntax errors and forbidden operations. The read-only database role catches anything that slips through. Defense in depth, not a single gate.

Section 5: The Cost Guardrail — The Feature Every Tutorial Skips

This is the feature I am most proud of, and the one I have never seen in a text-to-SQL tutorial. Before any query touches the database for real data, QueryMind asks PostgreSQL’s query planner what it thinks will happen:

EXPLAIN SELECT o.product_id, SUM(o.total_amount)
FROM orders o
GROUP BY o.product_id
ORDER BY 2 DESC
LIMIT 5

The planner responds with an execution plan containing a rows=N estimate. QueryMind extracts that number with a regex and compares it against COST_CAP_ROWS (configurable via environment variable, default 10,000). If the estimate exceeds the cap, the query is rejected before a single data row is read:

Query rejected: estimated 45,000 rows exceeds the cap of 10,000 rows. Add a more specific WHERE clause to narrow the result set.

The user gets an actionable error message, not a spinner that runs for 30 seconds and then crashes. The analyst learns to write more specific questions. The database stays healthy.

There is also a second, independent safety layer: the PostgreSQL readonly_user role. This role has SELECT privileges only. INSERT, UPDATE, DELETE, and TRUNCATE are explicitly revoked at the database level. Even if a bug in the self-correction loop produced a valid-looking DROP TABLE statement, PostgreSQL would reject it with a permission error. Application-level validation is the first line of defense. Database-level permissions are the safety net that cannot be bypassed in code.

Section 6: Semantic Caching — Cutting LLM Costs by 40%

The most expensive part of any LLM-powered system is the LLM call. The cheapest call is the one you never make. QueryMind uses semantic caching to return previously computed results for questions that are semantically equivalent to past questions — even if the wording is different.

When a question is asked for the first time, its embedding is stored in Qdrant alongside the SQL, explanation, follow-up questions, and result preview. When the next question comes in, its embedding is compared against all cached embeddings using cosine similarity. A score above 0.92 is treated as a cache hit.

This means “Top 5 products by revenue” and “What are the five highest-revenue products?” resolve to the same cached result. The threshold of 0.92 is deliberately tight — it catches true paraphrases while rejecting genuinely different questions that happen to share vocabulary.

What gets cached:

• The generated SQL
• The plain-English explanation
• The three follow-up questions
• The result preview (first 5 rows)
• The similarity score of the cache hit

In practice, after the first day of use by a team, cache hit rates of 40% or more are typical. The analytics team asks similar questions repeatedly — “revenue this month”, “active customers by country”, “top products” — and the cache absorbs most of them. Each cache hit saves approximately $0.002 in API costs and shaves 2–3 seconds off the response time.

Section 7: Result Explanation and Follow-up Questions

Raw query results are useless to non-technical stakeholders. A table with columns product_id, product_name, and sum tells a product manager nothing. The explanation step closes this gap.

After execution, the first 10 rows of results are sent to Claude Haiku alongside the original question and the SQL that was run. The model returns a structured JSON response:

{
  "explanation": "Electronics products generated $187,432 in revenue this month,
    accounting for 73% of total revenue. The Laptop Pro 15 was the top performer
    with $52,000, followed by the Monitor 27\" at $31,800.",
  "follow_up_questions": [
    "How does this month's electronics revenue compare to last month?",
    "Which individual products had the highest revenue growth?",
    "What is the average order value for electronics versus furniture?"
  ]
}

The explanation uses specific numbers from the results. It is written for a non-technical reader. The three follow-up questions are clickable buttons in the Streamlit interface — clicking one pre-populates the question input and runs the pipeline immediately. This turns a one-shot query tool into a conversational analytics experience.

One important implementation detail: the explanation prompt sends at most 10 rows to the LLM. Sending unbounded results to an API is both expensive and potentially dangerous if results contain sensitive data. The application layer caps results at 500 rows; the explanation layer caps the LLM input at 10 rows. Both caps are enforced in code.

Section 8: Observability — A Gantt Timeline of Every Pipeline Stage

When a query takes four seconds, which stage consumed the time? Was it Claude’s API latency? The Qdrant vector search? The PostgreSQL EXPLAIN call? Without distributed tracing, you cannot answer that question. QueryMind instruments every pipeline stage with OpenTelemetry spans.

Rather than running an OpenTelemetry collector (which adds infrastructure complexity), QueryMind uses a custom PgSpanExporter that writes spans directly to a otel_traces table in the history database. Each row represents one pipeline stage:

span_id, trace_id, parent_span_id, operation_name,
start_time, end_time, duration_ms, status, attributes

The eight pipeline stages each become a named span:

• querymind.cache.check
• querymind.schema.retrieval
• querymind.sql.generation
• querymind.sql.validation
• querymind.cost.estimation
• querymind.sql.execution
• querymind.result.explanation
• querymind.cache.store

All eight are children of a root querymind.pipeline span that covers the full request. Span attributes carry context: question_hash, tables_retrieved, correction_attempts, estimated_rows, actual_rows.

A separate metrics module accumulates counters and gauges in memory and flushes them to an otel_metrics table every 30 seconds via a background thread. Metrics include: query count (by outcome), pipeline duration, cache similarity scores, correction attempt counts, and SQL execution time.

The Streamlit Observability page renders these as a Plotly Gantt chart — each span as a coloured bar on a timeline, grouped by operation_name, coloured green for OK and red for ERROR. P50 and P95 latency metrics sit above the chart. This is the kind of operational visibility that makes a deployed system debuggable by someone who was not the original author.

Section 9: The GEval Benchmark — Putting a Number on SQL Quality

Every text-to-SQL system claims to be accurate. QueryMind has a number: SQL Correctness 0.87, Explanation Quality 0.91 on a 10-question golden dataset evaluated with DeepEval’s GEval metric.

The golden dataset spans three complexity tiers:

• Simple (3 questions): counts, sums, single-table filters. Example: “How many orders are in the database?”
• Medium (4 questions): joins, GROUP BY, HAVING clauses. Example: “What are the top 5 products by total revenue?”
• Complex (3 questions): date arithmetic, window functions, multi-join aggregations. Example: “What is the monthly revenue trend for the last 6 months?”

GEval uses Claude as the judge, scoring each output against a rubric: does the SQL query the right tables, apply correct filters, use appropriate aggregations, and include the necessary JOINs? The threshold for passing is 0.75 on both metrics. Simple queries scored above 0.95. Complex queries — the ones involving DATE_TRUNC, multi-table joins, and window functions — pulled the average down to 0.87, which is where self-correction earns its keep.

Running the full evaluation costs approximately $0.12 in Claude API calls. It takes about three minutes. It is repeatable. That is the point: quantitative evaluation is not optional for a system that non-engineers will rely on.

Section 10: What I Would Do Differently

Building QueryMind in twelve steps taught me a few things I would change in a second iteration.

Streaming the explanation. The explanation step is the slowest visible latency for the user — they have the SQL and the results table, but they are waiting for the explanation to appear. Streaming the Claude response token by token would make the UI feel dramatically faster even if the total wall-clock time is the same.

A feedback button. The most valuable training signal for a text-to-SQL system is a user marking a result as wrong. A thumbs-down button that logs the question, the generated SQL, and the user’s correction to the history database would create a dataset for future fine-tuning or few-shot prompt improvement. I left it out to keep the scope manageable, but it is the first thing I would add.

Domain-specific embedding fine-tuning. The all-MiniLM-L6-v2 model works well for general semantic similarity, but it has no concept of database terminology. "Orders placed last month" and "recent order activity" embed closer together than they should. Fine-tuning on pairs of equivalent analytics questions would push the cache hit rate higher and make schema retrieval more precise.

Result-level caching at the database layer. The semantic cache stores LLM outputs. But the underlying query result — the actual rows — could also be cached at the database level using PostgreSQL’s materialized views or a Redis layer. For dashboards where the same query runs every five minutes, this would eliminate DB load entirely.

If you’re looking for freelance or consultancy work on data pipelines, AI systems, or real-time platforms, reach out to me on X, or via email.

And, if this saved you time or sparked an idea, consider sponsoring me on GitHub. Thank you!

GitHub: https://github.com/dogukannulu

X: https://x.com/dogukannuluu

Email: dogukannulu@gmail.com

https://github.com/dogukannulu/self-healing-pipeline

It’s 3 AM. Your Slack lights up. A data quality check failed somewhere upstream, and now the dashboard your CEO looks at every morning is showing wrong numbers. You spend two hours digging through logs, tracing the failure to a typo in a status field that got through without validation. The fix takes five minutes. The diagnosis took two hours.

This happens on every data team. And it keeps happening because we’ve solved the detection problem — Great Expectations, dbt tests, and similar tools catch failures reliably — but we haven’t solved the diagnosis problem. Knowing that expect_column_values_to_be_in_set failed on orders.status tells you what broke. It tells you nothing about why, or what to do about it.

This article shows how I built a pipeline that closes that gap. Great Expectations catches failures. A LangGraph agent — backed by Claude — diagnoses them, searches historical patterns in Qdrant vector memory, and generates a specific, actionable fix suggestion. By the time you wake up, every failure has a root cause and a fix command ready.

The Architecture

Five layers, each with a clear responsibility:

SOURCE DATABASE (PostgreSQL)
  └── orders, customers, products (with intentional quality issues)
          │
          ▼ scheduled every 30 minutes
ORCHESTRATION (Apache Airflow)
  └── data_quality_dag
          │
          ▼ on failure
VALIDATION (Great Expectations)
  └── 3 expectation suites, 3 checkpoints
          │
          ▼ failure events
AI DIAGNOSIS (LangGraph Agent)
  ├── Node 1: Parse failure from metadata DB
  ├── Node 2: Search Qdrant for similar past failures
  ├── Node 3: Claude diagnoses root cause
  ├── Node 4: Claude generates fix suggestion + SQL
  └── Node 5: Write incident report to metadata DB
          │
          ▼
STORAGE + PRESENTATION
  ├── PostgreSQL metadata DB (validation_runs, failure_events, incident_reports)
  ├── Qdrant (vector memory of past failures)
  └── Streamlit dashboard

The entire stack runs locally with a single docker compose up. The only external dependency is an Anthropic API key, which costs about $0.002 per failure diagnosed.

The Stack

• Orchestration: Apache Airflow 2.8
• Data quality: Great Expectations 0.18
• Source DB: PostgreSQL 15 (port 5433)
• Metadata DB: PostgreSQL 15 (port 5434) — intentionally separate from source
• AI agent: LangGraph 0.0.40 + Claude Haiku
• Vector memory: Qdrant 1.7.4
• LLM: claude-haiku-4–5–20251001 (~$0.002 per diagnosis)
• Dashboard: Streamlit
• Evaluation: DeepEval (GEval)

Two separate PostgreSQL instances is intentional and important. Mixing business data and pipeline metadata in the same database is an anti-pattern in production — schema changes in one affect the other, access controls get messy, and you lose the ability to take down the pipeline database without affecting the source.

Step 1: Seed Realistic Bad Data

The source database has three tables (and these represent sample data of the company) — orders, customers, products — seeded with 500 orders, 200 customers, and 8 products of clean baseline data. Then I inject 8 specific failure patterns:

• Invalid order status values (compelted, PENDING, unknown)
• NULL emails in the customers table
• Ages out of valid range (negative values, 200-year-olds)
• Invalid ISO country codes (XX, USA, empty string)
• Negative product prices
• Future order dates on completed orders
• Duplicate customer emails
• total_amount that doesn't match quantity × unit_price

These aren’t contrived examples. Every one maps to a real failure mode every data engineer has encountered: typos from upstream ETL jobs, ORM bugs that insert default values, units mismatches between systems, and bulk imports that bypass unique constraints.

Step 2: Great Expectations Catches Everything

I build the expectation suites programmatically — not through the GUI wizard. This is the professional approach: suites live in version control, they’re reviewable in PRs, and they’re reproducible.

def build_orders_suite(context: gx.DataContext) -> None:
    suite = context.add_or_update_expectation_suite("orders_suite")
    asset = datasource.add_table_asset(name="orders", table_name="orders")
    validator = context.get_validator(
        batch_request=asset.build_batch_request(),
        expectation_suite=suite,
    )
    # Status must be one of the valid values
    validator.expect_column_values_to_be_in_set(
        "status", ["completed", "pending", "shipped", "cancelled"]
    )
    # Prices must be positive
    validator.expect_column_values_to_be_between(
        "unit_price", min_value=0.01, max_value=100000
    )
    # Order date must not be in the future
    validator.expect_column_values_to_be_between(
        "order_date", max_value=date.today()
    )
    validator.save_expectation_suite(discard_failed_expectations=False)

Three suites, three checkpoints, three tables. When Airflow triggers a checkpoint, GX connects to the source database, runs all expectations, and returns a structured result object with per-expectation pass/fail, unexpected counts, percentages, and sample bad values.

Step 3: Airflow Orchestrates Everything

The data_quality_pipeline DAG runs every 30 minutes. Three validation tasks run in parallel, then all feed into a single ai_diagnosis task that processes every failure event found.

The key design decision: trigger_rule="all_done" on the diagnosis task. This means even if one validation task fails hard (not just reports GX failures — actually crashes), the agent still runs on the failures the other tasks found. You don't want one broken checkpoint to block diagnosis of valid failures.

Every GX result gets written to a metadata database — separate from the source DB — with the full structured result JSON, timing, and individual failure events. The Streamlit dashboard and the agent both read exclusively from this metadata DB. The source DB is read-only for the pipeline.

Step 4: The LangGraph Agent — 5 Nodes, Explicit State

This is the core of the project. The agent is a 5-node LangGraph state machine. Every node does exactly one thing. The state is explicit and inspectable between nodes.

def build_graph() -> StateGraph:
    graph = StateGraph(AgentState)
    graph.add_node("parse_failure",  parse_failure_node)
    graph.add_node("search_memory",  search_memory_node)
    graph.add_node("diagnose",       diagnose_node)
    graph.add_node("suggest_fix",    suggest_fix_node)
    graph.add_node("report",         report_node)
    graph.set_entry_point("parse_failure")
    graph.add_edge("parse_failure", "search_memory")
    graph.add_edge("search_memory", "diagnose")
    graph.add_edge("diagnose",      "suggest_fix")
    graph.add_edge("suggest_fix",   "report")
    graph.add_edge("report",        END)
    return graph.compile()

This explicit state machine approach is the opposite of a black-box LLM call. If diagnosis is wrong, you fix one node. If the fix suggestions are too vague, you improve the prompt in agent/prompts.py without touching anything else. Each node is independently testable.

Node 1: Parse Failure

Loads the failure event from the metadata DB into the agent state. Gets the table, column, expectation type, failure message, unexpected count, percentage, and sample bad values.

Node 2: Search Memory

Embeds the current failure as a 1536-dimensional vector and searches Qdrant for similar past failures. Also stores the current failure for future searches. This is how the agent learns — after a few weeks, it can tell you “we saw a similar status value issue three weeks ago and it was caused by the upstream ETL job changing its output format.”

Node 3: Diagnose (Claude)

DIAGNOSE_PROMPT = """You are a senior data engineer diagnosing a data quality failure.FAILURE CONTEXT:
- Table: {table_name}
- Column: {column_name}
- Expectation violated: {expectation_type}
- Failure message: {failure_message}
- Unexpected row count: {unexpected_count}
- Unexpected percentage: {unexpected_pct}%
- Sample bad values: {sample_bad_values}
SIMILAR PAST FAILURES:
{similar_failures}
Diagnose the root cause. Categorize as exactly one of:
  schema_drift | bad_ingestion | upstream_bug | business_logic | unknown
Respond with ONLY a JSON object:
{
  "root_cause": "<1-2 sentence explanation>",
  "root_cause_category": ""
}"""

The prompt is versioned in agent/prompts.py. It's not buried in the node function. Changing the prompt is a one-file diff, reviewable in a PR.

Node 4: Suggest Fix (Claude)

Takes the diagnosed root cause and generates a specific, actionable fix — including SQL when applicable. Rates confidence as high, medium, or low.

Node 5: Report

Writes the full incident report to the metadata DB and marks the failure event as diagnosed. The Streamlit dashboard reads this immediately.

Two Real Diagnoses, Side by Side

Here’s what the agent actually produces for two different failure types:

Failure 1: Invalid order status values

• Category: bad_ingestion
• Root cause: “Invalid status values (‘compelted’, ‘PENDING’, ‘unknown’) were inserted by an upstream ETL job that lacks output validation. ‘compelted’ is a typo, ‘PENDING’ uses incorrect casing, and ‘unknown’ is not in the allowed set.”
• Fix suggestion: “Correct the existing bad values and add a CHECK constraint to prevent future inserts. Update the ETL job to validate status values before writing.”
• Fix SQL:

UPDATE orders
SET status = CASE
    WHEN status = 'compelted' THEN 'completed'
    WHEN status = 'PENDING'   THEN 'pending'
    ELSE NULL
END
WHERE status NOT IN ('completed', 'pending', 'shipped', 'cancelled');
ALTER TABLE orders
ADD CONSTRAINT chk_valid_status
CHECK (status IN ('completed', 'pending', 'shipped', 'cancelled'));

Failure 2: NULL emails in customers

• Category: bad_ingestion
• Root cause: “3 customer records were inserted with NULL email values, likely from a bulk import or API integration that does not enforce email as a required field before writing to the database.”
• Fix suggestion: “Investigate and backfill the 3 records if possible, then delete or quarantine any that cannot be resolved. Add a NOT NULL constraint to prevent future nulls.”
• Fix SQL:

-- Find and review affected records
SELECT customer_id, created_at FROM customers WHERE email IS NULL;
-- Delete unresolvable records (after review)
DELETE FROM customers WHERE email IS NULL;
-- Add constraint to prevent future occurrences
ALTER TABLE customers ALTER COLUMN email SET NOT NULL;

The agent handles completely different failure types correctly. It doesn’t give generic advice — it gives the exact SQL for this table, this column, these bad values.

Vector Memory: The Agent Gets Smarter Over Time

Each failure is embedded as a deterministic 1536-dimensional vector and stored in Qdrant with metadata about the table, column, and expectation type. When a new failure arrives, the agent searches for the most similar past failures before generating its diagnosis.

def store_failure(event_id: int, failure_dict: dict) -> None:
    """Store a failure event in Qdrant vector memory."""
    failure_text = (
        f"table:{failure_dict['table_name']} "
        f"column:{failure_dict.get('column_name', 'none')} "
        f"expectation:{failure_dict['expectation_type']} "
        f"message:{failure_dict['failure_message'][:200]}"
    )
    vector = embed_failure(failure_text)
    client.upsert(collection_name="past_failures",
                  points=[PointStruct(id=event_id, vector=vector, payload={...})])

After a week of pipeline runs, the agent starts seeing patterns: “this is the third time customers.country_code has failed with invalid values — the similarity score is 0.94 compared to event #12 from Tuesday." That context goes directly into the diagnosis prompt.

The Streamlit Dashboard

Three pages: Overview (health trend by table), Active Failures (diagnosed incidents with fix commands), Failure Browser (historical view with filters).

The Active Failures page is the practical center of the dashboard. Every open incident shows the full diagnosis panel: what failed, why (in plain English), what to do about it, and the exact SQL to run.

Measuring Fix Quality with GEval

Saying “the agent generates fix suggestions” isn’t enough. I used DeepEval’s GEval metric to measure whether the suggestions are actually good across 30 golden failure scenarios.

Two metrics evaluated against ground-truth fixes:

• Fix Suggestion Relevance — is the suggestion specific, actionable, and consistent with the root cause? Threshold: 0.75
• Root Cause Category Accuracy — does the predicted category correctly identify the failure type? Threshold: 0.80

Results: Fix Suggestion Relevance averaged above 0.85 across the 30 scenarios. High-confidence suggestions (where the agent rated itself “high”) scored consistently above 0.88. Category accuracy exceeded 0.80 semantically — the agent correctly distinguishes between schema drift, bad ingestion, upstream bugs, and business logic failures.

One honest observation: the distinction between bad_ingestion and upstream_bug is genuinely ambiguous without full context about the source system. The agent defaults to bad_ingestion when there's no explicit mention of a source system bug. This is actually reasonable behavior — it matches what a human engineer would guess given the same information.

The Test Suite: 44 Tests Across 3 Layers

Three test suites, each testing a different layer:

Unit tests (16): Agent nodes tested in isolation with mocked Claude responses. Memory embedding tested for correct dimension, normalization, and determinism. Metadata store functions tested with mocked psycopg2. These run in 1.28 seconds with no Docker dependency.

Integration tests (20): End-to-end against the real Docker stack. Bad data seeded → GX detects failures → metadata DB updated → agent diagnoses → incident report written → status transitions verified.

Evaluation tests (8): GEval quality scoring across 30 golden scenarios. These are the tests that actually measure whether the AI layer is doing its job.

What I’d Do Differently

Real embeddings. The current implementation uses a deterministic hash-based pseudo-embedding for local development. In production, swap this for real semantic embeddings — either sentence-transformers running locally or a proper embeddings API. The Qdrant integration is already in place; it's a one-function change.

Conditional agent routing. The current graph is linear: all 5 nodes always run. A production agent would skip the fix suggestion step for unknown-category failures and route high-confidence diagnoses to an auto-remediation path that actually executes the SQL (with human approval).

Feedback loop. Engineers should be able to mark a fix suggestion as correct or incorrect. That feedback should update the Qdrant vectors so similar future failures get better diagnoses. Right now the memory only stores what failed — not whether the suggested fix worked.

Alerting. The pipeline detects and diagnoses, but doesn’t alert. A Slack webhook in the report_node would complete the loop — the on-call engineer gets a message with the root cause and fix SQL before they even open a laptop.

Key Takeaways

• Detection ≠ diagnosis. GX tells you what broke. The agent tells you why and what to do. Both layers are necessary.
• Explicit state machines beat black-box LLM calls. LangGraph’s 5-node graph is inspectable, debuggable, and independently testable. If one node misbehaves, you fix that node.
• Separate your metadata from your source data. Two PostgreSQL instances, strict separation. This is a production best practice, not over-engineering.
• Measure the AI layer. GEval gives you a number. “The agent seems to work” isn’t good enough for production — you need to know it works on 30 representative scenarios before you trust it on new ones.
• Cost is not a barrier. $0.50 a week to automatically diagnose every data quality failure your pipeline catches. The ROI on the first 3 AM page you don’t have to take is immediate.

If you’re looking for freelance or consultancy work on data pipelines, AI systems, or real-time platforms, reach out to me on X, or via email.

And, if this saved you time or sparked an idea, consider sponsoring me on GitHub. Thank you!

GitHub: https://github.com/dogukannulu

X: https://x.com/dogukannuluu

Email: dogukannulu@gmail.com

Repository

https://github.com/dogukannulu/data-lineage

Overview

I built a system that automatically investigates Airflow pipeline failures end to end: from the moment a DAG task fails, through a graph walk of the DBT lineage tree, all the way to a written root-cause report stored in PostgreSQL and tracked in MLflow. The agent runs in under 30 seconds at the median, produces a structured hypothesis chain, and gets evaluated by a 50-incident golden test set graded by Claude itself acting as a judge. The entire stack -8 services-is managed by both Docker Compose for day-to-day development and Terraform for reproducible infrastructure-as-code deployments.

This article is the full technical story of how I built it, every design decision I made and regretted, and what the numbers actually looked like when I ran the evaluation harness.

Tech Stack

Full list of every dependency with its purpose in the project:

• Apache Airflow 2.9 — orchestration of 12 client DAGs, each running twice daily, with on_failure_callback wired to fire the investigation webhook
• LangGraph — 6-node state machine that sequences the investigation from triage through to report writing
• Anthropic Claude claude-sonnet-4–6 — hypothesis generation in HYPOTHESIZE and LLM-as-judge scoring in the GEval harness
• FastAPI — webhook receiver on /webhook/airflow and REST query endpoints for investigations and lineage
• uvicorn — ASGI server running the FastAPI app inside Docker
• Neo4j 5 Community — graph database holding the DBT lineage graph (84 nodes, 216 edges across 12 clients)
• PostgreSQL 16 — relational store for pipeline metadata (5 tables, 34,534 seeded rows)
• SQLModel — typed ORM models that double as Pydantic validation schemas
• Alembic — database migrations applied on container startup
• MLflow — investigation run tracking without the heavy ML dependency tree
• Pydantic — data validation across all API models and agent state
• Gradio — 4-tab browser dashboard for browsing incidents and triggering manual investigations
• structlog — structured JSON logging throughout the backend
• Docker— orchestrates the 8 services in a single-command local environment
• Terraform (kreuzwerker/docker provider) — declarative infrastructure management; 16 resources across containers, volumes, images, and network; the same stack can be reproduced with terraform apply and torn down with terraform destroy
• DBT — 5 model files (raw → staging → intermediate → mart) replicated across 12 clients = 60 compiled models
• GEval (LLM-as-judge pattern, implemented directly) — 3-metric evaluation harness over 50 golden incidents

Introduction

The setup I was working with: a hypothetical SaaS analytics company called UDog Analytics that runs identical data pipelines for 12 clients, which I’ll call client_A through client_L. Each client has the same DBT model chain — raw extraction, staging, intermediate transformation, and two mart models — all orchestrated by Apache Airflow running twice daily. When something breaks in that chain, an on-call data engineer has to triage the failure, trace it through the lineage graph, form a hypothesis about the root cause, verify it against run history, write up a fix, and log everything somewhere.

The challenge was that this triage process was manual, repetitive, and slow. The same five failure patterns — schema changes, null check violations, freshness staleness, timeouts, and upstream dependency cascades — accounted for the vast majority of incidents. Someone had to look at the same Airflow logs, walk the same DBT lineage graph, and write the same fix suggestion dozens of times per month across a dozen clients.

I wanted to see how far I could push automation on this. Not just alerting — actual investigation. The system I built accepts the Airflow failure webhook, runs a 6-node LangGraph state machine, queries Neo4j for lineage context, pulls 30 days of historical run data from PostgreSQL, calls Claude to generate ranked hypotheses, verifies them with SQL queries, and writes a complete incident report in under 30 seconds.

The evaluation question was equally interesting: how do you measure whether an AI agent is giving good answers, not just any answers? The answer I landed on was a GEval harness — 50 hand-seeded golden incidents with known root causes and known fixes, evaluated across three Claude-judged metrics.

Here is what I built, in full detail.

Architecture Overview

The system runs as 8 Docker Compose services on a shared bridge network called udog_net:

1. PostgreSQL 16 (port 5433 external, 5432 internal) — the primary relational store. Holds pipeline run history, task-level results, DBT test results, the lineage node registry, and the incidents table. Both Airflow and the FastAPI backend connect to this instance.
2. Neo4j 5 Community (ports 7474 for browser, 7687 for Bolt) — the graph database holding the DBT lineage. Every DBT model and source for every client is a node. Edges are FEEDS_INTO (direct data flow), DEPENDS_ON (mirrors DBT ref() declarations), and UPSTREAM_OF (derived transitively). The graph has 84 nodes and 216 total edges.
3. Airflow webserver (port 8082) and Airflow scheduler — together they run the 12 client DAGs. The webserver is exposed for the UI; the scheduler runs in a separate container sharing the same Airflow metadata database.
4. Airflow init (one-shot container) — runs Alembic migrations on the Airflow metadata database and creates the admin user before the webserver starts.
5. FastAPI backend (port 8001 external) — the investigation engine. Receives webhooks from Airflow, runs the LangGraph state machine synchronously, and serves REST endpoints for querying incidents and lineage. The investigation runs synchronously inside the POST handler — the Airflow on_failure_callback already runs in a subprocess, so blocking it for the 5–30 seconds the investigation takes does not hold up the scheduler. The trade-off is a simpler response contract and easier testing.
6. MLflow (port 5050 external) — tracks every investigation as a run with params, metrics, and 3 artifacts. Uses SQLite as its backend store and a local volume for artifact storage.
7. Gradio dashboard (port 7860) — 4-tab browser UI for browsing incidents, looking up lineage chains, and manually triggering investigations.
8. MCP server (port 8090) — a lightweight FastAPI application that exposes 3 tools following the Model Context Protocol, making the investigation system accessible to Claude Code and similar LLM-powered developer tools.

The design rationale for this split: I kept Neo4j and PostgreSQL as separate databases rather than picking one for everything. Neo4j is the right tool for multi-hop graph traversal — asking “what are all the ancestors of this model up to 10 hops away” is a single Cypher MATCH with a variable-length path. Doing the same in PostgreSQL requires recursive CTEs that become difficult to reason about at depth. PostgreSQL is the right tool for relational queries: time-series run history, filtered DBT test results, incident indexing. Both databases play to their strengths.

Infrastructure as Code with Terraform

Beyond Docker Compose, the project includes a complete Terraform configuration in infra/docker/ using the kreuzwerker/docker provider. It manages 16 resources: 5 containers (PostgreSQL, Neo4j, MLflow, Airflow webserver and scheduler), 5 named volumes, 4 pulled images, and 1 bridge network.

The key design choice was using port offsets (+1 from the docker-compose defaults) so both stacks can run simultaneously on the same machine without conflicts. PostgreSQL gets port 5434 under Terraform vs 5433 under docker-compose; Neo4j browser gets 7475 vs 7474; and so on.

All sensitive values — passwords, API keys, Fernet keys — are declared as sensitive = true variables and read from terraform.tfvars, which is git-ignored. A terraform.tfvars.example template is committed for reference.

The full lifecycle is three commands:

terraform init
terraform plan # previews 16 resources across containers, volumes, images, network
terraform apply # provisions all infrastructure
terraform destroy

The Data Model

PostgreSQL holds 5 tables, all defined as SQLModel classes and migrated via Alembic.

Pipeline Runs (pipeline_runs):

• id — integer primary key
• dag_id — the Airflow DAG identifier, e.g. client_A__daily_pipeline
• run_id — unique Airflow run identifier string
• client_id — client_A through client_L
• status — enum: success, failed, running, skipped
• triggered_by — enum: schedule, manual, backfill
• started_at, ended_at — timestamps indexed for range queries
• duration_seconds — float, nullable

This table is the backbone of the PROFILE node. It holds 2,400 seeded rows covering a full calendar year of simulated runs across all 12 clients.

Pipeline Tasks (pipeline_tasks):

• id, run_id (FK), task_id — the Airflow task identifier
• status — task-level outcome
• started_at, ended_at, duration_seconds
• error_message — raw exception text, up to 4096 characters

DBT Test Results (dbt_test_results):

• id, run_id (FK), model_name — which DBT model the test ran against
• test_name, test_type — e.g. not_null, accepted_values
• status — enum: pass, fail, warn
• rows_failed — integer count of failing rows

Lineage Nodes (lineage_nodes):

• id, node_name (unique), node_type — Source, Model, or Metric
• client_id, owner_team, layer — raw, staging, intermediate, or mart
• last_run_status, neo4j_node_id — bridge to the graph database

Incidents (incidents):

• id, run_id (FK), triggered_at
• affected_models — JSON array of model name strings
• root_cause_type — enum with 8 values: schema_change, null_check, freshness, timeout, dependency_failure, data_volume, flaky, unknown
• root_cause_summary, fix_suggestion — text fields
• hypothesis_chain — JSON array of the full LangGraph hypothesis objects
• severity — P1, P2, or P3
• status — open, investigating, resolved, false_positive
• mlflow_run_id — links back to the MLflow tracking server
• similar_incident_ids — JSON array of related incident IDs for deduplication

The incidents table serves double duty: it is the operational output of the agent (every investigation writes a row here) and the ground truth store for the GEval evaluation harness (50 seeded rows with known root causes).

Building the Lineage Graph in Neo4j

The DBT project has 5 model files per client: raw_orders (Source), raw_inventory (Source), stg_orders, stg_inventory, int_order_fulfillment, mart_daily_revenue, and mart_inventory_health. With 12 clients that comes to 84 nodes and, across FEEDS_INTO, DEPENDS_ON, and transitively-derived UPSTREAM_OF edges, 216 total relationships.

The loader (lineage/loader.py) runs at startup and is idempotent — it uses Cypher MERGE statements so re-running it never creates duplicates. The edge structure per client looks like this:

raw_orders → stg_orders → int_order_fulfillment → mart_daily_revenue

raw_inventory → stg_inventory → int_order_fulfillment

raw_inventory → stg_inventory → mart_inventory_health

The primary Cypher query the TRAVERSE node uses to walk upstream from any failed model is:

MATCH path = (ancestor)-[:FEEDS_INTO*1..10]->(target {name: $name})
RETURN
ancestor.name AS name,
labels(ancestor)[0] AS node_type,
ancestor.layer AS layer,
ancestor.client_id AS client_id,
length(path) AS depth
ORDER BY depth ASC

This returns every ancestor node ordered by hop distance. A failure on mart_daily_revenue returns int_order_fulfillment at depth 1, stg_orders and stg_inventory at depth 2, and raw_orders and raw_inventory at depth 3. The TRAVERSE node also runs the inverse query to get the downstream blast radius — any model that depends on the failed model and would be stale once it fails.

The loader also derives UPSTREAM_OF edges transitively using a second Cypher pass:

MATCH path = (upstream {client_id: $client_id})-[:FEEDS_INTO*2..]->(downstream {client_id: $client_id})
MERGE (upstream)-[:UPSTREAM_OF]->(downstream)

This pre-materialises multi-hop ancestor relationships so some queries can use the direct UPSTREAM_OF edge instead of a variable-length path scan.

The LangGraph Investigation Agent-All 6 Nodes in Detail

The agent is implemented as a LangGraph StateGraph. Every node receives the full AgentState TypedDict and returns an updated copy. The state flows linearly from TRIAGE through VERIFY, then branches conditionally: if the confidence threshold is met, the graph routes to REPORT; if not, it routes to CLARIFY.

The AgentState type definition:

class Hypothesis(TypedDict):
    rank: int
    hypothesis: str
    confidence: float
    verified: bool
    evidence_query: str | None
    class AgentState(TypedDict, total=False):
    # Webhook input
    run_id: str
    dag_id: str
    client_id: str
    failed_task_id: str
    exception_message: str
    # TRIAGE output
    error_type: str
    severity: str # P1 / P2 / P3
    # TRAVERSE output
    lineage_chain: list[dict[str, Any]]
    affected_models: list[str]
    # PROFILE output
    run_history: list[dict[str, Any]]
    dbt_failures: list[dict[str, Any]]
    # HYPOTHESIZE output
    hypotheses: list[Hypothesis]
    # VERIFY output
    verified_cause: str
    fix_suggestion: str
    # REPORT output
    incident_id: int | None
    mlflow_run_id: str | None
    # CLARIFY branch
    needs_clarification: bool
    clarification_msg: str

Node 1: TRIAGE

TRIAGE is a pure Python function with no external I/O — it runs in microseconds. Its job is to classify the exception message into one of 5 error types and assign a P1/P2/P3 severity.

Classification uses a list of compiled regex patterns evaluated in order. The first match wins:

_ERROR_PATTERNS: list[tuple[re.Pattern, str]] = [
(re.compile(r"schema|column.*(not found|missing|mismatch)", re.I), "schema_change"),
(re.compile(r"not.null|null.*check|null.*value", re.I), "null_check"),
(re.compile(r"fresh|stale|sla|last.*loaded", re.I), "freshness"),
(re.compile(r"timeout|timed out|exceed.*limit|pool.*exhaust", re.I), "timeout"),
(re.compile(r"upstream.*fail|dependent.*not.*material|cannot execute", re.I), "dependency"),
]
_MART_TASKS = {"run_mart_daily_revenue", "run_mart_inventory_health"}
_INT_TASKS = {"run_int_order_fulfillment"}
def triage_node(state: AgentState) -> AgentState:
  exception_msg = state.get("exception_message", "")
  failed_task_id = state.get("failed_task_id", "")
  error_type = _classify_error(exception_msg)
  severity = _assign_severity(error_type, failed_task_id)
  return {**state, "error_type": error_type, "severity": severity}

Severity rules: P1 for schema_change or freshness failures in the mart layer (these directly break client-facing reports). P2 for null_check or dependency failures in intermediate or mart. P3 for everything else — timeouts, unknowns, and staging failures that have not yet propagated downstream.

Node 2: TRAVERSE

TRAVERSE queries Neo4j to populate two fields: lineage_chain (the upstream ancestors of the failed model) and affected_models (the downstream blast radius). It derives the DBT model name from the Airflow task ID by stripping the “run_” prefix and prepending the client ID. A failure on task run_mart_daily_revenue for client_A maps to model client_A__mart_daily_revenue.

The node calls three lineage queries in sequence: get_upstream_chain to walk ancestors, get_downstream_impact to enumerate affected models, and find_root_failure to identify the topmost node in the affected set that has no upstream predecessors among the failed models. Root failure nodes are prepended to the lineage chain so Claude has them at the top of its context window in HYPOTHESIZE.

Node 3: PROFILE

PROFILE pulls two data sets from PostgreSQL using synchronous SQLModel sessions: the past 30 days of pipeline_run records for the failing client (up to 200 rows), and all dbt_test_result rows with status FAIL for the affected models within the same 30-day window (up to 100 rows).

These two data sets give Claude the statistical context it needs: how often has this client’s pipeline failed recently, and which DBT data quality tests have been failing on the models in the blast radius? The PROFILE node computes total_runs, failed_runs, and failure_rate from the run history and passes these as summary statistics into the HYPOTHESIZE prompt.

Node 4: HYPOTHESIZE

HYPOTHESIZE is where the LLM reasoning happens. It constructs a structured prompt from all prior node outputs and calls Claude claude-sonnet-4–6. The prompt instructs Claude to respond with a JSON array only — no prose — and to rank hypotheses by confidence descending.

def hypothesize_node(state: AgentState) -> AgentState:
      run_history  = state.get("run_history", [])
      total_runs   = len(run_history)
      failed_runs  = sum(1 for r in run_history if r.get("status") == "failed")
      failure_rate = failed_runs / total_runs if total_runs else 0.0

      prompt = _USER_TEMPLATE.format(
          max_hypotheses=settings.agent_max_hypotheses,
          dag_id=state.get("dag_id", ""),
          client_id=state.get("client_id", ""),
          failed_task_id=state.get("failed_task_id", ""),
          error_type=state.get("error_type", "unknown"),
          severity=state.get("severity", "P3"),
          exception_message=state.get("exception_message", ""),
          lineage_chain=json.dumps(state.get("lineage_chain", []), indent=2),
          affected_models=json.dumps(state.get("affected_models", []), indent=2),
          dbt_failures=json.dumps(state.get("dbt_failures", [])[:10], indent=2),
          total_runs=total_runs,
          failed_runs=failed_runs,
          failure_rate=failure_rate,
      )

      message = _client.messages.create(
          model=settings.anthropic_model,
          max_tokens=1024,
          system=_SYSTEM_PROMPT,
          messages=[{"role": "user", "content": prompt}],
      )

      raw = message.content[0].text.strip()
      # Strip markdown code fences if Claude wrapped the JSON
      if raw.startswith("```"):
          raw = raw.split("```", 2)[1]
          if raw.startswith("json"):
              raw = raw[4:]
          raw = raw.strip()

      hypotheses: list[Hypothesis] = json.loads(raw)
      for h in hypotheses:
          h.setdefault("verified", False)
          h.setdefault("evidence_query", None)

      return {**state, "hypotheses": hypotheses}

The system prompt sets Claude’s persona as a senior data reliability engineer at UDog Analytics and instructs it to respond with valid JSON only. The markdown fence stripping handles a common Claude behaviour where it wraps JSON responses in triple-backtick blocks even when told not to — this was particularly necessary when using Claude Haiku as the GEval judge model.

Each hypothesis in the returned array has a rank, a hypothesis string, a confidence float between 0 and 1, a verified boolean (initially false), and an evidence_query field — either null or a SQL SELECT statement that VERIFY can run to confirm the hypothesis.

Node 5: VERIFY

VERIFY iterates through the hypotheses and executes any non-null evidence_query against PostgreSQL. Before executing any query, it runs it through _is_safe_query(), which enforces that only SELECT statements are permitted.

_SELECT_ONLY = re.compile(r"^\s*SELECT\b", re.I)
  _CONFIDENCE_THRESHOLD = 0.55

  def _is_safe_query(query: str) -> bool:
      return bool(_SELECT_ONLY.match(query.strip()))

  def verify_node(state: AgentState) -> AgentState:
      hypotheses: list[Hypothesis] = state.get("hypotheses", [])
      updated: list[Hypothesis] = []

      with Session(_sync_engine) as session:
          for h in hypotheses:
              query = h.get("evidence_query")
              if query:
                  evidence = _run_evidence_query(query, session)
                  h = dict(h)
                  h["verified"] = bool(evidence and "error" not in evidence)
              updated.append(h)

      verified_cause, confidence = _pick_verified_cause(updated)
      needs_clarification = confidence < _CONFIDENCE_THRESHOLD
      fix_suggestion = _build_fix_suggestion(verified_cause, state)

      return {
          **state,
          "hypotheses": updated,
          "verified_cause": verified_cause,
          "fix_suggestion": fix_suggestion,
          "needs_clarification": needs_clarification,
      }

The confidence threshold is 0.55. If the best hypothesis after SQL verification still has confidence below that value, needs_clarification is set to True and the graph routes to CLARIFY instead of REPORT. Fix suggestions are generated from a template dictionary keyed on the verified_cause string — for example, a schema_change fix suggests updating the failing model’s SQL to align with the new upstream schema and re-running the staging layer.

Node 6: REPORT

REPORT writes the complete incident record to the incidents table in PostgreSQL. It maps the severity string (P1/P2/P3) and verified_cause string to their enum equivalents, builds a human-readable root_cause_summary string, and persists the full hypothesis chain as a JSON column.

The REPORT node also links the incident to its MLflow run via the mlflow_run_id field on the AgentState, which was set by the FastAPI webhook handler before invoking the graph. An important implementation detail: when the webhook receives a simulated or new run_id that does not exist in pipeline_runs, a helper function _ensure_pipeline_run() creates that record first. Without this, the _resolve_run_id lookup in REPORT silently falls back to run_id=1, causing GET /investigations/{run_id} to return 404 for any non-seeded investigation.

The CLARIFY Node

CLARIFY is reached when confidence is too low to commit to a root cause. It extracts the top hypothesis by confidence, formats a clarification message including the hypothesis text and confidence percentage, and writes it to state. The graph terminates after CLARIFY — no incident row is written. The FastAPI response signals status: “clarification_needed” back to the Airflow callback, and the Gradio dashboard surfaces the clarification message for the on-call engineer to act on.

The LangGraph Graph Assembly

The graph wires the 6 nodes with linear edges for the happy path and a conditional branch after VERIFY:

def _route_after_verify(state: AgentState) -> str:
      if state.get("needs_clarification", False):
          return "clarify"
      return "report"

  def build_investigation_graph() -> StateGraph:
      graph = StateGraph(AgentState)
      graph.add_node("triage",      triage_node)
      graph.add_node("traverse",    traverse_node)
      graph.add_node("profile",     profile_node)
      graph.add_node("hypothesize", hypothesize_node)
      graph.add_node("verify",      verify_node_with_mlflow)
      graph.add_node("report",      report_node)
      graph.add_node("clarify",     clarify_node)

      graph.add_edge(START, "triage")
      graph.add_edge("triage",      "traverse")
      graph.add_edge("traverse",    "profile")
      graph.add_edge("profile",     "hypothesize")
      graph.add_edge("hypothesize", "verify")

      graph.add_conditional_edges(
          "verify",
          _route_after_verify,
          {"report": "report", "clarify": "clarify"},
      )

      graph.add_edge("report",  END)
      graph.add_edge("clarify", END)
      return graph.compile()

  investigation_graph = build_investigation_graph()

The compiled graph is a module-level singleton. The FastAPI webhook handler imports it directly and calls investigation_graph.invoke(initial_state) — no instantiation overhead per request.

The FastAPI Backend

The webhook receiver at POST /webhook/airflow is the entry point for all investigations. It accepts the AirflowWebhookPayload, ensures the pipeline run record exists in PostgreSQL, constructs the initial AgentState, starts an MLflow run via InvestigationTracker, invokes the LangGraph graph synchronously, logs the final state to MLflow, and returns the WebhookResponse. The route returns HTTP 202 Accepted to signal that the work has been acknowledged, but the investigation itself completes before the response is sent.

The Airflow on_failure_callback that fires the webhook:

def _fire_webhook(context: dict) -> None:
      ti = context["task_instance"]
      payload = {
          "dag_id":          context["dag"].dag_id,
          "run_id":          context["run_id"],
          "client_id":       context["dag"].dag_id.split("__")[0],
          "task_id":         ti.task_id,
          "execution_date":  context["execution_date"].isoformat(),
          "exception":       str(context.get("exception", "")),
          "log_url":         ti.log_url,
      }
      try:
          resp = requests.post(
              f"{BACKEND_URL}/webhook/airflow",
              json=payload,
              headers={"X-API-Key": API_KEY},
              timeout=10,
          )
          resp.raise_for_status()
          log.info("Webhook fired successfully: %s", resp.json())
      except requests.RequestException as exc:
          log.error("Failed to fire investigation webhook: %s", exc)

The client_id is derived by splitting the DAG ID on the double underscore separator — client_A__daily_pipeline becomes client_A. This convention is baked into the DAG factory that generates all 12 DAGs from a single _make_dag(client_id) function.

Authentication uses a shared API key passed as the X-API-Key header. The FastAPI deps.py verifies this on every route. The key is mounted from a .env file via pydantic-settings. No secrets are hardcoded anywhere in the source.

Additional REST routes handle querying: GET /investigations/ (list with client_id and days filters), GET /investigations/{run_id} (fetch by Airflow run ID), and GET /lineage/{model_name}/upstream|downstream|subgraph.

MLflow Observability

Every investigation produces one MLflow run with the following structure:

Params (logged at run start and updated after triage):

• run_id, dag_id, client_id — webhook identifiers

• error_type, severity, verified_cause — set after the relevant nodes complete

Metrics:

• lineage_depth — number of nodes in the upstream lineage chain

• hypotheses_generated — count of hypotheses Claude produced

• hypothesis_confidence_max — highest confidence value across all hypotheses

• investigation_duration_s — wall-clock time from MLflow run start to log_final_state call

Artifacts (3 files logged to the run):

1. full_incident_report.json — a flattened serialisation of the complete AgentState: all lineage nodes, the full run history summary, every hypothesis, the verified cause, and the fix suggestion. This is the primary debugging artifact.
2. hypothesis_chain.json — just the hypotheses list, making it easy to compare hypothesis quality across different incident types without loading the full report.
3. lineage_subgraph.graphml — a NetworkX directed graph of the lineage chain and affected models exported in GraphML format. This can be loaded directly into Gephi or any other graph analysis tool for visual exploration.

Using MLflow for agent observability rather than just for ML model training turned out to be a natural fit. The params/metrics/artifacts structure maps cleanly onto an investigation: params are the context the agent received, metrics quantify the depth and quality of the investigation, and artifacts are the structured outputs. The MLflow UI gives you a dashboard where you can sort investigations by hypothesis_confidence_max to find the uncertain cases, or by investigation_duration_s to find the slow ones.

The evaluation harness logs its results to a separate geval-evaluation experiment in the same MLflow instance, so you can compare individual investigation quality against aggregate evaluation metrics in one place.

The Gradio Dashboard

The Gradio dashboard has 4 tabs:

Tab 1: Recent Incidents. A dropdown to select a client (or “All clients”) and a day-range slider from 1 to 30. On refresh, it fetches up to 50 incidents from the backend and renders a severity-tagged summary list. The first 5 incidents are also shown as raw JSON in a code block for debugging.

Tab 2: Incident Detail. A text input for an Airflow run_id. On fetch, it renders a full incident report including severity, root cause type, affected models, root cause summary, and fix suggestion. Below it shows the hypothesis chain as JSON — the ranked list of hypotheses Claude generated, with their confidence scores and verification status.

Tab 3: Lineage Lookup. A model name input (e.g. client_A__mart_daily_revenue), a direction dropdown (upstream ancestors, downstream blast radius, or full subgraph), and a depth slider. Returns the lineage chain as JSON, which is useful for ad hoc lineage exploration without opening the Neo4j browser.

Tab 4: Trigger Investigation. A manual investigation trigger for testing and demonstration. Select a client, an error type, and a failing task, then click the button. The dashboard generates a synthetic run_id, fires the webhook, and shows the response including the incident ID and MLflow run ID.

The MCP Server

MCP stands for Model Context Protocol — a standard for exposing tools to LLM-powered developer assistants in a way they can discover and invoke programmatically. The idea is that instead of writing a one-off script to query your investigation system, you register it as an MCP tool and any compatible assistant (Claude Code, for example) can call it directly from a conversation.

For data engineers this is significant. It means Claude Code can answer questions like “what failed in client_B’s pipeline yesterday?” or “show me the upstream lineage for mart_daily_revenue” without any copy-pasting of run IDs or manually navigating the Gradio UI. The engineer describes what they want; the assistant calls the right tool and surfaces the structured result.

The MCP server exposes 3 tools at port 8090:

Tool 1: investigate. Takes a run_id and optionally client_id, dag_id, task_id, and exception. First checks whether an investigation for this run_id already exists in the backend (GET /investigations/{run_id}). If it does, returns the cached report. If not, and if client_id and dag_id are provided, fires a synthetic webhook to trigger a new investigation. This tool is idempotent — calling it twice on the same run_id returns the same report.

Tool 2: lineage_lookup. Takes a model_name, a direction (upstream, downstream, or subgraph), and a depth. Proxies to the corresponding FastAPI lineage route. Returns the full lineage chain as a JSON object that the LLM can reason about inline.

Tool 3: incident_history. Takes a client_id, a days lookback window, and a result limit. Returns recent incidents for the client filtered by the time window. Useful for patterns like “has client_F had more schema_change failures than usual this week?”

All three tools are thin proxies over the FastAPI backend — they add no logic of their own, just translate MCP-format requests into backend HTTP calls and return the results. The MCP manifest is served at the root GET / endpoint and declares the tool schemas in JSON Schema format.

GEval Evaluation Harness

The hardest part of building an AI agent is not making it produce output — it is making it produce correct output, and knowing when it does not. For a data reliability agent, correctness means: did it identify the right root cause, did it trace the lineage correctly, and did it suggest a useful fix?

I used the GEval pattern — LLM-as-judge evaluation — implemented directly rather than through a third-party library. The harness lives in backend/app/evaluation/harness.py and runs against 50 golden incidents seeded into the PostgreSQL incidents table via scripts/seed.py. Each golden incident has a known root_cause_type, a known list of affected models (the first entry being the true root model), and a known fix_suggestion written by a human.

The 50 incidents are distributed across all 5 failure patterns (schema_change, null_check, freshness, timeout, dependency), all 12 clients, and both mart-layer and intermediate-layer failures, giving a representative sample of the full failure space the agent encounters.

To reduce evaluation cost, the LLM judge uses Claude Haiku (claude-haiku-4–5–20251001) rather than Sonnet. The judge is called once per incident for the Fix Relevance metric; Root Cause Accuracy and Lineage Correctness are computed without LLM calls. Running 50 cases with the Haiku judge costs approximately $0.50.

Three primary metrics govern whether the harness passes:

Metric 1: Root Cause Accuracy (target ≥ 80%). This is an exact string match between the agent’s verified_cause field and the golden ground truth root_cause_type. There is no partial credit — if the agent returns “timeout” for a “schema_change” incident, it counts as a miss. This is the strictest metric and the hardest to game, because the classification taxonomy is closed: 8 possible values, one correct answer per incident.

Metric 2: Lineage Correctness (target ≥ 90%). This checks whether the true root model name (the first entry in the golden incident’s affected_models list) appears anywhere in the agent’s lineage_chain output. The match is a substring check to handle naming variations. This is more lenient than Root Cause Accuracy because it only requires that the lineage traversal reached the right part of the graph, not that it identified the precise cause.

Metric 3: Fix Relevance (target ≥ 75%). This is the LLM-as-judge metric. Claude Haiku is given the agent’s fix_suggestion and the golden fix_suggestion side by side and asked to score the agent’s fix from 0.0 to 1.0 on relevance and correctness. The prompt asks Claude to respond with a JSON object containing a score and a brief reason. The average of these scores across all 50 incidents must reach 75%. This metric captures whether the fix is actionable, not just whether it mentions the right error type.

Results are logged to the geval-evaluation MLflow experiment after every harness run, making it straightforward to track evaluation quality across prompt iterations.

Unit Tests

Integration Tests

Results and Lessons

The system does what it was designed to do: take a structured failure signal from Airflow and produce a structured root cause report without human intervention, in under 30 seconds, with traceable reasoning.

A few things I would do differently:

The synchronous webhook handler keeps the response contract simple but means the Airflow callback waits for the full investigation. For a production deployment where investigations routinely exceed 30 seconds or where multiple failures arrive simultaneously, moving to a background task queue (Celery or similar) with a polling endpoint would be the right call.

The VERIFY node’s evidence queries are generated by Claude against a database schema it can’t see directly — it infers column names from the hypothesis context. This works well for common patterns but fails on queries that reference columns that exist in a different form than Claude expects. Passing an explicit schema summary in the HYPOTHESIZE prompt would improve verification accuracy.

The Terraform configuration covers the infrastructure services but not the custom-built containers (backend, Gradio, MCP server), since those require docker build context. A complete Terraform solution would use the docker_registry_image resource to build and push images to a local registry first. For this project, the split is clean: Terraform owns the data infrastructure, Docker Compose owns the full application stack.

The GEval harness runs 50 cases in a single synchronous pass. Parallelising across incidents with a thread pool would cut evaluation wall-clock time by 5–10x.

If you’re looking for freelance or consultancy work on data pipelines, AI systems, or real-time platforms, reach out to me on X, or via email.

And, if this saved you time or sparked an idea, consider sponsoring me on GitHub. Thank you!

GitHub: https://github.com/dogukannulu

X: https://x.com/dogukannuluu

Email: dogukannulu@gmail.com

Repository

https://github.com/dogukannulu/langgraph-chatbot-tavily-sqlite

Tech Stack

• LangChain
• LangGraph
• Tavily Search
• SQLite
• Black
• Google Gemini

Overview

• As the beginning of this project, we are going first to leverage uv and Makefile to automate processes.
• Then, create a virtual environment and install all requirements.
• Format all scripts with black.
• Create and run environment test script.
• Create and run demo script to call the model.
• Then, we will create LangGraph graph that will work as a chatbot with tool calling.
• We will create nodes, edges and will use conditional routing.
• We will also keep chat memory using SQLite and keep threads IDs there.

Introduction

In this comprehensive guide, we’ll build a sophisticated chatbot using LangGraph that features persistent memory, tool integration, and routing capabilities. Our chatbot will use Google’s Gemini model, integrate with Tavily for web search, maintain conversation history in SQLite, and provide a polished user experience.

Prerequisites and Tools Overview

What is UV?

UV is a modern Python package manager that’s significantly faster than pip. Developed by the creators of Ruff, UV can be 10–100x faster than traditional pip installations. Key benefits include:

• Speed: Ultra-fast package resolution and installation
• Reliability: Better dependency resolution than pip
• Modern: Built with Rust for performance
• Compatible: Drop-in replacement for pip in most cases

UV handles virtual environments, package installation, and dependency management more efficiently than traditional Python tools. We are going to use it for package installation, venv creation and script running.

You may find more information in this repo.

What is a Makefile?

A Makefile is a file containing a set of directives used by the make utility to automate tasks. In Python projects, Makefiles help standardize common development tasks:

• Task Automation: Run complex commands with simple make commands
• Consistency: Ensure all developers use the same commands
• Documentation: Serve as living documentation of common tasks
• Cross-platform: Work across different operating systems

Example: Instead of remembering uv pip install -r requirements.txt, we will simply run make install.

Core Technologies Overview

LangChain

LangChain is a framework for developing applications powered by language models. It provides:

• Chains: Link multiple LLM calls together
• Memory: Maintain conversation context
• Tools: Integrate external APIs and services
• Agents: Create autonomous decision-making systems

LangGraph

LangGraph is LangChain’s library for building stateful, multi-actor applications with LLMs. Key features:

• Graph-based Architecture: Define workflows as directed graphs
• State Management: Persistent state across conversation turns
• Conditional Routing: Dynamic flow control based on conditions
• Checkpoints: Save and restore conversation state

Tavily Search

Tavily is a search API designed specifically for AI applications:

• AI-Optimized: Results formatted for LLM consumption
• Real-time: Access to current web information
• Reliable: High availability and performance
• Easy Integration: Simple API with LangChain support

Now, we can start implementing each step of the project.

Get API Keys

Google API Key (Gemini)

1. Visit Google AI Studio: Go to https://aistudio.google.com/app/apikey
2. Sign in: Use your Google account
3. Create API Key: Click “Create API Key”
4. Copy Key: Save the generated key securely in your local
5. Set Billing: Ensure billing is enabled for your Google Cloud project

Tavily API Key

1. Visit Tavily: Go to https://tavily.com/
2. Sign Up: Create an account
3. Dashboard: Access your dashboard after verification
4. API Keys: Navigate to the API keys section
5. Generate Key: Create a new API key and save it securely in your local
6. Free Tier: Tavily offers free searches per month

Project Setup

Directory Structure

Let’s start by understanding our project structure:

advanced_chatbot/
├── main.py                          # Main application
├── requirements.txt                 # Dependencies
├── env_template.txt                # Environment template
├── test_development_setup.py       # Setup verification
├── demo_chatbot.py                 # Demo script
├── Makefile                        # Task automation
├── README.md                       # Documentation
├── chatbot.log                     # Generated logs
└── chatbot_checkpoint.db           # Generated database

Installation Process with Makefile

Our project includes a comprehensive Makefile that automates all setup and development tasks. You may run make help to see all commands it has by running make help. Let’s go through the complete setup process:

Step 1: Install UV Package Manager

You may refer to Makefile here.

# Use Makefile to install UV automatically
make install-uv

This command:

• Checks if UV is already installed
• Installs UV using Homebrew if available
• Provides instructions if Homebrew is not found
• Confirms successful installation with version info

Step 2: Create Virtual Environment

# Create virtual environment named 'chatbot'
make create-venv

This creates a new virtual environment called chatbot using UV, which is much faster than traditional python -m venv.

Step 3: Activate Virtual Environment

# Activate the virtual environment (this step cannot be automated by Makefile)
source chatbot/bin/activate

Note: Makefile cannot activate virtual environments directly due to shell limitations, so this must be done manually.

Step 4: Install Dependencies

# Install all project dependencies in requirements.txt
make install

This command:

• Checks if virtual environment is activated
• Uses UV if available (much faster), falls back to pip
• Installs all dependencies from requirements.txt
• Provides helpful error messages if environment isn’t activated

Step 5: Configure Environment Variables

Create a .env file from the template:

cp env_template.txt .env

Edit .env with your API keys:

# Google API Configuration
GOOGLE_API_KEY=your_actual_google_api_key_here
# Tavily Search API Configuration  
TAVILY_API_KEY=your_actual_tavily_api_key_here
# Optional configurations
CHECKPOINT_DB_PATH=chatbot_checkpoint.db
LOG_LEVEL=INFO
LOG_FILE=chatbot.log

Dependencies Analysis

Let’s examine our requirements.txt:

# Core LangChain and LangGraph dependencies
langchain>=0.3.0                    # Main framework
langchain-google-genai>=2.0.0       # Google Gemini integration
langchain-community>=0.3.0          # Community tools
langgraph>=0.2.0                    # Graph workflows
langgraph-checkpoint-sqlite>=0.1.0  # SQLite persistence
# Search and web tools
langchain-tavily>=0.1.0             # Web search
# Environment and configuration
python-dotenv>=1.0.0                # Environment variables
# Additional utilities
beautifulsoup4>=4.12.0              # HTML parsing
requests>=2.31.0                    # HTTP requests
# Development tools
black>=24.0.0                       # Code formatting

Development Workflow with Makefile

As the next step, you may create your own project files by obtaining them from the repo. Now, we are going to format all the scripts existing in our project.

Install Development Tools

# Install development dependencies (to format all our scripts)
make dev-install

This installs additional tool black for code formatting.

Code Formatting

# Auto-format all Python code with black
make format

Testing Setup

# Run setup verification tests
make test

This runs test_development_setup.py which verifies:

• All imports work correctly
• Environment configuration is proper
• Database connections function
• Main script components load successfully

So far, we created and activated our virtual environment, installed all requirements, formatted our code and tested imports. Now, we will move on with main chatbot application with Langchain/LangGraph

Core Implementation

State Management Design

Our chatbot uses a TypedDict to define its state structure. We are going to use this class for the state structure of our chatbot.

from typing import Annotated, Dict, Any
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    """
    State structure for our chatbot.
    
    Key Features:
    - messages: Conversation history with automatic message addition
    - session_info: Metadata and statistics
    - thread_id: Unique identifier for memory persistence
    """
    messages: Annotated[list, add_messages]  # Auto-manages message history
    session_info: Dict[str, Any]             # Session metadata
    thread_id: str                           # Memory persistence key

Key Design Decisions:

1. Annotated Messages: The add_messages annotation automatically handles message list management
2. Session Info: Tracks statistics like message count and tool usage. We are also going to log all this information
3. Thread ID: Enables conversation persistence across sessions. By using thread ID, we will be able to continue our chats with chat memory.

Logging System

We should implement a comprehensive logging since we would like to see the AI response, total message count, thread ID, whether tool was called per prompt or not, etc.

class ColoredFormatter(logging.Formatter):
    """Custom colored formatter for better log readability."""
    
    COLORS = {
        'DEBUG': '\033[36m',    # Cyan
        'INFO': '\033[32m',     # Green  
        'WARNING': '\033[33m',  # Yellow
        'ERROR': '\033[31m',    # Red
        'CRITICAL': '\033[35m', # Magenta
    }
    RESET = '\033[0m'
    
    def format(self, record):
        log_color = self.COLORS.get(record.levelname, self.RESET)
        record.levelname = f"{log_color}{record.levelname}{self.RESET}"
        return super().format(record)
    
    def setup_logging():
        """Setup enhanced logging with file and console handlers."""
        logger = logging.getLogger(__name__)
        logger.setLevel(logging.INFO)
        
        # File handler for persistent logs
        file_handler = logging.FileHandler("chatbot.log")
        file_formatter = logging.Formatter(
            "%(asctime)s | %(name)s | %(levelname)s | %(funcName)s:%(lineno)d | %(message)s"
        )
        file_handler.setFormatter(file_formatter)
        
        # Console handler with colors
        console_handler = logging.StreamHandler()
        console_formatter = ColoredFormatter(
            "%(asctime)s | %(levelname)s | %(funcName)s | %(message)s",
            datefmt="%H:%M:%S"
        )
        console_handler.setFormatter(console_formatter)
        
        logger.addHandler(file_handler)
        logger.addHandler(console_handler)
        
        return logger

Logging Features:

• Dual Output: Both file and console logging (file logging inside chatbot.log)
• Color Coding: Different colors for different log levels for better visibility
• Detailed Format: Includes timestamp, function name, and line number
• Structured Tracking: Every operation is logged for debugging

LLM Initialization

We initialize Google’s Gemini 2.5 Flash model. The reason of using this model is about the pricing and the responses are still satisfactory.

def initialize_llm():
    """Initialize the LLM with comprehensive error handling."""
    logger.info("Initializing Google Gemini 2.5 Flash model...")
    
    api_key = os.getenv("GOOGLE_API_KEY")
    if not api_key:
        logger.error("GOOGLE_API_KEY not found in environment variables")
        raise ValueError("GOOGLE_API_KEY not found")
    
    try:
        llm = init_chat_model(
            "gemini-2.5-flash", 
            model_provider="google_genai", 
            api_key=api_key,
            temperature=0.7  # Balanced creativity and consistency
        )
        logger.info("Successfully initialized Google Gemini 2.5 Flash model")
        return llm
    except Exception as e:
        logger.error(f"Failed to initialize LLM: {str(e)}")
        raise

Key Features:

• Environment Validation: Checks for API key presence
• Error Handling: Graceful failure with detailed logging
• Configuration: Temperature set to 0.7 for balanced responses including both creativity and consistency

Tool Integration

Our chatbot integrates Tavily search as an optional tool. Afterwards, we are going to leverage conditional routing. If the answer requires model to search web , it is going to fork into tool calling and chatbot will use Tavily search.

def initialize_tools():
    """Initialize tools with enhanced logging."""
    logger.info("Initializing available tools...")
    tools = []
    
    # Add Tavily search tool if API key is available
    tavily_api_key = os.getenv("TAVILY_API_KEY")
    if tavily_api_key:
        try:
            search_tool = TavilySearch(
                max_results=3,
                description="Search the web for current information, news, and real-time data. Use this when you need up-to-date information that you don't have in your training data.",
            )
            tools.append(search_tool)
            logger.info("Successfully added Tavily search tool")
        except Exception as e:
            logger.error(f"Failed to initialize Tavily search: {str(e)}")
    else:
        logger.warning("TAVILY_API_KEY not found - web search functionality disabled")
    
    logger.info(f"Tool initialization complete. Available tools: {len(tools)}")
    return tools

Tool Features:

• Optional Integration: Functions with or without Tavily API key
• Smart Configuration: Limited to 3 results for concise responses
• Descriptive: Clear description helps LLM decide when to use
• Graceful Degradation: Works even if search is unavailable

Graph Architecture Deep Dive

Node Implementation

Statistics Update Node

We are going to keep the statistics of the chat using this node. With every new message, the stats are updated and we also log those into the console as well as the log file.

def update_session_stats(state: State):
    """Update session statistics and user profile."""
    logger.debug("Updating session statistics...")
    
    # Initialize or update session info
    session_info = state.get("session_info", {})
    if "message_count" not in session_info:
        session_info["start_time"] = datetime.now().isoformat()
        session_info["message_count"] = 0
        session_info["tool_usage_count"] = 0
    
    session_info["message_count"] += 1
    session_info["last_activity"] = datetime.now().isoformat()
    
    # Count tool usage in recent messages
    messages = state.get("messages", [])
    if messages:
        last_message = messages[-1]
        if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
            session_info["tool_usage_count"] += len(last_message.tool_calls)
    
    return {"session_info": session_info}

Main Chatbot Node

We use this node as the main chatbot. This node is going to receive the Human prompt and provide the response accordingly.

def chatbot_node(state: State):
    """Main chatbot processing node with enhanced error handling."""
    logger.info("Processing user message through chatbot node...")
    
    try:
        # Generate dynamic system prompt
        system_prompt = get_system_prompt(state)
        
        # Prepare messages for LLM
        messages = [SystemMessage(content=system_prompt)] + state["messages"]
        
        # Get LLM response
        response = llm_with_tools.invoke(messages)
        
        # Log response details
        if hasattr(response, 'tool_calls') and response.tool_calls:
            tool_names = [tc.get("name", "unknown") for tc in response.tool_calls]
            logger.info(f"LLM requested tool usage: {', '.join(tool_names)}")
        
        return {"messages": [response]}
        
    except Exception as e:
        logger.error(f"Error in chatbot_node: {str(e)}")
        error_response = AIMessage(
            content="I apologize, but I encountered an error processing your request."
        )
        return {"messages": [error_response]}

Smart Routing Logic

Using this node, the graph decides whether to use a tool (Tavily) or not. If the tool is required, it is going to look for an answer by searching web.

def should_use_tools(state: State) -> str:
    """Determine if we should route to tools or end the conversation."""
    logger.debug("Evaluating routing decision...")
    
    messages = state.get("messages", [])
    if not messages:
        return "end"
    
    last_message = messages[-1]
    
    # Check if LLM requested tools
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        logger.info(f"Routing to tools - {len(last_message.tool_calls)} tool calls requested")
        return "tools"
    
    logger.debug("No tools requested, routing to end")
    return "end"

Graph Construction

This is going to be the final step. By using this function, we create the chatbot graph with all of the above nodes included.

def build_chatbot_graph():
    """Build the complete chatbot graph with SQLite persistence."""
    logger.info("Building LangGraph chatbot workflow...")
    
    # Create StateGraph
    graph_builder = StateGraph(State)
    
    # Add nodes
    graph_builder.add_node("update_stats", update_session_stats)
    graph_builder.add_node("chatbot", chatbot_node)
    
    # Add tools node if available
    if tools:
        tool_node = ToolNode(tools)
        graph_builder.add_node("tools", tool_node)
    
    # Define edges and routing
    graph_builder.add_edge(START, "update_stats")
    graph_builder.add_edge("update_stats", "chatbot")
    
    # Conditional routing from chatbot
    if tools:
        graph_builder.add_conditional_edges(
            "chatbot",
            should_use_tools,
            {"tools": "tools", "end": END}
        )
        graph_builder.add_edge("tools", "chatbot")
    else:
        graph_builder.add_edge("chatbot", END)
    
    # Initialize SQLite checkpointer for memory persistence
    db_path = "chatbot_checkpoint.db"
    checkpointer = SqliteSaver(
        sqlite3.connect(db_path, check_same_thread=False)
    )
    
    # Compile graph
    compiled_graph = graph_builder.compile(checkpointer=checkpointer)
    logger.info("LangGraph workflow compiled successfully")
    
    return compiled_graph

Graph Flow:

You may see a simple visualization on how the whole graph looks like.

START → update_stats → chatbot → [conditional routing]
                                      ↓
                               tools → chatbot (loop back)
                                      ↓  
                                     END

Advanced Features

Dynamic System Prompt Generation

A system prompt is a special instruction given to an AI model before the conversation starts. The purpose is to set the rules, goals, and behavior of the AI throughout the interaction. It will be really useful for the accuracy of the response.

def get_system_prompt(state: State) -> str:
    """Generate context-aware system prompt."""
    session_info = state.get("session_info", {})
    
    system_prompt = """You are a helpful, knowledgeable, and friendly AI assistant.
      CORE PRINCIPLES:
      - Be conversational and natural in your responses
      - Provide accurate, helpful information
      - Ask clarifying questions when needed
      - Maintain context from our conversation history
      TOOL USAGE:
      - You have access to web search capabilities
      - Use search tools ONLY when you need current information or real-time data
      - If you can answer from your knowledge, do so directly
      - When using tools, briefly explain what you're searching for
      CONVERSATION GUIDELINES:
      - Reference previous parts of our conversation when relevant
      - Be concise but thorough
      - Use a friendly, approachable tone
      - Adapt your explanations to the user's apparent level of understanding
      """
    
    if session_info.get("message_count", 0) > 1:
        system_prompt += f"\n- This is message #{session_info['message_count']} in our conversation"
    
    return system_prompt

Memory Persistence with SQLite

Our chatbot uses SQLite checkpoints to maintain conversation memory. With every prompt without providing the thread ID, a new ID will be created and saved into SQLite. Chatbot will display the specific thread ID in the console. We can set it as an argument for chatbot to be able to remember earlier prompts.

# Each conversation has a unique thread_id
thread_id = args.thread or f"chat_{uuid.uuid4().hex[:8]}"
# Configuration for memory persistence
config = {"configurable": {"thread_id": thread_id}}
# Graph execution with memory
result = graph.invoke(inputs, config)

Memory Features:

• Thread-based: Each conversation thread has isolated memory
• Persistent: Conversations survive application restarts
• Automatic: No manual checkpoint management required
• Efficient: SQLite provides fast, reliable storage

Next Steps

Since we went through each step of the main script, we can move forward with next steps to run the scripts.

Running Demo

# Run feature demonstration
make demo

This executes demo_chatbot.py which showcases all key features without requiring user input. It will provide a simple “Hello” as the user input and check if the chatbot script runs properly and provides response.

Cleaning Generated Files

# Clean all generated files (logs, database, cache)
make clean

This removes:

• chatbot.log (log files)
• chatbot_checkpoint.db (SQLite database)
• __pycache__ directories
• .pytest_cache directories
• All .pyc and .pyo files

Running the Chatbot

Since we completed all previous steps, we can run the main script. We have two way of doing this. We can either leverage Makefile again or running commands ourselves.

Makefile:

# Single message mode
# It will provide a single "Hello" message to the model and get an answer
make run

# Interactive conversation mode
# By running this, we can start an interactive chat with a simple "Hello" input
make run-interactive

Terminal Commands:

# Start conversation
uv run main.py --message "Hello"

# You will get the first response
# If you want to keep the chat memory, you can provide the next message with thread
# You may see thread ID in the logs in console or in console.log (see image below)
uv run main.py --thread chat_732b9e45 --message "Next Message"


# Or if you want to start an interactive chat, run this command
# Then, you can keep sending prompts
uv run main.py --message 'Your message' --interactive

Usage Examples

In this last section, we will go through some sample messages.

Single Message Mode

# Using Makefile command
make run
# Or direct Python execution
uv run main.py --message "What's the latest news about AI?"

Interactive Mode

# Using Makefile command
make run-interactive
# Or direct Python execution
uv run main.py --message "Hello!" --interactive

Continuing Previous Conversations

# Resume conversation with specific thread ID
uv run main.py --thread abc123 --message "What did we discuss earlier?" --interactive

Debug Mode

# Enable detailed logging
uv run main.py --message "Hello!" --debug --interactive

Conclusion

This LangGraph Chatbot demonstrates modern AI application development with:

• Professional Setup: UV package manager and Makefile automation
• Robust Architecture: Graph-based workflow with conditional routing
• Memory Persistence: SQLite checkpoints for conversation continuity
• Tool Integration: Optional web search capabilities with Tavily
• Production-Ready: Comprehensive logging, error handling, and testing

The modular design makes it easy to extend with additional tools, modify the conversation flow, or integrate with different LLM providers. The comprehensive Makefile ensures consistent development workflows across different environments.

Key Takeaways:

1. Makefile Automation: Simplifies complex development workflows
2. LangGraph Power: Enables sophisticated conversation flows
3. Memory Management: Persistent state enhances user experience with SQLite
4. Tool Integration: Extends LLM capabilities with external services (Tavily)
5. Production Considerations: Logging, testing, and error handling are crucial

This implementation serves as a solid foundation for building more complex conversational AI applications. We will continue with other applications.

If you’re looking for freelance or consultancy work on data pipelines, AI systems, or real-time platforms, reach out to me on X, or via email.

And, if this saved you time or sparked an idea, consider sponsoring me on GitHub. Thank you!

GitHub: https://github.com/dogukannulu

X: https://x.com/dogukannuluu

Email: dogukannulu@gmail.com

Repository

GitHub - dogukannulu/stock-sentiment-platform

The Problem

Financial sentiment data is everywhere. Earnings summaries, analyst notes, news wires — all of it moves markets. The problem is that nobody is processing it in real time for individual tickers, correlating it with live price data, and making it queryable through plain English.

Commercial solutions exist, but they’re priced for hedge funds. Open-source alternatives are either stale demos or toy scripts that fall apart under real load.

I wanted to build something production-grade: proper schemas, a real message broker, time-series storage, version-controlled SQL transformations, a live dashboard, an AI analyst agent, and an evaluation suite that proves the sentiment scoring actually works.

Here’s everything I built, how I built it, and the lessons I learned along the way.

Architecture Overview

Yahoo Finance (yfinance)      Alpha Vantage News API
        |                              |
  price_producer.py           news_producer.py
        |                              |
  Kafka: price-ticks          Kafka: news-articles
  (Avro schema)               (Avro schema)
                    |
            Confluent Schema Registry
                    |
            Sentiment Job (Python threads)
  (Claude Haiku scores each article → TimescaleDB)
                    |
            TimescaleDB
(price_ticks · news_articles · sentiment_events)
                    |
              dbt models
(staging views → sentiment_signals mart → analytics schema)
                    |
        ┌───────────┴────────────┐
   Grafana Dashboard      AI Market Analyst Agent
   (4 live panels)        (LangGraph + FastAPI /brief)

The Stack

• Apache Kafka + Zookeeper
• Confluent Schema Registry (Avro)
• Threaded Kafka Consumers (Python)
• Anthropic Claude API (claude-haiku-4–5)
• TimescaleDB (Postgres 15)
• dbt-postgres
• Grafana
• LangGraph + FastAPI
• DeepEval GEval
• Docker

Step 1: Setting Up the Infrastructure

Everything except the Python processes runs in Docker Compose — Kafka, Zookeeper, Kafka UI, Confluent Schema Registry, TimescaleDB, and Grafana. The full stack comes up with one command:

docker compose up -d
docker ps
# 6 containers: zookeeper, kafka, kafka-ui, schema-registry, timescaledb, grafana

The key design decision here: Kafka acts as the durable buffer between data ingestion and processing. If the sentiment job goes down while Claude is being slow or rate-limited, messages queue up in Kafka and are processed when the job restarts. The producers never block.

Why Avro + Schema Registry? Because raw JSON is a footgun at scale. Schema Registry enforces data contracts at the message level — if a producer sends a message with a missing required field, the registry rejects it before the message even enters the topic. Here’s the price tick schema:

{
  "type": "record",
  "name": "PriceTick",
  "fields": [
    {"name": "event_time", "type": "string"},
    {"name": "ticker",     "type": "string"},
    {"name": "price",      "type": "double"},
    {"name": "volume",     "type": ["null", "long"], "default": null},
    {"name": "source",     "type": "string", "default": "yahoo_finance"}
  ]
}

Step 2: TimescaleDB Schema

TimescaleDB is Postgres with automatic time-based partitioning. That means a query like “give me NVDA sentiment for the last 24 hours” hits a partition scan instead of a full table scan — orders of magnitude faster at scale.

Three hypertables power this whole system:

-- Raw price ticks from Yahoo Finance
CREATE TABLE IF NOT EXISTS price_ticks (
    event_time  TIMESTAMPTZ NOT NULL,
    ticker      TEXT        NOT NULL,
    price       NUMERIC     NOT NULL,
    volume      BIGINT,
    source      TEXT        DEFAULT 'yahoo_finance'
);
SELECT create_hypertable('price_ticks', 'event_time', if_not_exists => TRUE);

-- Raw news articles from Alpha Vantage
CREATE TABLE IF NOT EXISTS news_articles (
    event_time          TIMESTAMPTZ NOT NULL,
    ticker              TEXT        NOT NULL,
    article_id          TEXT        UNIQUE,
    title               TEXT,
    body                TEXT,
    source              TEXT,
    url                 TEXT,
    av_sentiment_score  NUMERIC,
    av_sentiment_label  TEXT
);
SELECT create_hypertable('news_articles', 'event_time', if_not_exists => TRUE);
-- Enriched sentiment events (output of Claude scoring)
CREATE TABLE IF NOT EXISTS sentiment_events (
    event_time      TIMESTAMPTZ NOT NULL,
    ticker          TEXT        NOT NULL,
    source          TEXT        NOT NULL,
    sentiment_score NUMERIC,
    sentiment_label TEXT,
    raw_text        TEXT,
    price_at_event  NUMERIC,
    metadata        JSONB
);
SELECT create_hypertable('sentiment_events', 'event_time', if_not_exists => TRUE);

Step 3: The Kafka Producers

Two independent producers run in separate terminals (or as separate processes):

price_producer.py polls Yahoo Finance every 60 seconds for 5 tickers: AAPL, GOOGL, MSFT, TSLA, NVDA.

def fetch_and_produce(producer):
    for ticker in TICKERS:
        data = yf.Ticker(ticker)
        info = data.fast_info
        price = info.last_price
    if price is None:
            logger.warning(f"No price data for {ticker}, skipping.")
            continue
        message = {
            "event_time": datetime.now(timezone.utc).isoformat(),
            "ticker": ticker,
            "price": round(float(price), 4),
            "volume": int(info.three_month_average_volume) if info.three_month_average_volume else None,
            "source": "yahoo_finance",
        }
        producer.send(KAFKA_TOPIC, value=message)
        logger.info(f"Sent price tick: {ticker} @ {price}")

news_producer.py polls the Alpha Vantage News Sentiment API every 60 minutes (free tier: 25 calls/day), deduplicates by article_id, and sends each article to the news-articles topic:

for ticker in TICKERS:
    url = (
        f"https://www.alphavantage.co/query"
        f"?function=NEWS_SENTIMENT&tickers={ticker}"
        f"&limit=10&apikey={AV_API_KEY}"
    )
    resp = requests.get(url, timeout=10)
    data = resp.json()
    for item in data.get("feed", []):
        article_id = item.get("url", "")[-80:]
        if article_id in seen_ids:
            continue
        seen_ids.add(article_id)
        # ... send to Kafka

Before running the producers, register the Avro schemas:

python producers/kafka_admin.py

Step 4: The Sentiment Job — Where Claude Does the Work

This is the core AI layer. A Python process runs two threads: one consuming price ticks (simple DB write), one consuming news articles (Claude API call + DB write).

The prompt engineering:

SENTIMENT_PROMPT = """Analyze the sentiment of this financial news article \
toward the stock ticker mentioned.
Respond ONLY with a JSON object, no other text:
{{"score": <float between -1.0 and 1.0>, "label": "<positive|negative|neutral>", \
"reasoning": "<one sentence>"}}Article: {text}"""

The prompt does three things:

1. Grounds Claude on financial sentiment specifically, not general sentiment
2. Forces structured JSON output — no parsing ambiguity. Here, we would have used Pydantic, but simple prompt works as expected for this project.
3. Asks for a reasoning field, which becomes invaluable for debugging and the GEval evaluation later.

The Claude call inside NewsEnricher.map():

message = self.client.messages.create(
    model="claude-haiku-4-5-20251001",
    max_tokens=200,
    messages=[{"role": "user", "content": SENTIMENT_PROMPT.format(text=text)}],
)
raw_response = message.content[0].text.strip()
# Strip markdown code fences if Claude wraps the JSON in ```json ... ```
if raw_response.startswith("```"):
    raw_response = raw_response.split("```")[1]
    if raw_response.startswith("json"):
        raw_response = raw_response[4:]
    raw_response = raw_response.strip()
sentiment = json.loads(raw_response)

That code-fence stripping tripped me up early on. Claude occasionally wraps JSON responses in json blocks despite explicit instructions not to — always strip defensively.

Why threaded consumers instead of PyFlink?

My original plan used PyFlink. After spending an afternoon fighting apache-beam dependency failures on Python 3.12, I scrapped it. For a single-node local pipeline, threaded consumers give identical behavior — the business logic (Claude API call, DB write) is identical either way. No JVM, no classpath hell, zero configuration overhead.

def main():
    news_thread = threading.Thread(
        target=_run_consumer,
        args=("news-articles", "sentiment-job-news", NewsEnricher()),
        daemon=True,
    )
    price_thread = threading.Thread(
        target=_run_consumer,
        args=("price-ticks", "sentiment-job-price", PriceWriter()),
        daemon=True,
    )
    news_thread.start()
    price_thread.start()
    try:
        news_thread.join()
        price_thread.join()
    except KeyboardInterrupt:
        logger.info("Shutting down sentiment job.")

Step 5: TimescaleDB + dbt — From Raw Events to Business Signals

Raw sentiment events are noisy. A single article might be an outlier. The sentiment_signals mart aggregates by ticker per hour, computes the average sentiment score, counts positive/negative/neutral articles, and assigns a signal classification.

The dbt project has three layers:

staging/
  stg_price_ticks.sql          ← clean view over price_ticks (last 7 days)
  stg_sentiment_events.sql     ← clean view over sentiment_events (last 7 days)
marts/
  sentiment_signals.sql        ← hourly aggregates, bullish/bearish/neutral signal
analytics/
  sentiment_by_symbol.sql      ← per-symbol summary with momentum
  trending_tickers.sql         ← tickers with most sentiment movement
  price_sentiment_correlation.sql  ← correlation coefficient by ticker

The mart SQL is where the signal gets defined:

case
    when s.avg_sentiment >  0.3 then 'bullish'
    when s.avg_sentiment < -0.3 then 'bearish'
    else 'neutral'
end as signal

Running all 6 models takes about 0.6 seconds:

cd dbt && dbt run
# 06:12:43  6 of 6 OK created sql table model analytics.price_sentiment_correlation
# 06:12:43  Finished running 6 models in 0 hours 0 minutes and 0.61 seconds

Step 6: The Grafana Dashboard

This is the payoff moment. Four panels, all querying TimescaleDB directly, refreshing every 30 seconds.

Panel 1 — Sentiment Score Over Time (Time Series)

SELECT
  time_bucket('15 minutes', event_time) AS time,
  ticker,
  avg(sentiment_score) AS avg_sentiment
FROM sentiment_events
WHERE $__timeFilter(event_time)
GROUP BY 1, 2
ORDER BY 1

Panel 2 — Current Signal per Ticker (Stat panel)

SELECT ticker, signal, avg_sentiment
FROM sentiment_signals
WHERE event_hour = date_trunc('hour', now())
ORDER BY ticker

Panel 3 — Price vs Sentiment Overlay (dual Y-axis)

SELECT
  time_bucket('1 minute', p.event_time) AS time,
  p.ticker,
  avg(p.price) AS price,
  avg(s.sentiment_score) AS sentiment
FROM price_ticks p
LEFT JOIN sentiment_events s
  ON  time_bucket('15 minutes', p.event_time)
    = time_bucket('15 minutes', s.event_time)
  AND p.ticker = s.ticker
WHERE $__timeFilter(p.event_time)
GROUP BY 1, 2
ORDER BY 1

Panel 4 — News Article Volume (Bar chart)

SELECT
  time_bucket('1 hour', event_time) AS time,
  ticker,
  count(*) AS article_count
FROM news_articles
WHERE $__timeFilter(event_time)
GROUP BY 1, 2
ORDER BY 1

Step 7: The AI Market Analyst Agent

Beyond the dashboard, I built a natural-language query interface using LangGraph and FastAPI. Ask it anything about the market data, it figures out what SQL to run, queries the analytics tables, and synthesizes a market brief.

The agent has a single SQL tool:

@tool
def query_sentiment_data(sql: str) -> str:
    """Execute a read-only SQL query against the analytics schema."""
    conn = psycopg2.connect(**DB_CONFIG)
    try:
        with conn.cursor() as cur:
            cur.execute(sql)
            columns = [d[0] for d in cur.description]
            rows = cur.fetchmany(50)
            return json.dumps({"columns": columns, "rows": rows}, default=str)
    finally:
        conn.close()

The agent is built with create_react_agent from LangGraph:

def build_agent():
    llm = ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0)
    return create_react_agent(llm, [query_sentiment_data], prompt=SYSTEM_PROMPT)

Real query and output:

curl -s -X POST http://localhost:8000/brief \
  -H "Content-Type: application/json" \
  -d '{"question": "Which ticker has the most bearish sentiment right now?"}' \
  | jq .brief

**Symbol:** TSLA
**Sentiment:** Strongly negative at -0.78, deteriorating over last 3 hours
**Price:** $172.31, down 1.2% in the last hour
**Signal:** Bearish sentiment confirmed by price action
**Confidence:** Medium (12 articles in window)

Step 8: Evaluating the AI — GEval with Claude as Judge

The hardest question with any LLM pipeline: “Is the AI actually getting this right?”

I used DeepEval’s GEval metric framework, which lets you define evaluation criteria in plain English and score outputs against them. The twist: I used Claude Haiku as both the scorer and the judge, creating a closed evaluation loop with zero external dependencies.

Two metrics on a 50-article golden dataset (hand-labeled):

Label Faithfulness — Does the label (positive/negative/neutral) match the reasoning?

faithfulness_metric = GEval(
    name="Label Faithfulness",
    criteria="""The sentiment label must be consistent with the reasoning.
    A 'positive' label must have reasoning explaining why the article is positive.
    A 'negative' label must have reasoning explaining bearish content.
    A 'neutral' label should indicate balanced or factual reporting.""",
    evaluation_params=[LLMTestCaseParams.ACTUAL_OUTPUT],
    model=_judge,
)

Score Direction Accuracy — Does the sign of the score (+/−) match the human label?

score_calibration_metric = GEval(
    name="Score Direction Accuracy",
    criteria="""The numeric score must match the expected sentiment direction.
    Positive scores (>0) for positive sentiment.
    Negative scores (<0) for negative sentiment.
    Near-zero scores (-0.3 to 0.3) for neutral sentiment.""",
    evaluation_params=[LLMTestCaseParams.ACTUAL_OUTPUT, LLMTestCaseParams.EXPECTED_OUTPUT],
    model=_judge,
)

Results on the golden dataset:

• faithfulness (10 examples) -> 100%
• accuracy (50 examples) -> 96% (48/50 correct sign)

Lessons Learned

PyFlink on Python 3.12 is a non-starter. The apache-beam dependency fails at install time with ModuleNotFoundError: No module named 'pkg_resources'. I replaced the entire Flink runtime with threaded Kafka consumers in about 30 minutes. Identical behavior, zero JVM dependency. Don't fight your environment.

Claude occasionally wraps JSON in code fences. Even with explicit instructions to respond “ONLY with a JSON object, no other text,” Claude will sometimes wrap the response in ```json blocks. Always strip defensively — one line of handling saves hours of debugging in production.

DeepEval’s GEval requires a judge model, and it defaults to OpenAI. If you don’t have an OpenAI key, set model=AnthropicModel(...) explicitly. Using the same model as scorer and judge is methodologically fine for a closed evaluation — you're testing consistency, not ground truth. But, if we were evaluating the accuracy, using different models (even different LLM providers) would definitely be suggested.

Kafka consumer group IDs matter more than you think. Two processes using the same group_id will split partition ownership, causing each to see only half the messages. Make group IDs explicit and unique per logical consumer. Similarly, zombie consumer processes from previous runs can hold partition leases for 30 seconds after they die — always pkill -f sentiment_job.py before a clean restart.

TimescaleDB’s time_bucket() is what makes Grafana queries elegant. Instead of GROUP BY DATE_TRUNC('15 minutes', event_time), you write GROUP BY time_bucket('15 minutes', event_time). The function is index-aware and communicates intent better in code reviews.

If this saved you time or sparked an idea, consider sponsoring me on GitHub — and if you’re looking for freelance or consultancy work on data pipelines, AI systems, or real-time platforms, reach out to me on X, or via email.

GitHub: https://github.com/dogukannulu

X: https://x.com/dogukannuluu

Email: dogukannulu@gmail.com

Repository

GitHub - dogukannulu/aws_end_to_end_streaming_pipeline: An AWS Data Engineering End-to-End Project (Glue, Lambda, Kinesis, Redshift, QuickSight, Athena, EC2, S3)

You may see the part 2 (Data processing) below:

https://medium.com/@dogukannulu/aws-cloud-engineering-project-part-2-aws-glue-lambda-kinesis-redshift-athena-quicksight-c72b26876592

Overview

In this part of the whole pipeline, we are moving on to the Data Analysis part. We are going to check if the data has correctly arrived in the S3 bucket and the corresponding Glue table has been created by the Glue ETL job. If so, we will explore the data using AWS Athena first. For the Athena, we are going to use the data in the Glue table. We are going to connect AWS Athena to DBeaver and run several advanced SQL queries against the data.

Then, we will create a Redshift cluster with suitable parameters. We are going to use the data in the final S3 bucket for our Redshift cluster. Since our resulting parquet files will be uploaded as small batches, it is not possible to create the resulting tables according to their individual locations. We will, instead, create an external schema, database, and table in the end. We will also use the S3 directory to do this, not single locations.

We are going to use QuickSight to create our final dashboard. We will connect both AWS Athena and Redshift for the underlying data of the dashboard.

Monitor the Data

The first thing we have to do is to check if the data has arrived correctly. After our Glue ETL job is successfully completed, we should have our resulting parquet files under s3://aws-glue-clean-books-parquet-dogukan-ulu/clean_books_parquet/. We can see that the files have been uploaded correctly below.

Every time we run this whole pipeline, more parquet files will be added to this bucket and directory. Since the Glue table reads the data from this exact S3 directory, the underlying data in the Glue table will also be appended in each run. The parquet files should be cleaned and should look like this clean_books_data.parquet file.

Now we should also check if the Glue table exists under the database glue-etl-books-parquet-dogukan-ulu.

We can see below that there are two Glue tables:

• dirty_books_parquet: This is the table created by the Glue crawler in the previous part. It includes the dirty books data.
• clean_books_parquet: This is the resulting table of the Glue ETL Job. It includes the resulting cleaned parquet data.

We can also see the locations of the Glue tables (S3 locations).

Since we can see that our data has arrived correctly, we can move on to the AWS Athena stage.

AWS Athena

aws_end_to_end_streaming_pipeline/athena_queries at main · dogukannulu/aws_end_to_end_streaming_pipeline

Once we open the Athena home page on AWS, we can directly reach to the Glue tables.

We can connect to DBeaver using the Athena as the source data. It will help us run our queries however we want. You may take a look at the below article to have an insight into how to connect DBeaver to Athena.

https://medium.com/@dogukannulu/aws-athena-and-dbeaver-connection-6a4118985c6a

We can also use the Athena UI directly, but using DBeaver is another solution. Once connected, we will be able to see the Glue database and tables. We can run the following queries to better explore the data.

Find the availability ratio:

WITH numbers AS (
    SELECT
        COUNT(
            CASE
                WHEN availability = 1 THEN 1
            END
        ) AS number_of_available_books,
        COUNT(*) as total_number_of_books
    FROM `glue-etl-books-parquet-dogukan-ulu.clean_books_parquet`
)

SELECT
    number_of_available_books/total_number_of_books as availability_ratio
FROM numbers

Categorize the data by price:

WITH all_price_categories AS (
    SELECT
        title,
        CASE
            WHEN price <= 15 THEN 'Cheap'
            WHEN price <= 25 AND price < 40 THEN 'Middle'
            ELSE 'Expensive'
        END as price_category
    FROM `glue-etl-books-parquet-dogukan-ulu.clean_books_parquet`
)

SELECT
    title,
    price_category = 'Cheap' as is_cheap,
    price_category = 'Middle' as is_middle,
    price_category = 'Expensive' as is_expensive
FROM all_price_categories

Partition by number of reviews:

WITH max_price AS (
    SELECT
        title,
        price,
        num_reviews,
        ROW_NUMBER() OVER(PARTITION BY num_reviews ORDER BY price DESC) as price_row_number
    FROM `glue-etl-books-parquet-dogukan-ulu.clean_books_parquet`
)


SELECT
    title,
    price,
    num_reviews
FROM max_price
WHERE price_row_number = 1

We can see more SQL queries in the above link to explore the data.

AWS Redshift

aws_end_to_end_streaming_pipeline/redshift_queries at main · dogukannulu/aws_end_to_end_streaming_pipeline

First of all, we have to create an IAM role which will include the following policies:

• AmazonS3FullAccess
• AmazonRedshiftAllCommandsFullAccess
• AWSGlueServiceRole

After creating the IAM role, we should also create a subnet group since it will require us to use a subnet group while creating the cluster. Under Configuration -> Subnet groups, we can create a new subnet group. If we use the cluster for comparably simpler data loads, we can choose the smallest Node type and choose the number as 1 so that it won’t cost too much.

Then, we are going to define the username and password. This part is important since we will be using these while connecting to DBeaver.

After creating the username and password, we should uncheck the default settings and define the Network and security section ourselves.

We should choose the subnet group we already created and the VPC it is located. We can also choose a suitable security group. After creating the cluster, we will add Redshift as an inbound rule to our security group.

We should definitely check the Turn on Publicly accessible button so that we will be able to access our cluster from DBeaver.

We can leave all other options as default and create our cluster. It will take around 10 minutes. After creating, we can also run our queries using the Query editor on the Redshift dashboard, but we will use DBeaver. If you want to know how we can create a DBeaver and Redshift connection, you may take a look at the below article.

https://medium.com/@dogukannulu/aws-redshift-and-dbeaver-connection-39d3f4d2016f

Once connected, we have to create an external schema, database, and table. We are going to use this table to query the data.

We are going to first create the external schema and database.

create external schema redshift_parquet
from data catalog
database 'glue-etl-books-parquet-dogukan-ulu-external'
iam_role '<iam_role_arn>'
create external database if not exists;

We can check if the database glue-etl-books-parquet-dogukan-ulu-external is created under the Glue databases. If so, we can run the following query to create the external table.

CREATE EXTERNAL TABLE redshift_parquet.clean_books_parquet(
    title VARCHAR(100),
    price FLOAT,
    num_reviews INTEGER,
    availability INTEGER
)
stored as parquet
location 's3://aws-glue-clean-books-parquet-dogukan-ulu/clean_books_parquet/';

You should change the location part according to your S3 location. After the run is successful, we can select the table and see the results.

SELECT * FROM dev.redshift_parquet.clean_books_parquet limit 10;

We can also run all the queries in the AWS Athena section to better explore the data.

We checked the data correctness and created an external table for our Redshift cluster. Now, it’s time to create the resulting dashboard using this data.

Amazon QuickSight

The below article explains how to create an Amazon QuickSight account, connect to Redshift and Athena, and create data sets using those. In the end, it also explains how to use the QuickSight page efficiently.

https://medium.com/@dogukannulu/creating-quicksight-dashboards-using-amazon-redshift-and-athena-b503a8f937bf

For the dashboard, we can use both Athena and Redshift (even though they include exactly the same data, we can test our knowledge at this point).

The dashboard part depends completely on the use case of the data. I created a sample dashboard as below. You can also see the resulting Books Dashboard here.

If you’re looking for freelance or consultancy work on data pipelines, AI systems, or real-time platforms, reach out to me on X, or via email.

And, if this saved you time or sparked an idea, consider sponsoring me on GitHub. Thank you!

GitHub: https://github.com/dogukannulu

X: https://x.com/dogukannuluu

Email: dogukannulu@gmail.com

Repository

GitHub - dogukannulu/aws_end_to_end_streaming_pipeline: An AWS Data Engineering End-to-End Project (Glue, Lambda, Kinesis, Redshift, QuickSight, Athena, EC2, S3)

You may see the part 1 (Data ingestion) below:

https://medium.com/@dogukannulu/aws-cloud-engineering-project-part-1-aws-glue-lambda-kinesis-redshift-athena-quicksight-649fc4d7e0dc

Overview

In the previous part, we streamed data into Amazon Kinesis and uploaded the resulting data to the S3 bucket. In this part, we are going to create a Lambda function first. This function will be triggered by the S3 object upload. Once triggered, it will convert the data to parquet and upload the data to another S3 bucket.

We are going to create a Glue workflow which will include a Glue Crawler, a Glue ETL Job, and triggers. We are going to run this workflow once Lambda uploads the parquet file to the S3 bucket (either on schedule or manually, depending on the use case). The crawler will create a new Glue table and the ETL job will clean the dirty data. In the end, it will upload the resulting data to a different S3 bucket and will create a Glue table accordingly.

Amazon S3 Bucket

The first thing we have to do is to create an S3 bucket. We are going to use this bucket for the Lambda function to upload the resulting parquet data. From the S3 dashboard -> Create bucket:

We can leave all other parameters as default. In the end, we can click on Create bucket (aws-glue-dirty-books-parquet-dogukan-ulu for this project).

Lambda Function

aws_end_to_end_streaming_pipeline/lambda_function at main · dogukannulu/aws_end_to_end_streaming_pipeline

We are going to create the Lambda function first. The best and most stable way of creating a Lambda function is using a Docker container image. In the below article, you may see the steps for creating the Lambda function as a Docker container.

How to Create Amazon Lambda Function with the Container Image (Dockerfile)

We will use this Dockerfile and requirements file for this project.
One of the significant points for the Lambda function is creating an endpoint for Amazon S3. You can check the below article for how to create a VPC and create an S3 endpoint. You may either create a dedicated new VPC or use the default VPC. We have to add the endpoint for each. I explained the process in detail below.

Establishing a VPC for Amazon S3, Lambda, RDS and EC2

We should also have a proper IAM role that contains necessary S3 and CloudWatch permissions (AmazonS3FullAccess and CloudWatchFullAccess for example).

The other thing is having a certain security group. That doesn’t matter too much since we won’t need additional inbound rules.

To conclude the preparation part, we should have the following ones ready:

• IAM role with necessary S3 and CloudWatch permissions (AmazonS3FullAccess and CloudWatchFullAccess)
• The VPC with an S3 endpoint
• Security group dedicated to our Lambda function

After having all these ready, we can now walk through the Lambda function itself. We are going to create a class to obtain the objects coming from the source S3 bucket.

class S3Loader:
    def __init__(self):
        self.s3 = boto3.client('s3')

    def load_df_from_s3(self, bucket_name, key):
        """
        Read JSON from an S3 bucket & load into a pandas dataframe
        """
        print("Starting S3 object retrieval process...")
        
        try:
            get_response = self.s3.get_object(Bucket=bucket_name, Key=key)
            print("Object retrieved from S3 bucket successfully")
        except ClientError as e:
            print(f"S3 object cannot be retrieved: {e}")
            return None
                
        json_data = get_response['Body'].read().decode('utf-8')
        
        json_data_separated = json_data.replace('}{', '},{')
        json_data_final = f"[{json_data_separated}]"
        json_objects = json.loads(json_data_final)
        df = pd.DataFrame(json_objects)
        return df

Since the JSON data will be combined by Firehose and uploaded directly to the S3 bucket, the lambda function will modify that data and create the proper JSON data out of it. In the end, it will create a pandas data frame. After getting the data, we are going to create a parquet buffer out of it.

class S3Uploader:
    def __init__(self):
        self.s3 = boto3.client('s3')
        self.target_s3_bucket = "aws-glue-dirty-books-parquet-dogukan-ulu"
        self.target_s3_key = "dirty_books_parquet/books.parquet"
    
    def upload(self, buffer):
        """
        Upload Parquet data from memory buffer to S3
        """
        try:
            self.s3.put_object(Body=buffer.getvalue(), Bucket=self.target_s3_bucket, Key=self.target_s3_key)
            print("Parquet file uploaded into S3 successfully")
        except Exception as e:
            print(f"Error occured while uploading parquet into S3: {e}")

We also defined the target S3 bucket with prefix and key. So, our parquet data will be uploaded to s3://aws-glue-dirty-books-parquet-dogukan-ulu/dirty_books_parquet/books.parquet.

We should create a lambda_handler function since our Dockerfile will check that method.

def lambda_handler(event, context):
    try:
        bucket = event['Records'][0]['s3']['bucket']['name']
        key = event['Records'][0]['s3']['object']['key']
        print(f"S3 bucket is obtained from the event: {bucket}")
        print(f"Object key is obtained from the event: {key}")
        s3_loader = S3Loader()
        df = s3_loader.load_df_from_s3(bucket_name=bucket, key=key)
        
        parquet_buffer = io.BytesIO()
        df.to_parquet(parquet_buffer, index=False)
        print("JSON has been converted into parquet")
        
        s3_uploader = S3Uploader()
        s3_uploader.upload(parquet_buffer)
        return {
            'statusCode': 200,
            'body': json.dumps('Parquet conversion and upload successful')
        }
        
    except Exception as e:
        print(f"An error occurred: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps('An error occurred during processing')
        }

In this method, we get the data from the S3 event because we want this function to be triggered by the S3 object upload. You can see a sample S3 event here. Then, we created a parquet buffer and uploaded it to the target S3 bucket. Now, we have our Lambda function ready. After deploying our function to Amazon ECR following the instructions in the above article, we can use it to create our Lambda function.

We should choose Container Image as the creation method. We can name our function as json-to-parquet-lambda-function. We should copy the URI of the container image located in the ECR and paste it under the Container image URI.

We can choose the architecture according to our local machine and select the suitable IAM role we just created. Then, we can create our Lambda function.

After it is created, we should set the timeout as 1 minute from the basic settings (We can set a lower limit but just in case). We should also modify the VPC settings. We should choose the VPC that includes the S3 endpoint we recently created. We can choose as many subnets as we want and the dedicated security group for our function.

Since we don’t have any environmental variables, we don’t have to modify that part. We are going to add the S3 trigger now.

We should choose the source bucket as the one Firehose will upload the streaming data. We can define the prefix as json_data/ since the data will be uploaded to that directory.

After setting the trigger, we have our Lambda function ready now. Once triggered, we should see the logs of the script as below.

We should also see the parquet file in the S3 bucket as below.

We can decide on the directory to which we want to upload our parquet file. Once uploaded, we can query our file using S3 Select.

The result will be something as below. You can also see the resulting parquet file here.

AWS Glue Crawler

Our next plan is to take the dirty parquet file uploaded by the Lambda function by a Crawler and create a corresponding Glue table. We should have an IAM role that has an AmazonS3FullAccess policy. We can use it both for the crawler and the ETL job. We are going to create our Crawler first.

We can name it as books-parquet-to-glue-table.

If we want our crawler to crawl all the files, we can define the location as a directory under a certain S3 bucket.

We can choose a Glue Data Catalog table as well. But if our table hasn’t been created yet, we have to choose Not yet as the first option. We are going to define our S3 location. Depending on our use case, we can choose Crawl all sub-folders or we can define it so that it crawls only the newly uploaded files under the S3 directory. If new files are uploaded into the S3 bucket every time we run our pipeline and we only want to crawl them, we can choose Crawl new sub-folders only as above.

We should choose the IAM role that the Crawler will be using so that it includes necessary S3 permissions.

We should also choose a target database in which our resulting Glue table will be located.

We can choose a prefix for the resulting table. If we don’t choose, the name of our target table will be the same as the source S3 directory. We should choose the schedule as On demand since it will be triggered by a trigger in the workflow.

After defining all the parameters, we can create our Crawler. We have to create the Glue ETL Job since it will also be needed while creating our workflow. Our crawler will create a Glue table named dirty_books_parquet at the end.

AWS Glue ETL Job

aws_end_to_end_streaming_pipeline/glue_etl_job at main · dogukannulu/aws_end_to_end_streaming_pipeline

Now, it’s time to create our Glue ETL job. It will modify the dirty parquet data uploaded by the Lambda function and create a resulting clean parquet file. It will upload that file to an S3 bucket named aws-glue-clean-books-parquet-dogukan-ulu and create a corresponding Glue table clean_parquet under the database glue-etl-books-parquet-dogukan-ulu.

First of all, we have to create our Glue ETL job using a Spark script.

After creating the script, we can set the parameters as 2 DPU and the least powerful worker type since our workload will not be that much. We should name our job as well. Now, we can go through our ETL job script. First of all, the imports come as default in Glue ETL jobs. We should add the last two lines.

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame
from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType, FloatType

There are more default parameters as below that start Spark session.

args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

We should define a dictionary so that we can convert the num_reviews column from string to integer.

num_review_mapping = {
    "One": 1, "Two": 2, "Three": 3, "Four": 4, "Five": 5
}

Since our Glue ETL job will run after the crawler creates a new Glue table dirty_books_parquet, we can read the data in that table using a Glue dynamic frame. Then, we are going to convert it to a Spark data frame as below.

def create_initial_df():
    glue_dynamic_frame_initial = glueContext.create_dynamic_frame.from_catalog(database='glue-etl-books-parquet-dogukan-ulu', table_name='dirty_books_parquet')
    df = glue_dynamic_frame_initial.toDF()

    return df

We are going to drop the pound symbol from the price column (also convert it to float) and convert the num_reviews column to an integer instead of a string.

def map_num_reviews(review):
    return num_review_mapping.get(review, review)

map_num_reviews_udf = F.udf(map_num_reviews, IntegerType())

# Define the UDF for price formatting
def format_price(price):
    price = price.replace("£", "").replace(",", "")
    return round(float(price), 2)

format_price_udf = F.udf(format_price, FloatType())

In the end, we are going to create our final data frame. We should drop the upc column since it’s unnecessary. We will also convert the availability column from string to integer depending on the column content. We will also convert our Spark data frame to a Glue dynamic frame to be able to write it to a Glue table.

df_final = df.withColumn("price", format_price_udf("price")) \
             .drop("upc") \
             .withColumn("num_reviews", map_num_reviews_udf("num_reviews")) \
             .withColumn("availability", F.when(df["availability"] == "In stock", 1).otherwise(0))

glue_dynamic_frame_final = DynamicFrame.fromDF(df_final, glueContext, "glue_etl")

Last but not least, we are going to upload our resulting clean parquet file to the S3 bucket and create a corresponding Glue table.

s3output = glueContext.getSink(
  path="s3://aws-glue-clean-books-parquet-dogukan-ulu/clean_books_parquet",
  connection_type="s3",
  updateBehavior="UPDATE_IN_DATABASE",
  partitionKeys=[],
  compression="snappy",
  enableUpdateCatalog=True,
  transformation_ctx="s3output",
)

s3output.setCatalogInfo(
  catalogDatabase="glue-etl-books-parquet-dogukan-ulu", catalogTableName="clean_books_parquet"
)
s3output.setFormat("glueparquet")
s3output.writeFrame(glue_dynamic_frame_final)
job.commit()

Our resulting parquet file should look like this clean_books_data.parquet file.

AWS Glue Workflow

Now it’s time to create the Glue workflow so that this whole process is automated. You can check the below article to have an insight into how to create a Glue workflow.

Creating AWS Glue Workflows with Glue Crawler and Glue ETL Jobs

We should have the following components for our workflow:

• An initial trigger. If we want to use this workflow for demo purposes, we can choose it as an on-demand trigger (We can trigger it manually). If we want it to be automatic for an automated streaming pipeline, we should set it as Schedule and decide on the cron expression as below.

• We should have our crawler as the second component.
• We should add another trigger after the crawler with ANY notation.
• In the end, we will have our Glue ETL job.

This workflow will triggered initially. Then, our crawler will create a corresponding Glue table from the dirty parquet file. Once created, our Glue ETL job will start running. It will clean the data and create a resulting clean parquet file. It will upload it to the S3 bucket and create a corresponding Glue table.

In this part of the article, we created a new Lambda function using a Docker container image. Once we upload the JSON data coming from the Firehose, it is triggered. It converts that data and uploads it as a parquet file into another S3 bucket. Once it is uploaded, our Glue workflow can start running. It includes the Glue crawler and Glue ETL job. In the end, we cleaned the parquet data, uploaded it to S3, and created a corresponding Glue table.

In the last part, we are going to check if the data is uploaded correctly. Then, we will create the corresponding tables in Redshift and Athena. In the end, we are going to create a QuickSight dashboard using this data.

If you’re looking for freelance or consultancy work on data pipelines, AI systems, or real-time platforms, reach out to me on X, or via email.

And, if this saved you time or sparked an idea, consider sponsoring me on GitHub. Thank you!

GitHub: https://github.com/dogukannulu

X: https://x.com/dogukannuluu

Email: dogukannulu@gmail.com

Repository

https://github.com/dogukannulu/text-to-sql

Tech Stack

Before diving in, here is every tool, framework, and package used across the project:

LLM & Agent

• Claude Sonnet 4.6 (Anthropic) — SQL generation, self-correction, result explanation
• Claude Haiku 4.5 (Anthropic) — LLM-as-judge for hallucination evaluation
• LangChain 0.3 — LLM wrappers, prompt templates, structured output parsing
• LangGraph 0.2 — stateful agent graph (7 nodes, conditional routing)
• langchain-anthropic 0.2 — Claude integration for LangChain

Embeddings & Vector Search

• Voyage AI voyage-3–1024-dimensional text embeddings
• langchain-voyageai — LangChain wrapper for Voyage AI
• Qdrant — vector store (2 collections: schema + knowledge base)
• qdrant-client 1.8 — Python SDK for Qdrant

API & Backend

• FastAPI 0.110 — REST API (POST /query, GET /history, POST /schema/refresh)
• Uvicorn 0.29 — ASGI server
• python-multipart — multipart form support for FastAPI
• SQLModel 0.0.16 — ORM (combines SQLAlchemy + Pydantic)
• psycopg2-binary 2.9 — PostgreSQL adapter
• Alembic 1.13 — database migrations
• pydantic 2.6 — data validation and serialisation
• pydantic-settings 2.2 — settings management (.env)
• sqlglot 23.0 — SQL parser for syntax validation (no DB round-trip)

Frontend

• Streamlit 1.33 — chat UI, dataframe display, schema sidebar

Observability & Tracing

• Langfuse (Cloud) — LLM observability: traces, spans, token usage, scores
• langfuse 4.x Python SDK — @observe decorator, score_current_trace()

Evaluation

• deepeval / GEval — evaluation harness framework
• anthropic Python SDK — direct Anthropic API calls for LLM-as-judge

Infrastructure & Config

• Docker + Docker Compose — all 6 services in one compose file
• PostgreSQL 16 (postgres:16-alpine) — primary data store + Langfuse metadata DB

MCP Server

• mcp Python SDK (FastMCP) — MCP server exposing 3 tools to Claude Code
• requests 2.31 — HTTP client for MCP server → backend calls

Testing

• pytest 8.0 — test runner
• pytest-asyncio 0.23 — async test support
• httpx 0.27 — async HTTP client for integration tests

Developer Tooling

• ruff 0.3 — linter and formatter
• mypy 1.9 — static type checker
• python-dotenv — .env file loading

Why I built a Text-to-SQL agent from scratch

Whenever someone wants to easily build queries with pure text, NLQ or text to SQL comes into our minds. That’s why I wanted to know whether a modern LLM stack could close that loop entirely — not just generate SQL, but execute it, explain the results in plain English, and recover gracefully when the query is wrong.

So, I built a production-grade agent with a RAG pipeline, a 7-node LangGraph state machine, full observability, a REST API, a Streamlit UI, an MCP server for Claude Code, and a GEval evaluation harness. The database is fictional (UDogRetail — a dog-products e-commerce company), but every architectural decision reflects something I’d do on a real project.

This article walks through how I built it, every non-obvious decision I made, and everything that went wrong.

The full code is on GitHub: clone it, run docker compose up, and it works in under five minutes.

1. The UDogRetail dataset — designing a realistic test environment

Real databases and semantic layers are messy. They have tables with ambiguous names, columns with legacy conventions, and business rules that exist only in someone’s head. I wanted a test environment that reflected this complexity without needing access to production data.

UDogRetail has five business tables:

customers — 1,000 customer records with full_name, email, country, plan_tier

orders — 4,500 orders with a status lifecycle (pending → delivered → refunded)

order_items — ~10,000 line items with product categories and discount percentages

payments — 4,300 payment attempts across four providers (Stripe, PayPal, Klarna, Adyen)

subscriptions — 686 subscription contracts with monthly_fee and cancellation tracking

I generated seed data with a Python script (scripts/generate_seed.py) using random.seed(42) so every fresh clone produces identical data. The data has intentional quirks that make it a good test: mixed currencies (GBP, EUR, USD), nullable shipped_at columns, a multi-table revenue calculation requiring a JOIN across three tables, and a full_name column — not first_name/last_name — that the LLM reliably tries to hallucinate.

I also wrote nine knowledge-base markdown documents covering KPI definitions (MRR, churn rate, ARPU), business rules (how status = ‘active’ filters apply to subscriptions), and data quality notes. These matter because some questions are unanswerable without domain context — “what is our MRR?” requires knowing that MRR = SUM(monthly_fee) WHERE status = ‘active’, not just a count of subscriptions.

2. Architecture overview — how all the pieces fit together

The system has six Docker services, all wired together in a single docker-compose.yml.

The data flow:

• User → Streamlit (8501) → POST /query (FastAPI, 8000) → LangGraph Agent OR Claude Code → MCP Server → POST /query (FastAPI, 8000) → LangGraph Agent
• RETRIEVE (Qdrant)
• GENERATE (Claude)
• VALIDATE (sqlglot)
• EXECUTE (PostgreSQL)
• EXPLAIN ←── success
• CORRECT ←── error (up to 3 retries)
• CLARIFY ←── retries exhausted

Every query run is traced in Langfuse and saved to a query_run table in PostgreSQL. The MCP server exposes query_tool, schema_tool, and history_tool to Claude Code as a stdio subprocess.

The tech stack I chose and why:

LLM: Claude Sonnet 4.6 (Anthropic) — best structured-output reliability in my testing

Agent: LangGraph 0.2 — explicit state machine, every transition is readable code

Embeddings: Voyage AI voyage-3 — strong retrieval quality, 1024-dimensional vectors

Vector store: Qdrant — simple Docker setup, fast cosine similarity search

API: FastAPI + Uvicorn — async, Pydantic validation built-in

Frontend: Streamlit — functional chat UI in ~300 lines

Observability: Langfuse — token tracking + span-level latency per node

Evaluation: GEval / deepeval — LLM-as-judge for hallucination scoring

Config: pydantic-settings — .env locally, AWS SSM Parameter Store in production

3. Building the RAG pipeline — schema + document retrieval

The most important architectural decision in the whole project: ground the SQL generator in retrieved schema context, not LLM memory.

Without RAG, Claude generates SQL from training data. It knows generic SQL patterns perfectly, but it doesn’t know that UDogRetail uses full_name instead of first_name/last_name, or that MRR is defined as SUM(monthly_fee) WHERE status = ‘active’. Every query that requires either of these will hallucinate.

Schema indexing

I wrote a schema crawler (app/rag/crawler.py) that queries INFORMATION_SCHEMA to extract table names, column names, data types, nullable flags, and foreign key relationships. This runs at startup via the FastAPI lifespan handler.

Each table becomes one chunk in Qdrant’s udogretail_schema collection — five vectors total, one per table. The chunk format is deliberately verbose:

Table: customers

Description: Core customer registry. One row per registered user account.

Columns:

• id (uuid, NOT NULL) — Primary key
• full_name (character varying, NOT NULL)
• email (character varying, NOT NULL)
• country (character varying, NOT NULL)
• plan_tier (varchar, NOT NULL)
• created_at (timestamptz, NOT NULL)

Foreign keys: none

The full column name, type, and nullability are in every chunk. When a question is embedded and searched, the schema chunks that come back give Claude the exact column names — no guessing allowed.

Knowledge base indexing

Nine markdown documents live in backend/knowledge_base/. They cover things the schema can’t express: what “active subscription” means, how to calculate MRR, which currency to use when grouping revenue, and known data quality issues.

These are split with RecursiveCharacterTextSplitter at 400 tokens with 40-token overlap, producing ~100 chunks in the udogretail_docs collection.

Retrieval

At query time, retrieve_context() embeds the user’s question with voyage-3 and searches both collections (udogretail_schema and udogretail_docs) in parallel using ThreadPoolExecutor(max_workers=2). The top-3 results from each collection are merged and sorted by relevance score — always the most relevant schema tables and the most relevant business rules, delivered together to the GENERATE node.

4. The LangGraph agent — nodes, state, and self-correction

The agent is a StateGraph with seven nodes. Every node receives and returns an AgentState TypedDict — 17 fields covering the question, retrieved context, generated SQL, execution result, error history, and Langfuse trace ID.

class AgentState(TypedDict):
  question: str
  session_id: str
  retrieved_schema: list[str]
  retrieved_docs: list[str]
  generated_sql: Optional[str]
  sql_reasoning: Optional[str]
  sql_assumptions: list[str]
  sql_confidence: float
  validation_error: Optional[str]
  execution_result: Optional[ExecutionResult]
  execution_error: Optional[str]
  retry_count: int
  correction_history: list[CorrectionRecord]
  needs_clarification: bool
  clarification_message: Optional[str]
  final_explanation: Optional[str]
  langfuse_trace_id: Optional[str]

GENERATE

The GENERATE node calls Claude with a structured output schema (SQLGenerationResponse) using LangChain’s with_structured_output(model, include_raw=True). The include_raw=True flag is crucial — it gives back both the parsed Pydantic model AND the raw AIMessage with usage_metadata, so I can log token counts to Langfuse without a second API call.

The system prompt has one rule that saved me hours of debugging:

CRITICAL: Use ONLY the exact column names shown in the schema above.

Never guess or invent column names. If a name feels intuitive (e.g. first_name, last_name) but is not in the schema, it does NOT exist — use what is shown (e.g. full_name).

Without this constraint, Claude reliably generates c.first_name || ‘ ‘ || c.last_name for any customer name query. The column is full_name. The “CRITICAL:” label is not decoration — Claude follows explicitly labelled constraints more reliably than softly worded ones.

VALIDATE → EXECUTE

The VALIDATE node runs sqlglot.parse(sql, dialect=”postgres”) before touching the database. Syntax errors caught here are cheaper than round-tripping to PostgreSQL.

The EXECUTE node (app/tools/pg_executor.py) enforces read-only access with a FORBIDDEN_KEYWORDS check:

FORBIDDEN_KEYWORDS = frozenset({
"INSERT", "UPDATE", "DELETE", "DROP",
"TRUNCATE", "ALTER", "CREATE", "GRANT", "REVOKE"
})

Every token in the SQL is checked against this set before execution. The database also runs with SET LOCAL statement_timeout so runaway queries can’t hang the server.

The self-correction loop

The routing after EXECUTE is the heart of the agent:

def route_after_execute(state: AgentState) -> str:
  if state["execution_error"] is None:
    return "explain"
  if state["retry_count"] >= settings.max_retries:
    return "clarify"
    return "correct"

CORRECT re-prompts Claude with the original question, full schema context, the failed SQL, and the PostgreSQL error message. The CorrectionRecord is appended to correction_history so each retry knows what was tried before. CORRECT always routes back through VALIDATE — corrected SQL still needs a syntax check before hitting the database.

After MAX_RETRIES (default: 3), the CLARIFY node generates a user-friendly message with specific suggestions: rephrase with exact column names, break the question into parts, specify a currency for monetary questions.

5. Making it production-ready — FastAPI, Docker, Terraform

The API

The API has three route groups — /query, /history, and /schema. All I/O is typed with Pydantic:

• QueryRequest: question (5–500 chars), session_id (UUID string)
• QueryResponse: full result including SQL, rows, explanation, retry count, succeeded flag
• QueryRunSummary / QueryRunDetail: history endpoint shapes

Rate limiting (max 10 requests/minute per session_id) protects against runaway clients. CORS allows localhost:8501 (Streamlit) and localhost:8080 (MCP server) in local mode.

Docker

All six services are defined in docker-compose.yml. The backend Dockerfile runs Alembic migrations before starting uvicorn:

#!/bin/bash
# backend/start.sh
set -e
echo "==> Running Alembic migrations…"
cd /app/backend && alembic upgrade head
echo "==> Starting uvicorn…"
exec uvicorn app.main:app - host 0.0.0.0 - port 8000

This guarantees the query_run, correction_attempt, and schema_snapshot tables exist before the first request arrives — no race conditions between migration and serving.

Configuration

app/config.py uses pydantic-settings BaseSettings for local development and fetches secrets from AWS SSM Parameter Store when APP_ENV=production (in our case, they are kept in .env):

class Settings(BaseSettings):
  anthropic_api_key: SecretStr
  voyage_api_key: SecretStr
  postgres_password: SecretStr
  langfuse_secret_key: SecretStr

# All secrets use SecretStr — never logged as plaintext

The SSM path convention is /udogretail/text2sql/{FIELD_NAME}. Swapping from local to production is a single env var change.

6. Observability with Langfuse — tracing every agent run

Every POST /query call creates a Langfuse trace. Every node in the agent creates a child span. I used Langfuse SDK v4’s @observe decorator — it wraps functions automatically and attaches them to the active trace context:

@observe(name="generate", as_type="generation")
def generate(state: AgentState) -> AgentState:
# Claude call happens here
# Langfuse auto-captures input, output, latency

For Claude calls, I use with_structured_output(model, include_raw=True) to capture token counts from raw_response.usage_metadata and log them to the active span:

lf = get_lf_client()
lf.update_current_observation(
model="claude-sonnet-4–6",
usage={"input": input_tokens, "output": output_tokens},
)

The result: every trace shows retrieve → generate → validate → execute → explain with latency per node and token counts per Claude call. I also log two scores per trace — succeeded (1.0 or 0.0) and retry_count — so I can filter traces by quality in the Langfuse scores panel.

Three things I learned about Langfuse SDK v4:

1. @observe replaces the old CallbackHandler. The v4 SDK dropped langfuse.callback entirely. If you’re migrating, rewrite every LangChain integration with @observe.

2. docker compose restart doesn’t re-read .env. Use docker compose up -d service to recreate the container with new environment variables.

3. SDK v4 requires Langfuse Cloud or a v3+ self-hosted server. The langfuse/langfuse:2 Docker image is not compatible with SDK v4’s OTLP ingestion format. For anything new, use Langfuse Cloud.

7. Testing — unit tests and integration tests

A production agent needs a test suite that runs without thinking about it. I split the tests into two layers: unit tests that run offline in milliseconds, and integration tests that hit the running Docker stack.

Unit tests

Five unit test files cover the pure-logic components that have no external dependencies.

test_sql_validator.py — the most important unit test file. It verifies that validate_sql() correctly rejects every forbidden keyword regardless of case, including keywords buried inside CTEs:

@pytest.mark.parametrize("keyword", sorted(FORBIDDEN_KEYWORDS))
def test_forbidden_keyword_rejected(keyword: str) -> None:
  sql = f"{keyword} INTO orders VALUES ('x')"
  result = validate_sql(sql)
  assert not result.is_valid
  assert keyword in result.error_message
  def test_forbidden_keyword_in_cte_still_rejected() -> None:
  sql = "WITH x AS (DELETE FROM orders RETURNING id) SELECT * FROM x"
  result = validate_sql(sql)
  assert not result.is_valid

This is the security boundary of the entire agent — if validate_sql lets a write statement through, it hits PostgreSQL. The parameterised test ensures every keyword in FORBIDDEN_KEYWORDS is tested individually, including case-insensitive variants.

test_crawler.py — uses a mock SQLModel session to verify crawl_schema() returns the correct SchemaMetadata structure without touching a real database.

test_chunker.py — verifies that chunk_schema() produces one chunk per table and that the chunk text contains the table name, column names, and types.

test_config.py — verifies that get_settings() loads correctly from environment variables and that all secret fields are SecretStr instances (never plain strings).

test_agent_state.py — verifies that AgentState can be instantiated with all 17 fields correctly typed.

Run unit tests with no Docker required:

pytest tests/unit/ -v

Integration tests

Three integration test files run against the live Docker stack — they make real HTTP requests to http://localhost:8000.

test_query_endpoint.py — six tests covering the full /query flow: HTTP 200, correct response shape, succeeded=True for a simple question, session_id round-trip, HTTP 422 for a question under 5 characters, and that a query is saved to history.

test_schema_endpoints.py — verifies GET /schema/status returns vector counts for both Qdrant collections and POST /schema/refresh returns HTTP 200.

test_history_endpoint.py — verifies GET /history and GET /history/{id} return the correct shapes.

The design principle: integration tests assert on structure, not values. assert data[“succeeded”] is True and assert “SELECT” in data[“sql”].upper() — not assert data[“result”][0][“count”] == 1000. Values change when the seed data changes; structure doesn’t.

# Docker stack must be running

pytest tests/integration/ -v

20/20 integration tests pass against the live stack.

8. Evaluating the agent with GEval

I built a 32-case golden test set covering three complexity levels and ran three metrics:

SQL Correctness — does the query return the expected result columns?

Schema Faithfulness — does the generated SQL reference only tables and columns that actually exist in the schema? Checked by parsing the SQL with sqlglot and comparing every identifier against the live schema.

Hallucination — is the plain-English explanation faithful to the actual query results? Scored by an LLM-as-judge using claude-haiku-4–5:

prompt = f"""
Score from 0.0 to 1.0 whether this explanation is faithful to the results.
Results: {json.dumps(rows[:5])}
Explanation: {explanation}
Return only JSON: {{"score": float, "reasoning": str}}
"""

Running the harness:

python -m evaluation.harness --complexity simple
python -m evaluation.harness --limit 10

The test set includes five “trick” questions that require knowledge-base docs to answer correctly — questions like “what is our MRR?” (requires the business rule status = ‘active’) and “show customers with duplicate payments” (requires knowing this data quality issue is documented in the knowledge base). These are the questions that fail without RAG and pass with it.

9. The MCP server — making it a Claude Code citizen

We can invoke the FastAPI endpoint via Streamlit. But, we can also do the same using an MCP server using Claude Code.

The MCP server (mcp_server/server.py) uses FastMCP to expose three tools to Claude Code:

mcp = FastMCP(name="udogretail-text2sql")
@mcp.tool()
def query_tool(question: str, session_id: str = "") -> str:
"""Run a natural-language question through the Text-to-SQL agent."""
…
@mcp.tool()
def schema_tool(keyword: str) -> str:
"""Look up tables and columns matching a keyword."""
…
@mcp.tool()
def history_tool(limit: int = 5) -> str:
"""Fetch the last N query runs from agent history."""
…

The .mcp.json at the repo root registers it as a stdio subprocess for Claude Code:

{
"mcpServers": {
  "udogretail-text2sql": {
    "type": "stdio",
    "command": ".venv/bin/python",
    "args": ["mcp_server/server.py"],
    "env": {"API_BASE_URL": "http://localhost:8000"}
    }
  }
}

With the Docker stack running, I can ask Claude Code “how many active Pro subscribers are there?” and it calls query_tool directly, returning the SQL and result inline in the terminal.

One unexpected issue: a stdio MCP server exits immediately in Docker if stdin is closed. The fix is stdin_open: true and tty: true in the compose service definition. Without these, the container starts and exits with code 0 immediately.

10. Results, lessons, and what I’d do differently

The agent works. It handles simple count queries first-try, joins across three tables with appropriate GROUP BY, and self-corrects the most common errors — wrong column name, missing WHERE clause, incorrect JOIN direction. Langfuse traces confirm the self-correction loop fires roughly 20–30% of the time on harder questions. Clarification is reached rarely on well-formed questions.

What I’d do differently:

More aggressive schema chunk verbosity: The chunks include column types and FK relationships, but not value distributions. Knowing that status has values (‘pending’, ‘confirmed’, ‘shipped’, ‘delivered’, ‘cancelled’, ‘refunded’) — hardcoded in the chunk — would have prevented several early errors where Claude guessed wrong enum values.

Smarter retry routing: The current CORRECT node re-prompts from scratch. A better approach: if the error is “column does not exist”, retrieve fresh schema context before re-prompting. The schema is already there, but explicitly signalling what changed would improve correction rates.

Evaluation before prompt tuning: I tuned the generate prompt by feel during development. Running the GEval harness earlier would have given me a quantitative signal — I could have measured the effect of each prompt change instead of reasoning about it. The changes for prompts into production should happen after seeing better GEval results.

What surprised me:

The column-name hallucination problem is real and consistent. Every time I tested a “top customers by spend” query before adding the CRITICAL: constraint, Claude generated first_name/last_name. After the constraint, it stopped. The fix was four lines of prompt text.

Decimal serialization is a silent failure. PostgreSQL NUMERIC columns return Python Decimal objects, which can’t be serialized to JSON. The error only surfaces at save time — the query succeeds, the results are correct, and then the API returns 500. Sanitising rows to float before saving is one of those fixes that should be in every SQL-over-API stack.

Langfuse SDK version pinning matters. SDK v4 dropped backward compatibility with Langfuse server v2. If you’re self-hosting, make sure your server and SDK versions are aligned. For anything new, use Langfuse Cloud — no version management required.

If you’re looking for freelance or consultancy work on data pipelines, AI systems, or real-time platforms, reach out to me on X, or via email.

And, if this saved you time or sparked an idea, consider sponsoring me on GitHub. Thank you!

GitHub: https://github.com/dogukannulu

X: https://x.com/dogukannuluu

Email: dogukannulu@gmail.com

How Batch Pipelines Become Complex and Brittle Over Time

Every team that works with batch pipelines has experienced a well-known nightmare: getting notified about a job failure in the middle of the night. When something like this happens, dashboards suddenly become stale, stakeholders start asking questions, and the first thing teams usually try to do is restart the job, assuming it was a temporary problem.

And sure, at the beginning of our evaluations, errors do look operational. But it becomes obvious that this is not the real underlying issue. The real problem is the architecture, which easily becomes fragile once the data volume and enterprise requirements increase.

But I didn’t write this article to list the bad behaviors of batch pipelines. After all, they are, without a doubt, the backbone of power analytical systems, so the question isn’t why do we use them? It’s how do we structure them to support real-time decisions, compliance guarantees, and reliable data at scale?

In this article, we’ll explore why batch pipelines start simple but begin failing as data volumes and requirements grow. We will also see how a unified data movement layer with Estuary can help you rewrite this story.

Key Takeaways

• As systems scale, batch pipelines often become fragile due to hidden state, dependency chains, and delayed data visibility.
• Operational fixes and retries increase complexity over time instead of improving reliability.
• A unified data movement layer handles incremental change continuously, reducing backfills, downtime, and data inconsistency.

Why “Just Run the Batch Job” Is Not Working Anymore

As mentioned, most batch pipelines usually look simple in the beginning. Batch jobs are generally scheduled overnight; this way, teams can address any malfunctions before the start of the next day.

Jobs consist of three major steps for ETL or ELT pipelines: extract, transform, load. The approach seems fine for small-scale jobs and companies. Stakeholders are limited, data is not huge yet, and fixes are relatively easy.

Though this sounds pretty simple, it’s just an illusion supported by entity-relationship diagrams.

The reality is more complex:

• Even though most components look to be stateless, you still have stateful offsets.
• There are so many components that might break the order of data.
• Every source system is prone to changes in the schema. While this is typically a matter of communication with the source team, overlooking it can impact all downstream components.

When this happens, enterprises have high expectations. They want reliable and compliant data with clear ownership. It also needs to be so fresh that we can measure it in minutes.

Traditional batch architectures struggle to meet these demands.

How Do Batch Pipelines Become Complex and Fragile?

Problems in batch operations don’t usually appear out of nowhere. As setups get more complex, weaknesses start popping up slowly, making breakdowns more likely.

The Hidden Statefulness of Batch Pipelines

Even though we see batch jobs as stateless, there are plenty of states they are dependent on. A job may rely on successful completion of a previous run. In addition, source systems don’t promise anything regarding completeness, so we may get partial data with some job runs. We could also hit some rate limits on the source side, and each component might include internal watermarks.

Once these issues arise, teams typically try to find immediate workarounds to meet SLAs. But every workaround adds yet another state.

Here’s a very common pattern:

# You may see the pseudo-code reflecting the logic 
# behind an hourly Airflow task

last_cursor = read_from_metadata_store("api_cursor")

response = requests.get(
    API_ENDPOINT,
    params={"cursor": last_cursor, "limit": 1000}
)

write_to_s3(response.json())

new_cursor = response.json()["next_cursor"]
write_to_metadata_store("api_cursor", new_cursor)

Even though it looks like a stateless operation, every Airflow task is prone to retry once an issue occurs. The schemas might change in the source, or a temporary network issue may lead to partial results.

Estuary treats the state as a significant part of the flow. It’s not dependent on the orchestration metadata anymore. We can easily detect API changes on top of changelog tables:

SELECT
    payload,
    cursor,
    captured_at
FROM api_capture_events
ORDER BY captured_at;

Dependency Chains and Cascading Failures

As your data grows, so do your requirements. That’s why most teams start looking for better ways to handle it, which, in turn, increases internal dependencies between tasks.

When it comes to orchestration, it’s easy to view a job as a series of interdependent tasks. They all rely on one other, thanks to the trigger mechanism. Basically, once a task completes, it triggers the next one, and the chain reaction continues.

Take, for example, a batch pipeline that gets data from an API, writes the results into an S3 bucket, loads data into Snowflake, and creates analytical tables to be used in Looker.

The first step is running this Python script in EC2, which gets data from a sample API and writes the results to the S3 bucket:

import json
import time
import requests
import boto3
from datetime import datetime

API_URL = "https://api.example.com/v1/events"
API_TOKEN = "API_TOKEN"
S3_BUCKET = "raw-api-ingestion"
S3_PREFIX = "events"

s3 = boto3.client("s3")

def fetch_events(page):
    response = requests.get(
        API_URL,
        headers={"Authorization": f"Bearer {API_TOKEN}"},
        params={"page": page, "limit": 100}
    )
    response.raise_for_status()
    return response.json()

def write_to_s3(data, page):
    timestamp = datetime.utcnow()
    key = f"{S3_PREFIX}/dt={timestamp}/events_page_{page}.json"

    s3.put_object(
        Bucket=S3_BUCKET,
        Key=key,
        Body=json.dumps(data),
        ContentType="application/json"
    )

def ingest_data():
    page = 1
    while True:
        payload = fetch_events(page)

        if not payload["data"]:
            break

        write_to_s3(payload["data"], page)

        page += 1
        time.sleep(0.5)

if __name__ == "__main__":
    ingest_data()

Next, we need to create an external stage in Snowflake, which mirrors the data that we have in the S3 bucket.

We also need to set up an hourly Snowflake task that copies the data from the stage into a raw table:

COPY INTO snowflake_raw.api_events
FROM @api_stage
FILE_FORMAT = (TYPE = JSON);

In the end, we need to run SQL queries to aggregate the raw data. This is going to create tables that will be used in the analytical layer in Looker:

CREATE OR REPLACE TRANSIENT TABLE events_clean AS (
SELECT
    event_date,
    event_type,
    COUNT(*) AS total_events
FROM snowflake_analytics.events
GROUP BY event_date, event_type
);

In this example, the first constraint is the API rate limits that might vary based on the time of the day. Partial responses might result in duplicates or missing data. The Snowflake COPY task can’t know whether the ingested data is complete or not, so Looker dashboards are updated with the wrong data without errors. As a result, dashboards may display missing events or duplicate entries.

When teams notice discrepancies, they need to pause Looker dashboards, re-run each step, and even backfill the entire data.

If these issues occur at any step, dashboards display the aggregated data late. Coordinating warehouse or database access across teams might be necessary, and entire database may even need to be restarted if there are no quick solutions. The main stakeholders should also be informed, regardless of whether they are internal or external users.

Estuary is capable of getting API data incrementally instead of getting partial API responses with batch pipelines. At the same time, you will keep the correct order and state. The whole process is quite straightforward. For example, let’s consider the HubSpot API, which is known for its complex structure.

On Estuary’s main page, click on Sources, search for HubSpot Incremental, and create a capture.

While defining it, you can mention all details about your HubSpot account, including your credentials. You can also specify which objects to sync, such as companies and deals. These will be the streams inside your HubSpot capture.

Even though the HubSpot API is rate-limited, Estuary will throttle it and resume.

After the HubSpot capture is created, the destination in this particular case will be Snowflake. In the Estuary UI, you can refer to the Destinations section and choose Snowflake as the materialization.

Once you’ve selected Snowflake as the destination, you should fill in the required fields with your Snowflake account information, including your credentials, the destination database, and the destination schema. You’ll also need to specify the role that is going to ingest the records and the warehouse that you will use during the ingestion.

Now you can create a binding between the newly created HubSpot capture and Snowflake materialization. You can edit your materialization and add a new binding. While defining it, you can choose the write mode, primary key, destination table, streams to ingest, schema evolution, and so on.

Once the whole sync is created, Estuary will perform an initial backfill and sync your source data to your destination with a right-time data approach.

As you can see, with Estuary, the data flow becomes much more solid:

Schema Changes and the Rising Cost of Batch Pipelines

Any structure is prone to changes in the source system, including schema changes or legacy columns. When this happens (and the communication with source team is lacking, which is often the case), the data we use might lose its meaning to stakeholders due to missing columns.

Batch pipelines struggle to adjust since they mostly rely on outdated data instead of steady methods. Once new changes are introduced to the pipeline, the result is often inaccurate data or failure in all components.

This isn’t about how teams implement the batch pipeline; it’s about the architecture in the background. Since it relies on initial conditions, any new changes can crash the working mechanism and cause failures.

Practical Example: From Airflow DAGs to Right-Time Data

When it comes to scheduling or orchestrating, the first tool that comes to your mind is probably Airflow. Think about an hourly Airflow ELT DAG:

1. It gets the data from any kind of source database, such as MySQL or PostgreSQL.
2. Then, it ingests the data into a warehouse for analytical purposes (think of BigQuery for this specific example).
3. After the data arrives in BigQuery, it aggregates the raw data located in the staging layer that was ingested in the previous step.
4. Finally, these analytical tables are used in visualization tools, like Looker.

However, what seems like a suitable setup today might change tomorrow, so keep an eye out. Hourly jobs may easily interfere with one another, backfills often start right away, and rebooting jobs demand more processing power. In addition, job links might need to be tweaked by hand, and you may have to verify new data again.

This is what the DAG will look like:

with DAG(
    dag_id="raw_orders_ingestion_and_transform",
    schedule_interval="@hourly",
    catchup=True,
) as dag:
    ingest_orders = MySqlOperator(
        task_id="ingest_orders",
        sql="""
        SELECT *
        FROM orders
        WHERE updated_at >= '{{ data_interval_start }}'
          AND updated_at < '{{ data_interval_end }}'
        """
    )
    transform_orders = BigQueryOperator(
        task_id="transform_orders",
        sql="""
        CREATE OR REPLACE TABLE analytics.orders_clean AS
        SELECT
            order_id,
            user_id,
            amount,
            status
        FROM raw_orders
        WHERE created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
        """
    )
    compute_metrics = BigQueryOperator(
        task_id="compute_metrics",
        sql="""
        INSERT INTO analytics.hourly_revenue
        SELECT
            TIMESTAMP_TRUNC(created_at, HOUR) AS hour,
            SUM(amount) AS revenue
        FROM analytics.orders_clean
        GROUP BY 1
        """
    )
    
    ingest_orders >> transform_orders >> compute_metrics

In this example, the ingest_orders task assumes that the source orders table includes all changes made over the last hour. It ingests the data into raw_orders in BigQuery. Since the transform_orders task assumes the ingest_orders task has completed successfully, late-arriving records are now a part of the transformation layer, and they are already in the orders_clean table. Because metrics are also calculated based on the results of transform_orders, dashboards may end up displaying inaccurate information.

In the case of a backfill, historical data, which downstream jobs have already consumed, is rewritten. As a result, if a retry happens in one task, it invalidates the results in another task. The important point is that none of these trigger an alert. They quietly decrease data quality instead. This causes longer chains of dependency, and when the first step breaks, everything downstream collapses.

In Estuary, instead of long chains of dependencies, changes are detected only once and then, they start flowing continuously. For this specific example, the capture is going to be MySQL.

While setting up your MySQL source, you should define all your credentials first. Then, you can configure many additional advanced settings, such as tables that you would not like to backfill, specific databases you would like to connect, and the backfill chunk size.

Once your source MySQL is ready, you can move on to creating the destination materialization. In this example, we are going to use BigQuery as the destination, and you may choose your materialization as BigQuery.

Then, you should define the credentials for your BigQuery account. There are many available parameters you can configure, including the region where your BigQuery account is located, the dataset where you would like to see your ingested tables, and the Google Cloud Storage bucket that will be used for temporary data and specifications.

With the additional advanced options, you can even specify more granular settings for your BigQuery materialization.

After creating both the source and the destination, you can establish a new binding between the MySQL source and the BigQuery destination. In the end, Estuary handles your data ingestion incrementally with the right-time data approach .

With this new implementation, you don’t need ingestion tasks in your Airflow DAGs. Since the data ingestion is going to be handled by Estuary, you can focus on locating your transformation tasks and keeping your Airflow DAGs more reliable.

This approach replaces the whole error-prone system with a steady flow of data.

With Estuary, you can sync your entire data set through log-based CDC, which will result in a steady pipeline. Even when something is updated in the source (schema change, column additions/deletions, etc.), Estuary is capable of detecting each change and automatically propagating them downstream.

The changes will have an immediate effect. Source and engineering teams can implement updates without relying on additional dependencies. Your data will remain fresh, up-to-date, and accurate, and you will not need full backfills or other operations that will create a costly load on your pipeline.

This is what real-time data actually looks like.

The Hidden Cost of Batch-Heavy Architectures

Batch pipelines are not designed to fail suddenly. Instead, they become insufficient as requirements grow.

Compute Costs That Scale Non-Linearly

If an issue happens in your batch pipeline and there is no quick workaround to resolve it, teams need to do backfills. This will come with a compute cost (warehouses in Snowflake, for example). Since data volume increases every day, each backfill will require extra compute cost over time.

Engineering Required for Recovery

One of primary goals of data engineering is to automate pipelines as much as possible. However, when an error occurs, human intervention is required.

The first thing we need to do is to detect the root cause and then re-run all the jobs if necessary. Afterward, we should perform sanity checks on downstream tables to ensure data quality. Finally, we need to communicate the issue and provide reliable updates to the stakeholders.

All of this requires significant time and effort from engineers.

Compliance and Audit Risk

Once a batch pipeline starts failing, it can lead to missing or partial data. Files or tables provided to stakeholders will lack important information, dashboards will start throwing errors, and daily/weekly reports may not work. Since failures are often partial, pipeline lineage can also become flawed.

All these issues create significant compliance and audit risks.

Trust Loss

We aim to build solid, reliable pipelines; however, just as importantly, we want to build trust with our users. In case of a critical failure, all downstream components will be broken. This will definitely impact the trust between engineers and end-users in a negative way, and more questions will arise regarding data accuracy.

Trust is built through consistency. Once an error happens or data loses its quality, this consistency is lost, and so is the confidence in the pipeline.

Why Is a Unified Real-Time and Batch Approach Better?

A unified approach treats data movement as a continuous flow rather than a series of scheduled jobs. The more frequently we check data freshness, the sooner stale data or any related issues can be detected. Instead of waiting for batches, real-time data flows enable smoother operations as a whole.

Modern AI or ML systems in particular are dependent on fresh and accurate data. By using incremental check principles, we can make sure our data is accurate and up-to-date. Since our models are trained on it, its stability is critical to prevent drifts. Real-time data is also essential if we want embeddings to remain current while agents are operating. At the same time, regulatory and audit requirements make clear lineage definitions essential.

Once pipelines are split into batch and streaming, data definitions become different, costs are duplicated, and complexity becomes hard to maintain.

A unified data movement layer solves all this.

Continuous Sync and Lower Warehouse Costs

Since Estuary syncs data incrementally, heavy computation costs are significantly reduced. Compute workers only process incremental changes rather than repeatedly scanning full tables, which results in more efficient warehouse usage.

One Platform, One Model

Estuary uses an approach that handles batch and real-time processing together. This means the same reliability promises, schema evolution guarantees, and governance model apply consistently across both systems. The architecture is simplified and operational maintenance is significantly reduced.

Flexible Deployment for Enterprise Requirements

Most pipelines are deployed into multiple cloud providers rather than relying on a single one. This is often done due to hybrid environment requirements or strict compliance rules.

Estuary’s BYOC and hybrid deployment models address these enterprise needs without demanding any additional architectural changes to your system.

Conclusion

Estuary is more than just another pipeline tool. It creates a solid structure that makes data move smoothly across multiple systems, and it gives you the ability to leverage its power for analytics, AI/ML, and compliance.

Once batch pipelines start to become fragile due to the increasing data volumes, the issue is rarely about how teams operate or whether everything was implemented correctly. The root cause is the architecture itself, which was probably not designed to meet modern-day expectations.

The best solution lies in the unified data movement. It creates a fully reliable system where you can reduce costs significantly, minimize the risk of downtimes and errors, and rebuild trust with your stakeholders.

With Estuary, this transition is smooth and reliable.