data-lineage

/data-lineage — Pipeline Lineage Tracker

Track DataFrame transformation pipelines and visualize row count changes using insight_blueprint.lineage.

When to Use

• Setting up lineage tracking in a new notebook
• Exporting Mermaid diagrams from tracked pipelines
• Refactoring pipelines based on lineage analysis

When NOT to Use

• Creating or managing analysis designs (-> /analysis-design)
• Recording reasoning steps (-> /analysis-journal)
• Registering data sources (-> /catalog-register)

Prerequisites Check

Step 0: Python Package Check (MUST run before Step 1)

1. Run: python -c "from insight_blueprint.lineage import tracked_pipe; print('OK')"
2. If output is "OK": proceed to Step 1
3. If ImportError or command fails:
   • Ask the user: "data-lineage の tracked_pipe を使うには insight-blueprint Python パッケージが必要です。uv add insight-blueprint を実行しますか？（分析パイプラインの透明性追跡に推奨）"
   • If user approves: run uv add insight-blueprint, then re-check import
   • If user declines: inform user that tracked_pipe features are unavailable, continue with MCP tools only (export/Mermaid diagram generation via MCP is still available)
   • If install fails: show error, suggest manual install with pip install insight-blueprint

Note: MCP tools (Mermaid export, lineage graph generation) work without the Python package. Only tracked_pipe in user notebooks/scripts requires the package.

Workflow

Mode A: Setup — /data-lineage <design_id> setup

Help the user add tracked_pipe to an existing notebook.

Workflow:

1. Confirm the target AnalysisDesign (if design_id provided, call get_analysis_design)
2. Locate the target notebook/script (ask the user if unclear)
3. Read the notebook and identify DataFrame transformation steps (.pipe(), filter, merge, join, dropna, assign, query, etc.)
4. For each transformation, suggest wrapping with tracked_pipe:

from insight_blueprint.lineage import LineageSession, tracked_pipe

session = LineageSession(name="<pipeline_name>", design_id="<design_id>")

# Before:
df = df.dropna(subset=["price"])

# After:
df = df.pipe(tracked_pipe(
    lambda df: df.dropna(subset=["price"]),
    reason="price 欠損行を除外",
    session=session,
))