Agent Data Freshness: Three Retrieval Problems

import requests, os, time, hashlib, json

SCAVIO_KEY = os.environ["SCAVIO_API_KEY"]

class FreshRetrievalLayer:
    """Retrieval layer with TTL-aware caching."""

    def __init__(self):
        self.cache = {}

    def search(self, query, count=10, max_age_seconds=3600):
        """Search with time-aware cache. Stale data auto-expires."""
        cache_key = hashlib.md5(f"{query}:{count}".encode()).hexdigest()
        now = time.time()

        if cache_key in self.cache:
            cached_at, data = self.cache[cache_key]
            age = now - cached_at
            if age < max_age_seconds:
                return data, {"source": "cache", "age_seconds": int(age)}

        resp = requests.post(
            "https://api.scavio.dev/api/v1/search",
            headers={"x-api-key": SCAVIO_KEY},
            json={"query": query, "num_results": count}
        )
        data = resp.json()["results"]
        self.cache[cache_key] = (now, data)
        return data, {"source": "live", "age_seconds": 0}

# Use case-specific TTLs instead of one-size-fits-all
retrieval = FreshRetrievalLayer()

# Financial data: 5 minute cache (prices change fast)
stock_data, meta = retrieval.search(
    "NVDA stock price today", max_age_seconds=300
)

# Product research: 1 hour cache (changes slowly)
products, meta = retrieval.search(
    "best project management tools 2026", max_age_seconds=3600
)

# Reference docs: 24 hour cache (rarely changes)
docs, meta = retrieval.search(
    "Python asyncio documentation", max_age_seconds=86400
)

print(f"Data source: {meta['source']}, age: {meta['age_seconds']}s")

class ValidatedRetrievalLayer(FreshRetrievalLayer):
    """Retrieval with result validation to prevent cache poisoning."""

    def __init__(self):
        super().__init__()
        self.blocked_domains = set()

    def search_validated(self, query, count=10, max_age_seconds=3600):
        """Search with validation: filter bad results before caching."""
        results, meta = self.search(query, count + 5, max_age_seconds)

        validated = []
        for r in results:
            if not self._validate_result(r):
                continue
            validated.append(r)
            if len(validated) >= count:
                break

        return validated, meta

    def _validate_result(self, result):
        """Check if a result is trustworthy enough to cache."""
        url = result.get("url", "")
        title = result.get("title", "")
        snippet = result.get("description", "")

        # Block known bad domains
        for domain in self.blocked_domains:
            if domain in url:
                return False

        # Flag suspiciously empty results
        if len(snippet) < 20:
            return False

        # Flag results with spam indicators
        spam_signals = ["click here", "buy now", "limited time",
                        "act fast", "guaranteed"]
        spam_count = sum(1 for s in spam_signals if s in title.lower())
        if spam_count >= 2:
            return False

        return True

    def report_bad_result(self, url):
        """Mark a domain as unreliable. Invalidate cached results."""
        from urllib.parse import urlparse
        domain = urlparse(url).netloc
        self.blocked_domains.add(domain)
        # Purge cache entries containing this domain
        keys_to_purge = []
        for key, (ts, data) in self.cache.items():
            if any(domain in r.get("url", "") for r in data):
                keys_to_purge.append(key)
        for key in keys_to_purge:
            del self.cache[key]
        return f"Blocked {domain}, purged {len(keys_to_purge)} cache entries"

retrieval = ValidatedRetrievalLayer()
results, meta = retrieval.search_validated("AI agent deployment patterns")
print(f"Validated {len(results)} results from {meta['source']}")

class TemporalRetrievalLayer(ValidatedRetrievalLayer):
    """Retrieval with temporal awareness."""

    def search_temporal(self, query, count=10, recency_bias=True):
        """Search with temporal metadata for grounding."""
        results, meta = self.search_validated(query, count)

        temporal_results = []
        for r in results:
            enriched = {
                **r,
                "retrieval_timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
                "is_live_data": True,
            }
            # Extract date from result if available
            result_date = r.get("date", "")
            if result_date:
                enriched["result_date"] = result_date

            temporal_results.append(enriched)

        return temporal_results, meta

    def build_grounding_prompt(self, query, results):
        """Build a prompt that makes temporal context explicit."""
        context = (
            f"The following information was retrieved via live web search "
            f"on {time.strftime('%Y-%m-%d')} and reflects current data. "
            f"If this contradicts your training data, prefer the search "
            f"results as they are more recent.\n\n"
        )
        for i, r in enumerate(results, 1):
            date_note = f" (dated: {r.get('result_date', 'unknown')})"
            context += (
                f"[{i}] {r['title']}{date_note}\n"
                f"{r['description']}\n"
                f"URL: {r['url']}\n\n"
            )
        return context

retrieval = TemporalRetrievalLayer()
results, meta = retrieval.search_temporal(
    "Tavily acquisition status 2026"
)
grounding = retrieval.build_grounding_prompt(
    "Tavily acquisition status", results
)
print(grounding[:500])

class ProductionRetrievalLayer:
    """Complete retrieval layer solving all three freshness problems."""

    def __init__(self):
        self.retrieval = TemporalRetrievalLayer()
        self.query_log = []

    def retrieve(self, query, use_case="general"):
        """Production-grade retrieval with use-case-specific settings."""
        config = {
            "financial": {"count": 10, "max_age": 300, "recency": True},
            "research":  {"count": 15, "max_age": 3600, "recency": True},
            "reference": {"count": 5,  "max_age": 86400, "recency": False},
            "general":   {"count": 10, "max_age": 1800, "recency": True},
        }
        c = config.get(use_case, config["general"])

        results, meta = self.retrieval.search_temporal(
            query, count=c["count"], recency_bias=c["recency"]
        )

        grounding = self.retrieval.build_grounding_prompt(query, results)

        self.query_log.append({
            "query": query,
            "use_case": use_case,
            "results_count": len(results),
            "source": meta["source"],
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        })

        return {
            "results": results,
            "grounding_prompt": grounding,
            "meta": meta,
        }

# Usage in an agent
layer = ProductionRetrievalLayer()

# Financial query: 5min cache, high result count
financial = layer.retrieve(
    "NVDA earnings Q1 2026 results", use_case="financial"
)

# Reference query: 24hr cache, low result count
reference = layer.retrieve(
    "Python requests library POST example", use_case="reference"
)

print(f"Financial: {len(financial['results'])} results ({financial['meta']['source']})")
print(f"Reference: {len(reference['results'])} results ({reference['meta']['source']})")