Building a Production LLM System End-to-End: From Prompt to Retrieval, Tools, Evaluation, and Guardrails

One of the biggest gaps in LLM education is that most content explains individual concepts in isolation. Prompting is explained in one place, retrieval in another, evaluation in another, and observability somewhere else. Real systems do not work that way. In production, all of these layers interact.

This article shows how to think about an end-to-end LLM system as a composed architecture rather than a single model call. The example use case is a support assistant for internal operations: it answers policy questions, retrieves current documentation, calls tools when needed, logs traces, and applies guardrails before any sensitive action.

1. The Problem Statement

Assume we need to build an internal assistant that can:

• answer HR and IT policy questions
• retrieve current internal documentation
• look up employee account status through a tool
• refuse unauthorized or unsafe requests
• log enough information for debugging and review

This is not a pure chat problem. It is a system design problem with an LLM inside it.

2. The High-Level Architecture

A realistic request path often looks like this:

flowchart LR
  A[User Request] --> B[Request Classifier]
  B --> C[Retriever]
  C --> D[Prompt Builder]
  D --> E[LLM]
  E --> F{Tool Needed?}
  F -->|No| G[Structured Answer]
  F -->|Yes| H[Tool Validation]
  H --> I[Tool Execution]
  I --> J[LLM Synthesis]
  J --> G
  G --> K[Policy Validation]
  K --> L[Trace + Metrics]
  L --> M[Final Response]

This architecture is already enough to illustrate a core lesson: quality does not come from the model alone. It comes from how the system routes, constrains, validates, and observes the model.

3. Layer 1: Request Classification

The first question is often not “what is the answer?” but “what kind of request is this?”

Typical classes might include:

• answerable from documentation alone
• requires a live tool call
• sensitive and requires extra validation
• disallowed or out of scope

A simple classifier can reduce downstream cost and improve safety by selecting the right path early.

def classify_request(question, llm):
    prompt = f"""
    Classify this request into one of:
    - docs_only
    - tool_required
    - sensitive
    - disallowed

    Request: {question}
    Return only the label.
    """
    return llm.invoke(prompt).strip()

In a mature system, this classifier itself should be evaluated and monitored rather than assumed correct.

4. Layer 2: Retrieval

If the question depends on internal policy, the system should retrieve current documentation rather than relying on model memory.

def retrieve_context(question, retriever):
    docs = retriever.search(question, top_k=5)
    return docs

The retriever is only useful if:

• the corpus is current
• chunking preserves meaning
• metadata filters are correct
• retrieval quality has been measured

This is why RAG systems should evaluate retrieval independently before judging answer quality.

5. Layer 3: Prompt Building

Prompt construction should combine:

• task instructions
• retrieved evidence
• output schema expectations
• policy boundaries

def build_prompt(question, docs):
    context = "\n\n".join(doc["text"] for doc in docs)
    return f"""
    You are an internal operations assistant.
    Use only the provided policy context.
    If the context is insufficient, say that more information is required.
    If a live lookup is needed, indicate tool_required.

    Context:
    {context}

    Question:
    {question}
    """

This is where prompt engineering becomes system engineering. The prompt is not only an instruction. It is a policy boundary and output contract.

6. Layer 4: Tool Calling

If the model determines that a live account lookup is necessary, it should not execute anything directly. It should propose a structured tool call.

def propose_tool_call(question, llm):
    prompt = f"""
    If the request needs an account lookup, return JSON with:
    tool
    Otherwise return tool.

    Request: {question}
    """
    return llm.invoke(prompt)

Then the application validates before execution:

def run_tool(tool_call, user_context):
    validate_schema(tool_call)
    authorize(tool_call, user_context)
    return execute_tool(tool_call)

This separation is non-negotiable in production systems.

7. Layer 5: Final Synthesis

Once the system has either retrieved documentation or executed tools, the final response should be synthesized in a controlled format.

def synthesize_answer(question, docs, tool_result, llm):
    context = "\n\n".join(doc["text"] for doc in docs)
    return llm.invoke(f"""
    Answer the question using the policy context and tool result.
    Be concise, factual, and do not invent unsupported claims.

    Context:
    {context}

    Tool result:
    {tool_result}

    Question:
    {question}
    """)

A professional system often adds structured output requirements or policy assertions here as well.

8. Layer 6: Guardrails and Policy Validation

Before returning the answer or allowing a side effect, the system should validate whether:

• the answer exposes unauthorized data
• the tool path was permitted
• the result matches the user’s authorization scope
• the request should have been escalated instead

In high-risk systems, guardrails should live outside the model as deterministic checks where possible.

9. Layer 7: Observability

Every major step should emit trace data:

• prompt version
• retrieved document IDs
• tool calls
• validation status
• latency
• token usage
• final route chosen

def trace_event(request_id, payload, logger):
    logger.write({
        "request_id": request_id,
        **payload
    })

Observability is what makes postmortem analysis possible when the system fails.

10. Layer 8: Evaluation

This system should be evaluated on more than answer quality. At minimum:

• retrieval quality
• classifier accuracy
• tool-call validity
• groundedness of final answers
• refusal quality
• cost and latency per request

If one of these layers is weak, the user experience may fail even when the model itself is strong.

11. Putting It Together

A simplified orchestration loop might look like this:

def handle_request(question, retriever, llm, user_context, logger):
    route = classify_request(question, llm)
    trace_event("req_1", {"route": route}, logger)

    if route == "disallowed":
        return "I cannot help with that request."

    docs = retrieve_context(question, retriever)
    prompt = build_prompt(question, docs)
    draft = llm.invoke(prompt)

    if route == "tool_required":
        tool_call = propose_tool_call(question, llm)
        tool_result = run_tool(tool_call, user_context)
        answer = synthesize_answer(question, docs, tool_result, llm)
    else:
        answer = draft

    enforce_policy(answer, user_context)
    trace_event("req_1", {"answer": answer}, logger)
    return answer

This is still simplified, but it is already much closer to a real product architecture than a standalone prompt.

12. Common Failure Modes

An end-to-end system like this typically fails through composition mistakes:

• the classifier sends the request to the wrong route
• retrieval finds weak context
• prompt construction buries critical evidence
• tool calls are syntactically valid but semantically wrong
• guardrails run too late
• observability is too weak to diagnose failures

These are exactly the kinds of issues professionals learn to expect.

13. Why This Matters for Mastery

Real LLM mastery is not only knowing what a transformer is or when to use RAG. It is knowing how all the layers fit together in a system that must work under cost, latency, safety, and reliability constraints.

That is the shift from “knowing concepts” to “engineering systems.”

14. Layer 9: Request Classification — Production Implementation

from openai import OpenAI
from pydantic import BaseModel
from typing import Literal
import asyncio

client = OpenAI()

class RequestClassification(BaseModel):
    route:       Literal["docs_only", "tool_required", "sensitive", "disallowed"]
    confidence:  Literal["low", "medium", "high"]
    reason:      str

async def classify_request_structured(question: str) -> RequestClassification:
    response = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[{
            "role": "system",
            "content": """Classify user requests for an internal HR/IT assistant.

Routes:
- docs_only: answerable from documentation alone, no live data needed
- tool_required: needs live system lookup (account status, active tickets, current HR data)
- sensitive: touches PII, disciplinary records, salaries, or medical data
- disallowed: offensive, irrelevant to HR/IT, or out of scope

Be conservative: when in doubt, classify as 'sensitive'.""",
        }, {
            "role": "user",
            "content": f"Classify this request: {question}",
        }],
        response_format=RequestClassification,
    )
    return response.choices[0].message.parsed

---

15. Full Retrieval Layer with Evaluation

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from sentence_transformers import CrossEncoder
import numpy as np

class ProductionRetriever:
    def __init__(self, docs, embedding_model="text-embedding-3-large"):
        embeddings   = OpenAIEmbeddings(model=embedding_model)
        vectorstore  = Chroma.from_documents(docs, embeddings)
        bm25         = BM25Retriever.from_documents(docs, k=10)
        dense        = vectorstore.as_retriever(search_kwargs={"k": 10})
        self.hybrid  = EnsembleRetriever(
            retrievers=[bm25, dense],
            weights=[0.3, 0.7],
        )
        self.reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

    def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        # Step 1: Hybrid retrieval
        candidates = self.hybrid.invoke(query)

        # Step 2: Rerank with cross-encoder
        pairs  = [(query, doc.page_content) for doc in candidates]
        scores = self.reranker.predict(pairs)

        ranked = sorted(
            zip(candidates, scores),
            key=lambda x: x[1],
            reverse=True,
        )[:top_k]

        return [
            {
                "text":   doc.page_content,
                "score":  float(score),
                "source": doc.metadata.get("source", "unknown"),
                "id":     doc.metadata.get("id", ""),
            }
            for doc, score in ranked
        ]

    def compute_retrieval_metrics(
        self,
        queries:   list[str],
        relevant:  list[list[str]],  # relevant doc IDs per query
        k:         int = 5,
    ) -> dict:
        """Compute Recall@k and MRR for retrieval quality."""
        recalls, mrr_scores = [], []
        for query, rel_ids in zip(queries, relevant):
            results   = self.retrieve(query, top_k=k)
            found_ids = [r["id"] for r in results]
            recalls.append(
                len(set(found_ids) & set(rel_ids)) / max(len(rel_ids), 1)
            )
            for rank, id_ in enumerate(found_ids, 1):
                if id_ in rel_ids:
                    mrr_scores.append(1.0 / rank)
                    break
            else:
                mrr_scores.append(0.0)

        return {
            "recall_at_k": round(np.mean(recalls), 4),
            "mrr":         round(np.mean(mrr_scores), 4),
            "k":           k,
        }

---

16. Prompt Construction — Professional Template

from jinja2 import Template
from datetime import date

SYSTEM_TEMPLATE = Template("""
You are an internal operations assistant for .

Capabilities:
- Answer questions about HR and IT policies using the provided documentation
- Look up live employee account information when authorized
- Refuse requests outside your scope or permissions

Guidelines:
1. Base answers ONLY on the provided context documents
2. If the context is insufficient, say clearly: "I don't have enough information"
3. Never reveal information about other employees without explicit authorization
4. Today's date is 

Scope: 
""")

USER_TEMPLATE = Template("""
Context Documents:

---

Question: 

Instructions:
- Use only the context above to answer
- Cite document IDs for claims: e.g., "According to [DOC_001]..."
- If information is missing from the context, say so explicitly
""")

def build_production_prompt(
    question:     str,
    docs:         list[dict],
    company_name: str = "AcmeCorp",
    scope:        str = "HR and IT policies",
) -> list[dict]:
    system_msg = SYSTEM_TEMPLATE.render(
        company_name=company_name,
        today=date.today().isoformat(),
        scope=scope,
    )
    user_msg = USER_TEMPLATE.render(
        question=question,
        docs=[{
            "id":     f"DOC_{i+1:03d}",
            "source": d.get("source", "internal"),
            "score":  d.get("score", 0.0),
            "text":   d["text"][:500],  # truncate for context budget
        } for i, d in enumerate(docs)],
    )
    return [
        {"role": "system", "content": system_msg},
        {"role": "user",   "content": user_msg},
    ]

---

17. Structured Tool Calling

from openai import OpenAI
from pydantic import BaseModel
from typing import Optional
import json

client = OpenAI()

# Define tool schemas
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "lookup_account",
            "description": "Look up an employee account status in the HR system.",
            "parameters": {
                "type": "object",
                "properties": {
                    "employee_id": {
                        "type":        "string",
                        "description": "The employee ID (e.g., EMP12345)",
                        "pattern":     "^EMP\\d{5}$",
                    },
                    "fields": {
                        "type":  "array",
                        "items": {"type": "string"},
                        "enum":  ["name", "department", "status", "role"],
                        "description": "Which fields to retrieve (exclude salary/medical)",
                    },
                },
                "required": ["employee_id"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_policy",
            "description": "Search the policy knowledge base for relevant documents.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                    "policy_area": {
                        "type": "string",
                        "enum": ["hr", "it", "security", "finance", "general"],
                    },
                },
                "required": ["query"],
            },
        },
    },
]

class ToolCallResult(BaseModel):
    tool_name:   str
    arguments:   dict
    result:      Optional[str]
    success:     bool
    error:       Optional[str]

def propose_and_execute_tool(
    question:     str,
    user_context: dict,
    messages:     list[dict],
) -> tuple[str, list[ToolCallResult]]:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=TOOLS,
        tool_choice="auto",
    )

    msg          = response.choices[0].message
    tool_results = []

    if msg.tool_calls:
        for tc in msg.tool_calls:
            args   = json.loads(tc.function.arguments)
            result = ToolCallResult(
                tool_name=tc.function.name,
                arguments=args,
                result=None,
                success=False,
                error=None,
            )
            try:
                # Validate before execution
                if tc.function.name == "lookup_account":
                    if not user_context.get("can_lookup_accounts"):
                        raise PermissionError("Not authorized for account lookups")
                    result.result  = f"Account data for {args['employee_id']}: status=Active"
                    result.success = True
                elif tc.function.name == "search_policy":
                    result.result  = f"Policy results for '{args['query']}': [policy content]"
                    result.success = True
                else:
                    raise ValueError(f"Unknown tool: {tc.function.name}")
            except PermissionError as e:
                result.error = f"permission_denied: {e}"
            except Exception as e:
                result.error = f"tool_error: {e}"

            tool_results.append(result)

    return msg.content or "", tool_results

---

18. End-to-End Request Handler

import time
import uuid
from datetime import datetime

class ProductionLLMSystem:
    def __init__(self, retriever, llm_client, logger, guardrails, rate_limiter):
        self.retriever    = retriever
        self.llm          = llm_client
        self.logger       = logger
        self.guardrails   = guardrails
        self.rate_limiter = rate_limiter

    async def handle(self, question: str, user_id: str, user_context: dict) -> dict:
        request_id = str(uuid.uuid4())
        start_time = time.perf_counter()
        trace      = {"request_id": request_id, "user_id": user_id, "timestamp": datetime.utcnow().isoformat()}

        # Rate limiting
        rate_check = self.rate_limiter.check(user_id)
        if not rate_check["allowed"]:
            return {"error": "rate_limit", "retry_after": rate_check["retry_after"]}

        # Input guardrails
        safety = self.guardrails.check_input(question)
        trace["safety_check"] = safety
        if not safety["safe"]:
            self.logger.write({**trace, "outcome": "refused_input"})
            return {"answer": "I'm sorry, I cannot help with that request."}

        # Request classification
        classification = await classify_request_structured(question)
        trace["route"]  = classification.route

        if classification.route == "disallowed":
            self.logger.write({**trace, "outcome": "refused_disallowed"})
            return {"answer": "This request is outside the scope of this assistant."}

        # Retrieval
        docs = self.retriever.retrieve(question, top_k=5)
        trace["retrieval"] = {"doc_count": len(docs), "top_score": docs[0]["score"] if docs else 0}

        # Build messages
        messages = build_production_prompt(question, docs)

        # Tool calling if needed
        tool_results = []
        if classification.route == "tool_required":
            _, tool_results = propose_and_execute_tool(question, user_context, messages)
            trace["tools"] = [{"name": t.tool_name, "success": t.success} for t in tool_results]

        # Generate answer
        answer_messages = messages.copy()
        if tool_results:
            tool_summary = "\n".join(
                f"Tool {t.tool_name}: {t.result or t.error}"
                for t in tool_results
            )
            answer_messages.append({
                "role":    "user",
                "content": f"Tool results:\n{tool_summary}\n\nNow answer: {question}",
            })

        response = self.llm.invoke(answer_messages)
        answer   = response.content

        # Output guardrails
        output_check = self.guardrails.check_output(answer, user_context)
        if not output_check["safe"]:
            answer = "I've encountered an issue generating a safe response. Please contact your HR representative."
            trace["output_safety_triggered"] = True

        # Observability
        trace.update({
            "latency_ms":        round((time.perf_counter() - start_time) * 1000, 1),
            "prompt_tokens":     response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "outcome":           "success",
        })
        self.logger.write(trace)

        return {
            "answer":     answer,
            "sources":    [d["source"] for d in docs],
            "request_id": request_id,
        }

---

19. Deployment Configuration

# docker-compose.yml for production LLM system
version: "3.9"

services:
  llm-api:
    build: .
    environment:
      OPENAI_API_KEY:         ${OPENAI_API_KEY}
      LANGCHAIN_TRACING_V2:   "true"
      LANGCHAIN_API_KEY:      ${LANGSMITH_API_KEY}
      LANGCHAIN_PROJECT:      "production-support-assistant"
      REDIS_URL:              "redis://redis:6379"
      CHROMA_HOST:            "chromadb"
      LOG_LEVEL:              "INFO"
    ports:
      - "8000:8000"
    depends_on:
      - chromadb
      - redis
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout:  10s
      retries:  3

  chromadb:
    image: chromadb/chroma:latest
    volumes:
      - chroma_data:/chroma/chroma
    ports:
      - "8001:8000"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  chroma_data:

---

20. Load Testing

import asyncio
import aiohttp
import time
from statistics import mean, quantiles

TEST_QUERIES = [
    "What is the PTO policy?",
    "How do I submit a support ticket?",
    "What are the remote work guidelines?",
    "How long is the parental leave?",
    "How do I reset my corporate VPN?",
]

async def load_test(
    url:         str,
    concurrency: int = 20,
    n_requests:  int = 100,
) -> dict:
    latencies = []
    errors    = 0

    async def single_request(session):
        import random
        payload = {
            "question": random.choice(TEST_QUERIES),
            "user_id":  "load_test_user",
        }
        t0 = time.perf_counter()
        try:
            async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=30)) as resp:
                await resp.json()
                latencies.append((time.perf_counter() - t0) * 1000)
        except Exception:
            nonlocal errors
            errors += 1

    async with aiohttp.ClientSession() as session:
        semaphore = asyncio.Semaphore(concurrency)
        async def bounded(s): 
            async with semaphore: 
                await single_request(s)
        await asyncio.gather(*[bounded(session) for _ in range(n_requests)])

    if latencies:
        qs = quantiles(latencies, n=100)
        return {
            "n_requests":    n_requests,
            "n_errors":      errors,
            "error_rate":    f"{errors/n_requests*100:.1f}%",
            "p50_ms":        round(qs[49]),
            "p95_ms":        round(qs[94]),
            "p99_ms":        round(qs[98]),
            "mean_ms":       round(mean(latencies)),
        }
    return {"error": "No successful requests"}

# Run: asyncio.run(load_test("http://localhost:8000/ask", concurrency=20, n_requests=200))

---

System Reliability Patterns

Circuit Breaker for LLM Calls

import time
from enum import Enum
from threading import Lock

class CircuitState(Enum):
    CLOSED   = "closed"    # normal operation
    OPEN     = "open"      # failing, rejecting requests
    HALF_OPEN = "half_open"  # testing recovery

class LLMCircuitBreaker:
    def __init__(
        self,
        failure_threshold: int   = 5,
        recovery_timeout:  float = 60.0,
        success_threshold: int   = 2,
    ):
        self.failure_threshold  = failure_threshold
        self.recovery_timeout   = recovery_timeout
        self.success_threshold  = success_threshold
        self.state              = CircuitState.CLOSED
        self.failure_count      = 0
        self.success_count      = 0
        self.last_failure_time  = 0.0
        self._lock              = Lock()

    def call(self, llm_fn, *args, **kwargs):
        with self._lock:
            if self.state == CircuitState.OPEN:
                if time.time() - self.last_failure_time >= self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.success_count = 0
                else:
                    raise RuntimeError("Circuit breaker OPEN — LLM service unavailable")

        try:
            result = llm_fn(*args, **kwargs)
            with self._lock:
                self.failure_count = 0
                if self.state == CircuitState.HALF_OPEN:
                    self.success_count += 1
                    if self.success_count >= self.success_threshold:
                        self.state = CircuitState.CLOSED
            return result

        except Exception as e:
            with self._lock:
                self.failure_count    += 1
                self.last_failure_time = time.time()
                if self.failure_count >= self.failure_threshold:
                    self.state = CircuitState.OPEN
            raise

circuit_breaker = LLMCircuitBreaker(failure_threshold=5, recovery_timeout=60)

def resilient_llm_call(prompt: str, llm) -> str:
    return circuit_breaker.call(lambda: llm.invoke(prompt).content)

Graceful Degradation

def invoke_with_fallback(
    prompt:    str,
    primary:   object,
    fallbacks: list,
    logger     = None,
) -> str:
    """Try primary model, fall back through alternatives on failure."""
    for i, model in enumerate([primary] + fallbacks):
        try:
            result = model.invoke(prompt).content
            if i > 0 and logger:
                logger.warning(f"Used fallback model #{i} due to primary failure")
            return result
        except Exception as e:
            if i == len(fallbacks):
                raise RuntimeError(f"All models failed. Last error: {e}")
            continue

    return ""

---

Production Incident Response

Runbook: LLM System Degraded

## Incident: LLM Response Quality Degraded

### Detection
- Faithfulness score drops below 0.75 for > 15 minutes
- User-reported "wrong answer" rate exceeds 2%
- LLM-as-judge passing rate drops below 70%

### Investigation Steps

1. Check recent deployments:
   - Any prompt version changes in the last 24h?
   - Model version bump or provider update?
   - Corpus or retrieval index updated?

2. Inspect traces in LangSmith/Langfuse:
   - Filter by failure flag = True
   - Check if retrieval scores dropped
   - Identify pattern: specific query type, domain, or time period?

3. Check retrieval layer independently:
   - Run retrieval eval on golden query set
   - Compare top-k scores before and after incident

4. Check serving health:
   - Latency p95 (high latency → timeout → partial responses)
   - Error rate on LLM provider API
   - Queue depth (backing up → delayed responses)

### Remediation

- If prompt change: rollback to previous version
- If retrieval degraded: check corpus update, rebuild index
- If model provider issue: activate fallback model
- If latency: scale serving replicas

---

Conclusion

A production LLM system is a layered architecture, not a model prompt with extra glue. Every layer — classification, retrieval, prompt construction, tool calling, guardrails, observability, and evaluation — participates in the final quality of the product. The system described here is representative of what real engineering teams build: structured request routing, hybrid retrieval with reranking, Jinja-templated prompts with citation anchoring, validated tool execution with permission checks, privacy-compliant structured logging, automated quality gates in CI/CD, circuit breakers for resilience, and incident runbooks for operational maturity. Engineers who understand these interactions build systems that are controllable and trustworthy. Engineers who focus only on the model usually end up debugging failures caused by everything around it.

---

Layer Summary: Production LLM System Checklist

Layer	Component	Key Practice
Input	Request classifier	Evaluate classifier accuracy independently
Retrieval	Hybrid + reranker	Measure Recall@5 and MRR on golden queries
Prompt	Jinja template	Version-control every template change
Tool calling	Validated tool executor	Never execute without authorization check
Output	Schema validation	Track conformance rate as SLO metric
Guardrails	Input/output filters	Red-team quarterly; canary test per deploy
Observability	Structured traces	Emit trace per request; never log raw PII
Evaluation	RAGAS + LLM judge	Regression test gates block every deploy
Operations	Circuit breaker + runbook	Test failure path quarterly

---

Prompt Version Control

from pathlib import Path
import json
from datetime import datetime

class PromptRegistry:
    """Version-controlled prompt storage with rollback capability."""

    def __init__(self, registry_path: str = "./prompts"):
        self.base = Path(registry_path)
        self.base.mkdir(exist_ok=True)

    def save(self, name: str, template: str, metadata: dict = None) -> str:
        version = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
        record  = {
            "name":       name,
            "version":    version,
            "template":   template,
            "metadata":   metadata or {},
            "created_at": datetime.utcnow().isoformat(),
        }
        path = self.base / f"{name}_{version}.json"
        path.write_text(json.dumps(record, indent=2))
        # Update current symlink
        current = self.base / f"{name}_current.json"
        current.write_text(json.dumps(record, indent=2))
        return version

    def load(self, name: str, version: str = "current") -> dict:
        path = self.base / f"{name}_{version}.json"
        return json.loads(path.read_text())

    def rollback(self, name: str, target_version: str):
        record = self.load(name, target_version)
        self.save(name, record["template"], {"rollback_from": target_version})

registry = PromptRegistry()
v = registry.save(
    "hr_assistant_v3",
    "You are an HR assistant...\n{context}\n{question}",
    {"author": "team", "eval_score": 0.87},
)
print(f"Saved prompt version: {v}")

---

System Engineering Principles for LLM Systems

1. Fail loudly: Surface errors to monitoring immediately rather than returning degraded silent responses
2. Separate concerns: Classification, retrieval, generation, and validation are each independently testable units
3. Design for observability first: Emit trace before you optimize performance
4. Default to least privilege: Tools, retrieval, and model calls should have minimum required permissions
5. Measure before you optimize: Profile latency and cost by component before refactoring
6. Version everything: Prompts, schemas, retrieval configs, and model versions all need change tracking
7. Test failure paths: Circuit breakers, fallbacks, and guardrails must be exercised in CI
8. Treat eval as a product: The evaluation system is as important as the application itself

---

End-to-End Conclusion

Building a production LLM system is a software engineering discipline, not a prompting exercise. The eight layers—classification, retrieval, prompt construction, tool calling, output validation, guardrails, observability, and evaluation—each require careful design, testing, and monitoring. Teams that master this architecture build systems that are controllable, auditable, and improvable over time. Teams that treat an LLM as a magic box surrounded by duct-tape glue code build fragile systems that are impossible to debug or improve. The architecture in this article is where serious LLM engineering begins.

Recommended Learning Path

For engineers looking to deepen their production LLM expertise:

1. Master RAG: Build a production RAG system with hybrid retrieval, reranking, and RAGAS evaluation
2. Add observability: Integrate LangSmith tracing and cost monitoring from day one
3. Build guardrails: Implement input moderation, output validation, and access control
4. Evaluate rigorously: Set up CI/CD quality gates with automated regression testing
5. Optimize serving: Profile latency, implement caching, and deploy with vLLM
6. Build agentic flows: Add tool calling and multi-step reasoning with LangGraph
7. Scale responsibly: Add circuit breakers, fallbacks, and operational runbooks

Each step builds on the previous one. The full stack described in this article represents approximately 6-12 months of progressive learning for an experienced software engineer transitioning into LLM systems.