</> Local LLM Engineering Crash Course

def estimate_model_ram(params_billion, bits_per_weight=4, context_tokens=8192):
    weights_gb = params_billion * 1_000_000_000 * bits_per_weight / 8 / 1e9
    kv_cache_gb = context_tokens / 8192 * params_billion * 0.10
    overhead_gb = 1.2
    return round(weights_gb + kv_cache_gb + overhead_gb, 1)

models = [
    ("small-support", 3, 4),
    ("daily-driver", 8, 4),
    ("reasoning-heavy", 14, 4),
    ("server-grade", 32, 4),
]

for name, params, bits in models:
    print(f"{name:16} {params:>2}B Q{bits}: ~{estimate_model_ram(params, bits)} GB RAM")

from openai import OpenAI

local_client = OpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama",
)

response = local_client.chat.completions.create(
    model="llama3.1:8b",
    messages=[
        {"role": "system", "content": "Válaszolj röviden, magyarul."},
        {"role": "user", "content": "Mi a local LLM legnagyobb előnye egy webshop support botnál?"},
    ],
)

print(response.choices[0].message.content)

quantization = {
    "Q8_0": {"quality": 0.98, "ram": "high", "use": "quality-sensitive eval"},
    "Q6_K": {"quality": 0.96, "ram": "medium-high", "use": "good laptop/server default"},
    "Q5_K_M": {"quality": 0.94, "ram": "medium", "use": "balanced daily driver"},
    "Q4_K_M": {"quality": 0.90, "ram": "low", "use": "laptop-friendly default"},
    "Q3_K_M": {"quality": 0.82, "ram": "very-low", "use": "only if memory constrained"},
}

for q, meta in quantization.items():
    print(f"{q:7} quality~{meta['quality']:.2f} ram={meta['ram']:11} use={meta['use']}")

from dataclasses import dataclass
from openai import OpenAI

@dataclass
class LLMConfig:
    name: str
    base_url: str
    api_key: str
    default_model: str
    timeout_s: int = 30

local_cfg = LLMConfig(
    name="local-ollama",
    base_url="http://localhost:11434/v1",
    api_key="ollama",
    default_model="llama3.1:8b",
)

client = OpenAI(base_url=local_cfg.base_url, api_key=local_cfg.api_key, timeout=local_cfg.timeout_s)
print(f"provider={local_cfg.name} model={local_cfg.default_model} timeout={local_cfg.timeout_s}s")

def webshop_support_prompt(question, context):
    return f"""
Te a WebShop Pro lokális support asszisztense vagy.
Csak a CONTEXT alapján válaszolj. Ha nincs benne elég információ, mondd ezt:
"Erre nincs elég információm a lokális tudásbázisban."

Válaszformátum:
- válasz: 2-4 mondat magyarul
- forrás: dokumentum neve vagy "nincs"
- bizonyosság: low | medium | high

CONTEXT:
{context}

USER QUESTION:
{question}
""".strip()

print(webshop_support_prompt("Mikor kapom meg a rendelésem?", "shipping_policy.md: Budapesten 2 munkanap."))

documents = [
    {"id": "shipping-1", "source": "shipping_policy.md", "text": "Budapesten a standard szállítás 2 munkanap, vidéken 3 munkanap."},
    {"id": "returns-1", "source": "return_policy.md", "text": "A vásárló 14 napon belül jelezheti az elállást sérülésmentes termékre."},
    {"id": "warranty-1", "source": "warranty.md", "text": "Elektronikai termékekre 24 hónap jótállás vonatkozik."},
]

chunks = []
for doc in documents:
    chunks.append({
        "id": doc["id"],
        "text": doc["text"],
        "metadata": {"source": doc["source"], "privacy": "public-support"},
    })

print(f"Prepared {len(chunks)} chunks for local embedding")

def simple_retrieve(question, chunks, top_k=2):
    terms = set(question.lower().split())
    scored = []
    for chunk in chunks:
        score = sum(1 for t in terms if t in chunk["text"].lower())
        scored.append((score, chunk))
    return [chunk for score, chunk in sorted(scored, reverse=True, key=lambda x: x[0])[:top_k]]

question = "Hány nap alatt érkezik meg Budapestre a rendelés?"
hits = simple_retrieve(question, chunks)
context = "\n".join(f"{h['metadata']['source']}: {h['text']}" for h in hits)
print(context)

allowed_tools = {
    "lookup_order": {"required": ["order_id"]},
    "search_policy": {"required": ["query"]},
}

model_plan = {
    "tool": "lookup_order",
    "arguments": {"order_id": "ORD-1042"},
    "reason": "A felhasználó konkrét rendelés állapotát kérdezi.",
}

def validate_tool_call(plan):
    if plan["tool"] not in allowed_tools:
        raise ValueError("Unknown tool")
    missing = set(allowed_tools[plan["tool"]]["required"]) - set(plan["arguments"])
    if missing:
        raise ValueError(f"Missing args: {missing}")
    return True

print(validate_tool_call(model_plan), model_plan["tool"], model_plan["arguments"])

bench = [
    {"model": "3b-q4", "ttft_ms": 280, "tok_s": 42, "ram_gb": 3.2, "quality": 0.74},
    {"model": "8b-q4", "ttft_ms": 520, "tok_s": 24, "ram_gb": 6.1, "quality": 0.86},
    {"model": "14b-q4", "ttft_ms": 950, "tok_s": 14, "ram_gb": 9.8, "quality": 0.91},
]

def score(row):
    latency_penalty = row["ttft_ms"] / 1000
    speed_bonus = row["tok_s"] / 50
    return round(row["quality"] + speed_bonus - latency_penalty, 2)

for row in sorted(bench, key=score, reverse=True):
    print(row["model"], "score=", score(row), "ttft=", row["ttft_ms"], "ms", "ram=", row["ram_gb"], "GB")

def route_request(question, contains_pii=False, complexity=0.3, user_tier="standard"):
    if contains_pii:
        return "local-private"
    if complexity > 0.75:
        return "cloud-frontier"
    if user_tier == "free":
        return "local-small"
    return "local-balanced"

cases = [
    ("Mi a visszaküldési határidő?", False, 0.2, "standard"),
    ("Elemezd ezt a panaszt és írj kompenzációs tervet.", True, 0.8, "premium"),
    ("Írj SQL-t a rendelési kohorszokra.", False, 0.85, "premium"),
]

for question, pii, complexity, tier in cases:
    print(route_request(question, pii, complexity, tier), "-", question[:42])

# Lab indítása a repo gyökeréből
docker compose --profile local-llm up -d

# Hoston futó Ollama ellenőrzése
curl http://localhost:11434/api/tags

# Gateway dokumentáció
open http://localhost:8000/docs

eval_results = [
    {"provider": "local-8b", "grounded": 0.86, "policy": 0.94, "latency_ms": 1450, "cost": 0.00},
    {"provider": "cloud-frontier", "grounded": 0.94, "policy": 0.97, "latency_ms": 2100, "cost": 0.018},
]

for r in eval_results:
    pass_mark = r["grounded"] >= 0.85 and r["policy"] >= 0.90 and r["latency_ms"] < 2500
    print(f"{r['provider']:15} pass={pass_mark} grounded={r['grounded']} latency={r['latency_ms']}ms cost_units={r['cost']}")