データの来歴を追跡するprovenance設計｜Entity・Activity・Agent・DB実装

2026-10-06

設計値や部品表をDBへ保存できても、「この値はどのfileから来たのか」「どの変換programを通ったのか」「なぜ前回と違うのか」が分からなければ、誤りを見つけた後の調査が止まります。

必要なのは最新版だけでなく、入力、処理、出力の関係をデータとして残すprovenance（来歴）です。この記事では、W3C PROVの基本modelを使い、JSON、SQLite、Pythonで追跡可能な変換履歴を設計します。

provenanceが解決する問題

次のJSONは部品表の出力としては読めますが、生成過程が分かりません。

{
  "assembly_id": "assy-100",
  "revision": 3,
  "total_mass_kg": 128.4
}

この値だけでは、次の問いに答えられません。

どのCSV、CAD、DB recordを入力にしたか
いつ、どのversionのprogramで変換したか
自動処理か手動修正か
同じ入力から再生成できるか
誤った入力の影響がどの出力へ広がったか

通常の監査logは「誰が何を操作したか」を時系列で記録します。lineageはdata同士の由来をたどり、provenanceは入力、処理、生成物、責任主体を含む来歴を表します。実務では厳密な用語差より、必要な関係を機械的に追跡できることが重要です。

W3C PROVの3要素

W3C PROV-DMは、provenanceを表す中心概念としてEntity、Activity、Agentを定義しています。

要素	意味	設計dataの例
Entity	固定した側面を持つ物理的・デジタル・概念的な対象	CSV、図面revision、DB snapshot、計算結果
Activity	一定期間に実行され、Entityを使用・変換・生成する処理	CSV取込、単位換算、BOM展開、CAD export
Agent	ActivityやEntityに責任を持つ主体	利用者、組織、service account、software agent

重要なのは、fileだけを履歴対象にしないことです。変換programの実行もActivityとしてIDを持たせます。これにより、同じ入力を別versionのprogramで処理した結果を区別できます。

代表的な関係は次のとおりです。

used：Activityが入力Entityを使用した
wasGeneratedBy：出力EntityがActivityによって生成された
wasDerivedFrom：出力Entityが入力Entityから派生した
wasAssociatedWith：ActivityにAgentが関与した

すべてをRDFやontologyへ変換しなければならないわけではありません。最初は同じ意味を持つJSONとrelational tableで実装し、外部交換が必要になった段階でPROV-Oへ対応できます。

最小のprovenance JSON

CSVの部品表をJSONへ変換した例です。

{
  "schema_version": 1,
  "entities": [
    {
      "entity_id": "bom-csv-assy100-r3",
      "entity_type": "bom_source_csv",
      "content_sha256": "<sha256-hex>"
    },
    {
      "entity_id": "bom-json-assy100-r3",
      "entity_type": "bom_normalized_json",
      "content_sha256": "<sha256-hex>"
    }
  ],
  "activities": [
    {
      "activity_id": "run-20261006-0001",
      "activity_type": "csv_to_bom_json",
      "started_at": "2026-10-06T01:20:00Z",
      "ended_at": "2026-10-06T01:20:03Z",
      "software_version": "1.4.2",
      "mapping_version": 3
    }
  ],
  "agents": [
    {
      "agent_id": "service:bom-importer",
      "agent_type": "software_agent"
    }
  ],
  "relations": [
    {
      "type": "used",
      "activity_id": "run-20261006-0001",
      "entity_id": "bom-csv-assy100-r3"
    },
    {
      "type": "wasGeneratedBy",
      "entity_id": "bom-json-assy100-r3",
      "activity_id": "run-20261006-0001"
    },
    {
      "type": "wasDerivedFrom",
      "generated_entity_id": "bom-json-assy100-r3",
      "source_entity_id": "bom-csv-assy100-r3"
    },
    {
      "type": "wasAssociatedWith",
      "activity_id": "run-20261006-0001",
      "agent_id": "service:bom-importer"
    }
  ]
}

最上位はobjectで、entities、activities、agents、relationsはarrayです。各要素はobjectであり、IDとtypeをstring、versionをinteger、日時をRFC 3339形式のstringとして保持します。

hashは内容の同一性確認に使えますが、hashだけでは「誰が、どの処理で生成したか」は分かりません。provenanceの関係と組み合わせます。

入力・変換・出力を分離する

処理の流れは次のように表せます。

入力Entity: bom-csv-assy100-r3
  ↓ used
Activity: run-20261006-0001
  ↓ wasGeneratedBy
出力Entity: bom-json-assy100-r3

Activity ← wasAssociatedWith ← Agent
出力Entity ← wasDerivedFrom ← 入力Entity

Activityには、少なくとも次を持たせます。

実行ごとに変わらないactivity_id
処理の種類
開始・終了日時
software、mapping、schemaのversion
成功、失敗、取消などのstatus
実行環境を識別する非機密情報

同じ処理を再試行したとき、新しいActivityとして残すか、同じrequest IDへ結び付けるかを契約で決めます。業務処理の冪等性と、実行履歴の回数は別の問題です。

SQLiteで関係を保存する

Entity、Activity、Agentと、その関係を別tableへ分けます。

PRAGMA foreign_keys = ON;

CREATE TABLE entities (
    entity_id      TEXT PRIMARY KEY NOT NULL,
    entity_type    TEXT NOT NULL,
    content_sha256 TEXT,
    schema_version INTEGER NOT NULL CHECK (schema_version > 0),
    created_at     TEXT NOT NULL,
    invalidated_at TEXT
);

CREATE TABLE activities (
    activity_id      TEXT PRIMARY KEY NOT NULL,
    activity_type    TEXT NOT NULL,
    started_at       TEXT NOT NULL,
    ended_at         TEXT,
    status           TEXT NOT NULL CHECK (
        status IN ('running', 'succeeded', 'failed', 'cancelled')
    ),
    software_version TEXT NOT NULL,
    mapping_version  INTEGER NOT NULL CHECK (mapping_version > 0)
);

CREATE TABLE agents (
    agent_id   TEXT PRIMARY KEY NOT NULL,
    agent_type TEXT NOT NULL CHECK (
        agent_type IN ('person', 'organization', 'software_agent')
    )
);

CREATE TABLE activity_inputs (
    activity_id TEXT NOT NULL,
    entity_id   TEXT NOT NULL,
    input_role  TEXT NOT NULL,
    PRIMARY KEY (activity_id, entity_id, input_role),
    FOREIGN KEY (activity_id) REFERENCES activities(activity_id),
    FOREIGN KEY (entity_id) REFERENCES entities(entity_id)
);

CREATE TABLE entity_generations (
    entity_id   TEXT PRIMARY KEY NOT NULL,
    activity_id TEXT NOT NULL,
    FOREIGN KEY (entity_id) REFERENCES entities(entity_id),
    FOREIGN KEY (activity_id) REFERENCES activities(activity_id)
);

CREATE TABLE entity_derivations (
    generated_entity_id TEXT NOT NULL,
    source_entity_id    TEXT NOT NULL,
    PRIMARY KEY (generated_entity_id, source_entity_id),
    FOREIGN KEY (generated_entity_id) REFERENCES entities(entity_id),
    FOREIGN KEY (source_entity_id) REFERENCES entities(entity_id),
    CHECK (generated_entity_id <> source_entity_id)
);

CREATE TABLE activity_agents (
    activity_id TEXT NOT NULL,
    agent_id    TEXT NOT NULL,
    agent_role  TEXT NOT NULL,
    PRIMARY KEY (activity_id, agent_id, agent_role),
    FOREIGN KEY (activity_id) REFERENCES activities(activity_id),
    FOREIGN KEY (agent_id) REFERENCES agents(agent_id)
);

SQLiteでは外部キー制約をconnectionごとに有効化します。IDが存在しない関係を保存できないようにし、孤立したlineageを防ぎます。

entity_generations.entity_idを主キーにした例では、1つのEntityに生成Activityは1つです。複数の生成経路を同一Entity IDへ集約したい要件ならmodelを変更しますが、通常は生成物ごとに新しいEntity IDを発行する方が追跡しやすくなります。

Pythonで冪等に登録する

次の例は、同じIDと同じ属性の再登録を許可し、同じIDで内容が違う場合は衝突として拒否します。

import sqlite3


def ensure_entity(
    conn: sqlite3.Connection,
    *,
    entity_id: str,
    entity_type: str,
    content_sha256: str | None,
    schema_version: int,
    created_at: str,
) -> None:
    expected = (
        entity_type,
        content_sha256,
        schema_version,
        created_at,
    )

    current = conn.execute(
        """
        SELECT entity_type, content_sha256, schema_version, created_at
        FROM entities
        WHERE entity_id = ?
        """,
        (entity_id,),
    ).fetchone()

    if current is None:
        conn.execute(
            """
            INSERT INTO entities (
                entity_id, entity_type, content_sha256,
                schema_version, created_at
            ) VALUES (?, ?, ?, ?, ?)
            """,
            (
                entity_id,
                entity_type,
                content_sha256,
                schema_version,
                created_at,
            ),
        )
        return

    if tuple(current) != expected:
        raise ValueError("同じentity_idに異なる属性が登録されています")


def link_derivation(
    conn: sqlite3.Connection,
    generated_entity_id: str,
    source_entity_id: str,
) -> None:
    with conn:
        conn.execute(
            """
            INSERT OR IGNORE INTO entity_derivations (
                generated_entity_id,
                source_entity_id
            ) VALUES (?, ?)
            """,
            (generated_entity_id, source_entity_id),
        )

SQLへ値を渡すときは文字列連結せずplaceholderを使います。INSERT OR IGNOREは同じ関係の再登録を重複させません。ただし外部キー違反やID衝突を正常扱いにせず、原因別に記録します。

本番処理では、Entity、Activity、relation、処理結果の保存を同じtransactionへ入れます。業務dataだけcommitされ、provenanceが失敗する状態を避けます。

upstreamを再帰的に調べる

ある計算結果がどの入力から派生したかは、recursive CTEで遡れます。

WITH RECURSIVE lineage(entity_id, depth, path) AS (
    SELECT
        :start_entity_id,
        0,
        ',' || :start_entity_id || ','

    UNION ALL

    SELECT
        d.source_entity_id,
        lineage.depth + 1,
        lineage.path || d.source_entity_id || ','
    FROM entity_derivations AS d
    JOIN lineage
      ON d.generated_entity_id = lineage.entity_id
    WHERE lineage.depth < 50
      AND instr(
          lineage.path,
          ',' || d.source_entity_id || ','
      ) = 0
)
SELECT entity_id, depth
FROM lineage
ORDER BY depth, entity_id;

深さ上限と訪問済みpathを使い、誤った循環参照でqueryが終わらなくなることを防ぎます。登録時にも、派生関係がcycleを作らないか検証します。

逆向きにたどれば、ある入力fileの誤りが影響した出力、API response、計算結果、図面を抽出できます。

正常系と異常系

case	扱い
同じID・同じ属性を再登録	冪等な再実行として許可
同じID・異なるhash	ID衝突として隔離
存在しないEntityとのrelation	外部キー制約で拒否
未対応schema version	変換せず隔離
Activity終了前に出力を確定	transactionまたはstatusで防止
派生関係がcycleを形成	登録前validationで拒否
`ended_at`がNULL	実行中か異常終了かをstatusで区別

日時はUTCなどの基準と表記形式を固定します。NULLは「不明」「未終了」「該当なし」のどれかをfieldごとに決め、空文字と混同しません。IDにはfile名や表示名を直接使わず、rename後も変わらない値を発行します。

履歴を上書きしない

provenanceは、現在値を説明するための履歴です。変換規則を変更した場合、過去のActivityを更新せず、新しいActivityと出力Entityを追加します。

誤った出力を使えなくするときも、関係を削除して痕跡を消すのではなく、invalidated_at、理由、取消Activityなどを追加します。物理削除が必要な個人情報や機密dataは、保持方針を優先し、識別情報を分離・匿名化します。

セキュリティとAgent設計

AgentへAPI key、token、passwordを保存してはいけません。保存するのはservice account ID、key ID、実行roleなど、監査に必要な非秘密情報です。

人をAgentとして記録する場合も、氏名やemailを各recordへ複製せず、アクセス制御されたIDへ参照させます。provenanceにはfile path、顧客名、材料原価などが現れるため、業務dataと同じ権限管理、保持期限、監査logが必要です。

「誰が実行したか」と「誰が内容を承認したか」は別のroleです。operator、reviewer、approverなどをrelationへ持たせ、責任の意味を曖昧にしません。

CAD・BOM・APIへの展開

設計dataでは、次のようなlineageを同じmodelで表せます。

CAD model revisionからDXF・PDFを生成した履歴
部品表CSVから正規化DBと集計表を作った履歴
計算条件から選定結果と帳票を生成した履歴
API responseを変換して検索indexへ登録した履歴
単位換算・丸め・名称正規化を適用したversion

Entity IDを図面番号だけにするとrevisionを区別できません。図面という論理対象と、各revisionの固定snapshotを分けます。Activityにsoftware versionとmapping versionを残せば、将来programを更新しても、過去結果がどの規則で作られたか説明できます。