lineage neo4j langgraph mlflow observability fastapi

Agentes de investigación de linajes en Neo4j y LangGraph

Un agente de investigación de linajes en Neo4j y LangGraph que clasifica, recorre los bordes FEEDS_INTO por arrendatario, verifica hipótesis y registra auditorías en MLflow.

Editor: REACTAPP.TOPAñadido a este sitio 26 sept 202617 min de lectura3878 palabras

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

Los incidentes de datos rara vez comienzan en el panel de control que se ilumina en rojo. Comienzan más arriba: fuentes, transformaciones y uniones que nadie puede describir bajo presión. Esta estructura convierte el seguimiento de la cadena de datos en un agente de investigación: Neo4j almacena las conexiones FEEDS_INTO, LangGraph sigue un procedimiento de seis nodos (más aclaraciones), FastAPI lo sirve, MLflow registra las ejecuciones y Gradio proporciona una consola a los operadores.

Repositorio y visión general

El proyecto incluye infraestructura como código, un grafo de seguimiento de datos, un orquestador de investigaciones, una API HTTP, seguimiento de experimentos y una interfaz de usuario ligera. Las investigaciones promedio tienen como objetivo completarse en menos de medio minuto, con cadenas de hipótesis estructuradas evaluadas según un conjunto estándar de incidentes.

Stack tecnológica

Terraform para recursos en la nube, Neo4j para el seguimiento de origen, LangGraph para el flujo de control, FastAPI para la API, MLflow para parámetros/métricas/artefactos, Gradio para la experiencia de usuario del operador, y una red de jueces para la puntuación sin conexión.

Visión general de la arquitectura

Las alertas ingresan al proceso de triaje; el recorrido sigue los caminos de los ancestros; el perfilamiento resume el estado de los nodos; se hipotetizan las causas; se verifican pruebas de seguridad; el informe emite un resumen estructurado. Se aclaran las interrupciones cuando falta la identidad.

Infraestructura como código con Terraform

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

Planifique antes de aplicar; conecte las salidas — URLs de Bolt del grafo, nombres de buckets, secretos de webhook — a la configuración en tiempo de ejecución en lugar de pegar valores de la consola en los cuadernos de notas.

El modelo de datos

Los clientes, conjuntos de datos, tareas y tablas se convierten en nodos; los bordes FEEDS_INTO indican la dirección del movimiento de los datos. Las etiquetas de multiarrendatario evitan que el recorrido de un cliente se filtre al de otro.

Construcción del grafo de linaje en Neo4j

Recorridos de ancestros para determinar el radio de impacto:

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

Recorrido limitado con filtros de cliente:

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

El agente de investigación: seis nodos más herramientas de aclaración

TRIAGE

Clasifica la gravedad y selecciona un nodo de entrada en el grafo a partir del contenido de la alerta.

TRAVERSE

Expande los enlaces hacia arriba y hacia abajo dentro de los límites establecidos, respetando los filtros de arrendatario.

PROFILE

Añade información sobre la actualidad, el volumen y las firmas de errores a los nodos candidatos.

HYPOTHESIZE

Hipótesis estructuradas con rangos de prioridad:

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

Herramientas para identificar patrones de errores:

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

Lógica de los nodos de hipótesis:

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

Solo formas de lectura seguras:

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

Ruteo después de la verificación:

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

INFORME

Reúne una narrativa además de una cadena legible por máquina para los usuarios y los tickets.

ESCLARIFICAR

Interrumpe cuando el ID del cliente o el ID del activo es ambiguo; es mejor pausar que acceder al tenant incorrecto.

Montaje del grafo

Conecta START → triage → traverse → profile → hypothesize → verify → (bucle o informe) → END, con ESCLARIFICAR como desvío condicional. Compila con un checkpointer para que las investigaciones largas sobrevivan a los reinicios del proceso.

Backend FastAPI

Expone endpoints para ejecutar, reanudar y consultar el estado que se corresponden con invoke / Command(resume=…). Autentica a los operadores; nunca exponga credenciales brutas de Bolt al navegador.

Observabilidad con MLflow

Registre los parámetros de triaje al inicio; métricas sobre el número de pasos realizados, hipótesis contadas, verificación de éxitos/fracasos y tiempo de ejecución; archivos para el informe en formato Markdown, JSON con las hipótesis y transcripciones de Cypher.

Panel de control de Gradio

Divida las ventanas para la recepción de alertas, el estado en tiempo real y el informe final. Los widgets de reanudación envían respuestas claras al mismo identificador de hilo.

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)

Active los webhooks al finalizar el informe para enviarlo a la plataforma de tickets sin bloquear la ruta de retorno del gráfico; las reintentos deben realizarse en el proceso del webhook, no en el nodo principal.

Cierre

Un árbol de relaciones sin un procedimiento de investigación es solo un gráfico. El procedimiento sin ese árbol de relaciones no es más que conjetura. Juntos, junto con la infraestructura gestionada por Terraform, recorridos seguros para los usuarios, verificaciones mediante consultas SELECT únicamente y auditorías con MLflow, se convierten en una herramienta operativa para gestionar incidentes en lugar de un simple cuaderno de ejemplos.

Registre a los responsables de las decisiones junto a los nodos y los destinos de los webhooks.

Prefiera consultas de verificación que utilicen únicamente SELECT en bucles no confiables.

Tome instantáneas de los incidentes críticos antes de cambiar los mensajes de triaje.

Genere alertas para los bucles de verificación fallida de la misma manera que lo hacen los servicios ante códigos 5xx.

Separe a los clientes arrendatarios en las etiquetas del gráfico desde el primer día.

Documente las salidas de Terraform que consume el entorno en tiempo de ejecución del agente.

Registre a los responsables de las decisiones junto a los nodos y destinos de webhook.

Prefiera consultas de verificación que utilicen únicamente SELECT en bucles no confiables.

Tome instantáneas de los incidentes críticos antes de cambiar los mensajes de triaje.

Genere alertas para los bucles de verificación fallida de la misma manera que lo hacen los servicios ante códigos 5xx.

Separe a los clientes arrendatarios en las etiquetas del gráfico desde el primer día.

Documente las salidas de Terraform que consume el entorno en tiempo de ejecución del agente.

Registre a los responsables de las decisiones junto a los nodos y destinos de webhook.

Prefiera consultas de verificación que utilicen únicamente SELECT en bucles no confiables.

Tomar una instantánea de los incidentes críticos antes de cambiar los mensajes de triaje.

Generar alertas por bucles de verificación fallida de la misma manera que lo hacen las alertas para códigos 5xx.

Separar a los clientes arrendatarios en las etiquetas del gráfico desde el primer día.

Dокументar los resultados de Terraform que consume el entorno en tiempo real del agente.

Registrar a los responsables de las decisiones junto a los nodos y destinos de webhook.

Preferir consultas de verificación que utilicen únicamente SELECT en bucles no confiables.

Tomar una instantánea de los incidentes críticos antes de cambiar los mensajes de triaje.

Generar alertas por bucles de verificación fallida de la misma manera que lo hacen las alertas para códigos 5xx.

Separar a los clientes arrendatarios en las etiquetas del gráfico desde el primer día.

Dокументar los resultados de Terraform que consume el entorno en tiempo real del agente.

Registrar a los responsables de las decisiones junto a los nodos y destinos de webhook.

Preferir consultas de verificación que utilicen únicamente SELECT en bucles no confiables.

Tomar una instantánea de los incidentes críticos antes de cambiar los mensajes de triaje.

Se deben generar alertas por bucles de verificación fallida de la misma manera que lo hacen los servicios ante códigos 5xx.

Se deben separar los clientes arrendatarios en las etiquetas del gráfico desde el primer día.

Se deben documentar las salidas de Terraform que consume el entorno en tiempo real del agente.

Se deben registrar los responsables de las decisiones junto a los nodos y destinos de webhook.

En bucles no confiables, se debe preferir el uso de consultas de verificación que solo incluyan SELECT.

Se deben tomar instantáneas de los incidentes críticos antes de modificar los mensajes de triaje.

Se deben generar alertas por bucles de verificación fallida de la misma manera que lo hacen los servicios ante códigos 5xx.

Se deben separar los clientes arrendatarios en las etiquetas del gráfico desde el primer día.

Se deben documentar las salidas de Terraform que consume el entorno en tiempo real del agente.

Se deben registrar los responsables de las decisiones junto a los nodos y destinos de webhook.

En bucles no confiables, se debe preferir el uso de consultas de verificación que solo incluyan SELECT.

Se deben tomar instantáneas de los incidentes críticos antes de modificar los mensajes de triaje.

Se deben generar alertas por bucles de verificación fallida de la misma manera que lo hacen los servicios ante códigos 5xx.

Separar a los clientes en las etiquetas del gráfico desde el primer día.

Documentar las salidas de Terraform que consume el entorno en tiempo real del agente.

Registrar a los responsables de las decisiones junto a los nodos y destinos de los webhook.

Preferir consultas de verificación que utilicen solo SELECT en bucles no confiables.

Tomar instantáneas de los incidentes críticos antes de cambiar los mensajes de triaje.

Generar alertas en caso de bucles de verificación fallida de la misma manera que lo hacen las alertas por códigos 5xx.

Separar a los clientes en las etiquetas del gráfico desde el primer día.

Documentar las salidas de Terraform que consume el entorno en tiempo real del agente.

Mantener los responsables, configuraciones fijas y notas de reversión junto a cada cambio en el gráfico o el agente para que las demostraciones no superen la capacidad operativa.

Mantenga los responsables, las configuraciones y las notas de reversión junto a cada gráfico o cambio de agente para que las demostraciones no superen la capacidad operativa.

Lecturas relacionadas

Los asistentes de investigación empresarial necesitan EA, LangGraph, MCP y RAG — Las investigaciones multidestino unen tickets, registros y libros contables bajo una gobernanza definida; las herramientas demuestran su eficacia, RAG la explica y una arquitectura gradual supera a las demostraciones con chatbots.

Reforzamiento de los gráficos agentes: contratos de estado, aislamiento y puertas HITL — Fiabilidad en producción para LangGraph: estado plano de TypedDict, nodos de uso exclusivo, aristas de reintentos y errores, revisión humana y capacidades de observabilidad.

← Todos los artículos Enviar el brief