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

Originally published on tamiz.pro.

Most AI apps you'll ship in 2025–2026 aren't chatbots. They're agents — systems that make decisions, call tools, maintain state across turns, and fail in ways that span hundreds of milliseconds and dozens of LLM calls. The moment you cross from prototype to production, three things bite you: you can't see what your agent actually did, it forgets everything between requests, and your RAG retrieves garbage because you never built a proper retrieval pipeline.

This is a field guide for the engineers who already know how to wire together a LangChain chain and are now staring at a dashboard full of opaque latency numbers. We'll cover observability, memory, and RAG as a unified stack — not as isolated tutorials, but as the three pillars that hold up any serious AI application.

From Chatbot to Agent — What Actually Changes

A chatbot takes a message and returns a response. An agent loops: it perceives the user input, reasons over available tools, executes actions, observes results, and repeats until the task is complete or it gives up.

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]

That diagram is deceptively simple. Every arrow is a source of failure:

• Thought can hallucinate a tool that doesn't exist, call it with wrong arguments, or enter an infinite loop.
• Tool Call can fail silently — API timeout, auth error, malformed response — and the agent retries blindly.
• Observability Trace is where most teams have zero coverage. You ship, you watch, you panic.
• Memory Write either creates noise or loses critical context.

The core thesis: agent productionization is an engineering problem, not a prompting problem. Better prompts help. Structured observability, persistent memory, and grounded retrieval are what separate a demo from a shipped product.

1. Observability — See the Loop, Not Just the Output

The Myth of the Single Traced Request

With a stateless chatbot, one request equals one trace. With an agent, one user turn equals a tree of LLM calls, tool invocations, and conditional branches. Most teams measure "latency" as time from input to output — which is meaningless when the agent made 14 LLM calls and 3 tool invocations inside that window.

What you actually need to track:

Metric	Why It Matters
agent.turns	Number of loop iterations before answer or max-step break
agent.tools.called	Which tools fired, how many times, success/failure rate
agent.llm.calls	Per-call token usage, latency, and model selected
agent.loop.break_reason	success, max_steps, tool_error, hallucination_guard
user.cohort.retention	Are users coming back? (The ultimate signal)

Tracing That Actually Helps Debug

Here's a minimal but production-viable tracing layer using OpenTelemetry — the standard most observability platforms (Langfuse, Phoenix, Arize, SigNoz) understand natively:

# 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",
)

Then instrument the agent loop itself:

# 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."

Two things most teams get wrong here:

1. They trace the wrong granularity. A single span wrapping the whole agent is a black box. You need per-turn, per-tool, and per-LLM-call spans to debug whether the agent is stuck in a retry loop, calling the wrong tool, or being truncated by context limits.

2. They don't emit metrics alongside traces. Traces tell you what happened in one request. Metrics tell you what's happening across all requests. The agent.turns histogram is one of the highest-signal metrics you can ship — if your P99 turns is 7 and your max is 8, you have a systemic looping problem, not a bad prompt.

The Observability Stack Decision

Platform	Best For	Trade-off
Langfuse	Rapid iteration, open-source self-host	Less flexible for non-LangChain frameworks
Phoenix (Arize)	Deep visual trace analysis	Heavier infra, requires Arize account for cloud
OpenTelemetry + SigNoz	Full control, framework-agnostic	You own the plumbing
Datadog AI Observability	Enterprise, existing DD footprint	Expensive at scale, proprietary SDKs

For a greenfield agent product, I recommend starting with OpenTelemetry self-hosted (SigNoz or the OTLP collectors) and instrumenting manually as above. The moment you outgrow it, migrating to Langfuse or Phoenix is a configuration change, not a rewrite — because you're already speaking their language.

2. Memory — The Part Everyone Underestimates

The Three Layers Every Agent Needs

Memory in AI agents isn't one thing. It's three distinct systems that serve different purposes:

┌─────────────────────────────────────────────────────┐
│                 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   │
└─────────────────────────────────────────────────────┘

Short-Term: Conversation History with smart truncation

Naive approach: keep every message, chunk at the context limit. This fails because the most recent messages get the most weight in the LLM's attention, so truncating from the middle destroys coherence.

Better approach: recurved conversation summary — keep the full recent window and compress older turns into a summary:

# 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]

The key insight: compression is not optional at scale. A 10-turn conversation with tool outputs can easily exceed 50K tokens. Without summarization, you're either burning money on context or losing conversation coherence.

Medium-Term: Episodic Memory with Vector Search

Episodic memory answers: Has this user asked about this topic before? What tools did they use last time? What failed? Store recent interactions as embeddings and retrieve by similarity:

# 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]

Episodic memory has a critical performance consideration: write throughput vs. query freshness. In high-traffic agents, you don't want every turn blocking on an embedding call. Decouple writes:

# 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])

The agent loop never waits for memory writes. It enqueues and continues. Draining happens on a background task or at natural pause points (between turns, during idle periods).

Long-Term: Structured Profile + Semantic Storage

Long-term memory is fundamentally different. It's not about what happened recently — it's about who the user is and what they've established as true. This needs:

1. A structured profile store (PostgreSQL or similar) for hard facts: name, preferences, permissions, known constraints.
2. A vector store for semantic recall: past topics discussed, documents referenced, project contexts.
3. A conflict resolution layer: when new information contradicts old, the system must decide what to keep.

# 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()

The confidence field matters more than you'd think. An LLM can confidently state something wrong. When the agent later learns the user prefers "dark mode" but previously stored "light mode" with confidence 0.95, you need a merging strategy — not a blind overwrite.

3. Real RAG — Beyond the Tutorial Pipeline

Why Tutorial RAG Fails in Production

The canonical RAG tutorial does this:

1. Chunk your documents.
2. Embed each chunk.
3. Store in a vector DB.
4. On query: embed the question, retrieve top-k chunks, inject into prompt.

That works until:

• Your documents have tables, code blocks, and mixed formatting — chunks destroy structure.
• "Top-k" retrieves semantically similar but contextually irrelevant passages.
• Queries are vague and the retrieval returns noise.
• Your knowledge base is large (millions of chunks) and latency is unacceptable.
• Documents are updated and you're returning stale information.

Real RAG is not a retrieval step. It's a query understanding → routing → retrieval → re-ranking → synthesis pipeline.

The Production RAG Pipeline

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]

Step 1: Query Understanding (Not Just Embedding the Raw Query)

Raw queries are terrible retrieval signals. "How do I fix it?" means nothing without context. A query understanding layer decomposes and rewrites before retrieval:

# 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

Step 2: Multi-Strategy Retrieval

Don't rely on a single retrieval method. Use three in parallel and let the re-ranker choose:

# 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

Step 3: Re-Ranking — The Hidden Performance Multiplier

This is where most production RAG systems separate from the tutorials. Raw retrieval gives you 20 potentially relevant chunks. A re-ranker — a smaller, specialized model — scores them for actual relevance to this specific query:

# 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]

Re-ranking adds ~100-300ms of latency but typically improves retrieval quality by 20-40% (measured by pass@k against human-annotated relevance). That's the difference between an answer that feels right and one that makes the user trust the system.

Step 4: Context Selection and Prompt Assembly

Don't dump all retrieved chunks into the prompt. Select, order, and format:

# 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}"

Step 5: Evaluation — Because You Can't Improve What You Don't Measure

This is the step that separates shipped products from prototypes. You need an evaluation harness:

# 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
            )
        }

Run this evaluation weekly against a held-out test set. Track the metrics over time. If faithfulness drops while correctness holds steady, your retrieval is getting worse even though the LLM is compensating. That's a signal to improve your chunking or add more structured filters.

4. Tying It Together — The Production Agent Architecture

All three pillars converge in the agent loop. Here's the integrated architecture:

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

The agent orchestrator is the conductor. It decides:

• How much memory to inject (based on query complexity and user context)
• Whether to trigger RAG (or is this a conversational query?)
• Which tools are available and which to call
• When to break the loop
• What to persist to memory after completion

# 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

Common Pitfalls and How to Avoid Them

1. The Infinite Loop

Agents that don't terminate properly are the #1 production incident. Every agent loop must have:

• A hard step limit (configured per-agent, not global)
• A token budget (prompt + completion tokens per turn)
• A circuit breaker that fires when tool error rate exceeds a threshold in a sliding window

# 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

2. Memory Leaks (Literal and Figurative)

Episodic memory grows unbounded. Implement a retention policy:

• Auto-evict entries older than N days
• Compress low-confidence entries after 30 days
• Archive to cold storage (S3 + Parquet) after 90 days

# 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]
            )

3. RAG Drift

Your documents change. Your embeddings don't automatically reflect that. Set up a document change detection pipeline:

# 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

Run this on a cron or CI pipeline. Incremental re-embedding keeps your RAG fresh without full re-indexing costs.

When to Use What — Decision Framework

Scenario	Priority Focus	Why
Internal dev tool	Observability + RAG	Low user volume, high need for accuracy
Customer-facing chatbot	Memory + Observability	Scale, personalization, debugging at production volume
Data analysis agent	RAG + Memory	Grounding in private data, remembering past analyses
Autonomous workflow agent	Observability + Circuit breakers	Failure modes are expensive (API calls, money)
Prototype / MVP	Just get it working	Don't over-engineer. Add observability before you ship.

Frequently Asked Questions

Q: Do I really need all three layers of memory? My agent is small.

Even for small agents, short-term memory is non-negotiable — without it, every request is stateless and you lose conversation continuity. Episodic and profile memory are worth adding once you have more than ~100 active users or when you notice users repeating questions. Start simple; the architecture supports incremental addition.

Q: How do I measure RAG quality without manual annotation?

LLM-as-judge is the practical approach. Use a strong model (GPT-4o, Claude 3.5 Sonnet) to score faithfulness and correctness on a sample of your queries. Even 50–100 labeled examples give you a reliable baseline. The RAGEvaluator above implements this pattern. Re-run weekly and track the trend — absolute numbers matter less than direction.

Q: My agent is slow. Which pillar should I optimize first?

Check your traces first. If the bottleneck is LLM latency, that's a model choice issue. If it's retrieval latency, optimize your vector DB (use HNSW indexes, not brute-force). If it's the loop overhead, you likely have too many turns — check agent.turns histogram. In my experience, 80% of "slow agent" complaints trace to unoptimized RAG retrieval, not the LLM itself. Start there.

---

Building production AI agents isn't about chaining more LLM calls together. It's about building systems that can be observed, remembered, and grounded — so when things go wrong (and they will), you can find out why and fix them. The stack above is battle-tested across dozens of shipped agent products. Start with observability. Add memory when you need continuity. Invest in RAG when you need accuracy. The order matters — each layer makes the next one more valuable.

For deeper dives into production agent patterns, check out Tamiz's Insights where we publish field reports from shipped AI systems.