Sourcing and scoring

Every morning, crawl new listings from several sources, keep the ones worth a look, price them and put the profitable ones in front of an analyst. This is Millstream's reference pipeline and its end-to-end acceptance test: real estate renovation deals, though the shape fits any sourcing work.

The pipeline

flowchart LR
  T(["daily, 06:00"]) --> A[crawl source A] & B[crawl source B]
  A & B --> D{{dedupe}} --> S["score (agent)"] --> F1{{"score at least 70"}}
  F1 --> R[rentability] --> F2{{"margin at least 15%"}} --> V[/analyst review/] --> P[PDF report]

pipeline: property-sourcing
description: Daily sourcing of renovation opportunities

vars: { city: Lyon, maxPrice: 400000, minScore: 70, minMarginPct: 15 }
secrets: [sourceAToken]

triggers:
  daily: { cron: "0 6 * * *", tz: Europe/Paris }

streams:
  listings.raw: { retention: 90d, redact: [/seller/phone] }

steps:
  crawl-source-a:
    action: lab/crawl-source-a@1.2
    in: trigger.daily
    out: listings.raw
    config:
      filters: { city: "${vars.city}", maxPrice: "${vars.maxPrice}" }
      token: "${secrets.sourceAToken}"
  crawl-source-b:
    action: lab/crawl-source-b@0.9
    in: trigger.daily
    out: listings.raw
    config: { filters: { city: "${vars.city}", maxPrice: "${vars.maxPrice}" } }
  dedupe:
    action: dedupe
    in: listings.raw
    out: listings.unique
    config: { key: "item.addressNorm + ':' + string(item.surfaceM2)", window: 30d }
  score:
    action: lab/score-listing@2
    in: listings.unique
    out: listings.scored
    concurrency: 8
  keep-good:
    action: filter
    in: listings.scored
    out: listings.shortlist
    config: { expr: "item.score >= vars.minScore" }
  rentability:
    action: ./actions/rentability
    in: listings.shortlist
    out: listings.priced
  keep-profitable:
    action: filter
    in: listings.priced
    out: listings.selected
    config: { expr: "item.marginPct >= vars.minMarginPct" }
  analyst-review:
    action: review
    in: listings.selected
    out: listings.approved
    config: { inbox: analysts, timeout: 72h, onTimeout: reject, editable: [/marginPct, /notes] }
  report:
    action: lab/pdf-report@1
    in: listings.approved
    out: reports.pdf

budget: { runUsd: 25 }

What Millstream does for it

Streams and steps

• Both crawlers write listings.raw, so their output merges with no join step.
• The dedupe step was added later, between crawling and scoring, without touching another step; millstream replay --step dedupe --since 7d ran past items through it.
• Thresholds live in filter steps, so the scoring action stays reusable and every threshold is visible in the file.

The agent

• score runs in a sandbox with its declared network destinations only, and its answer is checked against the action's output schema.
• Each attempt's transcript is stored with the delivery, and its cost is recorded per model.
• budget.runUsd: 25 holds back agent deliveries once a run has spent it, and pauses the agent steps until someone resumes them.

The analyst

• The inbox shows each selected listing; the analyst may edit /marginPct and /notes before approving.
• A listing nobody decides within 72 hours is rejected, with reason timeout.
• Each decision records who, when, the reason and the diff of any edit.

Personal data

• Raw listings keep their payloads for 90 days.
• Evidence bundles exported for development replace the seller's phone number with a keyed hash.

How it learns

The M5 exit scenario ran this pipeline's fixtures through the whole loop, offline and deterministic. Analysts lowered the scores of poor listings; the insights projector found the pattern, and the improve pipeline drafted a lessons file for score. Its backtest on 46 sampled deliveries raised agreement with the reviewers from 0.58 to 1.00. In a second fixture, an agent step whose answers followed a rule was distilled into Go code: on 270 holdout items it covered 260 with agreement 1.000, and the next day cost 0.008385 USD against 0.227307 USD on the agent alone, 3.7%. A person accepted both proposals before they applied.

Figures to watch

Next

Pipelines · Self-improvement · Getting started