LLM Engineering in Production: Evaluation, Cost & Deployment

← Back to LLM Engineering & RAG course

Getting an LLM application to work in a notebook is straightforward. Getting it to work reliably in production — with acceptable latency, predictable cost, observable quality, and the ability to detect regressions — is an engineering discipline. This final lesson covers the four concerns that separate prototype LLM applications from production ones: evaluation, cost management, latency optimisation, and deployment.

Previous: Lesson 9 — LangChain Agents & Tools

Evaluation

Why Evaluation Is Hard for LLMs

Traditional ML evaluation has ground truth labels — the model either classified the image correctly or it didn't. LLM outputs are free-form text. "Is this a good answer?" requires a judge, and the most scalable judge is another LLM.

LLM-as-Judge Evaluation

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class EvalResult(BaseModel):
    score:    int    = Field(description="Score from 1 (poor) to 5 (excellent)")
    reasoning: str   = Field(description="One sentence explaining the score")
    passed:   bool   = Field(description="True if score >= 3")

EVAL_PROMPT = ChatPromptTemplate.from_template("""
You are an expert evaluator for question-answering systems.

Question: {question}
Reference Answer: {reference}
Candidate Answer: {candidate}

Evaluate the candidate answer on:
1. Factual accuracy vs the reference
2. Completeness
3. Conciseness (no unnecessary padding)

Return a JSON object with score (1-5), reasoning (one sentence), and passed (bool).
""")

evaluator = EVAL_PROMPT | ChatOpenAI(model="gpt-4o", temperature=0) | JsonOutputParser()

def evaluate(question: str, reference: str, candidate: str) -> dict:
    return evaluator.invoke({
        "question":  question,
        "reference": reference,
        "candidate": candidate
    })

result = evaluate(
    question="What is RAG?",
    reference="RAG (Retrieval-Augmented Generation) grounds LLM responses in retrieved documents from a knowledge base, enabling accurate answers over private or up-to-date data.",
    candidate="RAG is a technique where you fetch relevant documents and include them in the LLM prompt."
)
print(result)
# → {'score': 4, 'reasoning': 'Correct and concise but missing the private/up-to-date data motivation.', 'passed': True}

Ragas — RAG-Specific Metrics

Ragas is a framework for evaluating RAG pipelines on four dimensions without requiring hand-labelled reference answers:

bash

pip install ragas

python

from ragas import evaluate as ragas_evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall, context_precision
from datasets import Dataset

# Collect a test set of question + answer + retrieved contexts
test_data = {
    "question":         ["What is FAISS?", "How does chunk overlap work?"],
    "answer":           [rag_chain.invoke(q) for q in ["What is FAISS?", "How does chunk overlap work?"]],
    "contexts":         [[doc.page_content for doc in retriever.invoke(q)] for q in ["What is FAISS?", "How does chunk overlap work?"]],
    "ground_truth":     ["FAISS is an in-process vector similarity search library by Meta.", "Chunk overlap preserves context at chunk boundaries by repeating characters."]
}

dataset = Dataset.from_dict(test_data)
scores  = ragas_evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_recall, context_precision]
)
print(scores)

Building a Regression Test Suite

python

import json
from pathlib import Path
from datetime import datetime

TEST_CASES_PATH = Path("eval/test_cases.json")

def run_eval_suite(chain, test_cases: list[dict]) -> dict:
    results = []
    passed = 0

    for case in test_cases:
        candidate = chain.invoke(case["question"])
        result    = evaluate(case["question"], case["reference"], candidate)
        result["question"]  = case["question"]
        result["candidate"] = candidate
        results.append(result)
        if result["passed"]:
            passed += 1

    summary = {
        "timestamp": datetime.utcnow().isoformat(),
        "total":     len(test_cases),
        "passed":    passed,
        "pass_rate": round(passed / len(test_cases), 3),
        "results":   results
    }
    out_path = Path(f"eval/run_{datetime.utcnow():%Y%m%d_%H%M%S}.json")
    out_path.write_text(json.dumps(summary, indent=2))
    return summary

test_cases = json.loads(TEST_CASES_PATH.read_text())
summary    = run_eval_suite(rag_chain, test_cases)
print(f"Pass rate: {summary['pass_rate']:.0%} ({summary['passed']}/{summary['total']})")

Cost Management

Tracking Token Usage

python

from dataclasses import dataclass, field
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult

# GPT-4o-mini pricing (USD per 1M tokens)
PRICING = {
    "gpt-4o-mini":         {"input": 0.15, "output": 0.60},
    "gpt-4o":              {"input": 5.00, "output": 15.00},
    "text-embedding-3-small": {"input": 0.02, "output": 0.00},
}

@dataclass
class TokenUsageTracker(BaseCallbackHandler):
    model:         str = "gpt-4o-mini"
    input_tokens:  int = field(default=0)
    output_tokens: int = field(default=0)
    call_count:    int = field(default=0)

    def on_llm_end(self, response: LLMResult, **kwargs):
        usage = response.llm_output.get("token_usage", {})
        self.input_tokens  += usage.get("prompt_tokens", 0)
        self.output_tokens += usage.get("completion_tokens", 0)
        self.call_count    += 1

    @property
    def cost_usd(self) -> float:
        p = PRICING.get(self.model, {"input": 0, "output": 0})
        return (self.input_tokens / 1_000_000 * p["input"] +
                self.output_tokens / 1_000_000 * p["output"])

    def report(self) -> str:
        return (f"Calls: {self.call_count} | "
                f"Input: {self.input_tokens:,} | "
                f"Output: {self.output_tokens:,} | "
                f"Cost: ${self.cost_usd:.6f}")

# Attach the tracker as a callback
tracker = TokenUsageTracker(model="gpt-4o-mini")
llm     = ChatOpenAI(model="gpt-4o-mini", callbacks=[tracker])
chain   = prompt | llm | StrOutputParser()

chain.invoke({"question": "What is a transformer?"})
print(tracker.report())

Reducing Cost Without Reducing Quality

Three approaches that cut cost while maintaining answer quality:

python

# 1. Cache identical queries — identical prompts return cached responses
from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache

set_llm_cache(SQLiteCache(database_path=".langchain_cache.db"))
# Identical prompts now return cached responses instantly at zero cost

# 2. Use gpt-4o-mini for retrieval and formatting; gpt-4o only for final answers
retrieval_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
synthesis_llm = ChatOpenAI(model="gpt-4o",      temperature=0)

# 3. Reduce k — retrieve 3 chunks instead of 6
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})  # saves ~30% input tokens

Latency Optimisation

Profiling Your Chain

python

import time
from contextlib import contextmanager

@contextmanager
def timer(label: str):
    start = time.perf_counter()
    yield
    elapsed = (time.perf_counter() - start) * 1000
    print(f"{label}: {elapsed:.0f}ms")

with timer("retrieval"):
    docs = retriever.invoke("What is RAG?")

with timer("generation"):
    answer = chain.invoke("What is RAG?")

Typical breakdown for a gpt-4o-mini RAG call:

Embedding the query: 50–100ms
Vector search (Chroma, 10k docs): 5–20ms
LLM generation (200 token answer): 800–1500ms

The LLM is almost always the bottleneck. Streaming hides latency; caching eliminates it for repeated queries.

Async Parallel Processing

For bulk operations — indexing, batch evaluation, multi-document analysis — run LLM calls in parallel:

python

import asyncio
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

async def process_documents(docs: list[str]) -> list[str]:
    tasks = [
        chain.ainvoke({"question": f"Summarise: {doc}"})
        for doc in docs
    ]
    return await asyncio.gather(*tasks)

# Run 10 documents concurrently instead of sequentially
summaries = asyncio.run(process_documents(document_texts[:10]))

Deployment

FastAPI Application

A complete, production-ready FastAPI application exposing the RAG chain:

python

# app.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import create_retrieval_chain, create_history_aware_retriever
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.chat_history import InMemoryChatMessageHistory
import os

app = FastAPI(title="RAG API")

# Initialise the chain at startup
embeddings  = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
    collection_name="docs",
    embedding_function=embeddings,
    persist_directory=os.getenv("CHROMA_PATH", "./chroma_db")
)
llm       = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

ctx_prompt = ChatPromptTemplate.from_messages([
    ("system", "Reformulate the question to be standalone given history."),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])
qa_prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer using only the context.\n\nContext:\n{context}"),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

rag_chain = create_retrieval_chain(
    create_history_aware_retriever(llm, retriever, ctx_prompt),
    create_stuff_documents_chain(llm, qa_prompt)
)

sessions: dict[str, list] = {}

class ChatRequest(BaseModel):
    session_id: str
    question:   str

@app.post("/chat")
async def chat(req: ChatRequest):
    history = sessions.setdefault(req.session_id, [])
    result  = rag_chain.invoke({"input": req.question, "chat_history": history})
    history += [HumanMessage(content=req.question), AIMessage(content=result["answer"])]
    sources = list({d.metadata.get("source", "") for d in result.get("context", [])})
    return {"answer": result["answer"], "sources": sources}

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    history = sessions.setdefault(req.session_id, [])

    async def generate():
        full_answer = ""
        async for chunk in rag_chain.astream({"input": req.question, "chat_history": history}):
            if "answer" in chunk:
                full_answer += chunk["answer"]
                yield f"data: {chunk['answer']}\n\n"
        history += [HumanMessage(content=req.question), AIMessage(content=full_answer)]
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

@app.get("/health")
def health():
    return {"status": "ok"}

Dockerfile

dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

text

# requirements.txt
fastapi
uvicorn[standard]
langchain
langchain-openai
langchain-chroma
langchain-community
chromadb
tiktoken
python-dotenv

bash

# Build and run
docker build -t rag-api .
docker run -p 8000:8000 \
  -e OPENAI_API_KEY=$OPENAI_API_KEY \
  -e CHROMA_PATH=/data/chroma \
  -v $(pwd)/chroma_db:/data/chroma \
  rag-api

Production Checklist

Before going live with an LLM application, verify each of the following: