Infrastructure Requirements: Ingestion, Retrieval, and Generation Layers (Airweave)

The previous subtopics established WHEN you need agentic context retrieval; this subtopic covers WHAT it actually takes to build one, following this course's own detailed breakdown of its three layers.

The Ingestion layer handles getting data IN correctly. Per this course, this means: connect to apps without auth headaches (each source system — Gmail, Slack, Linear — has its own authentication scheme, and a real system needs to handle all of them uniformly); process different data sources properly before embedding (email vs code vs calendar) — you can't naively chunk-and-embed a calendar event the same way you'd chunk a code file or an email thread, since their structure and what matters about them differs fundamentally; and detect if a source is updated and refresh embeddings (ideally, without a full refresh) — re-embedding an entire source every time anything changes is wasteful; a good system detects and re-embeds only what's actually changed.

The Retrieval layer handles finding the RIGHT information once it's ingested: expand vague queries to infer what users actually want (real user questions are often underspecified compared to what they actually need answered); direct queries to the correct data sources (this is the agentic source-selection decision from the previous subtopic); layer multiple search strategies like semantic-based, keyword-based, and graph-based (pure embedding similarity, from vector-search-basics, isn't always the right tool — sometimes an exact keyword match or a relationship-graph traversal finds what semantic search alone would miss); ensure retrieving only what users are authorized to see (the access-control requirement from the previous subtopic, enforced here); and weigh old vs. new retrieved info (recent data matters more, but old context still counts) — a genuinely subtle requirement, since naively always preferring the newest information can discard genuinely relevant older context.

The Generation layer, comparatively simpler in this course's treatment, has one core requirement: provide a citation-backed LLM response — so the user can verify where each part of the answer actually came from, directly connecting to the citation approach from the context-engineering-workflow-build topic's Streamlit app.

This course grounds all of this in a named, real implementation rather than leaving it purely abstract: Airweave, a recently trending 100% open-source framework that provides the context retrieval layer for AI agents across 30+ apps and databases (as of 3 Dec, 2025). It implements everything discussed above: how to handle authentication across apps, how to process different data sources, how to gather info from multiple tools, how to weigh old vs. new info, how to detect updates and do real-time sync, and how to generate perplexity-like citation-backed responses. This course adds one honest, specific engineering nuance about the update-detection piece: for instance, to detect updates and initiate a re-sync, one might do timestamp comparisons. But this does not tell if the content actually changed (maybe only the permission was updated), and you might still re-embed everything unnecessarily — a reminder that even a solved-looking sub-problem (detect updates) has real subtlety once you dig into it (a timestamp changing doesn't mean the CONTENT changed).

# Modeling the 3 layers' concrete requirements, following the book's
# own breakdown -- including the update-detection nuance it flags.

def ingestion_layer_process(source_kind: str, raw_data: str) -> dict:
    """Different source KINDS need different processing before embedding."""
    handlers = {
        "email": lambda d: {"subject": d.split("\n")[0], "body": d},
        "calendar": lambda d: {"event_time": "parsed_time", "attendees": []},
        "code": lambda d: {"function_signatures": [], "raw": d},
    }
    return handlers.get(source_kind, lambda d: {"raw": d})(raw_data)

def detect_real_change(old_timestamp: str, new_timestamp: str,
                       old_content_hash: str, new_content_hash: str) -> bool:
    """The book's own flagged nuance: a timestamp change alone doesn't
    mean the CONTENT changed (e.g. only a permission was updated) --
    compare content hashes too, to avoid unnecessary re-embedding."""
    timestamp_changed = old_timestamp != new_timestamp
    content_actually_changed = old_content_hash != new_content_hash
    return timestamp_changed and content_actually_changed   # only re-embed if BOTH

def retrieval_layer_weigh_recency(results: list[dict], recency_weight: float = 0.3) -> list[dict]:
    """Weigh old vs new: boost recent items, but don't fully discard older,
    still-relevant context."""
    for r in results:
        r["final_score"] = r["relevance_score"] * (1 - recency_weight) + r["recency_score"] * recency_weight
    return sorted(results, key=lambda r: r["final_score"], reverse=True)

def generation_layer_with_citations(answer: str, sources: list[str]) -> dict:
    """Generation layer's core requirement: citation-backed response."""
    return {"answer": answer, "citations": sources}

print(ingestion_layer_process("email", "Subject: Refund status\nYour refund is processed."))
print("real change detected?", detect_real_change("t1", "t2", "hashA", "hashA"))  # False -- only timestamp changed
results = [{"text": "old policy", "relevance_score": 0.9, "recency_score": 0.2},
          {"text": "new policy", "relevance_score": 0.7, "recency_score": 0.95}]
print("weighted results:", retrieval_layer_weigh_recency(results))
print(generation_layer_with_citations("Refunds take 5 days.", ["gmail:thread_123", "docs:policy.pdf"]))