Convert real unstructured or semi-structured text records into synthetic counterparts that preserve semantic intent and structure while replacing all personally identifiable, datable, and location-specific details. Uses Claude to intelligently rewrite records with explicit "preserve / change" rules.
-
Install Anthropic and dependencies:
pip install anthropic pandas tqdm numpy scikit-learn
-
Draft a transformation prompt with explicit rules:
You are anonymising {record_type} for research/testing purposes.
Read this real record:
{real_record_json}
Rewrite it as a synthetic record following these rules:
- PRESERVE: semantic intent, tone, structure, technical content, logical flow
- PRESERVE: {preservation_rules}
- CHANGE: all names, places, dates, email addresses, phone numbers, URLs, IDs
- CHANGE: specific quoted text, proper nouns, organizational names
- MAINTAIN: field schema, word count approximately
Return only valid JSON, no markdown.
-
Write a batch transformation script with progress and resume:
import json
import anthropic
import pandas as pd
from pathlib import Path
import time
def transform_records(input_path, output_path, record_type,
preserve_rules, transform_fields, locale="en"):
client = anthropic.Anthropic()
if input_path.endswith('.jsonl'):
with open(input_path) as f:
real_records = [json.loads(line) for line in f]
else:
df = pd.read_csv(input_path)
real_records = df.to_dict(orient='records')
completed = set()
if Path(output_path).exists():
with open(output_path) as f:
completed = {i for i, _ in enumerate(f)}
synthetic_records = []
for idx, real_record in enumerate(real_records):
if idx in completed:
continue
record_json = json.dumps(real_record, indent=2)
prompt = f"""Transform this {record_type} into a synthetic version:
{record_json}
Rules:
- PRESERVE semantic intent, tone, structure, technical details
- PRESERVE:
- CHANGE ALL: names, locations, dates, emails, phone numbers, IDs, URLs
- CHANGE: specific quoted text and proper nouns
- MAINTAIN: approximate length and field schema
Return ONLY valid JSON, no markdown or explanation."""
:
message = client.messages.create(
model=,
max_tokens=,
messages=[{: , : prompt}]
)
response_text = message.content[].text.strip()
response_text.startswith():
response_text = response_text.split()[].lstrip().strip()
synthetic_record = json.loads(response_text)
(output_path, ) f:
f.write(json.dumps(synthetic_record) + )
synthetic_records.append(synthetic_record)
(idx + ) % == :
()
time.sleep()
json.JSONDecodeError e:
()
anthropic.APIError e:
()
time.sleep()
()
synthetic_records
__name__ == :
transform_records(
input_path=,
output_path=,
record_type=,
preserve_rules=,
transform_fields=[, , ]
)
-
Optional: QA check for leakage (flag records too similar to source):
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
def check_leakage(real_path, synth_path, threshold=0.75):
with open(real_path) as f:
real_records = [json.loads(line) for line in f]
with open(synth_path) as f:
synth_records = [json.loads(line) for line in f]
real_texts = [' '.join(str(v) for v in r.values()) for r in real_records]
synth_texts = [' '.join(str(v) for v in r.values()) for r in synth_records]
vectorizer = TfidfVectorizer()
all_texts = real_texts + synth_texts
tfidf = vectorizer.fit_transform(all_texts)
flagged = []
for i, synth_idx in enumerate(range(len(real_texts), len(all_texts))):
similarity = cosine_similarity(tfidf[synth_idx], tfidf[:len(real_texts)])
max_sim = np.max(similarity)
if max_sim > threshold:
flagged.append({
: i,
: (max_sim),
: (np.argmax(similarity))
})
flagged:
()
f flagged[:]:
()
:
()
flagged
-
Run transformation and QA:
python transform_real_to_synth.py
python check_leakage.py
