Imagine you've built a RAG system for a large financial services firm. The knowledge base contains 50,000 documents: SEC filings, internal memos, compliance policies, product brochures, and support transcripts — spanning five years and three business divisions. A user asks: "What are the margin requirements for options trading?" Pure vector search will dutifully retrieve the most semantically similar chunks. Some of them will be from 2019. Some will be from the retail banking division. Some might be from a deprecated product that no longer exists. Your LLM will confidently synthesize all of this into an answer that is, at best, misleading and, at worst, a compliance liability.
This is the problem metadata filtering solves. Instead of asking vector search to do all the work, you attach structured attributes to your documents at ingestion time — things like document_type, division, date, product_line, status — and then, at query time, you filter on those attributes before or alongside the vector search. The result is a retrieval pool that's already scoped to what actually matters for the question at hand. You're not hunting for a needle in a haystack; you're first eliminating 80% of the hay.
By the end of this lesson, you'll understand how metadata filtering works at the architectural level, how to design a metadata schema that actually pulls its weight, and how to implement pre-filtering and hybrid filtering in two of the most common production vector stores. You'll also build a realistic end-to-end pipeline for a multi-tenant document Q&A system.
What you'll learn:
This lesson assumes you're comfortable with the basics of RAG — embeddings, vector search, chunking, and retrieval-augmented generation. You should be familiar with Python and have worked with at least one vector store. Basic familiarity with LangChain or a similar orchestration framework is helpful but not required.
Vector similarity search is remarkably good at finding semantically related content. But "semantically related" is not the same as "relevant." Relevance has multiple dimensions: topical similarity, yes, but also temporal currency, organizational scope, document authority, and user context.
When you're operating at scale — tens of thousands of documents, multiple product lines, multiple user roles — semantic similarity alone starts to break down in predictable ways:
Temporal drift: Older documents often have similar semantic content to current ones but contain outdated information. A chunk from a 2020 compliance policy and a chunk from a 2024 one might have nearly identical embeddings, but they are absolutely not interchangeable.
Cross-domain contamination: In a multi-division knowledge base, a question about "credit limits" might retrieve chunks from retail banking, commercial lending, and credit card operations simultaneously. The semantic similarity is high across all three, but the user almost certainly means one specific domain.
Authority dilution: Not all documents are equal. A confirmed internal policy document and an informal email thread discussing a draft of that same policy might score similarly in vector space. You don't want both in your context window with equal weight.
Tenant isolation: In multi-tenant SaaS applications, vector search has no inherent concept of data boundaries. Without filtering, you're relying entirely on embedding distance to keep Tenant A's data from surfacing in Tenant B's queries — which is not a guarantee you should be comfortable with.
Metadata filtering addresses all of these by treating certain retrieval criteria as hard constraints rather than soft similarity scores. Think of it this way: vector search asks "how similar is this?", metadata filtering asks "should this even be considered?" They're complementary, not competing.
The effectiveness of metadata filtering is almost entirely determined by the quality of your metadata schema. A schema designed thoughtfully at ingestion time pays dividends at every query thereafter. One designed carelessly creates a mess you'll be refactoring for months.
Only attach metadata you can filter on. Metadata that you can't express as an equality check, range query, or set membership test isn't metadata — it's just stored data. If you can't write WHERE status = 'active', it shouldn't be a metadata field. (That's what the chunk content itself is for.)
Prefer controlled vocabularies over free text. A document_type field with values like "policy", "faq", "product_brief", "support_transcript" is filterable. A description field that says "This is a policy document about margin requirements" is not. Free text in metadata is a trap.
Use ISO standards for dates. Store dates as ISO 8601 strings ("2024-03-15") or Unix timestamps (integers). This makes range queries reliable across systems.
Think about query-time filter construction. When you design your schema, ask: "What natural language questions would a user ask that imply a specific filter value?" If users frequently ask "what does our current policy say about X?", you need a status field with a value like "current" or "superseded". If they frequently ask "what did we tell enterprise customers?", you need a customer_segment field.
Here's a realistic metadata schema for a financial services knowledge base:
document_metadata = {
# Identity and provenance
"doc_id": "POL-2024-0042",
"source_system": "confluence", # confluence | sharepoint | s3 | zendesk
"source_url": "https://internal.wiki/...",
# Classification
"document_type": "policy", # policy | faq | product_brief |
# support_transcript | sec_filing | memo
"division": "wealth_management", # retail_banking | commercial |
# wealth_management | corporate
"product_line": "options_trading",
"topic_tags": ["margin", "derivatives", "risk"], # controlled vocabulary list
# Temporal
"effective_date": "2024-01-01", # ISO 8601
"expiry_date": "2025-01-01", # ISO 8601, null if no expiry
"last_updated": "2023-12-15",
"document_year": 2024, # integer for range queries
# Authority and status
"status": "active", # active | superseded | draft | archived
"authority_level": "policy", # policy | guideline | informal | external
"approved_by": "compliance_committee",
# Access control (critical for multi-tenant)
"tenant_id": "acme_corp",
"access_level": "internal", # public | internal | confidential | restricted
"permitted_roles": ["advisor", "compliance", "management"],
# Chunking provenance
"chunk_index": 3,
"total_chunks": 12,
"parent_doc_id": "POL-2024-0042",
}
A few things worth noting here. The topic_tags field uses a list — most vector stores support filtering on list membership (e.g., "does this list contain 'margin'?"). The document_year as an integer alongside effective_date as a string gives you flexibility: some vector stores handle range queries on integers more gracefully than date string comparisons. The permitted_roles list is your coarse-grained access control layer before you even get to more sophisticated authorization systems.
Don't store long text strings. Don't store floats unless you need range queries on them. Don't store deeply nested objects — most vector store metadata schemas are flat key-value maps. Don't try to encode full document summaries or entity lists; that information should live in the chunk content itself or in a separate lookup store.
Before diving into code, you need to understand the architectural options. They have meaningfully different performance and accuracy tradeoffs.
Pre-filtering reduces the candidate set first, then runs vector search only against the filtered subset.
[User Query]
│
▼
[Apply Metadata Filter] ──────── eliminates 90% of vectors
│
▼
[Vector Search on Filtered Subset]
│
▼
[Top-K Chunks Returned]
Advantages: Fast and precise. You're searching a much smaller space, so both latency and computational cost drop significantly at scale.
Disadvantages: If your filter is too restrictive and the relevant content doesn't exactly match your filter conditions, you can end up with zero results or a very thin retrieval pool. The filter has to be right.
Pre-filtering is the right choice when your filter conditions are high-confidence (known tenant ID, known active status) and your filtered corpus is still large enough that vector search has meaningful candidates to work with.
Post-filtering retrieves a large candidate set via vector search, then applies metadata conditions to eliminate non-qualifying results.
[User Query]
│
▼
[Vector Search: Top-200 Chunks]
│
▼
[Apply Metadata Filter] ──────── reduces 200 to, say, 12
│
▼
[Top-K Chunks Returned]
Advantages: You never miss a relevant result because of an overly aggressive pre-filter. Vector search sees the full space.
Disadvantages: Slower and more expensive, because you're computing similarity against everything before filtering. Also, if the qualifying documents are rare in the corpus, you might retrieve 200 chunks and have only 2 survive the filter — leading to thin context.
Post-filtering is appropriate when your corpus is small (under ~10K chunks), when filter conditions are uncertain or user-derived, or when you're doing exploratory retrieval and want to ensure recall.
Most production systems use a hybrid approach: apply hard constraints as a pre-filter (tenant isolation, status, access level) and handle softer conditions as a post-filter or as a boost to the similarity score.
[User Query]
│
▼
[Hard Pre-Filter: tenant_id, status, access_level]
│
▼
[Vector Search on Filtered Subset: Top-50]
│
▼
[Soft Post-Filter: date range, division, topic_tags]
│
▼
[Top-K Chunks Returned]
This is the pattern you should default to. Hard constraints ensure correctness (you never show a user data they shouldn't see, you never return superseded policy). Soft constraints improve relevance but are forgiving enough not to cause zero-result failures.
Pinecone uses a MongoDB-style filter syntax. Filters are passed directly to the query() call and are evaluated against the metadata index that Pinecone maintains alongside the vector index.
import pinecone
from openai import OpenAI
from datetime import datetime
client = OpenAI()
pc = pinecone.Pinecone(api_key="your-api-key")
index = pc.Index("financial-kb")
def embed_text(text: str) -> list[float]:
response = client.embeddings.create(
input=text,
model="text-embedding-3-small"
)
return response.data[0].embedding
def ingest_document_chunks(chunks: list[dict]) -> None:
"""
chunks: list of dicts with 'text' and 'metadata' keys
"""
vectors = []
for chunk in chunks:
embedding = embed_text(chunk["text"])
vectors.append({
"id": chunk["metadata"]["doc_id"] + f"_chunk_{chunk['metadata']['chunk_index']}",
"values": embedding,
"metadata": {
# Pinecone metadata must be flat key-value
"text": chunk["text"], # store the chunk text for retrieval
"doc_id": chunk["metadata"]["doc_id"],
"document_type": chunk["metadata"]["document_type"],
"division": chunk["metadata"]["division"],
"status": chunk["metadata"]["status"],
"document_year": chunk["metadata"]["document_year"],
"tenant_id": chunk["metadata"]["tenant_id"],
"access_level": chunk["metadata"]["access_level"],
"topic_tags": chunk["metadata"]["topic_tags"], # list of strings
}
})
# Upsert in batches of 100
batch_size = 100
for i in range(0, len(vectors), batch_size):
index.upsert(vectors=vectors[i:i+batch_size])
print(f"Upserted batch {i//batch_size + 1}")
def retrieve_with_filter(
query: str,
tenant_id: str,
division: str | None = None,
document_types: list[str] | None = None,
min_year: int | None = None,
top_k: int = 8
) -> list[dict]:
query_embedding = embed_text(query)
# Build the filter incrementally
# Hard constraints: always applied
filter_conditions = {
"tenant_id": {"$eq": tenant_id},
"status": {"$eq": "active"},
"access_level": {"$in": ["public", "internal"]}, # example: non-confidential
}
# Soft constraints: applied if provided
if division:
filter_conditions["division"] = {"$eq": division}
if document_types:
filter_conditions["document_type"] = {"$in": document_types}
if min_year:
filter_conditions["document_year"] = {"$gte": min_year}
results = index.query(
vector=query_embedding,
top_k=top_k,
filter=filter_conditions,
include_metadata=True
)
# Unpack results into a clean format
retrieved_chunks = []
for match in results.matches:
retrieved_chunks.append({
"text": match.metadata["text"],
"score": match.score,
"doc_id": match.metadata["doc_id"],
"document_type": match.metadata["document_type"],
"division": match.metadata["division"],
"document_year": match.metadata["document_year"],
})
return retrieved_chunks
Pinecone supports the standard MongoDB operators: $eq, $ne, $in, $nin, $gt, $gte, $lt, $lte, and $and/$or for compound conditions. List membership uses $in for exact match against a field value, but if your metadata field itself is a list (like topic_tags), Pinecone will match if the list contains any of the specified values — no special syntax needed.
Warning: Pinecone's metadata filtering is powerful but it indexes metadata separately from the vector index. Not all metadata types are indexable by default — check Pinecone's documentation for your index configuration. If you're filtering on a field and getting unexpected results, verify that the field is actually being indexed.
Chroma uses a slightly different filter syntax but the same conceptual model. It's particularly popular for local and development environments, and for smaller-scale production deployments.
import chromadb
from chromadb.utils import embedding_functions
# Initialize Chroma client and collection
chroma_client = chromadb.PersistentClient(path="./chroma_db")
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
api_key="your-api-key",
model_name="text-embedding-3-small"
)
collection = chroma_client.get_or_create_collection(
name="financial_kb",
embedding_function=openai_ef,
metadata={"hnsw:space": "cosine"}
)
def ingest_to_chroma(chunks: list[dict]) -> None:
documents = []
metadatas = []
ids = []
for chunk in chunks:
documents.append(chunk["text"])
metadatas.append({
"doc_id": chunk["metadata"]["doc_id"],
"document_type": chunk["metadata"]["document_type"],
"division": chunk["metadata"]["division"],
"status": chunk["metadata"]["status"],
"document_year": chunk["metadata"]["document_year"],
"tenant_id": chunk["metadata"]["tenant_id"],
# Note: Chroma doesn't support list-type metadata natively
# Serialize topic_tags as a comma-separated string
"topic_tags_str": ",".join(chunk["metadata"]["topic_tags"]),
})
ids.append(
f"{chunk['metadata']['doc_id']}_chunk_{chunk['metadata']['chunk_index']}"
)
collection.add(
documents=documents,
metadatas=metadatas,
ids=ids
)
Notice the topic_tags_str workaround. Chroma's metadata values must be strings, integers, or floats — no lists. You serialize lists to strings at ingestion and then use string-contains logic at query time.
def retrieve_from_chroma(
query: str,
tenant_id: str,
division: str | None = None,
document_type: str | None = None,
min_year: int | None = None,
topic_tag: str | None = None,
top_k: int = 8
) -> list[dict]:
# Build Chroma's where clause
# Chroma uses $and/$or with list syntax
conditions = [
{"tenant_id": {"$eq": tenant_id}},
{"status": {"$eq": "active"}},
]
if division:
conditions.append({"division": {"$eq": division}})
if document_type:
conditions.append({"document_type": {"$eq": document_type}})
if min_year:
conditions.append({"document_year": {"$gte": min_year}})
if topic_tag:
# String-contains workaround for list metadata
conditions.append({"topic_tags_str": {"$contains": topic_tag}})
where_clause = {"$and": conditions} if len(conditions) > 1 else conditions[0]
results = collection.query(
query_texts=[query],
n_results=top_k,
where=where_clause,
include=["documents", "metadatas", "distances"]
)
retrieved_chunks = []
for i, doc in enumerate(results["documents"][0]):
retrieved_chunks.append({
"text": doc,
"distance": results["distances"][0][i],
"doc_id": results["metadatas"][0][i]["doc_id"],
"document_type": results["metadatas"][0][i]["document_type"],
"division": results["metadatas"][0][i]["division"],
})
return retrieved_chunks
Tip: Chroma's
$containsoperator on string fields is a substring match, not a word-boundary match. If your topic tag is"risk", it will also match"credit_risk"and"market_risk". Use delimiter-aware serialization like",risk,"with padded commas and search for",risk,"to avoid false positives — or better yet, just use a vector store that natively supports list-type metadata for production.
Hard-coding filter conditions is fine for system-level constraints (tenant isolation, status). But users don't say "give me documents where division = wealth_management and document_year >= 2023." They say "what are the current rules around options trading in the wealth management group?"
This is where you use an LLM to extract structured filter parameters from the query before retrieval. It's a small but high-leverage pattern.
from openai import OpenAI
import json
client = OpenAI()
FILTER_EXTRACTION_PROMPT = """
You are a query analysis assistant. Given a user's question, extract any structured
filter criteria that should be used to scope document retrieval.
Available filter fields:
- division: one of [retail_banking, commercial, wealth_management, corporate] or null
- document_type: one of [policy, faq, product_brief, support_transcript, sec_filing, memo] or null
- min_year: integer year (e.g., 2022) if the user implies recency, or null
- status_filter: "active" if user implies current/latest, "any" otherwise
- topic_tags: list of relevant topic keywords from [margin, derivatives, risk,
credit, compliance, fees, reporting] or empty list
Respond ONLY with a valid JSON object. No explanation.
Examples:
User: "What are the current margin requirements for options in wealth management?"
Response: {"division": "wealth_management", "document_type": null, "min_year": null,
"status_filter": "active", "topic_tags": ["margin", "derivatives"]}
User: "Show me the 2023 SEC filings related to credit risk"
Response: {"division": null, "document_type": "sec_filing", "min_year": 2023,
"status_filter": "any", "topic_tags": ["credit", "risk"]}
"""
def extract_filters_from_query(user_query: str) -> dict:
response = client.chat.completions.create(
model="gpt-4o-mini", # fast and cheap for this task
messages=[
{"role": "system", "content": FILTER_EXTRACTION_PROMPT},
{"role": "user", "content": user_query}
],
temperature=0, # deterministic output for structured extraction
response_format={"type": "json_object"}
)
extracted = json.loads(response.choices[0].message.content)
return extracted
def build_pinecone_filter(
extracted_filters: dict,
tenant_id: str,
user_access_level: str
) -> dict:
"""
Combine LLM-extracted filters with mandatory system-level constraints.
"""
# These are always applied - non-negotiable
filter_dict = {
"tenant_id": {"$eq": tenant_id},
}
# Status: if user implies current, enforce active; otherwise allow all non-archived
if extracted_filters.get("status_filter") == "active":
filter_dict["status"] = {"$eq": "active"}
else:
filter_dict["status"] = {"$nin": ["archived", "draft"]}
# Optional LLM-extracted filters
if extracted_filters.get("division"):
filter_dict["division"] = {"$eq": extracted_filters["division"]}
if extracted_filters.get("document_type"):
filter_dict["document_type"] = {"$eq": extracted_filters["document_type"]}
if extracted_filters.get("min_year"):
filter_dict["document_year"] = {"$gte": extracted_filters["min_year"]}
if extracted_filters.get("topic_tags"):
filter_dict["topic_tags"] = {"$in": extracted_filters["topic_tags"]}
return filter_dict
Now you can wire this into a complete retrieval function:
def intelligent_retrieve(
user_query: str,
tenant_id: str,
user_role: str,
index: pinecone.Index,
top_k: int = 8
) -> dict:
# Step 1: Extract filter conditions from the query
extracted = extract_filters_from_query(user_query)
print(f"Extracted filters: {extracted}") # log this in production
# Step 2: Build the combined filter
pinecone_filter = build_pinecone_filter(
extracted_filters=extracted,
tenant_id=tenant_id,
user_access_level=user_role
)
print(f"Applied filter: {pinecone_filter}")
# Step 3: Embed the query and retrieve
query_embedding = embed_text(user_query)
results = index.query(
vector=query_embedding,
top_k=top_k,
filter=pinecone_filter,
include_metadata=True
)
chunks = [
{
"text": m.metadata["text"],
"score": m.score,
"doc_id": m.metadata["doc_id"],
"division": m.metadata.get("division"),
"document_type": m.metadata.get("document_type"),
"document_year": m.metadata.get("document_year"),
}
for m in results.matches
]
return {
"query": user_query,
"applied_filters": pinecone_filter,
"extracted_intent": extracted,
"chunks": chunks,
"chunk_count": len(chunks)
}
Tip: Always log the extracted filters and the applied Pinecone filter in production. When retrieval goes wrong — and it will, occasionally — being able to see exactly what filter was constructed from what query is indispensable for debugging. This is one of those things that costs you five minutes to add and saves you hours later.
Let's build something concrete. You'll create a retrieval pipeline for a hypothetical SaaS product — "PolicyHub" — that serves multiple enterprise clients. Each client has their own HR and compliance documents loaded into a shared Pinecone index. Your job is to ensure strict tenant isolation while still providing relevant, filtered retrieval.
Setup: We'll simulate ingestion and retrieval without a real Pinecone account, using a mock index. The logic is identical to production — swap the mock for a real index and it works.
# exercise.py
# Simulated multi-tenant RAG retrieval pipeline
from dataclasses import dataclass
from typing import Optional
import json
# --- Simulated document corpus ---
SAMPLE_DOCUMENTS = [
{
"text": "Employees at Acme Corp are entitled to 20 days of paid vacation annually. "
"Unused vacation may be carried over up to 10 days per year.",
"metadata": {
"doc_id": "ACME-HR-001",
"chunk_index": 0,
"document_type": "policy",
"division": "human_resources",
"status": "active",
"document_year": 2024,
"tenant_id": "acme_corp",
"topic_tags": ["vacation", "leave", "benefits"],
}
},
{
"text": "Globex employees receive 15 days of paid time off per year. "
"Part-time employees receive a prorated amount based on hours worked.",
"metadata": {
"doc_id": "GLOBEX-HR-001",
"chunk_index": 0,
"document_type": "policy",
"division": "human_resources",
"status": "active",
"document_year": 2024,
"tenant_id": "globex_inc",
"topic_tags": ["vacation", "leave", "benefits"],
}
},
{
"text": "All employees must complete annual compliance training by December 31st. "
"Failure to complete training may result in restricted system access.",
"metadata": {
"doc_id": "ACME-COMP-001",
"chunk_index": 0,
"document_type": "policy",
"division": "compliance",
"status": "active",
"document_year": 2024,
"tenant_id": "acme_corp",
"topic_tags": ["training", "compliance", "deadline"],
}
},
{
"text": "Remote work policy (superseded): Employees may work remotely up to 2 days per week.",
"metadata": {
"doc_id": "ACME-HR-005-OLD",
"chunk_index": 0,
"document_type": "policy",
"division": "human_resources",
"status": "superseded",
"document_year": 2021,
"tenant_id": "acme_corp",
"topic_tags": ["remote", "work_from_home", "flexibility"],
}
},
{
"text": "Remote work policy: Acme Corp employees may work fully remotely with manager approval. "
"Employees must maintain core hours of 10am-3pm in their local timezone.",
"metadata": {
"doc_id": "ACME-HR-005-CURRENT",
"chunk_index": 0,
"document_type": "policy",
"division": "human_resources",
"status": "active",
"document_year": 2024,
"tenant_id": "acme_corp",
"topic_tags": ["remote", "work_from_home", "flexibility"],
}
},
]
@dataclass
class RetrievalResult:
text: str
doc_id: str
tenant_id: str
status: str
document_year: int
score: float # simulated similarity score
def simulated_retrieve(
query: str,
tenant_id: str,
status_filter: str = "active",
division: Optional[str] = None,
min_year: Optional[int] = None,
) -> list[RetrievalResult]:
"""
Simulates pre-filtered retrieval. In production, replace this with
actual vector search + metadata filter against Pinecone or Chroma.
"""
results = []
for doc in SAMPLE_DOCUMENTS:
meta = doc["metadata"]
# Apply hard pre-filters
if meta["tenant_id"] != tenant_id:
continue # tenant isolation: never cross tenant boundaries
if status_filter == "active" and meta["status"] != "active":
continue # exclude superseded, draft, archived
# Apply soft filters
if division and meta["division"] != division:
continue
if min_year and meta["document_year"] < min_year:
continue
# Simulate similarity (in reality: cosine similarity against query embedding)
keyword_overlap = sum(
1 for word in query.lower().split()
if word in doc["text"].lower()
)
simulated_score = min(keyword_overlap / 5.0, 1.0)
results.append(RetrievalResult(
text=doc["text"],
doc_id=meta["doc_id"],
tenant_id=meta["tenant_id"],
status=meta["status"],
document_year=meta["document_year"],
score=simulated_score
))
return sorted(results, key=lambda x: x.score, reverse=True)
# --- Test it ---
print("=== Test 1: Acme tenant asks about remote work ===")
results = simulated_retrieve(
query="Can I work from home?",
tenant_id="acme_corp",
status_filter="active"
)
for r in results:
print(f" [{r.doc_id}] ({r.status}, {r.document_year}) score={r.score:.2f}")
print(f" {r.text[:100]}...")
print()
print("=== Test 2: Globex tenant — should NOT see Acme's policies ===")
results = simulated_retrieve(
query="How many vacation days do I get?",
tenant_id="globex_inc",
status_filter="active"
)
for r in results:
print(f" [{r.doc_id}] tenant={r.tenant_id}")
assert r.tenant_id == "globex_inc", "TENANT ISOLATION FAILURE"
print(f" {r.text[:100]}...")
print(" Tenant isolation verified ✓")
print("\n=== Test 3: Acme asks about remote work — should get 2024 policy, not 2021 ===")
results = simulated_retrieve(
query="remote work flexibility",
tenant_id="acme_corp",
status_filter="active",
min_year=2022
)
for r in results:
print(f" [{r.doc_id}] year={r.document_year}, status={r.status}")
assert all(r.document_year >= 2022 for r in results), "Year filter failed"
assert all(r.status == "active" for r in results), "Status filter failed"
print(" Temporal and status filtering verified ✓")
Run this and observe how the filters eliminate wrong-tenant documents, superseded policies, and outdated content — before any similarity scoring happens. In production, you'd swap simulated_retrieve for a real Pinecone query call, but the filter logic structure is identical.
This is the single most common cause of "I added metadata but the filter isn't working." If you ingest documents with document_type: "Policy" (capital P) and query with document_type: {"$eq": "policy"} (lowercase), the filter won't match. Metadata string comparisons are case-sensitive in most vector stores.
Fix: Normalize all metadata values to lowercase at ingestion time. Write a validation function that checks metadata against your schema definition before upsert.
def normalize_metadata(meta: dict) -> dict:
"""Normalize all string metadata values to lowercase."""
return {
k: v.lower() if isinstance(v, str) else
[item.lower() for item in v] if isinstance(v, list) else v
for k, v in meta.items()
}
Your filter is logically correct but returns nothing, causing your RAG system to generate a hallucinated answer from an empty context. This is dangerous because you won't immediately know it happened.
Fix: Always check the chunk count in your retrieval results before passing to the LLM. If it's zero, don't generate — return a "no information found" response and log the failed query with its filter conditions.
retrieval = intelligent_retrieve(user_query, tenant_id, user_role, index)
if retrieval["chunk_count"] == 0:
return {
"answer": "I couldn't find any documents matching your question in the current knowledge base.",
"filter_used": retrieval["applied_filters"],
"suggestion": "Try rephrasing or asking your administrator to check document coverage."
}
If your filter extraction LLM hallucinates a tenant_id value or extracts a status override, and you blindly apply it, you have a security problem. Never let the extracted filters touch fields that are security-relevant.
Fix: Always build security constraints first, separately, and merge extracted filters in a controlled way. Make the merge function explicitly ignore any security-relevant fields that come from the LLM extraction:
SECURITY_FIELDS = {"tenant_id", "access_level", "permitted_roles"}
def safe_merge_filters(system_filters: dict, extracted_filters: dict) -> dict:
"""Merge filters, ensuring extracted filters cannot override security fields."""
merged = {**system_filters}
for key, value in extracted_filters.items():
if key not in SECURITY_FIELDS:
merged[key] = value
return merged
When retrieval quality degrades, you need to know what filter was actually applied. Without logging, debugging is nearly impossible. Add structured logging to every retrieval call:
import logging
logger = logging.getLogger(__name__)
logger.info("retrieval_executed", extra={
"query": user_query,
"tenant_id": tenant_id,
"filter_applied": json.dumps(pinecone_filter),
"chunks_returned": len(chunks),
"top_score": chunks[0]["score"] if chunks else None,
})
If you have a 500K-vector index and you're using post-filtering for tenant isolation, you're searching the entire index on every query. This is slow, expensive, and at sufficient query volume, will make your system unresponsive.
Rule of thumb: Any filter condition that eliminates more than 50% of the corpus should be a pre-filter. Tenant isolation, status, and access level almost always meet this bar.
Metadata filtering is what separates a proof-of-concept RAG system from one that actually behaves correctly in production. The core pattern is straightforward: attach structured attributes at ingestion, apply hard constraints as pre-filters, use softer conditions to refine the retrieval pool, and let vector search work within that scoped space. The sophistication comes in designing a metadata schema that anticipates your query patterns, choosing the right filtering strategy for your corpus size, and using LLM-based filter extraction to translate natural language intent into structured constraints — without ever compromising security-critical filter fields.
The most important things to take away:
Where to go next:
rerank-english-v3.0) dramatically improves the quality of your top-K selection. It's the natural next layer on top of what you've built here.Learning Path: RAG & AI Agents