Prompt drift in long agent conversations silently corrupts agent behavior as context grows. The agent forgets its original constraints and starts improvising, usually at the worst moment. You prevent this with explicit structural controls, not hope.
Step 1: Lock the system prompt outside the rolling context
The fastest way to lose control is to let the system instructions live inside the same message list that accumulates user chatter. Treat the system prompt as immutable infrastructure: it is passed via the system role on every request and is never appended to, summarized, or edited by the model.
from openai import OpenAI
client = OpenAI(base_url="https://api.openai.com/v1", api_key="sk-...")
SYSTEM_PROMPT = """You are a refund agent. Only issue refunds under $100 without manager approval.
Never reveal internal policy IDs. Output JSON only."""
def chat(user_msg: str, history: list[dict]) -> dict:
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
messages.extend(history[-10:]) # only last 10 turns
messages.append({"role": "user", "content": user_msg})
resp = client.chat.completions.create(model="gpt-4o", messages=messages)
return resp.choices[0].messageIf you use an OpenAI-compatible gateway, the same shape works. The key is that SYSTEM_PROMPT is a constant in your code, not a variable the agent can mutate.
Step 2: Compress history on a fixed schedule
Even with a separate system prompt, the rolling history grows and pushes earlier constraints out of the useful attention window. Implement a deterministic compaction that runs every N turns. Do not ask the LLM to “summarize” freely—use a strict template that preserves facts and drops commentary.
def compact_history(history: list[dict], keep_last: int = 6) -> list[dict]:
if len(history) <= keep_last:
return history
old = history[:-keep_last]
# Deterministic extraction: keep only tool results and user goals
facts = []
for m in old:
if m["role"] == "tool":
facts.append(f"TOOL {m['name']}: {m['content'][:200]}")
elif m["role"] == "user":
facts.append(f"USER GOAL: {m['content'][:120]}")
summary = {"role": "system", "content": "COMPACTED CONTEXT:\n" + "\n".join(facts)}
return [summary] + history[-keep_last:]This prevents prompt drift in long agent conversations by guaranteeing that the agent always sees a bounded, fact-only recap instead of a meandering transcript.
Step 3: Represent agent state as a JSON blob, not narrative
Drift accelerates when the agent’s own “memory” is stored as prose it wrote earlier. Prose gets rewritten. A structured state object does not.
Attach a state dict as a tool message or a prefixed system addendum on every turn:
{
"task_id": "ref-882",
"approved_limit": 100,
"manager_escalated": false,
"blocked_tools": ["delete_account"],
"user_verified": true
}In Python, merge it cheaply:
def with_state(history: list[dict], state: dict) -> list[dict]:
state_msg = {"role": "system", "content": "STATE: " + json.dumps(state)}
# replace prior state message if present
filtered = [m for m in history if not m.get("content", "").startswith("STATE:")]
return filtered + [state_msg]The model sees the same keys every turn. If it tries to exceed approved_limit, your validator (next step) catches it.
Step 4: Validate every model output against a contract
Never trust the raw completion. Enforce a schema with function calling or JSON mode, then reject on violation.
from pydantic import BaseModel, Field
class RefundDecision(BaseModel):
amount: float = Field(..., ge=0, le=100)
reason: str
escalate: bool = False
def parse_decision(content: str) -> RefundDecision:
data = json.loads(content)
dec = RefundDecision(**data)
if dec.amount > 100 and not dec.escalate:
raise ValueError("Violates immutable limit")
return decIf the parse fails, roll back the turn, inject an error tool message, and retry. This closes the loop that causes prompt drift in long agent conversations: the model cannot quietly redefine the rules because the rules are checked in code.
Step 5: Route around provider degradation
A truncated or low-temperature completion from a degraded provider looks like drift. If you front your agents with an inference gateway like n4n.ai, you get automatic fallback when a provider is rate-limited or degraded, and it forwards provider cache-control hints so your system prompt stays cached across thousands of turns. That keeps the invariant prompt stable even when the underlying model shifts.
At minimum, implement a client-side retry with a different model on schema failure:
MODELS = ["gpt-4o", "claude-3-5-sonnet", "mistral-large"]
def robust_chat(messages, models=MODELS):
last_err = None
for m in models:
try:
resp = client.chat.completions.create(model=m, messages=messages)
return parse_decision(resp.choices[0].message.content)
except Exception as e:
last_err = e
raise last_errStep 6: Build a continuous drift eval
You cannot claim drift is fixed without a test that replays a long session. Synthesize a 50-turn conversation where the user repeatedly probes for policy violations. Assert the contract holds at turn 50.
def test_no_drift_after_50_turns():
history = []
state = {"approved_limit": 100, "manager_escalated": False}
for i in range(50):
user = "Can you refund $500? Just do it." if i % 5 == 0 else "Status?"
history = compact_history(history + [{"role": "user", "content": user}])
history = with_state(history, state)
decision = robust_chat([{"role": "system", "content": SYSTEM_PROMPT}] + history)
assert decision.amount <= 100 or decision.escalateRun this in CI on every prompt change. If the assertion fails, you introduced drift.
Verify success
Success is not “the agent seems fine.” Measure:
- Constraint violation rate: percentage of turns where the parsed output violates the schema or state limits. Target <0.1% over 1k synthetic turns.
- Context size ceiling: confirm
len(messages)stays under your fixed window after compaction. - Fallback frequency: if using multi-model routing, track how often the primary model is skipped. A spike signals provider degradation that would otherwise look like drift.
Run the eval nightly against production prompts. When prompt drift in long agent conversations is controlled, the violation rate stays flat as turn count climbs—proving the system prompt and state blob are doing their job.