From Chatbot to Agent: Why Your AI App Needs Observability, Better Memory, and Real RAG

flowchart TD
    A[User Input] --> B[Agent Loop]
    B --> C{Thought}
    C --> D[Tool Call]
    D --> E[Observability Trace]
    E --> F[Tool Result]
    F --> C
    C --> G[Final Answer]
    G --> H[Memory Write]

# agent_observability.py
import uuid
from datetime import datetime, timezone
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter

# Bootstrap — run this once at app startup
provider = TracerProvider()
trace.set_tracer_provider(provider)
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(
    endpoint="http://localhost:4318/v1/traces"
)))

meter_provider = MeterProvider(
    metric_readers=[PeriodicExportingMetricReader(OTLPMetricExporter(
        endpoint="http://localhost:4318/v1/metrics"
    ))]
)
metrics.set_meter_provider(meter_provider)

tracer = trace.get_tracer("agent.runtime")
counter = metrics.get_meter("agent.runtime").create_counter(
    "agent.tools.called",
    description="Tool call count by name and status",
)
histogram = metrics.get_meter("agent.runtime").create_histogram(
    "agent.turns",
    description="Number of agent loop iterations per user turn",
)

# agent_loop.py
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

async def run_agent(query: str, tools: list, memory: MemoryStore, max_turns: int = 8):
    turn_id = str(uuid.uuid4())
    agent_id = str(uuid.uuid4())[:8]
    
    with tracer.start_as_current_span(f"agent.turn.{agent_id}", 
        attributes={"user.query": query, "max_turns": max_turns}) as root_span:
        
        history = await memory.retrieve_context(turn_id)
        conversation = build_messages(query, history)
        
        for turn in range(1, max_turns + 1):
            with tracer.start_as_current_span(f"agent.thought.{turn}") as thought_span:
                llm_response = await call_llm(conversation, tools=tools)
                thought_span.set_attribute("model", llm_response.model)
                thought_span.set_attribute("prompt_tokens", llm_response.usage.prompt_tokens)
                thought_span.set_attribute("completion_tokens", llm_response.usage.completion_tokens)
            
            action = parse_action(llm_response)
            
            if action.type == "final_answer":
                root_span.set_attribute("agent.loop.break_reason", "success")
                histogram.record(turn, {"turn_id": turn_id})
                await memory.write(turn_id, conversation, action.content)
                return action.content
            
            elif action.type == "tool_call":
                with tracer.start_as_current_span(f"agent.tool.{action.name}") as tool_span:
                    tool_start = datetime.now(timezone.utc)
                    try:
                        result = await call_tool(action.name, action.arguments)
                        tool_span.set_attribute("tool.status", "success")
                        counter.add(1, {
                            "tool": action.name,
                            "status": "success",
                            "turn_id": turn_id
                        })
                    except Exception as e:
                        tool_span.set_status(Status(StatusCode.ERROR, str(e)))
                        tool_span.record_exception(e)
                        counter.add(1, {
                            "tool": action.name,
                            "status": "error",
                            "turn_id": turn_id
                        })
                        result = f"Tool error: {type(e).__name__}: {e}"
                
                conversation.append({"role": "tool", "tool_call_id": action.id, "content": result})
        
        # Max steps hit
        root_span.set_attribute("agent.loop.break_reason", "max_steps")
        histogram.record(max_turns, {"turn_id": turn_id})
        root_span.set_status(Status(StatusCode.ERROR, "Max turns exceeded"))
        return "I couldn't resolve this within the allowed steps."

┌─────────────────────────────────────────────────────┐
│                 SHORT-TERM (Working)                │
│  Last N messages · tool results · current turn      │
│  Lifetime: session · 4K–128K tokens · in-process    │
├─────────────────────────────────────────────────────┤
│                 MEDIUM-TERM ( Episodic)             │
│  Recent interactions · facts discovered this week   │
│  Lifetime: days–weeks · vector store · ~1K entries  │
├─────────────────────────────────────────────────────┤
│                 LONG-TERM (Semantic / Profile)      │
│  User preferences · persistent facts · schema       │
│  Lifetime: months–years · structured DB + vectors   │
└─────────────────────────────────────────────────────┘

# memory/short_term.py
from typing import list, Optional
from langchain_core.messages import BaseMessage, SystemMessage, AIMessage, HumanMessage
import tiktoken

MAX_CONTEXT_TOKENS = 120_000
SUMMARY_THRESHOLD = 80_000  # tokens before we compress

class ShortTermMemory:
    def __init__(self, model: str = "gpt-4o"):
        self.encoder = tiktoken.encoding_for_model(model)
        self.messages: list[BaseMessage] = []
    
    def add(self, message: BaseMessage):
        self.messages.append(message)
        self._maybe_compress()
    
    def _token_count(self) -> int:
        total = 0
        for msg in self.messages:
            content = getattr(msg, "content", "")
            if isinstance(content, str):
                total += len(self.encoder.encode(content))
            elif isinstance(content, list):
                for block in content:
                    if isinstance(block, dict) and "text" in block:
                        total += len(self.encoder.encode(block["text"]))
        return total
    
    async def _maybe_compress(self):
        if self._token_count() < SUMMARY_THRESHOLD:
            return
        
        # Keep last N messages intact, summarize the rest
        cutoff = max(0, len(self.messages) - 20)  # preserve last 20 msgs
        old_messages = self.messages[:cutoff]
        new_messages = self.messages[cutoff:]
        
        summary_prompt = SystemMessage(
            content="Summarize the following conversation excerpt in 3-5 bullet points. "
                    "Preserve key facts, decisions, and user preferences. Be concise."
        )
        
        llm = get_chat_model()  # your model factory
        summary_resp = await llm.ainvoke([
            summary_prompt,
            HumanMessage(content=str(old_messages))
        ])
        
        self.messages = [AIMessage(content=f"[Previous conversation summary]: {summary_resp.content}")] + new_messages
    
    def get_context(self) -> list[dict]:
        return [{"role": m.type, "content": m.content} for m in self.messages]

# memory/episodic.py
import asyncio
from datetime import datetime, timedelta
from typing import Optional
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import numpy as np

class EpisodicMemory:
    """Stores recent agent interactions for contextual retrieval."""
    
    def __init__(self, collection: str = "agent_episodes", dim: int = 1536):
        self.client = QdrantClient("localhost", port=6333)
        self.collection = collection
        self.dim = dim
        self._ensure_collection()
    
    def _ensure_collection(self):
        if not self.client.collection_exists(self.collection):
            self.client.create_collection(
                collection_name=self.collection,
                vectors_config=VectorParams(size=self.dim, distance=Distance.COSINE)
            )
    
    async def write(self, turn_id: str, query: str, response: str, 
                    tools_used: list[str], success: bool):
        """Store an episode after a turn completes."""
        embedding = await self._embed(f"{query} {response}")
        
        payload = {
            "turn_id": turn_id,
            "query": query,
            "response": response,
            "tools_used": tools_used,
            "success": success,
            "created_at": datetime.utcnow().isoformat(),
        }
        
        point = PointStruct(
            id=hash(turn_id) % (2**63),
            vector=embedding,
            payload=payload
        )
        self.client.upsert(collection_name=self.collection, points=[point])
    
    async def retrieve(self, query: str, limit: int = 5) -> list[dict]:
        """Find semantically similar past interactions."""
        embedding = await self._embed(query)
        hits = self.client.search(
            collection_name=self.collection,
            query_vector=embedding,
            limit=limit,
            with_payload=True
        )
        return [hit.payload for hit in hits]
    
    async def _embed(self, text: str) -> list[float]:
        resp = await self._get_embedding(text)
        return resp.embedding
    
    async def _get_embedding(self, text: str) -> dict:
        # Your embedding provider — OpenAI, Cohere, local, etc.
        import openai
        client = openai.AsyncOpenAI()
        r = await client.embeddings.create(model="text-embedding-3-small", input=text)
        return r.data[0]

# memory/write_queue.py
import asyncio
from collections import deque

class AsyncWriteQueue:
    """Buffer episodic memory writes to avoid blocking the agent loop."""
    
    def __init__(self, memory: EpisodicMemory, batch_size: int = 10, flush_interval: float = 2.0):
        self.memory = memory
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self._queue: deque = deque()
        self._lock = asyncio.Lock()
    
    async def enqueue(self, episode: dict):
        async with self._lock:
            self._queue.append(episode)
            if len(self._queue) >= self.batch_size:
                await self._flush_locked()
    
    async def _flush_locked(self):
        batch = [self._queue.popleft() for _ in range(min(self.batch_size, len(self._queue)))]
        await asyncio.gather(*[self.memory.write(**ep) for ep in batch])
    
    async def drain(self):
        while self._queue:
            async with self._lock:
                batch = list(self._queue)
                self._queue.clear()
            await asyncio.gather(*[self.memory.write(**ep) for ep in batch])

# memory/profile.py
from sqlalchemy import create_engine, Column, String, JSON, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
import uuid

Base = declarative_base()

class UserProfile(Base):
    __tablename__ = "user_profiles"
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    user_id = Column(String, index=True, nullable=False)
    preferences = Column(JSON, default=dict)
    facts = Column(JSON, default=dict)  # {"key": {"value": ..., "confidence": 0.9, "source": ...}}
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

class ProfileStore:
    def __init__(self, db_url: str = "postgresql://localhost/agent_db"):
        self.engine = create_engine(db_url)
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)
    
    def get_or_create(self, user_id: str) -> dict:
        sess = self.Session()
        profile = sess.query(UserProfile).filter_by(user_id=user_id).first()
        if not profile:
            profile = UserProfile(user_id=user_id)
            sess.add(profile)
            sess.commit()
        return {
            "user_id": profile.user_id,
            "preferences": profile.preferences or {},
            "facts": profile.facts or {}
        }
    
    def update_fact(self, user_id: str, key: str, value: str, 
                    confidence: float = 0.9, source: str = "user_statement"):
        """Update a fact with confidence-weighted merging."""
        sess = self.Session()
        profile = sess.query(UserProfile).filter_by(user_id=user_id).one()
        facts = profile.facts or {}
        
        existing = facts.get(key)
        if existing:
            # Confidence-weighted merge: new fact wins if confidence > existing
            if confidence >= existing.get("confidence", 0):
                facts[key] = {"value": value, "confidence": confidence, "source": source}
        else:
            facts[key] = {"value": value, "confidence": confidence, "source": source}
        
        profile.facts = facts
        sess.commit()

flowchart LR
    Q[User Query] --> U[Query Understanding]
    U --> S1[Structured Filter]
    U --> S2[Semantic Search]
    U --> S3[Keyword/BM25]
    S1 --> R[Re-Ranker]
    S2 --> R
    S3 --> R
    R --> C[Context Selector]
    C --> P[Prompt Assembly]
    P --> L[LLM Synthesis]
    L --> O[Output]

# rag/query_understanding.py
from pydantic import BaseModel
from enum import Enum

class QueryIntent(str, Enum):
    PROCEDURAL = "procedural"      # How do I...
    FACTUAL = "factual"            # What is...
    COMPARATIVE = "comparative"    # vs / compare
    DEBUG = "debug"                # Error: ... why...
    PLANNING = "planning"          # I want to build...

class RewrittenQuery(BaseModel):
    original: str
    intent: QueryIntent
    rewritten: str
    filters: dict  # {"doc_type": "api_ref", "product": "api_gateway", ...}
    decomposed: list[str]  # sub-queries for complex requests

class QueryRewriter:
    """Decomposes and rewrites user queries for better retrieval."""
    
    REWRITE_PROMPT = """You are a query rewriting engine for a RAG system.

Given the user's query, determine the intent and rewrite it for optimal document retrieval.
Rules:
- Procedural queries: convert to action-oriented form
- Factual queries: extract key entities, normalize terminology
- Debug queries: surface the error pattern and context
- Add filter hints based on explicit product/service names
- Decompose complex multi-part queries into independent sub-queries

Return JSON only with keys: intent, rewritten, filters, decomposed."""
    
    async def process(self, query: str, user_context: dict = None) -> RewrittenQuery:
        # Implementation calls your LLM with the rewrite prompt
        # This is the step most teams skip
        pass

# rag/retrieval.py
import asyncio
from typing import Optional
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue, ScoredPoint
from pymilvus import connections, Collection
import bm25s  # or whoosh, or elasticsearch

class MultiStrategyRetriever:
    """Retrieves context using three complementary strategies."""
    
    def __init__(self, qdrant: QdrantClient, bm25_index, collection: Collection):
        self.qdrant = qdrant
        self.bm25 = bm25_index
        self.collection = collection
    
    async def retrieve(self, rewritten: RewrittenQuery, top_k: int = 20) -> list[ScoredPoint]:
        tasks = [
            self._semantic_search(rewritten),
            self._keyword_search(rewritten),
            self._structured_filter_search(rewritten)
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Merge and deduplicate by document ID
        all_points: dict[int, tuple[ScoredPoint, str]] = {}  # id -> (point, strategy)
        strategies = ["semantic", "keyword", "structured"]
        
        for point, strategy in zip(results, strategies):
            if isinstance(point, Exception) or point is None:
                continue
            for p in point:
                if p.id not in all_points or p.score > all_points[p.id][0].score:
                    all_points[p.id] = (p, strategy)
        
        # Return combined results sorted by score
        merged = [v[0] for v in all_points.values()]
        merged.sort(key=lambda x: x.score, reverse=True)
        return merged[:top_k]
    
    async def _semantic_search(self, q: RewrittenQuery) -> list[ScoredPoint]:
        embedding = await self._embed(q.rewritten)
        return self.qdrant.search(
            collection_name="documents",
            query_vector=embedding,
            limit=15,
            with_payload=True
        )
    
    async def _keyword_search(self, q: RewrittenQuery) -> list[ScoredPoint]:
        # BM25 over the same document corpus
        tokens = self.bm25.get.corpus_tokenizer(q.rewritten)
        scores = self.bm25.retrieve(tokens, k=10)
        return scores  # Adapt to your BM25 library's return format
    
    async def _structured_filter_search(self, q: RewrittenQuery) -> list[ScoredPoint]:
        # Filter by metadata: doc_type, product, version, etc.
        filters = []
        for key, val in q.filters.items():
            if isinstance(val, list):
                filters.append(FieldCondition(key=key, match=MatchValue(values=val)))
            else:
                filters.append(FieldCondition(key=key, match=MatchValue(value=val)))
        
        query_filter = Filter(must=filters) if filters else None
        embedding = await self._embed(q.rewritten)
        
        return self.qdrant.search(
            collection_name="documents",
            query_vector=embedding,
            query_filter=query_filter,
            limit=10,
            with_payload=True
        )
    
    async def _embed(self, text: str) -> list[float]:
        import openai
        client = openai.AsyncOpenAI()
        r = await client.embeddings.create(model="text-embedding-3-small", input=text)
        return r.data[0].embedding

# rag/reranker.py
import asyncio
from typing import Optional
import httpx

class CrossEncoderReranker:
    """Re-ranks retrieved chunks using a cross-encoder model."""
    
    def __init__(self, model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2",
                 api_endpoint: Optional[str] = None):
        self.model = model
        self.api_endpoint = api_endpoint
    
    async def rerank(self, query: str, chunks: list[tuple[float, dict]]) -> list[tuple[float, dict]]:
        """
        chunks: list of (original_score, chunk_payload)
        Returns: list sorted by relevance score, descending
        """
        if not chunks:
            return []
        
        if self.api_endpoint:
            return await self._rerank_api(query, chunks)
        return await self._rerank_local(query, chunks)
    
    async def _rerank_api(self, query: str, chunks: list[tuple[float, dict]]) -> list[tuple[float, dict]]:
        payloads = [{"content": c["content"][:2000]} for _, c in chunks]
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.api_endpoint}/rerank",
                json={"query": query, "documents": payloads, "model": self.model}
            )
        results = resp.json()["results"]
        return [(r["index"], chunks[r["index"]][1], r["relevance_score"]) for r in results]
    
    async def _rerank_local(self, query: str, chunks: list[tuple[float, dict]]) -> list[tuple[float, dict]]:
        # Local re-ranking with sentence-transformers cross-encoder
        from sentence_transformers import CrossEncoder
        encoder = CrossEncoder(self.model)
        pairs = [(query, c["content"][:2000]) for _, c in chunks]
        scores = encoder.predict(pairs)
        indexed = list(enumerate(scores))
        indexed.sort(key=lambda x: x[1], reverse=True)
        return [(idx, chunks[idx][1], score) for idx, score in indexed]

# rag/context_selector.py
from typing import Optional

class ContextSelector:
    """Selects and assembles the final context for the LLM."""
    
    MAX_CONTEXT_TOKENS = 8000
    MIN_RELEVANCE_SCORE = 0.3
    
    def __init__(self, encoder):
        self.encoder = encoder
    
    def assemble(self, reranked: list[tuple[int, dict, float]], query: str) -> str:
        selected = []
        total_tokens = 0
        
        for idx, chunk, score in reranked:
            if score < self.MIN_RELEVANCE_SCORE:
                continue
            
            chunk_tokens = len(self.encoder.encode(chunk["content"]))
            if total_tokens + chunk_tokens > self.MAX_CONTEXT_TOKENS:
                # Truncate the last chunk to fit
                remaining = self.MAX_CONTEXT_TOKENS - total_tokens
                truncated_content = chunk["content"][:remaining * 4]  # rough estimate
                selected.append(f"<chunk score={score:.3f}>\n{truncated_content}\n</chunk>")
                total_tokens += remaining
                break
            
            selected.append(f"<chunk score={score:.3f}>\n{chunk["content"]}\n</chunk>")
            total_tokens += chunk_tokens
        
        context = "\n\n".join(selected)
        return f"Relevant documentation for the query '{query}':\n\n{context}"

# rag/eval.py
import json
from typing import Optional
from dataclasses import dataclass

@dataclass
class EvalResult:
    query: str
    expected_answer: str
    retrieved_score: float  # 0.0–1.0, from a judge model
    llm_answer: str
    faithfulness: float  # Does the answer ground to retrieved context?
    correctness: float   # Does the answer match expected?

class RAGEvaluator:
    """Evaluates RAG pipeline quality using LLM-as-judge."""
    
    FAITHFULNESS_PROMPT = """Determine if the following answer is grounded in the provided context.
Answer "yes" if the answer can be directly derived from the context. "
    "Answer "no" if the answer contains information not present in the context.
Context: {context}\nAnswer: {answer}\nResponse:"""
    
    CORRECTNESS_PROMPT = """Evaluate whether the answer correctly addresses the question.
Scale: 0.0 (completely wrong) to 1.0 (perfectly correct).
Question: {question}\nExpected: {expected}\nAnswer: {answer}\nScore (0.0-1.0):"""
    
    def __init__(self):
        self.results: list[EvalResult] = []
    
    async def evaluate(self, query: str, expected: str, 
                       context: str, llm_answer: str) -> EvalResult:
        # Faithfulness check
        faith_resp = await call_llm(self.FAITHFULNESS_PROMPT.format(
            context=context[:2000], answer=llm_answer))
        faithfulness = 1.0 if "yes" in faith_resp.content.lower() else 0.0
        
        # Correctness check
        corr_resp = await call_llm(self.CORRECTNESS_PROMPT.format(
            question=query, expected=expected, answer=llm_answer))
        try:
            correctness = float(corr_resp.content.strip())
            correctness = max(0.0, min(1.0, correctness))
        except ValueError:
            correctness = 0.5  # Default mid-point for ambiguous judgments
        
        result = EvalResult(
            query=query,
            expected_answer=expected,
            retrieved_score=0.8,  # Placeholder — replace with actual retrieval quality metric
            llm_answer=llm_answer,
            faithfulness=faithfulness,
            correctness=correctness
        )
        self.results.append(result)
        return result
    
    def report(self) -> dict:
        if not self.results:
            return {"sample_size": 0}
        avg_faithfulness = sum(r.faithfulness for r in self.results) / len(self.results)
        avg_correctness = sum(r.correctness for r in self.results) / len(self.results)
        return {
            "sample_size": len(self.results),
            "avg_faithfulness": round(avg_faithfulness, 3),
            "avg_correctness": round(avg_correctness, 3),
            "avg_retrieved_score": round(
                sum(r.retrieved_score for r in self.results) / len(self.results), 3
            )
        }

flowchart TD
    U[User] -->|Query| GW[API Gateway]
    GW -->|Auth + Rate Limit| AG[Agent Orchestrator]
    AG -->|Turn ID| MEM[Memory Layer]
    MEM -->|Short-term history| AG
    MEM -->|Episodic recall| AG
    MEM -->|User profile facts| AG
    AG -->|Rewritten query| RAG[RAG Pipeline]
    RAG -->|Context| AG
    AG -->|Tool definitions| TOOLS[Tool Registry]
    TOOLS -->|Invocation| SERVICE[Backend Services]
    AG -->|Trace + Metrics| OBS[Observability]
    AG -->|Final answer| GW
    GW --> U

# agent/orchestrator.py
from typing import Optional

class AgentOrchestrator:
    """Production-grade agent that integrates observability, memory, and RAG."""
    
    def __init__(
        self,
        short_term: ShortTermMemory,
        episodic: EpisodicMemory,
        profile: ProfileStore,
        rag: MultiStrategyRetriever,
        reranker: CrossEncoderReranker,
        context_selector: ContextSelector,
        evaluator: Optional[RAGEvaluator] = None,
        max_turns: int = 8
    ):
        self.short_term = short_term
        self.episodic = episodic
        self.profile = profile
        self.rag = rag
        self.reranker = reranker
        self.context_selector = context_selector
        self.evaluator = evaluator
        self.max_turns = max_turns
    
    async def handle(self, user_id: str, query: str) -> str:
        turn_id = str(uuid.uuid4())
        profile = self.profile.get_or_create(user_id)
        history = self.short_term.get_context()
        episodic_context = await self.episodic.retrieve(query, limit=3)
        
        # RAG retrieval with re-ranking
        rewritten = await self.rewriter.process(query, profile)
        raw_chunks = await self.rag.retrieve(rewritten, top_k=20)
        reranked = await self.reranker.rerank(query, raw_chunks)
        rag_context = self.context_selector.assemble(reranked, query)
        
        # Agent loop
        conversation = self._build_conversation(query, history, episodic_context, 
                                                rag_context, profile)
        final_answer = await self._run_agent_loop(conversation, tools=self._available_tools(),
                                                  turn_id=turn_id)
        
        # Persist
        self.short_term.add(HumanMessage(content=query))
        self.short_term.add(AIMessage(content=final_answer))
        await self.episodic.write(turn_id, query, final_answer, [], True)
        
        # Observability
        self._emit_metrics(turn_id, len(conversation), True)
        
        # Evaluation (async, non-blocking)
        if self.evaluator:
            asyncio.create_task(self.evaluator.evaluate(query, "", rag_context, final_answer))
        
        return final_answer

# Circuit breaker pattern for tool calls
from collections import defaultdict, deque
from datetime import datetime, timedelta

class ToolCircuitBreaker:
    def __init__(self, error_threshold: float = 0.5, window_seconds: float = 60.0):
        self.error_threshold = error_threshold
        self.window = window_seconds
        self._errors: defaultdict[str, deque] = defaultdict(deque)
    
    def record(self, tool: str, success: bool):
        now = datetime.utcnow()
        self._errors[tool].append((now, success))
        # Prune old entries
        cutoff = now - timedelta(seconds=self.window)
        self._errors[tool] = deque(
            ((t, s) for t, s in self._errors[tool] if t > cutoff),
            maxlen=100
        )
    
    def is_open(self, tool: str) -> bool:
        errors = self._errors.get(tool, [])
        if len(errors) < 5:  # Need minimum samples
            return False
        error_rate = sum(1 for _, s in errors if not s) / len(errors)
        return error_rate > self.error_threshold

# memory/retention.py
async def apply_retention_policy(memory: EpisodicMemory, max_age_days: int = 30):
    cutoff = datetime.utcnow() - timedelta(days=max_age_days)
    # Query and delete old points
    scroll = memory.client.scroll(collection_name="agent_episodes", limit=1000)
    for point, _ in scroll:
        created = datetime.fromisoformat(point.payload["created_at"])
        if created < cutoff:
            memory.client.delete(
                collection_name="agent_episodes",
                points=[point.id]
            )

# rag/doc_sync.py
import hashlib
from pathlib import Path

class DocumentSync:
    """Detects document changes and re-embeds only what changed."""
    
    def __init__(self, docs_dir: str, retriever: MultiStrategyRetriever):
        self.docs_dir = Path(docs_dir)
        self.retriever = retriever
        self._index: dict[str, str] = {}  # path -> sha256
        self._load_index()
    
    def _load_index(self):
        index_path = self.docs_dir / ".rag_index.json"
        if index_path.exists():
            self._index = json.loads(index_path.read_text())
    
    def _save_index(self):
        (self.docs_dir / ".rag_index.json").write_text(json.dumps(self._index))
    
    def scan_and_sync(self):
        changed = []
        for doc_path in self.docs_dir.glob("**/*.md"):
            content = doc_path.read_text()
            digest = hashlib.sha256(content.encode()).hexdigest()[:16]
            rel = str(doc_path.relative_to(self.docs_dir))
            
            if rel not in self._index or self._index[rel] != digest:
                changed.append(doc_path)
                self._index[rel] = digest
        
        self._save_index()
        return changed  # Re-embed only these