Day 50 — How to Build the Context Layer for Your Data Stack. Step by Step.

 #DataSeries | #50 Subtitle: 50 posts ago I asked: what if your data could talk back? Today, we build the answer.

⸻ THE MILESTONE 50 days ago, I wrote Day 1. The topic was Conversational Analytics. The idea was simple: your data should not sit in a dashboard waiting for you to look at it. It should be able to speak. To ask questions. To act. We have spent 49 days building toward that. Data Contracts (Day 15). GraphRAG (Day 23). LLM Evals (Day 45). MCP (Day 47). A2A (Day 48). Context Engineering as a concept (Day 49). Today we stop talking about it and start building it. This is the Context Layer build guide. Four components. Real tools. Real code patterns. Everything you need to take an Agent from "works in a demo" to "works in production." ⸻ WHY THE CONTEXT LAYER EXISTS Here is the problem in one sentence. Your Agent is brilliant. And it knows nothing about your business. It does not know what "Revenue" means in your company. It does not remember the anomaly it found last Tuesday. It does not know which tables are fresh and which ones are stale. It does not know it is not allowed to write to the production database. MCP gave it tools (Day 47). A2A gave it coworkers (Day 48). The Context Layer gives it the institutional knowledge to use both reliably. Let us build it. ⸻ COMPONENT 1: MEMORY "An Agent without memory starts from zero every single time. That is not intelligence. That is amnesia." Memory has two tiers. You need both. — TIER 1: SHORT-TERM MEMORY (In-Session) — Short-term memory is the Agent's working memory. Everything that has happened inside the current task: the inputs it received, the tools it called, the outputs it produced, the decisions it made. In LangGraph, this is handled natively by MemorySaver. Every node in your StateGraph takes the current state as input, updates it, and passes it to the next node. Nothing is lost inside a session. The core state object looks like this: from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from typing import TypedDict, List class AgentState(TypedDict): goal: str messages: List[dict] tool_results: List[dict] decisions: List[str] # what the Agent decided and why errors: List[str] # what went wrong current_step: str checkpointer = MemorySaver() graph = StateGraph(AgentState) # Every node receives and returns AgentState # LangGraph handles persistence automatically The thread_id is your session key. Different users, different workflows, different thread_ids. LangGraph ensures complete isolation between them. — TIER 2: LONG-TERM MEMORY (Cross-Session) — Long-term memory is institutional knowledge. What did this Agent find last Monday? What anomaly was flagged in Q3? What decision did the team override two weeks ago? You store this as vector embeddings so the Agent can query it semantically — not just "give me the last 5 runs" but "what have we seen before that looks like this anomaly?" The best options in 2026: # Option A: Redis (fastest, production-grade, dual-tier) from redisvl.extensions.llmcache import SemanticCache from redis import Redis client = Redis(host="localhost", port=6379) memory_store = SemanticCache( name="agent_memory", redis_client=client, distance_threshold=0.15 # semantic similarity cutoff ) # Store a decision after the run memory_store.store( prompt="Q3 revenue anomaly detected in store_sales table", response="Root cause: ETL delay from upstream POS system. Fixed by backfill." ) # Retrieve next time a similar situation arises hits = memory_store.check(prompt="revenue drop in store_sales") # Returns the stored context if semantically similar enough # Option B: Cognee on LangGraph (semantic knowledge graph) import cognee from langgraph.prebuilt import create_react_agent from cognee.api.v1.search import SearchType # Ingest your business knowledge once await cognee.add("Your KPI definitions, past decisions, anomaly history") await cognee.cognify() # builds the knowledge graph automatically # Search semantically in any agent node async def retrieve_memory(state: AgentState): results = await cognee.search( SearchType.INSIGHTS, query=state["goal"] ) state["messages"].append({"role": "context", "content": str(results)}) return state One critical rule on memory: build a forgetting mechanism. Stale, superseded, or low-confidence memories must be pruned. If the Agent stored "Revenue definition = gross sales" and you changed the definition six months ago, old memory is now a liability. Cognee supports explicit pruning. Redis TTL handles expiry automatically. ⸻ COMPONENT 2: STATE "State is the briefing sheet your Agent reads every morning before it starts work." Memory tells the Agent what happened before. State tells the Agent what is happening right now. Before your Agent executes a single tool call, it builds a State object by reading your live data infrastructure. Pipeline health. Table freshness. KPI threshold breaches. Business calendar context. Here is the State builder node pattern: import json from datetime import datetime, timezone def build_state_node(state: AgentState) -> AgentState: """ Runs first. Gives the Agent its morning briefing. Reads from: Airflow API, dbt metadata, your KPI store. """ live_state = { "timestamp": datetime.now(timezone.utc).isoformat(), "pipeline_health": get_airflow_dag_status(), # from Airflow REST API "table_freshness": get_dbt_source_freshness(), # from dbt metadata "kpi_alerts": get_kpi_threshold_breaches(), # from your BI layer "business_context": { "quarter_end": is_quarter_end(), "fiscal_week": get_fiscal_week() } } state["messages"].insert(0, { "role": "system", "content": f"Current operational state:\n{json.dumps(live_state, indent=2)}" }) return state # Wire it as Node 0 in your graph — fires before anything else graph.add_node("build_state", build_state_node) graph.set_entry_point("build_state") The State object is rebuilt fresh every run. It is not cached. You want the Agent reading reality, not a stale snapshot. ⸻ COMPONENT 3: THE SEMANTIC LAYER "If the Agent does not have a dictionary, it will invent its own definitions. And it will be confidently wrong." We covered the Semantic Layer in Day 5 and Day 47. Now it has a new consumer: your Agent. The Semantic Layer is the bridge between your raw schema and business meaning. It tells the Agent: — "Revenue" means net recognized revenue, excluding refunds, recognized at delivery date. — "Active User" means any account with a login event in the last 7 calendar days. — "Churn" means voluntary cancellation only, not payment failures. Without this, the Agent generates SQL that is syntactically correct and semantically wrong. Every time. Here is how to wire it via MCP (Day 47): # Step 1: Extract your dbt metric definitions into a vector store from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings import yaml, glob # Load all dbt model docs and metric definitions docs = [] for path in glob.glob("./dbt/models/**/*.yml", recursive=True): with open(path) as f: content = yaml.safe_load(f) docs.append(str(content)) semantic_store = Chroma.from_texts( texts=docs, embedding=OpenAIEmbeddings(), collection_name="semantic_layer" ) # Step 2: Expose as an MCP tool the Agent can call def resolve_business_term(term: str) -> str: """ Agent calls this before writing any SQL or forming any conclusion. Returns: the official business definition of the term. """ results = semantic_store.similarity_search(term, k=3) return "\n".join([r.page_content for r in results]) # Step 3: Add to your Agent's MCP tool list tools = [resolve_business_term, ...] # alongside your other MCP tools The Agent now calls resolve_business_term("revenue") before generating any SQL. It gets the canonical definition. It uses it. The SQL is correct. This is the Semantic Layer finally getting the consumer it always deserved — not a human clicking on a BI tool, but an Agent making decisions at machine speed. ⸻ COMPONENT 4: GUARDRAILS "Guardrails are the component most teams implement last. They should be implemented first." At a 5% per-step failure rate, an Agent running 20 steps will fail on a large portion of runs without guardrails. Production requires end-to-end failure rates well below 1%. That level of reliability does not come from a better model. It comes from guardrails as engineering. Three layers. All required. — LAYER 1: INPUT GUARDRAILS — Check before the Agent acts. Reject bad inputs before they become bad actions. import re from guardrails import Guard # Define your input policy as code input_guard = Guard().use_many( DetectPII(on_fail="exception"), # block PII from reaching the LLM ValidTopics( # scope the Agent's mandate valid_topics=["data_analysis", "reporting", "pipeline_ops"], on_fail="exception" ) ) def input_guardrail_node(state: AgentState) -> AgentState: try: input_guard.validate(state["messages"][-1]["content"]) except Exception as e: state["errors"].append(f"Input blocked: {e}") state["current_step"] = "HALTED" return state — LAYER 2: TOOL GUARDRAILS (Human-in-the-Loop) — Gate irreversible actions. The Agent can always read. It needs a human to write. from langgraph.graph import END def should_interrupt(state: AgentState) -> str: """ Conditional edge: check if the next action is irreversible. If yes: pause and wait for human approval. If no: continue automatically. """ irreversible_actions = ["delete_table", "send_email", "push_to_prod", "write_to_db"] next_action = state.get("current_step", "") if any(action in next_action for action in irreversible_actions): return "INTERRUPT" # → routes to human review node return "CONTINUE" # → routes to execution node # interrupt_before fires BEFORE a node executes graph.add_conditional_edges( "plan_action", should_interrupt, {"INTERRUPT": "human_review", "CONTINUE": "execute_action"} ) — LAYER 3: OUTPUT GUARDRAILS — Validate before delivering. Check format, compliance, and accuracy. output_guard = Guard().use_many( ValidJSON(on_fail="fix"), # enforce structured output format GDPRCompliance(on_fail="exception"), # block non-compliant outputs FactualConsistency( # verify claims match source data sources=state["tool_results"], on_fail="reask" # reask the model, don't just fail ) ) def output_guardrail_node(state: AgentState) -> AgentState: final_output = state["messages"][-1]["content"] validated = output_guard.validate(final_output) state["messages"][-1]["content"] = validated.validated_output return state ⸻ WIRING IT ALL TOGETHER Here is the complete Context Layer graph. Every production Agent should follow this pattern. from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver graph = StateGraph(AgentState) # Node 0: Build live state (pipeline health, KPIs, freshness) graph.add_node("build_state", build_state_node) # Node 1: Retrieve long-term memory (what happened before) graph.add_node("retrieve_memory", retrieve_memory_node) # Node 2: Resolve business terms via Semantic Layer graph.add_node("semantic_resolve", semantic_resolve_node) # Node 3: Input guardrails (scope, PII, prompt injection) graph.add_node("input_guardrails", input_guardrail_node) # Node 4: Core agent execution (your LLM + MCP tools + A2A coworkers) graph.add_node("execute", agent_execute_node) # Node 5: Human review (fires only for irreversible actions) graph.add_node("human_review", human_review_node) # Node 6: Output guardrails (format, compliance, fact-check) graph.add_node("output_guardrails", output_guardrail_node) # Node 7: Write back to long-term memory graph.add_node("write_memory", write_memory_node) # Edges graph.set_entry_point("build_state") graph.add_edge("build_state", "retrieve_memory") graph.add_edge("retrieve_memory", "semantic_resolve") graph.add_edge("semantic_resolve", "input_guardrails") graph.add_edge("input_guardrails", "execute") graph.add_conditional_edges("execute", should_interrupt, {"INTERRUPT": "human_review", "CONTINUE": "output_guardrails"}) graph.add_edge("human_review", "output_guardrails") graph.add_edge("output_guardrails", "write_memory") graph.add_edge("write_memory", END) app = graph.compile(checkpointer=MemorySaver()) This is the Context Layer. Eight nodes. Every Agent in production in 2026 that is actually working reliably has some version of this pattern — whether the team calls it that or not. ⸻ THE HONEST TRUTH None of this is glamorous. Building Memory schemas is data modeling. Building State management is pipeline engineering. Building a Semantic Layer is the documentation work your team has been avoiding for three years. Building Guardrails is governance — the thing everyone agrees is important and nobody wants to do first. But here is what 50 days of this series has been building toward: The companies winning in 2026 are not winning because they have access to better models. They are winning because they did the boring work. They modeled their data properly (Day 3, Day 15). They built observability into their pipelines (Day 6). They created active metadata and governance (Day 21). They treated their data as a product with an owner and an SLA (Day 3). The Context Layer is just all of that work — finally given the consumer it always deserved. Not a dashboard. Not a report. Not a human squinting at a chart at 9am. An Agent that acts on your behalf, reliably, at scale. That is what 50 days of writing has been building toward. ⸻

(Resources) Context engineering in agents — LangChain Docs: https://docs.langchain.com/oss/python/langchain/context-engineering AI Agent Architecture: Build Systems That Work in 2026 (Redis): https://redis.io/blog/ai-agent-architecture/ AI Agents in 2026: Practical Architecture for Tools, Memory, Evals, and Guardrails: https://andriifurmanets.com/blogs/ai-agents-2026-practical-architecture-tools-memory-evals-guardrails LangGraph + Cognee: Persistent Semantic Memory: https://www.cognee.ai/blog/integrations/langgraph-cognee-integration-build-langgraph-agents-with-persistent-cognee-memory Contextual Engineering Guide (GitHub): https://github.com/FareedKhan-dev/contextual-engineering-guide #DataSeries #ContextEngineering #AgenticAI #LangGraph #SemanticLayer #Guardrails #MCP #A2A #DataEngineering #AITrends2026 #FutureOfWork #DataLeadership #DataGovernance #Day50