datapilot/workflow.py

from __future__ import annotations

import time
from typing import Any, TypedDict
from uuid import uuid4

import numpy as np
import pandas as pd
from langgraph.graph import END, START, StateGraph

from datapilot.config import Settings, get_settings
from datapilot.data import duckdb_overview
from datapilot.insights import deterministic_insights, optional_llm_narrative
from datapilot.modeling import (
    TrainingBundle,
    critic_decision,
    explain_model,
    train_models,
)
from datapilot.observability import log_to_mlflow
from datapilot.persistence import ArtifactStore, RunStore
from datapilot.quality import audit_quality, build_profile
from datapilot.reports import export_artifacts
from datapilot.schemas import AnalysisPlan, RunSummary, TaskType


class AgentState(TypedDict, total=False):
    run_id: str
    dataset_name: str
    frame: pd.DataFrame
    target: str
    settings: Settings
    profile: Any
    quality_issues: list[Any]
    evidence: list[Any]
    eda: dict[str, Any]
    statistics: dict[str, Any]
    plan: AnalysisPlan
    feature_plan: dict[str, Any]
    model_bundle: TrainingBundle
    critic: Any
    explainability: Any
    executive_summary: list[str]
    recommendations: list[str]
    summary_payload: dict[str, Any]
    artifacts: dict[str, str]
    trace: list[dict[str, Any]]


def _trace(state: AgentState, agent: str, started: float, detail: str) -> list[dict[str, Any]]:
    trace = list(state.get("trace", []))
    trace.append(
        {
            "agent": agent,
            "status": "completed",
            "duration_seconds": round(time.perf_counter() - started, 3),
            "detail": detail,
        }
    )
    return trace


def data_quality_agent(state: AgentState) -> dict[str, Any]:
    started = time.perf_counter()
    profile = build_profile(state["frame"], state["target"])
    issues, evidence = audit_quality(state["frame"], profile)
    return {
        "profile": profile,
        "quality_issues": issues,
        "evidence": evidence,
        "trace": _trace(
            state, "Data Quality Agent", started, f"Recorded {len(issues)} quality observations."
        ),
    }


def eda_agent(state: AgentState) -> dict[str, Any]:
    started = time.perf_counter()
    frame = state["frame"]
    overview = duckdb_overview(frame)
    overview["numeric_summary"] = (
        frame.select_dtypes(include=np.number).describe().round(4).to_dict()
    )
    overview["categorical_cardinality"] = {
        column: int(frame[column].nunique(dropna=True))
        for column in frame.select_dtypes(exclude=np.number).columns
    }
    return {
        "eda": overview,
        "trace": _trace(state, "EDA Agent", started, "Computed DuckDB-backed dataset overview."),
    }


def statistical_agent(state: AgentState) -> dict[str, Any]:
    started = time.perf_counter()
    frame = state["frame"]
    target = state["target"]
    numeric = frame.select_dtypes(include=np.number)
    correlations: dict[str, float] = {}
    if target in numeric.columns and len(numeric.columns) > 1:
        correlations = (
            numeric.corr(numeric_only=True)[target]
            .drop(labels=[target])
            .abs()
            .sort_values(ascending=False)
            .head(10)
            .round(4)
            .to_dict()
        )
    statistics = {
        "top_absolute_target_correlations": correlations,
        "target_distribution": frame[target].value_counts(dropna=False).head(20).to_dict(),
    }
    return {
        "statistics": statistics,
        "trace": _trace(
            state,
            "Statistical Analysis Agent",
            started,
            "Measured target distribution and associations.",
        ),
    }


def planning_agent(state: AgentState) -> dict[str, Any]:
    started = time.perf_counter()
    profile = state["profile"]
    if profile.task_type == TaskType.classification:
        metric = "balanced_accuracy"
        candidates = [
            "Logistic Regression",
            "Random Forest",
            "Extra Trees",
            "Histogram Gradient Boosting",
            "XGBoost (when installed)",
        ]
    else:
        metric = "r2"
        candidates = [
            "Linear Regression",
            "Random Forest",
            "Extra Trees",
            "Histogram Gradient Boosting",
            "XGBoost (when installed)",
        ]
    plan = AnalysisPlan(
        objective=f"Predict '{profile.target}' and produce reproducible, evidence-backed insights.",
        target=profile.target,
        task_type=profile.task_type,
        primary_metric=metric,
        validation_strategy="Training-only stratified cross-validation; untouched final test evaluation"
        if profile.task_type == TaskType.classification
        else "Training-only cross-validation; untouched final test evaluation",
        candidate_models=candidates,
        risk_controls=[
            "Drop rows with missing target before split",
            "Fit imputers, encoders, and scalers on training folds only",
            "Flag leakage-like names and identifier cardinality",
            "Require critic quality gate before explanation",
        ],
    )
    return {
        "plan": plan,
        "trace": _trace(state, "Planning Agent", started, f"Selected {metric} as primary metric."),
    }


def feature_engineering_agent(state: AgentState) -> dict[str, Any]:
    started = time.perf_counter()
    profile = state["profile"]
    feature_plan = {
        "numeric": "Median imputation followed by standard scaling",
        "categorical": "Most-frequent imputation followed by unknown-safe one-hot encoding",
        "fit_scope": "Preprocessing is fitted inside each sklearn Pipeline after splitting",
        "dropped": ["exact duplicate rows", "rows with missing target"],
        "feature_count": profile.columns - 1,
    }
    return {
        "feature_plan": feature_plan,
        "trace": _trace(
            state,
            "Feature Engineering Agent",
            started,
            "Created leakage-safe ColumnTransformer plan.",
        ),
    }


def modeling_agent(state: AgentState) -> dict[str, Any]:
    started = time.perf_counter()
    retry = state.get("model_bundle").retry_number + 1 if state.get("model_bundle") else 0
    bundle = train_models(
        state["frame"],
        state["target"],
        state["profile"].task_type,
        state["settings"],
        retry_number=retry,
    )
    return {
        "model_bundle": bundle,
        "trace": _trace(
            state,
            "Modeling Agent",
            started,
            f"Compared {len(bundle.results)} models; {bundle.best_model} ranked first.",
        ),
    }


def evaluation_critic_agent(state: AgentState) -> dict[str, Any]:
    started = time.perf_counter()
    decision = critic_decision(state["model_bundle"], state["profile"].task_type, state["settings"])
    detail = "Approved analysis." if decision.approved else "Rejected analysis and requested retry."
    return {
        "critic": decision,
        "trace": _trace(state, "Evaluation / Critic Agent", started, detail),
    }


def critic_route(state: AgentState) -> str:
    return "explainability" if state["critic"].approved else "retry_modeling"


def explainability_agent(state: AgentState) -> dict[str, Any]:
    started = time.perf_counter()
    result = explain_model(state["model_bundle"])
    return {
        "explainability": result,
        "trace": _trace(
            state, "Explainability Agent", started, f"Generated {result.method} explanations."
        ),
    }


def executive_insights_agent(state: AgentState) -> dict[str, Any]:
    started = time.perf_counter()
    summary, recommendations = deterministic_insights(state)
    llm_summary = optional_llm_narrative(state, state["evidence"], state["settings"])
    if llm_summary:
        summary = llm_summary
    return {
        "executive_summary": summary,
        "recommendations": recommendations,
        "trace": _trace(
            state,
            "Executive Insights Agent",
            started,
            "Created evidence-grounded narrative with deterministic metric provenance.",
        ),
    }


def build_graph():
    graph = StateGraph(AgentState)
    graph.add_node("data_quality", data_quality_agent)
    graph.add_node("eda", eda_agent)
    graph.add_node("statistics", statistical_agent)
    graph.add_node("planning", planning_agent)
    graph.add_node("feature_engineering", feature_engineering_agent)
    graph.add_node("modeling", modeling_agent)
    graph.add_node("critic", evaluation_critic_agent)
    graph.add_node("explainability", explainability_agent)
    graph.add_node("executive_insights", executive_insights_agent)
    graph.add_edge(START, "data_quality")
    graph.add_edge("data_quality", "eda")
    graph.add_edge("eda", "statistics")
    graph.add_edge("statistics", "planning")
    graph.add_edge("planning", "feature_engineering")
    graph.add_edge("feature_engineering", "modeling")
    graph.add_edge("modeling", "critic")
    graph.add_conditional_edges(
        "critic",
        critic_route,
        {"retry_modeling": "modeling", "explainability": "explainability"},
    )
    graph.add_edge("explainability", "executive_insights")
    graph.add_edge("executive_insights", END)
    return graph.compile()


def run_analysis(

    frame: pd.DataFrame,

    target: str,

    dataset_name: str,

    settings: Settings | None = None,

) -> RunSummary:
    settings = settings or get_settings()
    run_id = f"run_{uuid4().hex[:12]}"
    store = RunStore(settings)
    artifact_store = ArtifactStore(settings.artifact_root)
    store.save(run_id, dataset_name, "running", {"run_id": run_id, "status": "running"})
    initial: AgentState = {
        "run_id": run_id,
        "dataset_name": dataset_name,
        "frame": frame,
        "target": target,
        "settings": settings,
        "trace": [],
    }
    try:
        final = build_graph().invoke(initial)
        payload = _summary_payload(final, run_id, dataset_name)
        final["summary_payload"] = payload
        log_to_mlflow(payload, settings)
        artifacts = export_artifacts(run_id, final, artifact_store.run_directory(run_id))
        payload["artifacts"] = artifacts
        payload["status"] = "completed"
        store.save(run_id, dataset_name, "completed", payload)
        return RunSummary.model_validate(payload)
    except Exception as exc:
        store.save(
            run_id,
            dataset_name,
            "failed",
            {"run_id": run_id, "dataset_name": dataset_name, "status": "failed", "error": str(exc)},
        )
        raise
    finally:
        store.engine.dispose()


def _summary_payload(state: AgentState, run_id: str, dataset_name: str) -> dict[str, Any]:
    bundle = state["model_bundle"]
    return {
        "run_id": run_id,
        "status": "completed",
        "dataset_name": dataset_name,
        "profile": state["profile"].model_dump(mode="json"),
        "plan": state["plan"].model_dump(mode="json"),
        "quality_issues": [item.model_dump(mode="json") for item in state["quality_issues"]],
        "evidence": [item.model_dump(mode="json") for item in state["evidence"]],
        "model_results": [item.model_dump(mode="json") for item in bundle.results],
        "model_failures": [item.model_dump(mode="json") for item in bundle.failures],
        "best_model": bundle.best_model,
        "critic": state["critic"].model_dump(mode="json"),
        "explainability": state["explainability"].model_dump(mode="json"),
        "executive_summary": state["executive_summary"],
        "recommendations": state["recommendations"],
        "artifacts": {},
        "trace": state["trace"],
    }