AI agent patterns with Cache-Pot: semantic cache and memory

Cache-Pot provides three ready-to-use building blocks for AI-powered applications: semantic caching to cut LLM costs by reusing answers to similar questions, vector search for similarity-based retrieval in RAG pipelines, and agent memory for stateful sessions that survive across turns. All three are exposed over the same RESP2 connection your existing Redis client already speaks — no extra services, no SDKs to install.

Pattern 1: Semantic response caching

Save LLM responses keyed by meaning, not by exact text. When a user asks something semantically equivalent to a question you have already answered — even if the wording is completely different — Cache-Pot returns the stored response instead of making another model call.

import redis
import openai

r = redis.Redis(host='localhost', port=6379)
oai = openai.OpenAI()

def ask(question: str) -> str:
    # Check semantic cache first
    cached = r.execute_command('SCACHE.GET', question)
    if cached:
        return cached.decode()

    # Cache miss — call the LLM
    response = oai.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': question}]
    )
    answer = response.choices[0].message.content

    # Store with 1-hour TTL
    r.execute_command('SCACHE.SET', question, answer, 'TTL', 3600)
    return answer

SCACHE.GET and SCACHE.SET require an embeddings provider. Set CACHEPOT_EMBED_URL and CACHEPOT_EMBED_KEY on the Cache-Pot server before using these commands. A free local Ollama instance or an OpenAI API key both work.

You can tune how closely a new question must match a stored one by passing an explicit THRESHOLD:

# Require a very close match (default is 0.9)
cached = r.execute_command('SCACHE.GET', question, 'THRESHOLD', '0.95')

# Accept looser matches to increase the hit rate
cached = r.execute_command('SCACHE.GET', question, 'THRESHOLD', '0.80')

Cache-Pot tracks hit and miss counts and surfaces a running hit ratio on the dashboard at http://localhost:8080 — every hit is a model call you did not pay for.

Pattern 2: Vector similarity search

Store document embeddings and retrieve the most relevant ones for retrieval-augmented generation (RAG). Your application produces the embeddings; Cache-Pot stores and searches them with cosine similarity.

def index_document(doc_id: str, text: str):
    # Generate an embedding with your model of choice
    resp = oai.embeddings.create(input=text, model='text-embedding-3-small')
    vec = resp.data[0].embedding

    # Store the vector with a metadata snippet
    args = ['VSET', 'documents', doc_id] + [str(f) for f in vec] + ['META', text[:100]]
    r.execute_command(*args)

def search_documents(query: str, top_k: int = 5):
    # Embed the query and find the closest documents
    resp = oai.embeddings.create(input=query, model='text-embedding-3-small')
    vec = resp.data[0].embedding

    args = ['VSEARCH', 'documents'] + [str(f) for f in vec] + ['TOPK', str(top_k), 'WITHSCORES']
    return r.execute_command(*args)

All vectors in a collection must share the same dimension, fixed by the first VSET call. VDIM documents returns the dimension of an existing collection. VCARD documents returns the number of stored vectors.

A full RAG loop — index at write time, search at query time, inject top results into the prompt — needs only these two functions plus your LLM call:

def rag_answer(user_query: str) -> str:
    # Retrieve relevant chunks
    results = search_documents(user_query, top_k=3)

    # results is a flat list: [id, meta, score, id, meta, score, ...]
    context_chunks = [results[i + 1].decode() for i in range(0, len(results), 3)]
    context = '\n'.join(context_chunks)

    response = oai.chat.completions.create(
        model='gpt-4o-mini',
        messages=[
            {'role': 'system', 'content': f'Answer using this context:\n{context}'},
            {'role': 'user', 'content': user_query},
        ]
    )
    return response.choices[0].message.content

Pattern 3: Agent session memory

Store per-user or per-session state that persists across agent turns. REMEMBER writes a named field into a session namespace; RECALL reads one field or the entire session back.

def agent_turn(session_id: str, user_message: str) -> str:
    # Recall what we know about this session
    memory = r.execute_command('RECALL', session_id)
    context = dict(zip(memory[0::2], memory[1::2])) if memory else {}

    # Build a system prompt from stored facts
    user_name = context.get(b'user_name', b'').decode()
    turn_count = int(context.get(b'turn_count', b'0'))

    system_prompt = f"You are a helpful assistant. The user's name is {user_name}." if user_name else \
                    "You are a helpful assistant."

    response = oai.chat.completions.create(
        model='gpt-4o-mini',
        messages=[
            {'role': 'system', 'content': system_prompt},
            {'role': 'user', 'content': user_message},
        ]
    )
    answer = response.choices[0].message.content

    # Update memory for the next turn
    r.execute_command('REMEMBER', session_id, 'last_message', user_message)
    r.execute_command('REMEMBER', session_id, 'turn_count', str(turn_count + 1))

    return answer

Recall a single field instead of the whole session by passing the field name:

# Recall only the user's name
name = r.execute_command('RECALL', session_id, 'user_name')

REMEMBER and RECALL are backed by a per-session hash named mem:<session>. You can inspect or edit session memory directly with HGETALL mem:<session> using any Redis client or the Cache-Pot dashboard.

Putting it all together

Combine all three patterns in a single agent pipeline: check the semantic cache → recall session memory → search the vector store for context → generate a response → update the cache and memory. The entire loop runs over one RESP2 connection with no extra services.

def full_agent_turn(session_id: str, user_message: str) -> str:
    # 1. Check semantic cache — skip the LLM entirely if we have a close match
    cached = r.execute_command('SCACHE.GET', user_message, 'THRESHOLD', '0.92')
    if cached:
        return cached.decode()

    # 2. Recall session memory for personalisation
    memory = r.execute_command('RECALL', session_id)
    context = dict(zip(memory[0::2], memory[1::2])) if memory else {}

    # 3. Retrieve relevant documents from the vector store
    results = search_documents(user_message, top_k=3)
    context_chunks = [results[i + 1].decode() for i in range(0, len(results), 3)] if results else []

    # 4. Generate a response
    answer = oai.chat.completions.create(
        model='gpt-4o-mini',
        messages=[
            {'role': 'system', 'content': 'Context:\n' + '\n'.join(context_chunks)},
            {'role': 'user', 'content': user_message},
        ]
    ).choices[0].message.content

    # 5. Update the semantic cache and session memory
    r.execute_command('SCACHE.SET', user_message, answer, 'TTL', 3600)
    r.execute_command('REMEMBER', session_id, 'last_message', user_message)

    return answer

​Pattern 1: Semantic response caching

​Pattern 2: Vector similarity search

​Pattern 3: Agent session memory

​Putting it all together

Pattern 1: Semantic response caching

Pattern 2: Vector similarity search

Pattern 3: Agent session memory

Putting it all together