Implementing Conversational State Machines: Managing Multi-Turn Agent Workflows with Explicit State Transitions and Persistence

Introduction

Picture a customer support agent you're building for a software company. A user opens a ticket, explains they can't log in, gets asked for their username, provides it, gets asked whether they've tried a password reset, says yes, gets asked to describe the error message... and then their browser crashes. When they come back ten minutes later, your agent cheerfully greets them as if they've never spoken before.

This is the failure mode that conversational state machines solve. Without explicit state management, your multi-turn agent is just a glorified chatbot — it responds to the immediate input, has no reliable memory of what workflow step it's on, and can't recover from interruptions. Users get frustrated, agents give inconsistent guidance, and you lose all the hard-won context that made the conversation valuable. As you build more ambitious LLM applications, you'll hit this wall quickly: raw message history alone is not enough to orchestrate complex workflows.

By the end of this lesson, you'll be able to design explicit finite state machines for conversational agents, implement them in Python with proper persistence, handle transitions and guards, and recover gracefully from mid-conversation interruptions. We'll build a realistic loan pre-qualification agent that demonstrates every concept in a production-ready way.

What you'll learn:

• How to model multi-turn workflows as formal state machines with defined states, transitions, and guards
• How to separate conversation state from message history, and why this distinction matters
• How to implement durable state persistence using Redis and PostgreSQL so conversations survive restarts and crashes
• How to handle invalid transitions, timeout states, and user-initiated backtracking
• How to integrate state machine logic with LLM calls without leaking state management concerns into your prompts

---

Prerequisites

You should be comfortable making LLM API calls from Python (see Using the OpenAI API with Python if you need a refresher), and you should understand how the chat completions API structures message threads (covered in Conversation Design for LLMs: Structuring System Prompts, Roles, and Message Threads with the Chat Completions API). Basic familiarity with Redis and PostgreSQL is helpful but not required — we'll explain the persistence layer as we build it.

---

Why Message History Alone Falls Short

Before diving into the implementation, it's worth being precise about the problem. When you build a multi-turn agent by simply appending messages to a list and sending them to the LLM each turn, you're relying on the model to:

1. Infer what step of the workflow it's on from reading the conversation
2. Decide what to ask for next based on what's been discussed
3. Maintain internal consistency about what data has been collected
4. Handle edge cases like the user going backwards or skipping steps

This works surprisingly well for short, simple conversations. It falls apart at scale because the LLM's "understanding" of workflow state is implicit and probabilistic. The model might hallucinate that it already collected information it didn't, or it might ask for the same information twice, or it might accept a vague answer and move on when it should push for specifics. You also can't reliably resume a crashed conversation, audit what step a user was on, or enforce business rules like "don't proceed to step 4 unless step 3 returned a valid result."

The solution is to make state explicit in your application layer, separate from the message content. Your state machine tracks where you are in the workflow. The LLM is responsible only for natural language — extracting information from user messages and generating appropriate responses. This is a classic separation of concerns, and it's what separates a production-quality agent from a prototype.

---

Modeling the State Machine

A finite state machine has four components:

• States: discrete positions in a workflow (e.g., COLLECTING_INCOME, AWAITING_CONFIRMATION)
• Transitions: rules for moving from one state to another
• Guards: conditions that must be true for a transition to occur
• Actions: side effects triggered by entering or leaving a state (e.g., calling an API, logging)

Let's model our loan pre-qualification agent. The workflow goes like this: greet the user, collect personal info (name, annual income, employment status), collect loan details (amount requested, loan purpose), run an eligibility check, present results, and either close or loop back for adjustments.

from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Optional, Dict, Any, Callable
import time

class ConversationState(Enum):
    GREETING = "greeting"
    COLLECTING_PERSONAL_INFO = "collecting_personal_info"
    COLLECTING_LOAN_DETAILS = "collecting_loan_details"
    RUNNING_ELIGIBILITY_CHECK = "running_eligibility_check"
    PRESENTING_RESULTS = "presenting_results"
    HANDLING_ADJUSTMENT = "handling_adjustment"
    COMPLETED = "completed"
    ABANDONED = "abandoned"
    ERROR = "error"

@dataclass
class ConversationContext:
    session_id: str
    state: ConversationState = ConversationState.GREETING
    collected_data: Dict[str, Any] = field(default_factory=dict)
    eligibility_result: Optional[Dict[str, Any]] = None
    turn_count: int = 0
    created_at: float = field(default_factory=time.time)
    last_updated: float = field(default_factory=time.time)
    error_count: int = 0
    
    def update(self, **kwargs):
        for key, value in kwargs.items():
            if hasattr(self, key):
                setattr(self, key, value)
        self.last_updated = time.time()
        self.turn_count += 1

Notice that ConversationContext is separate from your message history. The collected_data dict accumulates structured facts. The state field tells you exactly where you are. The turn_count and last_updated fields enable timeout logic. This context object is what you'll persist to a database — message history goes somewhere else (or gets reconstructed from it).

Now let's define the valid transitions:

from typing import NamedTuple, List

class Transition(NamedTuple):
    from_state: ConversationState
    to_state: ConversationState
    guard: Optional[Callable] = None  # Returns True if transition is allowed
    action: Optional[Callable] = None  # Side effect to run on transition

def has_personal_info(ctx: ConversationContext) -> bool:
    required = {"full_name", "annual_income_usd", "employment_status"}
    return required.issubset(ctx.collected_data.keys())

def has_loan_details(ctx: ConversationContext) -> bool:
    required = {"loan_amount_usd", "loan_purpose"}
    return required.issubset(ctx.collected_data.keys())

def eligibility_check_done(ctx: ConversationContext) -> bool:
    return ctx.eligibility_result is not None

VALID_TRANSITIONS = [
    Transition(ConversationState.GREETING, ConversationState.COLLECTING_PERSONAL_INFO),
    Transition(
        ConversationState.COLLECTING_PERSONAL_INFO,
        ConversationState.COLLECTING_LOAN_DETAILS,
        guard=has_personal_info
    ),
    Transition(
        ConversationState.COLLECTING_LOAN_DETAILS,
        ConversationState.RUNNING_ELIGIBILITY_CHECK,
        guard=has_loan_details
    ),
    Transition(
        ConversationState.RUNNING_ELIGIBILITY_CHECK,
        ConversationState.PRESENTING_RESULTS,
        guard=eligibility_check_done
    ),
    Transition(ConversationState.PRESENTING_RESULTS, ConversationState.COMPLETED),
    Transition(ConversationState.PRESENTING_RESULTS, ConversationState.HANDLING_ADJUSTMENT),
    Transition(ConversationState.HANDLING_ADJUSTMENT, ConversationState.COLLECTING_LOAN_DETAILS),
    # Any state can transition to ABANDONED or ERROR
    *[Transition(s, ConversationState.ABANDONED) for s in ConversationState if s not in (ConversationState.COMPLETED, ConversationState.ABANDONED, ConversationState.ERROR)],
    *[Transition(s, ConversationState.ERROR) for s in ConversationState if s != ConversationState.ERROR],
]

---

Building the State Machine Engine

Now let's wrap these transitions in a proper engine that handles validation and execution:

class StateMachineError(Exception):
    pass

class InvalidTransitionError(StateMachineError):
    def __init__(self, from_state, to_state):
        super().__init__(f"Cannot transition from {from_state.value} to {to_state.value}")
        self.from_state = from_state
        self.to_state = to_state

class GuardFailedError(StateMachineError):
    def __init__(self, from_state, to_state, reason=""):
        msg = f"Guard prevented transition from {from_state.value} to {to_state.value}"
        if reason:
            msg += f": {reason}"
        super().__init__(msg)

class ConversationStateMachine:
    def __init__(self, transitions: List[Transition]):
        # Build a lookup: (from_state, to_state) -> Transition
        self._transition_map: Dict[tuple, Transition] = {}
        for t in transitions:
            self._transition_map[(t.from_state, t.to_state)] = t
        
        # Also build a quick lookup of valid "from" states
        self._valid_from: Dict[ConversationState, List[ConversationState]] = {}
        for (from_s, to_s) in self._transition_map:
            self._valid_from.setdefault(from_s, []).append(to_s)
    
    def transition(
        self,
        ctx: ConversationContext,
        target_state: ConversationState
    ) -> ConversationContext:
        """
        Attempt to move ctx to target_state.
        Returns updated context if successful.
        Raises InvalidTransitionError or GuardFailedError if not.
        """
        key = (ctx.state, target_state)
        
        if key not in self._transition_map:
            raise InvalidTransitionError(ctx.state, target_state)
        
        transition = self._transition_map[key]
        
        if transition.guard is not None:
            if not transition.guard(ctx):
                raise GuardFailedError(
                    ctx.state, target_state,
                    f"guard '{transition.guard.__name__}' returned False"
                )
        
        # Run action if defined
        if transition.action is not None:
            transition.action(ctx)
        
        # Update state
        ctx.state = target_state
        ctx.last_updated = time.time()
        
        return ctx
    
    def can_transition(
        self,
        ctx: ConversationContext,
        target_state: ConversationState
    ) -> bool:
        """Non-raising version for conditional logic."""
        key = (ctx.state, target_state)
        if key not in self._transition_map:
            return False
        t = self._transition_map[key]
        if t.guard is not None:
            return t.guard(ctx)
        return True
    
    def available_transitions(
        self, ctx: ConversationContext
    ) -> List[ConversationState]:
        """Returns all states reachable from current state."""
        candidates = self._valid_from.get(ctx.state, [])
        return [
            s for s in candidates
            if self.can_transition(ctx, s)
        ]

# Instantiate the engine
state_machine = ConversationStateMachine(VALID_TRANSITIONS)

This engine gives you clean separation: the machine knows nothing about LLMs or HTTP — it only knows about states, transitions, and guards. This is easy to unit test in isolation, which matters for testing your LLM application properly.

---

Persisting State: Redis for Sessions, PostgreSQL for History

Session state needs two properties: fast access (because you read it on every turn) and durability (because users might come back hours later). The standard pattern is to use Redis for the active session window with a configurable TTL, and PostgreSQL for permanent storage of completed or important conversations.

import json
import redis
import psycopg2
from psycopg2.extras import Json
from dataclasses import asdict

class StateStore:
    """
    Hybrid persistence: Redis for hot sessions, PostgreSQL for durable archive.
    """
    
    SESSION_TTL_SECONDS = 3600  # 1 hour of inactivity before abandonment
    
    def __init__(self, redis_url: str, postgres_dsn: str):
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.pg_conn = psycopg2.connect(postgres_dsn)
        self._ensure_schema()
    
    def _ensure_schema(self):
        with self.pg_conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS conversation_sessions (
                    session_id TEXT PRIMARY KEY,
                    state TEXT NOT NULL,
                    collected_data JSONB NOT NULL DEFAULT '{}',
                    eligibility_result JSONB,
                    turn_count INTEGER DEFAULT 0,
                    error_count INTEGER DEFAULT 0,
                    created_at DOUBLE PRECISION NOT NULL,
                    last_updated DOUBLE PRECISION NOT NULL
                );
                
                CREATE TABLE IF NOT EXISTS conversation_events (
                    id SERIAL PRIMARY KEY,
                    session_id TEXT NOT NULL REFERENCES conversation_sessions(session_id),
                    event_type TEXT NOT NULL,  -- 'state_transition', 'user_message', 'agent_message'
                    payload JSONB NOT NULL,
                    occurred_at DOUBLE PRECISION NOT NULL
                );
            """)
            self.pg_conn.commit()
    
    def _redis_key(self, session_id: str) -> str:
        return f"conversation:session:{session_id}"
    
    def save(self, ctx: ConversationContext):
        """Write to Redis (fast) and PostgreSQL (durable)."""
        data = asdict(ctx)
        data["state"] = ctx.state.value  # Enum -> string for JSON
        
        # Redis: serialize and set with TTL
        self.redis.setex(
            self._redis_key(ctx.session_id),
            self.SESSION_TTL_SECONDS,
            json.dumps(data)
        )
        
        # PostgreSQL: upsert
        with self.pg_conn.cursor() as cur:
            cur.execute("""
                INSERT INTO conversation_sessions
                    (session_id, state, collected_data, eligibility_result,
                     turn_count, error_count, created_at, last_updated)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (session_id) DO UPDATE SET
                    state = EXCLUDED.state,
                    collected_data = EXCLUDED.collected_data,
                    eligibility_result = EXCLUDED.eligibility_result,
                    turn_count = EXCLUDED.turn_count,
                    error_count = EXCLUDED.error_count,
                    last_updated = EXCLUDED.last_updated
            """, (
                ctx.session_id,
                ctx.state.value,
                Json(ctx.collected_data),
                Json(ctx.eligibility_result) if ctx.eligibility_result else None,
                ctx.turn_count,
                ctx.error_count,
                ctx.created_at,
                ctx.last_updated,
            ))
            self.pg_conn.commit()
    
    def load(self, session_id: str) -> Optional[ConversationContext]:
        """Try Redis first, fall back to PostgreSQL."""
        redis_data = self.redis.get(self._redis_key(session_id))
        
        if redis_data:
            data = json.loads(redis_data)
        else:
            # Cache miss: load from PostgreSQL and re-warm Redis
            with self.pg_conn.cursor() as cur:
                cur.execute(
                    "SELECT * FROM conversation_sessions WHERE session_id = %s",
                    (session_id,)
                )
                row = cur.fetchone()
                if not row:
                    return None
                cols = [desc[0] for desc in cur.description]
                data = dict(zip(cols, row))
        
        # Reconstruct the dataclass
        data["state"] = ConversationState(data["state"])
        if isinstance(data.get("collected_data"), str):
            data["collected_data"] = json.loads(data["collected_data"])
        
        ctx = ConversationContext(
            session_id=data["session_id"],
            state=data["state"],
            collected_data=data.get("collected_data", {}),
            eligibility_result=data.get("eligibility_result"),
            turn_count=data.get("turn_count", 0),
            created_at=data.get("created_at", time.time()),
            last_updated=data.get("last_updated", time.time()),
            error_count=data.get("error_count", 0),
        )
        
        # Re-warm Redis if we loaded from PG
        if not redis_data:
            self.save(ctx)
        
        return ctx
    
    def log_event(self, session_id: str, event_type: str, payload: dict):
        """Append an audit event — useful for debugging and analytics."""
        with self.pg_conn.cursor() as cur:
            cur.execute(
                """INSERT INTO conversation_events 
                   (session_id, event_type, payload, occurred_at)
                   VALUES (%s, %s, %s, %s)""",
                (session_id, event_type, Json(payload), time.time())
            )
            self.pg_conn.commit()

This persistence layer pairs well with the observability patterns described in Implementing LLM Observability: Tracing, Logging, and Monitoring Requests in Production — your conversation_events table becomes a natural audit log.

---

Connecting the LLM: Extraction and Response Generation

Now we connect the state machine to actual LLM calls. The key design decision is that the LLM serves two distinct roles:

1. Extraction: parse structured data from user messages (what did they say their income was?)
2. Generation: produce the next conversational response given the current state

These are separate calls with separate prompts. Mixing them makes both tasks worse. For extraction, we want structured output with forced JSON. For generation, we want natural, context-aware responses.

from openai import OpenAI
import json

client = OpenAI()

EXTRACTION_PROMPTS = {
    ConversationState.COLLECTING_PERSONAL_INFO: """
You are extracting structured data from a user message in a loan pre-qualification conversation.
Extract any of the following fields if present in the message:
- full_name (string)
- annual_income_usd (number, in USD — convert if user says e.g. "$85k" -> 85000)
- employment_status (one of: "employed", "self_employed", "unemployed", "retired")

Return ONLY a JSON object with the fields you found. If a field is missing or unclear, omit it.
Do not include null values. Do not guess.

User message: {user_message}
Already collected: {already_collected}
""",
    ConversationState.COLLECTING_LOAN_DETAILS: """
Extract loan request information from this user message:
- loan_amount_usd (number — convert "50k" -> 50000, "$200,000" -> 200000)
- loan_purpose (one of: "home_purchase", "home_refinance", "auto", "personal", "business", "debt_consolidation")

Return ONLY a JSON object with found fields. Omit missing or unclear fields.

User message: {user_message}
Already collected: {already_collected}
""",
}

def extract_data_from_message(
    state: ConversationState,
    user_message: str,
    already_collected: dict
) -> dict:
    """Use LLM to extract structured fields from a user message."""
    prompt_template = EXTRACTION_PROMPTS.get(state)
    if not prompt_template:
        return {}
    
    prompt = prompt_template.format(
        user_message=user_message,
        already_collected=json.dumps(already_collected)
    )
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Use a fast, cheap model for extraction
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0,  # Deterministic extraction
    )
    
    try:
        extracted = json.loads(response.choices[0].message.content)
        # Validate types — don't trust the model blindly
        if "annual_income_usd" in extracted:
            extracted["annual_income_usd"] = float(extracted["annual_income_usd"])
        if "loan_amount_usd" in extracted:
            extracted["loan_amount_usd"] = float(extracted["loan_amount_usd"])
        return extracted
    except (json.JSONDecodeError, ValueError, TypeError):
        return {}


SYSTEM_PROMPTS = {
    ConversationState.GREETING: """
You are a friendly loan pre-qualification assistant for Meridian Financial.
Greet the user warmly and explain you'll need a few minutes to collect some information
to check their loan eligibility. Keep it brief and inviting.
""",
    ConversationState.COLLECTING_PERSONAL_INFO: """
You are a loan pre-qualification assistant. You need to collect:
- Full name (if not yet collected)
- Annual income in USD (if not yet collected)  
- Employment status: employed, self-employed, unemployed, or retired (if not yet collected)

Already collected: {collected}
Still needed: {needed}

Ask naturally for the missing information. Ask for one or two things at a time, not all at once.
If the user gave you some information in their last message, acknowledge it before asking for more.
""",
    ConversationState.COLLECTING_LOAN_DETAILS: """
You are a loan pre-qualification assistant. The applicant has provided their personal info.
Now collect:
- Desired loan amount in USD (if not yet collected)
- Loan purpose (if not yet collected): home purchase, refinancing, auto, personal, business, or debt consolidation

Already collected: {collected}
Still needed: {needed}

Be conversational. If they gave partial info, acknowledge it and ask for what's missing.
""",
    ConversationState.RUNNING_ELIGIBILITY_CHECK: """
Tell the user you're running their eligibility check and it will take just a moment.
Be reassuring and warm. Keep this very short.
""",
    ConversationState.PRESENTING_RESULTS: """
You are presenting loan pre-qualification results to the applicant.
Results: {eligibility_result}
Applicant data: {collected}

If pre-qualified: Congratulate them warmly, state the maximum amount and estimated rate range,
and ask if they'd like to proceed with a full application or adjust any parameters.

If not pre-qualified: Be empathetic and constructive. Explain the reason briefly,
suggest what might improve their situation, and ask if they'd like to try different loan parameters.
""",
}

def generate_response(
    state: ConversationState,
    ctx: ConversationContext,
    message_history: list,
    user_message: str
) -> str:
    """Generate a natural language response for the current state."""
    prompt_template = SYSTEM_PROMPTS.get(state, "You are a helpful assistant.")
    
    # Compute what's still needed for collection states
    needed_personal = {"full_name", "annual_income_usd", "employment_status"} - set(ctx.collected_data.keys())
    needed_loan = {"loan_amount_usd", "loan_purpose"} - set(ctx.collected_data.keys())
    
    system_prompt = prompt_template.format(
        collected=json.dumps(ctx.collected_data, indent=2),
        needed=list(needed_personal | needed_loan),
        eligibility_result=json.dumps(ctx.eligibility_result or {}, indent=2),
    )
    
    messages = [{"role": "system", "content": system_prompt}]
    messages.extend(message_history[-10:])  # Last 10 turns for context
    if user_message:
        messages.append({"role": "user", "content": user_message})
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        temperature=0.7,
    )
    
    return response.choices[0].message.content

Notice we're using gpt-4o-mini for extraction (cheap, fast, structured) and gpt-4o for generation (higher quality responses). This two-model approach is a practical optimization you'll appreciate at scale — see Cost Optimization: Token Counting, Caching, and Model Selection for LLMs for a deeper treatment of this pattern.

---

The Main Conversation Loop

Now let's wire everything together into a ConversationAgent class that handles a complete turn:

import uuid

def run_eligibility_check(ctx: ConversationContext) -> dict:
    """
    Simulate a credit/eligibility check. In production, this calls your
    underwriting API or credit bureau integration.
    """
    income = ctx.collected_data.get("annual_income_usd", 0)
    loan_amount = ctx.collected_data.get("loan_amount_usd", 0)
    employment = ctx.collected_data.get("employment_status", "")
    
    debt_to_income_ratio = loan_amount / max(income, 1)
    
    if employment == "unemployed":
        return {"approved": False, "reason": "Current employment status", "max_amount": 0}
    
    if debt_to_income_ratio > 5.0:
        return {
            "approved": False,
            "reason": "Requested amount exceeds income-based eligibility",
            "max_amount": round(income * 4.5, -3),
        }
    
    base_rate = 6.5 if employment == "employed" else 7.2
    return {
        "approved": True,
        "max_amount": min(loan_amount, round(income * 4.5, -3)),
        "rate_range_low": base_rate,
        "rate_range_high": base_rate + 1.5,
        "term_options_years": [15, 20, 30],
    }


class ConversationAgent:
    
    def __init__(self, state_store: StateStore):
        self.store = state_store
        self.machine = state_machine
    
    def start_session(self) -> tuple[str, str]:
        """Create a new session and return (session_id, initial_greeting)."""
        session_id = str(uuid.uuid4())
        ctx = ConversationContext(session_id=session_id)
        
        # Transition to collecting state immediately after greeting
        greeting = generate_response(
            ConversationState.GREETING, ctx, [], ""
        )
        
        ctx = self.machine.transition(ctx, ConversationState.COLLECTING_PERSONAL_INFO)
        self.store.save(ctx)
        self.store.log_event(session_id, "session_started", {"initial_state": ctx.state.value})
        
        return session_id, greeting
    
    def handle_turn(self, session_id: str, user_message: str) -> str:
        """
        Process one user turn. Returns the agent's response.
        This is the main entry point for each message.
        """
        ctx = self.store.load(session_id)
        if ctx is None:
            raise ValueError(f"Session {session_id} not found")
        
        # Check for session timeout
        if time.time() - ctx.last_updated > StateStore.SESSION_TTL_SECONDS:
            ctx = self.machine.transition(ctx, ConversationState.ABANDONED)
            self.store.save(ctx)
            return "It looks like some time has passed since we last spoke. For your security, I've ended this session. Please start a new conversation when you're ready."
        
        # Handle explicit user commands
        if user_message.lower().strip() in ("quit", "exit", "cancel", "stop"):
            ctx = self.machine.transition(ctx, ConversationState.ABANDONED)
            self.store.save(ctx)
            return "No problem! I've cancelled your pre-qualification session. Feel free to come back anytime."
        
        # Build message history for context (simplified — in production, load from DB)
        message_history = []  # Would be loaded from conversation_events table
        
        # Run extraction if we're in a data-collection state
        newly_extracted = {}
        if ctx.state in (ConversationState.COLLECTING_PERSONAL_INFO, ConversationState.COLLECTING_LOAN_DETAILS):
            newly_extracted = extract_data_from_message(ctx.state, user_message, ctx.collected_data)
            if newly_extracted:
                ctx.collected_data.update(newly_extracted)
                self.store.log_event(session_id, "data_extracted", newly_extracted)
        
        # Attempt automatic state advancement
        ctx = self._advance_state_if_ready(ctx)
        
        # If we just moved into eligibility check, run it synchronously
        if ctx.state == ConversationState.RUNNING_ELIGIBILITY_CHECK:
            ctx.eligibility_result = run_eligibility_check(ctx)
            self.store.log_event(session_id, "eligibility_checked", ctx.eligibility_result)
            ctx = self.machine.transition(ctx, ConversationState.PRESENTING_RESULTS)
        
        # Generate response for current state
        response = generate_response(ctx.state, ctx, message_history, user_message)
        
        # Handle PRESENTING_RESULTS — check if user wants to adjust or complete
        if ctx.state == ConversationState.PRESENTING_RESULTS:
            ctx = self._handle_results_intent(ctx, user_message)
        
        # Persist updated context
        ctx.turn_count += 1
        ctx.last_updated = time.time()
        self.store.save(ctx)
        self.store.log_event(session_id, "agent_response", {
            "state": ctx.state.value,
            "response_preview": response[:100]
        })
        
        return response
    
    def _advance_state_if_ready(self, ctx: ConversationContext) -> ConversationContext:
        """Check if guards pass for natural next state and advance if so."""
        next_states = {
            ConversationState.COLLECTING_PERSONAL_INFO: ConversationState.COLLECTING_LOAN_DETAILS,
            ConversationState.COLLECTING_LOAN_DETAILS: ConversationState.RUNNING_ELIGIBILITY_CHECK,
        }
        
        target = next_states.get(ctx.state)
        if target and self.machine.can_transition(ctx, target):
            ctx = self.machine.transition(ctx, target)
        
        return ctx
    
    def _handle_results_intent(self, ctx: ConversationContext, user_message: str) -> ConversationContext:
        """
        Simple intent detection for the results state.
        In production, use an LLM classifier here.
        """
        msg_lower = user_message.lower()
        completion_signals = ["proceed", "apply", "yes", "great", "perfect", "let's do it"]
        adjustment_signals = ["adjust", "change", "different", "lower", "try", "what if"]
        
        if any(sig in msg_lower for sig in completion_signals):
            if self.machine.can_transition(ctx, ConversationState.COMPLETED):
                ctx = self.machine.transition(ctx, ConversationState.COMPLETED)
        elif any(sig in msg_lower for sig in adjustment_signals):
            if self.machine.can_transition(ctx, ConversationState.HANDLING_ADJUSTMENT):
                ctx = self.machine.transition(ctx, ConversationState.HANDLING_ADJUSTMENT)
                # Clear loan details so user can re-enter
                ctx.collected_data.pop("loan_amount_usd", None)
                ctx.collected_data.pop("loan_purpose", None)
                ctx = self.machine.transition(ctx, ConversationState.COLLECTING_LOAN_DETAILS)
        
        return ctx

---

Hands-On Exercise

Build a multi-turn technical support triage agent using the patterns in this lesson. The workflow should have these states:

1. GREETING → IDENTIFYING_ISSUE (collect: product name, issue category)
2. IDENTIFYING_ISSUE → GATHERING_DETAILS (collect: error message or symptom, steps already tried)
3. GATHERING_DETAILS → DIAGNOSING (run a simulated lookup in a solutions database)
4. DIAGNOSING → PRESENTING_SOLUTION or ESCALATING_TO_HUMAN
5. PRESENTING_SOLUTION → RESOLVED or NEEDS_MORE_HELP
6. ESCALATING_TO_HUMAN → terminal state with a ticket number

Step 1: Define your ConversationState enum and ConversationContext dataclass for this workflow.

Step 2: Write guards for each transition. The guard from IDENTIFYING_ISSUE to GATHERING_DETAILS should check that both product_name and issue_category are present.

Step 3: Write the extraction prompt for IDENTIFYING_ISSUE. It should extract product_name (free text) and issue_category (one of: "login", "performance", "data_loss", "billing", "feature_request", "other").

Step 4: Implement a mock run_diagnosis(ctx) function that returns either {"has_solution": True, "solution": "..."} or {"has_solution": False, "escalation_reason": "..."} based on the issue category.

Step 5: Wire it into a ConversationAgent class and test it with at least three full conversations: one that resolves, one that escalates, and one that the user abandons mid-way.

Bonus challenge: Add a TIMEOUT guard that automatically transitions any session inactive for more than 30 minutes to a special TIMED_OUT state, with a re-engagement message asking if they'd like to continue or start fresh.

---

Common Mistakes & Troubleshooting

Mistake 1: Storing state in the message history

This is the most common anti-pattern. Developers add agent messages like "I've collected your name and income, now I need your loan amount" and then try to parse the next step from those messages. This is fragile — the model might paraphrase itself, users might push back and confuse the flow, and you can't reliably reconstruct state from prose.

Fix: Never let state live inside message content. Always maintain the explicit ctx.state field and treat message history as display-only.

Mistake 2: Not validating extracted data before storing

The extraction LLM might return "annual_income_usd": "seventy thousand" (a string instead of a number) or "employment_status": "contract" (not in your enum). If you store this unchecked, your guards will silently fail.

Fix: Add a validation step between extraction and storage. Use Pydantic models or simple type-checking functions:

from pydantic import BaseModel, validator
from typing import Literal, Optional

class PersonalInfoExtraction(BaseModel):
    full_name: Optional[str] = None
    annual_income_usd: Optional[float] = None
    employment_status: Optional[Literal["employed", "self_employed", "unemployed", "retired"]] = None
    
    @validator("annual_income_usd")
    def income_must_be_positive(cls, v):
        if v is not None and v <= 0:
            return None  # Discard invalid values rather than crashing
        return v

# In extract_data_from_message, parse through Pydantic:
raw = json.loads(response.choices[0].message.content)
validated = PersonalInfoExtraction(**raw)
return validated.dict(exclude_none=True)

Mistake 3: Race conditions on concurrent requests

If a user sends two messages rapidly (or your frontend has a retry mechanism), you might process two turns simultaneously against the same session, causing a state corruption. Redis handles this, but you need to use it correctly.

Fix: Use a Redis lock around the load-process-save cycle:

import redis.lock

def handle_turn_with_lock(self, session_id: str, user_message: str) -> str:
    lock = self.store.redis.lock(f"lock:session:{session_id}", timeout=30)
    with lock:
        return self.handle_turn(session_id, user_message)

Mistake 4: Forgetting to handle LLM failures mid-transition

If your eligibility check triggers an LLM call that fails, you're left in RUNNING_ELIGIBILITY_CHECK with no result. The next user message will try to advance from that state and fail silently.

Fix: Increment ctx.error_count on failures, and define a maximum before transitioning to ERROR. See Handling LLM Errors, Timeouts, and Rate Limits in Python Applications for the full retry pattern.

try:
    ctx.eligibility_result = run_eligibility_check(ctx)
    ctx = self.machine.transition(ctx, ConversationState.PRESENTING_RESULTS)
except Exception as e:
    ctx.error_count += 1
    if ctx.error_count >= 3:
        ctx = self.machine.transition(ctx, ConversationState.ERROR)
        return "I'm having trouble running your eligibility check right now. Please try again in a few minutes."
    # Stay in current state and ask user to wait
    return "I'm having a brief issue. Could you give me just a moment?"

Mistake 5: Overly restrictive state transitions blocking legitimate user behavior

Users don't follow scripts. Someone might spontaneously say "actually, I want to change my income figure" while you're in COLLECTING_LOAN_DETAILS. A rigid machine rejects this.

Fix: Add explicit "correction" transitions back to earlier states, and detect correction intent in your input handling. You can model this as a CORRECTING state that routes back to whichever collection state is appropriate.

---

Managing Context Window Costs at Scale

One thing that bites teams late is the cost of including full message history in every LLM call. As conversations grow longer, your token usage balloons — and with a state machine, you often don't need the full history.

Because you have explicit state and structured collected_data, you can often compress message history aggressively. Instead of sending 20 turns of conversation for the generation call, summarize earlier turns into a single "conversation summary so far" message. The Implementing Conversational Memory: Managing Context Windows and Chat History at Scale lesson covers this in detail.

A practical heuristic: only send the last N turns verbatim (3-5 is usually enough for coherence), and inject a structured summary of collected_data in the system prompt rather than relying on the model to remember facts from early in the conversation. Your state machine already tracks the facts — use it.

---

Summary & Next Steps

You've built a production-grade conversational state machine from the ground up. Here's what you accomplished:

• Modeled a multi-turn workflow as a formal finite state machine with explicit states, transitions, and guards, keeping workflow logic completely separate from message content
• Implemented a validation engine that enforces allowed transitions and guards, raising meaningful errors when violations occur
• Built a hybrid persistence layer that uses Redis for fast session access and PostgreSQL for durable history and audit logs
• Connected LLM calls to the state machine with a clean separation between extraction (structured output, cheap model) and generation (natural language, quality model)
• Handled real-world concerns like session timeouts, user abandonment, concurrent requests, and LLM failures

The patterns here scale to significantly more complex workflows. As you extend them, consider these directions:

• Parallel sub-states: Some workflows have branches that can be completed in any order. You can model these as a set of required "checkboxes" within a single state rather than forcing a linear sequence.
• Human-in-the-loop escalation: Add an AWAITING_HUMAN_REVIEW state that suspends the agent and sends a notification. The human approves or rejects via a separate interface, which triggers a transition back into the machine. This is exactly the pattern described in Agentic Loop Design.
• Tool integration: When your states need to call external APIs (credit checks, inventory lookups, calendar availability), connect them via function calling and tool use — the state machine defines when tools are called, tools don't define state.
• Long-term memory across sessions: Your collected_data can seed a longer-term user profile. The Implementing Structured Agent Memory lesson covers how to build episodic, semantic, and working memory layers that persist across separate conversation sessions.

The core lesson is one of discipline: explicit state over implicit inference, persistence over reconstruction, separation of concerns over monolithic prompts. Once you internalize these principles, you can build agents that feel genuinely intelligent because they're actually reliable.