Python skripta koja komunicira sa lokalnim modelom

Dosad si sa modelom pričao kroz curl. To je dobro za razumevanje i za jednokratne provere, ali čim treba obraditi listu fajlova, ponoviti neuspeo zahtev ili proslediti izlaz dalje, bash postaje tesan.

Ovaj tekst gradi Python klijent — od najmanjeg primera do alatke koju stvarno možeš staviti u /usr/local/bin.

Priprema

Zvanična openai biblioteka radi sa svakim OpenAI-kompatibilnim servisom, pa i sa tvojim:

$ python3 -m venv ~/.venv/ai
$ source ~/.venv/ai/bin/activate
$ pip install openai

Sistemski Python nemoj dirati; na novijim distribucijama pip to ionako odbija.

Podešavanja idu u okruženje, ne u kod:

$ cat ~/.ai_env
export AI_BASE_URL=http://localhost:11434/v1
export AI_API_KEY=k_alen_3f8a2b91c7d45e6f
export AI_MODEL=qwen3:8b
$ chmod 600 ~/.ai_env
$ source ~/.ai_env

Ovim ista skripta radi i sa lokalnom Ollamom i sa vLLM-om iza proxy-ja, bez ijedne izmene.

Najmanji klijent

#!/usr/bin/env python3
import os
from openai import OpenAI

klijent = OpenAI(
    base_url=os.environ["AI_BASE_URL"],
    api_key=os.environ.get("AI_API_KEY", "nije-bitno"),
)

odgovor = klijent.chat.completions.create(
    model=os.environ["AI_MODEL"],
    messages=[
        {"role": "system", "content": "Odgovaraj kratko, na srpskom."},
        {"role": "user", "content": "Šta radi komanda ss -tulpn?"},
    ],
    temperature=0.2,
    max_tokens=256,
)

print(odgovor.choices[0].message.content)
$ python3 klijent.py
Prikazuje otvorene portove sa procesima koji ih drže.

Biblioteka očekuje ključ i kad ga servis ne traži, pa Ollami daš bilo šta. Struktura odgovora je ista kao u tekstu o API-ju, samo pristupaš atributima umesto poljima u JSON-u.

Potrošnja tokena je tu:

print(f"{odgovor.usage.prompt_tokens} + {odgovor.usage.completion_tokens} tokena")

Bez biblioteke

Ako ne želiš zavisnost, dovoljna je requests ili čak standardna biblioteka:

import json, urllib.request

def pitaj(tekst, model="qwen3:8b"):
    podaci = json.dumps({
        "model": model,
        "messages": [{"role": "user", "content": tekst}],
        "temperature": 0.2,
    }).encode()

    zahtev = urllib.request.Request(
        "http://localhost:11434/v1/chat/completions",
        data=podaci,
        headers={"Content-Type": "application/json"},
    )

    with urllib.request.urlopen(zahtev, timeout=300) as veza:
        return json.load(veza)["choices"][0]["message"]["content"]

Za skriptu koja ide na server bez virtuelnog okruženja ovo je često praktičnije. Rad sa JSON-om u Pythonu pokriva poseban tekst.

Streaming

tok = klijent.chat.completions.create(
    model=os.environ["AI_MODEL"],
    messages=[{"role": "user", "content": "Objasni šta je inode"}],
    stream=True,
)

for deo in tok:
    komad = deo.choices[0].delta.content
    if komad:
        print(komad, end="", flush=True)
print()

Provera if komad nije višak — prvi i poslednji komad u nizu često nemaju sadržaj, pa bez nje dobijaš None u ispisu.

Za obradu u pozadini streaming ne koristi. Ima smisla samo kad neko gleda u ekran; inače je jednostavniji oblik i pouzdaniji i kraći.

Razgovor sa istorijom

Server ne pamti ništa, pa istoriju vodiš sam:

class Razgovor:
    def __init__(self, klijent, model, sistemski=None, max_poruka=20):
        self.klijent = klijent
        self.model = model
        self.max_poruka = max_poruka
        self.poruke = []
        if sistemski:
            self.poruke.append({"role": "system", "content": sistemski})

    def pitaj(self, tekst, **opcije):
        self.poruke.append({"role": "user", "content": tekst})

        odgovor = self.klijent.chat.completions.create(
            model=self.model,
            messages=self.poruke,
            **opcije,
        )

        sadrzaj = odgovor.choices[0].message.content
        self.poruke.append({"role": "assistant", "content": sadrzaj})
        self._skrati()
        return sadrzaj

    def _skrati(self):
        sistemske = [p for p in self.poruke if p["role"] == "system"]
        ostale = [p for p in self.poruke if p["role"] != "system"]
        self.poruke = sistemske + ostale[-self.max_poruka:]
razgovor = Razgovor(klijent, os.environ["AI_MODEL"],
                    sistemski="Ti si pomoćnik za Linux administraciju.")

print(razgovor.pitaj("Kako da vidim zauzeće diska?"))
print(razgovor.pitaj("A sortirano po veličini?"))

Metod _skrati je ovde ključan. Bez skraćivanja istorija raste dok ne premaši kontekst, a onda zahtev pukne sa greškom 400. Skraćivanje po broju poruka je grubo ali dovoljno; tačnije bi bilo po broju tokena, uz brojanje kroz usage iz prethodnih odgovora.

Strukturiran izlaz

Ono zbog čega ovo uopšte radiš u Pythonu, a ne u bashu:

import json

SISTEMSKI = """Klasifikuješ zapise iz sistemskih logova.
Vraćaš isključivo JSON, bez ikakvog dodatnog teksta.
Polja: tip (string), ozbiljnost (broj 1-5), ip (string ili null)."""

def klasifikuj(zapis):
    odgovor = klijent.chat.completions.create(
        model=os.environ["AI_MODEL"],
        messages=[
            {"role": "system", "content": SISTEMSKI},
            {"role": "user", "content": zapis},
        ],
        temperature=0,
        response_format={"type": "json_object"},
    )
    return json.loads(odgovor.choices[0].message.content)
>>> klasifikuj("SSH login failed for root from 203.0.113.45")
{'tip': 'neuspela_prijava', 'ozbiljnost': 4, 'ip': '203.0.113.45'}

Ovo i dalje može da zakaže — model ume da vrati ispravan JSON sa poljima koja nisu ona koja si tražio. Provera pre upotrebe:

def klasifikuj_sigurno(zapis):
    try:
        rezultat = klasifikuj(zapis)
    except json.JSONDecodeError:
        return None

    if not {"tip", "ozbiljnost", "ip"} <= rezultat.keys():
        return None
    if not isinstance(rezultat["ozbiljnost"], int):
        return None
    if not 1 <= rezultat["ozbiljnost"] <= 5:
        return None

    return rezultat

Nikad ne veruj izlazu modela bez provere. Ispravan oblik nije isto što i tačan sadržaj.

Greške i ponovni pokušaji

from openai import APIConnectionError, APIStatusError, APITimeoutError
import sys, time

def pitaj_uporno(poruke, pokusaja=3, **opcije):
    for pokusaj in range(1, pokusaja + 1):
        try:
            odgovor = klijent.chat.completions.create(
                model=os.environ["AI_MODEL"],
                messages=poruke,
                timeout=300,
                **opcije,
            )
            return odgovor.choices[0].message.content

        except APIConnectionError:
            print(f"Servis nedostupan (pokušaj {pokusaj})", file=sys.stderr)
        except APITimeoutError:
            print(f"Isteklo vreme (pokušaj {pokusaj})", file=sys.stderr)
        except APIStatusError as g:
            if g.status_code in (401, 403, 404):
                print(f"Greška {g.status_code}, nema smisla ponavljati", file=sys.stderr)
                raise
            print(f"Greška {g.status_code} (pokušaj {pokusaj})", file=sys.stderr)

        if pokusaj < pokusaja:
            time.sleep(2 ** pokusaj)

    raise RuntimeError(f"Neuspeh posle {pokusaja} pokušaja")

Razlikovanje grešaka je ovde suština. Kod 503 znači da se model učitava i vredi sačekati; kod 401 znači pogrešan ključ i ponavljanje ničemu ne služi. Razmak između pokušaja se udvostručuje, da servis koji se diže ne bombarduješ.

Podrazumevani rok od deset minuta je kratak za dugačke odgovore na sporijem hardveru — postavi timeout izričito.

Gotova alatka

Sve zajedno, u obliku koji ide u /usr/local/bin:

#!/usr/bin/env python3
"""Pitaj lokalni model. Čita sa argumenta ili sa standardnog ulaza."""

import argparse
import os
import sys
from openai import OpenAI, APIConnectionError, APIStatusError

def main():
    p = argparse.ArgumentParser(description="Upit lokalnom modelu")
    p.add_argument("pitanje", nargs="*", help="tekst pitanja")
    p.add_argument("-s", "--sistemski", help="sistemski prompt")
    p.add_argument("-t", "--temperatura", type=float, default=0.2)
    p.add_argument("-n", "--max-tokena", type=int, default=1024)
    p.add_argument("-m", "--model", default=os.environ.get("AI_MODEL", "qwen3:8b"))
    p.add_argument("--stream", action="store_true", help="postepen ispis")
    args = p.parse_args()

    pitanje = " ".join(args.pitanje)
    if not sys.stdin.isatty():
        ulaz = sys.stdin.read().strip()
        pitanje = f"{pitanje}\n\n{ulaz}" if pitanje else ulaz

    if not pitanje:
        p.error("nema pitanja ni na argumentu ni na ulazu")

    poruke = []
    if args.sistemski:
        poruke.append({"role": "system", "content": args.sistemski})
    poruke.append({"role": "user", "content": pitanje})

    klijent = OpenAI(
        base_url=os.environ.get("AI_BASE_URL", "http://localhost:11434/v1"),
        api_key=os.environ.get("AI_API_KEY", "nije-bitno"),
    )

    try:
        if args.stream:
            tok = klijent.chat.completions.create(
                model=args.model, messages=poruke,
                temperature=args.temperatura, max_tokens=args.max_tokena,
                stream=True, timeout=300,
            )
            for deo in tok:
                komad = deo.choices[0].delta.content
                if komad:
                    print(komad, end="", flush=True)
            print()
        else:
            odgovor = klijent.chat.completions.create(
                model=args.model, messages=poruke,
                temperature=args.temperatura, max_tokens=args.max_tokena,
                timeout=300,
            )
            print(odgovor.choices[0].message.content)

    except APIConnectionError:
        print("Servis nije dostupan.", file=sys.stderr)
        sys.exit(1)
    except APIStatusError as g:
        print(f"Greška {g.status_code}.", file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        sys.exit(130)

if __name__ == "__main__":
    main()
$ sudo install -m 755 pitaj.py /usr/local/bin/pitaj

Provera sys.stdin.isatty() je ono što ovu skriptu čini upotrebljivom u lancu komandi:

$ pitaj "Objasni šta radi umask"

$ journalctl -p err -n 30 --no-pager | pitaj "Sažmi ove greške u pet tačaka"

$ pitaj --stream -s "Odgovaraj samo komandom, bez objašnjenja" \
    "kako naći fajlove veće od 100MB"

$ cat nginx.conf | pitaj "Nađi bezbednosne propuste u ovoj konfiguraciji"

Izlazni kod 130 je dogovor za prekid preko Ctrl+C i čini skriptu pristojnom prema ostatku lanca. O rukovanju argumentima i izlaznim kodovima ima više u tekstu o argumentima u bash skriptama — načela su ista.

Obrada u paketu

Kada obrađuješ mnogo zapisa, paralelno slanje se isplati samo ako je ispod vLLM. Ollama zahteve ionako reda, pa tu paralelizam ne donosi ništa.

from concurrent.futures import ThreadPoolExecutor
import json

def obradi_sve(zapisi, niti=4):
    with ThreadPoolExecutor(max_workers=niti) as izvrsilac:
        return list(izvrsilac.map(klasifikuj_sigurno, zapisi))

with open("/var/log/auth.log") as f:
    zapisi = [red.strip() for red in f if "Failed" in red][:100]

for zapis, rezultat in zip(zapisi, obradi_sve(zapisi)):
    if rezultat and rezultat["ozbiljnost"] >= 4:
        print(json.dumps(rezultat, ensure_ascii=False))

Oznaka ensure_ascii=False je ovde bitna — bez nje naši znakovi izlaze kao \u010d umesto č.

Broj niti drži blizu vrednosti --max-num-seqs na serveru. Više od toga samo puni red čekanja.

Praktični scenariji

Scenario 1: veza istekne na dugačkim odgovorima

Postavi timeout=300 ili više. Podrazumevana vrednost je kratka za sporiji hardver.

Scenario 2: JSON parsiranje puca

Model je vratio propratni tekst uz JSON. Dodaj response_format, a ako i to podbaci, izvuci deo između prve { i poslednje }.

Scenario 3: greška o dužini konteksta posle nekoliko pitanja

Istorija je narasla preko prozora. Smanji max_poruka ili podigni kontekst servisa, uz proveru memorije iz teksta o VRAM-u.

Scenario 4: skripta radi ručno, ne radi iz crona

Cron ne učitava ~/.ai_env. Postavi promenljive u samom cron zapisu ili u skripti, kako je opisano u tekstu o cronu.

Scenario 5: naši znakovi izlaze kao kodovi

Dodaj ensure_ascii=False u json.dumps.

Scenario 6: paralelna obrada nije ništa ubrzala

Ispod je Ollama, koja zahteve reda. Za paralelizam treba vLLM.

Kratka referenca

  • pip install openai — zvanična biblioteka radi i sa lokalnim servisom
  • OpenAI(base_url=..., api_key=...) — usmeravanje na svoj servis
  • chat.completions.create() — glavni poziv
  • .choices[0].message.content — tekst odgovora
  • .usage — potrošnja tokena
  • stream=True — postepen ispis, uz proveru praznog komada
  • response_format={"type": "json_object"} — JSON izlaz
  • timeout=300 — postavi izričito
  • APIStatusError.status_code — 401 ne ponavljaj, 503 ponovi
  • sys.stdin.isatty() — čitanje sa standardnog ulaza
  • ensure_ascii=False — naši znakovi u JSON izlazu
  • ThreadPoolExecutor — ima smisla samo uz vLLM
  • Istoriju skraćuj, inače premaši kontekst
  • Izlaz modela proveri pre upotrebe

Vežba

  1. Napiši najmanji klijent i ispiši potrošnju tokena uz odgovor.
  2. Dopuni ga streaming režimom i uporedi doživljaj sa običnim pozivom.
  3. Napravi klasu za razgovor i postavi četiri povezana pitanja zaredom.
  4. Napiši funkciju koja vraća JSON sa tri polja i proveri svako pre upotrebe.
  5. Instaliraj alatku u /usr/local/bin i propusti kroz nju izlaz komande journalctl.
  6. Zaustavi servis i potvrdi da skripta javi razumnu grešku umesto da pukne.

Sledeći tekst u serijalu: AI u bash skriptama — automatska analiza logova

Povezano:

Comments

Post a Comment

Popular posts from this blog

Početak u Linuxu — šta je i zašto se uči

Ovo je prva lekcija na blogu. Ne uči se nijedna komanda — cilj je da razumeš gde ideš i zašto, pre nego što otvoriš terminal. Linux koristiš i sada Verovatno misliš da nikada nisi radio na Linuxu. Zapravo si u kontaktu sa njim ceo dan: Sajtovi koje otvaraš — najveći deo interneta vrti se na Linux serverima Android telefon — ispod svega je Linux kernel Netflix, Facebook, Google, banke — sve na Linuxu Superračunari — svih pet stotina najbržih na svetu Ruteri, televizori, automobili, bankomati Cloud — AWS, Azure i Google Cloud u ogromnoj većini pokreću Linux instance Praktično svugde osim na desktopu. To je i razlog zašto se Linux uči: ne zato što je bolji od Windowsa za svakodnevni rad, nego zato što je operativni sistem infrastrukture . Šta Linux zapravo jeste Linux je kernel — sloj koji upravlja hardverom, memorijom i procesima. Sam po sebi ne radi ništa vidljivo. Ono što instaliraš zove se distribucija : kernel plus sve ostalo — komande, serv...
Read more

Konverzija tipova podataka u Pythonu

Casting u programiranju se odnosi na promenu tipa podatka promenljive sa jednog tipa na drugi. U Pythonu, možete izvršiti casting (konverziju) pomoću određenih funkcija ili konstruktorima za različite tipove podataka. Evo nekoliko primera castinga u Pythonu: Casting u ceo broj (int) Možete konvertovati druge tipove podataka u ceo broj koristeći funkciju int() . decimalni = 3.14 ceo_broj = int(decimalni) # Konverzija decimalnog broja u ceo broj Casting u decimalni broj (float) Možete konvertovati druge tipove podataka u decimalni broj koristeći funkciju float() . broj = 42 decimalni_broj = float(broj) # Konverzija celog broja u decimalni broj Casting u nisku (string) Možete konvertovati druge tipove podataka u nisku koristeći funkciju str() . broj = 42 niska = str(broj) # Konverzija celog broja u nisku Casting u kompleksni broj (complex) Možete konvertovati druge tipove podata...
Read more

groupadd

 Komanda groupadd koristi se za dodavanje novih grupa na Unix i Linux sistemima. Evo nekoliko primera kako biste razumeli njenu upotrebu: Dodavanje nove grupe: sudo groupadd marketing Ova komanda će kreirati novu grupu sa imenom "marketing". Dodavanje grupe sa određenim GID-om (Group ID): sudo groupadd -g 1001 sales Postavlja GID grupe na 1001 prilikom kreiranja grupe "sales". Dodavanje grupe sa ekstra informacijama (komentarom): sudo groupadd -r -g 1002 -c "Development Team" devs Ovde se koristi -r opcija da se rezerviše sistemski GID, a -c dodaje komentar za grupu "devs". Dodavanje grupe sa ekstra informacijama iz datoteke: sudo groupadd -g 1003 -r -K GID_MIN=1000 -K GID_MAX=1999 -K INACTIVE=30 -K EXPIRE=2023-01-01 testers Koristi se -K opcija da se postave različite opcije (min/max GID, neaktivnost, istek) za grupu "testers". Dodavanje grupe bez stvaranja home direktorijuma: sudo groupadd -r -r -K NIS...
Read more