LLM Pipeline

After scraping and storage, an optional LLM pipeline processes the data for semantic search. It converts raw scraped items into chunked, embedded documents stored in pgvector for similarity search.

graph LR
    DS[Dataset<br/>(Parquet / S3)] --> CH[IntelligentChunker]
    CH -->|"text chunks + metadata"| EM[Embedder]
    EM -->|"384-dim vectors"| VS[Vector Store<br/>(pgvector)]
    VS --> API[Search API]
    VS --> RR[Reranker]
    RR --> API

Running the Pipeline

# Process a completed scrape job into LLM-ready chunks
make llm-process JOB=my-run PROFILE=idealista

# Check pipeline status
make llm-status JOB=my-run

# Process only a specific stage
make llm-process JOB=my-run PROFILE=idealista STAGE=detail

# Re-process from scratch (clear prior output)
make llm-process JOB=my-run PROFILE=idealista FRESH=1

Pipeline Modules

Module	File	Responsibility
Chunker	llm/chunker.py	Splits fields into chunks with configurable strategies
Embedder	llm/embedder.py	Generates 384-dim vectors via sentence-transformers
Vector Store	llm/vector_store.py	Stores/retrieves vectors from pgvector (HNSW index)
Reranker	llm/reranker.py	Cross-encoder second-pass scoring for search results
Pipeline	llm/pipeline.py	Orchestrates chunk → embed → store
API	llm/api.py	FastAPI server for semantic search

LLM Configuration Model

The pipeline's behavior is driven entirely by YAML configuration in the site profile's llm: section. The Python-side models live in core/llm_config.py:

FieldLLMConfig — Per-Field Settings

Each scraped field can be individually configured:

Parameter	Type	Default	Description
chunk	bool	false	If true, split this field's value into chunks
chunk_strategy	"semantic" | "heading" | "per_item" | "grouped" | "token"	"semantic"	Algorithm to use
chunk_size	int	1500	Target chunk size in characters
chunk_overlap	int	200	Character overlap between adjacent chunks
max_tokens	int	800	Max tokens (token strategy only)
overlap_tokens	int	50	Token overlap (token strategy only)
items_per_chunk	int	3	Items per chunk (grouped strategy only)
max_items	int | None	None	Cap items from list (per_item strategy only)
template	str	"{item}"	Python format string for per_item/grouped
include_in_context	bool	false	Attach value as metadata to every chunk in the row
clean	bool	false	Strip excess whitespace before chunking
importance	float	0.5	Priority weight (0.0–1.0) for mixed-type queries
content_type	str	""	Semantic label for this content (stored in chunk metadata)

LLMConfig — Top-Level Pipeline Settings

llm:
  enabled: true
  stage: detail                       # which stage's Parquet to process
  compute_profile: booking_com        # derived-field compute function
  embedding_model: "BAAI/bge-small-en-v1.5"
  batch_size: 32
  fields:
    - name: title
      type: string
      llm_config:
        chunk: false
        include_in_context: true
        importance: 1.0
    - name: description
      type: text
      llm_config:
        chunk: true
        chunk_strategy: semantic
        chunk_size: 1500
        chunk_overlap: 200
        importance: 0.8

Chunking Strategies — Full Reference

Located in llm/chunker.py — all strategies are pure Python with zero heavy NLP dependencies. Each strategy is a standalone function registered in the STRATEGIES dict, making it trivial to add custom strategies.

STRATEGIES = {
    "semantic":   semantic_chunk,
    "heading":    heading_chunk,
    "per_item":   per_item_chunk,
    "grouped":    grouped_chunk,
    "token":      token_aware_chunk,
}

---

1. Semantic Chunking (semantic_chunk)

Splits text at the most natural boundary found within the target window, preferring paragraph → newline → sentence → word in that order.

def semantic_chunk(
    text: str,
    chunk_size: int = 1500,
    chunk_overlap: int = 200,
) -> list[str]:
    separators = ["\\n\\n", "\\n", ". ", " ", ""]

    if len(text) <= chunk_size:
        return [text] if text.strip() else []

    chunks: list[str] = []
    start = 0

    while start < len(text):
        end = min(start + chunk_size, len(text))

        if end < len(text):
            # Find best split in the last 25% of the window
            search_start = max(start + int(chunk_size * 0.75), start)
            best = end
            for sep in separators:
                pos = text.rfind(sep, search_start, end + len(sep) if sep else end)
                if pos != -1:
                    best = pos + len(sep) if sep else pos
                    break
            end = best

        chunk = text[start:end].strip()
        if chunk:
            chunks.append(chunk)

        start = end - chunk_overlap if end < len(text) else end

    return chunks

Best for: Long-form descriptive text (property descriptions, product details, review content).

Profile config example:

- name: description
  type: text
  llm_config:
    chunk: true
    chunk_strategy: semantic
    chunk_size: 1200
    chunk_overlap: 150
    content_type: description
    importance: 0.8
    clean: true

---

2. Heading Chunking (heading_chunk)

Splits markdown-formatted text at # / ## / ### headings, preserving the heading hierarchy as a heading_path breadcrumb (e.g. "Overview > Features > Pool") on each chunk.

def heading_chunk(markdown_text: str) -> list[dict]:
    heading_pattern = re.compile(r"^(#{1,6})\\s+(.+)$", re.MULTILINE)
    matches = list(heading_pattern.finditer(markdown_text))

    if not matches:
        return [{"content": markdown_text.strip(), "heading_path": ""}]

    chunks: list[dict] = []
    current_path: list[str] = []

    for i, match in enumerate(matches):
        level = len(match.group(1))
        title = match.group(2).strip()

        # Maintain heading path stack
        while current_path and len(current_path) >= level:
            current_path.pop()
        current_path.append(title)
        heading_path = " > ".join(current_path)

        # Content from this heading to the next (or EOF)
        content_start = match.end()
        content_end = (
            matches[i + 1].start() if i + 1 < len(matches) else len(markdown_text)
        )
        content = markdown_text[content_start:content_end].strip()

        if content:
            chunks.append({"content": content, "heading_path": heading_path})

    return chunks

Unlike other strategies, heading_chunk returns list[dict] (not list[str]) — each dict has content and heading_path keys. The IntelligentChunker handles this transparently.

Best for: Structured documentation, product specifications, any markdown content.

---

3. Per-Item Chunking (per_item_chunk)

Each list element becomes its own independent chunk.

def per_item_chunk(
    items: list[Any],
    template: str = "{item}",
    max_items: int | None = None,
) -> list[str]:
    chunks: list[str] = []
    for item in items[:max_items] if max_items else items:
        text = template.format(item=str(item))
        if text.strip():
            chunks.append(text)
    return chunks

Best for: Lists of reviews, product attributes, room names — where each item should be findable independently.

Profile config example (room names as individual chunks):

- name: room_names
  type: array<string>
  llm_config:
    chunk: true
    chunk_strategy: per_item
    template: "Room: {item}"
    content_type: room
    importance: 0.6

---

4. Grouped Chunking (grouped_chunk)

Groups multiple list elements into fixed-size batches.

def grouped_chunk(
    items: list[Any],
    items_per_chunk: int = 3,
    template: str | None = None,
) -> list[str]:
    chunks: list[str] = []
    for i in range(0, len(items), items_per_chunk):
        group = items[i : i + items_per_chunk]
        if template:
            group_text = template.format(
                items="\\n---\\n".join(str(item) for item in group)
            )
        else:
            group_text = "\\n---\\n".join(str(item) for item in group)
        if group_text.strip():
            chunks.append(group_text)
    return chunks

Best for: Dense attribute lists (facilities, amenities, features) where individual items are too small to chunk alone.

Profile config example (facilities grouped in batches of 10):

- name: facility_names
  type: array<string>
  llm_config:
    chunk: true
    chunk_strategy: grouped
    items_per_chunk: 10
    template: "Facilities: {items}"
    content_type: facilities
    importance: 0.5

---

5. Token-Aware Chunking (token_aware_chunk)

Splits by actual token count using tiktoken (cl100k_base encoding), with a fallback to semantic_chunk if tiktoken isn't installed.

def token_aware_chunk(
    text: str,
    max_tokens: int = 800,
    overlap_tokens: int = 50,
) -> list[str]:
    try:
        import tiktoken
        enc = tiktoken.get_encoding("cl100k_base")
    except ImportError:
        # Fallback: ~4 chars ≈ 1 token
        return semantic_chunk(text, chunk_size=max_tokens * 4,
                              chunk_overlap=overlap_tokens * 4)

    tokens = enc.encode(text)
    chunks: list[str] = []
    start = 0

    while start < len(tokens):
        end = min(start + max_tokens, len(tokens))
        chunks.append(enc.decode(tokens[start:end]))
        start = end - overlap_tokens

    return chunks

Best for: Content destined for LLM context windows with strict token limits.

Profile config example:

- name: long_review
  type: text
  llm_config:
    chunk: true
    chunk_strategy: token
    max_tokens: 800
    overlap_tokens: 50
    content_type: review

---

IntelligentChunker — Row-by-Row Processing

The IntelligentChunker class in llm/chunker.py orchestrates all strategies per row. Here's the complete processing flow for a single Parquet row:

flowchart TD
    A["Row dict from Parquet"] --> B[Iterate over field configs]
    B --> C{field.chunk?}
    C -->|"Yes"| D["Look up strategy_fn<br/>from STRATEGIES dict"]
    C -->|"No"| E{include_in_context?}
    E -->|"Yes"| F["Add to collected_metadata"]
    E -->|"No"| G[Skip field]

    D --> H["Run strategy_fn(value)"]
    H --> I["For each raw chunk:"]
    I --> J["Build chunk dict with<br/>content, chunk_type,<br/>importance, chunk_hash"]
    J --> K["Append to all_chunks"]

    K --> L[Compute derived fields<br/>via compute_row(profile)]
    L --> M[Build context_prefix<br/>from has_* booleans]
    M --> N["Augment each chunk:<br/>prefix content, attach metadata,<br/>set source_url"]
    N --> O["Return list[chunk_dict]<br/>ready for embedding"]

Key Behaviors

1. Context-only fields — fields with chunk: false + include_in_context: true are serialized as JSON and attached to every chunk in the row as metadata.

2. List-aware parsing — Parquet stores lists as JSON strings. The chunker auto-detects this and parses them back before passing to per_item / grouped strategies.

3. Derived fields — After chunking, the profile's registered compute_row function (in core/llm_compute.py) is called to produce has_pool, has_garden, etc.

4. Context prefix — Derived has_* booleans are converted to human-readable prefixes that are prepended to chunk content before hashing and embedding:

Has pool. Has parking. Has garden. Rating: 4.5 (120 reviews). Name: Villa Madrid.
Beautiful 3-bedroom villa with private pool in the heart of...

This gives the embedding model semantic context about the item's features without bloating the chunk size.

1. Dedup-safe hashing — Each chunk gets an MD5 hash of its content for dedup (UNIQUE (source_url, chunk_hash) in Postgres), so reprocessing a job skips already-stored chunks.

---

LLM Field Presets

Defined in core/llm_presets.py — profiles reference presets by name instead of repeating full field definitions. Presets are expanded at profile load time; custom fields with the same name override preset fields.

llm:
  fields:
    - preset: ecommerce_product_v1
    - name: custom_field      # overrides preset field with same name
      type: string

Preset Registry

PRESETS = {
    "ecommerce_product_v1":     ECOMMERCE_PRODUCT_V1,
    "review_v1":                REVIEW_V1,
    "hotel_listing_v1":         HOTEL_LISTING_V1,
    "marketplace_listing_v1":   MARKETPLACE_LISTING_V1,
    "real_estate_listing_v1":   REAL_ESTATE_LISTING_V1,
}

The resolve_presets() function expands all preset references into concrete field dicts, then applies user overrides:

def resolve_presets(fields: list[dict]) -> list[dict]:
    resolved: dict[str, dict] = {}
    for entry in fields:
        preset_name = entry.pop("preset", None)
        if preset_name:
            for pf in PRESETS.get(preset_name, []):
                resolved.setdefault(pf["name"], pf)
        if "name" in entry:
            resolved[entry["name"]] = entry  # user override wins
    return list(resolved.values())

---

Preset: ecommerce_product_v1

Covers luxury fashion product detail pages (e.g. Hermès). Designed for:

• Identity fields attached as metadata: name, sku, price
• Description — semantic chunks at 1200 chars with 150 overlap
• Product detail — semantic chunks at 1000 chars with 100 overlap
• Care instructions — semantic chunks at 800 chars with 80 overlap
• Attributes as JSON context (not chunked): material, gender, color, categories, made_in
• Logistics as JSON context: stock_info, shipping_delay, lead_time

# Fields: ecommerce_product_v1 (14 fields)
# Identity (set include_in_context=True, chunk=False)
name           type:string  importance:1.0
sku            type:string
price          type:float

# Chunked text (chunk=True, strategy=semantic)
description    chunk_size:1200  overlap:150  type:description  importance:0.8
product_detail chunk_size:1000  overlap:100  type:product_detail  importance:0.7
care_instruction chunk_size:800  overlap:80   type:care_instruction  importance:0.5

# Context-only (include_in_context=True, chunk=False)
material       type:json  importance:0.6
gender         type:json
color          type:json
categories     type:json
made_in        type:string
stock_info     type:json
shipping_delay type:json
lead_time      type:json

---

Preset: review_v1

Covers product/service review pages (G2, Trustpilot-style). Designed for:

• Product identity as metadata: name, category, rating, review_count
• Description — semantic chunk at 1200 chars
• Pros/cons as JSON context (structured, not chunked)
• Like/dislike text — semantic chunks at 800 chars
• Per-review metadata: review_rating, review_author

# Fields: review_v1 (10 fields)
name           type:string  importance:1.0      # product name
category       type:string
rating         type:float                        # aggregate rating
review_count   type:int
description    chunk:1200  overlap:150  type:description  importance:0.8
pros           type:json   importance:0.7         # context-only
cons           type:json   importance:0.7         # context-only
likes          chunk:800   type:likes             # individual review likes
dislikes       chunk:800   type:dislikes          # individual review dislikes
review_rating  type:float                         # per-review rating
review_author  type:string                        # per-review author

---

Preset: hotel_listing_v1

Covers Booking.com-style hotel detail pages. Covers:

• Identity & location: property_name (importance 1.0), city, country_code, formatted_address, star_rating
• Reviews: review_score, review_count, per-category breakdowns (staff, facilities, cleanliness, comfort, value, location)
• Descriptions: description (1500/200), fine_print (1000/100), review_summary (1200/150) — all semantic chunks
• Facilities: grouped chunk (10 per chunk) via "Facilities: {items}" template
• Rooms: per-item chunk via "Room: {item}" template
• House rules: checkin_from, checkout_until, pets_allowed
• Breakfast: breakfast_style, breakfast_price
• Company info: company_name, company_address

# Fields: hotel_listing_v1 (25 fields)
property_name      type:string   importance:1.0
city               type:string
country_code       type:string
formatted_address  type:string
star_rating        type:float
review_score       type:float
review_count       type:int
description        chunk:1500  overlap:200  type:description  importance:0.8
fine_print         chunk:1000  overlap:100  type:fine_print   importance:0.3
review_summary     chunk:1200  overlap:150  type:review_summary  importance:0.7
facility_names     chunk:grouped  items_per_chunk:10  template:"Facilities: {items}"  importance:0.5
room_names         chunk:per_item  template:"Room: {item}"  importance:0.6
room_details       chunk:1000  overlap:100  type:room_detail  importance:0.6
checkin_from       type:string          # context-only
checkout_until     type:string          # context-only
pets_allowed       type:bool            # context-only
company_name       type:string          # context-only
company_address    type:string          # context-only
review_scores_staff       type:float    # context-only
review_scores_facilities  type:float    # context-only
review_scores_cleanliness type:float    # context-only
review_scores_comfort     type:float    # context-only
review_scores_value       type:float    # context-only
review_scores_location    type:float    # context-only
breakfast_style    type:string          # context-only
breakfast_price    type:float           # context-only

---

Preset: marketplace_listing_v1

Covers Facebook Marketplace-style classified listings:

• Identity: title (importance 1.0), price, city, condition, category_id, inventory_type, listing_status
• Description — semantic chunk at 1200/150
• Attributes — per_item chunk via "{item}" template
• Status booleans: is_sold, is_shipping_offered, creation_time

# Fields: marketplace_listing_v1 (11 fields)
title             type:string  importance:1.0
price             type:string
city              type:string
condition         type:string
category_id       type:string
inventory_type    type:string
listing_status    type:string
description       chunk:1200  overlap:150  type:description  importance:0.8
attribute_data    chunk:per_item  template:"{item}"  type:attribute  importance:0.4
is_sold           type:bool
is_shipping_offered type:bool
creation_time     type:int

---

Preset: real_estate_listing_v1

Covers Idealista-style property detail pages:

• Identity & pricing: title (1.0), location, price (0.9), price_per_m2
• Key attributes: area_m2 (0.8), bedrooms (0.8), bathrooms, floor
• Description — semantic chunk at 1200/150
• Features & amenities — grouped chunks (5 per chunk)
• Advertiser: advertiser_name, is_professional
• Boolean flags: has_lift, has_terrace, has_pool, has_parking, has_aircon
• Energy: energy_consumption

# Fields: real_estate_listing_v1 (18 fields)
title              type:string  importance:1.0
location           type:string
price              type:string  importance:0.9
price_per_m2       type:string
area_m2            type:int     importance:0.8
bedrooms           type:int     importance:0.8
bathrooms          type:int
floor              type:int
description        chunk:1200  overlap:150  type:description  importance:0.8
features           chunk:grouped  items_per_chunk:5  template:"Features: {items}"  importance:0.6
amenities          chunk:grouped  items_per_chunk:5  template:"Amenities: {items}"  importance:0.5
advertiser_name    type:string
is_professional    type:bool
has_lift           type:bool
has_terrace        type:bool
has_pool           type:bool
has_parking        type:bool
has_aircon         type:bool
energy_consumption type:string

---

Derived Fields (Compute Functions)

After chunking, the IntelligentChunker calls a registered compute function to produce derived fields from raw scraped data. These are attached to chunk metadata and used to build the context prefix.

Registered in core/llm_compute.py via the @register_compute decorator:

@register_compute("booking_com")
def _compute_booking(row: dict) -> dict:
    facilities = json.loads(row.get("facility_names", "[]"))
    fac_lower = [f.lower() for f in facilities]
    return {
        "has_pool":       any("pool" in f or "swim" in f for f in fac_lower),
        "has_spa":        any("spa" in f for f in fac_lower),
        "has_fitness":    any("fitness" in f or "gym" in f for f in fac_lower),
        "has_parking":    any("parking" in f for f in fac_lower),
        "has_restaurant": any("restaurant" in f for f in fac_lower),
        "has_wifi":       any("internet" in f or "wifi" in f for f in fac_lower),
        "star_rating":    int(row.get("star_rating", 0)),
        "review_score":   f"{int(row.get('review_score', 0))}/10",
    }

Registered Compute Functions

Profile	Function	Derived Fields
booking_com	_compute_booking	has_pool, has_spa, has_fitness, has_parking, has_restaurant, has_wifi, has_breakfast, has_free_breakfast, star_rating, review_score, city, property_name
facebook_marketplace	_compute_facebook_marketplace	listing_title, listing_price, city, item_condition, category (from Redis), is_sold, has_shipping, inventory_type, listed_date
trustpilot	_compute_trustpilot	business_name, business_stars, review_rating, reviewer, review_title, is_recommended, has_reply
g2	_compute_g2	has_pros, has_cons, reviewer
hermes	_compute_hermes	price (normalized), in_stock_ecom, in_stock_retail, color_variant_count, size_variant_count, lead_time_display
idealista	_compute_idealista	price_numeric, area_m2, price_per_m2_computed, has_phone, city, variant

The compute_row(profile_name, row) function dispatches to the correct compute function by profile name (matched to the compute_profile setting in llm: config):

def compute_row(profile_name: str, row: dict) -> dict:
    fn = _registry.get(profile_name)
    if fn is None:
        return {}
    return fn(row)

Embedding

Uses BAAI/bge-small-en-v1.5 via sentence-transformers:

embedder = Embedder("BAAI/bge-small-en-v1.5", device="cpu")
vectors = embedder.encode(["text one", "text two"])
# Each vector: np.ndarray(384,) float32, L2-normalized

• 384 dimensions — compact enough for fast HNSW search
• CPU-only — PyTorch is installed with CPU support to avoid ~2.5 GB CUDA deps
• Model baked into Docker image — pre-downloaded at build time
• Batched — processes in mini-batches of 32

Vector Store

Uses pgvector with a HNSW index for approximate nearest-neighbor search:

CREATE TABLE llm_documents (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    job_id          TEXT NOT NULL,
    site            TEXT NOT NULL,
    source_url      TEXT NOT NULL,
    chunk_hash      TEXT NOT NULL,
    chunk_type      TEXT NOT NULL,
    content         TEXT NOT NULL,
    embedding       vector(384),
    metadata        JSONB DEFAULT '{}',
    importance      REAL DEFAULT 0.5,
    ...
);

-- HNSW index for fast ANN search
CREATE INDEX idx_llm_documents_embedding
    ON llm_documents USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

Reranker

Uses cross-encoder/ms-marco-MiniLM-L-6-v2 for second-pass scoring:

reranker = Reranker()
scores = reranker.score(query, candidate_chunks)
# Returns relevance scores (0–1) for each candidate

The reranker is applied after the initial vector search to refine results. It's more accurate than cosine similarity but slower, so it's used on the top-N candidates only.