How to Implement AI-Powered Data Anomaly Detection with LangChain 0.3, Pandas 2.2, and Slack Alerts

import sys
import subprocess
import os
from pathlib import Path
import logging

# Configure logging for setup script
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

def install_dependencies():
    """Install required Python dependencies with version pinning for reproducibility."""
    required_packages = [
        "langchain==0.3.15",  # Pinned LangChain 0.3.x for stability
        "langchain-openai==0.2.14",  # OpenAI integration for LangChain 0.3
        "pandas==2.2.3",  # Pinned Pandas 2.2.x
        "slack-sdk==3.23.0",  # Slack Web API SDK
        "python-dotenv==1.0.1",  # Environment variable management
        "pyarrow==17.0.0",  # Parquet support for Pandas 2.2
        "kafka-python==2.0.2"  # Optional: Kafka streaming support
    ]

    logger.info("Installing dependencies: %s", required_packages)
    try:
        # Use sys.executable to ensure pip matches current Python environment
        subprocess.check_call(
            [sys.executable, "-m", "pip", "install", "-q"] + required_packages,
            stderr=subprocess.DEVNULL
        )
        logger.info("All dependencies installed successfully.")
    except subprocess.CalledProcessError as e:
        logger.error("Failed to install dependencies: %s", e)
        sys.exit(1)
    except Exception as e:
        logger.error("Unexpected error during installation: %s", e)
        sys.exit(1)

def setup_env_file():
    """Create .env file with required configuration variables, prompting user for inputs."""
    env_path = Path(".env")
    if env_path.exists():
        logger.warning(".env file already exists, skipping creation.")
        return

    logger.info("Setting up .env file. Please provide the following values:")
    openai_api_key = input("Enter your OpenAI API key: ").strip()
    slack_bot_token = input("Enter your Slack Bot User OAuth Token (xoxb-...): ").strip()
    slack_channel_id = input("Enter your Slack Channel ID (e.g., C1234567890): ").strip()

    env_contents = f"""# LangChain Configuration
OPENAI_API_KEY={openai_api_key}
LANGCHAIN_TRACING_V2=false  # Set to true for LangSmith debugging

# Slack Configuration
SLACK_BOT_TOKEN={slack_bot_token}
SLACK_CHANNEL_ID={slack_channel_id}

# Data Configuration
DATA_DIR=./data
ANOMALY_THRESHOLD=0.95  # Confidence threshold for LangChain alerts
"""
    try:
        with open(env_path, "w") as f:
            f.write(env_contents)
        logger.info(".env file created successfully at %s", env_path.absolute())
    except IOError as e:
        logger.error("Failed to write .env file: %s", e)
        sys.exit(1)

if __name__ == "__main__":
    logger.info("Starting environment setup for AI Anomaly Detection pipeline...")
    install_dependencies()
    setup_env_file()
    logger.info("Setup complete. Proceed to Step 2: Data Ingestion with Pandas 2.2.")

import os
import pandas as pd
import logging
from pathlib import Path
from dotenv import load_dotenv
from typing import Optional, Union
import pyarrow.parquet as pq  # Pandas 2.2 uses PyArrow as default Parquet backend

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class DataIngestor:
    """Handle data ingestion from CSV, Parquet, or Kafka streams using Pandas 2.2."""

    def __init__(self, data_dir: Union[str, Path] = os.getenv("DATA_DIR", "./data")):
        self.data_dir = Path(data_dir)
        self.data_dir.mkdir(exist_ok=True)
        logger.info("Initialized DataIngestor with data directory: %s", self.data_dir.absolute())

    def ingest_csv(self, file_path: Union[str, Path], parse_dates: Optional[list] = None) -> pd.DataFrame:
        """
        Ingest CSV file using Pandas 2.2's enhanced type inference.

        Args:
            file_path: Path to CSV file
            parse_dates: List of columns to parse as datetime

        Returns:
            Preprocessed Pandas DataFrame
        """
        file_path = Path(file_path)
        if not file_path.exists():
            logger.error("CSV file not found: %s", file_path.absolute())
            raise FileNotFoundError(f"CSV file {file_path} does not exist.")

        try:
            # Pandas 2.2 enables nullable dtypes by default for better type safety
            df = pd.read_csv(
                file_path,
                parse_dates=parse_dates,
                dtype_backend="numpy_nullable",  # Pandas 2.2 feature: avoids object dtype bloat
                engine="c",  # Faster than Python engine for large files
                low_memory=False  # Handle files larger than 1GB
            )
            logger.info("Ingested CSV with %d rows, %d columns", df.shape[0], df.shape[1])
            return self._preprocess_df(df)
        except pd.errors.ParserError as e:
            logger.error("Failed to parse CSV %s: %s", file_path, e)
            raise
        except Exception as e:
            logger.error("Unexpected error ingesting CSV %s: %s", file_path, e)
            raise

    def ingest_parquet(self, file_path: Union[str, Path]) -> pd.DataFrame:
        """
        Ingest Parquet file using Pandas 2.2's PyArrow backend.

        Args:
            file_path: Path to Parquet file

        Returns:
            Preprocessed Pandas DataFrame
        """
        file_path = Path(file_path)
        if not file_path.exists():
            logger.error("Parquet file not found: %s", file_path.absolute())
            raise FileNotFoundError(f"Parquet file {file_path} does not exist.")

        try:
            # Pandas 2.2 uses PyArrow as default Parquet backend, 3x faster than fastparquet
            df = pd.read_parquet(
                file_path,
                engine="pyarrow",
                dtype_backend="numpy_nullable"
            )
            logger.info("Ingested Parquet with %d rows, %d columns", df.shape[0], df.shape[1])
            return self._preprocess_df(df)
        except pq.ParquetError as e:
            logger.error("Failed to parse Parquet %s: %s", file_path, e)
            raise
        except Exception as e:
            logger.error("Unexpected error ingesting Parquet %s: %s", file_path, e)
            raise

    def _preprocess_df(self, df: pd.DataFrame) -> pd.DataFrame:
        """Internal method to preprocess DataFrames: handle missing values, sort by timestamp."""
        # Drop duplicate rows, a common issue in streaming data
        duplicate_count = df.duplicated().sum()
        if duplicate_count > 0:
            logger.warning("Dropped %d duplicate rows from DataFrame", duplicate_count)
            df = df.drop_duplicates()

        # Sort by timestamp if present, critical for time-series anomaly detection
        if "timestamp" in df.columns:
            df["timestamp"] = pd.to_datetime(df["timestamp"])
            df = df.sort_values("timestamp").reset_index(drop=True)
            logger.info("Sorted DataFrame by timestamp column.")

        # Fill numeric missing values with rolling median (Pandas 2.2's rolling is 20% faster)
        numeric_cols = df.select_dtypes(include=["number"]).columns
        for col in numeric_cols:
            if df[col].isna().any():
                median_fill = df[col].rolling(window=10, min_periods=1).median()
                df[col] = df[col].fillna(median_fill)
                logger.info("Filled missing values in numeric column %s with rolling median", col)

        return df

if __name__ == "__main__":
    # Example usage: ingest sample e-commerce clickstream data
    ingestor = DataIngestor()
    try:
        # Download sample data if not present (1.2GB clickstream dataset)
        sample_csv = ingestor.data_dir / "clickstream_sample.csv"
        if not sample_csv.exists():
            logger.info("Downloading sample clickstream data...")
            import requests
            url = "https://example.com/clickstream_sample.csv"  # Replace with real sample URL
            response = requests.get(url, stream=True)
            with open(sample_csv, "wb") as f:
                for chunk in response.iter_content(chunk_size=8192):
                    f.write(chunk)
            logger.info("Sample data downloaded to %s", sample_csv.absolute())

        df = ingestor.ingest_csv(sample_csv, parse_dates=["timestamp"])
        logger.info("Sample data preview:\n%s", df.head().to_string())
    except Exception as e:
        logger.error("Failed to run sample ingestion: %s", e)
        sys.exit(1)

import os
import pandas as pd
import logging
from dotenv import load_dotenv
from langchain_community.document_loaders import PandasLoader
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from typing import List, Dict, Any
import numpy as np

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class LangChainAnomalyDetector:
    """Detect anomalies using LangChain 0.3's LLM integration and Pandas 2.2."""

    def __init__(
        self,
        llm_model: str = "gpt-4o-mini",  # Cost-effective LLM for anomaly context
        confidence_threshold: float = float(os.getenv("ANOMALY_THRESHOLD", 0.95))
    ):
        self.confidence_threshold = confidence_threshold
        try:
            self.llm = ChatOpenAI(
                model=llm_model,
                temperature=0,  # Deterministic output for anomaly detection
                max_retries=3  # Handle rate limits
            )
            logger.info("Initialized LangChain ChatOpenAI with model: %s", llm_model)
        except Exception as e:
            logger.error("Failed to initialize LLM: %s", e)
            raise

        # Prompt template for anomaly context: provides business logic to LLM
        self.prompt_template = PromptTemplate(
            input_variables=["metric_name", "current_value", "historical_mean", "historical_std", "timestamp", "business_context"],
            template="""You are a senior data engineer analyzing time-series anomalies for an e-commerce platform.

Metric: {metric_name}
Current Value: {current_value}
Historical Mean: {historical_mean:.2f}
Historical Standard Deviation: {historical_std:.2f}
Timestamp: {timestamp}
Business Context: {business_context}

Determine if this is a true positive anomaly or a false positive. Return a JSON object with:
- is_anomaly: boolean (true if anomaly)
- confidence: float between 0 and 1
- reason: string explaining the decision, including business context if relevant.

Only return the JSON object, no additional text."""
        )

        self.chain = LLMChain(llm=self.llm, prompt=self.prompt_template)
        logger.info("Initialized LLMChain with anomaly detection prompt.")

    def calculate_statistical_anomalies(self, df: pd.DataFrame, metric_col: str) -> pd.DataFrame:
        """
        Calculate statistical anomalies using 3-sigma rule (baseline for LangChain comparison).

        Args:
            df: Input DataFrame with time-series data
            metric_col: Column name of the metric to analyze

        Returns:
            DataFrame with statistical anomaly flags
        """
        if metric_col not in df.columns:
            logger.error("Metric column %s not found in DataFrame", metric_col)
            raise ValueError(f"Column {metric_col} does not exist.")

        # Calculate rolling mean and std (30-minute window for clickstream data)
        df[f"{metric_col}_rolling_mean"] = df[metric_col].rolling(window=30, min_periods=5).mean()
        df[f"{metric_col}_rolling_std"] = df[metric_col].rolling(window=30, min_periods=5).std()

        # 3-sigma threshold
        df[f"{metric_col}_upper"] = df[f"{metric_col}_rolling_mean"] + 3 * df[f"{metric_col}_rolling_std"]
        df[f"{metric_col}_lower"] = df[f"{metric_col}_rolling_mean"] - 3 * df[f"{metric_col}_rolling_std"]

        # Flag statistical anomalies
        df["statistical_anomaly"] = (
            (df[metric_col] > df[f"{metric_col}_upper"]) | 
            (df[metric_col] < df[f"{metric_col}_lower"])
        )

        anomaly_count = df["statistical_anomaly"].sum()
        logger.info("Found %d statistical anomalies (3-sigma) for metric %s", anomaly_count, metric_col)
        return df

    def detect_anomalies(self, df: pd.DataFrame, metric_col: str, business_context: str = "General e-commerce traffic") -> List[Dict[str, Any]]:
        """
        Run LangChain-powered anomaly detection on statistical anomalies.

        Args:
            df: DataFrame with statistical anomalies pre-calculated
            metric_col: Metric column to analyze
            business_context: Business context to provide to LLM (e.g., "Black Friday sale")

        Returns:
            List of confirmed anomaly dictionaries
        """
        statistical_anomalies = df[df["statistical_anomaly"]].copy()
        if statistical_anomalies.empty:
            logger.info("No statistical anomalies found, skipping LangChain analysis.")
            return []

        confirmed_anomalies = []
        logger.info("Analyzing %d statistical anomalies with LangChain...", len(statistical_anomalies))

        for idx, row in statistical_anomalies.iterrows():
            try:
                # Prepare input for LangChain prompt
                input_data = {
                    "metric_name": metric_col,
                    "current_value": row[metric_col],
                    "historical_mean": row[f"{metric_col}_rolling_mean"],
                    "historical_std": row[f"{metric_col}_rolling_std"],
                    "timestamp": row["timestamp"].isoformat(),
                    "business_context": business_context
                }

                # Run LLM chain
                result = self.chain.run(input_data)
                # Parse JSON result (handle LLM output variations)
                import json
                result_json = json.loads(result.replace("", "").replace("", "").strip())

                if result_json.get("is_anomaly", False) and result_json.get("confidence", 0) >= self.confidence_threshold:
                    anomaly = {
                        "timestamp": row["timestamp"].isoformat(),
                        "metric": metric_col,
                        "value": row[metric_col],
                        "deviation": row[metric_col] - row[f"{metric_col}_rolling_mean"],
                        "confidence": result_json["confidence"],
                        "reason": result_json["reason"],
                        "business_context": business_context
                    }
                    confirmed_anomalies.append(anomaly)
                    logger.info("Confirmed anomaly at %s: %s", anomaly["timestamp"], anomaly["reason"])

            except json.JSONDecodeError as e:
                logger.warning("Failed to parse LLM output for row %d: %s", idx, e)
                continue
            except Exception as e:
                logger.error("Error analyzing row %d: %s", idx, e)
                continue

        logger.info("Confirmed %d true positive anomalies out of %d statistical anomalies", len(confirmed_anomalies), len(statistical_anomalies))
        return confirmed_anomalies

if __name__ == "__main__":
    # Example usage with sample data
    from data_ingestor import DataIngestor  # Import from Step 2 script
    load_dotenv()

    ingestor = DataIngestor()
    df = ingestor.ingest_csv(ingestor.data_dir / "clickstream_sample.csv", parse_dates=["timestamp"])

    detector = LangChainAnomalyDetector()
    df = detector.calculate_statistical_anomalies(df, metric_col="page_views")
    anomalies = detector.detect_anomalies(
        df, 
        metric_col="page_views", 
        business_context="Black Friday marketing campaign running from 2024-11-29 to 2024-12-02"
    )
    logger.info("Detected %d confirmed anomalies: %s", len(anomalies), anomalies[:2])

import os
import logging
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from dotenv import load_dotenv
from typing import Dict, Any

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class SlackAlerter:
    """Send anomaly alerts to Slack via Block Kit API."""

    def __init__(self):
        self.client = WebClient(token=os.getenv("SLACK_BOT_TOKEN"))
        self.channel_id = os.getenv("SLACK_CHANNEL_ID")
        logger.info("Initialized SlackAlerter for channel: %s", self.channel_id)

    def send_anomaly_alert(self, anomaly: Dict[str, Any]) -> bool:
        """Send a Block Kit alert for a confirmed anomaly."""
        blocks = [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"🚨 Anomaly Detected: {anomaly['metric']}",
                    "emoji": True
                }
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Timestamp:*
{anomaly['timestamp']}"},
                    {"type": "mrkdwn", "text": f"*Value:*
{anomaly['value']}"},
                    {"type": "mrkdwn", "text": f"*Deviation:*
{anomaly['deviation']:.2f}"},
                    {"type": "mrkdwn", "text": f"*Confidence:*
{anomaly['confidence']:.0%}"}
                ]
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Reason:* {anomaly['reason']}"
                }
            },
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "Acknowledge", "emoji": True},
                        "style": "primary",
                        "value": "ack_anomaly"
                    },
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "View Raw Data", "emoji": True},
                        "url": "https://example.com/s3/bucket/clickstream"  # Replace with your S3 link
                    }
                ]
            }
        ]

        try:
            response = self.client.chat_postMessage(
                channel=self.channel_id,
                blocks=blocks,
                text=f"Anomaly Detected: {anomaly['metric']}"  # Fallback for notifications
            )
            logger.info("Slack alert sent successfully: %s", response["ts"])
            return True
        except SlackApiError as e:
            logger.error("Failed to send Slack alert: %s", e.response["error"])
            return False
        except Exception as e:
            logger.error("Unexpected error sending Slack alert: %s", e)
            return False

if __name__ == "__main__":
    # Test alert with sample anomaly
    sample_anomaly = {
        "timestamp": "2024-11-29T10:00:00",
        "metric": "page_views",
        "value": 15234,
        "deviation": 12345.67,
        "confidence": 0.98,
        "reason": "Spike aligns with Black Friday marketing campaign launch, 300% above historical mean.",
        "business_context": "Black Friday sale"
    }
    alerter = SlackAlerter()
    alerter.send_anomaly_alert(sample_anomaly)

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

# Enable in-memory caching for LangChain LLM calls
set_llm_cache(InMemoryCache())

# Reuse cached responses for identical anomaly inputs
detector = LangChainAnomalyDetector()
anomalies = detector.detect_anomalies(df, "page_views", "Black Friday sale")

import pandas as pd

# Convert existing DataFrame to Pandas 2.2 nullable dtypes
df = pd.read_csv("legacy_data.csv")
df = df.convert_dtypes(dtype_backend="numpy_nullable")
print(df.dtypes)  # Int64, Float64, string instead of object

import slack_sdk
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def send_slack_alert(channel_id: str, alert_text: str):
    client = slack_sdk.WebClient(token=os.getenv("SLACK_BOT_TOKEN"))
    response = client.chat_postMessage(channel=channel_id, text=alert_text)
    if not response["ok"]:
        raise Exception(f"Slack API error: {response['error']}")
    return response

ai-anomaly-detection-langchain-pandas/
├── data/                     # Sample data directory
│   └── clickstream_sample.csv
├── src/
│   ├── __init__.py
│   ├── setup.py              # Environment setup script (Code Example 1)
│   ├── data_ingestor.py      # Pandas 2.2 data ingestion (Code Example 2)
│   ├── anomaly_detector.py   # LangChain 0.3 anomaly detection (Code Example 3)
│   └── slack_alert.py        # Slack alert integration
├── .env.example              # Example environment variables
├── requirements.txt          # Pinned dependencies
├── README.md                 # Tutorial instructions
└── benchmarks/               # Benchmark results and scripts
    └── anomaly_benchmark.csv