Build a RAG-Powered Persona Agent

Build a FastAPI persona agent with RAG retrieval and multi-turn memory.

Introduction

30 Second Summary

Ever chatted with an AI that felt completely generic? Real conversations have personality. A great conversational agent needs to remember what you said, stay in character, and respond like a real person would.

In this project, you will build a persona-based conversational AI that uses Retrieval-Augmented Generation to maintain consistent personality across multi-turn interactions. You'll wire together a vector store, conversation memory, and an LLM into a production-grade API.

What You'll Build

A FastAPI service that ingests personality documents into a ChromaDB vector store, retrieves relevant persona context on each conversation turn, and generates personality-aligned responses via Groq's Llama 3.3 70B model.

By the end of this project, you'll have:

  • A modular RAG pipeline with decoupled persona retrieval, conversation memory, and LLM generation layers that you can test and swap independently.
  • An async conversation memory backed by MongoDB Atlas that maintains multi-turn context using a sliding window strategy.
  • Production-grade engineering patterns including structured JSON logging with correlation IDs, a custom exception hierarchy, and Pydantic-validated request and response models.
  • Secret Mission: a hot-swap endpoint that switches personas mid-conversation while preserving the full conversation history.

Are there any prerequisites?

You'll need Python 3.11 installed, a free Groq account with an API key, and basic familiarity with Python and REST APIs. Step 1 walks you through everything else, including creating a free MongoDB Atlas account.

Before We Start

Your persona agent needs three external services working together: a vector store for personality data, a document database for conversation memory, and a language model for generating responses.

In this step, you will install every dependency, create your free cloud database, and confirm that all connections are live before writing any application code.

In this step, get ready to:
  • Create your project and install all dependencies.
  • Set up a free MongoDB Atlas cluster for conversation storage.
  • Verify your connections to MongoDB and Groq.
Create your project and install dependencies

Every Python project starts with a dedicated folder and a virtual environment to keep its packages isolated from the rest of your system.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir persona-agent
cd persona-agent
  • Create a Python virtual environment by running:
python3.11 -m venv venv
  • Activate your virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you are inside the persona-agent folder before activating. Run pwd (macOS) or cd (Windows) to check your current directory.

Help me activate my Python virtual environment.

Now you will create a requirements.txt file that lists every package your project needs.

  • Open Cursor by pressing Cmd+Space (macOS) or the Windows key (Windows), typing Cursor, and pressing Enter.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Select the persona-agent folder on your Desktop, then click Open.
  • Create a new file called requirements.txt in the root of your project by right-clicking in the Explorer sidebar and selecting New File.
  • Paste the following into requirements.txt:
fastapi==0.138.1
uvicorn[standard]==0.34.3
pydantic-settings==2.9.1
python-dotenv==1.1.0
groq==1.5.0
pymongo==4.17.0
chromadb==1.5.9
sentence-transformers==5.6.0

What are these packages?

  • fastapi and uvicorn give you the async web server for your API endpoints.
  • pydantic-settings and python-dotenv load and validate configuration from environment variables.
  • groq is the SDK for calling Groq's LLM API (Llama 3.3 70B).
  • pymongo is the official MongoDB driver with native async support for conversation storage.
  • chromadb is the vector database that stores your persona embeddings locally.
  • sentence-transformers provides the embedding model that converts text into vectors.
  • Save requirements.txt.
  • Install all dependencies by running this command in your terminal:
pip install -r requirements.txt

This will take a few minutes. The sentence-transformers package downloads a machine learning model (~80 MB), so the install is larger than a typical Python project.

Seeing installation errors?

Make sure your virtual environment is active (you see (venv) in your prompt). If you closed your terminal, navigate back to the persona-agent folder and run the activate command again.

Help me fix my pip install errors.

Set up MongoDB Atlas

Your persona agent stores conversation history in MongoDB Atlas. Atlas provides a free-forever cloud database with 512 MB of storage, which is more than enough for this project.

Once you are on your Atlas dashboard, you will create a Free cluster.

  • Click the Create button to start creating a new cluster.
  • Select the M0 (Free) option.
  • Choose any cloud provider (AWS, GCP, or Azure) and select a region close to you.
  • Enter a cluster name. You can use persona-agent or leave the default.
  • Click Create to deploy your cluster.

Atlas opens a Security Quickstart wizard. You need to create a database user and allow your IP address.

  • Enter a Username for your database user (for example, persona-admin).
  • Enter a Password or copy the auto-generated one. Save this password somewhere safe.
  • Click Create Database User.
  • Click Add My Current IP Address.
  • Click Finish and Close.
  • Click Go to Overview.

Your cluster takes less than 15 seconds to deploy. Now you need to get your connection string.

  • On the Clusters page, click Connect next to your cluster name.
  • Click Drivers.
  • Select Python as the driver.
  • Copy the connection string shown. It looks like: mongodb+srv://your-username:<password>@your-cluster.mongodb.net/?retryWrites=true&w=majority.
  • Replace <password> with the database user password you just created.

Why MongoDB Atlas for conversation memory?

MongoDB stores conversation turns as flexible documents. Unlike a relational database, you do not need to define a rigid schema upfront. This makes it ideal for chat data where each turn has variable metadata.

The Free tier gives you 512 MB of storage and auto-pauses after 30 days of inactivity. No credit card is required.

Now create the .env file that stores your connection credentials.

  • In Cursor, create a new file called .env in the root of your persona-agent folder.
  • Paste the following configuration into .env:
GROQ_API_KEY=your-api-key-here
MONGODB_URI=mongodb+srv://your-username:your-password@your-cluster.mongodb.net/?retryWrites=true&w=majority
APP_NAME=persona-agent
LOG_LEVEL=INFO
PERSONA_COLLECTION=personas
CONVERSATION_DB=persona_agent
  • Replace your-api-key-here with your Groq API key from https://console.groq.com/keys.
  • Replace the MONGODB_URI value with the connection string you just copied (with your real password filled in).
  • Save .env.

Keep your .env file private

Never commit .env to version control. If you use Git, add .env to your .gitignore file to prevent accidentally sharing your credentials.

Verify your connections

Before building any application logic, you will confirm that your MongoDB Atlas and Groq connections both work. This prevents debugging connection issues later when the code is more complex.

  • Create a new file called test_connections.py in the root of your project.
  • Paste the following test script into test_connections.py:
import asyncio
import os
from dotenv import load_dotenv
from pymongo import AsyncMongoClient
from groq import AsyncGroq

# Load environment variables from .env
load_dotenv()

async def test_mongodb():
    """Ping MongoDB Atlas to confirm the connection works."""
    uri = os.getenv("MONGODB_URI")
    client = AsyncMongoClient(uri)
    try:
        await client.admin.command("ping")
        print("MongoDB Atlas: Connection successful!")
    except Exception as e:
        print(f"MongoDB Atlas: Connection failed - {e}")
    finally:
        await client.close()

async def test_groq():
    """Send a simple request to Groq to confirm the API key works."""
    client = AsyncGroq(api_key=os.getenv("GROQ_API_KEY"))
    try:
        response = await client.chat.completions.create(
            messages=[{"role": "user", "content": "Say hello in one word."}],
            model="llama-3.3-70b-versatile",
            max_tokens=10,
        )
        print(f"Groq API: Connection successful! Response: {response.choices[0].message.content}")
    except Exception as e:
        print(f"Groq API: Connection failed - {e}")

async def main():
    await test_mongodb()
    await test_groq()

if __name__ == "__main__":
    asyncio.run(main())

What does this script do?

  • test_mongodb() connects to your Atlas cluster using AsyncMongoClient and sends a ping command. If the connection string and credentials are correct, Atlas replies with a success signal.
  • test_groq() sends a minimal chat request to the llama-3.3-70b-versatile model via the AsyncGroq client. If your API key is valid, you get a one-word response back.
  • Save test_connections.py.
  • Run the test script:
python test_connections.py

You should see two success messages confirming both MongoDB Atlas and Groq are connected.

✔️ I see two success messages

Both connections are working. You are ready to start building.

ⓧ MongoDB connection failed

  • Double-check your MONGODB_URI in .env. Make sure you replaced <password> with your actual database user password (not your Atlas login password).
  • Confirm your IP address is in the Atlas IP Access List. Go back to Atlas, click Network Access in the left sidebar, and verify your current IP is listed.
  • If you changed networks since creating the cluster, click Add IP Address and add your current IP.

Help me fix my MongoDB Atlas connection.

ⓧ Groq connection failed

  • Verify your GROQ_API_KEY in .env. Copy it fresh from https://console.groq.com/keys.
  • Make sure there are no extra spaces or quotes around the key value in .env.
  • Confirm your Groq account is active and not rate-limited.

Help me fix my Groq API connection.

✔️ Awesome, I've got everything!

Great. Double check you have saved both requirements.txt and .env.

ⓧ I'd like to double check the full code

Here is what your requirements.txt should look like:

fastapi==0.138.1
uvicorn[standard]==0.34.3
pydantic-settings==2.9.1
python-dotenv==1.1.0
groq==1.5.0
pymongo==4.17.0
chromadb==1.5.9
sentence-transformers==5.6.0

Here is what your .env should look like (with your own values filled in):

GROQ_API_KEY=your-api-key-here
MONGODB_URI=mongodb+srv://your-username:your-password@your-cluster.mongodb.net/?retryWrites=true&w=majority
APP_NAME=persona-agent
LOG_LEVEL=INFO
PERSONA_COLLECTION=personas
CONVERSATION_DB=persona_agent

Your environment is ready with all dependencies installed and both external services confirmed. Next up, you will build the configuration and logging layer that ties everything together.

Build the Configuration and Logging Layer

Your environment is set up, your dependencies are installed, and your MongoDB Atlas cluster is ready. Now it's time to lay the foundation that every other module in your project will depend on.

Production services need consistent configuration management, structured logging, and predictable error handling. In this step, you'll build all three using Pydantic Settings for config, Python's built-in logging module for structured JSON output, and a custom exception hierarchy for clean error responses.

In this step, get ready to:
  • Create a configuration module that loads and validates all environment variables.
  • Build structured JSON logging with correlation IDs for request tracing.
  • Define a custom exception hierarchy with a FastAPI error handler.
Create the settings and logging modules

Every module in your persona agent needs access to configuration values like your API key, database URI, and model names. Instead of scattering environment variable reads across files, you'll centralize everything in one validated settings class.

  • Create the app/ package directory and an empty __init__.py file by running these commands:
mkdir app
touch app/__init__.py
  • Confirm the directory was created by running:
ls app/

You should see __init__.py in the output.

  • Create a new file called app/config.py in Cursor by right-clicking the app folder in the Explorer sidebar, selecting New File, and naming it config.py.
  • Add the following code to app/config.py:
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "persona-agent"
    log_level: str = "INFO"
    groq_api_key: str
    mongodb_uri: str
    conversation_db: str = "persona_agent"
    persona_collection: str = "personas"
    chroma_persist_dir: str = "./chroma_data"
    embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"
    llm_model: str = "llama-3.3-70b-versatile"
    max_history_turns: int = 10
    persona_top_k: int = 5

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"


settings = Settings()

What does this code do?

The Settings class inherits from Pydantic's BaseSettings, which automatically reads environment variables and validates their types.

  • groq_api_key and mongodb_uri have no defaults, so the app fails fast if they're missing from your .env file.
  • Fields like max_history_turns and persona_top_k have sensible defaults that you can override via environment variables later.
  • The Config inner class tells Pydantic to load variables from your .env file automatically.
  • settings = Settings() at module level creates a singleton that any file can import.
  • Save app/config.py.
  • Verify it loads correctly by running this in your terminal:
python -c "from app.config import settings; print(settings.app_name)"

You should see persona-agent printed to the terminal.

Seeing a validation error?

If you see a ValidationError mentioning groq_api_key or mongodb_uri, your .env file is missing one of the required variables. Double-check that your .env is in the project root (same folder where you run the command) and contains both GROQ_API_KEY and MONGODB_URI.

help me debug my Pydantic Settings validation error

Now you need a logging module that outputs structured JSON instead of plain text. Structured logs make it possible to search and filter by fields like correlation ID or log level in production.

  • Create a new file called app/logging_config.py using the same right-click method in the Explorer sidebar.
  • Add the imports, correlation ID variable, and JSON formatter class:
import logging
import json
import uuid
from contextvars import ContextVar

correlation_id_var: ContextVar[str] = ContextVar("correlation_id", default="")


class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "correlation_id": correlation_id_var.get(""),
        }
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_entry)

What does this code do?

correlation_id_var is a ContextVar that stores a unique ID per request. Because FastAPI handles multiple requests concurrently, a regular global variable would get overwritten. ContextVar keeps each async task's value isolated.

The JSONFormatter class overrides Python's default log formatter to output each log entry as a JSON object with consistent fields like timestamp, level, and correlation_id.

  • Below the JSONFormatter class, add the utility functions that other modules will use:

def setup_logging(level: str = "INFO"):
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())
    root_logger = logging.getLogger()
    root_logger.handlers.clear()
    root_logger.addHandler(handler)
    root_logger.setLevel(getattr(logging, level.upper(), logging.INFO))


def get_logger(name: str) -> logging.Logger:
    return logging.getLogger(name)


def new_correlation_id() -> str:
    cid = str(uuid.uuid4())
    correlation_id_var.set(cid)
    return cid

What do these functions do?

  • setup_logging() wires up the JSON formatter to Python's root logger and sets the log level. Calling root_logger.handlers.clear() first prevents duplicate handlers if setup runs more than once.
  • get_logger() is a convenience wrapper so every module gets a named logger with one import.
  • new_correlation_id() generates a UUID and stores it in the ContextVar, so every log entry within that request automatically includes the same ID.
  • Save app/logging_config.py.
  • Test the logging module by running:
python -c "from app.logging_config import setup_logging, get_logger; setup_logging(); logger = get_logger('test'); logger.info('Hello from persona-agent')"

You should see a JSON-formatted log entry with fields like timestamp, level, logger, and message printed to the terminal.

Seeing a plain text log instead of JSON?

If the output looks like a standard Python log line (not JSON), make sure setup_logging() is called before logger.info(). The formatter only applies after setup runs. Also check that you saved the file with the JSONFormatter class included.

help me debug my structured JSON logging setup

✔️ Awesome, I've got everything!

Great. Make sure both app/config.py and app/logging_config.py are saved.

ⓧ I'd like to double check the full code

Here is the complete app/config.py:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "persona-agent"
    log_level: str = "INFO"
    groq_api_key: str
    mongodb_uri: str
    conversation_db: str = "persona_agent"
    persona_collection: str = "personas"
    chroma_persist_dir: str = "./chroma_data"
    embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"
    llm_model: str = "llama-3.3-70b-versatile"
    max_history_turns: int = 10
    persona_top_k: int = 5

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"


settings = Settings()

Here is the complete app/logging_config.py:

import logging
import json
import uuid
from contextvars import ContextVar

correlation_id_var: ContextVar[str] = ContextVar("correlation_id", default="")


class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "correlation_id": correlation_id_var.get(""),
        }
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_entry)


def setup_logging(level: str = "INFO"):
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())
    root_logger = logging.getLogger()
    root_logger.handlers.clear()
    root_logger.addHandler(handler)
    root_logger.setLevel(getattr(logging, level.upper(), logging.INFO))


def get_logger(name: str) -> logging.Logger:
    return logging.getLogger(name)


def new_correlation_id() -> str:
    cid = str(uuid.uuid4())
    correlation_id_var.set(cid)
    return cid
Create the exception hierarchy and error handlers

When something goes wrong in your API, you want predictable error responses. A custom exception hierarchy lets each module raise specific errors that FastAPI automatically catches and formats into consistent JSON responses.

  • Create a new file called app/exceptions.py in the app/ folder.
  • Add the base exception class and its four subclasses:
from fastapi import Request
from fastapi.responses import JSONResponse


class AppError(Exception):
    def __init__(self, message: str, status_code: int = 500):
        self.message = message
        self.status_code = status_code
        super().__init__(self.message)


class PersonaNotFoundError(AppError):
    def __init__(self, persona_id: str):
        super().__init__(
            message=f"Persona '{persona_id}' not found",
            status_code=404,
        )


class VectorStoreError(AppError):
    def __init__(self, message: str):
        super().__init__(message=message, status_code=503)


class LLMGenerationError(AppError):
    def __init__(self, message: str):
        super().__init__(message=message, status_code=502)


class ConversationNotFoundError(AppError):
    def __init__(self, session_id: str):
        super().__init__(
            message=f"Conversation session '{session_id}' not found",
            status_code=404,
        )

What does this hierarchy do?

AppError is the base class that carries a message and status_code. Each subclass pre-fills these for a specific failure mode:

  • PersonaNotFoundError (404) fires when a persona ID doesn't exist in the vector store.
  • VectorStoreError (503) fires when ChromaDB is unreachable or fails an operation.
  • LLMGenerationError (502) fires when the Groq API call fails.
  • ConversationNotFoundError (404) fires when a session ID has no history in MongoDB.
  • Below the exception classes, add the FastAPI error handler function:

async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.__class__.__name__,
            "message": exc.message,
        },
    )

What does the error handler do?

When you register this handler with FastAPI, any AppError (or subclass) raised anywhere in your app gets caught automatically. Instead of a generic 500 error, the client receives a clean JSON body with the error class name and message.

  • Save app/exceptions.py.
  • Verify the module imports cleanly by running:
python -c "from app.exceptions import AppError, PersonaNotFoundError, VectorStoreError, LLMGenerationError, ConversationNotFoundError, app_error_handler; print('All exceptions loaded OK')"

You should see All exceptions loaded OK printed.

Seeing an ImportError?

Make sure your virtual environment is still activated (you should see the environment name in your terminal prompt). The fastapi import at the top of exceptions.py requires FastAPI to be installed. Also check for typos in the class names and that the file is saved inside the app/ folder (not the project root).

help me fix my import error in app/exceptions.py

✔️ Awesome, I've got everything!

Great. Make sure app/exceptions.py is saved.

ⓧ I'd like to double check the full code

Here is the complete app/exceptions.py:

from fastapi import Request
from fastapi.responses import JSONResponse


class AppError(Exception):
    def __init__(self, message: str, status_code: int = 500):
        self.message = message
        self.status_code = status_code
        super().__init__(self.message)


class PersonaNotFoundError(AppError):
    def __init__(self, persona_id: str):
        super().__init__(
            message=f"Persona '{persona_id}' not found",
            status_code=404,
        )


class VectorStoreError(AppError):
    def __init__(self, message: str):
        super().__init__(message=message, status_code=503)


class LLMGenerationError(AppError):
    def __init__(self, message: str):
        super().__init__(message=message, status_code=502)


class ConversationNotFoundError(AppError):
    def __init__(self, session_id: str):
        super().__init__(
            message=f"Conversation session '{session_id}' not found",
            status_code=404,
        )


async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.__class__.__name__,
            "message": exc.message,
        },
    )

Your configuration, logging, and error handling layers are all in place. Next up, you'll build the persona vector store that powers the RAG retrieval pipeline.

Build the Persona Vector Store

Your configuration, logging, and exception layers are in place. Now you need to give your persona agent its memory of who it should be.

In this step, you will build the RAG layer that stores persona documents as vector embeddings and retrieves the most relevant personality fragments at query time. This is what makes the agent's responses sound like a specific person rather than a generic chatbot.

In this step, get ready to:
  • Create the persona ingestion module with ChromaDB and Sentence Transformers.
  • Create the persona retrieval function for top-k context lookup.
  • Test ingestion and retrieval with a sample Marcus Aurelius persona.
Create the persona ingestion module

The persona store is responsible for turning personality descriptions into searchable vectors. When a user sends a message, the system needs to find the most relevant personality fragments to include in the prompt.

What is a vector store?

A vector store is a database that stores data as numerical vectors (embeddings). Instead of searching by exact keywords, you search by meaning. Two sentences about "handling hardship with grace" and "overcoming obstacles with resilience" would be stored near each other in vector space, even though they share no exact words.

This project uses ChromaDB as the vector store and Sentence Transformers to generate the embeddings.

  • Create a new file called persona_store.py inside the app/ folder.
  • Add the imports and singleton initialization functions by pasting this code:
import chromadb
from sentence_transformers import SentenceTransformer

from app.config import settings
from app.exceptions import VectorStoreError, PersonaNotFoundError
from app.logging_config import get_logger

logger = get_logger(__name__)

_client: chromadb.ClientAPI | None = None
_embedding_model: SentenceTransformer | None = None


def get_chroma_client() -> chromadb.ClientAPI:
    global _client
    if _client is None:
        _client = chromadb.PersistentClient(path=settings.chroma_persist_dir)
        logger.info("ChromaDB client initialized", extra={"path": settings.chroma_persist_dir})
    return _client


def get_embedding_model() -> SentenceTransformer:
    global _embedding_model
    if _embedding_model is None:
        _embedding_model = SentenceTransformer(settings.embedding_model)
        logger.info("Sentence Transformer model loaded", extra={"model": settings.embedding_model})
    return _embedding_model

What does this code do?

The module-level _client and _embedding_model variables act as singletons. The first time you call get_chroma_client() or get_embedding_model(), the heavy object is created once and reused for every subsequent call.

The PersistentClient stores data on disk at ./chroma_data so your persona embeddings survive restarts. SentenceTransformer loads the all-MiniLM-L6-v2 model that converts text into 384-dimensional vectors.

First-time model download

The first time get_embedding_model() runs, Sentence Transformers downloads the model weights from Hugging Face Hub. This is about 80 MB and takes a minute or two depending on your connection. Subsequent calls use the locally cached model.

Now add the ingestion function below your singleton functions. This function takes persona documents and stores them as embeddings.

  • Add the ingest_persona function below get_embedding_model() by pasting this code:
def ingest_persona(persona_id: str, documents: list[dict]) -> int:
    try:
        client = get_chroma_client()
        model = get_embedding_model()
        collection = client.get_or_create_collection(name=f"persona_{persona_id}")

        texts = [doc["content"] for doc in documents]
        metadatas = [{"category": doc["category"], "persona_id": persona_id} for doc in documents]
        ids = [f"{persona_id}_{i}" for i in range(len(documents))]

        embeddings = model.encode(texts).tolist()

        collection.upsert(
            ids=ids,
            embeddings=embeddings,
            documents=texts,
            metadatas=metadatas,
        )
        logger.info("Persona ingested", extra={"persona_id": persona_id, "doc_count": len(documents)})
        return len(documents)
    except Exception as e:
        logger.error("Failed to ingest persona", extra={"persona_id": persona_id, "error": str(e)})
        raise VectorStoreError(f"Failed to ingest persona: {e}")

What does this code do?

The ingest_persona function creates (or retrieves) a ChromaDB collection named persona_{persona_id}. Each persona gets its own collection so you can manage them independently.

It extracts the text content from each document, generates embeddings using the Sentence Transformer model, and upserts them into the collection. The upsert method inserts new documents or updates existing ones with the same ID, making re-ingestion safe.

  • Save app/persona_store.py.

You will not see a visible change yet. The retrieval function in the next substep completes the module.

Seeing an import error?

If Python reports it cannot find chromadb or sentence_transformers, make sure your virtual environment is activated. Check that you installed all dependencies from requirements.txt in the previous step.

help me debug why my imports are failing for chromadb or sentence_transformers

Create the persona retrieval function

Ingestion stores the persona. Retrieval is what makes RAG work at conversation time. Given a user message, you need to find the most relevant personality fragments to feed into the LLM prompt.

  • In app/persona_store.py, add the retrieve_persona_context function below ingest_persona:
def retrieve_persona_context(persona_id: str, query: str, top_k: int = 5) -> list[str]:
    try:
        client = get_chroma_client()
        model = get_embedding_model()

        collection_name = f"persona_{persona_id}"
        try:
            collection = client.get_collection(name=collection_name)
        except Exception:
            raise PersonaNotFoundError(persona_id)

        query_embedding = model.encode([query]).tolist()

        results = collection.query(
            query_embeddings=query_embedding,
            n_results=top_k,
        )
        documents = results.get("documents", [[]])[0]
        logger.info(
            "Persona context retrieved",
            extra={"persona_id": persona_id, "results_count": len(documents)},
        )
        return documents
    except PersonaNotFoundError:
        raise
    except Exception as e:
        logger.error("Failed to retrieve persona context", extra={"error": str(e)})
        raise VectorStoreError(f"Retrieval failed: {e}")

What does this code do?

The function encodes the user's message into a vector, then queries the persona's ChromaDB collection for the top_k most similar documents. ChromaDB returns these ranked by cosine similarity.

If the collection does not exist (meaning no persona with that ID was ingested), the function raises PersonaNotFoundError. The outer except PersonaNotFoundError: raise pattern ensures that specific error passes through without being caught by the generic exception handler.

  • Save app/persona_store.py.

Confused about the double try/except?

The inner try/except catches the specific case where get_collection fails because the persona was never ingested. The outer try/except catches anything else unexpected (network issues, corrupt data) and wraps it in a VectorStoreError.

help me understand the exception handling pattern in retrieve_persona_context

✔️ Awesome, I've got everything!

Great. Double check you have saved app/persona_store.py with both functions.

ⓧ I'd like to double check the full code

Here is the complete app/persona_store.py file:

import chromadb
from sentence_transformers import SentenceTransformer

from app.config import settings
from app.exceptions import VectorStoreError, PersonaNotFoundError
from app.logging_config import get_logger

logger = get_logger(__name__)

_client: chromadb.ClientAPI | None = None
_embedding_model: SentenceTransformer | None = None


def get_chroma_client() -> chromadb.ClientAPI:
    global _client
    if _client is None:
        _client = chromadb.PersistentClient(path=settings.chroma_persist_dir)
        logger.info("ChromaDB client initialized", extra={"path": settings.chroma_persist_dir})
    return _client


def get_embedding_model() -> SentenceTransformer:
    global _embedding_model
    if _embedding_model is None:
        _embedding_model = SentenceTransformer(settings.embedding_model)
        logger.info("Sentence Transformer model loaded", extra={"model": settings.embedding_model})
    return _embedding_model


def ingest_persona(persona_id: str, documents: list[dict]) -> int:
    try:
        client = get_chroma_client()
        model = get_embedding_model()
        collection = client.get_or_create_collection(name=f"persona_{persona_id}")

        texts = [doc["content"] for doc in documents]
        metadatas = [{"category": doc["category"], "persona_id": persona_id} for doc in documents]
        ids = [f"{persona_id}_{i}" for i in range(len(documents))]

        embeddings = model.encode(texts).tolist()

        collection.upsert(
            ids=ids,
            embeddings=embeddings,
            documents=texts,
            metadatas=metadatas,
        )
        logger.info("Persona ingested", extra={"persona_id": persona_id, "doc_count": len(documents)})
        return len(documents)
    except Exception as e:
        logger.error("Failed to ingest persona", extra={"persona_id": persona_id, "error": str(e)})
        raise VectorStoreError(f"Failed to ingest persona: {e}")


def retrieve_persona_context(persona_id: str, query: str, top_k: int = 5) -> list[str]:
    try:
        client = get_chroma_client()
        model = get_embedding_model()

        collection_name = f"persona_{persona_id}"
        try:
            collection = client.get_collection(name=collection_name)
        except Exception:
            raise PersonaNotFoundError(persona_id)

        query_embedding = model.encode([query]).tolist()

        results = collection.query(
            query_embeddings=query_embedding,
            n_results=top_k,
        )
        documents = results.get("documents", [[]])[0]
        logger.info(
            "Persona context retrieved",
            extra={"persona_id": persona_id, "results_count": len(documents)},
        )
        return documents
    except PersonaNotFoundError:
        raise
    except Exception as e:
        logger.error("Failed to retrieve persona context", extra={"error": str(e)})
        raise VectorStoreError(f"Retrieval failed: {e}")
Test ingestion and retrieval with a sample persona

Time to prove the vector store works end to end. You will ingest a Marcus Aurelius persona with Stoic philosophy traits, then query it to verify relevant fragments come back.

  • Create a file called test_persona.py in your project root (the same level as requirements.txt) by pasting this code:
from app.persona_store import ingest_persona, retrieve_persona_context

# Sample Marcus Aurelius persona documents
documents = [
    {"category": "traits", "content": "I am Marcus Aurelius, Roman Emperor and Stoic philosopher. I value discipline, duty, and rational thought above all else."},
    {"category": "background", "content": "I ruled the Roman Empire from 161 to 180 AD during a period of wars and plague. Despite immense power, I remained committed to philosophical contemplation."},
    {"category": "style", "content": "I speak with measured calm and gravitas. I prefer short, reflective statements. I often pose questions to guide others toward their own wisdom."},
    {"category": "knowledge", "content": "My Meditations are private philosophical writings exploring Stoic principles: accept what you cannot control, focus on virtue, and remember that all things are impermanent."},
    {"category": "knowledge", "content": "A Stoic believes that the obstacle is the way. Hardship is not something to avoid but something to grow through. Virtue is the only true good."},
]

# Ingest the persona
count = ingest_persona("marcus-aurelius", documents)
print(f"Ingested {count} documents")

# Test retrieval
results = retrieve_persona_context("marcus-aurelius", "How should I handle adversity?")
print(f"\nRetrieved {len(results)} relevant fragments:")
for i, doc in enumerate(results, 1):
    print(f"\n--- Fragment {i} ---")
    print(doc)

What does this test do?

The script loads five persona documents spanning traits, background, speaking style, and knowledge. It ingests them into a ChromaDB collection called persona_marcus-aurelius, then queries with a question about adversity.

The retrieval should rank the Stoic philosophy fragments (about hardship being the way, and about accepting what you cannot control) highest because they are semantically closest to the query about handling adversity.

  • Run the test script by executing this command in your terminal:
python test_persona.py

You should see output confirming 5 documents were ingested, followed by 5 retrieved fragments ranked by relevance to the adversity query.

What should I see?

The first time you run this, you will see download progress bars as Sentence Transformers fetches the all-MiniLM-L6-v2 model. After that, you should see:

  • "Ingested 5 documents" printed to the terminal.
  • "Retrieved 5 relevant fragments:" followed by the five persona documents, with the Stoic hardship/virtue fragments appearing near the top.
  • A chroma_data/ folder created in your project directory containing the persisted vector data.

Seeing a slow download or timeout?

The all-MiniLM-L6-v2 model is about 80 MB. If the download stalls, check your internet connection and try again. The model is cached locally after the first successful download, so future runs will be instant.

If you see an error about pydantic_settings or missing environment variables, make sure your .env file is in the project root and your virtual environment is active.

help me debug why my test_persona.py script is failing

Your persona vector store is working. You can ingest personality documents and retrieve the most relevant ones based on semantic similarity. Next up, you will build the conversation memory layer so your agent can remember what was said earlier in the chat.

Build the Conversation Memory Layer

Your persona vector store can now retrieve relevant personality fragments for any user message. The next challenge is giving your agent memory of the conversation itself.

Without conversation history, your agent would treat every message as a fresh interaction. It would forget what the user said two messages ago. In this step, you will build a MongoDB-backed conversation store that records every turn and retrieves recent history for prompt context.

In this step, get ready to:
  • Create the conversation store module with async MongoDB operations.
  • Implement a sliding window memory strategy for efficient history retrieval.
Create the conversation store module

Each conversation turn consists of a user message and the assistant's response, tied to a session ID. Your store needs to write these turns to MongoDB and read them back in chronological order.

Why PyMongo's native AsyncMongoClient?

The traditional async MongoDB driver (Motor) was deprecated in May 2026. PyMongo 4.17.0 now ships with AsyncMongoClient natively. This means one fewer dependency and you are learning the current recommended path for async MongoDB in Python.

  • Create a new file called conversation_store.py inside the app/ folder.
  • Add the imports, the MongoDB client singleton, and the collection accessor by pasting this code:
from datetime import datetime, timezone

from pymongo import AsyncMongoClient

from app.config import settings
from app.logging_config import get_logger

logger = get_logger(__name__)

_client: AsyncMongoClient | None = None


def get_mongo_client() -> AsyncMongoClient:
    global _client
    if _client is None:
        _client = AsyncMongoClient(settings.mongodb_uri)
        logger.info("MongoDB async client initialized")
    return _client


def get_conversations_collection():
    client = get_mongo_client()
    db = client[settings.conversation_db]
    return db["conversations"]

What does this code do?

  • get_mongo_client() creates a singleton AsyncMongoClient on first call. Every subsequent call returns the same client instance, avoiding repeated connection overhead.
  • get_conversations_collection() navigates to the persona_agent database (set in your .env as CONVERSATION_DB) and returns the conversations collection. MongoDB creates the database and collection automatically on first write.

Now you need a function that writes a single conversation turn. Each turn captures who said what and when.

  • Add the add_turn function below the get_conversations_collection function:
async def add_turn(session_id: str, persona_id: str, user_message: str, assistant_message: str):
    collection = get_conversations_collection()
    turn = {
        "session_id": session_id,
        "persona_id": persona_id,
        "user_message": user_message,
        "assistant_message": assistant_message,
        "timestamp": datetime.now(timezone.utc),
    }
    await collection.insert_one(turn)
    logger.info("Conversation turn stored", extra={"session_id": session_id})

What does this code do?

  • add_turn() is an async function that inserts a document into the conversations collection. Each document stores the session ID, persona ID, both sides of the exchange, and a UTC timestamp.
  • The session_id field groups turns that belong to the same conversation. Later, you will query by session ID to reconstruct the full chat history.
  • Save app/conversation_store.py.

Seeing a red underline on AsyncMongoClient?

Make sure your virtual environment is active and that pymongo==4.17.0 is installed. Older versions of PyMongo do not include the async client.

help me fix my AsyncMongoClient import error

Implement a sliding window memory strategy

Sending every message from a long conversation into the LLM prompt would quickly exceed token limits and increase cost. A sliding window retrieves only the most recent N turns, keeping context manageable while preserving conversational flow.

  • Add the get_history function below add_turn in app/conversation_store.py:
async def get_history(session_id: str, max_turns: int = 10) -> list[dict]:
    collection = get_conversations_collection()
    cursor = collection.find(
        {"session_id": session_id},
        {"_id": 0, "user_message": 1, "assistant_message": 1, "timestamp": 1},
    ).sort("timestamp", -1).limit(max_turns)

    turns = []
    async for doc in cursor:
        turns.append(doc)

    turns.reverse()
    logger.info("History retrieved", extra={"session_id": session_id, "turns": len(turns)})
    return turns

What does this code do?

  • The query sorts by timestamp descending and limits results to max_turns. This grabs the most recent turns efficiently without loading the entire conversation.
  • turns.reverse() flips the results back to chronological order. The LLM needs history oldest-first so the conversation reads naturally in the prompt.
  • The projection ({"_id": 0, "user_message": 1, ...}) returns only the fields the prompt assembly needs, keeping memory usage low.

You also need a helper that looks up which persona a session is currently using. This becomes important later when the prompt assembler needs to know the active persona without the caller passing it explicitly.

  • Add the get_session_persona function below get_history:
async def get_session_persona(session_id: str) -> str | None:
    collection = get_conversations_collection()
    doc = await collection.find_one(
        {"session_id": session_id},
        {"_id": 0, "persona_id": 1},
        sort=[("timestamp", -1)],
    )
    return doc["persona_id"] if doc else None

What does this code do?

get_session_persona() finds the most recent turn for a given session and returns its persona_id. If no conversation exists yet, it returns None. This is useful for features like persona switching where you need to know the current active persona.

  • Save app/conversation_store.py.
  • Verify the module loads without errors by running this in your terminal:
python -c "from app.conversation_store import get_mongo_client, add_turn, get_history; print('conversation_store OK')"

You should see conversation_store OK printed with no import errors.

Seeing an import error?

Check that your .env file contains MONGODB_URI and GROQ_API_KEY. The Settings class validates all required fields on import, so a missing key causes an immediate failure.

Also confirm your virtual environment is active. If you see ModuleNotFoundError: No module named 'pymongo', your shell is using the system Python instead.

help me debug my conversation_store import error

✔️ Awesome, I've got everything!

Great. Double check you have saved app/conversation_store.py before moving on.

ⓧ I'd like to double check the full code

from datetime import datetime, timezone

from pymongo import AsyncMongoClient

from app.config import settings
from app.logging_config import get_logger

logger = get_logger(__name__)

_client: AsyncMongoClient | None = None


def get_mongo_client() -> AsyncMongoClient:
    global _client
    if _client is None:
        _client = AsyncMongoClient(settings.mongodb_uri)
        logger.info("MongoDB async client initialized")
    return _client


def get_conversations_collection():
    client = get_mongo_client()
    db = client[settings.conversation_db]
    return db["conversations"]


async def add_turn(session_id: str, persona_id: str, user_message: str, assistant_message: str):
    collection = get_conversations_collection()
    turn = {
        "session_id": session_id,
        "persona_id": persona_id,
        "user_message": user_message,
        "assistant_message": assistant_message,
        "timestamp": datetime.now(timezone.utc),
    }
    await collection.insert_one(turn)
    logger.info("Conversation turn stored", extra={"session_id": session_id})


async def get_history(session_id: str, max_turns: int = 10) -> list[dict]:
    collection = get_conversations_collection()
    cursor = collection.find(
        {"session_id": session_id},
        {"_id": 0, "user_message": 1, "assistant_message": 1, "timestamp": 1},
    ).sort("timestamp", -1).limit(max_turns)

    turns = []
    async for doc in cursor:
        turns.append(doc)

    turns.reverse()
    logger.info("History retrieved", extra={"session_id": session_id, "turns": len(turns)})
    return turns


async def get_session_persona(session_id: str) -> str | None:
    collection = get_conversations_collection()
    doc = await collection.find_one(
        {"session_id": session_id},
        {"_id": 0, "persona_id": 1},
        sort=[("timestamp", -1)],
    )
    return doc["persona_id"] if doc else None

Your conversation memory layer is ready to record and replay dialogue history. Next up, you will build the LLM generation layer that assembles persona context, conversation history, and the user message into a single prompt for Groq.

Build the LLM Layer and Wire Up the API

You have a persona vector store that retrieves personality fragments and a conversation memory layer that tracks multi-turn history. Now you need the piece that ties them together: the LLM generation layer and a FastAPI application that orchestrates the full pipeline.

In this step, you will build the prompt assembly logic, the Groq client wrapper, and a complete API with endpoints for chatting, ingesting personas, and checking system health.

In this step, get ready to:
  • Build a prompt assembly function that merges persona context, conversation history, and the user message into a structured message list.
  • Create an async Groq client wrapper that generates persona-aligned responses.
  • Wire up the full FastAPI application with Pydantic schemas and three endpoints.
Create the prompt assembly module

The prompt assembly function is the glue between retrieval and generation. It takes persona fragments from ChromaDB, conversation turns from MongoDB, and the current user message, then formats them into the message list that the LLM expects.

  • Create a new file called app/llm_client.py inside the app/ folder.
  • Add the imports, client singleton, and prompt assembly function by pasting this code:
import os

from groq import AsyncGroq

from app.config import settings
from app.exceptions import LLMGenerationError
from app.logging_config import get_logger

logger = get_logger(__name__)

_groq_client: AsyncGroq | None = None


def get_groq_client() -> AsyncGroq:
    global _groq_client
    if _groq_client is None:
        _groq_client = AsyncGroq(api_key=settings.groq_api_key)
        logger.info("Groq async client initialized")
    return _groq_client


def assemble_prompt(
    persona_name: str,
    persona_context: list[str],
    conversation_history: list[dict],
    user_message: str,
) -> list[dict]:
    system_content = f"""You are {persona_name}. Stay completely in character at all times.
Use the following personality context to guide your tone, knowledge, and behavior:

---
{chr(10).join(persona_context)}
---

Rules:
- Never break character or acknowledge you are an AI
- Respond naturally as this persona would
- Draw on the personality context for tone, vocabulary, and perspective
- Keep responses conversational and engaging"""

    messages = [{"role": "system", "content": system_content}]

    for turn in conversation_history:
        messages.append({"role": "user", "content": turn["user_message"]})
        messages.append({"role": "assistant", "content": turn["assistant_message"]})

    messages.append({"role": "user", "content": user_message})
    return messages

What does this code do?

get_groq_client() creates a singleton AsyncGroq instance so the same connection is reused across requests. It reads the API key from settings.groq_api_key which loads from your .env file.

assemble_prompt() builds a structured message list. The system message injects persona context and behavioral rules. Conversation history is replayed as alternating user/assistant messages. The current user message goes last.

  • Save app/llm_client.py.

Seeing an import error for groq?

Make sure your virtual environment is active and you installed all packages from requirements.txt in the earlier step. The groq package must be installed for the import to resolve.

help me fix my groq import error

Add the LLM generation function

The generation function sends the assembled prompt to Groq's llama-3.3-70b-versatile model and returns the response text. It wraps the API call in error handling so failures surface as a clean LLMGenerationError.

  • In app/llm_client.py, add the generate_response() function below assemble_prompt():
async def generate_response(messages: list[dict]) -> str:
    try:
        client = get_groq_client()
        chat_completion = await client.chat.completions.create(
            messages=messages,
            model=settings.llm_model,
            temperature=0.7,
            max_tokens=1024,
        )
        response = chat_completion.choices[0].message.content
        logger.info("LLM response generated", extra={"model": settings.llm_model})
        return response
    except Exception as e:
        logger.error("LLM generation failed", extra={"error": str(e)})
        raise LLMGenerationError(f"Failed to generate response: {e}")

What does this function do?

generate_response() takes the full message list from assemble_prompt() and sends it to Groq. It uses temperature=0.7 for natural-sounding responses and max_tokens=1024 to keep replies conversational without running too long.

If anything goes wrong (network timeout, rate limit, invalid key), the function catches the exception and re-raises it as an LLMGenerationError. This feeds into the exception handler you built earlier, returning a clean 502 JSON error to the client.

  • Save app/llm_client.py.
  • You will not see a visible change yet. The next substep wires this function into the API where you can test it end-to-end.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/llm_client.py with both functions.

ⓧ I'd like to double check the full code

import os

from groq import AsyncGroq

from app.config import settings
from app.exceptions import LLMGenerationError
from app.logging_config import get_logger

logger = get_logger(__name__)

_groq_client: AsyncGroq | None = None


def get_groq_client() -> AsyncGroq:
    global _groq_client
    if _groq_client is None:
        _groq_client = AsyncGroq(api_key=settings.groq_api_key)
        logger.info("Groq async client initialized")
    return _groq_client


def assemble_prompt(
    persona_name: str,
    persona_context: list[str],
    conversation_history: list[dict],
    user_message: str,
) -> list[dict]:
    system_content = f"""You are {persona_name}. Stay completely in character at all times.
Use the following personality context to guide your tone, knowledge, and behavior:

---
{chr(10).join(persona_context)}
---

Rules:
- Never break character or acknowledge you are an AI
- Respond naturally as this persona would
- Draw on the personality context for tone, vocabulary, and perspective
- Keep responses conversational and engaging"""

    messages = [{"role": "system", "content": system_content}]

    for turn in conversation_history:
        messages.append({"role": "user", "content": turn["user_message"]})
        messages.append({"role": "assistant", "content": turn["assistant_message"]})

    messages.append({"role": "user", "content": user_message})
    return messages


async def generate_response(messages: list[dict]) -> str:
    try:
        client = get_groq_client()
        chat_completion = await client.chat.completions.create(
            messages=messages,
            model=settings.llm_model,
            temperature=0.7,
            max_tokens=1024,
        )
        response = chat_completion.choices[0].message.content
        logger.info("LLM response generated", extra={"model": settings.llm_model})
        return response
    except Exception as e:
        logger.error("LLM generation failed", extra={"error": str(e)})
        raise LLMGenerationError(f"Failed to generate response: {e}")
Create the schemas and wire up the FastAPI app

With all four layers complete (config, persona store, conversation store, LLM client), it is time to define the API contract and build the FastAPI application that orchestrates them.

  • Create a new file called app/schemas.py inside the app/ folder.
  • Define the Pydantic request and response models by pasting this code:
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime


class ChatRequest(BaseModel):
    session_id: str = Field(..., description="Unique conversation session identifier")
    persona_id: str = Field(..., description="ID of the persona to use")
    message: str = Field(..., min_length=1, description="User message")


class ChatResponse(BaseModel):
    session_id: str
    persona_id: str
    response: str
    correlation_id: str
    timestamp: datetime


class PersonaDocument(BaseModel):
    category: str = Field(..., description="Category: traits, background, style, knowledge")
    content: str = Field(..., min_length=1, description="Persona content fragment")


class PersonaIngestRequest(BaseModel):
    persona_id: str = Field(..., description="Unique persona identifier")
    persona_name: str = Field(..., description="Display name for the persona")
    documents: list[PersonaDocument] = Field(..., min_length=1)


class PersonaIngestResponse(BaseModel):
    persona_id: str
    documents_ingested: int
    message: str


class HealthResponse(BaseModel):
    status: str
    mongodb: str
    chromadb: str

What do these schemas define?

Pydantic models enforce type validation on every request and response automatically. FastAPI uses them to parse incoming JSON, reject malformed requests, and generate OpenAPI documentation.

ChatRequest requires a session ID (for history), a persona ID (for retrieval), and a non-empty message. PersonaIngestRequest accepts a batch of persona documents categorized by type. HealthResponse reports the status of both MongoDB and ChromaDB.

  • Save app/schemas.py.
  • Now create the main application file. Create app/main.py inside the app/ folder and paste the following code:
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.config import settings
from app.logging_config import setup_logging, get_logger, new_correlation_id
from app.exceptions import AppError, app_error_handler
from app.schemas import (
    ChatRequest,
    ChatResponse,
    PersonaIngestRequest,
    PersonaIngestResponse,
    HealthResponse,
)
from app.persona_store import ingest_persona, retrieve_persona_context, get_chroma_client
from app.conversation_store import add_turn, get_history, get_mongo_client
from app.llm_client import assemble_prompt, generate_response
from datetime import datetime, timezone


@asynccontextmanager
async def lifespan(app: FastAPI):
    setup_logging(settings.log_level)
    logger = get_logger(__name__)
    logger.info("Application starting", extra={"app_name": settings.app_name})
    yield
    logger.info("Application shutting down")


app = FastAPI(
    title=settings.app_name,
    version="1.0.0",
    lifespan=lifespan,
)
app.add_exception_handler(AppError, app_error_handler)

logger = get_logger(__name__)


@app.get("/health", response_model=HealthResponse)
async def health_check():
    mongo_status = "healthy"
    chroma_status = "healthy"

    try:
        client = get_mongo_client()
        await client.admin.command("ping")
    except Exception:
        mongo_status = "unhealthy"

    try:
        get_chroma_client().heartbeat()
    except Exception:
        chroma_status = "unhealthy"

    status = "healthy" if mongo_status == "healthy" and chroma_status == "healthy" else "degraded"
    return HealthResponse(status=status, mongodb=mongo_status, chromadb=chroma_status)

What does this set up?

The lifespan context manager runs setup code when the server starts and cleanup code when it shuts down. Here it initializes structured logging.

app.add_exception_handler(AppError, app_error_handler) registers the handler you built in app/exceptions.py. Any unhandled AppError (or subclass) automatically returns a consistent JSON error response.

The /health endpoint pings both MongoDB and ChromaDB. If either fails, it reports a degraded status so monitoring tools can alert you.

  • Now add the business endpoints. Below the health_check function in app/main.py, paste the following code:
@app.post("/persona/ingest", response_model=PersonaIngestResponse)
async def ingest_persona_endpoint(request: PersonaIngestRequest):
    correlation_id = new_correlation_id()
    logger.info(
        "Ingesting persona",
        extra={"persona_id": request.persona_id, "correlation_id": correlation_id},
    )
    documents = [{"category": doc.category, "content": doc.content} for doc in request.documents]
    count = ingest_persona(request.persona_id, documents)
    return PersonaIngestResponse(
        persona_id=request.persona_id,
        documents_ingested=count,
        message=f"Successfully ingested {count} documents for persona '{request.persona_name}'",
    )


@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    correlation_id = new_correlation_id()
    logger.info(
        "Chat request received",
        extra={
            "session_id": request.session_id,
            "persona_id": request.persona_id,
            "correlation_id": correlation_id,
        },
    )

    # Retrieve relevant persona fragments based on the user's message
    persona_context = retrieve_persona_context(
        persona_id=request.persona_id,
        query=request.message,
        top_k=settings.persona_top_k,
    )

    # Fetch recent conversation history for this session
    history = await get_history(
        session_id=request.session_id,
        max_turns=settings.max_history_turns,
    )

    # Assemble the full prompt with persona context + history + message
    messages = assemble_prompt(
        persona_name=request.persona_id,
        persona_context=persona_context,
        conversation_history=history,
        user_message=request.message,
    )

    # Generate the persona-aligned response via Groq
    response_text = await generate_response(messages)

    # Store this turn in conversation history for future context
    await add_turn(
        session_id=request.session_id,
        persona_id=request.persona_id,
        user_message=request.message,
        assistant_message=response_text,
    )

    return ChatResponse(
        session_id=request.session_id,
        persona_id=request.persona_id,
        response=response_text,
        correlation_id=correlation_id,
        timestamp=datetime.now(timezone.utc),
    )

How does the /chat pipeline work?

Each chat request triggers four sequential operations. First, retrieve_persona_context() queries ChromaDB for persona fragments relevant to the user's message. Second, get_history() fetches recent conversation turns from MongoDB.

Then assemble_prompt() combines everything into a structured message list. Finally, generate_response() sends it to Groq and stores the result back in MongoDB for future turns.

  • Save app/main.py.
  • Create one more file in the project root (not inside app/): run.py. Paste the following code:
import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

What does run.py do?

This is a convenience entry point that starts the uvicorn ASGI server pointing at your FastAPI app. The reload=True flag watches for file changes and auto-restarts the server during development.

  • Save run.py.
  • Start the server by running this command in your terminal:
python run.py
  • You should see JSON log output including a line that says Application starting and uvicorn reporting it is serving on http://0.0.0.0:8000.
  • Open a new terminal tab, then test the health endpoint by running:
curl http://localhost:8000/health
  • You should see a JSON response with "status": "healthy", "mongodb": "healthy", and "chromadb": "healthy".

Server not starting?

If you see a ModuleNotFoundError, make sure you are running the command from the project root (the folder containing run.py and the app/ directory). Also confirm your virtual environment is active.

If MongoDB reports "unhealthy", check that your MONGODB_URI in .env matches the connection string from your Atlas cluster and that your IP is in the Atlas network access list.

help me troubleshoot my persona agent server startup

✔️ Awesome, I've got everything!

Great. Make sure you have saved all four files: app/schemas.py, app/llm_client.py, app/main.py, and run.py.

ⓧ I'd like to double check the full code

import os

from groq import AsyncGroq

from app.config import settings
from app.exceptions import LLMGenerationError
from app.logging_config import get_logger

logger = get_logger(__name__)

_groq_client: AsyncGroq | None = None


def get_groq_client() -> AsyncGroq:
    global _groq_client
    if _groq_client is None:
        _groq_client = AsyncGroq(api_key=settings.groq_api_key)
        logger.info("Groq async client initialized")
    return _groq_client


def assemble_prompt(
    persona_name: str,
    persona_context: list[str],
    conversation_history: list[dict],
    user_message: str,
) -> list[dict]:
    system_content = f"""You are {persona_name}. Stay completely in character at all times.
Use the following personality context to guide your tone, knowledge, and behavior:

---
{chr(10).join(persona_context)}
---

Rules:
- Never break character or acknowledge you are an AI
- Respond naturally as this persona would
- Draw on the personality context for tone, vocabulary, and perspective
- Keep responses conversational and engaging"""

    messages = [{"role": "system", "content": system_content}]

    for turn in conversation_history:
        messages.append({"role": "user", "content": turn["user_message"]})
        messages.append({"role": "assistant", "content": turn["assistant_message"]})

    messages.append({"role": "user", "content": user_message})
    return messages


async def generate_response(messages: list[dict]) -> str:
    try:
        client = get_groq_client()
        chat_completion = await client.chat.completions.create(
            messages=messages,
            model=settings.llm_model,
            temperature=0.7,
            max_tokens=1024,
        )
        response = chat_completion.choices[0].message.content
        logger.info("LLM response generated", extra={"model": settings.llm_model})
        return response
    except Exception as e:
        logger.error("LLM generation failed", extra={"error": str(e)})
        raise LLMGenerationError(f"Failed to generate response: {e}")
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime


class ChatRequest(BaseModel):
    session_id: str = Field(..., description="Unique conversation session identifier")
    persona_id: str = Field(..., description="ID of the persona to use")
    message: str = Field(..., min_length=1, description="User message")


class ChatResponse(BaseModel):
    session_id: str
    persona_id: str
    response: str
    correlation_id: str
    timestamp: datetime


class PersonaDocument(BaseModel):
    category: str = Field(..., description="Category: traits, background, style, knowledge")
    content: str = Field(..., min_length=1, description="Persona content fragment")


class PersonaIngestRequest(BaseModel):
    persona_id: str = Field(..., description="Unique persona identifier")
    persona_name: str = Field(..., description="Display name for the persona")
    documents: list[PersonaDocument] = Field(..., min_length=1)


class PersonaIngestResponse(BaseModel):
    persona_id: str
    documents_ingested: int
    message: str


class HealthResponse(BaseModel):
    status: str
    mongodb: str
    chromadb: str
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.config import settings
from app.logging_config import setup_logging, get_logger, new_correlation_id
from app.exceptions import AppError, app_error_handler
from app.schemas import (
    ChatRequest,
    ChatResponse,
    PersonaIngestRequest,
    PersonaIngestResponse,
    HealthResponse,
)
from app.persona_store import ingest_persona, retrieve_persona_context, get_chroma_client
from app.conversation_store import add_turn, get_history, get_mongo_client
from app.llm_client import assemble_prompt, generate_response
from datetime import datetime, timezone


@asynccontextmanager
async def lifespan(app: FastAPI):
    setup_logging(settings.log_level)
    logger = get_logger(__name__)
    logger.info("Application starting", extra={"app_name": settings.app_name})
    yield
    logger.info("Application shutting down")


app = FastAPI(
    title=settings.app_name,
    version="1.0.0",
    lifespan=lifespan,
)
app.add_exception_handler(AppError, app_error_handler)

logger = get_logger(__name__)


@app.get("/health", response_model=HealthResponse)
async def health_check():
    mongo_status = "healthy"
    chroma_status = "healthy"

    try:
        client = get_mongo_client()
        await client.admin.command("ping")
    except Exception:
        mongo_status = "unhealthy"

    try:
        get_chroma_client().heartbeat()
    except Exception:
        chroma_status = "unhealthy"

    status = "healthy" if mongo_status == "healthy" and chroma_status == "healthy" else "degraded"
    return HealthResponse(status=status, mongodb=mongo_status, chromadb=chroma_status)


@app.post("/persona/ingest", response_model=PersonaIngestResponse)
async def ingest_persona_endpoint(request: PersonaIngestRequest):
    correlation_id = new_correlation_id()
    logger.info(
        "Ingesting persona",
        extra={"persona_id": request.persona_id, "correlation_id": correlation_id},
    )
    documents = [{"category": doc.category, "content": doc.content} for doc in request.documents]
    count = ingest_persona(request.persona_id, documents)
    return PersonaIngestResponse(
        persona_id=request.persona_id,
        documents_ingested=count,
        message=f"Successfully ingested {count} documents for persona '{request.persona_name}'",
    )


@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    correlation_id = new_correlation_id()
    logger.info(
        "Chat request received",
        extra={
            "session_id": request.session_id,
            "persona_id": request.persona_id,
            "correlation_id": correlation_id,
        },
    )

    persona_context = retrieve_persona_context(
        persona_id=request.persona_id,
        query=request.message,
        top_k=settings.persona_top_k,
    )

    history = await get_history(
        session_id=request.session_id,
        max_turns=settings.max_history_turns,
    )

    messages = assemble_prompt(
        persona_name=request.persona_id,
        persona_context=persona_context,
        conversation_history=history,
        user_message=request.message,
    )

    response_text = await generate_response(messages)

    await add_turn(
        session_id=request.session_id,
        persona_id=request.persona_id,
        user_message=request.message,
        assistant_message=response_text,
    )

    return ChatResponse(
        session_id=request.session_id,
        persona_id=request.persona_id,
        response=response_text,
        correlation_id=correlation_id,
        timestamp=datetime.now(timezone.utc),
    )
import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

Your persona agent API is live. All three layers (retrieval, memory, generation) are wired into a single orchestrated pipeline. Next up, you will test the full conversation flow by ingesting a persona and chatting with it across multiple turns.

Secret mission

Hot-Swap Personas Mid-Conversation

What happens when you want a different persona to take over the conversation without losing the chat history? In this secret mission, you'll build a /persona/switch endpoint that hot-swaps the active persona mid-conversation while preserving full conversation continuity.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running or delete them entirely. This project uses the MongoDB Atlas Free tier and local storage, so there are no ongoing costs.

Resources you used:

  • MongoDB Atlas Free cluster (persona_agent database with conversations collection).
  • Local ChromaDB data stored in the chroma_data/ folder.
  • Local project folder with Python virtual environment and all installed packages.

Keep everything running

No action needed. Choose this if you plan to keep building on this project or experiment further with new personas.

  • Your MongoDB Atlas Free cluster auto-pauses after 30 days of zero connections. There is no cost to leave it running.
  • Your local chroma_data/ folder stays on disk until you delete it manually.
  • Your virtual environment and project files remain on your Desktop.

Pause - I'll come back to this later

Free up local resources while preserving your cloud data for later. Your Atlas cluster will auto-pause after 30 days of inactivity at no cost.

  • Stop the FastAPI server if it is still running by pressing Ctrl+C in the terminal.
  • Deactivate your virtual environment by running:
deactivate
  • Your MongoDB Atlas Free cluster will auto-pause after 30 days without connections. No manual action is required.

Delete - I don't want to use this again

Remove all project resources and start fresh. This permanently deletes your conversation history, persona embeddings, and project code.

Delete your MongoDB Atlas cluster:

  • Open the MongoDB Atlas console at cloud.mongodb.com.
  • In the sidebar, click Clusters under the Database heading.
  • Click the ... (ellipsis) next to your cluster.
  • Click Terminate from the drop-down menu.
  • Enter the cluster name to confirm, then click Terminate.

Delete local ChromaDB data:

rm -rf chroma_data/

Delete the entire project folder (macOS/Linux):

cd ~/Desktop
rm -rf persona-agent

On Windows, run:

cd ~\Desktop
rmdir /s /q persona-agent

Nice Work!

Nice Work!

You just built a production-grade persona agent API powered by Retrieval-Augmented Generation. Your service retrieves personality context from a vector store, tracks multi-turn conversations, and generates personality-aligned responses on every call.

You've learned how to:

  • Implement RAG-based persona retrieval using ChromaDB and Sentence Transformers to give an LLM consistent personality context.
  • Create async conversation memory with MongoDB Atlas and PyMongo's native AsyncMongoClient, maintaining multi-turn state across requests.
  • Assemble multi-source prompts combining persona context, conversation history, and user input into a structured call to Groq's Llama 3.3 70B model.
  • Secret Mission: Implemented hot-swappable personas via a /persona/switch endpoint that preserves conversation continuity while changing personality mid-session.

Ready to quiz yourself?