Building Conversational Flows with OpenAI Agents SDK: Multi-Turn State Management

Conversations Are State Machines

Every structured conversation follows a pattern: greet the user, collect information, confirm details, execute an action, and close. This is a state machine. The OpenAI Agents SDK does not force a specific state management approach, which gives you the flexibility to implement exactly the pattern your use case needs.

This guide shows you how to build structured conversational flows with explicit state tracking, slot filling, and flow control.

Defining Conversation State

Start with a clear state model that tracks where the user is in the flow and what data has been collected.

from pydantic import BaseModel
from enum import Enum
from typing import Any


class FlowState(str, Enum):
    GREETING = "greeting"
    COLLECTING_INFO = "collecting_info"
    CONFIRMING = "confirming"
    EXECUTING = "executing"
    COMPLETED = "completed"
    CANCELLED = "cancelled"


class SlotValue(BaseModel):
    value: Any | None = None
    confirmed: bool = False
    attempts: int = 0


class BookingState(BaseModel):
    flow_state: FlowState = FlowState.GREETING
    slots: dict[str, SlotValue] = {}
    required_slots: list[str] = ["date", "time", "service", "name", "phone"]
    errors: list[str] = []

    def get_missing_slots(self) -> list[str]:
        return [
            slot for slot in self.required_slots
            if slot not in self.slots or self.slots[slot].value is None
        ]

    def all_slots_filled(self) -> bool:
        return len(self.get_missing_slots()) == 0

    def get_slot_summary(self) -> str:
        lines = []
        for slot_name in self.required_slots:
            slot = self.slots.get(slot_name)
            if slot and slot.value:
                status = "confirmed" if slot.confirmed else "pending"
                lines.append(f"- {slot_name}: {slot.value} ({status})")
            else:
                lines.append(f"- {slot_name}: [not provided]")
        return "\n".join(lines)

Building the Slot Filling Agent

Create tools that let the agent update the conversation state as it collects information.

from agents import Agent, Runner, function_tool, RunContextWrapper


@function_tool
async def set_slot(ctx: RunContextWrapper[BookingState], slot_name: str, value: str) -> str:
    """Set a slot value collected from the user."""
    state: BookingState = ctx.context
    if slot_name not in state.required_slots:
        return f"Unknown slot: {slot_name}. Valid slots: {state.required_slots}"

    state.slots[slot_name] = SlotValue(value=value, confirmed=False)
    missing = state.get_missing_slots()
    if missing:
        return f"Slot '{slot_name}' set to '{value}'. Still need: {', '.join(missing)}"
    else:
        state.flow_state = FlowState.CONFIRMING
        return f"Slot '{slot_name}' set to '{value}'. All slots filled. Ask user to confirm."


@function_tool
async def get_state(ctx: RunContextWrapper[BookingState]) -> str:
    """Get current booking state and missing information."""
    state: BookingState = ctx.context
    summary = state.get_slot_summary()
    missing = state.get_missing_slots()
    return f"Current state: {state.flow_state.value}\n{summary}\nMissing: {missing or 'none'}"


@function_tool
async def confirm_booking(ctx: RunContextWrapper[BookingState]) -> str:
    """Confirm the booking after user approval."""
    state: BookingState = ctx.context
    if not state.all_slots_filled():
        return f"Cannot confirm. Missing: {state.get_missing_slots()}"
    for slot in state.slots.values():
        slot.confirmed = True
    state.flow_state = FlowState.EXECUTING
    return "Booking confirmed. Proceeding with execution."


@function_tool
async def cancel_flow(ctx: RunContextWrapper[BookingState]) -> str:
    """Cancel the current booking flow."""
    state: BookingState = ctx.context
    state.flow_state = FlowState.CANCELLED
    return "Booking cancelled."

The Conversational Agent

Wire the tools into an agent with instructions that guide the conversation flow.

booking_agent = Agent(
    name="booking_assistant",
    instructions="""You are a booking assistant. Follow this flow:

1. GREETING: Welcome the user and ask what service they need.
2. COLLECTING_INFO: Ask for missing information one field at a time.
   Use set_slot to record each piece of information.
   Required: date, time, service, name, phone.
3. CONFIRMING: Summarize the booking and ask the user to confirm.
4. EXECUTING: Tell the user the booking is confirmed.

Rules:
- Ask for ONE piece of information at a time.
- If the user provides multiple details in one message, set all of them.
- Always use get_state to check what is still missing.
- If the user wants to cancel, use cancel_flow.
- Be conversational and helpful, not robotic.""",
    tools=[set_slot, get_state, confirm_booking, cancel_flow],
)

Running Multi-Turn Conversations

The key to multi-turn flows is preserving conversation history and state across calls.

import asyncio
from agents.items import TResponseInputItem


async def run_booking_flow():
    state = BookingState()
    history: list[TResponseInputItem] = []

    print("Booking Assistant: Welcome! How can I help you today?")

    while state.flow_state not in (FlowState.COMPLETED, FlowState.CANCELLED):
        user_input = input("You: ")
        if not user_input.strip():
            continue

        history.append({"role": "user", "content": user_input})

        result = await Runner.run(
            booking_agent,
            input=history,
            context=state,
        )

        # Update history with full turn
        history = result.to_input_list()

        print(f"Assistant: {result.final_output}")

        if state.flow_state == FlowState.EXECUTING:
            state.flow_state = FlowState.COMPLETED
            print("\n--- Booking Complete ---")
            print(state.get_slot_summary())

asyncio.run(run_booking_flow())

Handling Edge Cases in Flows

Real conversations are messy. Users change their mind, provide partial information, or go off-topic.

@function_tool
async def update_slot(ctx: RunContextWrapper[BookingState], slot_name: str, new_value: str) -> str:
    """Update a previously set slot value (user changed their mind)."""
    state: BookingState = ctx.context
    if slot_name not in state.slots:
        return f"Slot '{slot_name}' has not been set yet. Use set_slot instead."

    old_value = state.slots[slot_name].value
    state.slots[slot_name] = SlotValue(value=new_value, confirmed=False)
    # Reset to collecting state if we were in confirming
    if state.flow_state == FlowState.CONFIRMING:
        state.flow_state = FlowState.COLLECTING_INFO
    return f"Updated '{slot_name}' from '{old_value}' to '{new_value}'."

FAQ

How do I handle conversation timeouts?

Track a last_active timestamp in your state object. Before processing each turn, check if the elapsed time exceeds your timeout threshold. If it does, reset the state and start fresh with a greeting that acknowledges the gap — something like "It has been a while since we spoke. Would you like to continue where we left off?"

Can I mix free-form conversation with structured slot filling?

Yes. Design your agent instructions to handle both modes. When the user asks a question unrelated to the booking flow, the agent can answer it normally without calling any slot-filling tools. The state persists unchanged until the user returns to the flow. Include a get_state call periodically to remind the agent what information is still needed.

How do I validate slot values (e.g., date format, phone number)?

Add validation logic inside the set_slot tool. Before storing the value, parse and validate it. Return a clear error message if validation fails, and increment the attempts counter on the slot. If attempts exceed a threshold, offer the user an alternative format or skip that slot with a default.

---

#OpenAIAgentsSDK #ConversationalAI #StateManagement #SlotFilling #MultiTurn #Python #AgenticAI #LearnAI #AIEngineering