lineage neo4j langgraph mlflow observability fastapi

Agents d’enquête sur les lignées familiales sur Neo4j et LangGraph

Un agent d’enquête sur les lignées dans Neo4j et LangGraph qui trie, parcourt les arêtes FEEDS_INTO par locataire, vérifie les hypothèses et enregistre les audits dans MLflow.

Éditeur : REACTAPP.TOPAjouté sur ce site 26 sept. 202617 min de lecture3878 mots

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

Les données relatives aux incidents ne commencent rarement par le tableau de bord qui s’illumine en rouge. Elles proviennent d’étapes antérieures : des flux de données, des transformations et des jointures que personne ne parvient à décrire sous pression. Cette pile technologique transforme la traçabilité en un outil d’enquête : Neo4j gère les arêtes FEEDS_INTO, LangGraph suit une procédure de six nœuds (à préciser), FastAPI le sert, MLflow enregistre les exécutions, et Gradio fournit aux opérateurs une console.

Reposoir et aperçu

Le projet regroupe l’infrastructure en code, un graphe de traçabilité, un outil d’orchestration des enquêtes, une API HTTP, un suivi d’expériences et une interface utilisateur légère. Les enquêtes moyennes visent à être menées en moins d’une demi-minute, avec des chaînes d’hypothèses structurées évaluées à l’aide d’un ensemble d’incidents de référence.

Pile technologique

Terraform pour les ressources cloud, Neo4j pour la traçabilité des origines, LangGraph pour le flux de contrôle, FastAPI pour l’API, MLflow pour les paramètres/métriques/artefacts, Gradio pour l’interface utilisateur des opérateurs, ainsi qu’un réseau d’évaluation pour le scoring hors ligne.

Aperçu de l’architecture

Les alertes entrent dans la phase de triage ; les parcours suivent les chemins des ancêtres ; le profilage résume l’état de santé des nœuds ; on formule des hypothèses sur les causes ; on vérifie par des tests de sécurité ; les rapports génèrent un résumé structuré. On précise les interruptions en cas d’identité manquante.

Infrastructure as Code avec Terraform

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

Planifier avant d’appliquer ; connecter les sorties — adresses URL des graphes, noms de buckets, secrets de webhook — à la configuration en temps réel plutôt que d’insérer des valeurs issues de la console dans les notebooks.

Le modèle de données

Les clients, les ensembles de données, les tâches et les tables deviennent des nœuds ; les arêtes FEEDS_INTO indiquent la direction du flux des données. Les étiquettes multi-locataires empêchent que les informations d’un client ne s’infiltrent dans celles d’un autre.

Création du graphe de filiation dans Neo4j

Recherche des ancêtres pour déterminer la portée de l’impact :

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

Traversée ciblée avec des filtres de client :

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

L’agent d’enquête — six nœuds plus des outils de clarification

TRIAGE

Il classe le niveau de gravité et sélectionne un nœud d’entrée dans le graphe à partir du contenu de l’alerte.

TRAVERSE

Il étend la recherche vers les sources et les destinations dans les limites prévues, en respectant les filtres de locataire.

PROFILE

Il attribue des informations sur la fraîcheur, le volume et les signatures d’erreurs aux nœuds candidats.

HYPOTHESIZE

Hypothèses structurées avec des niveaux de priorité :

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

Outils pour identifier les schémas d’erreurs :

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

Logique des nœuds d’hypothèses :

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

Seules les formes lisible en toute sécurité :

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

Acheminement après vérification :

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

RAPPORT

Assemble un récit ainsi qu’une chaîne lisible par machine destinée aux utilisateurs et aux tickets.

CLARIFIER

Interrompt le processus lorsque l’identifiant du client ou de l’actif est ambigu — il vaut mieux faire une pause plutôt que de parcourir le mauvais tenant.

Assemblage du graphe

Connecte START → triage → parcours → profil → hypothèse → vérification → (boucle ou rapport) → FIN, avec CLARIFIER en tant que détour conditionnel. Compilez avec un point de contrôle afin que les enquêtes longues survivent aux redémarrages du processus.

Backend FastAPI

Expose des points de terminaison pour exécuter, reprendre et consulter l’état, qui correspondent à invoke / Command(resume=…). Authentifiez les opérateurs ; ne exposez jamais de credentials brutes de Bolt au navigateur.

Observabilité MLflow

Enregistrement des paramètres de tri au démarrage ; métriques relatives aux étapes parcourues, au nombre d’hypothèses testées, à la vérification des réussites/échecs et au temps d’exécution ; fichiers générés pour le rapport en Markdown, les données JSON des hypothèses et les transcriptions Cypher.

Tableau de bord Gradio

Fenêtres séparées pour la réception des alertes, l’état en temps réel et le rapport final. Les widgets de reprise fournissent des réponses claires dans le même identifiant de thread.

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)

Déclenchement à la fin de l’élaboration du rapport pour le transmettre au système de gestion des tickets, sans bloquer le chemin de retour du graphe ; les tentatives de réessai doivent avoir lieu dans le processus dédié aux webhooks, et non sur le nœud principal.

Conclusion

Un arbre de dépendances sans procédure d’enquête n’est qu’un graphe visuel. Une procédure sans arbre de dépendances relève du hasard. Ensemble — avec une infrastructure gérée par Terraform, des parcours sécurisés pour les utilisateurs, des vérifications limitées aux requêtes SELECT et un audit via MLflow — ils forment un outil opérationnel d’assistance en cas d’incident, et non simplement un cahier de démonstration.

Indiquer les responsables des décisions à côté des nœuds et des cibles des webhooks.

Préférer des requêtes de vérification utilisant uniquement SELECT dans les boucles non fiables.

Télécharger des snapshots des incidents critiques avant de modifier les messages de triage.

Émettre des alertes pour les boucles d’échec de vérification de la même manière que les services émettent des alertes pour les codes 5xx.

Séparer les clients et leurs tenants dans les étiquettes du graphe dès le premier jour.

Dokumenter les sorties de Terraform que consomme le runtime de l’agent.

Indiquer les responsables des décisions à côté des nœuds et des cibles webhook.

Préférer des requêtes de vérification utilisant uniquement SELECT dans les boucles non fiables.

Télécharger des snapshots des incidents critiques avant de modifier les messages de triage.

Émettre des alertes pour les boucles d’échec de vérification de la même manière que les services émettent des alertes pour les codes 5xx.

Séparer les clients et leurs tenants dans les étiquettes du graphe dès le premier jour.

Dokumenter les sorties de Terraform que consomme le runtime de l’agent.

Indiquer les responsables des décisions à côté des nœuds et des cibles webhook.

Préférer des requêtes de vérification utilisant uniquement SELECT dans les boucles non fiables.

Prendre une capture d’écran des incidents critiques avant de modifier les prompts de triage.

Émettre une alerte en cas de boucles d’échec de vérification de la même manière que les services le font pour les codes 5xx.

Séparer les clients et locataires dans les étiquettes du graphe dès le premier jour.

Dokumenter les sorties de Terraform que consomme le runtime de l’agent.

Indiquer les responsables des décisions à côté des nœuds et des cibles webhook.

Préférer des requêtes de vérification ne contenant que SELECT dans les boucles non fiables.

Prendre une capture d’écran des incidents critiques avant de modifier les prompts de triage.

Émettre une alerte en cas de boucles d’échec de vérification de la même manière que les services le font pour les codes 5xx.

Séparer les clients et locataires dans les étiquettes du graphe dès le premier jour.

Dokumenter les sorties de Terraform que consomme le runtime de l’agent.

Indiquer les responsables des décisions à côté des nœuds et des cibles webhook.

Préférer des requêtes de vérification ne contenant que SELECT dans les boucles non fiables.

Prendre une capture d’écran des incidents critiques avant de modifier les prompts de triage.

Alertez en cas de boucles d’échec de vérification de la même manière que les services le font pour les codes 5xx.

Séparez les clients-locataires dans les étiquettes du graphe dès le premier jour.

Dokumentez les sorties de Terraform que consomme le runtime de l’agent.

Indiquez les responsables des décisions à côté des nœuds et des cibles webhook.

Préférez les requêtes de vérification utilisant uniquement SELECT dans les boucles non fiables.

Tenez une capture d’écran des incidents critiques avant de modifier les prompts de triage.

Alertez en cas de boucles d’échec de vérification de la même manière que les services le font pour les codes 5xx.

Séparez les clients-locataires dans les étiquettes du graphe dès le premier jour.

Dokumentez les sorties de Terraform que consomme le runtime de l’agent.

Indiquez les responsables des décisions à côté des nœuds et des cibles webhook.

Préférez les requêtes de vérification utilisant uniquement SELECT dans les boucles non fiables.

Tenez une capture d’écran des incidents critiques avant de modifier les prompts de triage.

Alertez en cas de boucles d’échec de vérification de la même manière que les services le font pour les codes 5xx.

Séparez les clients et locataires dans les étiquettes du graphe dès le premier jour.

Dokumentez les sorties de Terraform que consomme le runtime de l’agent.

Indiquez les responsables des décisions à côté des nœuds et des cibles webhook.

Préférez les requêtes de vérification utilisant uniquement SELECT dans les boucles non fiables.

Tenez une capture d’écran des incidents critiques avant de modifier les prompts de triage.

Avertissez en cas de boucles d’échec de vérification de la même manière que les services avertissent en cas de codes 5xx.

Séparez les clients et locataires dans les étiquettes du graphe dès le premier jour.

Dokumentez les sorties de Terraform que consomme le runtime de l’agent.

Gardez les responsables, les paramètres fixes et les notes de rollback à côté de chaque modification de graphe ou d’agent afin que les démonstrations ne devancent pas la faisabilité opérationnelle.

Conservez les responsables, les paramètres et les notes de réversion à côté de chaque graphique ou modification d’agent afin que les démonstrations ne devancent pas la faisabilité opérationnelle.

← Tous les articles Envoyer le brief