⚡ Redis Support Agent Memory#

DeskBuddy (Redis Session-Memory Agent) combines multi-turn conversational memory with autonomous tool calling (calculator and get_datetime), backed by Redis as an external session store.

Unlike client-held memory (where the browser must re-upload the entire chat transcript on every request), DeskBuddy requires the client to send only a lightweight session_id (such as "demo") and the latest message. The backend loads that session's conversation list from Redis under history:{session_id} (or falls back to an in-memory dictionary if Redis is not running), runs a bounded 5-lap tool-calling loop with gpt-4o-mini, persists the updated history back to the store, and returns the final answer alongside a full trace of invoked tools and the total stored message count.

1. Stateless Application Tier + Stateful Redis Tier

In production, Flask servers run multiple worker processes or containers behind a load balancer. Storing chat history in a Python global variable breaks as soon as Request 1 hits Worker A and Request 2 hits Worker B. Externalizing session lists into Redis under history:{session_id} using LRANGE, DELETE, and RPUSH lets any worker serve any user turn seamlessly.

2. Graceful Degradation with Zero-Config Fallback

_get_redis() probes the Redis socket once with a fast 1.5-second connection timeout (client.ping()). If a Redis server is reachable, all session operations use Redis lists; if not, _load_history and _save_history transparently fall back to _fallback_memory and report storage_backend="in-memory (redis fallback)" in the response so the demo works out of the box.

3. Bounded Multi-Turn Tool Loop (Safety Fuse)

A query like "What is 23*47, and what time is it right now?" requires calling both calculator and get_datetime before answering. The loop for _ in range(5): calls the model, appends any assistant tool_calls plus each tool's role="tool" JSON output to history, and repeats until not msg.tool_calls — with the 5-iteration cap preventing runaway infinite loops.

4. Sandboxed Expression Evaluation & Error Isolation

Inside call_tool(), the calculator evaluates arithmetic expressions with eval(expr, {"__builtins__": {}}) (stripping Python built-ins) and wraps tool execution in a try/except block that returns {"error": ...} as a normal dict rather than crashing the request if the model passes malformed syntax.

src/python/03_redis.py implements DeskBuddy, an autonomous tool-calling OpenAI agent whose per-session conversation history is persisted in Redis under history:{session_id} (with automatic in-memory fallback when Redis is unavailable).

1. Core AI Concepts in Stateful Tool-Calling Agents

• Externalized Session Memory (history:{session_id}): LLMs are stateless across API calls. Storing each session's ordered message list in Redis (`r.lrange` / `r.rpush` of JSON-serialized messages) lets the agent remember prior turns, user facts, and previous tool outputs across requests.
• OpenAI Function-Calling Schemas (TOOL_DEFS): Declaring structured JSON schemas for external functions (calculator and get_datetime) allows the LLM to emit structured tool_calls instead of guessing math or current timestamps.
• Bounded Think → Act → Observe Loop: The agent iterates up to 5 laps—calling the LLM, executing every requested tool, appending both the assistant's tool_calls message and each role="tool" result message to history, and re-querying the LLM until it produces a final natural-language answer.

2. Key Entities in src/python/03_redis.py

• RedisAgentRequest, RedisAgentToolCall, & RedisAgentResponse: Pydantic schemas defining the input (session_id: str = "demo", message: str, reset_session: bool = False), each tool invocation trace (tool: str, args: dict, result: dict), and the response payload (session_id, message, answer, tool_calls, stored_messages_count, and storage_backend).
• _get_redis(), _load_history(), _save_history(), & _clear_history():
  • _get_redis() probes redis.Redis(host=host, port=port, decode_responses=True, socket_connect_timeout=1.5) once via client.ping() and falls back to _fallback_memory if unreachable.
  • _load_history(session_id) reads all JSON items from the Redis list r.lrange(f"history:{session_id}", 0, -1) (or _fallback_memory) and returns (history_list, backend_label).
  • _save_history(session_id, history) replaces the Redis list by calling r.delete(key) followed by r.rpush(key, json.dumps(m)) for each message in history.
  • _clear_history(session_id) deletes f"history:{session_id}" from Redis and _fallback_memory.
• TOOL_DEFS & call_tool(name: str, args: dict) -> dict: Define and execute calculator(expression) (using a restricted eval(expr, {"__builtins__": {}})) and get_datetime() (returning datetime.datetime.now().isoformat()).
• run_redis_agent(session_id: str, message: str, reset_session: bool = False) -> dict:
  • Step ① — Optional session reset: If reset_session is true, clears the session via _clear_history(session_id) and returns early if message is empty.
  • Step ② — Load conversation history: Loads prior messages via _load_history(session_id) and appends {"role": "user", "content": message}.
  • Step ③ — Bounded 5-lap agent loop: Calls llm.chat.completions.create(model=model, messages=history, tools=TOOL_DEFS). When msg.tool_calls is empty, breaks out of the loop; otherwise appends msg.model_dump(exclude_none=True), executes each tool via call_tool(), records the step in tool_calls_trace, and appends the corresponding {"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result)} message to history.
  • Step ④ — Persist updated history: Appends the final {"role": "assistant", "content": answer} message, saves the full conversation via _save_history(session_id, history), and returns the response dictionary.

3. End-to-End Stateful Agent Flow

1. run_redis_agent() loads the existing message list for session_id from Redis (or fallback memory) and appends the new user message.
2. During the bounded loop, OpenAI inspects the conversation history and TOOL_DEFS, emitting tool_calls for arithmetic or current-time queries that call_tool() executes locally and feeds back as role="tool" messages.
3. Once the LLM synthesizes its final answer, the updated message history is persisted back to history:{session_id} so subsequent turns in the same session retain full multi-turn context.

DeskBuddy — Tool-Calling Agent with Redis Session Memory Service.

"""DeskBuddy — Tool-Calling Agent with Redis Session Memory Service.

This service demonstrates an autonomous tool-calling agent backed by per-session memory:
1. Connects to Redis (`redis.Redis`) to store and retrieve conversation history per `session_id`
   under `history:{session_id}` (falls back to an in-memory session dictionary when Redis is not running).
2. Exposes two tools (`calculator` and `get_datetime`) via OpenAI function-calling schemas (`TOOL_DEFS`).
3. Runs a bounded agent loop (max 5 laps safety fuse: Think -> Act -> Observe), executing requested tools
   and feeding results back until the LLM produces a final natural-language answer.
"""

import datetime
import json
import os

import redis
from dotenv import find_dotenv, load_dotenv
from openai import OpenAI
from pydantic import BaseModel

load_dotenv(find_dotenv())

_fallback_memory: dict[str, list[dict]] = {}
_redis_client = None
_redis_checked = False


# ---------------------------------------------------------------------------
# Request & Response Schemas (Pydantic)
# ---------------------------------------------------------------------------
class RedisAgentRequest(BaseModel):
    """Input payload for DeskBuddy Redis memory agent."""

    session_id: str = "demo"
    message: str = "What is 23*47, and what time is it right now?"
    reset_session: bool = False


class RedisAgentToolCall(BaseModel):
    """Record of a single tool invocation during the agent loop."""

    tool: str
    args: dict
    result: dict


class RedisAgentResponse(BaseModel):
    """Output payload from DeskBuddy Redis memory agent."""

    session_id: str
    message: str
    answer: str
    tool_calls: list[RedisAgentToolCall]
    stored_messages_count: int
    storage_backend: str


# ---------------------------------------------------------------------------
# Step 1: Redis Memory Store (with In-Memory Fallback)
# ---------------------------------------------------------------------------
def _get_redis():
    """Return a connected Redis client if reachable, or None to use in-memory fallback."""
    global _redis_client, _redis_checked
    if _redis_checked:
        return _redis_client
    _redis_checked = True
    host = os.getenv("REDIS_HOST", "localhost")
    port = int(os.getenv("REDIS_PORT", "6379"))
    try:
        client = redis.Redis(host=host, port=port, decode_responses=True, socket_connect_timeout=1.5)
        client.ping()
        _redis_client = client
    except Exception:
        _redis_client = None
    return _redis_client


def _load_history(session_id: str) -> tuple[list[dict], str]:
    """Load conversation history for `session_id`."""
    key = f"history:{session_id}"
    r = _get_redis()
    if r is not None:
        return [json.loads(m) for m in r.lrange(key, 0, -1)], "redis"
    return list(_fallback_memory.get(key, [])), "in-memory (redis fallback)"


def _save_history(session_id: str, history: list[dict]) -> None:
    """Persist conversation history for `session_id`."""
    key = f"history:{session_id}"
    r = _get_redis()
    if r is not None:
        r.delete(key)
        for m in history:
            r.rpush(key, json.dumps(m))
    else:
        _fallback_memory[key] = list(history)


def _clear_history(session_id: str) -> None:
    """Delete conversation history for `session_id`."""
    key = f"history:{session_id}"
    r = _get_redis()
    if r is not None:
        r.delete(key)
    _fallback_memory.pop(key, None)


# ---------------------------------------------------------------------------
# Step 2: Tool Definitions & Execution
# ---------------------------------------------------------------------------
TOOL_DEFS = [
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "Evaluate a math expression, e.g. '23*47' or '(100-8)/4'",
            "parameters": {
                "type": "object",
                "properties": {"expression": {"type": "string"}},
                "required": ["expression"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_datetime",
            "description": "Get the current date and time",
            "parameters": {"type": "object", "properties": {}},
        },
    },
]


def call_tool(name: str, args: dict) -> dict:
    """Execute the requested tool (`calculator` or `get_datetime`)."""
    try:
        if name == "calculator":
            expr = args.get("expression", "")
            result = eval(expr, {"__builtins__": {}})
            return {"result": result}
        if name == "get_datetime":
            return {"now": datetime.datetime.now().isoformat()}
        return {"error": f"unknown tool: {name}"}
    except Exception as e:
        return {"error": f"tool call failed: {e}"}


# ---------------------------------------------------------------------------
# Step 3: The Agent Loop (Think -> Act -> Observe -> Repeat)
# ---------------------------------------------------------------------------
def run_redis_agent(
    session_id: str = "demo",
    message: str = "What is 23*47, and what time is it right now?",
    reset_session: bool = False,
) -> dict:
    """Run one turn of the DeskBuddy agent with per-session conversation memory."""
    # ① Optionally clear session history if requested
    if reset_session:
        _clear_history(session_id)
        if not message.strip():
            _, backend = _load_history(session_id)
            return {
                "session_id": session_id,
                "message": "",
                "answer": f"Session '{session_id}' memory cleared.",
                "tool_calls": [],
                "stored_messages_count": 0,
                "storage_backend": backend,
            }

    # ② Load this session's conversation history and append the new user message
    history, backend = _load_history(session_id)
    history.append({"role": "user", "content": message})

    llm = OpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url=os.getenv("OPENAI_BASE_URL") or None,
        timeout=20.0,
    )
    model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

    # ③ Bounded agent loop (max 5 laps safety fuse)
    tool_calls_trace: list[dict] = []
    msg = None
    for _ in range(5):
        resp = llm.chat.completions.create(
            model=model,
            messages=history,
            tools=TOOL_DEFS,
        )
        msg = resp.choices[0].message

        if not msg.tool_calls:
            break  # LLM answered in words -> done

        history.append(msg.model_dump(exclude_none=True))
        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            result = call_tool(tc.function.name, args)
            tool_calls_trace.append({
                "tool": tc.function.name,
                "args": args,
                "result": result,
            })
            history.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": json.dumps(result),
            })

    # ④ Save updated conversation history back to Redis/memory and return response
    answer = (msg.content if msg else "") or ""
    history.append({"role": "assistant", "content": answer})
    _save_history(session_id, history)

    return {
        "session_id": session_id,
        "message": message,
        "answer": answer,
        "tool_calls": tool_calls_trace,
        "stored_messages_count": len(history),
        "storage_backend": backend,
    }

Flask Blueprint server and route handler (redis_route) in src/python/app.py.

"""Flask server for 04-ai-model-training-chromadb."""

import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


mod_ml = _load_service("01_model_training.py", "model_training")
mod_chroma = _load_service("02_chromadb.py", "chromadb_rag")
mod_redis = _load_service("03_redis.py", "redis_agent")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/model-training", methods=["POST"])
def model_training_service():
    # ① Parse delivery order features from the JSON request body
    data = request.get_json(force=True, silent=True) or {}
    try:
        distance_km = float(data.get("distance_km", 4.5))
        prep_time_min = float(data.get("prep_time_min", 15.0))
        rider_available = int(data.get("rider_available", 1))
        is_raining = int(data.get("is_raining", 1))
        if distance_km <= 0 or prep_time_min <= 0:
            return jsonify({"error": "Distance and prep time must be positive."}), 400
        # ② Run RandomForestRegressor inference in 01_model_training.py and return ETA
        res = mod_ml.predict_delivery_eta(distance_km, prep_time_min, rider_available, is_raining)
        res["result"] = f"Predicted Arrival: {res['eta_minutes']} min 🍔"
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/chromadb", methods=["POST"])
def chromadb_service():
    # ① Parse student question from JSON payload and reject empty queries
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Query ChromaDB notes collection via 02_chromadb.py and return answer + sources
        res = mod_chroma.ask_chromadb_notes(question)
        res["result"] = res["answer"]
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/redis", methods=["POST"])
def redis_agent_service():
    # ① Parse session_id, message, and optional reset flag from JSON payload
    data = request.get_json(force=True, silent=True) or {}
    session_id = (data.get("session_id") or "demo").strip()
    message = (data.get("message") or "").strip()
    reset_session = bool(data.get("reset_session", False))
    if not session_id or (not reset_session and not message):
        return jsonify({"error": "Session ID and message are required"}), 400
    try:
        # ② Run DeskBuddy agent turn (or session reset) backed by Redis in 03_redis.py
        res = mod_redis.run_redis_agent(session_id, message, reset_session)
        res["result"] = res["answer"]
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Use these numbered takeaways to revise Redis-backed agent memory and raw OpenAI tool calling:

1. Why Externalize Agent Memory to Redis: In-process Python dictionaries vanish on container restarts and are not shared across multiple Gunicorn/Flask worker processes; storing session history in Redis makes the web tier stateless and horizontally scalable.
2. Key-Namespaced Session Storage (history:{session_id}): Prefixing keys with history: and serializing the OpenAI message list via json.dumps() / json.loads() maps each conversation session to an O(1) Redis string lookup.
3. Fast Failover with Socket Timeouts: Configuring socket_connect_timeout=1.5 and caching _redis_checked = True ensures that if Redis is offline, the app falls back to _fallback_memory after a single quick check instead of hanging on every turn.
4. Raw OpenAI Function Calling Protocol: Passing tools=TOOL_DEFS and tool_choice="auto" lets the model emit structured msg.tool_calls containing a unique tc.id, tc.function.name, and JSON-encoded tc.function.arguments.
5. Appending Both Assistant Tool Requests and Tool Outputs: OpenAI's chat completion API requires that every role="tool" message referencing a tool_call_id be immediately preceded by the role="assistant" message that requested that tool_calls array.
6. Persisting Full Trajectories vs. Final Turns: Saving the final assistant reply alongside the user message in history:{session_id} allows future turns to reference calculations and timestamps computed in earlier turns.
7. Sandboxed Expression Evaluation: Restricting eval() with {"__builtins__": {}} and a whitelist of math attributes prevents arbitrary code execution while giving the LLM exact floating-point arithmetic.