Microservice: Внутренний аналитический сервис

pm-analytics

WarningОграничение публичной документации

В открытом доступе представлена демонстрационная версия метода. В настоящей публичной документации отображены не все шаги, технические сценарии и приватные эндпоинты для системы цифровых симуляторов бизнес-процессов.

• Полная спецификация метода: Будет доступна только во внутреннем контуре разработки (Confluence / Swagger Enterprise).

1 Общее назначение и архитектурная роль сервиса

Микросервис python-pm4py-worker — автономное математическое ядро на базе FastAPI / gRPC (grpcio). У сервиса отсутствуют публичные HTTP-порты: он изолирован во внутренней сети Docker и принимает вызовы исключительно от go-pm-backend по протоколу gRPC.

Архитектурная изоляция:

• Без состояния (Stateless & No Auth): не валидирует JWT, не ходит в MinIO и PostgreSQL.
• Прямой контур ClickHouse: выполняет оптимизированные аналитические выборки из logs.process_events с агрегацией по case_id.
• Математический расчёт: парсит XML-схему BPMN, конвертирует в сеть Петри и запускает Alignment-Based Conformance / Object-Centric Process Mining (OCPM) для классификации отклонений.

2 Основные бизнес-функции

1. AnalyzeConformance (gRPC Handler): прием XML-схемы (bpmn_xml_content), фильтров сценария и порога шума.
2. FetchAndAggregateClickHouseTraces: выполнение оптимизированного SQL-запроса с оконными функциями для извлечения упорядоченных трасс.
3. BPMNtoPetriNetCompiler: компиляция сырого XML в объект сети Петри с кэшированием в памяти воркера по хэшу схемы (SHA256).
4. AlignmentConformanceEngine: расчет оптимального выравнивания (Optimal Alignment) между реальной трассой и моделью с извлечением метрик c, p, m, r и классификацией аномалий.
5. OCPMObjectCentricProjection: поддержка объектно-центрического представления (привязка событий не к монолитному case_id, а к мульти-объектам order_id, operator_id, item_id).

3 Структура базы данных и хранилищ (Реестр СУБД)

Table 1: : Сводный реестр используемых СУБД, хранилищ данных и режимов доступа воркера

Компонент / СУБД	Логическая роль в сервисе	Схема / Таблица	Режим доступа из Python
ClickHouse (logs)	Источник сырых событий процесса	logs.process_events	Чтение через clickhouse-connect (HTTP/TCP, асинхронные батчи)
Локальная память (RAM)	Кэш скомпилированных сетей Петри	dict[str, tuple(net, im, fm)]	In-memory LRU-кэш по хэшу SHA256(bpmn_xml_content)
MinIO / PostgreSQL / Kafka	Не используются	—	Полная изоляция от внешних транзакционных контуров

4 Оптимизированный SQL-запрос выборки трасс из ClickHouse

Запрос извлекает упакованные массивы активностей и временных меток, минимизируя сетевой оверхед и ускоряя построение DataFrame в Python:

SELECT
    case_id,
    groupArray(activity) AS activities,
    groupArray(timestamp) AS timestamps,
    groupArray(operator_id) AS operators,
    groupArray(attributes_json) AS attributes
FROM (
    SELECT
        case_id,
        activity,
        timestamp,
        operator_id,
        attributes_json
    FROM logs.process_events
    WHERE timestamp >= now() - INTERVAL 30 DAY
      -- Опциональная фильтрация по сценарию через предикат в attributes_json
    ORDER BY case_id ASC, timestamp ASC
)
GROUP BY case_id
HAVING length(activities) >= 2;

5 Реализация математического ядра на Python (PM4Py + Alignments)

Listing 1: : Python-реализация воркера с кэшированием сетей Петри, вычиткой ClickHouse и расчетом Alignment Conformance

import hashlib
import json
import logging
from typing import Dict, List, Tuple
import xml.etree.ElementTree as ET

import clickhouse_connect
import pandas as pd
import pm4py
from pm4py.algo.conformance.alignments.petri_net import algorithm as alignments
from pm4py.objects.conversion.bpmn import converter as bpmn_converter
from pm4py.objects.petri_net.obj import Marking, PetriNet

logger = logging.getLogger("pm4py_worker")
_petri_cache: Dict[str, Tuple[PetriNet, Marking, Marking]] = {}


class ProcessMiningWorkerService:

  def __init__(self, ch_host: str = "clickhouse", ch_port: int = 8123):
    self.ch_client = clickhouse_connect.get_client(
        host=ch_host, port=ch_port, database="logs"
    )

  def _compile_bpmn_to_petri(
      self, bpmn_xml: str
  ) -> Tuple[PetriNet, Marking, Marking]:
    schema_hash = hashlib.sha256(bpmn_xml.encode("utf-8")).hexdigest()
    if schema_hash in _petri_cache:
      return _petri_cache[schema_hash]

    # Парсинг BPMN XML во временный файл / буфер через PM4Py
    # В реальном контуре используется временный файл или прямой импорт
    bpmn_graph = pm4py.read_bpmn_string(bpmn_xml)
    net, initial_marking, final_marking = (
        bpmn_converter.apply(  # type: ignore
            bpmn_graph, variant=bpmn_converter.Variants.BPMN_TO_PETRI_NET
        )
    )

    _petri_cache[schema_hash] = (net, initial_marking, final_marking)
    return net, initial_marking, final_marking

  def fetch_traces_from_ch(
      self, scenario_id: str, days_back: int = 30
  ) -> pd.DataFrame:
    query = f"""
        SELECT
            case_id,
            groupArray(activity) AS activities,
            groupArray(timestamp) AS timestamps,
            groupArray(operator_id) AS operators
        FROM (
            SELECT case_id, activity, timestamp, operator_id
            FROM logs.process_events
            WHERE timestamp >= now() - INTERVAL {days_back} DAY
            ORDER BY case_id ASC, timestamp ASC
        )
        GROUP BY case_id
        HAVING length(activities) >= 2
        LIMIT 5000;
        """
    result = self.ch_client.query(query)
    # Распаковка массивов в плоский DataFrame для PM4Py
    rows = []
    for row in result.result_rows:
      case_id, acts, tss, ops = row[0], row[1], row[2], row[3]
      for act, ts, op in zip(acts, tss, ops):
        rows.append({
            "case:concept:name": str(case_id),
            "concept:name": str(act),
            "time:timestamp": pd.to_datetime(ts),
            "org:resource": str(op),
        })
    return pd.DataFrame(rows)

  def compute_alignments_and_classify(
      self,
      df_events: pd.DataFrame,
      net: PetriNet,
      im: Marking,
      fm: Marking,
      noise_threshold: float = 0.2,
  ) -> dict:
    if df_events.empty:
      return {
          "global_fitness_score": 1.0,
          "total_cases_analyzed": 0,
          "health_status": "HEALTHY",
          "detailed_cases": [],
      }

    # Преобразование в Event Log PM4Py
    event_log = pm4py.format_dataframe(
        df_events,
        case_id="case:concept:name",
        activity_key="concept:name",
        timestamp_key="time:timestamp",
    )

    # Расчет Alignment-based Conformance
    aligned_traces = alignments.apply_log(event_log, net, im, fm)

    detailed_cases = []
    fit_scores = []

    for trace_idx, aligned_trace in enumerate(aligned_traces):
      # Анализ cost и токенов из выравнивания
      fitness = aligned_trace.get("fitness", 0.0)
      fit_scores.append(fitness)

      alignment_path = aligned_trace.get("alignment", [])
      # Подсчет consumed (c), produced (p), missing (m), remaining (r)
      c_cnt, p_cnt, m_cnt, r_cnt = 0, 0, 0, 0
      anomalies = []
      classification = "COMPLIANT"

      for step in alignment_path:
        model_move, log_move = step[0]
        if model_move != ">>" and log_move != ">>":
          c_cnt += 1
          p_cnt += 1
        elif model_move != ">>" and log_move == ">>":
          m_cnt += 1
          anomalies.append({
              "type": "MISSING_TOKEN_STATE_MACHINE",
              "severity": "CRITICAL",
              "message": f"Пропущен шаг в логе: {model_move}",
          })
          classification = "CRITICAL_BACKEND_BUG"
        elif model_move == ">>" and log_move != ">>":
          r_cnt += 1
          anomalies.append({
              "type": "HIDDEN_API_ROUTE",
              "severity": "LOW_WARNING",
              "message": f"Внеплановый шаг в логе (обход BPMN): {log_move}",
          })
          if classification == "COMPLIANT":
            classification = "UNDOCUMENTED_API_FEATURE"

      case_id_val = (
          event_log[trace_idx].attributes.get(
              "concept:name", f"case_{trace_idx}"
          )
          if trace_idx < len(event_log)
          else f"case_{trace_idx}"
      )

      detailed_cases.append({
          "case_id": str(case_id_val),
          "fitness_score": round(float(fitness), 4),
          "is_compliant": fitness >= (1.0 - noise_threshold),
          "metrics": {
              "consumed": c_cnt,
              "produced": p_cnt,
              "missing": m_cnt,
              "remaining": r_cnt,
          },
          "classification": classification,
          "anomalies": anomalies,
      })

    global_fitness = sum(fit_scores) / len(fit_scores) if fit_scores else 1.0
    health_status = (
        "HEALTHY" if global_fitness >= (1.0 - noise_threshold) else "ATTENTION_REQUIRED"
    )

    return {
        "global_fitness_score": round(global_fitness, 4),
        "total_cases_analyzed": len(detailed_cases),
        "health_status": health_status,
        "detailed_cases": detailed_cases,
    }

6 Объектно-центрический подход (OCPM Projection)

В отличие от классического Process Mining (где один case_id привязан к одному документу), Object-Centric Process Mining (OCPM) в воркере строит граф взаимодействий между независимыми объектами: Order (заказ), ReceiptImage (чек), Operator (пользователь) и AIGatewayRequest (вызов Whisper/Qwen). Пример SQL-извлечения OCPM-событий из ClickHouse:

SELECT
    case_id AS event_id,
    activity,
    timestamp,
    operator_id,
    JSONExtractString(attributes_json, 'order_id') AS obj_order_id,
    JSONExtractString(attributes_json, 'receipt_hash') AS obj_receipt_hash
FROM logs.process_events
WHERE timestamp >= now() - INTERVAL 7 DAY
  AND JSONExtractString(attributes_json, 'order_id') != '';

PM4Py строит объектно-центрическую модель (OCEL — Object-Centric Event Log), предотвращая «клонирование кейсов» при параллельной обработке нескольких чеков одним оператором.

7 Диаграмма последовательности (python-pm4py-worker)

%%{init: {
  'theme': 'base',
  'themeVariables': {
    'actorBkg': '#EBF8FF',
    'actorBorder': '#A3BFFA',
    'actorTextColor': '#000000',
    'rectBkg': '#FAF7F0',
    'rectBorder': '#E3DDCF',
    'noteBkgColor': '#E6F4EA',
    'noteBorderColor': '#34D399',
    'noteTextColor': '#064E3B',
    'signalColor': '#2D3748',
    'signalLineColor': '#4A5568'
  }
}}%%
sequenceDiagram
    autonumber
    participant Orchestrator as go-pm-backend (gRPC Client)
    participant GrpcSrv as GRPC Server (FastAPI / grpcio)
    participant Compiler as BPMNtoPetriNetCompiler (LRU Cache)
    participant CH as ClickHouse DBMS (logs.process_events)
    participant Engine as PM4Py Alignment / OCPM Engine

    Orchestrator->>GrpcSrv: gRPC AnalyzeConformance(ConformanceRequest)
    activate GrpcSrv
    
    note over GrpcSrv: Шаг 2: Валидация аргументов и извлечение SHA256(bpmn_xml_content)
    GrpcSrv->>Compiler: GetPetriNet(bpmn_xml_string)
    activate Compiler
    
    note over Compiler: Проверка in-memory кэша. При промахе — конвертация BPMN_TO_PETRI_NET
    Compiler-->>GrpcSrv: Tuple(Net, InitialMarking, FinalMarking)
    deactivate Compiler

    note over GrpcSrv: Шаг 3: Построение оптимизированного SQL с groupArray и window-функциями
    GrpcSrv->>CH: SELECT case_id, groupArray(activity), ... FROM logs.process_events
    activate CH
    CH-->>GrpcSrv: Raw result_rows (до 5000 кейсов за окно)
    deactivate CH

    note over GrpcSrv: Шаг 4: Распаковка в Pandas DataFrame и вызов OCPM Projection
    GrpcSrv->>Engine: alignments.apply_log(event_log, net, im, fm)
    activate Engine
    
    note over Engine: Шаг 5: Расчет токенов c, p, m, r и классификация (COMPLIANT / BUG / FEATURE)
    Engine-->>GrpcSrv: Aligned traces & Diagnostic structures
    deactivate Engine

    note over GrpcSrv: Шаг 6: Сборка SummaryStats и сериализация protobuf
    GrpcSrv-->>Orchestrator: ConformanceResponse (gRPC Status OK)
    deactivate GrpcSrv

Figure 1: Сквозной конвейер вычислений Alignment и OCPM в воркере python-pm4py-worker

8 Таблица расшифровки шагов воркера (python-pm4py-worker)

Table 2: : Пошаговый сценарий конформного анализа и обработки исключений воркером pm4py

Шаг	Операция	Входные данные	Выходные данные / Артефакт	Обработка исключений (Fallback)
1	Прием gRPC-запроса	ConformanceRequest от go-pm-backend	Распакованный XML и пороги	Возврат gRPC INVALID_ARGUMENT при пустом bpmn_xml_content
2	Компиляция BPMN	bpmn_xml_content	(PetriNet, InitialMarking, FinalMarking) из LRU-кэша	При ошибке парсинг XML — INTERNAL: BPMN_XML_PARSE_ERROR
3	Выгрузка из ClickHouse	scenario_id, окно дат	Сырой pd.DataFrame трасс	При таймауте ClickHouse (>5s) — пустой датафрейм / деградация
4	Расчет Alignment	event_log + PetriNet	Массив выравниваний (aligned_traces)	Fallback на эвристический Token Replay при OOM на сложных петлях
5	Классификация аномалий	Метрики c, p, m, r	CaseDiagnostic с classification и списком anomalies	Присвоение CRITICAL_BACKEND_BUG при missing > 0
6	Формирование gRPC-ответа	Структуры SummaryStats + repeated CaseDiagnostic	Сериализованный protobuf в Go	Логирование в stdout с log_stream: process

9 Справочник ошибок воркера (python-pm4py-worker)

Table 3: : Справочная матрица gRPC-статусов, каноничных кодов ошибок воркера и стратегий их обработки

gRPC Статус	Каноничный код (canonical_code)	Уровень	Сообщение в логе (message)	Причина сбоя и действие
3 / INVALID_ARGUMENT	PM4PY_EMPTY_BPMN_XML	ERROR	BPMN XML payload is empty or malformed.	Оркестратор передал битый XML из MinIO
13 / INTERNAL	PM4PY_ALIGNMENT_SOLVER_OOM	CRITICAL	ILP Solver out of memory on complex loops.	Переключение на эвристический алгоритм Token Replay
14 / UNAVAILABLE	CLICKHOUSE_QUERY_TIMEOUT	ERROR	ClickHouse query exceeded 5000ms deadline.	Сокращение окна выборки до INTERVAL 3 DAY
2 / UNKNOWN	OCPM_OBJECT_MAPPING_FAIL	WARN	Failed to extract multi-object keys from attributes_json.	Откат к монолитному case_id-графу