Context Engineering Is What Your Agent Actually Needs

mkdir context-engine && cd context-engine
npm init -y
npm install ai @ai-sdk/openai zod
npm install -D typescript @types/node tsx
npx tsc --init --target ES2022 --module NodeNext --moduleResolution NodeNext --outDir dist

mkdir context_engine && cd context_engine
python -m venv .venv && source .venv/bin/activate
pip install openai numpy tiktoken

export OPENAI_API_KEY="sk-..."

import { z } from "zod";
 
// Zod validates config at runtime -- catches missing fields before they silently degrade output
const SystemPromptConfig = z.object({
  role: z.string(),
  constraints: z.array(z.string()),
  personality: z.string().optional(),
  escalationRules: z.array(z.string()).optional(),
  outputFormat: z.string().optional(),
});
 
type SystemPromptConfig = z.infer<typeof SystemPromptConfig>;
 
function buildSystemPrompt(
  config: SystemPromptConfig,
  injectedContext: { memories?: string; knowledge?: string }
): string {
  const sections: string[] = [];
 
  // Role first: models weight the beginning of context most heavily
  sections.push(`<role>\n${config.role}\n</role>`);
 
  // Constraints next: behavioral boundaries the model must not cross
  if (config.constraints.length > 0) {
    sections.push(
      `<constraints>\n${config.constraints.map((c) => `- ${c}`).join("\n")}\n</constraints>`
    );
  }
 
  // Memories injected per-request -- this section doesn't exist at design time
  if (injectedContext.memories) {
    sections.push(
      `<customer_context>\nRelevant information from previous interactions:\n${injectedContext.memories}\n</customer_context>`
    );
  }
 
  // RAG results injected per-request -- grounds the model in facts, prevents hallucination
  if (injectedContext.knowledge) {
    sections.push(
      `<knowledge>\nRelevant documentation:\n${injectedContext.knowledge}\n</knowledge>`
    );
  }
 
  // Escalation rules: safety net for situations the agent shouldn't handle alone
  if (config.escalationRules?.length) {
    sections.push(
      `<escalation>\n${config.escalationRules.map((r) => `- ${r}`).join("\n")}\n</escalation>`
    );
  }
 
  // XML sections help the model attend to specific blocks vs. parsing a wall of text
  return sections.join("\n\n");
}
 
// Usage
const systemPrompt = buildSystemPrompt(
  {
    role: "You are a customer support agent for Acme Corp. You help customers with billing questions, account issues, and product information.",
    constraints: [
      "Never reveal internal pricing formulas or discount logic",
      "Always verify customer identity before accessing account details",
      "If you cannot resolve an issue in 3 attempts, offer to escalate to a human agent",
    ],
    escalationRules: [
      "Escalate immediately if customer mentions legal action",
      "Escalate if customer sentiment is consistently negative for 3+ turns",
    ],
  },
  {
    memories: "Customer prefers email communication. Has been a subscriber since 2024. Previously had a billing dispute resolved in their favor.",
    knowledge: "Current promotion: 20% off annual plans through March 31. Refund policy: full refund within 30 days, prorated after.",
  }
);

from dataclasses import dataclass, field
 
@dataclass
class SystemPromptConfig:
    role: str
    constraints: list[str] = field(default_factory=list)
    personality: str | None = None
    escalation_rules: list[str] = field(default_factory=list)
    output_format: str | None = None
 
def build_system_prompt(
    config: SystemPromptConfig,
    memories: str | None = None,
    knowledge: str | None = None,
) -> str:
    sections = []
 
    # Role first: models weight the beginning of context most heavily
    sections.append(f"<role>\n{config.role}\n</role>")
 
    # Hard behavioral boundaries the model must not cross
    if config.constraints:
        items = "\n".join(f"- {c}" for c in config.constraints)
        sections.append(f"<constraints>\n{items}\n</constraints>")
 
    # Per-request injection -- doesn't exist at design time, populated from memory store
    if memories:
        sections.append(
            f"<customer_context>\nRelevant information from previous interactions:\n{memories}\n</customer_context>"
        )
 
    # Per-request injection -- RAG results ground the model in facts
    if knowledge:
        sections.append(
            f"<knowledge>\nRelevant documentation:\n{knowledge}\n</knowledge>"
        )
 
    # Safety net for situations the agent shouldn't handle alone
    if config.escalation_rules:
        items = "\n".join(f"- {r}" for r in config.escalation_rules)
        sections.append(f"<escalation>\n{items}\n</escalation>")
 
    return "\n\n".join(sections)
 
# Usage
config = SystemPromptConfig(
    role="You are a customer support agent for Acme Corp.",
    constraints=[
        "Never reveal internal pricing formulas",
        "Verify customer identity before accessing account details",
    ],
    escalation_rules=[
        "Escalate immediately if customer mentions legal action",
    ],
)
 
system_prompt = build_system_prompt(
    config,
    memories="Customer prefers email. Subscriber since 2024.",
    knowledge="Current promotion: 20% off annual plans through March 31.",
)

interface RetrievedChunk {
  content: string;
  score: number;
  source: string;
  tokenCount: number;
}
 
interface TokenBudget {
  knowledge: number;  // Hard ceiling -- RAG cannot exceed this
  memory: number;     // Hard ceiling -- memory injection
  history: number;    // Hard ceiling -- conversation history
  tools: number;      // Hard ceiling -- tool schemas
  system: number;     // Hard ceiling -- system prompt
}
 
// These numbers enforce the 60/20/20 split described below
const DEFAULT_BUDGET: TokenBudget = {
  system: 2000,
  knowledge: 3000,
  memory: 1000,
  history: 8000,
  tools: 4000,
};
 
async function retrieveWithBudget(
  query: string,
  budget: number,
  retriever: { search(query: string, topK: number): Promise<RetrievedChunk[]> }
): Promise<string> {
  // Over-fetch, then trim -- cheaper than multiple retrieval calls
  const chunks = await retriever.search(query, 10);
 
  // 0.7 threshold: below this, chunks inject noise that dilutes attention
  const relevant = chunks.filter((c) => c.score >= 0.7);
 
  // Greedy packing: highest-score chunks first until budget exhausted
  const selected: RetrievedChunk[] = [];
  let tokenCount = 0;
 
  for (const chunk of relevant) {
    if (tokenCount + chunk.tokenCount > budget) break;
    selected.push(chunk);
    tokenCount += chunk.tokenCount;
  }
 
  if (selected.length === 0) return "";
 
  // Source attribution helps the model cite its reasoning
  return selected
    .map((c, i) => `[Source ${i + 1}: ${c.source}]\n${c.content}`)
    .join("\n\n---\n\n");
}

from dataclasses import dataclass
 
@dataclass
class RetrievedChunk:
    content: str
    score: float
    source: str
    token_count: int
 
# Hard ceilings per layer -- prevents any single layer from starving the others
DEFAULT_BUDGET = {
    "system": 2000,
    "knowledge": 3000,
    "memory": 1000,
    "history": 8000,
    "tools": 4000,
}
 
async def retrieve_with_budget(
    query: str,
    budget: int,
    retriever,  # has async search(query, top_k) -> list[RetrievedChunk]
) -> str:
    # Over-fetch, then trim -- cheaper than multiple retrieval calls
    chunks = await retriever.search(query, top_k=10)
 
    # 0.7 threshold: below this, chunks inject noise that dilutes attention
    relevant = [c for c in chunks if c.score >= 0.7]
 
    # Greedy packing: highest-score chunks first until budget exhausted
    selected = []
    token_count = 0
 
    for chunk in relevant:
        if token_count + chunk.token_count > budget:
            break
        selected.append(chunk)
        token_count += chunk.token_count
 
    if not selected:
        return ""
 
    sections = []
    for i, c in enumerate(selected, 1):
        sections.append(f"[Source {i}: {c.source}]\n{c.content}")
 
    return "\n\n---\n\n".join(sections)

interface Memory {
  id: string;
  content: string;
  category: "preference" | "fact" | "interaction" | "feedback";
  createdAt: Date;
  embedding: number[];
}
 
interface MemorySearchResult {
  memory: Memory;
  score: number;
}
 
class MemoryInjector {
  private minScore = 0.3;   // Lower than RAG (0.7) because memories are broader signals
  private maxMemories = 10;  // Cap prevents flooding context with stale facts
  private tokenBudget = 1000;
 
  async injectMemories(
    customerId: string,
    currentMessage: string,
    memoryStore: {
      search(
        customerId: string,
        query: string,
        options: { minScore: number; limit: number }
      ): Promise<MemorySearchResult[]>;
    }
  ): Promise<string> {
    // The current message IS the search query -- "billing address" retrieves billing-related memories
    const results = await memoryStore.search(customerId, currentMessage, {
      minScore: this.minScore,
      limit: this.maxMemories,
    });
 
    if (results.length === 0) return "";  // No memories = no tokens wasted
 
    // Grouping by category creates structure the model can attend to selectively
    const grouped = new Map<string, string[]>();
    for (const { memory } of results) {
      const items = grouped.get(memory.category) || [];
      items.push(memory.content);
      grouped.set(memory.category, items);
    }
 
    const sections: string[] = [];
 
    if (grouped.has("preference")) {
      sections.push(
        `Customer preferences:\n${grouped.get("preference")!.map((m) => `- ${m}`).join("\n")}`
      );
    }
 
    if (grouped.has("fact")) {
      sections.push(
        `Known facts:\n${grouped.get("fact")!.map((m) => `- ${m}`).join("\n")}`
      );
    }
 
    if (grouped.has("interaction")) {
      sections.push(
        `Previous interactions:\n${grouped.get("interaction")!.map((m) => `- ${m}`).join("\n")}`
      );
    }
 
    return sections.join("\n\n");
  }
}

from dataclasses import dataclass
from collections import defaultdict
 
@dataclass
class Memory:
    id: str
    content: str
    category: str  # "preference" | "fact" | "interaction" | "feedback"
    created_at: str
    embedding: list[float]
 
@dataclass
class MemorySearchResult:
    memory: Memory
    score: float
 
class MemoryInjector:
    def __init__(self, min_score: float = 0.3, max_memories: int = 10):
        self.min_score = min_score
        self.max_memories = max_memories
 
    async def inject_memories(
        self,
        customer_id: str,
        current_message: str,
        memory_store,  # has async search(customer_id, query, min_score, limit)
    ) -> str:
        results = await memory_store.search(
            customer_id,
            current_message,
            min_score=self.min_score,
            limit=self.max_memories,
        )
 
        if not results:
            return ""
 
        # Group by category
        grouped: dict[str, list[str]] = defaultdict(list)
        for result in results:
            grouped[result.memory.category].append(result.memory.content)
 
        sections = []
 
        if "preference" in grouped:
            items = "\n".join(f"- {m}" for m in grouped["preference"])
            sections.append(f"Customer preferences:\n{items}")
 
        if "fact" in grouped:
            items = "\n".join(f"- {m}" for m in grouped["fact"])
            sections.append(f"Known facts:\n{items}")
 
        if "interaction" in grouped:
            items = "\n".join(f"- {m}" for m in grouped["interaction"])
            sections.append(f"Previous interactions:\n{items}")
 
        return "\n\n".join(sections)

import { generateText } from "ai";
import { openai } from "@ai-sdk/openai";
 
interface Message {
  role: "user" | "assistant" | "system";
  content: string;
}
 
interface CompressedHistory {
  summary: string | null;
  recentMessages: Message[];
  originalCount: number;
  compressedTokens: number;
}
 
class HistoryCompressor {
  private maxTokens: number;
  private recentTurnCount: number;
  // gpt-4o-mini for summaries: 16x cheaper than gpt-4o, good enough for compression
  private summaryModel = openai("gpt-4o-mini");
 
  constructor(maxTokens = 8000, recentTurnCount = 10) {
    this.maxTokens = maxTokens;
    this.recentTurnCount = recentTurnCount;
  }
 
  async compress(messages: Message[]): Promise<CompressedHistory> {
    const estimatedTokens = this.estimateTokens(messages);
 
    // No compression needed
    if (estimatedTokens <= this.maxTokens) {
      return {
        summary: null,
        recentMessages: messages,
        originalCount: messages.length,
        compressedTokens: estimatedTokens,
      };
    }
 
    // Split point: recent turns stay verbatim (customer expects you to remember what was just said)
    const recentMessages = messages.slice(-this.recentTurnCount * 2); // *2 for user+assistant pairs
    const olderMessages = messages.slice(0, -this.recentTurnCount * 2);
 
    if (olderMessages.length === 0) {
      return {
        summary: null,
        recentMessages: messages,
        originalCount: messages.length,
        compressedTokens: estimatedTokens,
      };
    }
 
    // Summarize older messages
    const conversationText = olderMessages
      .map((m) => `${m.role}: ${m.content}`)
      .join("\n");
 
    const { text: summary } = await generateText({
      model: this.summaryModel,
      system: `You are a conversation summarizer. Extract key facts, decisions,
and unresolved issues from the conversation. Be concise but preserve all
actionable information. Format as bullet points.`,
      prompt: `Summarize this conversation:\n\n${conversationText}`,
    });
 
    return {
      summary,
      recentMessages,
      originalCount: messages.length,
      compressedTokens: this.estimateTokens(recentMessages) + (summary.length / 4),
    };
  }
 
  private estimateTokens(messages: Message[]): number {
    // Rough estimate: 1 token ~= 4 chars for English text.
    // For production, use tiktoken-node or the AI SDK's token counting.
    // This heuristic overestimates code/JSON by ~20% and underestimates CJK by ~40%.
    return messages.reduce((sum, m) => sum + Math.ceil(m.content.length / 4), 0);
  }
}

from openai import AsyncOpenAI
from dataclasses import dataclass
 
@dataclass
class CompressedHistory:
    summary: str | None
    recent_messages: list[dict]
    original_count: int
    compressed_tokens: int
 
class HistoryCompressor:
    def __init__(
        self,
        max_tokens: int = 8000,
        recent_turn_count: int = 10,
    ):
        self.max_tokens = max_tokens
        self.recent_turn_count = recent_turn_count
        self.client = AsyncOpenAI()
 
    async def compress(self, messages: list[dict]) -> CompressedHistory:
        estimated = self._estimate_tokens(messages)
 
        if estimated <= self.max_tokens:
            return CompressedHistory(
                summary=None,
                recent_messages=messages,
                original_count=len(messages),
                compressed_tokens=estimated,
            )
 
        # Split into older (summarize) and recent (keep verbatim)
        split_idx = -self.recent_turn_count * 2
        recent = messages[split_idx:]
        older = messages[:split_idx]
 
        if not older:
            return CompressedHistory(
                summary=None,
                recent_messages=messages,
                original_count=len(messages),
                compressed_tokens=estimated,
            )
 
        conversation_text = "\n".join(
            f"{m['role']}: {m['content']}" for m in older
        )
 
        response = await self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "You are a conversation summarizer. Extract key facts, "
                        "decisions, and unresolved issues. Be concise but preserve "
                        "all actionable information. Format as bullet points."
                    ),
                },
                {"role": "user", "content": f"Summarize:\n\n{conversation_text}"},
            ],
        )
 
        summary = response.choices[0].message.content
 
        return CompressedHistory(
            summary=summary,
            recent_messages=recent,
            original_count=len(messages),
            compressed_tokens=(
                self._estimate_tokens(recent) + len(summary or "") // 4
            ),
        )
 
    def _estimate_tokens(self, messages: list[dict]) -> int:
        # Fast approximation. For production accuracy, use:
        # import tiktoken; enc = tiktoken.encoding_for_model("gpt-4o")
        # return sum(len(enc.encode(m.get("content", ""))) for m in messages)
        return sum(len(m.get("content", "")) // 4 for m in messages)

interface ToolDefinition {
  name: string;
  description: string;
  inputSchema: Record<string, unknown>;
  tokenCost: number; // Estimated tokens for this tool's schema
}
 
interface ToolSet {
  id: string;
  name: string;
  tools: ToolDefinition[];
}
 
class ToolResolver {
  private tokenBudget: number;
 
  constructor(tokenBudget = 4000) {
    this.tokenBudget = tokenBudget;
  }
 
  async resolveTools(
    agentConfig: {
      toolsetIds: string[];
      toolIds: string[];
    },
    toolsetStore: { get(id: string): Promise<ToolSet> },
    conversationContext?: string // Current conversation topic for relevance filtering
  ): Promise<ToolDefinition[]> {
    const allTools: ToolDefinition[] = [];
 
    // Toolsets group related tools (e.g., "Billing", "CRM") -- load only what's relevant
    for (const tsId of agentConfig.toolsetIds) {
      const toolset = await toolsetStore.get(tsId);
      allTools.push(...toolset.tools);
    }
 
    // Fallback: if no toolsets configured, resolve individual tool IDs
    if (agentConfig.toolsetIds.length === 0 && agentConfig.toolIds.length > 0) {
      // Agent-scoped resolution -- implementation depends on your registry
    }
 
    // Hard budget enforcement: surplus tools get dropped, not squeezed in
    let tokenCount = 0;
    const selected: ToolDefinition[] = [];
 
    for (const tool of allTools) {
      if (tokenCount + tool.tokenCost > this.tokenBudget) {
        console.warn(
          `Tool budget exceeded. Included ${selected.length}/${allTools.length} tools.`
        );
        break;
      }
      selected.push(tool);
      tokenCount += tool.tokenCost;
    }
 
    return selected;
  }
}

from dataclasses import dataclass
 
@dataclass
class ToolDefinition:
    name: str
    description: str
    input_schema: dict
    token_cost: int  # Estimated tokens for schema
 
@dataclass
class ToolSet:
    id: str
    name: str
    tools: list[ToolDefinition]
 
class ToolResolver:
    def __init__(self, token_budget: int = 4000):
        self.token_budget = token_budget
 
    async def resolve_tools(
        self,
        toolset_ids: list[str],
        tool_ids: list[str],
        toolset_store,  # has async get(id) -> ToolSet
    ) -> list[ToolDefinition]:
        all_tools: list[ToolDefinition] = []
 
        # Resolve from toolsets (MCP-style grouped tools)
        for ts_id in toolset_ids:
            toolset = await toolset_store.get(ts_id)
            all_tools.extend(toolset.tools)
 
        # Agent-scoped fallback
        if not toolset_ids and tool_ids:
            pass  # Resolve individual tools from registry
 
        # Budget enforcement
        token_count = 0
        selected: list[ToolDefinition] = []
 
        for tool in all_tools:
            if token_count + tool.token_cost > self.token_budget:
                print(
                    f"Tool budget exceeded. "
                    f"Included {len(selected)}/{len(all_tools)} tools."
                )
                break
            selected.append(tool)
            token_count += tool.token_cost
 
        return selected

import { generateText } from "ai";
import { openai } from "@ai-sdk/openai";
 
interface ContextEngineConfig {
  model: Parameters<typeof openai>[0];
  maxContextTokens: number;
  budget: TokenBudget;
}
 
class ContextEngine {
  private systemPromptBuilder: typeof buildSystemPrompt;
  private memoryInjector: MemoryInjector;
  private historyCompressor: HistoryCompressor;
  private toolResolver: ToolResolver;
  private config: ContextEngineConfig;
 
  constructor(config: ContextEngineConfig) {
    this.config = config;
    this.memoryInjector = new MemoryInjector();
    this.historyCompressor = new HistoryCompressor(config.budget.history);
    this.toolResolver = new ToolResolver(config.budget.tools);
    this.systemPromptBuilder = buildSystemPrompt;
  }
 
  async processMessage(
    message: string,
    context: {
      customerId: string;
      agentConfig: SystemPromptConfig;
      toolsetIds: string[];
      conversationHistory: Message[];
      // Injected dependencies
      memoryStore: any;
      knowledgeRetriever: any;
      toolsetStore: any;
    }
  ) {
    // 1. Parallel retrieval saves 200-300ms vs. sequential -- critical for voice agents
    const [memories, knowledge, tools] = await Promise.all([
      this.memoryInjector.injectMemories(
        context.customerId,
        message,
        context.memoryStore
      ),
      retrieveWithBudget(
        message,
        this.config.budget.knowledge,
        context.knowledgeRetriever
      ),
      this.toolResolver.resolveTools(
        { toolsetIds: context.toolsetIds, toolIds: [] },
        context.toolsetStore
      ),
    ]);
 
    // 2. Compress history if needed
    const history = await this.historyCompressor.compress(
      context.conversationHistory
    );
 
    // 3. Build system prompt with injected context
    const systemPrompt = this.systemPromptBuilder(context.agentConfig, {
      memories,
      knowledge,
    });
 
    // 4. Assemble messages array
    const messages: Message[] = [];
 
    // Add compressed history summary if it exists
    if (history.summary) {
      messages.push({
        role: "system",
        content: `Previous conversation summary:\n${history.summary}`,
      });
    }
 
    // Add recent messages
    messages.push(...history.recentMessages);
 
    // Add current message
    messages.push({ role: "user", content: message });
 
    // 5. maxSteps: 10 allows multi-turn tool use (call tool, read result, call another)
    const result = await generateText({
      model: openai(this.config.model),
      system: systemPrompt,
      messages,
      tools: this.convertTools(tools),
      maxSteps: 10,
    });
 
    return {
      response: result.text,
      toolCalls: result.steps.flatMap((s) => s.toolCalls),
      tokenUsage: result.usage,
    };
  }
 
  private convertTools(tools: ToolDefinition[]) {
    // Convert to AI SDK tool format
    const converted: Record<string, any> = {};
    for (const tool of tools) {
      converted[tool.name] = {
        description: tool.description,
        parameters: tool.inputSchema,
      };
    }
    return converted;
  }
}
 
// Usage
const engine = new ContextEngine({
  model: "gpt-4o",
  maxContextTokens: 128000,
  budget: DEFAULT_BUDGET,
});

import asyncio
from openai import AsyncOpenAI
from dataclasses import dataclass
 
@dataclass
class ContextEngineConfig:
    model: str = "gpt-4o"
    max_context_tokens: int = 128000
    budget: dict = None
 
    def __post_init__(self):
        if self.budget is None:
            self.budget = DEFAULT_BUDGET
 
class ContextEngine:
    def __init__(self, config: ContextEngineConfig):
        self.config = config
        self.client = AsyncOpenAI()
        self.memory_injector = MemoryInjector()
        self.history_compressor = HistoryCompressor(
            max_tokens=config.budget["history"]
        )
        self.tool_resolver = ToolResolver(
            token_budget=config.budget["tools"]
        )
 
    async def process_message(
        self,
        message: str,
        customer_id: str,
        agent_config: SystemPromptConfig,
        toolset_ids: list[str],
        conversation_history: list[dict],
        memory_store,
        knowledge_retriever,
        toolset_store,
    ) -> dict:
        # 1. Parallel retrieval saves 200-300ms vs. sequential -- critical for voice agents
        memories, knowledge, tools = await asyncio.gather(
            self.memory_injector.inject_memories(
                customer_id, message, memory_store
            ),
            retrieve_with_budget(
                message, self.config.budget["knowledge"], knowledge_retriever
            ),
            self.tool_resolver.resolve_tools(
                toolset_ids, [], toolset_store
            ),
        )
 
        # 2. Compress history
        history = await self.history_compressor.compress(
            conversation_history
        )
 
        # 3. Build system prompt
        system_prompt = build_system_prompt(
            agent_config, memories=memories, knowledge=knowledge
        )
 
        # 4. Assemble messages
        messages = []
 
        if history.summary:
            messages.append({
                "role": "system",
                "content": f"Previous conversation summary:\n{history.summary}",
            })
 
        messages.extend(history.recent_messages)
        messages.append({"role": "user", "content": message})
 
        # 5. Call LLM
        tool_defs = [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.input_schema,
                },
            }
            for t in tools
        ]
 
        response = await self.client.chat.completions.create(
            model=self.config.model,
            messages=[
                {"role": "system", "content": system_prompt},
                *messages,
            ],
            tools=tool_defs if tool_defs else None,
        )
 
        return {
            "response": response.choices[0].message.content,
            "usage": {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
            },
        }