Build a deployed, evaluated search for SAP products and business partners that catches duplicates before they're created, measured with a business KPI.
Every SAP company has people who look up master data all day. A buyer needs the material number for a part. A service agent needs the right customer. An accounts payable clerk needs the supplier that sent an invoice. They type a few words into a search box and hope the record uses the same words.
When the search fails, two things happen:
Time is lost. People scroll lists, ask a colleague or phone the master data team.
Duplicates are born. If nobody finds "Hex bolt M8x40 zinc plated", someone creates "Bolt, hexagon head, M8 x 40, galvanised". The same holds for a supplier entered as "Müller" once and "Mueller" later.
Duplicates are the expensive part. Stock of one part sits under two numbers. Spend reports split one supplier in two. MRP plans two items. A data steward later spends days finding and merging records that better search would have prevented.
Who feels it: buyers and planners (procure-to-pay, plan-to-produce), service and sales staff (order-to-cash), the master data team, and finance when reports don't add up.
Today's numbers have to come from your own company. In this project we use a made-up baseline: a buyer needs a median of 95 seconds to find the right record, measured by watching users before go-live. Your real number may be very different, and it is the first thing to measure.
8:10. Anna, a buyer, types "hearing protection" into the new search page. With an embedding model running, "Ear muffs, 30 dB" and "Foam ear plugs" come up first, though neither shares a word with her search. She clicks This is the one. The system notes that she found it in 12 seconds.
8:25. She types "isopropyl alcohol". The system answers no good match, because that chemical is restricted to lab buyers. She doesn't see a record she isn't allowed to see.
9:40. Anna needs a new supplier, "Muller Industrietechnik GmbH". Before she asks for it to be created, she runs the duplicate check. Two existing records come back: "Müller Industrietechnik" and "Mueller Industrietechnik", both in Stuttgart. She uses the first one. No new record is created.
9:41. Two existing records for one company is itself a problem. Anna sends the pair to the data steward queue.
14:00. Sam, the data steward, reviews the pair and approves it as a duplicate with a reason. The approved pair goes to a file the master data team works through in SAP. The search system itself never changes data in SAP.
The solution has three jobs: find the right record fast, warn before a duplicate is created, and route suspected duplicates to a person who decides.
A KPI that a business owner can sign means a metric, a baseline, a target and a method. The success metrics topic explains how to agree one. For this project:
Time to find
Duplicate creation avoided
Metric
Median seconds from search to choosing a record
Duplicate checks where the user picked an existing record instead of creating one
Baseline
95 s (made-up sample; measure yours by shadowing)
Duplicates found by stewards per month before go-live
Target
Agree it with the business, for example under 30 s
Agree it, for example half of last year's duplicates
How it is measured
The system's own event log: search time and click time
The same log, confirmed by steward reviews
Guardrail
No user sees a record outside their authorizations
Wrong duplicate warnings stay rare (precision)
The log matters. The system records every search, every click and every duplicate check, so the KPI comes from facts, not from a survey.
flowchart LR
S[SAP: product and<br/>business partner APIs] -->|read only| D[Data copy<br/>products, partners]
D --> I[Search index<br/>keywords + meaning]
U[User in browser] --> A[Search API]
A --> G{Access filter<br/>deleted, restricted}
G --> I
A --> L[(Event log)]
A --> Q[Steward queue]
Q -->|approved list| M[Master data team<br/>changes SAP]
L --> K[KPI report]
Where SAP fits. As of October 2026, SAP already covers parts of this:
SAP Master Data Governance (MDG) has a duplicate check built into creating a business partner. SAP Learning describes it checking address or identification data, such as an ID or tax number.
Joule in MDG on SAP S/4HANA Cloud Private Edition lets users search and display records and submit business partners in natural language, according to SAP's product page. It needs a Joule Base subscription and MDG.
SAP HANA Cloud can store the embeddings next to business data, if you build your own search on SAP's platform.
So the first question is not "how do we build this?" but "does MDG already cover our case, and is it licensed?" This project is worth building when the need is broader: product search for buyers, one search across objects, or an AI assistant that must find the right record first.
The pain (30 s). Search "hearing protection" in today's search: nothing. Show the two hex bolt records that are the same part.
The search (45 s). Same query in the new page, with the embedding model running: ear protection comes first. Then "M-1016": exact product number, found at once. Then "canteen menu": "no good match".
The safety (30 s). Switch the user to Anna and search a restricted chemical: nothing. Switch to Ben, a lab buyer: found.
The duplicate check (45 s). Enter "Muller Industrietechnik GmbH, Stuttgart" as a new supplier: two existing records come back. Send them to the steward. Show that only Sam can approve, and not his own request.
The numbers (30 s). Show the evaluation report and the KPI report, and say clearly which numbers are from tests and which from real use.
Pick one answer for each question. The explanation appears after you choose.
1Why is duplicate creation usually the bigger cost of failed master data search?
Answer: C. A failed search often ends with someone creating a record that already exists. Stock, spend reports and MRP then treat one thing as two, and a steward must merge it later.
2Which KPI design can a business owner sign?
Answer: B. A signable KPI names the metric, the baseline, the target and how it is measured. The system's own log of search and click times gives the facts.
3A buyer searches for a chemical restricted to lab staff. What should happen?
Answer: A. The access filter runs before ranking, so records a user may not see in SAP never appear. A copy of SAP data must respect the same authorizations.
4Your company already licenses SAP Master Data Governance. What do you check first?
Answer: B. SAP Learning describes MDG's duplicate check as part of creating a business partner. Build only what isn't covered, such as product search for buyers or one search across objects.
5Who may approve that two supplier records are duplicates?
Answer: D. A similarity score is a hint, not a decision. The design routes suspected duplicates to a steward, enforces a second pair of eyes, and the search system never changes SAP data itself.
6The demo shows "M8" and "M10" bolts side by side as top results. What is the risk?
Answer: C. Embeddings place similar descriptions close together, even when a size differs. Keyword search and a rule that compares numbers keep different sizes apart.
7Before go-live, which question must be answered with SAP, not only internally?
Answer: C. SAP's API policy sets rules for extracting and replicating data through its APIs. Whether a design fits depends on the contract, so confirm it with SAP and record the answer.
Ask a question
Testing: only staff see this
Stuck on something in this layer? Ask it here. Questions are answered in the order they arrive, and the answer appears under My questions.
Sign in (free) to ask a question. You can ask anonymously.
You will turn the hybrid search from Semantic search over SAP master data into a complete, small product: a search service for SAP products and business partners that a buyer can use in a browser, that catches duplicates before they are created, that a data steward controls, and that reports its own KPI.
flowchart LR
S["Stage 1: SAP data<br/>sap_data.py"] --> R["Stage 2: retrieval<br/>search_core.py"]
R --> A["Stage 3: API<br/>app.py"]
A --> U["Stage 4: web page<br/>served by app.py"]
A --> D["Stage 5: deploy<br/>manifest.yml"]
R --> E["Stage 6: evaluation<br/>evaluate.py"]
E -->|thresholds| R
A -->|events.jsonl| K["Stage 7: KPI<br/>kpi.py"]
Seven files of Python do the work, one per stage, plus a test file and three small deployment files. Everything runs on made-up, SAP-shaped data without any account. Real data from SAP's sandbox and a real embedding model are optional switches.
Prerequisites. You should have done these published topics, or be ready to look back at them:
Before you start: complete Set up your computer for this course, Set up for Unit 3 and Set up for Unit 6. They give you the orchestrate-course folder, its .venv, numpy, Sentence Transformers, FastAPI, requests, python-dotenv and your SAP_API_KEY in .env. This project adds no new libraries.
In this stage you get products and business partners into two JSON files. Every later stage reads only those files, so it doesn't matter whether they came from made-up data or SAP's sandbox.
flowchart LR
A["--sample<br/>made-up records"] --> F["data/products.json<br/>data/partners.json"]
B["--sandbox<br/>API_PRODUCT_SRV<br/>API_BUSINESS_PARTNER"] --> F
SAP's Developer Center tutorials show the Business Partner service at /sap/opu/odata/sap/API_BUSINESS_PARTNER, with BusinessPartnerFullName, SearchTerm1 and the address fields CityName and Country. They also note that the demo APIs on the SAP Business Accelerator Hub are read-only. That suits this project: it only reads.
The made-up data has 30 products and 14 business partners. It reuses the 20 products from Unit 3 and adds near-duplicates on purpose: two M8 bolts, two "Müller/Mueller" suppliers, two "Acme" suppliers. Two lab chemicals and two suppliers carry an access label (HAZ or KEY) that only some users may see. These labels are invented stand-ins for whatever authorization your SAP system uses.
The data is regenerated by the script, the logs hold users' searches, and the exports hold steward decisions. None of them belong in a public portfolio.
Wrote 30 products (1 marked for deletion) to data/products.json
Wrote 14 business partners to data/partners.json
Source: sample
The files are in projects/master-data-search/data/. Open partners.json: each record has an id, a text (name and city) and an auth label, usually empty.
What success looks like: two Wrote ... lines and Source: sandbox. The sandbox holds SAP's demo records, not gloves and bolts, so the labelled test set in Stage 6 won't fit them. Run --sample again before Stage 2 to follow this walkthrough; come back to sandbox data in the exercise.
This stage is the search engine: the "two rankers and a gate" from Unit 3, with three additions a real service needs.
One engine, two kinds of record. Products and business partners share the engine; the gate keeps them apart.
An access gate per user. Each demo user may see certain access labels. The gate removes everything else before ranking.
A duplicate check. Given the text of a record someone wants to create, it returns existing records that are very similar and carry the same numbers. "M12x40" is never a duplicate of "M8x40", however similar the words.
flowchart LR
Q[Query + user + kind] --> X{Exact ID?}
X -->|yes| R[That record]
X -->|no| G{Gate: kind,<br/>not deleted,<br/>user's labels}
G --> K[BM25 keywords]
G --> M[Embeddings<br/>above threshold]
K --> F[RRF fusion]
M --> F
F --> O[Top 5 or<br/>no good match]
The model (default): sentence-transformers/all-MiniLM-L6-v2, the English model from Unit 3, which turns each text into 384 numbers. It should find "Ear muffs" for "hearing protection".
--offline: no download. It splits text into overlapping three-letter pieces and hashes them into 512 numbers. That compares spelling, not meaning. It catches "Mueller" for "Müller" and typos, but it can't connect "hearing protection" with "ear muffs".
The offline stand-in exists so that everything runs even where huggingface.co is blocked, and so that a small deployment doesn't need the model's large libraries. Stage 6 measures what it costs in quality.
In projects/master-data-search, create search_core.py, paste the code below and save.
"""Stage 2: the search engine. Keyword (BM25) + meaning (embeddings) + filters + fusion (RRF),
a "no good match" answer, and a duplicate check for new records.
Other stages import this file. To try it on its own:
python projects/master-data-search/search_core.py "hearing protection"
python projects/master-data-search/search_core.py "acme fasteners chicago" --kind partner
python projects/master-data-search/search_core.py "isopropanol" --user ben
python projects/master-data-search/search_core.py "hex bolt m8" --offline # no model download
"""
import argparse
import hashlib
import json
import math
import re
import sys
from collections import Counter
from pathlib import Path
import numpy as np
HERE = Path(__file__).parent
DATA = HERE / "data"
CONFIG = HERE / "config.json"
MODEL = "sentence-transformers/all-MiniLM-L6-v2"
# Demo users and the access labels each may see ("" = unrestricted records).
# In production these come from SAP's identity and authorization setup (see "Production gap").
USERS = {
"anna": {"role": "buyer", "auth": {""}},
"ben": {"role": "lab buyer", "auth": {"", "HAZ"}},
"sam": {"role": "data steward", "auth": {"", "HAZ", "KEY"}},
}
# Thresholds. evaluate.py --write-config replaces them with values chosen on the eval set.
DEFAULTS = {"min_meaning": {"model": 0.35, "offline": 0.45}, "duplicate": {"model": 0.80, "offline": 0.75}}
def load_config() -> dict:
if CONFIG.exists():
return json.loads(CONFIG.read_text(encoding="utf-8"))
return json.loads(json.dumps(DEFAULTS)) # a copy, so callers can change it safely
def load_records() -> list[dict]:
rows = []
for name in ("products", "partners"):
path = DATA / f"{name}.json"
if not path.exists():
sys.exit(f"{path} is missing. Run: python projects/master-data-search/sap_data.py --sample")
rows += json.loads(path.read_text(encoding="utf-8"))["records"]
return rows
# ------------------------------------------------------------------ keyword ranking (BM25)
def tokens(text: str) -> list[str]:
return re.findall(r"[a-z0-9äöüßąćęłńóśźż]+", text.lower())
class BM25:
def __init__(self, texts: list[str], k1: float = 1.2, b: float = 0.75):
self.docs = [Counter(tokens(t)) for t in texts]
self.lengths = [sum(d.values()) for d in self.docs]
self.avg = sum(self.lengths) / max(len(self.docs), 1)
self.k1, self.b = k1, b
df = Counter(w for d in self.docs for w in d)
n = len(self.docs)
self.idf = {w: math.log(1 + (n - f + 0.5) / (f + 0.5)) for w, f in df.items()}
def scores(self, query: str) -> np.ndarray:
out = np.zeros(len(self.docs))
words = set(tokens(query))
for i, (doc, length) in enumerate(zip(self.docs, self.lengths)):
norm = self.k1 * (1 - self.b + self.b * length / self.avg)
for w in words:
f = doc.get(w, 0)
if f:
out[i] += self.idf[w] * f * (self.k1 + 1) / (f + norm)
return out
# ------------------------------------------------------------------ embeddings
class ModelEmbedder:
"""Real meaning: a Sentence Transformers model running on your computer."""
mode = "model"
def __init__(self, name: str = MODEL):
try:
from sentence_transformers import SentenceTransformer
except ImportError:
sys.exit("sentence-transformers is not installed (Set up for Unit 3). Or add --offline.")
print(f"Loading {name} (downloads once, then uses the copy on disk)...", file=sys.stderr)
self.model = SentenceTransformer(name)
self.name = name
def __call__(self, texts: list[str]) -> np.ndarray:
return np.asarray(self.model.encode(texts, normalize_embeddings=True))
class OfflineEmbedder:
"""No download: character trigrams hashed into 512 numbers. It matches spellings, not meanings,
so it catches typos and variants ("Mueller" vs "Müller") but not "ear muffs" for "hearing protection"."""
mode = "offline"
name = "char-trigram-hash-512"
def __call__(self, texts: list[str]) -> np.ndarray:
out = np.zeros((len(texts), 512), dtype=np.float32)
for row, text in enumerate(texts):
t = " " + " ".join(tokens(fold(text))) + " "
for i in range(len(t) - 2):
h = int(hashlib.md5(t[i:i + 3].encode()).hexdigest()[:8], 16)
out[row, h % 512] += 1.0
n = np.linalg.norm(out[row])
if n:
out[row] /= n
return out
def fold(text: str) -> str:
"""Spell German umlauts the way people type them without a German keyboard."""
return (text.lower().replace("ä", "ae").replace("ö", "oe").replace("ü", "ue").replace("ß", "ss"))
def make_embedder(offline: bool):
return OfflineEmbedder() if offline else ModelEmbedder()
# ------------------------------------------------------------------ the search engine
def rrf(rankings: list[list[int]], k: int = 60) -> list[tuple[int, float]]:
fused = Counter()
for ranking in rankings:
for rank, i in enumerate(ranking, start=1):
fused[i] += 1 / (k + rank)
return fused.most_common()
def numbers(text: str) -> set[str]:
"""Sizes and codes such as 8, 40, 6204, 10 from 'M8x40', '6204-2RS', '10 A'."""
return set(re.findall(r"\d+(?:[.,]\d+)?", text))
class SearchEngine:
def __init__(self, records: list[dict], embedder, config: dict | None = None):
self.records = records
self.embedder = embedder
self.config = config or load_config()
texts = [r["text"] for r in records]
self.bm25 = BM25([fold(t) for t in texts])
self.vectors = embedder(texts)
self.by_id = {r["id"].lower(): i for i, r in enumerate(records)}
def threshold(self, name: str) -> float:
return float(self.config[name][self.embedder.mode])
def allowed(self, user: str, kind: str, include_deleted: bool = False) -> np.ndarray:
"""The gate: right object type, not deleted, and only records this user may see."""
labels = USERS[user]["auth"]
return np.array([r["kind"] == kind and (include_deleted or not r["deleted"]) and r["auth"] in labels
for r in self.records])
def search(self, query: str, user: str = "anna", kind: str = "product", top: int = 5,
method: str = "hybrid", min_meaning: float | None = None) -> dict:
"""Returns {"results": [...], "no_match": bool, "exact": bool}."""
gate = self.allowed(user, kind)
hit = self.by_id.get(query.strip().lower())
if hit is not None and gate[hit]: # exact ID first: IDs are never embedded
return {"results": [self._row(hit, 1.0, "exact id")], "no_match": False, "exact": True}
kw = self.bm25.scores(fold(query))
sem = self.vectors @ self.embedder([query])[0]
floor = self.threshold("min_meaning") if min_meaning is None else min_meaning
kw_rank = [i for i in np.argsort(-kw) if gate[i] and kw[i] > 0][:20]
sem_rank = [i for i in np.argsort(-sem) if gate[i] and sem[i] >= floor][:20]
if method == "keyword":
ranked = [(i, float(kw[i])) for i in kw_rank]
elif method == "meaning":
ranked = [(i, float(sem[i])) for i in sem_rank]
else:
ranked = rrf([kw_rank, sem_rank])
rows = [self._row(i, s, self._why(i, kw, sem, floor)) for i, s in ranked[:top]]
return {"results": rows, "no_match": not rows, "exact": False}
def duplicate_check(self, text: str, kind: str, user: str = "sam", top: int = 3,
threshold: float | None = None) -> list[dict]:
"""Likely duplicates of a record someone is about to create. Includes deleted records,
because re-creating an old part is a duplicate too. Different numbers (M8 vs M10) never match."""
limit = self.threshold("duplicate") if threshold is None else threshold
gate = self.allowed(user, kind, include_deleted=True)
sem = self.vectors @ self.embedder([text])[0]
new_numbers = numbers(text)
out = []
for i in np.argsort(-sem):
if not gate[i] or sem[i] < limit:
continue
if numbers(self.records[i]["text"]) != new_numbers: # the size guard
continue
out.append(self._row(i, float(sem[i]), "similar text, same numbers"))
if len(out) == top:
break
return out
def _why(self, i: int, kw: np.ndarray, sem: np.ndarray, floor: float) -> str:
parts = []
if kw[i] > 0:
parts.append("shares words")
if sem[i] >= floor:
parts.append(f"similar ({sem[i]:.2f})")
return ", ".join(parts)
def _row(self, i: int, score: float, why: str) -> dict:
r = self.records[i]
return {"id": r["id"], "text": r["text"], "group": r["group"], "deleted": r["deleted"],
"score": round(float(score), 4), "why": why}
def main() -> None:
parser = argparse.ArgumentParser(description="Search the master data from the command line.")
parser.add_argument("query")
parser.add_argument("--kind", default="product", choices=["product", "partner"])
parser.add_argument("--user", default="anna", choices=sorted(USERS))
parser.add_argument("--method", default="hybrid", choices=["keyword", "meaning", "hybrid"])
parser.add_argument("--offline", action="store_true", help="no model download; spelling-based vectors")
parser.add_argument("--duplicates", action="store_true", help="treat the text as a new record to check")
args = parser.parse_args()
engine = SearchEngine(load_records(), make_embedder(args.offline))
print(f"Mode: {engine.embedder.mode} ({engine.embedder.name}), user: {args.user}, kind: {args.kind}")
if args.duplicates:
found = engine.duplicate_check(args.query, args.kind, user=args.user)
print(f"Possible duplicates of '{args.query}':" if found else "No likely duplicate found.")
for r in found:
flag = " (marked for deletion)" if r["deleted"] else ""
print(f" {r['score']:.3f} {r['id']:9s} {r['text']}{flag}")
return
answer = engine.search(args.query, args.user, args.kind, method=args.method)
if answer["no_match"]:
print("No good match. Try other words, or ask the master data team.")
for r in answer["results"]:
print(f" {r['score']:.4f} {r['id']:9s} {r['group']:9s} {r['text']} [{r['why']}]")
if __name__ == "__main__":
main()
The scores are RRF scores, so they only order the list. The right brackets say why each record is there. Notice the M10 bolt in second place. The deleted bolt M-1013 never appears: the gate removed it.
The first line says Loading sentence-transformers/all-MiniLM-L6-v2 ..., then Mode: model. Check whether "Ear muffs, 30 dB" and "Foam ear plugs" now appear, marked similar. Your scores will differ from anyone else's.
Mode: offline (char-trigram-hash-512), user: anna, kind: partner
0.0328 10100001 DE Müller Industrietechnik GmbH, Stuttgart [shares words, similar (0.56)]
0.0323 10100002 DE Mueller Industrietechnik GmbH, Stuttgart [shares words, similar (0.56)]
The search folds "ü" to "ue" for both sides, and the spelling vectors forgive the missing "l". Both records come back: the planted duplicate is visible.
You get No likely duplicate found, although three M8 bolts exist. The duplicate threshold is a starting guess of 0.75. Stage 6 chooses it from labelled cases; then this check finds them.
The API puts the engine behind a web address, so a web page, a Joule agent or another system can call it. It also does the two things a command line can't: it logs every event for the KPI, and it enforces the steward's approval in code.
sequenceDiagram
participant B as Browser
participant A as app.py
participant E as search engine
participant L as events.jsonl
B->>A: GET /api/search?q=... (X-User: anna)
A->>E: search(q, anna, product)
E-->>A: results or no match
A->>L: search event + search_id
A-->>B: results + search_id
B->>A: POST /api/select (search_id, record)
A->>L: select event + seconds_to_find
The endpoints:
Endpoint
Who
What it does
GET /health
anyone
Says the service is up, which mode it runs in and how many records it holds
GET /api/search
any user
Searches in the user's scope; logs the search
POST /api/select
any user
Records which result the user chose, and how many seconds after the search
POST /api/duplicate-check
any user
Returns likely duplicates of a new record's text; logs the check
POST /api/duplicate-outcome
any user
Records whether the user used an existing record or will create a new one in SAP
POST /api/reviews
any user
Proposes that some records are duplicates
POST /api/reviews/{id}/decision
data steward only
Approves or rejects, with a reason; never your own request
In projects/master-data-search, create app.py, paste the code below and save. It is long because the web page for Stage 4 is inside it, as the text in PAGE.
"""Stages 3 and 4: the search API and a one-page web UI, with an event log for the KPI and a
data steward approval queue. It reads master data; it never writes to SAP.
How to run (from your orchestrate-course folder):
python projects/master-data-search/app.py --offline # no model download; open http://127.0.0.1:8000
python projects/master-data-search/app.py # with the embedding model
python projects/master-data-search/app.py --port 8080 # another port if 8000 is taken
API documentation: http://127.0.0.1:8000/docs
"""
import argparse
import csv
import json
import os
import threading
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, Field
from search_core import USERS, SearchEngine, load_records, make_embedder
HERE = Path(__file__).parent
LOGS = Path(os.environ.get("LOG_DIR", HERE / "logs"))
EXPORTS = Path(os.environ.get("EXPORT_DIR", HERE / "exports"))
_lock = threading.Lock()
def now() -> str:
return datetime.now(timezone.utc).isoformat(timespec="milliseconds")
def log_event(event: dict) -> None:
"""Append one line to logs/events.jsonl. The KPI stage reads only this file."""
LOGS.mkdir(parents=True, exist_ok=True)
with _lock, open(LOGS / "events.jsonl", "a", encoding="utf-8") as f:
f.write(json.dumps({"ts": now(), **event}, ensure_ascii=False) + "\n")
def check_user(x_user: str | None) -> str:
"""Demo identity from the X-User header. Production takes it from SAP's identity service."""
if x_user not in USERS:
raise HTTPException(401, f"Unknown user. Send the X-User header with one of: {', '.join(USERS)}")
return x_user
class Selection(BaseModel):
search_id: str
record_id: str
rank: int = Field(ge=1, le=20)
class DuplicateCheck(BaseModel):
text: str = Field(min_length=3, max_length=120)
kind: Literal["product", "partner"] = "product"
class DuplicateOutcome(BaseModel):
check_id: str
outcome: Literal["used_existing", "created_anyway"]
record_id: str | None = None
class ReviewRequest(BaseModel):
record_ids: list[str] = Field(min_length=2, max_length=5)
note: str = Field(default="", max_length=300)
class Decision(BaseModel):
decision: Literal["approve", "reject"]
reason: str = Field(min_length=3, max_length=300)
def create_app(engine: SearchEngine, source: str) -> FastAPI:
app = FastAPI(title="Master data search", version="1.0")
searches: dict[str, float] = {} # search_id -> start time, to measure time to find
reviews: dict[str, dict] = {} # review queue (in memory; see "Production gap")
@app.get("/", response_class=HTMLResponse)
def home():
return PAGE
@app.get("/health")
def health():
return {"status": "ok", "mode": engine.embedder.mode, "records": len(engine.records), "source": source}
@app.get("/api/search")
def search(q: str, kind: Literal["product", "partner"] = "product", x_user: str | None = Header(None)):
user = check_user(x_user)
if not 1 <= len(q.strip()) <= 120:
raise HTTPException(422, "The search text must be 1 to 120 characters.")
start = time.perf_counter()
answer = engine.search(q, user, kind)
search_id = uuid.uuid4().hex[:12]
searches[search_id] = time.time()
log_event({"event": "search", "search_id": search_id, "user": user, "kind": kind, "query": q,
"results": [r["id"] for r in answer["results"]], "no_match": answer["no_match"],
"latency_ms": round((time.perf_counter() - start) * 1000, 1)})
return {"search_id": search_id, **answer}
@app.post("/api/select")
def select(body: Selection, x_user: str | None = Header(None)):
user = check_user(x_user)
started = searches.get(body.search_id)
if started is None:
raise HTTPException(404, "Unknown search_id. Search again.")
seconds = round(time.time() - started, 1)
log_event({"event": "select", "search_id": body.search_id, "user": user, "record_id": body.record_id,
"rank": body.rank, "seconds_to_find": seconds})
return {"ok": True, "seconds_to_find": seconds}
@app.post("/api/duplicate-check")
def duplicate_check(body: DuplicateCheck, x_user: str | None = Header(None)):
user = check_user(x_user)
found = engine.duplicate_check(body.text, body.kind, user=user)
check_id = uuid.uuid4().hex[:12]
log_event({"event": "duplicate_check", "check_id": check_id, "user": user, "kind": body.kind,
"text": body.text, "candidates": [r["id"] for r in found]})
return {"check_id": check_id, "candidates": found}
@app.post("/api/duplicate-outcome")
def duplicate_outcome(body: DuplicateOutcome, x_user: str | None = Header(None)):
user = check_user(x_user)
log_event({"event": "duplicate_outcome", "check_id": body.check_id, "user": user,
"outcome": body.outcome, "record_id": body.record_id})
note = ("Good: no new record needed." if body.outcome == "used_existing"
else "Noted. Create the record in SAP as usual; this tool does not create it.")
return {"ok": True, "message": note}
@app.post("/api/reviews")
def request_review(body: ReviewRequest, x_user: str | None = Header(None)):
"""Anyone may propose that records are duplicates. Only a data steward may decide."""
user = check_user(x_user)
visible = USERS[user]["auth"]
unknown = [r for r in body.record_ids if r.lower() not in engine.by_id
or engine.records[engine.by_id[r.lower()]]["auth"] not in visible]
if unknown:
raise HTTPException(422, f"Unknown record IDs: {unknown}")
review_id = uuid.uuid4().hex[:8]
reviews[review_id] = {"id": review_id, "record_ids": body.record_ids, "note": body.note,
"requested_by": user, "status": "pending"}
log_event({"event": "review_requested", "review_id": review_id, "user": user, "record_ids": body.record_ids})
return reviews[review_id]
@app.get("/api/reviews")
def list_reviews(x_user: str | None = Header(None)):
check_user(x_user)
return list(reviews.values())
@app.post("/api/reviews/{review_id}/decision")
def decide(review_id: str, body: Decision, x_user: str | None = Header(None)):
"""The approval gate. Enforced here, in code: steward role, and never your own request."""
user = check_user(x_user)
item = reviews.get(review_id)
if item is None:
raise HTTPException(404, "Unknown review.")
if USERS[user]["role"] != "data steward":
raise HTTPException(403, "Only a data steward can decide on duplicates.")
if item["requested_by"] == user:
raise HTTPException(403, "Four eyes: someone else must decide on your own request.")
if item["status"] != "pending":
raise HTTPException(409, f"Already {item['status']}.")
item.update(status=body.decision + "d", decided_by=user, reason=body.reason, decided_at=now())
log_event({"event": "review_decided", "review_id": review_id, "user": user, "decision": body.decision})
if body.decision == "approve":
# The approved merge goes to a file for the master data team to carry out in SAP
# (for example through SAP Master Data Governance). This service never changes SAP data.
EXPORTS.mkdir(exist_ok=True)
path = EXPORTS / "approved_duplicates.csv"
new = not path.exists()
with open(path, "a", newline="", encoding="utf-8") as f:
w = csv.writer(f)
if new:
w.writerow(["review_id", "record_ids", "requested_by", "decided_by", "reason", "decided_at"])
w.writerow([review_id, ";".join(item["record_ids"]), item["requested_by"], user,
body.reason, item["decided_at"]])
return item
return app
PAGE = """<!doctype html><html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1"><title>Master data search</title>
<style>
body{font-family:system-ui,sans-serif;max-width:760px;margin:24px auto;padding:0 16px;color:#1d2433}
h1{font-size:1.4rem} h2{font-size:1.1rem;margin-top:28px} input,select,button{font-size:1rem;padding:6px}
input[type=text]{width:60%} li{margin:8px 0} .muted{color:#667085;font-size:.9rem}
button{cursor:pointer} .box{border:1px solid #d0d5dd;border-radius:8px;padding:12px;margin-top:12px}
</style></head><body>
<h1>Master data search</h1>
<p class="muted">Demo on made-up or sandbox data. Searching and checking never change anything in SAP.</p>
<label>User <select id="user"><option>anna</option><option>ben</option><option>sam</option></select></label>
<label>Find <select id="kind"><option value="product">products</option><option value="partner">business partners</option></select></label>
<div class="box"><input id="q" type="text" placeholder="e.g. hearing protection, acme chicago, M-1016">
<button onclick="search()">Search</button><div id="results"></div></div>
<h2>About to create a new record?</h2>
<div class="box"><input id="newtext" type="text" placeholder="Description or name of the new record">
<button onclick="checkDup()">Check for duplicates</button><div id="dups"></div></div>
<h2>Data steward queue</h2><div class="box"><button onclick="loadReviews()">Refresh</button><div id="reviews"></div></div>
<script>
const $=id=>document.getElementById(id);
const hdr=()=>({"X-User":$("user").value,"Content-Type":"application/json"});
const esc=s=>String(s).replace(/[&<>"]/g,c=>({"&":"&","<":"<",">":">",'"':"""}[c]));
let lastSearch=null, lastCheck=null;
async function call(path,opts={}){const r=await fetch(path,{headers:hdr(),...opts});const j=await r.json();
if(!r.ok){alert(j.detail||r.status);throw new Error(r.status)} return j}
async function search(){const j=await call(`/api/search?kind=${$("kind").value}&q=${encodeURIComponent($("q").value)}`);
lastSearch=j.search_id; if(j.no_match){$("results").innerHTML="<p>No good match. Try other words, or ask the master data team.</p>";return}
$("results").innerHTML="<ol>"+j.results.map((r,i)=>`<li><b>${esc(r.id)}</b> ${esc(r.text)} <span class="muted">${esc(r.group)} · ${esc(r.why)}</span>
<button onclick="pick(${esc(JSON.stringify(r.id))},${i+1})">This is the one</button></li>`).join("")+"</ol>"}
async function pick(id,rank){const j=await call("/api/select",{method:"POST",body:JSON.stringify({search_id:lastSearch,record_id:id,rank})});
$("results").innerHTML=`<p>Selected ${esc(id)} after ${j.seconds_to_find} s.</p>`}
async function checkDup(){const j=await call("/api/duplicate-check",{method:"POST",body:JSON.stringify({text:$("newtext").value,kind:$("kind").value})});
lastCheck=j.check_id; if(!j.candidates.length){$("dups").innerHTML=`<p>No likely duplicate found.</p>
<button onclick="outcome('created_anyway')">I will create it in SAP</button>`;return}
const ids=j.candidates.map(r=>r.id);
$("dups").innerHTML="<p>Possible duplicates:</p><ul>"+j.candidates.map(r=>`<li><b>${esc(r.id)}</b> ${esc(r.text)}
${r.deleted?"<i>(marked for deletion)</i>":""} <button onclick="outcome('used_existing',${esc(JSON.stringify(r.id))})">Use this one</button></li>`).join("")+
`</ul><button onclick="outcome('created_anyway')">None fits: I will create a new one in SAP</button>
<button onclick="review(${esc(JSON.stringify(ids))})">Ask a data steward to review these</button>`}
async function outcome(o,id=null){const j=await call("/api/duplicate-outcome",{method:"POST",body:JSON.stringify({check_id:lastCheck,outcome:o,record_id:id})});
$("dups").innerHTML=`<p>${esc(j.message)}</p>`}
async function review(ids){await call("/api/reviews",{method:"POST",body:JSON.stringify({record_ids:ids,note:"from duplicate check"})});
$("dups").innerHTML="<p>Sent to the data steward queue.</p>"; loadReviews()}
async function loadReviews(){const j=await call("/api/reviews"); if(!j.length){$("reviews").innerHTML="<p class='muted'>Empty.</p>";return}
$("reviews").innerHTML="<ul>"+j.map(r=>`<li>${esc(r.record_ids.join(", "))} · asked by ${esc(r.requested_by)} · <b>${esc(r.status)}</b>
${r.status==="pending"?`<button onclick="decide('${r.id}','approve')">Approve</button><button onclick="decide('${r.id}','reject')">Reject</button>`:""}</li>`).join("")+"</ul>"}
async function decide(id,d){const reason=prompt("Reason for the decision?"); if(!reason)return;
await call(`/api/reviews/${id}/decision`,{method:"POST",body:JSON.stringify({decision:d,reason})}); loadReviews()}
$("q").addEventListener("keydown",e=>{if(e.key==="Enter")search()});
</script></body></html>"""
def main() -> None:
parser = argparse.ArgumentParser(description="Run the master data search API and web page.")
parser.add_argument("--offline", action="store_true", help="no model download; spelling-based vectors")
parser.add_argument("--port", type=int, default=int(os.environ.get("PORT", 8000)))
args = parser.parse_args()
import uvicorn
records = load_records()
source = json.loads((HERE / "data" / "products.json").read_text(encoding="utf-8"))["source"]
engine = SearchEngine(records, make_embedder(args.offline))
host = "0.0.0.0" if "PORT" in os.environ else "127.0.0.1" # Cloud Foundry sets PORT
print(f"Serving {len(records)} records ({source} data, {engine.embedder.mode} mode) on http://{host}:{args.port}")
uvicorn.run(create_app(engine, source), host=host, port=args.port, access_log=False)
if __name__ == "__main__":
main()
The tests call the API in memory with FastAPI's TestClient, without starting a server. They use the offline embedder, so they need no download.
Create test_app.py in the same folder, paste this and save:
"""Contract tests for app.py: no server, no model download, no account.
How to run (from your orchestrate-course folder):
python projects/master-data-search/test_app.py
"""
import os
import sys
import tempfile
from pathlib import Path
os.environ["LOG_DIR"] = tempfile.mkdtemp() # keep test events out of the real log
os.environ["EXPORT_DIR"] = tempfile.mkdtemp() # and test approvals out of the real export
sys.path.insert(0, str(Path(__file__).parent))
from fastapi.testclient import TestClient # noqa: E402
from app import create_app # noqa: E402
from search_core import OfflineEmbedder, SearchEngine, load_records # noqa: E402
client = TestClient(create_app(SearchEngine(load_records(), OfflineEmbedder()), "test"))
A, B, S = {"X-User": "anna"}, {"X-User": "ben"}, {"X-User": "sam"}
results = []
def check(name: str, ok: bool) -> None:
results.append(ok)
print(f" {'PASS' if ok else 'FAIL'} {name}")
print("Contract tests for projects/master-data-search/app.py")
check("health answers ok", client.get("/health").json()["status"] == "ok")
check("search without a user is refused (401)", client.get("/api/search?q=bolt").status_code == 401)
r = client.get("/api/search?q=hex bolt m8", headers=A).json()
check("search finds the M8 bolt first", r["results"][0]["id"] == "M-1001")
check("deleted products never appear in search", all(x["id"] != "M-1013" for x in r["results"]))
check("exact product number is found", client.get("/api/search?q=M-1016", headers=A).json()["exact"])
check("nonsense gets 'no good match'",
client.get("/api/search?q=canteen menu for next week", headers=A).json()["no_match"])
check("anna cannot see restricted chemicals",
all(x["id"] != "M-1025" for x in client.get("/api/search?q=isopropyl alcohol", headers=A).json()["results"]))
check("ben can see restricted chemicals",
client.get("/api/search?q=isopropyl alcohol", headers=B).json()["results"][0]["id"] == "M-1025")
check("too long a query is refused (422)", client.get("/api/search?q=" + "x" * 200, headers=A).status_code == 422)
sel = client.post("/api/select", headers=A, json={"search_id": r["search_id"], "record_id": "M-1001", "rank": 1})
check("selecting a result records the time to find", "seconds_to_find" in sel.json())
d = client.post("/api/duplicate-check", headers=A,
json={"text": "Muller Industrietechnik GmbH, Stuttgart", "kind": "partner"}).json()
check("duplicate check finds the Mueller partners", {"10100001", "10100002"} <= {c["id"] for c in d["candidates"]})
m12 = client.post("/api/duplicate-check", headers=A, json={"text": "Hex bolt M12x40 zinc plated"}).json()
check("a different size is not a duplicate", m12["candidates"] == [])
rev = client.post("/api/reviews", headers=A, json={"record_ids": ["10100001", "10100002"]}).json()
deny = client.post(f"/api/reviews/{rev['id']}/decision", headers=B, json={"decision": "approve", "reason": "same firm"})
check("a non-steward cannot approve (403)", deny.status_code == 403)
own = client.post("/api/reviews", headers=S, json={"record_ids": ["M-1004", "M-1005"]}).json()
self_ok = client.post(f"/api/reviews/{own['id']}/decision", headers=S, json={"decision": "approve", "reason": "mine"})
check("a steward cannot approve their own request (403)", self_ok.status_code == 403)
ok = client.post(f"/api/reviews/{rev['id']}/decision", headers=S, json={"decision": "approve", "reason": "same VAT ID"})
check("a steward approves someone else's request", ok.json()["status"] == "approved")
again = client.post(f"/api/reviews/{rev['id']}/decision", headers=S, json={"decision": "reject", "reason": "changed mind"})
check("a decided review cannot be decided again (409)", again.status_code == 409)
print(f"\n{sum(results)} of {len(results)} passed")
sys.exit(0 if all(results) else 1)
Run:
python projects/master-data-search/test_app.py
What success looks like:
Contract tests for projects/master-data-search/app.py
PASS health answers ok
PASS search without a user is refused (401)
PASS search finds the M8 bolt first
PASS deleted products never appear in search
PASS exact product number is found
PASS nonsense gets 'no good match'
PASS anna cannot see restricted chemicals
PASS ben can see restricted chemicals
PASS too long a query is refused (422)
PASS selecting a result records the time to find
PASS duplicate check finds the Mueller partners
PASS a different size is not a duplicate
PASS a non-steward cannot approve (403)
PASS a steward cannot approve their own request (403)
PASS a steward approves someone else's request
PASS a decided review cannot be decided again (409)
16 of 16 passed
Five of these tests guard the business rules, not the code: deleted records stay hidden, access labels hold, and the approval gate can't be bypassed. Keep them passing after every change.
Serving 44 records (sample data, offline mode) on http://127.0.0.1:8000
INFO: Started server process [12345]
...
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
The terminal is now busy running the server. Leave it open.
Open a second terminal (Terminal > New Terminal), turn on .venv again, and call the search:
You get JSON with a search_id, "no_match": false and two results, "Foam ear plugs" first. %20 is how a space is written in a web address.
Open http://127.0.0.1:8000/docs in a browser. FastAPI lists every endpoint. Choose GET /api/search, Try it out, type a query, set x-user to ben, and choose Execute.
Look at the log the API wrote, projects/master-data-search/logs/events.jsonl. Each line is one event, for example:
The simplest thing a user can try is a web page. app.py already serves one at /: a search box, a duplicate check and the steward queue. It is plain HTML and a little JavaScript, with no build step.
Leave User on anna and Find on products. Type ear plugs and press Enter.
Choose This is the one next to "Foam ear plugs, 200 pairs".
What success looks like: the page says Selected M-1021 after 4.2 s. (your seconds differ). That number is the KPI in the making: the log now has a select event with seconds_to_find.
Type M-1016 and press Enter: one result, the exact product number.
Type canteen menu for next week: the page says No good match.
#Step 2: Check for a duplicate and send it to the steward
Set Find to business partners.
Under About to create a new record?, type Muller Industrietechnik GmbH, Stuttgart and choose Check for duplicates.
Two candidates appear. Choose Ask a data steward to review these. The page says Sent to the data steward queue.
Under Data steward queue, the request shows as pending. Choose Approve while still logged in as anna, and type a reason. An alert says Only a data steward can decide on duplicates.
Switch User to sam, choose Refresh, then Approve, and type Same tax number in both records. The status changes to approved.
Open projects/master-data-search/exports/approved_duplicates.csv. It has one line with both IDs, who asked, who decided and why. That file is what the master data team works from in SAP.
Make the browser window narrow. The page has no fixed widths, so it stays usable. Buyers often search from a tablet in the warehouse; check it before a demo.
Offline mode in the cloud. The embedding model needs Sentence Transformers and its machine learning libraries, which are large. The deployed app runs --offline, so it needs only FastAPI, uvicorn and numpy and fits in 512 MB. Stage 6 shows what this costs in search quality; the Production gap shows the SAP way to get real embeddings in the cloud.
The log on a container is temporary. As Unit 6 explained, files written inside a Cloud Foundry container don't survive a restart, and each instance has its own. For a demo on the trial that's acceptable; for real use, events go to a database.
In projects/master-data-search, create these four files. For each: right-click the folder, choose New File, type the name exactly (including the dot in .cfignore), paste and save.
requirements.txt, the libraries the cloud installs. These are the versions we tested on 11 October 2026; to use yours, run pip show fastapi uvicorn numpy and copy the Version: lines.
fastapi==0.142.2
uvicorn==0.54.0
numpy==2.5.3
runtime.txt, the Python version. Cloud Foundry's Python buildpack reads it; 3.13.x means the newest 3.13 it has.
python-3.13.x
manifest.yml, how to run the app. Use spaces, not tabs.
Cloud Foundry's Python buildpack doesn't invent a start command, and its documentation recommends listening on the port in the PORT variable. app.py reads PORT and then listens on all addresses. Test that on your computer first.
You will score keyword, meaning and hybrid search on 28 labelled searches, score the duplicate check on 13 labelled new records, let the script choose both thresholds from the results, and save a report for your portfolio.
flowchart LR
S[eval_set.jsonl<br/>28 searches<br/>13 new records] --> E[evaluate.py]
E --> T[Threshold sweep]
T --> C[config.json]
E --> R[reports/eval_report_mode.md]
C --> A[app.py and<br/>search_core.py]
The labelled set includes the hard cases on purpose:
Words the record doesn't use: "hearing protection", "something to write with".
Typos and spelling variants: "nitril glvoes", "mueller industrietechnik".
Access cases: "isopropyl alcohol" is right for Ben and must be "no good match" for Anna.
No answer at all: "canteen menu", "company car lease".
Near-duplicates that aren't: a new M12 bolt, a 6206 bearing, a 20 A fuse.
What the numbers mean:
Number
Meaning
hit@1, hit@3
Share of answerable searches with a right record in first place, or in the top 3
MRR
Average of 1 / position of the first right record; 1.0 means always first
Correct "no good match"
Unanswerable searches where the system honestly said so
Wrong "no good match"
Answerable searches where the system gave up
Leaks
Results the user may not see. Must be 0; anything else stops the project
Precision, recall (duplicates)
Of the warnings shown, how many were real duplicates; of the real duplicates, how many were found
Using 41 labelled cases from eval_set.jsonl
Search (offline mode), meaning threshold 0.2
method hit@1 hit@3 MRR no-match ok wrong no-match leaks
keyword 15/24 18/24 0.69 4/4 6 0
meaning 19/24 19/24 0.79 3/4 3 0
hybrid 19/24 20/24 0.81 3/4 2 0
hybrid missed: eye protection (as anna); hearing protection (as anna); something to write with (as anna); solvent for cleaning parts (as ben); company car lease (as anna): should say no good match
Duplicate check, threshold 0.6: precision 0.93, recall 1.00 (14 found, 1 false alarms, 0 missed)
Saved the report to reports/eval_report_offline.md
Saved thresholds for offline mode to config.json: meaning 0.2, duplicate 0.6
How to read it:
Leaks are 0 for every method. The gate holds. This is the first line to check, always.
Hybrid beats keyword on hit@3 (20 against 18 of 24) and MRR (0.81 against 0.69).
The misses are all "meaning" searches. "Eye protection", "hearing protection", "something to write with" need a model that knows what words mean. Spelling vectors can't do it.
The script chose a low meaning threshold, 0.2. It counts top-3 hits plus correct "no good match" answers, and a low threshold wins on that count. The price is visible: run python projects/master-data-search/search_core.py "hearing protection" --offline again. It now shows three bearings, because they barely clear 0.2. A user would find that confusing. The score went up and the experience got worse; the exercise fixes the scoring.
The duplicate check improved. At 0.6 it finds all 14 labelled duplicates with one false alarm. Run the M8 check from Stage 2 again:
Compare reports/eval_report_model.md with reports/eval_report_offline.md. Write down three things for your evaluation report:
Does the model fix the "eye protection" and "hearing protection" misses?
Which meaning threshold did it choose, and does "canteen menu" still get "no good match"?
Did the duplicate check's precision or recall change?
We couldn't download the model in our test environment, so we can't print your numbers here. Expect them to differ from the offline ones; that difference is the case for or against paying for the model's size in production.
Restart the API (Ctrl+C, then start it again). It reads config.json at start, so it now uses the chosen thresholds.
28 labelled searches and 13 labelled new records, written to eval_set.jsonl by --init
check_labels
Stops if a label names a record that isn't in data/, for example after loading sandbox data
score_search
Runs each search as its user; counts hits, reciprocal rank, correct and wrong "no good match", and leaks
score_duplicates
Compares the duplicate check's candidates with the labels: true and false alarms, misses
Threshold sweep
Tries thresholds from 0.10 to 0.90; keeps the best meaning threshold by hits plus correct "no match", and the best duplicate threshold by F1, the stricter one on ties
--write-config
Saves both thresholds for this mode into config.json
The KPI comes from the system's own event log, not from a survey. Every select event carries seconds_to_find, and every duplicate_outcome says whether the user used an existing record.
KPI
Computed from
Search success
Searches with a select event, divided by all searches
Median seconds to find
seconds_to_find of all select events
Duplicate creations avoided
duplicate_outcome events with used_existing
Created despite a flagged candidate
created_anyway after a check that showed candidates: either a false alarm or a missed warning, worth a look
Latency p95
latency_ms of searches; 95 of 100 were faster
Real numbers need real users over weeks. To see the arithmetic now, --simulate replays the labelled searches 300 times as if users had clicked the first right result. Those numbers are made up and must never appear in a report as results.
Create kpi.py in the project folder, paste this and save:
"""Stage 7: compute the business KPIs from the system's own event log.
How to run (from your orchestrate-course folder):
python projects/master-data-search/kpi.py # read logs/events.jsonl (your real clicks)
python projects/master-data-search/kpi.py --simulate --offline # make up 300 sessions first, then report
python projects/master-data-search/kpi.py --baseline 95 # "before" median seconds to find, measured
Simulated sessions are made up to show the arithmetic. Only real users' events prove anything.
"""
import argparse
import json
import random
import statistics
from collections import Counter
from datetime import datetime, timedelta, timezone
from pathlib import Path
HERE = Path(__file__).parent
LOG = HERE / "logs" / "events.jsonl"
SIM_LOG = HERE / "logs" / "events_simulated.jsonl"
REPORTS = HERE / "reports"
SAMPLE_BASELINE = 95.0 # made-up "before" median, seconds to find the right record by shadowing users
def simulate(path: Path, offline: bool, sessions: int, seed: int = 7) -> None:
"""Replay the labelled cases as if users had run them: they click the first right record in the
top 5 (more seconds the lower it sits), or give up. Duplicate checks end in 'use existing' when a
right candidate is shown, most of the time."""
from evaluate import BUILT_IN
from search_core import SearchEngine, load_records, make_embedder
rnd = random.Random(seed)
engine = SearchEngine(load_records(), make_embedder(offline))
searches = [c for c in BUILT_IN if c["type"] == "search"]
dups = [c for c in BUILT_IN if c["type"] == "duplicate"]
t = datetime(2026, 10, 5, 8, 0, tzinfo=timezone.utc)
events = []
for n in range(sessions):
t += timedelta(minutes=rnd.randint(1, 20))
if rnd.random() < 0.8:
c = rnd.choice(searches)
out = engine.search(c["query"], c["user"], c["kind"])
ids = [r["id"] for r in out["results"]]
sid = f"sim{n}"
events.append({"ts": t.isoformat(), "event": "search", "search_id": sid, "user": c["user"],
"kind": c["kind"], "query": c["query"], "results": ids, "no_match": out["no_match"],
"latency_ms": round(rnd.uniform(5, 40), 1)})
rank = next((pos for pos, i in enumerate(ids, 1) if i in c["relevant"]), None)
if rank:
events.append({"ts": t.isoformat(), "event": "select", "search_id": sid, "user": c["user"],
"record_id": ids[rank - 1], "rank": rank,
"seconds_to_find": round(8 + 7 * rank + rnd.uniform(0, 10), 1)})
else:
c = rnd.choice(dups)
found = [r["id"] for r in engine.duplicate_check(c["query"], c["kind"])]
cid = f"simd{n}"
events.append({"ts": t.isoformat(), "event": "duplicate_check", "check_id": cid, "user": "anna",
"kind": c["kind"], "text": c["query"], "candidates": found})
right = [i for i in found if i in c["relevant"]]
used = bool(right) and rnd.random() < 0.85
events.append({"ts": t.isoformat(), "event": "duplicate_outcome", "check_id": cid, "user": "anna",
"outcome": "used_existing" if used else "created_anyway",
"record_id": right[0] if used else None})
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("".join(json.dumps(e) + "\n" for e in events), encoding="utf-8")
print(f"Wrote {sessions} simulated sessions to logs/{path.name} (made up, for practice only)")
def pct(values: list[float], q: float) -> float:
values = sorted(values)
return values[min(len(values) - 1, int(round(q * (len(values) - 1))))]
def main() -> None:
parser = argparse.ArgumentParser(description="Business KPIs from the search event log.")
parser.add_argument("--simulate", action="store_true", help="write made-up sessions, then report on them")
parser.add_argument("--sessions", type=int, default=300)
parser.add_argument("--offline", action="store_true", help="with --simulate: no model download")
parser.add_argument("--baseline", type=float, help="measured 'before' median seconds to find a record")
args = parser.parse_args()
path = SIM_LOG if args.simulate else LOG
if args.simulate:
simulate(path, args.offline, args.sessions)
if not path.exists():
raise SystemExit("No events yet. Use the web page first, or run with --simulate --offline.")
events = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
kinds = Counter(e["event"] for e in events)
searches = [e for e in events if e["event"] == "search"]
selects = [e for e in events if e["event"] == "select"]
outcomes = [e for e in events if e["event"] == "duplicate_outcome"]
checks = {e["check_id"]: e for e in events if e["event"] == "duplicate_check"}
if not searches:
raise SystemExit("The log has no searches yet.")
found_in = {e["search_id"] for e in selects}
seconds = [e["seconds_to_find"] for e in selects]
baseline = args.baseline if args.baseline is not None else SAMPLE_BASELINE
avoided = [o for o in outcomes if o["outcome"] == "used_existing"]
flagged_but_created = [o for o in outcomes if o["outcome"] == "created_anyway"
and checks.get(o["check_id"], {}).get("candidates")]
latency = [e["latency_ms"] for e in searches]
rows = [
("Searches", f"{len(searches)}"),
("Search success (a result was chosen)", f"{len(found_in) / len(searches):.0%}"),
("'No good match' answers", f"{sum(e['no_match'] for e in searches) / len(searches):.0%}"),
("Median seconds to find (after)", f"{statistics.median(seconds):.0f}" if seconds else "n/a"),
("Median seconds to find (before)", f"{baseline:.0f}" + ("" if args.baseline else " (made-up sample)")),
("Chosen result's average position", f"{statistics.mean(e['rank'] for e in selects):.1f}" if selects else "n/a"),
("Duplicate checks run", f"{kinds['duplicate_check']}"),
("Duplicate creations avoided (used existing)", f"{len(avoided)}"),
("Created despite a flagged candidate", f"{len(flagged_but_created)}"),
("Search latency p95 (ms)", f"{pct(latency, 0.95):.0f}"),
]
width = max(len(k) for k, _ in rows)
print(f"\nKPIs from logs/{path.name}")
for k, v in rows:
print(f" {k:{width}s} {v}")
if seconds:
saved = baseline - statistics.median(seconds)
print(f"\n Time to find: {baseline:.0f} s -> {statistics.median(seconds):.0f} s "
f"({saved / baseline:.0%} less) on searches that found a record.")
REPORTS.mkdir(exist_ok=True)
out = REPORTS / ("kpi_report_simulated.md" if args.simulate else "kpi_report.md")
title = "KPI report (SIMULATED sessions)" if args.simulate else "KPI report"
out.write_text(f"# {title}\n\nSource: logs/{path.name}\n\n| KPI | Value |\n| --- | --- |\n"
+ "".join(f"| {k} | {v} |\n" for k, v in rows), encoding="utf-8")
print(f"\nSaved the report to reports/{out.name}")
if __name__ == "__main__":
main()
Wrote 300 simulated sessions to logs/events_simulated.jsonl (made up, for practice only)
KPIs from logs/events_simulated.jsonl
Searches 244
Search success (a result was chosen) 70%
'No good match' answers 21%
Median seconds to find (after) 20
Median seconds to find (before) 95 (made-up sample)
Chosen result's average position 1.1
Duplicate checks run 56
Duplicate creations avoided (used existing) 24
Created despite a flagged candidate 11
Search latency p95 (ms) 39
Time to find: 95 s -> 20 s (79% less) on searches that found a record.
Saved the report to reports/kpi_report_simulated.md
Two cautions are built into the wording. "On searches that found a record" means the median ignores searches where the user gave up, so it flatters the result; report search success next to it. And the simulated latency is random; real latency comes from real requests.
Start the API (Stage 3), open the page, and run ten searches as different users, clicking This is the one when you find the right record. Run two duplicate checks.
Run the report on the real log:
python projects/master-data-search/kpi.py
You see the same table, now from logs/events.jsonl, with your own times. If you measured how long the same ten lookups take in your usual tool, pass it: --baseline 60.
Replays labelled cases through the real engine; clicks the first right result with more seconds for lower positions; uses an existing record in most duplicate checks that show a right candidate
Event filters
Split the log into searches, selections, checks and outcomes
KPI table
Success rate, medians, counts and the p95 latency, printed and saved as Markdown
--baseline
The measured "before" median; without it the report labels 95 s as a made-up sample
Two decisions in this system change SAP data, and the system makes neither of them.
Creating a record. The duplicate check only warns. If the user decides to create a new record anyway, they create it in SAP through the usual app and process; duplicate-outcome only records the decision.
Merging or retiring duplicates. Anyone may propose that records are duplicates. Only a data steward may decide, never on their own request, and always with a reason.
How it is enforced in code, not in a prompt or a policy document:
No write path to SAP exists.sap_data.py calls SAP only with requests.get, and no other file in the project talks to SAP. There is no SAP user or key with write rights anywhere in the project.
The decision endpoint checks the role and the requester before it changes anything, and returns 403 otherwise. Two contract tests prove both checks.
A decision is final in the queue (409 on a second decision), and every decision is logged with who, when and why.
The output of an approval is a file for the master data team, who carry it out in SAP with their own authorizations, for example through SAP MDG.
The same pattern carries into every later project: the AI proposes, a person with the right role approves, and the change happens through SAP's own process.
Replace the X-User header with real sign-in, such as the identity service on SAP BTP that Unit 6's deployment topic sketches. The user's roles come from the token, not from a dictionary.
Replace the invented auth labels with the authorization data your SAP system really uses for products and business partners, copied into the index as filter columns, as Grounding on SAP data without breaking authorizations explains. Re-run the leak test after every change.
Decide how the duplicate check scopes its search. Here it searches only what the user may see, so a buyer can create a duplicate of a restricted record without being warned. A real design might check the full index and return only "a similar record exists; the master data team will review".
Load changed records only, not everything every night. Unit 3 showed LastChangeDateTime on A_Product; check the equivalent for business partners on the SAP Business Accelerator Hub.
Check the extraction against SAP's API policy. Version 4.2026a sets per-API limits and preconditions for bulk extraction and replication, and allows large-scale extraction or replication only through SAP-endorsed pathways. Whether a nightly delta load of master data fits is a question for your SAP account team; record the answer.
Thirty products fit in memory. Hundreds of thousands still do on one server, but the embedding step needs time and a store. In SAP's platform, SAP HANA Cloud keeps vectors in a REAL_VECTOR column next to the business data and compares them with COSINE_SIMILARITY; SAP HANA Cloud vector engine covers it. The LangChain integration for SAP HANA Cloud adds metadata filters, specific_metadata_columns for fast filter columns, an HNSW index for large catalogues, and HanaInternalEmbeddings, which computes embeddings in the database with VECTOR_EMBEDDING when NLP is enabled on the instance.
Move the event log and the review queue from files and memory into a database. Both are lost when a Cloud Foundry container restarts, and each instance would keep its own.
Watch latency, error rates and the share of "no good match" answers. A sudden rise in "no good match" often means the data load failed. Observability for AI systems covers the tools.
Re-run the evaluation after every change of model, threshold or data, and before every release.
Offline mode costs only a small app instance. The model needs more memory, or an embedding service; in SAP HANA Cloud, in-database embeddings depend on your edition and contract.
Labelled cases cost business time. Budget a few hours of a buyer's and a steward's time to write and check them.
Read through published APIs, keep the index side by side, and change nothing in S/4HANA. The approved duplicates go back through SAP's own processes.
Check SAP MDG first for business partners. SAP Learning describes an integrated duplicate check in central governance that flags potential duplicates by address or identification. An SAP Community post from 2017, published under Technology Blogs by SAP, describes the SAP HANA-based variant, built on search rule sets and SAP HANA fuzzy search, with stopwords and term mappings to tune it. Joule in MDG on SAP S/4HANA Cloud Private Edition adds natural language search and business partner requests, according to sap.com.
Need
Build (this project)
SAP
Duplicate check when creating business partners
Possible, but you maintain it
MDG duplicate check, if licensed
Product search for buyers across descriptions and languages
Good fit
Check what your S/4HANA apps and licences already offer
Put the project in your public portfolio repository from Set up for Unit 14. It should contain:
README.md with the problem (failed searches and duplicates), the KPI (time to find, duplicates avoided, with baseline and target), and how to run each stage.
Architecture diagram: the Stage overview mermaid diagram, or your own.
Run instructions that work on a fresh clone with --sample and --offline and no account.
Evaluation report: reports/eval_report_offline.md and, if you ran it, reports/eval_report_model.md, with three sentences on what you learned.
A one-page design summary: objects and fields, the gate, the thresholds, the approval flow, and what the Production gap would change.
A 3-minute demo script: the one in the foundational layer of this topic works; rehearse it with the web page.
Leave out data/, logs/, exports/ and .env. Label any simulated KPI report as simulated in its file name and title, as kpi.py does.
Save your work:
git add .gitignore projects/master-data-search
git status
Check that git status lists no data/, logs/ or exports/ files, then:
git commit -m "Project: SAP master data semantic search"
#Exercise: make the evaluation punish junk results
Stage 6 showed a flaw: the threshold sweep rewards a low meaning threshold, and offline search then shows bearings for "hearing protection". You will add a guardrail metric, junk shown, and make the sweep respect it. The improved harness carries into the next projects.
Open projects/master-data-search/evaluate.py and find score_search.
In the dictionary s = {...} at its start, change "misses": []} to "misses": [], "junk": 0}.
Find the line s["false_no_match"] += out["no_match"]. Below it, add these two lines, with the same indentation:
if out["results"] and not (first and first <= 3):
s["junk"] += 1
An answerable search that shows results but no right one in the top 3 counts as junk.
In the if not c["relevant"]: block, below s["unanswerable"] += 1, add the same idea for unanswerable searches that show anything:
s["junk"] += not out["no_match"]
In main, find sweep.append((s["hit3"] + s["no_match_ok"], -t, t, s)) and change the first value so junk costs a point:
Print the junk count: in main, find {s['leaks']}") at the end of the line that prints each method's numbers, and change it to {s['leaks']} junk {s['junk']}").
Run python projects/master-data-search/search_core.py "hearing protection" --offline again and note whether the bearings are gone.
In your evaluation report, write two sentences: which threshold the new rule chose, and what it cost in hit@3.
Done when: the evaluation prints a junk count for each method, the chosen offline meaning threshold is higher than 0.2, and your evaluation report explains the trade-off in two sentences.
Pick one answer for each question. The explanation appears after you choose.
1Why does every stage after Stage 1 read only data/products.json and data/partners.json?
Answer: B. sap_data.py turns made-up records and sandbox pages into the same record format. Retrieval, API, evaluation and KPI never need to know where the data came from.
2The engine filters by kind, deletion flag and access label before ranking. What breaks if you rank first and filter the top 5 afterwards?
Answer: B. If the top 5 happen to be deleted or restricted records, filtering afterwards leaves nothing, although allowed matches sat lower down. The gate must run first.
3Why does duplicate_check drop candidates whose numbers differ, even with a high similarity?
Answer: C. Similar wording with a different size is a classic false duplicate. The size guard compares the numbers exactly, so a new M12 bolt is never matched to M8 or M10 bolts.
4In the offline evaluation, the sweep chose a meaning threshold of 0.2 and "hearing protection" then showed bearings. What went wrong?
Answer: C. The scoring rewarded a low threshold because it added top-3 hits without penalizing wrong results shown. The exercise adds a junk count so the sweep pays for confusing results.
5How does the system make sure only a data steward approves duplicates?
Answer: A. The rule lives in decide on the server, and contract tests prove it. A hidden button alone is not a control, because anyone can call the API directly.
6A colleague wants the deployed demo to use the real embedding model instead of --offline. What do you check first?
Answer: A. The model brings large libraries and needs more memory than the 512 MB offline app. Compare the model and offline evaluation reports to see whether the gain is worth that cost.
7Your KPI report shows a median of 20 seconds to find, against a 95-second baseline. What must sit next to it in a report?
Answer: B. seconds_to_find exists only when a user chose a result. A high median gain with low search success can hide many failed searches.
8A customer asks you to run this nightly against their S/4HANA system with real data. Which change is not optional before go-live?
Answer: A. The demo identity and invented labels can't protect real data. The gate must use real authorization data, and the extraction route must fit SAP's API policy, confirmed with SAP.
Ask a question
Testing: only staff see this
Stuck on something in this layer? Ask it here. Questions are answered in the order they arrive, and the answer appears under My questions.
Sign in (free) to ask a question. You can ask anonymously.
Related topics
Picked from the terms, links and tags this topic shares with others.
Orchestrate Lab is a free practice ERP in your browser, built around the fictional Calder Pumpworks. These are the places in it where you do what this topic teaches.
Stages build on each other: a stage link opens the core path at that stage. To start there without doing the earlier stages yourself, use Do this stage for me on the stage before it. A chapter starts inside your own company once the core path is done (or the lab finishes the core for you first); drills and exercises start from their own prepared copy. Everything is saved in this browser only.
SAP HANA Cloud Vector Engine (SAP Learning)— REAL_VECTOR type; COSINE_SIMILARITY from -1 to 1; semantic and similarity search use cases; complements relational and full-text search
SAP HANA Cloud Vector Engine integration (LangChain documentation)— langchain-hana package with HanaDB; filter operators; specific_metadata_columns; create_hnsw_index; HanaInternalEmbeddings uses VECTOR_EMBEDDING and needs NLP enabled; reranking through CROSS_ENCODE
Python buildpack (Cloud Foundry documentation)— detected by requirements.txt; Python version in runtime.txt; no default start command (Procfile, cf push -c or manifest command); listen on PORT, 8080 expected
SAP API Policy v.4.2026a (help.sap.com)— only published APIs; per-API rate limits and preconditions for bulk extraction or replication; large-scale extraction or replication only through SAP-endorsed pathways; restriction on AI systems that plan, select or execute sequences of API calls