lineage neo4j langgraph mlflow observability fastapi

Abstammungssuche-Tools auf Neo4j und LangGraph

Ein Agent für Abstammungsuntersuchungen in Neo4j und LangGraph, der die FEEDS_INTO-Kanten pro Mieter sortiert und durchsucht, Hypothesen überprüft und Audits in MLflow protokolliert.

Verlag: REACTAPP.TOPAuf dieser Website hinzugefügt 26. Sept. 202617 Min. Lesezeit3878 Wörter

Quellen: Production-Grade AI-Powered Data Lineage Observability Agent with LangGraph, Airflow, DBT, Neo4j, MLflow, Gradio, GEval

Datenvorfälle beginnen selten am Anzeigefeld, das rot aufleuchtet. Sie entstehen weiter oben in der Kette – bei Eingabedaten, Transformationen und Verknüpfungen, die unter Druck niemand richtig beschreiben kann. Diese Technologiekombination verwandelt die Abhängigkeitsverfolgung in ein Ermittlungstool: Neo4j speichert die Verbindungen FEEDS_INTO, LangGraph führt einen Prozess mit sechs Knoten durch, FastAPI stellt die Daten bereit, MLflow protokolliert die Ausführungen und Gradio bietet den Bedienern eine Konsole.

Repository und Übersicht

Das Projekt enthält Infrastructure-as-Code, ein Abhängigkeitsgraphen, einen Ermittlungsorchestrator, eine HTTP-API, Funktion für das Tracking von Experimenten sowie eine leichte Benutzeroberfläche. Ziel ist es, dass mittlere Ermittlungen in weniger als einer halben Minute abgeschlossen werden, wobei strukturierte Hypothesenketten anhand eines standardisierten Vorgehens bewertet werden.

Technologiestack

Terraform für Cloud-Ressourcen, Neo4j für die Herkunftsverfolgung, LangGraph für den Kontrollfluss, FastAPI für die API, MLflow für Parameter/Metriken/Artefakte, Gradio für die Benutzeroberfläche des Operators sowie ein Bewertungsnetzwerk für die Offline-Bewertung.

Überblick über die Architektur

Alarme werden zur Triage eingereicht; die Durchquerung prüft Vorfahrenpfade; das Profiling fasst den Zustand der Knoten zusammen; Hypothesen ordnen mögliche Ursachen zu; Sicherheitsprüfungen werden durchgeführt; Berichte liefern eine strukturierte Zusammenfassung. Bei fehlender Identität werden Unterbrechungen geklärt.

Infrastruktur als Code mit Terraform

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

Planen Sie vor der Anwendung; verbinden Sie Ausgaben – Graph-Bolt-URLs, Bucket-Names, Webhook-Secrets – mit der Laufzeitkonfiguration anstelle dessen, dass Sie Konsoleingaben in Notebooks einfügen.

Das Datenmodell

Kunden, Datensätze, Aufgaben und Tabellen werden zu Knoten; die Kanten FEEDS_INTO geben die Richtung der Datenbewegung an. Mehrfachnutzer-Labels verhindern, dass die Auswertungen eines Kunden in die eines anderen durchdringen.

Aufbau des Lineage-Graphs in Neo4j

Vorfahren-Auswertungen zur Bestimmung des Ausbreitungsradius:

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

Begrenzte Durchsuchung mit Kundenfiltern:

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

Der Ermittlungsagent – sechs Knoten plus Klärungsfunktion

TRIAGE

Klassifiziert die Schweregrad und wählt aus dem Alarminhalt einen Eingangsknoten im Graphen aus.

TRAVERSE

Erweitert die Suche in aufwärts und abwärts gerichteten Richtungen innerhalb der definierten Schritte, wobei die Nutzerfilter berücksichtigt werden.

PROFILE

Fügt potenziellen Knoten Informationen zu Aktualität, Volumen und Fehlermustern hinzu.

HYPOTHESIZE

Gestrukturierte Hypothesen mit Rangfolgen:

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

Hilfsmittel für Fehlermuster:

_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}

Logik der Hypothesenknoten:

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}

VERIFY

Nur sichere Lesformen:

_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,
      }

Routing nach Überprüfung:

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()

BERICHT

Erstellt eine für Menschen verständliche Beschreibung sowie eine maschinenlesbare Kette für Tickets.

KLÄRUNG

Unterbricht den Ablauf, wenn die Kunden-ID oder Asset-ID unklar ist – es ist besser, pausieren zu lassen, als den falschen Benutzerbereich zu durchlaufen.

Graphen-Aufbau

Vernetze START → Triage → Durchlaufen → Profilierung → Hypothesenbildung → Überprüfung → (Schleife oder Bericht) → END, wobei die KLÄRUNG als bedingte Umleitung fungiert. Verwende einen Checkpointer, damit lange Untersuchungen auch nach Prozessneustarts erhalten bleiben.

FastAPI-Backend

Biete Endpunkte für Ausführen, Wiederaufnehmen und Statusanzeige an, die auf invoke / Command(resume=…) verweisen. Authentifiziere die Benutzer; gib niemals unverschlüsselte Bolt-Anmeldeinformationen an den Browser weiter.

MLflow-Überwachbarkeit

Protokollieren Sie zu Beginn die Triage-Parameter; Metriken für die durchlaufenen Schritte, gezählte Hypothesen, Überprüfung von Erfolgen/Fehlern sowie Laufzeit; Artefakte für den Markdown-Bericht, Hypothesen im JSON-Format sowie Transkripte der Cypher-Ausführungen.

Gradio-Dashboard

Trennen Sie die Fenster für die Eingabe von Alarmen, den aktuellen Zustand und den endgültigen Bericht. Die Wiederbelebungs-Widgets liefern klare Antworten zurück zum gleichen Thread-ID.

Webhooks

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)

Lösen Sie bei Abschluss des Berichts Aktionen im Ticketing-System aus, ohne den Rückweg des Graphen zu blockieren; Wiederholungsversuche sollten im Webhook-Worker stattfinden, nicht im Hot Node.

Fazit

Eine Abfolge von Ereignissen ohne Untersuchungsverfahren ist nur ein schöner Graph. Ein Verfahren ohne Abfolge ist reine Spekulation. In Kombination mit infrastrukturbezogenen Terraform-Strukturen, sichereren Durchläufen für Mieter, Überprüfungen ausschließlich mit SELECT-Anfragen sowie MLflow-Audits entsteht so ein funktionsfähiger Hilfsmechanismus für Incidents statt nur ein Demo-Notebook.

Legen Sie die Entscheidungsträger neben die Knoten sowie die Webhook-Ziele fest.

Verwenden Sie in unsicheren Schleifen lieber Verifizierungsabfragen, die nur SELECT verwenden.

Nehmen Sie vor der Änderung der Triage-Anfragen Snapshots der kritischen Vorfälle auf.

Geben Sie bei Fehlern in den Verifizierungsabfragen Alarm, genauso wie Dienste bei 5xx-Fehlern Alarm geben.

Trennen Sie die Client-Mieter bereits ab dem ersten Tag durch Graphen-Labels voneinander.

Dokumentieren Sie die von der Agenten-Laufzeit genutzten Terraform-Ausgaben.