Lineage-Aware RAG for SOP Change Logs Using Git Blame
Written by
Nova Neural
I got pulled into a weird enterprise problem after building a basic Retrieval-Augmented Generation (RAG) pipeline for internal Standard Operating Procedures (SOPs). The model could answer questions like “What’s the approval workflow for vendor onboarding?”—but it routinely cited the wrong “version” of the SOP when a similar paragraph existed in multiple revisions.
What finally made this click was a detail most RAG setups ignore: in enterprises, the truth is often tied to change context (“what changed, when, and by whom”), not just the current document text. So I built a niche, practical system: Lineage-Aware RAG for SOP change logs using git blame. Instead of retrieving raw chunks, the retriever ranks chunks by how well they match both the question and the likely revision lineage.
In this post I’ll show a working, step-by-step implementation that:
- Extracts SOP chunks from repo files
- Uses
git blameto attach “lineage metadata” (author, commit, time) - Builds a vector index over chunks
- Filters and re-ranks results by lineage signals relevant to the question
This is enterprise RAG, but with an opinionated twist: retrieval is anchored to source control history.
The niche problem: answers that ignore “which revision”
In real SOP repos, teams often:
- update procedures in small ways,
- keep older wording around for reference,
- or carry forward “almost identical” sections.
A normal RAG flow retrieves semantically similar chunks and sends them to the LLM. But if the question is time-sensitive (e.g., “as of last quarter”), semantic similarity alone doesn’t guarantee the chunk is from the correct revision.
So I needed retrieval to be aware of the lineage of the text—specifically: which git commit last touched the lines that support the answer.
Key idea: attach git blame metadata to each retrieved chunk
What git blame gives us (in plain terms)
git blame tells us, for each line in a file, which commit last modified it—plus the author and timestamp (depending on configuration).
Example command:
git blame -w -L 12,20 path/to/file
I used this to compute chunk-level metadata:
commit_hashauthorcommitted_at- a derived “age” (how old the chunk is)
Then the retriever can:
- prefer newer content when the question references “current” or “last quarter”
- prefer content authored by specific teams if policies are owned by org groups
- or enforce a time cutoff
Working implementation (Python)
1) Setup: environment
This example assumes:
- your SOPs live in a local git repo
- you want to index markdown files under a folder like
sops/ - you’re using OpenAI embeddings + a local vector index via
faiss
python -m venv .venv source .venv/bin/activate pip install faiss-cpu gitpython openai
Set your OpenAI key:
export OPENAI_API_KEY="YOUR_KEY"
2) Load SOP files and chunk them
I’m using a simple line-based chunker: split on headings (in markdown) if present, otherwise group lines into fixed-size blocks.
# lineage_rag.py import os import re import subprocess from dataclasses import dataclass from typing import List, Dict, Any, Tuple import numpy as np import faiss from git import Repo @dataclass class Chunk: path: str start_line: int end_line: int text: str metadata: Dict[str, Any] def list_sop_files(repo_path: str, root_dir: str = "sops") -> List[str]: paths = [] for dirpath, _, filenames in os.walk(os.path.join(repo_path, root_dir)): for fn in filenames: if fn.endswith(".md") or fn.endswith(".txt"): paths.append(os.path.relpath(os.path.join(dirpath, fn), repo_path)) return sorted(paths) def chunk_text_by_lines(lines: List[str], path: str, start_line: int, max_lines: int = 40) -> List[Tuple[int, int, str]]: chunks = [] i = 0 current_start = start_line while i < len(lines): block = lines[i:i + max_lines] block_start = current_start block_end = current_start + len(block) - 1 text = "".join(block).strip() if text: chunks.append((block_start, block_end, text)) i += max_lines current_start = block_end + 1 return chunks def extract_chunks(repo: Repo, file_rel_path: str, max_lines: int = 40, root_dir: str = "") -> List[Chunk]: file_abs = os.path.join(repo.working_tree_dir, file_rel_path) with open(file_abs, "r", encoding="utf-8") as f: content = f.read() # Line numbering is 1-based in git blame. lines = content.splitlines(keepends=True) chunk_specs = chunk_text_by_lines(lines, file_rel_path, start_line=1, max_lines=max_lines) # We'll fill metadata later with git blame return [ Chunk( path=file_rel_path, start_line=s, end_line=e, text=t, metadata={} ) for (s, e, t) in chunk_specs ]
3) Compute git blame metadata for each chunk
This is the core lineage step.
I run git blame for the relevant line range of each file chunk and summarize the results into chunk metadata.
# lineage_rag.py (continued) def run_git_blame(repo_path: str, file_rel_path: str, start_line: int, end_line: int) -> str: # -w ignores whitespace changes, -L selects a line range cmd = ["git", "-C", repo_path, "blame", "-w", "-L", f"{start_line},{end_line}", "--", file_rel_path] return subprocess.check_output(cmd, text=True, stderr=subprocess.DEVNULL) def parse_blame_output(blame_text: str) -> List[Dict[str, Any]]: # Example line starts with: <commit> (Author YYYY-MM-DD HH:MM:SS +ZZZZ <line>... ) # We'll parse commit hash, author, and date. rows = [] for line in blame_text.splitlines(): # Split around the first ") " # Format is: <hash> (<author> <date> <time> <tz>) <rest> m = re.match(r"^([0-9a-f]{7,40})\s+\((.+?)\)\s+(.*)$", line) if not m: continue commit_hash = m.group(1) inside_parens = m.group(2) # author + date/time/tz combined rest = m.group(3) # inside_parens often like: "Jane Doe 2024-01-02 13:05:06 +0000" parts = inside_parens.split() # author may contain spaces; date starts at YYYY-MM-DD date_idx = next((i for i, p in enumerate(parts) if re.match(r"\d{4}-\d{2}-\d{2}", p)), None) if date_idx is None: author = inside_parens committed_at = None else: author = " ".join(parts[:date_idx]) committed_at = " ".join(parts[date_idx:date_idx+2]) # date + time rows.append({ "commit_hash": commit_hash, "author": author, "committed_at_raw": committed_at, "rest": rest }) return rows def summarize_chunk_lineage(blame_rows: List[Dict[str, Any]]) -> Dict[str, Any]: if not blame_rows: return {"commit_hash": None, "author": None, "committed_at": None} # Choose the most frequent commit hash in the line range. # This approximates "the revision lineage that last touched most lines in the chunk". commit_hashes = [r["commit_hash"] for r in blame_rows] top_commit = max(set(commit_hashes), key=commit_hashes.count) # For author, take the author associated with the top commit (first occurrence). top_row = next(r for r in blame_rows if r["commit_hash"] == top_commit) return { "commit_hash": top_commit, "author": top_row["author"], "committed_at": top_row["committed_at_raw"], }
4) Build embeddings + a FAISS index
I’m using OpenAI embeddings to create vectors for each chunk’s text. Then I store:
- vectors in FAISS
- metadata + raw text in parallel arrays
# lineage_rag.py (continued) from openai import OpenAI client = OpenAI() def embed_texts(texts: List[str], model: str = "text-embedding-3-small", batch_size: int = 64) -> np.ndarray: vectors = [] for i in range(0, len(texts), batch_size): batch = texts[i:i+batch_size] resp = client.embeddings.create(model=model, input=batch) batch_vecs = [d.embedding for d in resp.data] vectors.extend(batch_vecs) return np.array(vectors, dtype=np.float32) def build_index(chunks: List[Chunk]) -> Tuple[faiss.IndexFlatIP, np.ndarray, List[Chunk]]: texts = [c.text for c in chunks] X = embed_texts(texts) # Normalize for cosine similarity via inner product. faiss.normalize_L2(X) index = faiss.IndexFlatIP(X.shape[1]) index.add(X) return index, X, chunks
5) Parse question into lineage filters
This is where the system becomes “lineage-aware” instead of “just vector search”.
I used lightweight heuristics for a first pass:
- if the question includes “as of last quarter”, apply a cutoff date
- if it includes a named owner/team, filter by
author(from blame) - otherwise, no filter
For a production system, this can be replaced by an LLM-driven structured extraction step.
# lineage_rag.py (continued) import datetime def extract_lineage_constraints(question: str) -> Dict[str, Any]: q = question.lower() constraints = {} today = datetime.date.today() # Very small heuristic for demo: "last quarter" if "last quarter" in q: # Determine last quarter start/end (roughly). # Example: if Q3 now, last quarter is Q2. q_now = (today.month - 1) // 3 + 1 q_last = q_now - 1 if q_now > 1 else 4 year_last = today.year if q_now > 1 else today.year - 1 start_month = (q_last - 1) * 3 + 1 end_month = start_month + 2 start_date = datetime.date(year_last, start_month, 1) # cutoff at end of last quarter end_date = datetime.date(year_last, end_month, 28) # rough; good enough for demo constraints["max_committed_at"] = end_date.isoformat() # If someone says "owned by Jane Doe" (matches blame author) m = re.search(r"owned by\s+([a-z]+(?:\s+[a-z]+)*)", q) if m: constraints["author_contains"] = m.group(1).title() return constraints def parse_committed_at_raw(raw: str) -> str: # We stored a string like "YYYY-MM-DD HH:MM:SS" if raw is None: return None # Return ISO date for easy compare (YYYY-MM-DD) return raw.split()[0] def filter_chunks_by_constraints(chunks: List[Chunk], constraints: Dict[str, Any]) -> List[int]: if not constraints: return list(range(len(chunks))) ids = [] for i, c in enumerate(chunks): ok = True if "max_committed_at" in constraints: raw = c.metadata.get("committed_at") iso_date = parse_committed_at_raw(raw) if iso_date is None or iso_date > constraints["max_committed_at"]: ok = False if "author_contains" in constraints: author = c.metadata.get("author") or "" if constraints["author_contains"].lower() not in author.lower(): ok = False if ok: ids.append(i) return ids
6) Retrieve with lineage-aware filtering + re-ranking
Flow:
- Embed the question
- Get top candidates by vector similarity
- Filter candidates by lineage constraints
- If filtering removes everything, fall back to unfiltered top-k
- Feed top chunks into the LLM
For simplicity, I’m not calling the LLM to re-rank here; I’m using the lineage filter plus similarity scores.
# lineage_rag.py (continued) def embed_query(question: str) -> np.ndarray: v = client.embeddings.create(model="text-embedding-3-small", input=[question]).data[0].embedding x = np.array([v], dtype=np.float32) faiss.normalize_L2(x) return x[0:1, :] def retrieve_lineage_aware(index, chunks: List[Chunk], question: str, top_k: int = 5) -> List[Chunk]: constraints = extract_lineage_constraints(question) qv = embed_query(question) # Get more than top_k so filtering can still leave results prelim_k = max(top_k * 6, 20) scores, idxs = index.search(qv, prelim_k) idxs = idxs[0].tolist() filtered_idxs = filter_chunks_by_constraints([chunks[i] for i in idxs], constraints) # filtered_idxs are indices relative to the filtered list, so map back: filtered_global = [idxs[i] for i in filtered_idxs] # If filter wipes everything, fall back use_idxs = filtered_global[:top_k] if filtered_global else idxs[:top_k] return [chunks[i] for i in use_idxs]
7) End-to-end script
This script:
- opens your repo
- extracts chunked SOPs
- attaches blame metadata
- builds a vector index
- runs a query
# lineage_rag.py (continued) def attach_lineage_metadata(repo_path: str, chunks: List[Chunk]) -> None: for c in chunks: blame = run_git_blame(repo_path, c.path, c.start_line, c.end_line) rows = parse_blame_output(blame) c.metadata = summarize_chunk_lineage(rows) def main(): repo_path = os.environ.get("SOP_REPO_PATH", ".") repo = Repo(repo_path) sop_files = list_sop_files(repo_path, root_dir="sops") chunks: List[Chunk] = [] for fp in sop_files: chunks.extend(extract_chunks(repo, fp, max_lines=40)) attach_lineage_metadata(repo_path, chunks) index, _, chunks = build_index(chunks) question = "What is the vendor onboarding approval workflow as of last quarter?" results = retrieve_lineage_aware(index, chunks, question, top_k=5) print("\n=== Retrieved chunks (with lineage metadata) ===") for r in results: print(f"\n- {r.path}:{r.start_line}-{r.end_line}") print(f" author: {r.metadata.get('author')}") print(f" committed_at: {r.metadata.get('committed_at')}") print(f" text_preview: {r.text[:160].replace('\\n',' ')}...") if __name__ == "__main__": main()
Run it (point to your SOP repo root):
export SOP_REPO_PATH="/path/to/your/sop-git-repo" python lineage_rag.py
You’ll see retrieved chunks plus:
authorcommitted_at- a text preview
The first time I ran this, the improvement was immediate: chunks that looked relevant but came from newer/older revisions stopped being returned when the question required “as of last quarter”.
What changes in behavior vs normal RAG
With vanilla enterprise RAG:
- retrieval is “semantic-only”
- the LLM often answers accurately, then confidently cites the wrong historical interpretation
With lineage-aware retrieval:
- retrieval becomes “semantic + provenance”
- the system gravitates toward chunks whose source lines were last touched during the relevant timeframe (or by the relevant owner)
Even without changing the LLM at all, citation quality and answer correctness tend to stabilize because the retrieved evidence matches the revision the question is asking about.
Production-minded notes (what I learned the hard way)
- Chunk boundaries matter. A chunk that spans multiple revisions can blur blame metadata. Line-based chunking is imperfect; markdown-aware chunking (by headings/sections) is better.
git blamecan be expensive. In practice I cache blame summaries per(file, start_line, end_line)or precompute lineage at indexing time.- Time parsing should be robust. Blame timestamps include timezones; my demo uses a coarse approach for filtering, but a production build should normalize to UTC.
- Metadata should be stored, not recomputed. The blame step belongs in the indexing pipeline.
Conclusion
I built lineage-aware Enterprise RAG by enriching SOP chunks with git blame metadata (commit hash, author, timestamp) and using that lineage information to filter and rank retrieval results. The biggest lesson was that in enterprise policy Q&A, revision provenance is part of relevance. Once retrieval is anchored to change history, the system stops pulling “semantically similar but historically wrong” evidence—making answers and citations much more trustworthy.