Conversation Branching and History Management

Beyond Linear Conversations

Most agent conversations are linear — message follows message in sequence. But real-world usage patterns are messier. Users change their minds, want to explore alternative approaches, or realize a prior instruction was wrong. Linear history does not accommodate this well.

The OpenAI Agents SDK provides tools for non-linear conversation management: pop_item() for undo, session input callbacks for history customization, and patterns for branching conversations that explore multiple paths.

pop_item() for Undo and Correction

The pop_item() method removes the last item from a session and returns it. This is the building block for undo functionality.

flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus<br/>classify"]
    PLAN["Plan and tool<br/>selection"]
    AGENT["Agent loop<br/>LLM plus tools"]
    GUARD{"Guardrails<br/>and policy"}
    EXEC["Execute and<br/>verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus<br/>next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff

Basic Undo

from agents import Agent, Runner
from agents.extensions.sessions import SQLiteSession

session = SQLiteSession(db_path="./branching.db")

agent = Agent(
    name="WritingAssistant",
    instructions="Help the user write and revise content.",
)

async def undo_last_turn(session_id: str):
    """Remove the last assistant response and the user message that triggered it."""
    # Pop the assistant response
    assistant_item = await session.pop_item(session_id)
    # Pop the user message
    user_item = await session.pop_item(session_id)

    print(f"Undid: User said '{user_item}' -> Assistant said '{assistant_item}'")
    return user_item, assistant_item

Undo with Retry

A common pattern: undo the last exchange and regenerate with the same input:

async def retry_last_turn(session_id: str):
    """Undo the last turn and regenerate the response."""
    # Remove last assistant response
    await session.pop_item(session_id)
    # Remove last user message
    user_item = await session.pop_item(session_id)

    if user_item is None:
        print("Nothing to retry")
        return None

    # Re-run with the same input — may get a different response
    result = await Runner.run(
        agent,
        user_item.get("content", ""),
        session=session,
        session_id=session_id,
    )
    return result.final_output

Multi-Step Undo

Undo multiple turns to return to an earlier point in the conversation:

async def undo_n_turns(session_id: str, n: int):
    """Undo the last n conversation turns (each turn = user + assistant)."""
    removed = []
    for _ in range(n * 2):  # Each turn has 2 items
        item = await session.pop_item(session_id)
        if item is None:
            break
        removed.append(item)

    print(f"Removed {len(removed)} items, undid {len(removed) // 2} turns")
    return removed

Session Input Callbacks for History Customization

Session input callbacks let you transform the conversation history before it is sent to the model. This is powerful for injecting context, filtering history, or restructuring the conversation.

Adding System Context

Inject dynamic context at the start of every conversation:

from datetime import datetime

def add_system_context(items: list) -> list:
    """Prepend dynamic system context to every conversation."""
    system_context = {
        "role": "system",
        "content": f"""Current time: {datetime.utcnow().isoformat()}
        Active promotions: 20% off annual plans
        System status: All services operational""",
    }
    return [system_context] + items

Filtering Sensitive Content

Remove or redact sensitive information before it reaches the model:

import re

def redact_sensitive_data(items: list) -> list:
    """Redact PII from conversation history before sending to model."""
    redacted = []
    for item in items:
        if isinstance(item, dict) and "content" in item:
            content = item["content"]
            # Redact credit card numbers
            content = re.sub(r'\bd{4}[s-]?d{4}[s-]?d{4}[s-]?d{4}\b',
                           '[REDACTED_CC]', content)
            # Redact SSNs
            content = re.sub(r'\bd{3}-d{2}-d{4}\b',
                           '[REDACTED_SSN]', content)
            redacted.append({**item, "content": content})
        else:
            redacted.append(item)
    return redacted

Summarizing Old History

Keep recent messages in full detail but summarize older ones:

async def summarize_old_history(items: list, keep_recent: int = 10) -> list:
    """Keep recent items verbatim, summarize older ones."""
    if len(items) <= keep_recent:
        return items

    old_items = items[:-keep_recent]
    recent_items = items[-keep_recent:]

    # Create a summary of the old items
    old_text = "\n".join(
        f"{item.get('role', 'unknown')}: {item.get('content', '')}"
        for item in old_items
        if isinstance(item, dict)
    )

    summary = {
        "role": "system",
        "content": f"Summary of earlier conversation ({len(old_items)} messages):\n{old_text[:2000]}",
    }

    return [summary] + recent_items

Branching Conversations

Branching lets users explore different paths from the same point. Think of it as "save points" in a conversation — you can return to a prior state and take a different direction.

Implementing Branches with Session Copying

import copy

class BranchableSession:
    """Session wrapper that supports conversation branching."""

    def __init__(self, base_session):
        self.base_session = base_session
        self.branches: dict[str, str] = {}  # branch_name -> session_id

    async def create_branch(
        self, source_session_id: str, branch_name: str
    ) -> str:
        """Create a branch — copy current session state to a new session_id."""
        branch_session_id = f"{source_session_id}:branch:{branch_name}"

        # Copy all items from source to branch
        items = await self.base_session.get_items(source_session_id)
        if items:
            await self.base_session.add_items(branch_session_id, items)

        self.branches[branch_name] = branch_session_id
        return branch_session_id

    async def list_branches(self, source_session_id: str) -> list[str]:
        """List all branches for a session."""
        return [
            name for name, sid in self.branches.items()
            if sid.startswith(source_session_id)
        ]

    async def merge_branch(
        self, branch_name: str, target_session_id: str
    ) -> None:
        """Replace target session with branch content."""
        branch_sid = self.branches[branch_name]
        branch_items = await self.base_session.get_items(branch_sid)

        await self.base_session.clear_session(target_session_id)
        if branch_items:
            await self.base_session.add_items(target_session_id, branch_items)

    async def delete_branch(self, branch_name: str) -> None:
        """Delete a branch."""
        branch_sid = self.branches.pop(branch_name)
        await self.base_session.clear_session(branch_sid)

Using Branches

from agents.extensions.sessions import SQLiteSession

base = SQLiteSession(db_path="./branching_demo.db")
branching = BranchableSession(base)

agent = Agent(
    name="StrategyAdvisor",
    instructions="Help the user explore different business strategies.",
)

async def explore_strategies():
    sid = "strategy-session-1"

    # Initial conversation
    await Runner.run(agent, "I'm launching a SaaS product.", session=base, session_id=sid)
    await Runner.run(agent, "Should I use freemium or paid-only?", session=base, session_id=sid)

    # Create branches to explore both paths
    freemium_sid = await branching.create_branch(sid, "freemium")
    paid_sid = await branching.create_branch(sid, "paid-only")

    # Explore freemium path
    result_free = await Runner.run(
        agent,
        "Let's explore the freemium model. What's the conversion rate expectation?",
        session=base,
        session_id=freemium_sid,
    )
    print(f"Freemium path: {result_free.final_output}")

    # Explore paid-only path
    result_paid = await Runner.run(
        agent,
        "Let's explore paid-only. What pricing tier structure works best?",
        session=base,
        session_id=paid_sid,
    )
    print(f"Paid path: {result_paid.final_output}")

    # User decides on freemium — merge that branch back
    await branching.merge_branch("freemium", sid)
    print("Merged freemium branch into main conversation")

History Pruning Strategies

For long-running sessions, you need strategies to keep history manageable without losing important context.

Strategy 1: Rolling Window

Keep only the N most recent items:

async def prune_to_window(session, session_id: str, window_size: int = 30):
    """Keep only the most recent items."""
    items = await session.get_items(session_id)
    if len(items) <= window_size:
        return  # No pruning needed

    keep = items[-window_size:]
    await session.clear_session(session_id)
    await session.add_items(session_id, keep)

Strategy 2: Importance-Based Pruning

Keep items that contain key information and prune filler:

def is_important(item: dict) -> bool:
    """Heuristic for whether an item contains important information."""
    content = str(item.get("content", ""))

    # Keep items with specific data
    if any(keyword in content.lower() for keyword in [
        "deadline", "budget", "decision", "agreed", "confirmed",
        "requirement", "must", "critical", "password", "account"
    ]):
        return True

    # Keep items with numbers (dates, amounts, IDs)
    if re.search(r'd{2,}', content):
        return True

    # Keep short items (likely questions or confirmations)
    if len(content) < 100:
        return True

    return False

async def importance_prune(session, session_id: str, max_items: int = 50):
    """Prune less important items while keeping critical ones."""
    items = await session.get_items(session_id)
    if len(items) <= max_items:
        return

    # Always keep the last 10 items
    recent = items[-10:]
    older = items[:-10]

    # From older items, keep only important ones
    important = [item for item in older if is_important(item)]

    # If still over limit, keep only the most recent important ones
    if len(important) + len(recent) > max_items:
        important = important[-(max_items - len(recent)):]

    pruned = important + recent
    await session.clear_session(session_id)
    await session.add_items(session_id, pruned)

Strategy 3: Summarize-and-Replace

Replace a block of old messages with a single summary message:

async def summarize_and_replace(
    session, session_id: str, summarizer_agent: Agent, keep_recent: int = 15
):
    """Replace old history with a summary, keep recent messages intact."""
    items = await session.get_items(session_id)
    if len(items) <= keep_recent + 5:
        return

    old_items = items[:-keep_recent]
    recent_items = items[-keep_recent:]

    # Use an agent to summarize the old conversation
    old_text = "\n".join(str(item) for item in old_items)
    summary_result = await Runner.run(
        summarizer_agent,
        f"Summarize this conversation history concisely, preserving all key facts, decisions, and action items:\n\n{old_text}",
    )

    summary_item = {
        "role": "system",
        "content": f"[Conversation summary]: {summary_result.final_output}",
    }

    await session.clear_session(session_id)
    await session.add_items(session_id, [summary_item] + recent_items)

These pruning strategies let you maintain long-running conversations without unbounded memory growth or context window overflow.

Sources:

• https://openai.github.io/openai-agents-python/sessions/
• https://platform.openai.com/docs/guides/conversation-state