lineage neo4j langgraph mlflow observability fastapi

Агенти з розслідування походження в Neo4j та LangGraph

Агент для дослідження ліній спадковості в Neo4j та LangGraph, який сортує, аналізує ребра FEEDS_INTO за кожним користувачем, перевіряє гіпотези та записує звіти про аудит у MLflow.

Видавець: REACTAPP.TOPДодано на сайт 26 вер. 2026 р.17 хв читання3878 слів

Джерела: Production-Grade AI-Powered Data Lineage Observability Agent with LangGraph, Airflow, DBT, Neo4j, MLflow, Gradio, GEval

Інциденти з даними рідко починаються на панелі керування, яка світиться червоним кольором. Вони виникають на ранніх етапах — у процесах надходження, перетворення та об’єднання даних, які ніхто не може описати під тиском. Ця стек-архітектура перетворює ланцюжок операцій з даними на інструмент для розслідувань: Neo4j зберігає краї типу FEEDS_INTO, LangGraph виконує процедуру з шістьма вузлами (та додатковими уточненнями), FastAPI надає доступ до цих даних, MLflow фіксує результати виконання, а Gradio забезпечує операторів консоллю.

Репозиторій та огляд

Проєкт включає інфраструктуру у вигляді коду, граф ланцюжка операцій з даними, інструмент для організації розслідувань, HTTP API, систему відстеження експериментів та легкий інтерфейс користувача. Середній час розслідування становить менше півхвилини завдяки структурованим ланцюжкам гіпотез, оціненим за стандартом „золотого набору інцидентів“.

Технологічна стек-архітектура

Terraform для хмарних ресурсів, Neo4j для визначення походження даних, LangGraph для керування потоком виконання, FastAPI для API, MLflow для параметрів/метрик/артефактів, Gradio для користувацького інтерфейсу оператора, а також мережа суддів для оцінки в автономному режимі.

Огляд архітектури

Попередження надходять на етап сортування; алгоритм просування відстежує шляхи до предків; профілювання підсумовує стан вузлів; формуються гіпотези щодо причин; виконуються перевірки на безпеку; формується структурований звіт. При відсутності ідентифікатора уточнюються перерви.

Інфраструктура як код за допомогою Terraform

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

Спочатку розробляється план, потім він застосовується; результати обробки — URL-адреси графу, назви бакетів, секретні дані webhook — вбудовуються до конфігурації робочого середовища, а не копіюються з консолі до ноутбуків.

Модель даних

Клієнти, набори даних, завдання та таблиці стають вузлами; ребра FEEDS_INTO відображають напрямок руху даних. Мітки багатокористувацького доступу запобігають проникненню даних одного клієнта в систему іншого.

Створення графа лінійності в Neo4j

Процес пошуку предків для визначення зони ураження:

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

Обмежений перебір з фільтрами клієнтів:

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

Агент розслідування — шість вузлів та додаткові елементи

TRIAGE

Класифікує ступінь серйозності та вибирає вузол у графі з контенту сповіщення.

TRAVERSE

Розширює пошук у напрямку вгору/вниз в межах встановлених обмежень, дотримуючись фільтрів користувачів.

PROFILE

Додає до потенційних вузлів інформацію про свіжість, обсяг та ознаки помилок.

HYPOTHESIZE

Структуровані гіпотези з рангами:

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

Інструменти для аналізу шаблонів помилок:

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

Логіка вузлів гіпотез:

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

Лише безпечні формати для читання:

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

Маршрутизація після перевірки:

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

ЗВІТ

Створює наратив та ланцюг даних, зрозумілий для людей та системи обробки замовлень.

УЗГОДЖЕННЯ

Перериває процес, якщо ідентифікатор клієнта чи активу є неоднозначним — краще зупинитися, ніж працювати з неправильним об’єктом.

Складання графа

Підключення етапів: START → триаж → просування → профілювання → формулювання гіпотези → перевірка → (цикл або звіт) → END, з можливістю узгодження як умовного відхилення. Компілюється з механізмом контролю, щоб тривалі розслідування залишалися актуальними після перезапуску процесу.

Бекенд FastAPI

Надає ендпоїнти для запуску, відновлення та отримання статусу, які відповідають invoke / Command(resume=…). Аутентифікує операторів; ніколи не відкривати сирі облікові дані Bolt для браузера.

Оперативність MLflow

Записуйте параметри сортування журналу на початку; метрики щодо кількості пройдених етапів, кількості гіпотез, перевірка успіху/невдачі, час виконання; матеріали для звіту у форматі Markdown, JSON із гіпотезами та транскрипції даних у форматі Cypher.

Панель керування Gradio

Розділіть вікна для отримання сповіщень, перегляду поточного стану та остаточного звіту. Віджети для продовження роботи надають уточнені відповіді у той самий ідентифікатор потоку.

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)

Запускайте процес після завершення звіту у системі обробки тикетів, не блокуючи шлях повернення даних від графа; спроби повторного запуску мають відбуватися у роботі webhook, а не на основному вузлі.

Заключення

Лінійна структура без процедури розслідування — це просто граф. Процедура без лінійної структури — це припущення. Разом із інфраструктурою, створеною за допомогою Terraform, безпечними процесами обробки даних, перевірками лише за допомогою SELECT та аудитом у форматі MLflow вони стають ефективним інструментом для роботи з інцидентами, а не просто демонстраційним нотбуком.