Llama 3.1 in the Enterprise: Self-Hosted LLM Deployment Architecture

# =============================================================================
# vLLM SERVER LAUNCH CONFIGURATION (run as shell command or Python subprocess)
# =============================================================================
# Llama 3.1 70B AWQ — single A100 80GB
# vllm serve meta-llama/Llama-3.1-70B-Instruct-AWQ \
#     --quantization awq \
#     --tensor-parallel-size 1 \
#     --max-model-len 65536 \
#     --gpu-memory-utilization 0.90 \
#     --max-num-batched-tokens 8192 \
#     --max-num-seqs 256 \
#     --enable-chunked-prefill \
#     --port 8000 \
#     --host 0.0.0.0
#
# Llama 3.1 70B BF16 — two A100 80GB with tensor parallelism
# vllm serve meta-llama/Llama-3.1-70B-Instruct \
#     --tensor-parallel-size 2 \
#     --max-model-len 131072 \
#     --gpu-memory-utilization 0.92 \
#     --max-num-batched-tokens 16384 \
#     --port 8000

# =============================================================================
# PRODUCTION INFERENCE CLIENT
# =============================================================================
import os
import time
import logging
from typing import Optional, Generator
from openai import OpenAI  # Works against vLLM's OpenAI-compatible API
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import APIError, APITimeoutError, RateLimitError

logger = logging.getLogger(__name__)


class LlamaInferenceClient:
    """
    Production inference client for self-hosted Llama 3.1 via vLLM.
    Uses the OpenAI-compatible API endpoint exposed by vLLM.
    Includes retry logic, timeout handling, and structured logging.
    """

    def __init__(
        self,
        base_url: str = "http://llama-inference-svc:8000/v1",
        model: str = "meta-llama/Llama-3.1-70B-Instruct-AWQ",
        timeout: float = 60.0,
        max_retries: int = 3
    ):
        self.model = model
        self.client = OpenAI(
            api_key="not-needed-for-vllm",  # vLLM does not require an API key by default
            base_url=base_url,
            timeout=timeout,
            max_retries=0  # We handle retries ourselves for better observability
        )
        self.max_retries = max_retries
        logger.info(f"LlamaInferenceClient initialised: model={model}, base_url={base_url}")

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=30),
        retry=retry_if_exception_type((APIError, APITimeoutError))
    )
    def complete(
        self,
        system_prompt: str,
        user_message: str,
        temperature: float = 0.0,
        max_tokens: int = 2048,
        top_p: float = 0.95,
        stop_sequences: Optional[list[str]] = None
    ) -> dict:
        """
        Single-turn completion with structured response and latency logging.
        Returns dict with: content, model, usage (prompt_tokens, completion_tokens, total_tokens), latency_ms
        """
        start_time = time.monotonic()

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_message}
        ]

        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            top_p=top_p,
            stop=stop_sequences
        )

        latency_ms = (time.monotonic() - start_time) * 1000
        content = response.choices[0].message.content

        result = {
            "content": content,
            "model": response.model,
            "usage": {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens
            },
            "latency_ms": round(latency_ms, 2),
            "finish_reason": response.choices[0].finish_reason
        }

        logger.info(
            f"inference_complete | model={self.model} | "
            f"prompt_tokens={result['usage']['prompt_tokens']} | "
            f"completion_tokens={result['usage']['completion_tokens']} | "
            f"latency_ms={result['latency_ms']} | "
            f"finish_reason={result['finish_reason']}"
        )

        if result["finish_reason"] == "length":
            logger.warning(f"Response truncated at max_tokens={max_tokens} — consider increasing limit")

        return result

    def stream_complete(
        self,
        system_prompt: str,
        user_message: str,
        temperature: float = 0.0,
        max_tokens: int = 2048
    ) -> Generator[str, None, None]:
        """Streaming completion — yields text chunks as they arrive."""
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_message}
        ]
        stream = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=True
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield delta


if __name__ == "__main__":
    client = LlamaInferenceClient()

    result = client.complete(
        system_prompt="You are a senior contract analyst. Extract key terms from contracts in structured JSON.",
        user_message="Extract the payment terms, liability cap, and termination clauses from this contract: [contract text]",
        temperature=0.0,
        max_tokens=1024
    )
    print(f"Response ({result['latency_ms']}ms, {result['usage']['total_tokens']} tokens):")
    print(result["content"])