lineage neo4j langgraph mlflow observability fastapi

Агенты для аналізу генеалогічных лямоў у Neo4j і LangGraph

Агент для аналізу генеалагіі ў Neo4j і LangGraph, які сортавае, пераходзіць па рэгламентах FEEDS_INTO па кожнам ужыткавальніку, перакананае гіпатэзы і фіксуе аудыты ў MLflow.

Выдавец: REACTAPP.TOPДададзена на сайт 26 вер 202617 хв чытання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

Аб’явіце канцэнты run, resume і status, якія адпавядаюць invoke / Command(resume=…). Автентыфікуйце аператараў; ніколі не аб’являйце сырых крантэнаў Bolt у браузеры.

Візуабельнасць MLflow

Запісваюцься параметры сортавання логаў на пачатку; паказываюцца метрыкі: колькість пройдзеных етапаў, колькість гіпотэз, падтверджэння успеху/неспяшчання, час выконання; таксонамі для дакумента ў формате Markdown, JSON з гіпотэзамі і транскрыпцыі сухопутных перадач.

Панель керування 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 — яны стаюць функцыйнальным адпамогчыкам у разв’язванні інцыдэтаў, а не проста демонстрацыйным нотбукам.

Запісваюцца імены адпавядачаў за рашэнні празае ўзлам і цілями webhook.

У недазвереных цыклах лепшае выкарыстоўваць толькі запыты верыфікацыі типу SELECT.

Зробіце копію ключовых інцидэнаў прычымо да змены прапанаў трыажу.

Падаюць апавешчэння пра цыклы з невялікай верыфікацыёй так сама, як і праблемы типу 5xx.

З самага пачатку раздзеляйце кліентаў-аренднікаў у атрыбутах графа.