Streamlining Your Build Process with AWS CodeBuild
Optimizing Software Builds with AWS CodeBuild The build process is a critical step in software development, ensuring code quality and preparing applications for deployment. AWS CodeBuild simplifies and automates this process, reducing manual intervention and enhancing efficiency. What Happens During a Build? A successful build involves multiple steps to transform raw source code into a deployable package: Retrieving Dependencies: The build pulls external libraries and modules from package managers like Node Package Manager or Maven.Compiling the Code: The source code is transformed into an executable format.Packaging the Artifact: The output is structured as a deployable unit, such as a Docker image, Linux RPM package, or Windows MSI installer.Running Automated Tests: Unit tests verify that the code performs as expected before deployment. If any of these steps fail—due to missing dependencies, compilation errors, or test failures—it results in a broken build, impacting development timelines. Continuous Integration (CI) and Frequent Builds Continuous integration (CI) ensures that every code change is tested and merged into the project seamlessly. Running frequent builds helps detect issues early, reduce integration conflicts, and provide developers with confidence in code stability. A broken build is treated as a top priority, as it affects the entire development team. With a structured build process in place, developers can focus on new features rather than debugging code conflicts. Automating the Build with AWS CodeBuild AWS CodeBuild is a fully managed service that automates compilation, testing, and artifact creation. Instead of managing on-premises build servers, teams can leverage CodeBuild's scalable infrastructure. Key Benefits of AWS CodeBuildScalability: Automatically handles multiple builds in parallel, reducing wait times.Cost Efficiency: Pay only for the build time used, eliminating costs associated with idle build servers.Seamless AWS Integration: Works with AWS services like CloudWatch for monitoring and S3 for storing artifacts.Configuring a Build in AWS CodeBuild To use AWS CodeBuild, two key components need to be configured: Build Project: Defines the source code location, build environment (such as Docker images), and storage settings.Buildspec File: The buildspec.yml file specifies build steps, environment variables, and artifact packaging. Logs and build outputs are stored in AWS CloudWatch, providing a detailed view of build performance and potential issues. Once the build is complete, the artifact is stored in S3, ready for deployment. Why Choose AWS CodeBuild? AWS CodeBuild removes the complexity of managing build infrastructure, allowing development teams to focus on software quality and delivery. By automating the build process, businesses can accelerate deployment cycles and improve CI/CD workflows. Could your organization benefit from scalable and automated builds? Contact Webaroo today to implement AWS CodeBuild and optimize your development pipeline.
AI Agent Memory Systems: From Session to Persistent Context
AI
Agent Memory Systems: From Session to Persistent Context
Your AI agent remembers the last three messages. Great. But what
happens when the user comes back tomorrow? Next week? Next month?
Memory isn’t just about token windows—it’s about building systems
that retain context across sessions, learn from interactions, and recall
relevant information at the right time. This is the difference between a
chatbot and an actual assistant.
This guide covers the engineering behind AI agent memory: when to use
different storage strategies, how to implement them, and the production
patterns that scale.
The Memory Hierarchy
AI agents need multiple layers of memory, just like humans:
1. Working Memory (Current
Session)
What it is: The conversation happening right
now
Storage: In-context tokens, cached in LLM
provider
Lifetime: Current session only
Retrieval: Automatic (part of prompt)
Cost: Token usage per request
2. Short-Term Memory (Recent
Sessions)
What it is: Recent interactions from the past few
days
Storage: Fast key-value store (Redis,
DynamoDB)
Lifetime: Days to weeks
Retrieval: Query by user/session ID
Cost: Database queries
3. Long-Term Memory
(Historical Context)
What it is: All past interactions, decisions,
preferences
Storage: Vector database (Pinecone, Weaviate,
pgvector)
Lifetime: Permanent (or years)
Retrieval: Semantic search
Cost: Vector operations + storage
4. Knowledge Memory (Facts
& Training)
What it is: Domain knowledge, procedures,
policies
Storage: Vector database + structured DB
Lifetime: Updated periodically
Retrieval: RAG (Retrieval Augmented
Generation)
Cost: Embedding generation + queries
When Each Memory Type Makes
Sense
Working Memory Only: - Simple FAQ bots - Stateless
API wrappers - One-shot tasks - Budget-conscious projects
Working + Short-Term: - Customer support bots
(remember current issue across multiple sessions) - Project assistants
(track active tasks) - Debugging helpers (retain context during
troubleshooting)
Working + Short-Term + Long-Term: - Personal
assistants (learn user preferences over time) - Enterprise agents
(organizational memory) - Learning systems (improve from historical
interactions)
Full Stack (All Four): - Production AI assistants -
Multi-tenant SaaS platforms - High-value use cases where context =
competitive advantage
Implementation Patterns
Pattern 1: Session-Based
Memory
The simplest approach: store conversation history in a fast database,
retrieve it at the start of each session.
Architecture:
class SessionMemoryAgent:
def __init__(self, redis_client):
self.redis = redis_client
self.session_ttl = 3600 * 24 * 7 # 7 days
async def get_context(self, user_id: str, session_id: str) -> List[Message]:
"""Retrieve recent conversation history"""
key = f"session:{user_id}:{session_id}"
messages = await self.redis.lrange(key, 0, -1)
return [json.loads(m) for m in messages]
async def add_message(self, user_id: str, session_id: str, message: Message):
"""Append message to session history"""
key = f"session:{user_id}:{session_id}"
await self.redis.rpush(key, json.dumps(message.dict()))
await self.redis.expire(key, self.session_ttl)
async def chat(self, user_id: str, session_id: str, user_message: str) -> str:
# Load conversation history
history = await self.get_context(user_id, session_id)
# Build prompt with history
messages = [
{"role": "system", "content": "You are a helpful assistant."}
]
messages.extend([{"role": m.role, "content": m.content} for m in history])
messages.append({"role": "user", "content": user_message})
# Get response
response = await llm.chat(messages)
# Store both messages
await self.add_message(user_id, session_id,
Message(role="user", content=user_message, timestamp=time.time()))
await self.add_message(user_id, session_id,
Message(role="assistant", content=response, timestamp=time.time()))
return response
Advantages: - Simple to implement - Fast retrieval -
Predictable costs
Limitations: - No memory across sessions - No
semantic search - Limited to recent context
Pattern 2: Vector-Based
Episodic Memory
Store all interactions as embeddings. Retrieve relevant past
conversations based on semantic similarity.
Architecture:
class VectorMemoryAgent:
def __init__(self, vector_db, embedding_model):
self.db = vector_db
self.embedder = embedding_model
async def store_interaction(self, user_id: str, interaction: Interaction):
"""Store interaction with embedding"""
# Generate embedding of the interaction
text = f"{interaction.user_message}\n{interaction.assistant_response}"
embedding = await self.embedder.embed(text)
# Store in vector DB
await self.db.upsert(
id=interaction.id,
vector=embedding,
metadata={
"user_id": user_id,
"timestamp": interaction.timestamp,
"user_message": interaction.user_message,
"assistant_response": interaction.assistant_response,
"tags": interaction.tags,
"sentiment": interaction.sentiment
}
)
async def retrieve_relevant_context(
self,
user_id: str,
current_query: str,
limit: int = 5
) -> List[Interaction]:
"""Find semantically similar past interactions"""
# Embed current query
query_embedding = await self.embedder.embed(current_query)
# Search vector DB
results = await self.db.query(
vector=query_embedding,
filter={"user_id": user_id},
top_k=limit,
include_metadata=True
)
return [Interaction(**r.metadata) for r in results]
async def chat(self, user_id: str, message: str) -> str:
# Retrieve relevant past interactions
relevant_context = await self.retrieve_relevant_context(user_id, message)
# Build prompt with retrieved context
context_summary = "\n\n".join([
f"Past conversation (relevance: {ctx.score:.2f}):\nUser: {ctx.user_message}\nAssistant: {ctx.assistant_response}"
for ctx in relevant_context
])
prompt = f"""You are assisting a user. Here are some relevant past interactions:
{context_summary}
Current user message: {message}
Respond to the current message, using past context where relevant."""
response = await llm.generate(prompt)
# Store this interaction
interaction = Interaction(
id=str(uuid.uuid4()),
user_id=user_id,
user_message=message,
assistant_response=response,
timestamp=time.time()
)
await self.store_interaction(user_id, interaction)
return response
Advantages: - Semantic retrieval (finds relevant
context even if words differ) - Works across sessions - Scales to large
histories
Limitations: - Embedding costs - Query latency -
Requires tuning (top_k, relevance threshold)
Pattern 3: Hybrid Memory
System
Combine session storage with vector-based long-term memory. Best of
both worlds.
Architecture:
class HybridMemoryAgent:
def __init__(self, redis_client, vector_db, embedding_model):
self.redis = redis_client
self.vector_db = vector_db
self.embedder = embedding_model
self.session_ttl = 3600 * 24 # 1 day
self.session_limit = 20 # Max messages in working memory
async def get_working_memory(self, user_id: str, session_id: str) -> List[Message]:
"""Get recent conversation (working memory)"""
key = f"session:{user_id}:{session_id}"
messages = await self.redis.lrange(key, -self.session_limit, -1)
return [json.loads(m) for m in messages]
async def get_long_term_memory(self, user_id: str, query: str) -> List[Interaction]:
"""Get relevant historical context (long-term memory)"""
query_embedding = await self.embedder.embed(query)
results = await self.vector_db.query(
vector=query_embedding,
filter={"user_id": user_id},
top_k=3,
include_metadata=True
)
return [Interaction(**r.metadata) for r in results if r.score > 0.7]
async def chat(self, user_id: str, session_id: str, message: str) -> str:
# 1. Load working memory (recent conversation)
working_memory = await self.get_working_memory(user_id, session_id)
# 2. Load long-term memory (relevant past context)
long_term_memory = await self.get_long_term_memory(user_id, message)
# 3. Build layered prompt
prompt_parts = ["You are a helpful assistant."]
if long_term_memory:
context = "\n".join([
f"- {ctx.user_message[:100]}... (response: {ctx.assistant_response[:100]}...)"
for ctx in long_term_memory
])
prompt_parts.append(f"\nRelevant past interactions:\n{context}")
# 4. Construct messages
messages = [{"role": "system", "content": "\n\n".join(prompt_parts)}]
messages.extend([{"role": m.role, "content": m.content} for m in working_memory])
messages.append({"role": "user", "content": message})
# 5. Generate response
response = await llm.chat(messages)
# 6. Store in both memory systems
await self.store_working_memory(user_id, session_id, message, response)
await self.store_long_term_memory(user_id, message, response)
return response
async def store_working_memory(self, user_id: str, session_id: str,
user_msg: str, assistant_msg: str):
"""Store in Redis (short-term)"""
key = f"session:{user_id}:{session_id}"
await self.redis.rpush(key, json.dumps({
"role": "user",
"content": user_msg,
"timestamp": time.time()
}))
await self.redis.rpush(key, json.dumps({
"role": "assistant",
"content": assistant_msg,
"timestamp": time.time()
}))
await self.redis.expire(key, self.session_ttl)
async def store_long_term_memory(self, user_id: str,
user_msg: str, assistant_msg: str):
"""Store in vector DB (long-term)"""
interaction_text = f"User: {user_msg}\nAssistant: {assistant_msg}"
embedding = await self.embedder.embed(interaction_text)
await self.vector_db.upsert(
id=str(uuid.uuid4()),
vector=embedding,
metadata={
"user_id": user_id,
"user_message": user_msg,
"assistant_response": assistant_msg,
"timestamp": time.time()
}
)
Advantages: - Fast recent context (Redis) - Deep
historical context (vector DB) - Balances cost and capability
Challenges: - More complex to implement - Two
systems to maintain - Deciding what goes where
Production Considerations
Memory Compression
Long conversations exceed token limits. Compress older messages.
class CompressingMemoryAgent:
async def compress_history(self, messages: List[Message]) -> List[Message]:
"""Compress old messages to fit token budget"""
if len(messages) <= 10:
return messages
# Keep recent messages verbatim
recent = messages[-5:]
# Summarize older messages
older = messages[:-5]
summary_text = "\n".join([f"{m.role}: {m.content}" for m in older])
summary = await llm.generate(f"""Summarize this conversation history in 2-3 sentences:
{summary_text}
Summary:""")
compressed = [
Message(role="system", content=f"Previous conversation summary: {summary}")
]
compressed.extend(recent)
return compressed
Privacy & Data Retention
Memory means storing user data. Handle it responsibly.
class PrivacyAwareMemoryAgent:
def __init__(self, vector_db):
self.db = vector_db
self.retention_days = 90
async def anonymize_interaction(self, interaction: Interaction) -> Interaction:
"""Remove PII before storing"""
# Use a PII detection service/library
anonymized_user_msg = await pii_detector.redact(interaction.user_message)
anonymized_assistant_msg = await pii_detector.redact(interaction.assistant_response)
return Interaction(
id=interaction.id,
user_id=hash_user_id(interaction.user_id), # Hash instead of plaintext
user_message=anonymized_user_msg,
assistant_response=anonymized_assistant_msg,
timestamp=interaction.timestamp
)
async def delete_old_memories(self, user_id: str):
"""Implement data retention policy"""
cutoff_time = time.time() - (self.retention_days * 24 * 3600)
await self.db.delete(
filter={
"user_id": user_id,
"timestamp": {"$lt": cutoff_time}
}
)
async def delete_user_data(self, user_id: str):
"""GDPR/CCPA compliance: delete all user data"""
await self.db.delete(filter={"user_id": user_id})
await self.redis.delete(f"session:{user_id}:*")
Memory Indexing Strategies
How you index matters.
class IndexedMemoryAgent:
async def store_with_rich_metadata(self, interaction: Interaction):
"""Index by multiple dimensions for better retrieval"""
embedding = await self.embedder.embed(interaction.user_message)
# Extract metadata for filtering
tags = await self.extract_tags(interaction.user_message)
sentiment = await self.analyze_sentiment(interaction.user_message)
entities = await self.extract_entities(interaction.user_message)
await self.db.upsert(
id=interaction.id,
vector=embedding,
metadata={
"user_id": interaction.user_id,
"timestamp": interaction.timestamp,
"tags": tags, # ["billing", "technical-issue"]
"sentiment": sentiment, # "negative", "neutral", "positive"
"entities": entities, # {"product": "Pro Plan", "company": "Acme"}
"resolved": interaction.resolved, # bool
"category": interaction.category
}
)
async def retrieve_with_filters(self, user_id: str, query: str,
category: str = None,
resolved: bool = None):
"""Retrieve with semantic search + metadata filters"""
query_embedding = await self.embedder.embed(query)
filters = {"user_id": user_id}
if category:
filters["category"] = category
if resolved is not None:
filters["resolved"] = resolved
results = await self.db.query(
vector=query_embedding,
filter=filters,
top_k=5
)
return results
Memory Consistency Across
Agents
In multi-agent systems, agents need to share memory.
class SharedMemoryCoordinator:
"""Coordinate memory across multiple specialized agents"""
def __init__(self, vector_db, redis_client):
self.vector_db = vector_db
self.redis = redis_client
async def write_to_shared_memory(self, interaction: Interaction,
agent_id: str):
"""Any agent can write to shared memory"""
embedding = await self.embedder.embed(
f"{interaction.user_message} {interaction.assistant_response}"
)
await self.vector_db.upsert(
id=interaction.id,
vector=embedding,
metadata={
**interaction.dict(),
"agent_id": agent_id, # Track which agent handled it
"shared": True
}
)
async def retrieve_shared_context(self, query: str,
exclude_agent: str = None):
"""Retrieve context from all agents, optionally excluding one"""
query_embedding = await self.embedder.embed(query)
filters = {"shared": True}
if exclude_agent:
filters["agent_id"] = {"$ne": exclude_agent}
results = await self.vector_db.query(
vector=query_embedding,
filter=filters,
top_k=5
)
return results
Monitoring Memory Health
Track memory system performance.
class MemoryMetrics:
def __init__(self):
self.context_relevance = Histogram(
'memory_context_relevance_score',
'Relevance score of retrieved context'
)
self.retrieval_latency = Histogram(
'memory_retrieval_latency_seconds',
'Time to retrieve context'
)
self.storage_size = Gauge(
'memory_storage_size_bytes',
'Total size of stored memories',
['user_id']
)
async def record_retrieval(self, user_id: str, query: str):
start_time = time.time()
results = await self.vector_db.query(
vector=await self.embedder.embed(query),
filter={"user_id": user_id},
top_k=5
)
latency = time.time() - start_time
self.retrieval_latency.observe(latency)
if results:
avg_relevance = sum(r.score for r in results) / len(results)
self.context_relevance.observe(avg_relevance)
return results
The Bottom Line
Memory isn’t a feature—it’s a system. The difference between a demo
and a production AI agent is how well it remembers, retrieves, and
applies context.
Start simple: Session-based memory for most use
cases.
Add layers: Vector storage when you need semantic
retrieval across time.
Go hybrid: Combine fast short-term storage with deep
long-term memory for production systems.
And always remember: stored data = stored responsibility. Handle it
accordingly.
The best AI agents don’t just remember everything—they remember the
right things at the right time.
