Working with Structured Data

Real-world AI datasets are messy. Documents have inconsistent encodings. JSON responses have missing fields. CSVs have mixed types in the same column. Before you can run any AI processing, you need to clean and normalize your data — and do it efficiently for large volumes.

JSON: The Universal AI Data Format

LLM APIs speak JSON. Vector databases return JSON. Your evaluation pipelines produce JSON. Mastering Python's JSON handling is non-negotiable:

import json
                      from pathlib import Path
                      from typing import Any
                      
                      # Parse JSON from various sources
                      json_string = '{"model": "gpt-4o", "tokens": 150}'
                      data: dict = json.loads(json_string)
                      
                      # Load from file
                      with open("results.json", encoding="utf-8") as f:
                          results: list[dict] = json.load(f)
                      
                      # Dump to string or file
                      output = json.dumps(data, indent=2, ensure_ascii=False)
                      Path("output.json").write_text(output, encoding="utf-8")
                      
                      
                      # Handling real-world messiness
                      def safe_parse_json(text: str) -> dict | None:
                          """Parse JSON with error recovery."""
                          try:
                              return json.loads(text)
                          except json.JSONDecodeError as e:
                              # LLMs sometimes wrap JSON in markdown code blocks
                              if "```json" in text:
                                  start = text.find("```json") + 7
                                  end = text.find("```", start)
                                  try:
                                      return json.loads(text[start:end].strip())
                              if "```" in text:
                                  start = text.find("```") + 3
                                  end = text.find("```", start)
                                  try:
                                      return json.loads(text[start:end].strip())
                              return None

CSV Processing for AI Datasets

CSV is still the lingua franca for training data, evaluation sets, and labeled examples. Python's built-in csv module reads rows as dictionaries keyed by header (DictReader), which is the safe way to handle columns without relying on position. For anything numeric or large you'd reach for pandas, but for row-by-row streaming of an AI dataset the standard library is lighter and memory-friendly.

import csv
                      from pathlib import Path
                      from dataclasses import dataclass
                      from typing import Iterator
                      
                      @dataclass
                      class EvalRecord:
                          prompt_id: str
                          prompt: str
                          expected_output: str
                          model: str
                          actual_output: str = ""
                          score: float = 0.0
                      
                      
                      def load_eval_dataset(csv_path: Path) -> list[EvalRecord]:
                          """Load an evaluation dataset from CSV, handling common issues."""
                          records = []
                      
                          with open(csv_path, encoding="utf-8-sig", newline="") as f:
                              # utf-8-sig handles BOM markers from Excel exports
                              reader = csv.DictReader(f)
                      
                              for row_num, row in enumerate(reader, start=2):  # 2 = first data row
                                  # Clean each field
                                  try:
                                      record = EvalRecord(
                                          prompt_id=row.get("prompt_id", f"auto_{row_num}").strip(),
                                          prompt=row["prompt"].strip(),
                                          expected_output=row.get("expected", "").strip(),
                                          model=row.get("model", "unknown").strip(),
                                      )
                                      if not record.prompt:
                                          continue  # skip empty rows
                                      records.append(record)
                                  except KeyError as e:
                                      print(f"Row {row_num}: missing required field {e} — skipping")
                      
                          print(f"Loaded {len(records)} evaluation records from {csv_path.name}")
                          return records
                      
                      
                      def stream_large_csv(csv_path: Path) -> Iterator[dict]:
                          """Stream a large CSV row-by-row without loading into memory."""
                          with open(csv_path, encoding="utf-8-sig", newline="") as f:
                              reader = csv.DictReader(f)
                              for row in reader:
                                  yield dict(row)  # yield one row at a time

Data Cleaning Patterns for AI Pipelines

Garbage in, garbage out is especially true for AI: messy text inflates token counts and confuses models. The cleaning steps below are the ones worth standardizing — normalizing whitespace, stripping control characters, collapsing duplicates, and handling missing fields — so every record entering your pipeline has a predictable shape.

import re
                      import unicodedata
                      from typing import Optional
                      
                      def clean_document_text(text: str) -> str:
                          """
                          Normalize text for consistent AI processing.
                          Common issues: BOM markers, control characters, excessive whitespace,
                          Unicode normalization inconsistencies.
                          """
                          if not text:
                              return ""
                      
                          # Normalize Unicode (NFC: composed form)
                          text = unicodedata.normalize("NFC", text)
                      
                          # Remove null bytes and control characters (except newlines and tabs)
                          text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)
                      
                          # Normalize line endings
                          text = text.replace("\r\n", "\n").replace("\r", "\n")
                      
                          # Collapse multiple blank lines to at most two
                          text = re.sub(r"\n{3,}", "\n\n", text)
                      
                          # Normalize multiple spaces (preserve newlines)
                          lines = [re.sub(r" {2,}", " ", line).strip() for line in text.split("\n")]
                          text = "\n".join(lines)
                      
                          return text.strip()
                      
                      
                      def extract_metadata(text: str) -> dict[str, Optional[str]]:
                          """Extract metadata patterns from document text."""
                          return {
                              "title": re.search(r"(?:title|heading):\s*(.+)", text, re.I)
                                     and re.search(r"(?:title|heading):\s*(.+)", text, re.I).group(1).strip(),
                              "date": re.search(r"\d{4}-\d{2}-\d{2}", text)
                                    and re.search(r"\d{4}-\d{2}-\d{2}", text).group(),
                              "word_count": str(len(text.split())),
                              "char_count": str(len(text)),
                          }
                      
                      
                      def chunk_text(
                          text: str,
                          chunk_size: int = 500,
                          chunk_overlap: int = 50,
                      ) -> list[str]:
                          """
                          Split text into overlapping chunks for RAG indexing.
                          Chunks on word boundaries to avoid mid-word splits.
                          """
                          words = text.split()
                          chunks = []
                          step = chunk_size - chunk_overlap
                      
                          for start in range(0, len(words), step):
                              chunk_words = words[start:start + chunk_size]
                              chunk = " ".join(chunk_words)
                              if chunk.strip():
                                  chunks.append(chunk)
                      
                              if start + chunk_size >= len(words):
                                  break
                      
                          return chunks

Pandas for AI Data Analysis

When you need aggregation, filtering, and analysis across large datasets, pandas is the tool:

import pandas as pd
                      import json
                      from pathlib import Path
                      
                      # Load evaluation results
                      results_df = pd.read_json("eval_results.jsonl", lines=True)
                      
                      # Basic exploration
                      print(results_df.shape)         # (rows, cols)
                      print(results_df.dtypes)        # column types
                      print(results_df.head())        # first 5 rows
                      print(results_df.describe())    # summary stats for numerics
                      
                      # Clean and filter
                      results_df["score"] = pd.to_numeric(results_df["score"], errors="coerce")
                      results_df = results_df.dropna(subset=["score", "prompt"])
                      results_df = results_df[results_df["score"] >= 0]  # remove negative scores
                      
                      # Analysis by model
                      model_stats = results_df.groupby("model").agg(
                          avg_score=("score", "mean"),
                          total_requests=("score", "count"),
                          p95_latency=("latency_ms", lambda x: x.quantile(0.95)),
                      )
                      print(model_stats.to_string())
                      
                      # Find worst-performing prompts
                      worst = (
                          results_df
                          .sort_values("score")
                          .head(20)[["prompt_id", "score", "model", "finish_reason"]]
                      )
                      
                      # Export for review
                      worst.to_csv("worst_performing.csv", index=False)
                      results_df.to_parquet("eval_results_clean.parquet", index=False)

Validation Pipeline

Cleaning fixes shape; validation enforces correctness. Running each cleaned record through a Pydantic model (from Lesson 1.2) gives you a single gate that rejects malformed rows with a clear reason and hands downstream code guaranteed-valid, typed objects. The pipeline below reads raw data, cleans it, validates it, and separates the good records from the rejects for review.

from pydantic import BaseModel, Field, ValidationError
                      from typing import Literal
                      
                      class RawEvalResult(BaseModel):
                          """Schema for raw results from an evaluation run."""
                          prompt_id: str
                          prompt: str = Field(min_length=1)
                          model: str
                          score: float = Field(ge=0.0, le=1.0)
                          output: str
                          tokens_used: int = Field(ge=0)
                          finish_reason: Literal["end_turn", "max_tokens", "stop_sequence", "error"]
                          latency_ms: float = Field(ge=0.0)
                      
                      
                      def validate_results(raw_records: list[dict]) -> tuple[list[RawEvalResult], list[dict]]:
                          """Validate a batch of results, separating valid from invalid."""
                          valid = []
                          invalid = []
                      
                          for record in raw_records:
                              try:
                                  valid.append(RawEvalResult.model_validate(record))
                              except ValidationError as e:
                                  invalid.append({"record": record, "errors": e.errors()})
                      
                          print(f"Valid: {len(valid)} | Invalid: {len(invalid)}")
                          return valid, invalid