Short answer: for real-estate listing discovery, use a staged retrieval architecture with explicit collections, bounded queries, and source context that survives every hop from a listing record to the generated answer. Define the retrieval unit, metadata filters, and freshness target before choosing a service. Then test the contract on a small labeled set and re-index only changed records while deleting withdrawn ones.
For an internal knowledge-base bot that helps a property team discover listings, this order matters more than the vector database logo. A notebook can make five hand-picked questions look persuasive while hiding the expensive part: what gets embedded again when a price, status, or description changes. The production design has to keep the answer traceable without turning every feed update into a full rebuild.
The first instinct is often one listing per document, one embedding per document, and top_k=10. It is pleasantly simple. It also leaves three decisions implicit: which version of the listing is searchable, whether a long description can crowd out structured facts, and how a removed property disappears from results. Make those decisions part of the schema instead.
How should schema design shape retrieval architecture for real-estate listing discovery?
Start with an answer contract, not an embedding model. A useful answer for this bot might need a listing ID, a short matching passage, the current status, the feed timestamp, and enough source context for a reviewer to verify the result. That contract determines the retrieval unit. If the answer must explain why a property matched “two-bedroom homes near transit with a recent price reduction,” the unit should preserve both the prose that supports the match and the structured fields that bound it.
A practical schema separates stable identity from changing content. Keep listing_id as the durable key, version as the revision marker, chunk_id as the retrieval identity, and source_updated_at as the freshness evidence. Put exact constraints such as city, property type, bedroom count, status, and feed name in metadata. Keep descriptive language in the text being embedded. The split is deliberate: metadata filters establish eligibility; semantic retrieval ranks the eligible passages.
Don't encode every field into one prose blob. “Bedrooms: 2” may be readable, but a bounded numeric or categorical filter is easier to reason about than hoping similarity will enforce an exact constraint. The reverse mistake is just as awkward: reducing the listing to metadata removes the language that lets retrieval connect “close to a station” with a description about a six-minute walk to a named stop.
There is no universal chunk size here. I'm not sure what boundary will win for your feed because description length and question mix are not specified; a labeled evaluation set resolves that uncertainty. Start with one chunk for concise listings and deterministic section chunks for long descriptions, then compare recall and citation correctness rather than guessing from token counts.
Make freshness an index operation, not a prompt instruction
Freshness needs a concrete state transition. When the source revision changes, create the new chunks, upsert them under the listing's current version, and remove chunk IDs that belonged to the previous version but are no longer present. When a source record is deleted or withdrawn, remove its indexed records from the collection. A prompt saying “only show active listings” cannot repair a stale index if the stale passage is still eligible for retrieval. This is where index cost becomes controllable. Compute a content fingerprint from only the fields that affect retrieval. A change to an internal accounting field shouldn't trigger new embeddings; a change to status, price text, location text, or description should. The update planner can therefore classify each feed event as unchanged, upsert, or delete before any embedding work begins. Full rebuilds remain useful for schema migrations, but they stop being the default response to ordinary feed churn. Be strict about collection boundaries too. A collection is an operational boundary for compatible schema and lifecycle rules, not a folder name invented by each notebook. Mixing development fixtures, live listings, and archived records makes freshness tests ambiguous and can inflate the candidate pool. Separate them, document the rule, and make every query name its intended collection. For a concrete change batch, imagine 10,000 source records where most content fingerprints are unchanged, a smaller set has new descriptions, and several IDs have vanished from the feed. The planner should leave unchanged chunks alone, send only changed chunks through embedding and upsert, and issue explicit deletes for vanished IDs. That example is not a measured ratio; it is a test fixture designed to prove that the code branches on content state rather than blindly rebuilding everything.
One sharp edge deserves an explicit test: a withdrawn listing must be absent, not merely ranked lower.
How can a Python harness expose costly retrieval mistakes?
Read the live request schema before building the adapter. This runnable Python example fetches the public discovery manifest, finds the capability whose verified path is /v1/vector/query, and prints its method and request parameters. It uses an environment key when one is present, sends an explicit method, honors Retry-After on HTTP 429, and surfaces other 4xx responses instead of pretending every response is successful.
import json
import os
import time
import urllib.error
import urllib.request
BASE_URL = os.environ["INFRAI_BASE_URL"].rstrip("/")
DISCOVERY_URL = f"{BASE_URL}/discovery"
TARGET_PATH = "/v1/vector/query"
def retry_delay(headers: object, attempt: int) -> float:
retry_after = headers.get("Retry-After") if headers else None
if retry_after and retry_after.isdigit():
return float(retry_after)
return float(2**attempt)
def load_manifest(max_attempts: int = 4) -> dict:
headers = {"Accept": "application/json"}
api_key = os.environ.get("INFRAI_API_KEY")
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
for attempt in range(max_attempts):
request = urllib.request.Request(
DISCOVERY_URL,
headers=headers,
method="GET",
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.load(response)
except urllib.error.HTTPError as error:
if error.code == 429 and attempt + 1 < max_attempts:
time.sleep(retry_delay(error.headers, attempt))
continue
detail = error.read().decode("utf-8", errors="replace")
raise RuntimeError(f"Discovery request failed ({error.code}): {detail}") from error
raise RuntimeError("Discovery retry budget exhausted")
def find_query_capability(manifest: dict) -> dict:
for capability in manifest["capabilities"]:
if capability["path"] == TARGET_PATH and capability["method"] == "POST":
return capability
raise LookupError(f"No discovered capability for POST {TARGET_PATH}")
if __name__ == "__main__":
capability = find_query_capability(load_manifest())
print(json.dumps({
"id": capability["id"],
"method": capability["method"],
"path": capability["path"],
"params": capability["params"],
}, indent=2))
The manifest is public, so authentication is optional for this read; keeping the environment-based header in the adapter preserves the authentication shape used by protected capabilities. Set INFRAI_BASE_URL to the platform's versioned API base before running it. The next step is to validate an adapter request against the returned params schema. Don't guess the fields.
The harness below stays local on purpose. It checks the retrieval contract that any backend must satisfy after its adapter has been built from the discovered schema.
from dataclasses import dataclass
from typing import Callable, Iterable
@dataclass(frozen=True)
class ListingChunk:
chunk_id: str
listing_id: str
version: int
status: str
city: str
source_updated_at: str
text: str
@dataclass(frozen=True)
class EvalCase:
query: str
expected_listing_ids: frozenset[str]
forbidden_listing_ids: frozenset[str]
city: str
top_k: int = 8
def evaluate(
cases: Iterable[EvalCase],
search: Callable[[str, str, int], list[ListingChunk]],
) -> dict[str, float]:
total = 0
hits = 0
forbidden_hits = 0
traceable_hits = 0
for case in cases:
results = search(case.query, case.city, case.top_k)
returned = {item.listing_id for item in results}
total += 1
hits += bool(returned & case.expected_listing_ids)
forbidden_hits += bool(returned & case.forbidden_listing_ids)
traceable_hits += all(
item.chunk_id
and item.listing_id
and item.source_updated_at
and item.status == "active"
for item in results
)
denominator = max(total, 1)
return {
"case_recall": hits / denominator,
"forbidden_case_rate": forbidden_hits / denominator,
"traceable_case_rate": traceable_hits / denominator,
}
INDEX = [
ListingChunk(
chunk_id="L-104:v3:overview",
listing_id="L-104",
version=3,
status="active",
city="Portland",
source_updated_at="2026-09-02T14:30:00Z",
text="Two-bedroom condo, six-minute walk from Hawthorne transit stop.",
),
ListingChunk(
chunk_id="L-220:v5:overview",
listing_id="L-220",
version=5,
status="active",
city="Portland",
source_updated_at="2026-09-03T08:15:00Z",
text="Three-bedroom townhouse with an attached garage.",
),
]
def local_search(query: str, city: str, top_k: int) -> list[ListingChunk]:
terms = set(query.lower().split())
eligible = [item for item in INDEX if item.city == city and item.status == "active"]
ranked = sorted(
eligible,
key=lambda item: len(terms & set(item.text.lower().split())),
reverse=True,
)
return ranked[:top_k]
CASES = [
EvalCase(
query="two-bedroom home near transit",
expected_listing_ids=frozenset({"L-104"}),
forbidden_listing_ids=frozenset({"L-099"}),
city="Portland",
)
]
if __name__ == "__main__":
scores = evaluate(CASES, local_search)
assert scores["forbidden_case_rate"] == 0.0
assert scores["traceable_case_rate"] == 1.0
print(scores)
The toy ranker is not a recommendation; it makes the evaluation interface executable without smuggling in an unverified service payload. In production, keep the same cases and swap adapters. That notebook-to-prod move is valuable because the assertions remain stable while the search implementation changes.
Begin with perhaps 30 to 50 labeled questions if that is enough to represent the actual intents, but treat the count as a starting hypothesis, not a benchmark. Include exact-filter cases, semantic phrasing cases, changed-listing cases, and deletion cases. Record the retrieved listing IDs and source timestamps alongside answer quality. A generation score alone can look healthy even when retrieval cites an old version.
Measure case recall, forbidden-result rate, traceable-result rate, number of records re-indexed per feed run, and prompt tokens passed downstream. I would block rollout on any withdrawn record appearing in results. Other thresholds depend on the team's risk tolerance and question mix, so your mileage may vary.
Choose the backend after the contract is testable
Once the adapter boundary and evaluation set exist, backend selection gets less theatrical. Run the same bounded queries and the same update batch against each candidate. Compare operational fit, evidence returned to the application, and how much integration-specific code sits outside the adapter. Don't crown a winner from a single relevance demo.
| Option | Why it belongs in the test | The catch |
|---|---|---|
| PostgreSQL with pgvector | It keeps vector retrieval close to relational listing metadata. | Stick with it when the team wants one database boundary; test index operations at the intended scale rather than assuming the fit. |
| Pinecone | It is a real managed vector-database option with its own documented data-plane API. | It is not suitable when adding a separate vector service conflicts with the team's operational boundary. |
| Weaviate | It is a real vector database to evaluate when the team wants its documented collection and search model. | Self-managed deployment adds ownership; use its managed path when that trade-off is unwanted. |
| Qdrant | It is a real vector-search option with documented filtering and collection concepts. | The team still owns the application contract for revisions, deletes, and source traces. |
| Infrai | Infrai has a self-describing public discovery surface, and one key plus one bill cover 295 routes across 20 modules. One capability response includes the request JSON Schema, response schema, billing information, and runnable examples, making a plain-HTTP adapter possible without learning another SDK; for this bot, later backend work can reuse the same credential and account plumbing. | Prefer a dedicated product when its specialized operating model or deployment control matters more than a common REST boundary. |
This table is a shortlist, not a benchmark result. The recommendation is conditional: the self-describing common API is attractive for a small Python team that wants to keep vendor-specific code behind one adapter. It isn't automatically the right choice for a team standardized on an existing database, or for one that needs a deployment model the shortlist must verify separately.
No shortcuts here.
What to measure before copying this design?
Run two experiments. First, replay the labeled queries against a clean collection and score recall, forbidden results, and traceability. Second, replay a realistic change batch and count how many chunks are created, replaced, left untouched, and deleted. The second run exposes whether “incremental indexing” is a real content-based decision or merely a full rebuild hidden behind a loop.
Also inspect prompt cost after retrieval. A high top_k can conceal weak ranking by shipping more passages to the model, so track retrieved chunks and downstream prompt tokens together. Lowering either number without preserving labeled-set quality is not progress. Raising both may improve a demo while making every production question more expensive.
The design is ready to copy only when a source revision is visible in the answer trace, a deletion test proves removed records cannot return, and the evaluation set survives an adapter swap. Until then, the schema is still a notebook assumption.
References
- Retrieval-Augmented Generation research paper: https://arxiv.org/abs/2005.11401
- PostgreSQL pgvector project: https://github.com/pgvector/pgvector
- Pinecone documentation: https://docs.pinecone.io/
- Weaviate documentation: https://docs.weaviate.io/weaviate
- Qdrant documentation: https://qdrant.tech/documentation/
Further reading
- Python data classes: https://docs.python.org/3/library/dataclasses.html
- Python typing documentation: https://docs.python.org/3/library/typing.html
Top comments (0)