This tutorial provides a complete guide to implementing the Context Engineering framework for AI agents: the architecture presented is optimized to prevent context rot, reduce execution latency, and lower token-processing overhead by dynamically managing LLM attention windows.
What is Context Engineering for AI Agents?
Context Engineering is the practice of dynamically curating, pruning, and optimizing the prompt context of a Large Language Model (LLM) at runtime. Rather than feeding a static, bloated document into the prompt (“prompt stuffing”), context engineering continuously manages active memory to maintain a high signal-to-noise ratio, ensuring optimal reasoning accuracy and lower latency.
The Technical Problem: Why “Prompt Stuffing” Fails
Modern LLMs advertise massive context windows (up to 1M+ tokens), but reliance on these large windows introduces architectural bottlenecks:
- Quadratic Scaling Latency (
O(n2)O(n2)): Transformer attention computation complexity scales quadratically with sequence length, leading to significant delays in Time-to-First-Token (TTFT) and overall token generation speeds. - Context Rot & Attention Decay: As sequence history grows, model recall degrades (known as “lost in the middle” phenomena). The model struggles to locate and act on critical operational rules or historical configurations.
- High Token Costs: Passing long raw logs, full file schemas, and redundant conversation history through every agent iteration wastes resources.
The 4 Core Pillars of Context Engineering
To optimize LLM resource efficiency, this architecture implements the four pillars defined in Anthropic’s engineering framework:
┌────────────────────────────────────────────────────────────────────────┐
│ CONTEXT ENGINEERING LOOP │
├───────────────────────┬───────────────────────┬────────────────────────┤
│ 1. Just-In-Time (JIT) │ 2. Active Compaction │ 3. Structured Note-Taking │
│ Paging & Metadata │ & Tool Stripping │ & External Memory │
└───────────────────────┴───────────────────────┴────────────────────────┘
│
▼
4. Sub-Agent Task Isolation- Just-In-Time (JIT) File Access: The agent uses metadata tools to check schemas first, only pulling specific data slices (read_file_chunk) when needed.
- Conversation History Compaction: The orchestrator monitors token usage. When thresholds are exceeded, it condenses historical dialogue turns and removes raw tool output strings.
- Structured Note-Taking (State Persistence): Key data points are stored externally in a state file (memory.json) rather than keeping them in the active prompt history.
- Sub-Agent Isolation: Computationally heavy analytics tasks are routed to a temporary sub-agent with a clean context, returning only a concise summary to the main agent.
Step-by-Step Python Implementation Guide
This step-by-step setup guides you through implementing this context engineering architecture with a local-first Python stack.
Directory Structure
context_engineered_agent/
│
├── data/
│ ├── user_activity.csv # Large Mock Dataset
│ └── memory.json # Persistent Agentic State
│
├── config.py # Token & API Configurations
├── utils.py # Token Heuristics & JSON Extractors
├── tools.py # JIT & Persistent Storage Tools
├── llm_client.py # Local Ollama HTTP Connection Client
├── agent.py # Main Orchestration Loop
└── main.py # Entry Point RunnerCode language: PHP (php)Step 1: Configuration Management (config.py)
Define execution thresholds. We set a low token threshold (1500 tokens) to demonstrate the active compaction engine using standard datasets.
# config.py
MAX_CONTEXT_TOKENS = 1500 # Low limit to trigger compaction in testing
OLLAMA_MODEL = "llama3"
OLLAMA_URL = "http://localhost:11434/api/chat"Code language: PHP (php)Step 2: Creating Token Utilities and Parsers (utils.py)
LLMs often output markdown formatting around raw JSON payloads. We implement a parser to extract JSON blocks reliably and apply a character-to-token heuristic to manage memory without external dependencies.
# utils.py
import re
import json
def estimate_tokens(text: str) -> int:
"""Calculates approximate token volume using standard char-to-token distribution."""
return max(1, len(text) // 4)
def calculate_history_tokens(messages) -> int:
"""Sums the token volume of a conversation thread."""
return sum(estimate_tokens(msg.get("content", "")) for msg in messages)
def extract_json(text: str):
"""Parses JSON blocks from raw LLM text outputs."""
text = text.strip()
match = re.search(r"```(?:json)?\s*({.*?})\s*```", text, re.DOTALL)
if match:
try:
return json.loads(match.group(1))
except json.JSONDecodeError:
pass
try:
return json.loads(text)
except json.JSONDecodeError:
pass
start = text.find("{")
end = text.rfind("}")
if start != -1 and end != -1:
try:
return json.loads(text[start:end+1])
except json.JSONDecodeError:
pass
return NoneCode language: Python (python)Step 3: Implementing Just-In-Time (JIT) File Access (tools.py)
Instead of dumping the raw user_activity.csv into the prompt, the agent uses hierarchical metadata and pagination tools. This approach keeps the context window clear of raw data overhead.
# tools.py
import os
import csv
import json
class ToolManager:
def __init__(self, data_dir="data", memory_file="memory.json"):
self.data_dir = data_dir
self.memory_file = os.path.join(data_dir, memory_file)
if not os.path.exists(self.memory_file):
with open(self.memory_file, "w") as f:
json.dump({}, f)
def list_files(self):
"""Returns files available in data directory."""
return os.listdir(self.data_dir)
def get_file_metadata(self, filename):
"""JIT Tool: Checks column names and line counts without loading raw records."""
filepath = os.path.join(self.data_dir, filename)
if not os.path.exists(filepath):
return {"error": f"File {filename} not found."}
size = os.path.getsize(filepath)
with open(filepath, "r") as f:
reader = csv.reader(f)
headers = next(reader, [])
line_count = 1 + sum(1 for _ in reader)
return {
"filename": filename,
"size_bytes": size,
"total_lines": line_count,
"headers": headers
}
def read_file_chunk(self, filename, start_line, limit=10):
"""JIT Tool: Loads a specific range of lines to prevent token bloat."""
filepath = os.path.join(self.data_dir, filename)
if not os.path.exists(filepath):
return {"error": f"File {filename} not found."}
chunk = []
with open(filepath, "r") as f:
reader = csv.reader(f)
headers = next(reader, None)
if headers:
chunk.append(headers)
for _ in range(start_line - 1):
next(reader, None)
for _ in range(limit):
row = next(reader, None)
if row is None:
break
chunk.append(row)
return {
"chunk_lines": len(chunk),
"data": chunk,
"range": f"lines {start_line} to {start_line + len(chunk) - 2}"
}
def write_note(self, key, value):
"""Structured Memory: Saves verified insights to a persistent JSON state."""
data = {}
if os.path.exists(self.memory_file):
with open(self.memory_file, "r") as f:
try:
data = json.load(f)
except json.JSONDecodeError:
pass
data[key] = value
with open(self.memory_file, "w") as f:
json.dump(data, f, indent=2)
return {"status": "success", "message": f"Saved {key} to persistent state."}
def read_notes(self):
"""Reads persistent agentic notes."""
if os.path.exists(self.memory_file):
with open(self.memory_file, "r") as f:
return json.load(f)
return {}Code language: Python (python)Step 4: The Conversational Compaction Engine (agent.py)
The orchestration logic measures active context before each inference step. If the token count exceeds the configured limit (MAX_CONTEXT_TOKENS), the orchestrator triggers a background condensation step. This step replaces older conversation turns and raw tool outputs with a concise progress report, keeping the context window high-signal.
# agent.py (Extract of the compaction method)
def _ensure_compact_context(self):
"""Checks token usage and compacts history when limits are exceeded."""
current_tokens = calculate_history_tokens(self.history)
print(f" -> Context Load: ~{current_tokens} tokens.")
if current_tokens > MAX_CONTEXT_TOKENS:
print(" -> [Compaction] Context threshold met. Summarizing history...")
# Keep System Prompt [0] and the most recent interaction steps
keep_count = 2
mid_history = self.history[1:-keep_count]
recent_history = self.history[-keep_count:]
compaction_prompt = (
"You are an agent memory compaction module. Summarize the progress of this run. "
"Highlight discovered configurations, metrics, and immediate objectives. "
"Keep the output summary under 120 words.\n\n"
f"History to condense:\n{json.dumps(mid_history, indent=2)}"
)
summary = self.llm.chat([{"role": "user", "content": compaction_prompt}])
# Reconstruct history with compacted summary and persistent memory notes
self.history = [
self.history[0], # Core System Instructions
{"role": "system", "content": f"[CONSOLIDATED INTERACTION BACKLOG]\n{summary}"},
{"role": "system", "content": f"Active memory notes: {json.dumps(self.tools.read_notes())}"}
]
self.history.extend(recent_history)
print(f" -> [Compaction Complete] Reduced to ~{calculate_history_tokens(self.history)} tokens.")Code language: Python (python)Step 5: Sub-Agent Task Offloading
When the primary agent encounters a computationally heavy or repetitive analytic task, it spins up an isolated sub-agent. This sub-agent executes its task, writes the results, and hands back a short summary, keeping its detailed workspace history separate from the main controller.
# agent.py (Extract of the sub-agent pipeline)
def _run_isolated_sub_agent(self, sub_task):
"""Spawns an isolated sub-agent with a clean context."""
print(f"\n--- [Sub-Agent] Spawning isolated runner for: '{sub_task}' ---")
sub_system = (
"You are an analytical sub-agent. Solve the task step-by-step. "
"Output JSON with either 'tool' (get_file_metadata, read_file_chunk) or 'response'."
)
sub_history = [
{"role": "system", "content": sub_system},
{"role": "user", "content": sub_task}
]
for step in range(3):
raw = self.llm.chat(sub_history)
parsed = extract_json(raw)
if not parsed:
continue
if "response" in parsed:
print(f"--- [Sub-Agent] Finished: {parsed['response']} ---")
return {"status": "success", "result": parsed["response"]}
tool_name = parsed.get("tool")
args = parsed.get("arguments", {})
if tool_name in ["get_file_metadata", "read_file_chunk"]:
res = self._execute_tool(tool_name, args)
else:
res = {"error": f"Tool {tool_name} is restricted for sub-agents."}
sub_history.append({"role": "assistant", "content": raw})
sub_history.append({"role": "user", "content": json.dumps(res)})
return {"status": "failure", "reason": "Execution steps exceeded."}Code language: Python (python)FAQ
How does Context Engineering differ from Prompt Engineering?
Prompt engineering focuses on organizing instructions and static input text before inference. Context engineering actively manages memory context at runtime, using techniques like dynamic file pagination, tool trace compaction, and external database updates to control the active token count.
Why does long conversation history degrade LLM performance?
Transformers use self-attention mechanics to compute relationships between every token in a sequence. As length scales, the attention signal dilutes, making it harder for the model to retrieve instructions. Keeping the context slim and focused on high-signal data helps maintain higher reasoning accuracy and lower latency.
How do sub-agents help optimize context budgets?
If a primary orchestrator handles every data operation directly, all raw tool payloads, schemas, and query steps remain in its active chat history. Routing complex sub-tasks to an isolated runner confines those intermediate tokens to a temporary context. Once the sub-task is complete, the sub-agent returns a concise summary, and the detailed log history is discarded.
Github Codes
You can download the codes on our Github repo, just download and follow the README steps. If need help you can always reach out our team on Discord 🤙
Start your free 30-day trial at regolo.ai and deploy LLMs with complete privacy by design.
👉 Talk with our Engineers or Start your 30 days free →
- Discord – Share your thoughts
- GitHub Repo – Code of blog articles ready to start
- Follow Us on X @regolo_ai
- Open discussion on our Subreddit Community
Built with ❤️ by the Regolo team. Questions? regolo.ai/contact or chat with us on Discord