Chat con un LLM locale: client Python per Ollama

In questo capitolo costruisci un client Python per Ollama: un modulo che parla con il modello locale installato nel capitolo 2, gestisce una conversazione a più turni, mostra la risposta mentre viene generata e non si blocca al primo errore di rete. È la base su cui poggiano la memoria del capitolo 4 e, più avanti, la ricerca nei documenti. Se non hai ancora chiaro che cosa sia un agente, riparti dal capitolo 1 della serie.

Il client usa solo requests per le chiamate HTTP, tenacity per i tentativi ripetuti e pydantic-settings per la configurazione. Le funzioni dell'API REST che usiamo (/api/chat, /api/embed, /api/tags, /api/version) sono descritte nella documentazione ufficiale di Ollama. Per progetti che vanno oltre l'esperimento, trovi una panoramica dei servizi in chatbot e agenti AI e nella sezione AI.

Architettura del client Python per Ollama

┌─────────────────────────────────────────────────────────────────┐
│                      OllamaClient                                │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐  │
│  │  Config     │  │  Session    │  │  Request Engine         │  │
│  │  (Pydantic) │──│  Manager    │──│  - requests.Session     │  │
│  └─────────────┘  └─────────────┘  │  - Streaming (NDJSON)  │  │
│         ▲                ▲         │  - Retry (tenacity)     │  │
│         │                │         │  - Timeout (connect/    │  │
│  ┌──────┴──────┐  ┌──────┴──────┐  │    read/total)          │  │
│  │  Models     │  │  Exceptions │  └─────────────────────────┘  │
│  │  (dataclass)│  │  (custom)   │              ▲                 │
│  └─────────────┘  └─────────────┘              │                 │
└────────────────────────────────────────────────┼─────────────────┘
                                                 │
                    ┌────────────────────────────┼─────────────────┐
                    │        Ollama API          │                 │
                    │    http://localhost:11434  │                 │
                    │    /api/chat (stream)      │                 │
                    │    /api/generate           │                 │
                    │    /api/embed              │                 │
                    └────────────────────────────┘                 │

Dipendenze

# requirements.txt
requests>=2.32.0
tenacity>=8.2.0
pydantic>=2.8.0
pydantic-settings>=2.3.0
python-json-logger>=2.0.0
tiktoken>=0.7.0  # per token counting

pip install -r requirements.txt

Configurazione tipizzata (Pydantic Settings)

# config.py
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Literal, Optional


class OllamaConfig(BaseSettings):
    """Configurazione client Ollama con validazione."""
    
    model_config = SettingsConfigDict(
        env_prefix="OLLAMA_",
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )
    
    # Connessione
    base_url: str = "http://localhost:11434"
    connect_timeout: float = 5.0
    read_timeout: float = 120.0
    total_timeout: float = 300.0
    
    # Modello default
    default_model: str = "mistral:7b-instruct-q4_k_m"
    
    # Parametri generazione default
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    top_p: float = Field(default=0.9, ge=0.0, le=1.0)
    top_k: int = Field(default=40, ge=0)
    num_predict: int = Field(default=-1, ge=-1)  # -1 = illimitato
    num_ctx: int = Field(default=4096, ge=512, le=131072)
    repeat_penalty: float = Field(default=1.1, ge=0.0, le=2.0)
    seed: int = Field(default=-1, ge=-1)  # -1 = random
    num_gpu: int = Field(default=999, ge=-1)  # layer su GPU (999 = tutti)
    
    # Retry
    max_retries: int = 3
    retry_base_delay: float = 1.0
    retry_max_delay: float = 30.0
    
    # Streaming
    stream_chunk_size: int = 1024
    
    # Logging
    log_level: str = "INFO"
    log_json: bool = True


config = OllamaConfig()

Eccezioni personalizzate

# exceptions.py
from typing import Optional, Any


class OllamaError(Exception):
    """Eccezione base per errori client Ollama."""
    
    def __init__(
        self,
        message: str,
        *,
        status_code: Optional[int] = None,
        response_body: Optional[Any] = None,
        request_id: Optional[str] = None,
    ):
        super().__init__(message)
        self.message = message
        self.status_code = status_code
        self.response_body = response_body
        self.request_id = request_id


class OllamaConnectionError(OllamaError):
    """Errore di connessione (daemon down, network, DNS)."""
    pass


class OllamaTimeoutError(OllamaError):
    """Timeout (connect, read, total)."""
    pass


class OllamaModelError(OllamaError):
    """Modello non trovato, non caricato, OOM."""
    pass


class OllamaValidationError(OllamaError):
    """Parametri invalidi, payload malformato."""
    pass


class OllamaStreamError(OllamaError):
    """Errore durante streaming (connessione interrotta, JSON malformato)."""
    pass

Modelli dati (dataclasses)

# models.py
from dataclasses import asdict, dataclass, field
from typing import Literal, Optional, List, Dict, Any
from datetime import datetime
import uuid


@dataclass(slots=True)
class Message:
    """Singolo messaggio conversazione."""
    role: Literal["system", "user", "assistant", "tool"]
    content: str
    tool_calls: Optional[List[Dict[str, Any]]] = None
    tool_call_id: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_api(self) -> Dict[str, Any]:
        """Serializza per API Ollama (/api/chat)."""
        msg = {"role": self.role, "content": self.content}
        if self.tool_calls:
            msg["tool_calls"] = self.tool_calls
        if self.tool_call_id:
            msg["tool_call_id"] = self.tool_call_id
        return msg


@dataclass(slots=True)
class GenerationOptions:
    """Parametri generazione (mappano 1:1 API Ollama options)."""
    temperature: float = 0.7
    top_p: float = 0.9
    top_k: int = 40
    num_predict: int = -1
    num_ctx: int = 4096
    repeat_penalty: float = 1.1
    seed: int = -1
    num_gpu: int = 999  # layer da portare su GPU (999 = tutti quelli possibili)
    stop: Optional[List[str]] = None
    mirostat: int = 0
    mirostat_eta: float = 0.1
    mirostat_tau: float = 5.0
    
    def to_dict(self) -> Dict[str, Any]:
        # asdict() perché con slots=True le istanze non hanno __dict__
        opts = {k: v for k, v in asdict(self).items() if v is not None}
        if opts.get("seed") == -1:
            opts.pop("seed")  # -1 = "casuale": lascia decidere a Ollama
        return opts


@dataclass(slots=True)
class ChatResponse:
    """Risposta completa (non-streaming) o chunk streaming."""
    model: str
    created_at: str  # timestamp ISO 8601 restituito da Ollama
    message: Message
    done: bool
    total_duration: Optional[int] = None  # ns
    load_duration: Optional[int] = None
    prompt_eval_count: Optional[int] = None
    eval_count: Optional[int] = None
    eval_duration: Optional[int] = None
    request_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    
    @property
    def tokens_per_second(self) -> Optional[float]:
        if self.eval_count and self.eval_duration:
            return self.eval_count / (self.eval_duration / 1e9)
        return None


@dataclass(slots=True)
class Conversation:
    """Sessione conversazione con history e metadati."""
    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    messages: List[Message] = field(default_factory=list)
    system_prompt: Optional[str] = None
    model: str = "mistral:7b-instruct-q4_k_m"
    options: GenerationOptions = field(default_factory=GenerationOptions)
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def add_message(self, role: str, content: str, **kwargs) -> Message:
        msg = Message(role=role, content=content, **kwargs)
        self.messages.append(msg)
        self.updated_at = datetime.now()
        return msg
    
    def get_api_messages(self) -> List[Dict[str, Any]]:
        msgs = []
        if self.system_prompt:
            msgs.append({"role": "system", "content": self.system_prompt})
        msgs.extend(m.to_api() for m in self.messages)
        return msgs
    
    def estimate_tokens(self, encoding_name: str = "cl100k_base") -> int:
        """Stima token totali (tiktoken)."""
        import tiktoken
        enc = tiktoken.get_encoding(encoding_name)
        total = 0
        for m in self.get_api_messages():
            total += len(enc.encode(m["content"]))
        return total
    
    def trim_to_context(self, max_tokens: int, encoding_name: str = "cl100k_base") -> int:
        """Rimuove messaggi più vecchi finché non sta nel context window. Ritorna token rimossi."""
        import tiktoken
        enc = tiktoken.get_encoding(encoding_name)
        removed = 0
        while self.estimate_tokens(encoding_name) > max_tokens and len(self.messages) > 1:
            # Mantieni system prompt + ultimi messaggi
            removed_msg = self.messages.pop(0)
            removed += len(enc.encode(removed_msg.content))
        return removed

Client principale con streaming, retry, logging

# client.py
import json
import logging
import time
from contextlib import contextmanager
from typing import Generator, Iterator, Optional, Union
from urllib.parse import urljoin

import requests
from requests.adapters import HTTPAdapter
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential_jitter,
    retry_if_exception_type,
    before_sleep_log,
)
from pythonjsonlogger import jsonlogger

from config import OllamaConfig, config
from models import Message, GenerationOptions, ChatResponse, Conversation
from exceptions import (
    OllamaError,
    OllamaConnectionError,
    OllamaTimeoutError,
    OllamaModelError,
    OllamaValidationError,
    OllamaStreamError,
)


# === Logging JSON strutturato ===
def setup_logging() -> logging.Logger:
    logger = logging.getLogger("ollama_client")
    logger.setLevel(config.log_level)
    logger.handlers.clear()
    
    handler = logging.StreamHandler()
    if config.log_json:
        formatter = jsonlogger.JsonFormatter(
            fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
            datefmt="%Y-%m-%dT%H:%M:%S",
        )
    else:
        formatter = logging.Formatter(
            "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
            datefmt="%H:%M:%S",
        )
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    return logger


logger = setup_logging()


# === Session HTTP con connection pooling ===
def create_session() -> requests.Session:
    session = requests.Session()
    adapter = HTTPAdapter(
        pool_connections=10,
        pool_maxsize=20,
        max_retries=0,  # Gestiamo retry noi con tenacity
    )
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session


# === Retry policy ===
def retry_policy(max_retries: int = None):
    max_retries = max_retries or config.max_retries
    return retry(
        reraise=True,
        stop=stop_after_attempt(max_retries),
        wait=wait_exponential_jitter(
            initial=config.retry_base_delay,
            max=config.retry_max_delay,
        ),
        retry=retry_if_exception_type((
            OllamaConnectionError,
            OllamaTimeoutError,
            requests.ConnectionError,
            requests.Timeout,
        )),
        before_sleep=before_sleep_log(logger, logging.WARNING),
    )


# === Client principale ===
class OllamaClient:
    """Client production-ready per Ollama API."""
    
    def __init__(self, config_override: Optional[OllamaConfig] = None):
        self.config = config_override or config
        self.session = create_session()
        self._default_options = GenerationOptions(
            temperature=self.config.temperature,
            top_p=self.config.top_p,
            top_k=self.config.top_k,
            num_predict=self.config.num_predict,
            num_ctx=self.config.num_ctx,
            repeat_penalty=self.config.repeat_penalty,
            seed=self.config.seed,
            num_gpu=self.config.num_gpu,
        )
    
    @property
    def chat_url(self) -> str:
        return urljoin(self.config.base_url.rstrip("/") + "/", "api/chat")
    
    @property
    def generate_url(self) -> str:
        return urljoin(self.config.base_url.rstrip("/") + "/", "api/generate")
    
    @property
    def embed_url(self) -> str:
        return urljoin(self.config.base_url.rstrip("/") + "/", "api/embed")
    
    @property
    def tags_url(self) -> str:
        return urljoin(self.config.base_url.rstrip("/") + "/", "api/tags")
    
    # --- Health check ---
    def health_check(self) -> bool:
        """Verifica se daemon Ollama risponde."""
        try:
            resp = self.session.get(
                urljoin(self.config.base_url, "/api/version"),
                timeout=self.config.connect_timeout,
            )
            return resp.status_code == 200
        except requests.RequestException:
            return False
    
    def list_models(self) -> list[dict]:
        """Lista modelli installati (equivale a `ollama list`)."""
        resp = self._request("GET", self.tags_url)
        return resp.json().get("models", [])
    
    # --- Core request con retry ---
    @retry_policy()
    def _request(
        self,
        method: str,
        url: str,
        *,
        json: Optional[dict] = None,
        stream: bool = False,
        timeout: Optional[tuple] = None,
    ) -> requests.Response:
        timeout = timeout or (
            self.config.connect_timeout,
            self.config.read_timeout,
        )
        try:
            resp = self.session.request(
                method=method,
                url=url,
                json=json,
                stream=stream,
                timeout=timeout,
            )
            resp.raise_for_status()
            return resp
        except requests.ConnectionError as e:
            raise OllamaConnectionError(
                f"Impossibile connettersi a {self.config.base_url}: {e}",
            ) from e
        except requests.Timeout as e:
            raise OllamaTimeoutError(
                f"Timeout ({timeout[0]}s connect, {timeout[1]}s read): {e}",
            ) from e
        except requests.HTTPError as e:
            self._handle_http_error(e)
    
    def _handle_http_error(self, e: requests.HTTPError) -> None:
        resp = e.response
        status = resp.status_code
        body = None
        try:
            body = resp.json()
        except Exception:
            body = resp.text
        
        if status == 404:
            raise OllamaModelError(
                "Modello non trovato o endpoint inesistente",
                status_code=status,
                response_body=body,
            )
        elif status == 400:
            raise OllamaValidationError(
                f"Richiesta malformata: {body}",
                status_code=status,
                response_body=body,
            )
        elif status == 500:
            raise OllamaModelError(
                f"Errore interno Ollama (possibile OOM): {body}",
                status_code=status,
                response_body=body,
            )
        else:
            raise OllamaError(
                f"HTTP {status}: {body}",
                status_code=status,
                response_body=body,
            )
    
    # --- Chat non-streaming ---
    def chat(
        self,
        messages: list[Message],
        *,
        model: Optional[str] = None,
        options: Optional[GenerationOptions] = None,
        stream: bool = False,
        timeout: Optional[tuple] = None,
    ) -> Union[ChatResponse, Generator[ChatResponse, None, None]]:
        """
        Chat completion.
        
        Args:
            messages: Lista messaggi (role, content)
            model: Override modello default
            options: Override parametri generazione
            stream: Se True, ritorna generator streaming
            timeout: Tupla (connect, read) in secondi; default da config
            
        Returns:
            ChatResponse (stream=False) o Generator[ChatResponse] (stream=True)
        """
        payload = {
            "model": model or self.config.default_model,
            "messages": [m.to_api() for m in messages],
            "stream": stream,
            "options": (options or self._default_options).to_dict(),
        }
        
        if stream:
            return self._stream_chat(payload, timeout)
        else:
            resp = self._request("POST", self.chat_url, json=payload, timeout=timeout)
            return self._parse_chat_response(resp.json())
    
    # --- Streaming generator ---
    def _stream_chat(
        self, payload: dict, timeout: Optional[tuple] = None
    ) -> Generator[ChatResponse, None, None]:
        resp = self._request(
            "POST", self.chat_url, json=payload, stream=True, timeout=timeout
        )
        
        for line in resp.iter_lines(decode_unicode=True):
            if not line:
                continue
            try:
                chunk = json.loads(line)
                yield self._parse_chat_response(chunk)
            except json.JSONDecodeError as e:
                logger.warning("Chunk JSON malformato", extra={"line": line, "error": str(e)})
                raise OllamaStreamError(f"Chunk streaming malformato: {line}") from e
    
    def _parse_chat_response(self, data: dict) -> ChatResponse:
        msg_data = data.get("message", {})
        return ChatResponse(
            model=data.get("model", ""),
            created_at=data.get("created_at", ""),
            message=Message(
                role=msg_data.get("role", "assistant"),
                content=msg_data.get("content", ""),
                tool_calls=msg_data.get("tool_calls"),
            ),
            done=data.get("done", False),
            total_duration=data.get("total_duration"),
            load_duration=data.get("load_duration"),
            prompt_eval_count=data.get("prompt_eval_count"),
            eval_count=data.get("eval_count"),
            eval_duration=data.get("eval_duration"),
        )
    
    # --- Embeddings (per RAG, Cap. 5) ---
    def embed(
        self,
        texts: Union[str, list[str]],
        model: str = "nomic-embed-text:latest",
    ) -> list[list[float]]:
        if isinstance(texts, str):
            texts = [texts]
        payload = {"model": model, "input": texts}
        resp = self._request("POST", self.embed_url, json=payload)
        return resp.json().get("embeddings", [])
    
    # --- Context manager per cleanup ---
    def close(self):
        self.session.close()
    
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.close()


# === Helper per conversazione multi-turn ===
class ConversationManager:
    """Gestisce conversazioni persistenti con trimming automatico context window."""
    
    def __init__(self, client: OllamaClient, conversation: Optional[Conversation] = None):
        self.client = client
        self.conversation = conversation or Conversation()
    
    def add_system_prompt(self, prompt: str):
        self.conversation.system_prompt = prompt
    
    def _messages_with_system(self) -> list[Message]:
        """History + system prompt in testa (chat() non lo aggiunge da solo)."""
        msgs = list(self.conversation.messages)
        if self.conversation.system_prompt:
            msgs.insert(0, Message("system", self.conversation.system_prompt))
        return msgs

    def user(self, content: str) -> ChatResponse:
        self.conversation.add_message("user", content)
        return self._generate()
    
    def assistant(self, content: str) -> Message:
        msg = self.conversation.add_message("assistant", content)
        return msg
    
    def _generate(self) -> ChatResponse:
        # Auto-trim se supera context window
        max_ctx = self.conversation.options.num_ctx
        if self.conversation.estimate_tokens() > max_ctx * 0.9:
            removed = self.conversation.trim_to_context(max_ctx)
            logger.info("Context window trimming", extra={"tokens_removed": removed})
        
        response = self.client.chat(
            messages=self._messages_with_system(),
            model=self.conversation.model,
            options=self.conversation.options,
            stream=False,
        )
        
        # Aggiungi risposta a history
        self.conversation.add_message(
            "assistant",
            response.message.content,
        )
        return response
    
    def stream_user(self, content: str) -> Generator[ChatResponse, None, None]:
        self.conversation.add_message("user", content)
        
        max_ctx = self.conversation.options.num_ctx
        if self.conversation.estimate_tokens() > max_ctx * 0.9:
            removed = self.conversation.trim_to_context(max_ctx)
            logger.info("Context window trimming (stream)", extra={"tokens_removed": removed})
        
        full_content = ""
        for chunk in self.client.chat(
            messages=self._messages_with_system(),
            model=self.conversation.model,
            options=self.conversation.options,
            stream=True,
        ):
            full_content += chunk.message.content
            yield chunk
        
        # Aggiungi risposta completa a history
        self.conversation.add_message("assistant", full_content)
    
    def get_history(self) -> list[Message]:
        return self.conversation.messages.copy()
    
    def clear_history(self, keep_system: bool = True):
        if keep_system and self.conversation.system_prompt:
            sys_prompt = self.conversation.system_prompt
            self.conversation = Conversation(
                system_prompt=sys_prompt,
                model=self.conversation.model,
                options=self.conversation.options,
            )
        else:
            self.conversation = Conversation(
                model=self.conversation.model,
                options=self.conversation.options,
            )


# === Esempio uso ===
if __name__ == "__main__":
    # Config da env (.env) o default
    # OLLAMA_BASE_URL=http://localhost:11434
    # OLLAMA_DEFAULT_MODEL=llama3.1:8b-instruct-q4_k_m
    # OLLAMA_TEMPERATURE=0.7
    # OLLAMA_NUM_CTX=8192
    
    with OllamaClient() as client:
        # Health check
        if not client.health_check():
            logger.error("Ollama non raggiungibile. Avvia: ollama serve")
            exit(1)
        
        print(f"Modelli disponibili: {[m['name'] for m in client.list_models()]}")
        
        # Conversazione semplice
        conv = ConversationManager(client)
        conv.add_system_prompt("Sei un esperto Python. Rispondi con codice commentato.")
        
        # Non-streaming
        resp = conv.user("Scrivi una funzione per calcolare fibonacci iterativo.")
        print(f"\n[Assistant]: {resp.message.content}")
        print(f"Tokens/sec: {resp.tokens_per_second:.1f}")
        
        # Streaming
        print("\n--- Streaming ---")
        for chunk in conv.stream_user("Ora fai la versione ricorsiva con memoization."):
            print(chunk.message.content, end="", flush=True)
        print()
        
        # History
        print(f"\nHistory ({len(conv.get_history())} messaggi):")
        for m in conv.get_history():
            print(f"  {m.role}: {m.content[:60]}...")

Tabella parametri generazione: riferimento rapido

La colonna "Default client" riporta i valori del nostro OllamaConfig, non quelli di Ollama. La documentazione dei parametri indica per Ollama valori diversi: per esempio temperature 0.8, repeat_penalty 1.0 e num_ctx 2048. Se non li imposti nel client, valgono quelli del server.

| Parametro | Tipo | Default client | Range | Effetto | Quando modificare | | --- | --- | --- | --- | --- | --- | | temperature | float | 0.7 | 0.0-2.0 | Casualità: 0 = quasi deterministico, valori alti = risposte più varie | 0.1-0.3 per codice e estrazione dati, 0.7-1.0 per testi creativi | | top_p | float | 0.9 | 0.0-1.0 | Nucleus sampling: taglia la coda della distribuzione | 0.95 per più varietà, 0.5 per risposte più focalizzate | | top_k | int | 40 | 0-100 | Considera solo i k token più probabili | Abbassalo per risposte più prudenti | | num_predict | int | -1 | -1, 1+ | Massimo di token generati (-1 = nessun limite) | Limita output verbosi, per esempio 512 per un riassunto | | num_ctx | int | 4096 | 512-131072 | Finestra di contesto (input + output) | Aumentala per documenti lunghi (8192, 16384, 32768), compatibilmente con la RAM/VRAM | | repeat_penalty | float | 1.1 | 0.0-2.0 | Penalizza le ripetizioni (1.0 = nessuna penalità) | 1.05-1.2 per evitare loop; oltre 1.5 la qualità può calare | | seed | int | -1 | -1, 0+ | Seed per la riproducibilità (-1 = non inviato) | Fissalo per test e debug | | num_gpu | int | 999 | -1, 0+ | Layer del modello da portare su GPU | Riducilo se la VRAM non basta, per esempio 20 o 32 | | stop | list[str] | null | - | Sequenze che interrompono la generazione | ["</s>", "Human:"] per modelli con formato chat particolare | | mirostat | int | 0 | 0, 1, 2 | Algoritmo Mirostat (0 = disattivato) | Solo per sperimentare con il controllo dell'entropia | | mirostat_eta | float | 0.1 | 0.01-1.0 | Learning rate di Mirostat | Conta solo con mirostat attivo | | mirostat_tau | float | 5.0 | 1.0-10.0 | Entropia obiettivo di Mirostat | Conta solo con mirostat attivo |

Pattern avanzati

1. Timeout diversi per richieste diverse

Il parametro timeout di chat() sostituisce i valori di default solo per quella chiamata.

# Chat veloce (timeout basso)
resp = client.chat(
    messages=[Message("user", "Ciao")],
    timeout=(2.0, 10.0),  # connect=2s, read=10s
)

# Documenti lunghi (timeout alto)
resp = client.chat(
    messages=[Message("user", "Riassumi questo testo: ...")],
    options=GenerationOptions(num_ctx=32768, num_predict=2048),
    timeout=(5.0, 300.0),  # connect=5s, read=5min
)

2. Output JSON

Chiedere il JSON nel prompt non basta a ottenerlo. L'API di Ollama accetta il campo format ("json" oppure uno schema JSON) e vincola l'output. Il nostro chat() non lo espone: per usarlo aggiungi un argomento format e copialo nel payload. Non usare stop=["}"]: il token di stop viene tagliato dall'output e il JSON resterebbe senza l'ultima graffa.

payload = {
    "model": "mistral:7b-instruct-q4_k_m",
    "messages": [{"role": "user", "content": "Valuta questa PR: ..."}],
    "stream": False,
    "format": "json",
    "options": {"temperature": 0.1},
}

3. Conteggio dei token

tiktoken fornisce una stima, non il conteggio esatto del tokenizer del tuo modello. Per il valore reale leggi prompt_eval_count ed eval_count nella risposta di Ollama.

from models import Conversation, Message

conv = Conversation(messages=[Message("user", "Spiegami lo streaming")])
tokens = conv.estimate_tokens()  # stima con cl100k_base
print(f"Token stimati: {tokens}")
print("Costo di inferenza locale: 0 euro (a parte l'energia)")

Passo per passo: dal file vuoto alla prima chat

Prepara l'ambiente. Crea una cartella di progetto, un ambiente virtuale (python3 -m venv .venv) e installa le dipendenze del file requirements.txt con pip install -r requirements.txt.
Verifica che Ollama risponda. Con curl localhost:11434/api/version devi ottenere un JSON con la versione. Se la connessione è rifiutata, avvia il servizio con ollama serve.
Controlla che il modello sia scaricato. ollama list mostra i modelli installati. Il nome indicato in OLLAMA_DEFAULT_MODEL deve coincidere con uno di quelli.
Crea i quattro moduli. Salva il codice di questa pagina in config.py, exceptions.py, models.py e client.py, nella stessa cartella.
Imposta la configurazione. Metti nel file .env solo ciò che vuoi cambiare, per esempio OLLAMA_DEFAULT_MODEL e OLLAMA_NUM_CTX. Il resto prende i valori predefiniti.
Lancia l'esempio. Esegui python client.py. Dovresti vedere l'elenco dei modelli, la risposta completa e poi la seconda risposta in streaming, token dopo token.
Prova un errore voluto. Ferma Ollama e rilancia lo script: il client esegue i tentativi con attesa crescente e poi segnala l'errore invece di restare appeso.

Checklist prima di usare il client

Il servizio Ollama è raggiungibile all'indirizzo di OLLAMA_BASE_URL.
Il modello indicato esiste in ollama list.
num_ctx è coerente con la memoria disponibile.
read_timeout è abbastanza alto per i modelli lenti.
Il client viene chiuso con with OllamaClient() as client:.
I log non contengono dati personali o segreti.

Errori da evitare

Dimenticare il system prompt nella chiamata. L'endpoint /api/chat non lo aggiunge da solo: va inserito come primo messaggio, come fa ConversationManager.
Lasciare crescere la history senza limiti. Prima o poi supera num_ctx e il modello perde l'inizio della conversazione.
Ripetere all'infinito i tentativi. Il retry serve per i guasti temporanei di rete, non per un modello inesistente o una richiesta errata.
Fidarsi della stima di tiktoken. È un'approssimazione: per i conteggi esatti usa i campi della risposta.
Esporre l'API di Ollama su Internet. Il servizio non ha autenticazione: tienilo su localhost o dietro un reverse proxy con controllo degli accessi.

Consigli pratici

Parti con temperature bassa (0.2) quando ti serve un risultato ripetibile, e alzala solo per testi creativi.
Tieni stream=True per le interfacce interattive: l'utente vede la risposta subito, anche se il modello è lento.
Annota in un file di log tokens_per_second: è il modo più semplice per confrontare modelli e quantizzazioni.
Cambia un parametro alla volta e confronta le risposte con lo stesso prompt e lo stesso seed.
Isola le impostazioni in .env, senza ricopiare i valori nel codice, e non committare mai quel file.

Domande frequenti

Perché non uso direttamente la libreria ufficiale ollama per Python? Si può. Scrivere il client con requests ti mostra come funziona il protocollo (richieste, streaming, errori) e ti lascia pieno controllo su retry e log.

Il client funziona con modelli diversi da Mistral? Sì. Basta cambiare OLLAMA_DEFAULT_MODEL con un nome che compare in ollama list, oppure passare model= alla chiamata.

Che differenza c'è tra /api/chat e /api/generate? /api/chat accetta una lista di messaggi con ruoli e serve per le conversazioni. /api/generate riceve un solo prompt di testo ed è più adatto a richieste singole.

Perché lo streaming si interrompe con un errore JSON? Perché la connessione è stata chiusa a metà di una riga. Il client solleva OllamaStreamError e registra la riga ricevuta, così puoi riprovare.

Troubleshooting client Python

| Errore | Causa | Fix | | --- | --- | --- | | ConnectionError: [Errno 111] Connection refused | ollama serve non avviato | Avvia il servizio e verifica con curl localhost:11434/api/version | | ReadTimeout su streaming | Modello lento, read_timeout basso | Aumenta OLLAMA_READ_TIMEOUT=300 e limita num_predict | | JSONDecodeError su un chunk | Riga troncata o connessione interrotta | Registra la riga grezza e riprova la richiesta | | OllamaModelError con HTTP 500 | Memoria GPU insufficiente o modello danneggiato | Riduci num_gpu o num_ctx; se persiste, ollama rm modello e ollama pull modello | | Conteggio token diverso dal previsto | tiktoken usa cl100k_base, non il tokenizer del modello | Considera la stima approssimata e usa prompt_eval_count della risposta | | Memoria che cresce in un loop lungo | requests.Session non chiusa o history non limitata | Usa with OllamaClient() as client: e il trimming di ConversationManager |

Prossimo capitolo

Nel capitolo 4: dai una memoria al tuo agente vediamo come conservare la conversazione tra una sessione e l'altra con LangChain, SQLite e file JSON, e come tenerla dentro la finestra di contesto.

Codice completo

I quattro moduli di questa pagina (config.py, exceptions.py, models.py, client.py) formano il pacchetto completo: copiali nella stessa cartella e il comando python client.py esegue l'esempio finale.

Serie "Il Mio Primo Agente Digitale". Dipendenze: requests, tenacity, pydantic e pydantic-settings, python-json-logger, tiktoken. Richiede Python 3.10 o superiore (per dataclass(slots=True)).