Context engineering works seamlessly with 7 memory managers, 7 state managers, and 14 ML/AI infrastructure modules.
Context Engineering Framework
The Context module provides sophisticated tools for managing and optimizing the context provided to AI agents, ensuring relevant information is available while staying within token limits.
fromagenticaiframework.contextimport(ContextManager,ContextType,ContextPriority,ContextItem)# Create context managercontext_mgr=ContextManager(max_tokens=8000,model="gpt-4"# For accurate token counting)# Add context itemscontext_mgr.add(ContextItem(content="You are a helpful AI assistant.",type=ContextType.SYSTEM,priority=ContextPriority.CRITICAL))context_mgr.add(ContextItem(content="User is asking about Python programming.",type=ContextType.TASK,priority=ContextPriority.HIGH))
fromagenticaiframework.contextimportContextType# Available context typesContextType.SYSTEM# System instructionsContextType.CONVERSATION# Chat historyContextType.KNOWLEDGE# Retrieved knowledgeContextType.TASK# Current task detailsContextType.TOOL_RESULT# Tool execution resultsContextType.USER_PROFILE# User preferencesContextType.METADATA# Additional metadata
fromagenticaiframework.contextimportContextPriority# Priority determines what stays during compressionContextPriority.CRITICAL# Never removed (system prompts)ContextPriority.HIGH# Removed last (recent conversation)ContextPriority.MEDIUM# Normal priorityContextPriority.LOW# Removed first (old history)
importlogginglogger=logging.getLogger(__name__)# Build optimized context for promptcontext=context_mgr.build_context(query="How do I create a class in Python?",max_tokens=4000,include_types=[ContextType.SYSTEM,ContextType.CONVERSATION,ContextType.KNOWLEDGE])# Get as messages for chat APImessages=context.to_messages()# Get as single stringtext=context.to_text()# Get token countlogger.info(f"Context tokens: {context.token_count}")
# Create session for conversationsession=context_mgr.create_session(session_id="user-123-conv-456",ttl_minutes=60)# Add to sessionsession.add_message("user","Hello!")session.add_message("assistant","Hi there! How can I help?")# Retrieve session contextsession_context=session.get_context()
fromagenticaiframework.contextimport(SemanticContextIndex,IndexConfig)# Create semantic indexindex=SemanticContextIndex(config=IndexConfig(embedding_model="text-embedding-3-small",dimension=1536,similarity_metric="cosine"))# Index documentsindex.add_documents([{"id":"doc1","content":"Python is a programming language..."},{"id":"doc2","content":"Machine learning uses algorithms..."},{"id":"doc3","content":"Data structures include arrays..."}])
importlogginglogger=logging.getLogger(__name__)# Search for relevant contextresults=index.search(query="How do I train a model?",top_k=5,min_score=0.7)forresultinresults:logger.info(f"Document: {result.id}")logger.info(f"Score: {result.score:.3f}")logger.info(f"Content: {result.content[:100]}...")
fromagenticaiframework.contextimportContextRetrievalStrategy# Different retrieval strategiesstrategy=ContextRetrievalStrategy.SEMANTIC# Pure semanticstrategy=ContextRetrievalStrategy.KEYWORD# Keyword matchingstrategy=ContextRetrievalStrategy.HYBRID# Combinedstrategy=ContextRetrievalStrategy.RECENT# Most recentstrategy=ContextRetrievalStrategy.MMR# Maximum Marginal Relevance# Use with context managercontext_mgr.set_retrieval_strategy(ContextRetrievalStrategy.MMR,diversity_factor=0.3# Balance relevance vs diversity)
# Configure sliding window for conversationswindow.configure_sliding(keep_recent=10,# Keep last 10 messageskeep_system=True,# Always keep system promptsummary_threshold=20# Summarize after 20 messages)# Add messages (auto-manages window)formessageinconversation_history:window.add_message(message)# Automatically removes old messages and creates summaries
importlogginglogger=logging.getLogger(__name__)fromagenticaiframework.contextimport(ContextCompressor,CompressionStrategy,CompressionConfig)# Create compressorcompressor=ContextCompressor(config=CompressionConfig(target_ratio=0.5,# Compress to 50% of originalpreserve_key_info=True))# Compress textoriginal="Long context with lots of details..."compressed=compressor.compress(original)logger.info(f"Original: {len(original)} chars")logger.info(f"Compressed: {len(compressed)} chars")
# Compress progressively as context growscontext_mgr.enable_progressive_compression(trigger_threshold=0.8,# Compress at 80% utilizationtarget_utilization=0.6,# Compress down to 60%strategy=CompressionStrategy.SELECTIVE)
fromagenticaiframeworkimportAgentfromagenticaiframework.contextimport(ContextManager,SemanticContextIndex,ContextWindow,ContextCompressor,ContextType,ContextPriority,ContextItem,CompressionStrategy,ContextRetrievalStrategy)# Initialize context componentscontext_mgr=ContextManager(max_tokens=8000,model="gpt-4")semantic_index=SemanticContextIndex()context_window=ContextWindow.for_model("gpt-4")compressor=ContextCompressor(target_ratio=0.5)# Index knowledge baseknowledge_docs=load_knowledge_base()semantic_index.add_documents(knowledge_docs)# Create context-aware agentclassContextAwareAgent:def__init__(self,agent:Agent):self.agent=agentself.session_history=[]asyncdefprocess(self,user_input:str)->str:# 1. Add system contextcontext_mgr.add(ContextItem(content=self.agent.instructions,type=ContextType.SYSTEM,priority=ContextPriority.CRITICAL))# 2. Retrieve relevant knowledgerelevant_docs=semantic_index.search(query=user_input,top_k=3,min_score=0.7)fordocinrelevant_docs:context_mgr.add(ContextItem(content=doc.content,type=ContextType.KNOWLEDGE,priority=ContextPriority.MEDIUM))# 3. Add conversation historyformsginself.session_history[-10:]:# Last 10 messagescontext_mgr.add(ContextItem(content=f"{msg['role']}: {msg['content']}",type=ContextType.CONVERSATION,priority=ContextPriority.HIGH))# 4. Add current user inputcontext_mgr.add(ContextItem(content=f"User: {user_input}",type=ContextType.TASK,priority=ContextPriority.HIGH))# 5. Build optimized contextcontext=context_mgr.build_context(query=user_input,max_tokens=6000# Leave room for response)# 6. Check if compression neededifcontext.token_count>6000:context=compressor.compress_context(context,target_tokens=5000)# 7. Generate responseresponse=awaitself.agent.run(messages=context.to_messages())# 8. Update historyself.session_history.append({"role":"user","content":user_input})self.session_history.append({"role":"assistant","content":response})# 9. Clear context for next turncontext_mgr.clear()returnresponse# Usageagent=Agent(name="KnowledgeAgent",model="gpt-4",instructions="You are a helpful assistant with access to a knowledge base.")context_agent=ContextAwareAgent(agent)# Conversation with context managementresponse1=awaitcontext_agent.process("What is machine learning?")response2=awaitcontext_agent.process("How does it relate to AI?")response3=awaitcontext_agent.process("Can you give me an example?")
# 1. Use semantic retrieval instead of stuffingrelevant_context=index.search(query,top_k=3)# 2. Compress old conversationcompressed=compressor.compress_conversation(history,keep_recent=5)# 3. Prioritize context itemscontext_mgr.add(item,priority=ContextPriority.HIGH)# 4. Set appropriate limitswindow=ContextWindow(max_tokens=4000)