better

def politica_devolucions() -> str:

    """Text vigent de la política de devolucions."""

    return open("docs/devolucions.md").read()

if __name__ == "__main__":

    mcp.run()   # transport stdio per defecte

```

Durant el desenvolupament, el servidor s'aixeca amb la CLI del SDK, que l'obre en un inspector on pots cridar les eines a mà abans de connectar-hi cap agent:

Amb transport `stdio`, qui llança el servidor és el client, com a subprocés, i li parla per l'entrada i la sortida estàndard. La conseqüència pràctica sorprèn el primer dia: **dins d'un servidor stdio no hi pot haver cap `print()`**, perquè stdout és el canal del protocol. Els missatges de registre van a stderr.

Per provar les eines a mà abans de connectar-hi cap agent, l'ecosistema té un inspector web. Aquesta és l'ordre que funciona, des de la carpeta del servidor i amb l'entorn actiu:

```bash

uv run mcp dev server.py

npx @modelcontextprotocol/inspector --cli python server.py \

    --connect-timeout 90000 --method tools/list

```

> ⚠️ **No facis servir `mcp dev server.py`**: llança el servidor en un entorn aïllat que només conté el paquet `mcp`, sense les dependències del teu projecte, i es queda a mig connectar. El mode web, sense `--cli`, es rendeix als 15 segons, que en màquines lentes és menys del que triga el servidor a arrencar.

Fixa't que l'inspector no és imprescindible: `tools/list` és una crida del protocol com qualsevol altra, i un script de vint línies amb el client del SDK et dona la mateixa informació sense dependre de Node.

Decisions clau de disseny:

- **Granularitat de les eines**: una eina per operació de negoci, no una per endpoint HTTP. El model raona millor amb `stock(sku)` que amb `http_get(url)`.

### Abstraccions multi-proveïdor i SDKs

> 📝 Els dos exemples d'aquesta secció, LangChain i LiteLLM, són per **comparar formes**, no per executar-los: cap de les dues llibreries no forma part de l'entorn de treball d'aquests materials, que treballa directament amb l'SDK d'OpenAI contra un servidor propi. El que has de saber llegir és què canvia entre una abstracció i l'altra, i quin preu té cadascuna.

Per a codi multi-proveïdor, la solució habitual és el **patró estratègia** que ofereix LangChain: defineix tipus de missatge propis (`SystemMessage`, `HumanMessage`, `AIMessage`) i cada integració de proveïdor els tradueix al format natiu. La lògica de l'aplicació és idèntica entre proveïdors; el que canvia és la instanciació:

```python

    confiança: float

    resum: str

client = instructor.from_provider(PROVEÏDOR_I_MODEL)

resultat = client.chat.completions.create(

# `client` és el client OpenAI ja configurat: instructor l'embolcalla i en conserva

# el `base_url`, cosa que importa quan apuntes a un servidor propi i no a l'API pública

client_estructurat = instructor.from_openai(client)

resultat = client_estructurat.chat.completions.create(

    model=MODEL,

    messages=messages,

    response_model=AnàlisiSentiment,

    max_retries=2,

from instructor.exceptions import InstructorRetryException

try:

    resultat = client.chat.completions.create(..., max_retries=2)

    resultat = client_estructurat.chat.completions.create(..., max_retries=2)

except InstructorRetryException:

    resultat = None  # fallback: cua de revisió o valor per defecte

```

Quan `instructor` no és accessible o el model no admet el seu mecanisme (p.ex. un model local sense `response_format`), les alternatives per ordre de fiabilitat:

**`response_format` directe**: si el model implementa `response_format={"type": "json_schema", ...}` però no uses `instructor`, pots cridar l'API directament i validar amb Pydantic. És l'opció menys invasiva.

**`response_format` directe**: si el model implementa `response_format={"type": "json_schema", ...}` però no uses `instructor`, pots cridar l'API directament i validar amb Pydantic. És l'opció menys invasiva, i la que fa que la restricció d'esquema i la validació quedin totes dues a la vista.

```python

ESQUEMA = {

    "name": "sentiment",

    "schema": AnàlisiSentiment.model_json_schema(),

    "strict": True,   # decodificació restringida: el servidor no pot generar res fora de l'esquema

}

resposta = client.chat.completions.create(

    model=MODEL, messages=messages,

    response_format={"type": "json_schema",

                     "json_schema": {"name": "sentiment", "schema": AnàlisiSentiment.model_json_schema()}}

    response_format={"type": "json_schema", "json_schema": ESQUEMA},

)

resultat = AnàlisiSentiment.model_validate_json(resposta.choices[0].message.content)

```

Amb `"strict": True` cal tenir present que *tots* els camps han d'aparèixer a `required`: aquí "requerit" vol dir que la clau ha d'existir, no que hagi de tenir valor, i l'opcionalitat real s'expressa amb `"type": ["string", "null"]`. Si el servidor d'inferència no accepta aquesta forma, la sortida de recanvi és treure `strict` i `additionalProperties` i confiar en la validació Pydantic: la garantia passa de "el servidor no pot generar-ho malament" a "ho detectem a la frontera", que és una xarxa de seguretat que en tot cas has de tenir.

**Function calling com a esquema**: definir una eina fictícia amb l'esquema desitjat i forçar el model a cridar-la. Quan el model respecta `tool_choice`, els arguments sempre són JSON vàlid; alguns LLMs locals l'ignoren.

```python

Quan el LLM és opcional (enriquiment, classificació no bloquejant), la indisponibilitat pot ser transparent per a l'usuari. Quan és el camí crític, no ho pot ser.

```python

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

from tenacity import (

    retry, retry_if_exception_type, stop_after_attempt,

    wait_exponential, wait_exponential_jitter,

)

from openai import RateLimitError, APIStatusError

@retry(

@retry(

    retry=retry_if_exception_type((RateLimitError, APIStatusError)),

    stop=stop_after_attempt(8),        # async: més paciència

    wait=wait_exponential(multiplier=2, min=2, max=120),

    wait=wait_exponential_jitter(initial=2, max=120),   # jitter: evita que tots reintentin alhora

)

def cridar_model_async(messages): ...

```

El *jitter* del cas asíncron no és un detall: si el servei cau i tots els workers reintenten amb el mateix backoff exponencial, tornen a picar a la porta tots junts i el tomben una altra vegada just quan es recuperava. Afegir soroll aleatori a l'espera reparteix els reintents en el temps.

**Errors que no s'han de reintentar**: 400 (prompt invàlid), 401 (credencials), 404 (model no trobat). El resultat no canviarà.

LiteLLM Router implementa fallback de proveïdor natiu: si el model principal retorna errors persistents, enruta automàticament a un model de backup sense canvis al codi de l'aplicació.

    urgència: Literal["alta", "mitjana", "baixa"]

    resum: str

ESQUEMA = {

    "name": "ticket_classificat",

    "schema": TicketClassificat.model_json_schema(),

    "strict": True,

}

def classificar_ticket(text: str) -> TicketClassificat:

    resposta = client.chat.completions.parse(

    resposta = client.chat.completions.create(

        model=MODEL,

        messages=[

            {"role": "system", "content":

             "Urgència alta: servei caigut o pèrdua de dades."},

            {"role": "user", "content": text},

        ],

        response_format=TicketClassificat,

        response_format={"type": "json_schema", "json_schema": ESQUEMA},

        temperature=0,

    )

    return resposta.choices[0].message.parsed

    return TicketClassificat.model_validate_json(resposta.choices[0].message.content)

```

Les dues peces van juntes i cap no substitueix l'altra: `response_format` restringeix el que el servidor pot generar, i `model_validate_json` comprova a la frontera que el que ha arribat és el que esperaves. Vegeu [Sortida estructurada i validació](llm_systems.md#sortida-estructurada-i-validació) per a les alternatives quan el servidor no admet aquesta forma.

---

## 2. Assistent conversacional

        f"Objectiu: {perfil['objectiu']} | "

        f"Restriccions: {perfil.get('restriccions', 'cap')}"

    )

    resposta = client.chat.completions.parse(

    resposta = client.chat.completions.create(

        model=MODEL,

        messages=[

            {"role": "system", "content":

             "adaptat al perfil. Cada sessió ha de durar entre 30 i 60 minuts."},

            {"role": "user", "content": perfil_text},

        ],

        response_format=PlaSetmanal,

        response_format={"type": "json_schema", "json_schema": {

            "name": "pla_setmanal", "schema": PlaSetmanal.model_json_schema(), "strict": True,

        }},

        temperature=0.7,   # aquí la variació entre crides és desitjada

    )

    return resposta.choices[0].message.parsed  # la validació llança ValueError si falla

    # el `field_validator` llança ValueError si alguna durada surt del rang:

    # l'esquema fixa la forma, la regla de negoci la comprova Pydantic

    return PlaSetmanal.model_validate_json(resposta.choices[0].message.content)

```

---

### Arquitectura

Capes actives: Inferència (vLLM / batch API del proveïdor) + Cua de tasques + Emmagatzematge de resultats.

Capes actives: Inferència (vLLM / batch API del proveïdor) + Pool de workers + Emmagatzematge de resultats.

```text

Dataset d'entrada (fitxer / BD)

Productor: encua ítems

    │

    ▼

Cua de tasques (Redis / Celery)

    │  workers consumeixen ítems en paral·lel

Pool de workers (fils del procés, o una cua Redis si cal repartir-ho entre màquines)

    │  els workers consumeixen ítems en paral·lel

    ▼

Worker

    ├── crida al servei d'inferència

| Inferència | ✓ | vLLM per throughput alt; batch API si s'usa proveïdor comercial |

| Client d'inferència | ✓ | |

| Emmagatzematge vectorial | — | Típicament no necessari |

| Orquestració | — | La cua gestiona el flux i la concurrència |

| Orquestració | — | El pool de workers gestiona el flux i la concurrència |

| Observabilitat | ✓ | Progrés, errors per ítem, cost acumulat |

| Desplegament | ✓ | Workers escalables horitzontalment |

```python

# Exemple: classificació massiva de correu electrònic corporatiu

from celery import Celery

from concurrent.futures import ThreadPoolExecutor, as_completed

from pydantic import BaseModel

from typing import Literal

app = Celery("batch_llm", broker="redis://localhost:6379/0")

class CorreuClassificat(BaseModel):

    categoria: Literal["facturació", "tècnic", "comercial", "spam"]

    urgència: Literal["alta", "baixa"]

    resum: str

@app.task(bind=True, max_retries=3)

def classificar_correu(self, correu_id: str, cos_correu: str):

ESQUEMA = {

    "name": "correu_classificat",

    "schema": CorreuClassificat.model_json_schema(),

    "strict": True,

}

def classificar_correu(correu: dict) -> None:

    """Processa un ítem. No llança mai: un error d'un correu no pot tombar el batch."""

    try:

        resposta = client.chat.completions.parse(

        resposta = client.chat.completions.create(

            model=MODEL,

            messages=[

                {"role": "system", "content":

                 "Classifica el correu electrònic corporatiu per categoria i urgència. "

                 "Urgència alta: incidències crítiques o sol·licituds de direcció."},

                {"role": "user", "content": cos_correu},

                {"role": "user", "content": correu["cos"]},

            ],

            response_format=CorreuClassificat,

            response_format={"type": "json_schema", "json_schema": ESQUEMA},

            temperature=0,

        )

        desar_resultat(correu_id, resposta.choices[0].message.parsed)

        marcar_completat(correu_id)

    except Exception as exc:

        marcar_error(correu_id, str(exc))

        raise self.retry(exc=exc, countdown=60)

def classificar_safata(correus: list[dict]):

    for correu in correus:

        if estat_ítem(correu["id"]) != "completat":  # idempotència

            classificar_correu.delay(correu["id"], correu["cos"])

        resultat = CorreuClassificat.model_validate_json(resposta.choices[0].message.content)

        desar_resultat(correu["id"], resultat)

        marcar_completat(correu["id"])

    except Exception as exc:          # noqa: BLE001 — es registra i es continua

        marcar_error(correu["id"], str(exc))

def classificar_safata(correus: list[dict], workers: int = 8) -> None:

    # idempotència: els que ja s'han completat en una execució anterior no es tornen a fer

    pendents = [c for c in correus if estat_ítem(c["id"]) != "completat"]

    # el nombre de workers s'ajusta a la concurrència que aguanta el servidor

    # d'inferència, no a la que aguanta la teva màquina

    with ThreadPoolExecutor(max_workers=workers) as executor:

        futurs = {executor.submit(classificar_correu, c): c["id"] for c in pendents}

        for fet, futur in enumerate(as_completed(futurs), start=1):

            futur.result()

            if fet % 100 == 0:

                print(f"processats {fet}/{len(pendents)}")

```

Aquest esquelet cobreix el que fa falta per a volums de milers d'ítems: concurrència limitada, idempotència i errors aïllats per ítem. Quan el volum creix fins al punt que el procés ha de sobreviure a reinicis, repartir-se entre màquines o reprendre's al mig, el següent pas és una cua de tasques de veritat (Celery o RQ amb Redis), que és el mateix disseny amb l'estat fora del procés. I si treballes contra una API comercial, comprova abans si té una **Batch API**: sol costar la meitat per token i t'estalvia tota aquesta maquinària.

---

## 7. Assistent conversacional amb eines

### Entrades i sortides

**Entrada:** text pla (petició) + estat inicial del graf (`TypedDict`, el JSON intern de LangGraph que cada node pot llegir i modificar). L'estat és el mecanisme de comunicació entre nodes: cada node llegeix el que necessita i escriu el que produeix.

**Entrada:** text pla (petició) i, si el flux es modela com un graf d'estats, l'estat inicial (un `TypedDict` que cada node pot llegir i modificar). L'estat és el mecanisme de comunicació entre nodes: cada node llegeix el que necessita i escriu el que produeix.

**Sortida:** text pla o markdown (resultat del node final) + historial d'execució del graf en JSON (seqüència de nodes executats, estat en cada pas) per a auditoria. Si s'activa el checkpointing, l'estat persisteix a la BD entre execucions.

### Decisions de disseny clau

**Estat compartit com a contracte**: l'estat del graf (`TypedDict`) és la interfície entre nodes. Cada node llegeix el que necessita i escriu el que produeix. Dissenyar bé l'estat evita que els nodes s'acoplin entre si.

**Estat compartit com a contracte**: quan hi ha graf, el seu estat (`TypedDict`) és la interfície entre nodes. Cada node llegeix el que necessita i escriu el que produeix. Dissenyar bé l'estat evita que els nodes s'acoplin entre si.

**Per què no un `if/else` en el backend**: un dispatcher amb `if/else` pot fer routing senzill, però no pot gestionar paral·lelisme, checkpointing d'estat, ni fluxos amb bucles condicionals (revisor que torna al redactor si la qualitat no és suficient). LangGraph expressa aquests patrons de forma declarativa.

**Per què no un `if/else` en el backend**: sovint sí, i l'exemple d'aquesta secció ho és. Un dispatcher amb `if/else` fa routing senzill perfectament; el que no fa és paral·lelisme amb síntesi, checkpointing d'estat ni bucles condicionals (un revisor que torna la feina al redactor si la qualitat no és suficient). Un graf d'estats expressa aquests tres patrons de forma declarativa, i és quan n'has de menester algun que el framework es paga.

**Agents especialitzats, no eines especialitzades**: la raó per tenir múltiples agents és que cada especialista necessita un context i instruccions radicalment diferents. Un agent de facturació ha de tenir accés a la BD de factures i conèixer la política de reemborsaments; un agent tècnic necessita documentació tècnica i eines de diagnosi. Barrejar-ho en un sol agent amb moltes eines degrada la qualitat i complica el control.

### Exemple

```python

from langgraph.graph import StateGraph, END

from typing import TypedDict, Literal

El nucli del patró no és cap llibreria: és un classificador que tria, i un especialista per branca amb el seu propi system prompt i les seves pròpies eines.

```python

# Exemple: routing de tickets de suport a agents especialitzats

class EstatTicket(TypedDict):

    missatge: str

    categoria: str

    resposta: str

def supervisor(estat: EstatTicket) -> dict:

ESPECIALISTES = {

    "facturació": "Ets un especialista en facturació. Tens accés a la política de "

                  "reemborsaments i pots consultar l'historial de pagaments.",

    "tècnic": "Ets un especialista tècnic. Diagnostica el problema i proposa una "

              "solució pas a pas.",

    "comercial": "Ets un agent comercial. Informa sobre plans, preus i condicions "

                 "contractuals.",

}

def classificar(missatge: str) -> str:

    categoria = client.chat.completions.create(

        model=MODEL,

        messages=[

            {"role": "system", "content":

             "Classifica el ticket de suport. Respon únicament amb una paraula: "

             "'facturació', 'tècnic' o 'comercial'."},

            {"role": "user", "content": estat["missatge"]},

            {"role": "user", "content": missatge},

        ],

        temperature=0,

    ).choices[0].message.content.strip()

    return {"categoria": categoria}

    return categoria if categoria in ESPECIALISTES else "tècnic"   # ruta per defecte

def agent_facturació(estat: EstatTicket) -> dict:

    resposta = client.chat.completions.create(

def atendre(missatge: str) -> str:

    categoria = classificar(missatge)

    return client.chat.completions.create(

        model=MODEL,

        messages=[

            {"role": "system", "content":

             "Ets un especialista en facturació. Tens accés a la política de reemborsaments "

             "i pots consultar l'historial de pagaments. Resol el dubte del client."},

            {"role": "user", "content": estat["missatge"]},

            {"role": "system", "content": ESPECIALISTES[categoria]},

            {"role": "user", "content": missatge},

        ],

    ).choices[0].message.content

    return {"resposta": resposta}

def agent_tècnic(estat: EstatTicket) -> dict:

    resposta = client.chat.completions.create(

        model=MODEL,

        messages=[

            {"role": "system", "content":

             "Ets un especialista tècnic. Diagnostica el problema i proposa solució pas a pas."},

            {"role": "user", "content": estat["missatge"]},

        ],

    ).choices[0].message.content

    return {"resposta": resposta}

print(atendre("No he rebut la factura del mes passat."))

```

def agent_comercial(estat: EstatTicket) -> dict:

    resposta = client.chat.completions.create(

        model=MODEL,

        messages=[

            {"role": "system", "content":

             "Ets un agent comercial. Informa sobre plans, preus i condicions contractuals."},

            {"role": "user", "content": estat["missatge"]},

        ],

    ).choices[0].message.content

    return {"resposta": resposta}

def ruta_supervisor(estat: EstatTicket) -> Literal["facturació", "tècnic", "comercial"]:

    return estat["categoria"]

graf = StateGraph(EstatTicket)

graf.add_node("supervisor", supervisor)

graf.add_node("facturació", agent_facturació)

graf.add_node("tècnic", agent_tècnic)

graf.add_node("comercial", agent_comercial)

graf.set_entry_point("supervisor")

graf.add_conditional_edges("supervisor", ruta_supervisor, {

    "facturació": "facturació",

    "tècnic": "tècnic",

    "comercial": "comercial",

})

graf.add_edge("facturació", END)

graf.add_edge("tècnic", END)

graf.add_edge("comercial", END)

Cada especialista podria ser al seu torn un [bucle d'agent](#5-agent-amb-eines) amb el seu catàleg d'eines, i el `dict` seguiria sent el mateix. Fixa't que el supervisor no comparteix el seu system prompt amb els especialistes: entre agents només hi viatja el resultat.

app = graf.compile()

**Què hi afegeix LangGraph.** Mentre el flux sigui "classifica i despatxa", el codi de sobre ja és la solució i un framework només hi posa vocabulari. El que compres quan te'n vas a un graf d'estats és el que aquest `if` implícit no sap fer: arestes condicionals que **tornen enrere** (un revisor que retorna la feina al redactor mentre no arribi al llistó), execució en **paral·lel** de branques independents amb un node de síntesi al final, i **checkpointing** de l'estat perquè una execució llarga es pugui aturar, esperar una aprovació humana i reprendre's des d'on era.

# ús

resultat = app.invoke({

    "missatge": "No he rebut la factura del mes passat.",

    "categoria": "",

    "resposta": "",

})

print(resultat["resposta"])

```python

# Fragment il·lustratiu: LangGraph no forma part de l'entorn de treball

graf.add_conditional_edges("revisor", prou_bo, {True: END, False: "redactor"})

app = graf.compile(checkpointer=SqliteSaver.from_conn_string("estat.db"))

```

Aquestes tres coses són la frontera real del cas 9. Si el teu flux no en necessita cap, el que tens és el cas 5 amb diversos prompts.

---

## Com es passa d'un cas al següent