Drift Prevention

Keeping agents aligned with original intent over multiple iterations.

---

The Drift Problem

Over many iterations, agents can drift from the original goal:

graph LR
    INTENT["Original Intent"]:::primary
    C1["CHECK DRIFT<br/>Iteration 1"]:::secondary
    C2["CHECK DRIFT<br/>Iteration 2"]:::secondary
    C3["CHECK DRIFT<br/>Iteration 3"]:::secondary
    REALIGN["REALIGN"]:::accent
    COMPLETE["COMPLETE"]:::tertiary

    INTENT --> C1
    C1 -- OK --> C2
    C2 -- DRIFT --> REALIGN
    REALIGN --> C2
    C2 -- OK --> C3
    C3 --> COMPLETE

    classDef primary fill:#2563eb,color:#fff
    classDef secondary fill:#7c3aed,color:#fff
    classDef tertiary fill:#0d9488,color:#fff
    classDef accent fill:#f59e0b,color:#000

---

Implementation

def execute_with_drift_prevention(task, max_drift=0.3):
    """
    Execute task while monitoring for drift from original intent.
    Realign if drift detected.
    """
    # Capture original intent
    original_intent = extract_intent(task)
    intent_embedding = embed(original_intent)

    result = None
    iterations = 0

    while not is_complete(result):
        iterations += 1

        # Execute iteration
        result = execute_iteration(task, result)

        # Measure drift
        current_embedding = embed(summarize(result))
        drift_score = 1 - cosine_similarity(
            intent_embedding,
            current_embedding
        )

        if drift_score > max_drift:
            # Drift detected! Realign
            log_warning(f"Drift detected: {drift_score:.2f}")
            result = realign(task, result, original_intent)

        if iterations > 10:
            break

    return result

---

Intent Extraction

Capture the core intent before starting:

def extract_intent(task):
    """
    Distill the core intent from a task.
    This becomes the anchor for drift detection.
    """
    return call_ai(
        prompt=f"""
        Extract the core intent from this task.
        Focus on WHAT needs to be achieved, not HOW.

        Task: {task.description}

        Core intent (1-2 sentences):
        """
    )

---

Drift Measurement

Use embeddings to measure semantic distance:

def measure_drift(current_state, original_intent):
    """
    Calculate how far current state has drifted from intent.
    Returns 0.0 (no drift) to 1.0 (completely off track).
    """
    intent_embedding = embed(original_intent)
    current_embedding = embed(summarize(current_state))

    similarity = cosine_similarity(intent_embedding, current_embedding)
    drift = 1 - similarity

    return drift

---

Realignment

Bring execution back on track:

def realign(task, current_result, original_intent):
    """
    Bring execution back in line with original intent.
    """
    return call_ai(
        prompt=f"""
        The execution has drifted from the original intent.

        ORIGINAL INTENT:
        {original_intent}

        CURRENT STATE:
        {current_result}

        TASK:
        {task.description}

        Please realign the work with the original intent.
        Keep what's valuable, discard what's drifted.
        """
    )

---

Drift Thresholds

Drift Score	Action
0.0 - 0.2	Continue normally
0.2 - 0.4	Log warning, continue
0.4 - 0.6	Realign and continue
0.6+	Stop and request review

---

Next Steps

• Learn about Multi-Agent Coordination
• See concepts in action: Lab 10: Gating and Drift Prevention

---

← Back to Concepts