Module 6: Prompt Composition and Chaining
6. Multi-Turn Strategies and Memory
Overview
Multi-turn conversations pose a unique challenge: the history grows with every exchange, and eventually exceeds the context window limit. On top of that, not every previous message is equally relevant to the current question.
In this capsule you'll learn three strategies for managing conversation history: selective memory (include only the N most recent messages), history summarization (compress old messages), and relevance-based selection (pick messages by semantic similarity).
The Growing History Problem
Turn 1: User: "Hi, I'm Ana" → 5 tokens
Bot: "Hi Ana, how can I help you?" → 10 tokens
Turn 2: User: "I have a problem with my API" → 10 tokens
Bot: "Tell me more about the problem..." → 15 tokens
...
Turn 20: Accumulated history → ~8,000 tokens
Turn 50: Accumulated history → ~25,000 tokens
Turn 100: Accumulated history → ~60,000 tokens ← Dangerous for gpt-4o-mini
ON TOP OF THAT:
- Messages from 50 turns ago are probably no longer relevant
- The cost per call grows linearly with the history
- The more unnecessary context, the more "noise" there can be in the answers
Architecture of a Conversational System
MULTI-TURN SYSTEM
┌─────────────────────────────────────────────────┐
│ MEMORY │
│ │
│ ┌────────────────┐ ┌─────────────────────┐ │
│ │ Full history │ │ User profile │ │
│ │ (all msgs) │ │ (persistent data) │ │
│ └───────┬────────┘ └──────────┬──────────┘ │
│ │ │ │
│ ▼ │ │
│ ┌────────────────┐ │ │
│ │ Context │ │ │
│ │ Selection │ │ │
│ │ Strategy │ │ │
│ └───────┬────────┘ │ │
└──────────┼────────────────────────┼─────────────┘
│ │
▼ ▼
┌─────────────────────────────────────┐
│ BUILT CONTEXT │
│ [System prompt with the profile] │
│ [Summary of the prior conversation]│
│ [Last N relevant messages] │
│ [The user's current message] │
└────────────────────┬────────────────┘
│
▼
LLM Response
Base Implementation
from openai import OpenAI
from dataclasses import dataclass, field
from typing import Optional
import json
import time
client = OpenAI()
@dataclass
class Message:
"""Represents one message in the conversation."""
role: str
content: str
timestamp: float = field(default_factory=time.time)
tokens: int = 0
def to_dict(self) -> dict:
return {"role": self.role, "content": self.content}
@dataclass
class UserProfile:
"""Persistent user profile for personalization."""
name: str = ""
preferences: dict = field(default_factory=dict)
business_context: str = ""
language: str = "english"
topic_history: list[str] = field(default_factory=list)
class ConversationalSystem:
"""
Conversational system with memory management.
Combines several strategies based on the size of the history.
"""
def __init__(
self,
system_prompt: str,
max_context_tokens: int = 4000,
n_recent_messages: int = 6,
summarize_after_n: int = 10
):
self.system_prompt = system_prompt
self.max_context_tokens = max_context_tokens
self.n_recent_messages = n_recent_messages
self.summarize_after_n = summarize_after_n
self.history: list[Message] = []
self.accumulated_summary: str = ""
self.profile: UserProfile = UserProfile()
def add_message(self, role: str, content: str):
"""Adds a message to the history."""
msg = Message(role=role, content=content)
# Estimate tokens
msg.tokens = len(content.split()) * 1.3 # Approximation
self.history.append(msg)
def build_context(self, current_query: str) -> list[dict]:
"""
Builds the optimal context for the LLM call.
Combines summary + recent messages + profile.
"""
messages = []
# System prompt with the profile context
full_system = self.system_prompt
if self.profile.name:
full_system += f"\n\nUser: {self.profile.name}"
if self.profile.business_context:
full_system += f"\nContext: {self.profile.business_context}"
messages.append({"role": "system", "content": full_system})
# Add the summary if there is one
if self.accumulated_summary:
messages.append({
"role": "system",
"content": f"Summary of the prior conversation:\n{self.accumulated_summary}"
})
# Add the last N messages
recent_messages = self.history[-self.n_recent_messages:]
messages.extend([m.to_dict() for m in recent_messages])
return messages
def chat(self, user_message: str) -> str:
"""Processes one conversation turn."""
# Check whether the old history needs summarizing
if len(self.history) >= self.summarize_after_n:
self._summarize_old_history()
# Add the user's message
self.add_message("user", user_message)
# Build the context
context_messages = self.build_context(user_message)
# Call the LLM
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=context_messages,
temperature=0.7,
max_tokens=500
)
answer = response.choices[0].message.content
# Add the answer to the history
self.add_message("assistant", answer)
# Update the profile if this is the first message
if len(self.history) <= 2:
self._update_profile(user_message)
return answer
def _summarize_old_history(self):
"""Compresses the old messages into a summary."""
# Take the messages that are NOT the N most recent ones
messages_to_summarize = self.history[:-self.n_recent_messages]
if not messages_to_summarize:
return
# Convert to text
conversation_str = "\n".join([
f"{m.role.upper()}: {m.content}"
for m in messages_to_summarize
])
# Include the previous summary if there is one
if self.accumulated_summary:
conversation_str = f"Previous summary:\n{self.accumulated_summary}\n\nAdditional conversation:\n{conversation_str}"
# Generate the new summary
self.accumulated_summary = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Summarize this conversation in 5 sentences at most.
Preserve: the user's personal information (name, company, etc.),
decisions made, problems solved, and important context.
Conversation:
{conversation_str}"""
}],
temperature=0,
max_tokens=200
).choices[0].message.content
# Drop the summarized messages from the history (keep only the recent ones)
self.history = self.history[-self.n_recent_messages:]
def _update_profile(self, message: str):
"""Extracts the user's information from the first message."""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Extract the user's information from this message if it's available.
If there is no information, leave the fields empty.
Message: {message}
JSON: {{"name": str, "company": str, "context": str}}"""
}],
temperature=0,
response_format={"type": "json_object"}
).choices[0].message.content
try:
data = json.loads(response)
if data.get("name"):
self.profile.name = data["name"]
if data.get("company") or data.get("context"):
self.profile.business_context = data.get("context", data.get("company", ""))
except Exception:
pass
def get_stats(self) -> dict:
"""Statistics about the current state of the conversation."""
return {
"messages_in_history": len(self.history),
"has_summary": bool(self.accumulated_summary),
"summary_length": len(self.accumulated_summary),
"profile_name": self.profile.name,
"estimated_tokens": sum(m.tokens for m in self.history)
}
# Usage example:
if __name__ == "__main__":
bot = ConversationalSystem(
system_prompt="You are a Python programming assistant. You help with code, errors and best practices.",
n_recent_messages=6,
summarize_after_n=8
)
conversation = [
"Hi, I'm Pedro, I work at a fintech startup",
"I have a problem with my FastAPI, I can't get the async routes to work",
"How do I configure uvicorn correctly?",
"The error says: RuntimeError: no running event loop",
"Can you show me a complete example?",
]
for message in conversation:
print(f"\nUser: {message}")
answer = bot.chat(message)
print(f"Bot: {answer[:200]}...")
stats = bot.get_stats()
print(f"[History: {stats['messages_in_history']} msgs, Summary: {bool(stats['has_summary'])}]")Strategy: Relevance-Based Selection with Embeddings
import math
from typing import Optional
def cosine_similarity(vec1: list[float], vec2: list[float]) -> float:
"""Computes the cosine similarity between two vectors."""
dot_product = sum(a * b for a, b in zip(vec1, vec2))
mag1 = math.sqrt(sum(a ** 2 for a in vec1))
mag2 = math.sqrt(sum(b ** 2 for b in vec2))
if mag1 * mag2 == 0:
return 0.0
return dot_product / (mag1 * mag2)
def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
"""Gets the embedding of a text."""
response = client.embeddings.create(
model=model,
input=text
)
return response.data[0].embedding
class SemanticMemory:
"""
Memory system that selects messages by semantic relevance.
Useful when the history is long and heterogeneous.
"""
def __init__(self, k_messages: int = 4):
self.history: list[tuple[dict, list[float]]] = [] # (message, embedding)
self.k_messages = k_messages
def add_message(self, role: str, content: str):
"""Adds a message with its pre-computed embedding."""
message = {"role": role, "content": content}
embedding = get_embedding(content)
self.history.append((message, embedding))
def select_relevant(
self,
query: str,
n_recent: int = 2,
n_semantic: int = 2
) -> list[dict]:
"""
Selects messages by combining:
- The N most recent ones (always included for coherence)
- The K most semantically similar to the current query
Args:
query: The user's current question or message
n_recent: How many recent messages to always include
n_semantic: How many additional messages by similarity
Returns:
List of selected messages, in chronological order
"""
if not self.history:
return []
# Recent messages (always included)
recent_messages = [msg for msg, _ in self.history[-n_recent:]]
recent_indices = set(range(len(self.history) - n_recent, len(self.history)))
# Get the embedding of the current query
query_emb = get_embedding(query)
# Compute the similarity for the NON-recent messages
semantic_candidates = []
for i, (msg, emb) in enumerate(self.history[:-n_recent]):
if i not in recent_indices:
sim = cosine_similarity(query_emb, emb)
semantic_candidates.append((sim, i, msg))
# Sort by similarity and take the top K
semantic_candidates.sort(reverse=True)
semantic_messages = [
(i, msg)
for sim, i, msg in semantic_candidates[:n_semantic]
if sim > 0.7 # Relevance threshold
]
# Combine and sort chronologically
all_indices = set(recent_indices)
all_messages_with_index = []
for i, msg in semantic_messages:
if i not in all_indices:
all_messages_with_index.append((i, msg))
all_indices.add(i)
for i in range(len(self.history) - n_recent, len(self.history)):
if i < len(self.history):
msg, _ = self.history[i]
all_messages_with_index.append((i, msg))
# Sort by index (chronological order)
all_messages_with_index.sort(key=lambda x: x[0])
return [msg for _, msg in all_messages_with_index]
def chat_with_semantic_memory(
self,
message: str,
system_prompt: str = "You are a helpful assistant."
) -> str:
"""Answers using semantic context selection."""
# Select the relevant context
relevant_messages = self.select_relevant(message)
# Build the messages for the API
messages = [{"role": "system", "content": system_prompt}]
messages.extend(relevant_messages)
messages.append({"role": "user", "content": message})
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
temperature=0.7,
max_tokens=400
)
answer = response.choices[0].message.content
# Add to the history
self.add_message("user", message)
self.add_message("assistant", answer)
return answer
# Example:
memory = SemanticMemory(k_messages=4)
system = "You are a Python technical assistant."
answers = []
messages = [
"My name is Carlos and I work with FastAPI",
"How do I handle errors in FastAPI?",
"What is Pydantic?",
"What's the difference between async and sync in Python?",
"Back to the topic of errors in FastAPI, how do I make a global error handler?"
]
for msg in messages:
resp = memory.chat_with_semantic_memory(msg, system)
print(f"\nUser: {msg}")
print(f"Bot: {resp[:150]}...")
print(f"Messages in history: {len(memory.history)}")Strategy: Context Pruning
def build_smart_context(
history: list[dict],
current_query: str,
max_tokens: int = 4000,
always_include_n: int = 4
) -> list[dict]:
"""
Builds the optimal context for the LLM using several strategies:
1. Always include the N most recent messages
2. If there's room, include relevant messages based on keywords
3. Add a summary of the excluded messages if there's room
Args:
history: Complete list of messages {role, content}
current_query: The user's current message
max_tokens: Token budget for the context
always_include_n: N recent messages always included
Returns:
List of messages optimized for the LLM call
"""
used_tokens = 0
selected_context = []
# Step 1: Always include the N most recent messages
recent = history[-always_include_n:]
for msg in recent:
msg_tokens = len(msg["content"].split()) * 1.3
used_tokens += msg_tokens
selected_context = list(recent)
if used_tokens >= max_tokens:
return selected_context
# Step 2: Try to include additional relevant messages
# Compute the keywords of the current query
keywords = set(current_query.lower().split()) - {"of", "the", "a", "an", "in", "that", "for", "to", "and"}
previous_messages = history[:-always_include_n]
candidates_with_relevance = []
for i, msg in enumerate(previous_messages):
content_lower = msg["content"].lower()
relevance = sum(1 for kw in keywords if kw in content_lower)
msg_tokens = len(msg["content"].split()) * 1.3
candidates_with_relevance.append((relevance, i, msg, msg_tokens))
# Sort by relevance
candidates_with_relevance.sort(reverse=True)
# Add relevant messages until the budget runs out
additional_messages = []
for relevance, i, msg, msg_tokens in candidates_with_relevance:
if relevance == 0:
break # No more relevant ones
if used_tokens + msg_tokens <= max_tokens:
additional_messages.append((i, msg))
used_tokens += msg_tokens
# Sort the additional messages chronologically and combine
additional_messages.sort(key=lambda x: x[0])
if additional_messages:
# Add an indicator that some earlier context was omitted
if len(history) > len(recent) + len(additional_messages):
selected_context = [
{"role": "system", "content": "[Note: There are earlier messages omitted due to context space]"}
]
else:
selected_context = []
selected_context.extend([msg for _, msg in additional_messages])
selected_context.extend(recent)
return selected_contextMulti-Turn Conversation with a Persistent Profile
class ProfiledAssistant:
"""
Conversational assistant that learns about the user over the course of the conversation.
The profile gets built automatically and personalizes the answers.
"""
def __init__(self, assistant_name: str, domain: str):
self.assistant_name = assistant_name
self.domain = domain
self.history: list[dict] = []
self.profile: dict = {
"name": "",
"technical_level": "unknown", # beginner/intermediate/advanced
"preferences": [],
"frequent_topics": [],
"solved_problems": []
}
self._summary = ""
def _system_prompt(self) -> str:
"""Builds the personalized system prompt."""
prompt = f"""You are {self.assistant_name}, an assistant who is an expert in {self.domain}.
USER PROFILE:
- Name: {self.profile.get('name', 'unknown')}
- Technical level: {self.profile.get('technical_level', 'unknown')}
- Preferences: {', '.join(self.profile.get('preferences', [])[:3])}
Adapt your language to the user's technical level.
If the level is beginner: use simple analogies, avoid jargon.
If it's intermediate: use technical terms with short explanations.
If it's advanced: get straight to the point, assume the basics are known."""
if self._summary:
prompt += f"\n\nCONTEXT FROM THE PRIOR CONVERSATION:\n{self._summary}"
return prompt
def _update_profile(self, user_msg: str, assistant_resp: str):
"""Updates the profile based on the exchange."""
# Detect the technical level from the vocabulary used
advanced_words = ["async", "asyncio", "dependency injection", "SOLID", "microservices", "kubernetes", "latency", "throughput"]
basic_words = ["what is", "i don't understand", "what's it for", "explain to me", "how does it work"]
msg_lower = user_msg.lower()
if any(w in msg_lower for w in advanced_words):
self.profile["technical_level"] = "advanced"
elif any(w in msg_lower for w in basic_words) and self.profile["technical_level"] == "unknown":
self.profile["technical_level"] = "beginner"
elif self.profile["technical_level"] == "unknown":
self.profile["technical_level"] = "intermediate"
# Extract the name if we don't have it yet
if not self.profile["name"]:
if "i'm " in msg_lower:
parts = user_msg.split()
for i, word in enumerate(parts):
if word.lower() == "i'm" and i + 1 < len(parts):
name_candidate = parts[i+1].replace(",", "").replace(".", "")
if name_candidate[0].isupper():
self.profile["name"] = name_candidate
break
def chat(self, message: str) -> str:
"""Processes one conversation turn."""
# Add the user's message
self.history.append({"role": "user", "content": message})
# Build the messages for the API
messages = [{"role": "system", "content": self._system_prompt()}]
# Include the last 8 messages (4 turns)
messages.extend(self.history[-8:])
# Call the LLM
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
temperature=0.7,
max_tokens=500
)
answer = response.choices[0].message.content
# Add the answer to the history
self.history.append({"role": "assistant", "content": answer})
# Update the profile
self._update_profile(message, answer)
# Compress the history if it's very long
if len(self.history) > 16:
messages_to_summarize = self.history[:-8]
conv_str = "\n".join([f"{m['role'].upper()}: {m['content'][:100]}" for m in messages_to_summarize])
self._summary = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"Summarize the key points in 3-4 sentences:\n{conv_str}"}],
temperature=0,
max_tokens=150
).choices[0].message.content
self.history = self.history[-8:]
return answer
# Demo:
assistant = ProfiledAssistant("CodeBot", "Python and FastAPI programming")
print("=== DEMO CONVERSATION ===\n")
turns = [
"Hi, I'm María and I'm learning Python",
"What is a decorator function?",
"Can you give me a simple example?",
"Now I want to learn about async. Is it hard?",
"And how do I use async in FastAPI?"
]
for turn in turns:
print(f"María: {turn}")
answer = assistant.chat(turn)
print(f"Bot: {answer[:200]}...")
print(f"[Profile: level={assistant.profile['technical_level']}, name={assistant.profile['name']}]\n")Troubleshooting
Problem 1: Silent context overflow
Symptom: The answers start becoming incoherent, or the model "forgets" information from the start of the conversation.
Cause: The history exceeds the context window without any warning.
Solution:
def check_tokens_before_calling(messages: list[dict], limit: int = 120000) -> bool:
"""Checks that the messages don't exceed the limit before calling."""
total = sum(len(m["content"].split()) * 1.3 for m in messages)
if total > limit:
print(f"⚠ WARNING: {total:.0f} estimated tokens exceeds the limit of {limit}")
return False
return TrueProblem 2: Loss of coherence from aggressive summarization
Symptom: The summary drops important details (the user's name, specific decisions).
Solution:
def summarize_preserving_critical(history: list[dict]) -> str:
"""A summary that explicitly preserves the critical data."""
conv_str = "\n".join([f"{m['role'].upper()}: {m['content']}" for m in history])
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"""Summarize this conversation (5 sentences max).
CRITICAL: You MUST preserve exactly:
- The user's name (if it was mentioned)
- Any numeric data (IDs, versions, quantities)
- Decisions made or commitments given
- Specific errors mentioned
Conversation:
{conv_str}"""}],
temperature=0,
max_tokens=200
).choices[0].message.contentProblem 3: Expensive embeddings for a long history
Symptom: Computing embeddings for 100+ messages is expensive and slow.
Solution: Pre-compute and cache the embeddings:
import hashlib
from functools import lru_cache
@lru_cache(maxsize=500)
def get_cached_embedding(text: str) -> tuple:
"""Cached embedding using an LRU cache (bounded in memory)."""
emb = get_embedding(text)
return tuple(emb) # a tuple is hashable for lru_cacheExercises
Exercise 1: Implement an adaptive memory window
The size of the memory window should adapt to the complexity of the topic:
- Simple questions: last 4 messages
- Complex questions that refer to prior context: last 8 messages
- Questions that explicitly reference something mentioned earlier: search the whole history
See solution
def build_adaptive_context(
history: list[dict],
query: str
) -> list[dict]:
"""Adaptive context window based on the query."""
# Detect references to the past
past_references = ["you said before", "you mentioned that", "we talked about", "as i said", "do you remember that"]
has_reference = any(r in query.lower() for r in past_references)
# Detect complexity
complex_words = ["compare", "analyze", "difference", "explain", "relate"]
is_complex = any(p in query.lower() for p in complex_words)
if has_reference:
# Search the whole history with keywords
keywords = set(query.lower().split()) - {"of", "the", "a", "an"}
candidates = []
for i, msg in enumerate(history[:-4]):
content = msg["content"].lower()
score = sum(1 for kw in keywords if kw in content)
if score > 0:
candidates.append((score, i, msg))
candidates.sort(reverse=True)
extras = [msg for _, _, msg in candidates[:3]]
return extras + history[-4:]
elif is_complex:
return history[-8:]
else:
return history[-4:]Exercise 2: Session persistence
Implement a system that saves the conversation state (history + summary + profile) to a JSON file, and can load it back to continue the conversation in another session.
See solution
import json
from pathlib import Path
class PersistentAssistant(ProfiledAssistant):
def save_session(self, file: str):
"""Saves the current state to JSON."""
state = {
"history": self.history,
"profile": self.profile,
"summary": self._summary,
"timestamp": time.time()
}
with open(file, "w") as f:
json.dump(state, f, ensure_ascii=False, indent=2)
print(f"Session saved to {file}")
def load_session(self, file: str):
"""Loads a previous session."""
path = Path(file)
if not path.exists():
print(f"No session found at {file}")
return
with open(file) as f:
state = json.load(f)
self.history = state["history"]
self.profile = state["profile"]
self._summary = state["summary"]
age = (time.time() - state["timestamp"]) / 3600
print(f"Session loaded ({age:.1f}h old, {len(self.history)} messages)")
# Usage:
import time
assistant = PersistentAssistant("Bot", "Python")
assistant.chat("Hi, I'm Roberto, I work with Django")
assistant.save_session("session_roberto.json")
# In another session:
assistant2 = PersistentAssistant("Bot", "Python")
assistant2.load_session("session_roberto.json")
resp = assistant2.chat("Do you remember my name?")
print(resp) # It should remember that he's RobertoSummary
- Selective memory: The last N messages are always included. Simple and effective for short conversations.
- History summarization: Compress the old messages as the history grows. Explicitly preserve the critical data.
- Relevance-based selection: Use embeddings to select messages by semantic similarity to the current query.
- Context pruning: Combine recent + relevant while respecting the token budget.
- User profile: Build and maintain a profile that personalizes the answers without taking up much context.