Application Development in the AI Era

The Shift from Data-Serving to Intelligence-Serving

For most of software history, applications were designed to serve data to humans. A dashboard showed numbers. A report showed trends. A table showed records. Humans interpreted the data and made decisions.

The AI era inverted this: applications now serve data to AI systems, which generate insights that humans consume. The architectural implications are profound.

The Old Application Architecture

User Request
    │
    ▼
Web Server (Flask/Django)
    │
    ▼
Database Query (SQL)
    │
    ▼
Template Rendering / JSON API
    │
    ▼
Browser (Human reads the data)

This architecture optimized for human readability: tables, charts, pagination.

The New AI-First Application Architecture

User Query (Natural Language)
    │
    ▼
API Gateway
    │
    ├──────────────────────┐
    ▼                      ▼
Orchestration Layer    Auth/Rate Limit
(LangChain / LlamaIndex)
    │
    ├──── Vector DB (Semantic Search)
    ├──── SQL Generator (Text-to-SQL)
    ├──── Tool Calls (APIs, Functions)
    └──── LLM (Reasoning/Generation)
    │
    ▼
Structured Response
    │
    ▼
User (Human reads AI-generated insight)

Pattern 1: Text-to-SQL Applications

The most immediately impactful pattern: letting users query data warehouses in plain English.

# FastAPI + LangChain Text-to-SQL implementation
from fastapi import FastAPI
from langchain_community.utilities import SQLDatabase
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import boto3

app = FastAPI()

# Connect to your data warehouse
db = SQLDatabase.from_uri("snowflake://user:pass@account/database/schema")

# Build the chain
prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an expert SQL analyst. Given the following table schemas:

{schema}

Generate a valid SQL query to answer the user's question.
Return ONLY the SQL query, no explanation.
Always limit results to 1000 rows unless asked for more."""),
    ("human", "{question}")
])

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def get_schema(_):
    return db.get_table_info()

chain = (
    {"schema": get_schema, "question": lambda x: x["question"]}
    | prompt
    | llm
    | StrOutputParser()
)

@app.post("/query")
async def natural_language_query(question: str):
    """Convert natural language to SQL and execute."""
    sql = chain.invoke({"question": question})

    # Safety: only allow SELECT statements
    if not sql.strip().upper().startswith("SELECT"):
        return {"error": "Only SELECT queries are allowed"}

    result = db.run(sql)
    return {
        "question": question,
        "sql": sql,
        "result": result
    }

Real-world example:

• User asks: "What were our top 5 products by revenue last quarter in the APAC region?"
• System generates: SELECT product_name, SUM(revenue) as total_revenue FROM orders WHERE region = 'APAC' AND order_date BETWEEN '2024-01-01' AND '2024-03-31' GROUP BY product_name ORDER BY total_revenue DESC LIMIT 5
• Returns structured insight to user

Pattern 2: RAG (Retrieval-Augmented Generation) Data Applications

For unstructured data (documents, logs, support tickets), RAG pipelines make data queryable.

# End-to-end RAG pipeline for data documentation
import boto3
from langchain_aws import BedrockEmbeddings, ChatBedrock
from langchain_community.vectorstores import OpenSearchVectorSearch
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain_core.documents import Document

# Step 1: Ingest data documentation into vector store
def ingest_data_catalog(catalog_entries: list[dict]):
    """Convert data catalog entries into searchable vectors."""
    embeddings = BedrockEmbeddings(
        client=boto3.client('bedrock-runtime'),
        model_id="amazon.titan-embed-text-v1"
    )

    documents = []
    for entry in catalog_entries:
        doc_text = f"""
        Table: {entry['table_name']}
        Description: {entry['description']}
        Owner: {entry['owner']}
        Columns: {', '.join([f"{c['name']} ({c['type']}): {c['description']}" 
                              for c in entry['columns']])}
        Sample Data: {entry.get('sample_values', 'N/A')}
        Business Context: {entry.get('business_context', 'N/A')}
        """
        documents.append(Document(
            page_content=doc_text,
            metadata={"table": entry['table_name'], "owner": entry['owner']}
        ))

    # Split and embed
    splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    splits = splitter.split_documents(documents)

    vectorstore = OpenSearchVectorSearch.from_documents(
        splits,
        embeddings,
        opensearch_url="https://your-opensearch-endpoint"
    )
    return vectorstore

# Step 2: Query the catalog
def answer_data_question(question: str, vectorstore):
    """Answer questions about your data using RAG."""
    llm = ChatBedrock(
        client=boto3.client('bedrock-runtime'),
        model_id="anthropic.claude-3-sonnet-20240229-v1:0"
    )

    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
        return_source_documents=True
    )

    result = qa_chain.invoke({"query": question})
    return {
        "answer": result["result"],
        "sources": [doc.metadata["table"] for doc in result["source_documents"]]
    }

# Usage
answer = answer_data_question("Which tables contain customer transaction history?", vectorstore)

Pattern 3: Feature Stores as Application Infrastructure

ML-powered applications need precomputed features served at low latency. Feature stores bridge the offline data lake and online application layer.

# Feast feature store integration
from feast import FeatureStore, Entity, FeatureView, Field, FileSource
from feast.types import Float64, Int64, String
import pandas as pd

# Define features computed offline
customer_stats = FeatureView(
    name="customer_stats",
    entities=["customer_id"],
    ttl=timedelta(hours=1),
    schema=[
        Field(name="total_orders_30d", dtype=Int64),
        Field(name="avg_order_value_30d", dtype=Float64),
        Field(name="days_since_last_order", dtype=Int64),
        Field(name="preferred_category", dtype=String),
        Field(name="churn_risk_score", dtype=Float64),
    ],
    source=FileSource(path="s3://feature-store/customer_stats.parquet")
)

store = FeatureStore(repo_path="./feature_repo")

# In your application: serve features at <5ms
@app.get("/recommend/{customer_id}")
async def get_recommendations(customer_id: str):
    # Fetch precomputed features instantly
    features = store.get_online_features(
        features=[
            "customer_stats:total_orders_30d",
            "customer_stats:churn_risk_score",
            "customer_stats:preferred_category"
        ],
        entity_rows=[{"customer_id": customer_id}]
    ).to_dict()

    # Use features for real-time decision
    churn_risk = features["churn_risk_score"][0]
    category = features["preferred_category"][0]

    if churn_risk > 0.7:
        return {"type": "retention_offer", "category": category}
    else:
        return {"type": "upsell", "category": category}

Pattern 4: Streaming Intelligence Applications

Real-time ML inference embedded in streaming pipelines.

# Flink/Kafka Streams + ML inference
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

# Source: Real-time transaction stream
t_env.execute_sql("""
    CREATE TABLE transactions (
        transaction_id STRING,
        customer_id STRING,
        amount DOUBLE,
        merchant_category STRING,
        timestamp TIMESTAMP(3),
        WATERMARK FOR timestamp AS timestamp - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'raw_transactions',
        'properties.bootstrap.servers' = 'kafka:9092',
        'format' = 'json'
    )
""")

# AI-powered fraud detection sink
t_env.execute_sql("""
    CREATE TABLE fraud_alerts (
        transaction_id STRING,
        customer_id STRING,
        amount DOUBLE,
        fraud_probability DOUBLE,
        alert_reason STRING
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'fraud_alerts',
        'properties.bootstrap.servers' = 'kafka:9092',
        'format' = 'json'
    )
""")

# Real-time fraud scoring via ML model call
t_env.execute_sql("""
    INSERT INTO fraud_alerts
    SELECT 
        transaction_id,
        customer_id,
        amount,
        ML_PREDICT('fraud_detection_model', amount, merchant_category) as fraud_probability,
        CASE 
            WHEN amount > 10000 THEN 'High value transaction'
            WHEN merchant_category = 'gambling' THEN 'High-risk merchant'
            ELSE 'Behavioral anomaly'
        END as alert_reason
    FROM transactions
    WHERE ML_PREDICT('fraud_detection_model', amount, merchant_category) > 0.85
""")

The Data Contract for AI Applications

AI applications have unique data requirements that traditional applications didn't:

Summary

Modern application development in the data engineering context means:

1. Designing schemas for machines, not just humans — AI needs semantic context
2. Building feature pipelines that serve ML models at production latency
3. Creating RAG infrastructure to make institutional knowledge queryable
4. Instrumenting AI pipelines with the same observability as traditional software
5. Governing AI-generated outputs as carefully as human-generated data