5.3 OpenAI Assistants API v2

Versió de Referència

OpenAI Assistants API v2 (2024). La versió 2 introdueix millores significatives respecte a v1: suport per a múltiples fitxers per missatge, vector stores administrats per OpenAI, i millores en el Code Interpreter. Documentació: platform.openai.com/docs/assistants

---

🎯 Assistants API vs LangChain/LlamaIndex

Característica	Assistants API	LangChain/LlamaIndex
Infraestructura	Gestionada per OpenAI	Tu l'has de muntar
Vector Store	Inclòs, automàtic	Has de configurar ChromaDB, etc.
Code Interpreter	Inclòs (sandbox segur)	Risc de seguretat si no sandboxes
Cost	Tokens + storage	Tokens + cost infraestructura
Flexibilitat	Baixa (locked-in OpenAI)	Alta (qualsevol LLM)
Debugging	Limitat	LangSmith, traces completes
Multi-model	No (només OpenAI)	Sí (GPT, Claude, Llama...)

Quan usar Assistants API? - Prototipatge ràpid sense voler gestionar infraestructura - Necessites Code Interpreter en un sandbox segur - El teu projecte ja usa l'ecosistema OpenAI completament

---

🏗️ Arquitectura de l'Assistants API

ASSISTANT (configuració persistent)
    ↓ usa
THREAD (conversa/sessió)
    ↓ conté
MESSAGES (historial de missatge)
    ↓ executa
RUN (instància d'execució)
    ↓ pot usar
TOOLS: File Search | Code Interpreter | Function Calling
    ↓ accedeix a
VECTOR STORES + FILES (documents indexats per OpenAI)

---

💻 Implementació Completa

Pas 1: Crear l'Assistant

from openai import OpenAI

client = OpenAI()  # Usa OPENAI_API_KEY de les variables d'entorn

# ── Crear l'assistant (una sola vegada, és persistent) ─────────

assistant = client.beta.assistants.create(
    name="Assistent ASIX",
    instructions="""
    Ets un expert en administració de sistemes i xarxes (ASIX).

    Pots ajudar amb:
    - Configuració de xarxes (VLANs, routing, firewalls)
    - Administració de servidors Linux/Windows
    - Seguretat informàtica i ciberseguretat
    - Scripting Bash i PowerShell

    Quan uses Code Interpreter, prefereix Python per a scripts.
    Respon sempre en català o castellà, mai en anglès.
    """,
    model="gpt-4o",
    tools=[
        {"type": "file_search"},        # Cerca en documents pujats
        {"type": "code_interpreter"},   # Executa codi Python en sandbox
    ]
)

print(f"Assistant creat: {assistant.id}")
# Guarda aquest ID! L'usaràs per a totes les converses.
ASSISTANT_ID = assistant.id

Pas 2: Gestió de Fitxers i Vector Store

import os

# ── Pujar fitxers a OpenAI ─────────────────────────────────────

file_manual = client.files.create(
    file=open("manual_cisco.pdf", "rb"),
    purpose="assistants"
)
print(f"Fitxer pujat: {file_manual.id}")

# ── Crear un Vector Store i afegir-hi fitxers ─────────────────

vector_store = client.beta.vector_stores.create(
    name="Documentació Tècnica ASIX",
    expires_after={"anchor": "last_active_at", "days": 30}
)

# Afegir fitxers al vector store
batch = client.beta.vector_stores.file_batches.upload_and_poll(
    vector_store_id=vector_store.id,
    files=[
        open("manual_cisco.pdf", "rb"),
        open("guia_linux.pdf", "rb"),
        open("procediments_seguretat.md", "rb"),
    ]
)
print(f"Status: {batch.status} — {batch.file_counts.completed} fitxers indexats")

# ── Associar el vector store a l'assistant ────────────────────

client.beta.assistants.update(
    assistant_id=ASSISTANT_ID,
    tool_resources={
        "file_search": {"vector_store_ids": [vector_store.id]}
    }
)
print("✅ Vector store associat a l'assistant")

Pas 3: Converses amb Threads

import time

# ── Crear un Thread (una sessió de conversa) ───────────────────

thread = client.beta.threads.create()
print(f"Thread creat: {thread.id}")

# ── Funció per fer preguntes i obtenir respostes ───────────────

def ask_assistant(question: str, thread_id: str) -> str:
    """Envia una pregunta a l'assistant i espera la resposta."""

    # 1. Afegir el missatge de l'usuari al thread
    client.beta.threads.messages.create(
        thread_id=thread_id,
        role="user",
        content=question
    )

    # 2. Iniciar l'execució (Run)
    run = client.beta.threads.runs.create(
        thread_id=thread_id,
        assistant_id=ASSISTANT_ID,
    )

    # 3. Esperar que el Run acabi (polling)
    # En producció, usar webhooks o streaming en lloc de polling!
    max_wait = 120  # Timeout de 2 minuts
    elapsed = 0

    while run.status in ("queued", "in_progress", "requires_action"):
        if elapsed > max_wait:
            client.beta.threads.runs.cancel(thread_id=thread_id, run_id=run.id)
            return "Error: Timeout en l'execució"

        time.sleep(1)
        elapsed += 1
        run = client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id)
        print(f"  Status: {run.status} ({elapsed}s)", end="\r")

    print()  # Nova línia

    if run.status == "failed":
        return f"Error: {run.last_error}"

    if run.status == "requires_action":
        # Function calling: el model vol cridar les nostres funcions
        return handle_required_action(run, thread_id)

    # 4. Obtenir la resposta
    messages = client.beta.threads.messages.list(
        thread_id=thread_id,
        order="desc",
        limit=1
    )

    last_message = messages.data[0]

    # Processar la resposta (pot tenir text + citations + imatges)
    response_text = []
    for block in last_message.content:
        if block.type == "text":
            text = block.text.value
            # Substituir citations inline [[0]] per referències llegibles
            for annotation in block.text.annotations:
                if annotation.type == "file_citation":
                    text = text.replace(
                        annotation.text,
                        f"[Font: {annotation.file_citation.file_id[:8]}...]"
                    )
            response_text.append(text)
        elif block.type == "image_file":
            response_text.append(f"[Imatge generada: {block.image_file.file_id}]")

    return "\n".join(response_text)


# ── Exemple de conversa multi-torn ────────────────────────────

print("Pregunta 1:")
r1 = ask_assistant(
    "Quins són els passos per configurar un trunk entre dos switches Cisco?",
    thread.id
)
print(r1)

print("\nPregunta 2 (seguiment):")
r2 = ask_assistant(
    "I si vull que la VLAN nativa sigui la 99?",  # L'assistant recorda el context!
    thread.id
)
print(r2)

Pas 4: Function Calling amb Assistants API

import json

# ── Definir funcions que pot cridar l'assistant ───────────────

def get_server_status(hostname: str) -> dict:
    """Comprova l'estat d'un servidor."""
    # En un sistema real, aquí hi hauria la crida real
    return {
        "hostname": hostname,
        "status": "online",
        "cpu_usage": 45.2,
        "memory_usage": 62.8,
        "disk_usage": 78.1
    }

# Mapa de funcions disponibles
FUNCTIONS_MAP = {
    "get_server_status": get_server_status,
}

# Actualitzar l'assistant amb les funcions
client.beta.assistants.update(
    assistant_id=ASSISTANT_ID,
    tools=[
        {"type": "file_search"},
        {"type": "code_interpreter"},
        {
            "type": "function",
            "function": {
                "name": "get_server_status",
                "description": "Comprova l'estat d'un servidor de producció.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "hostname": {
                            "type": "string",
                            "description": "Nom del servidor (p.ex. 'web01.empresa.com')"
                        }
                    },
                    "required": ["hostname"]
                }
            }
        }
    ]
)

def handle_required_action(run, thread_id: str) -> str:
    """Gestiona les crides a funcions quan el Run ho requereix."""
    tool_outputs = []

    for tool_call in run.required_action.submit_tool_outputs.tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)

        print(f"  Executant: {func_name}({func_args})")

        if func_name in FUNCTIONS_MAP:
            result = FUNCTIONS_MAP[func_name](**func_args)
            tool_outputs.append({
                "tool_call_id": tool_call.id,
                "output": json.dumps(result)
            })
        else:
            tool_outputs.append({
                "tool_call_id": tool_call.id,
                "output": f"Error: funció '{func_name}' no disponible"
            })

    # Enviar els resultats de les funcions
    run = client.beta.threads.runs.submit_tool_outputs_and_poll(
        thread_id=thread_id,
        run_id=run.id,
        tool_outputs=tool_outputs
    )

    messages = client.beta.threads.messages.list(thread_id=thread_id, limit=1)
    return messages.data[0].content[0].text.value

---

💰 Gestió de Costos

L'Assistants API cobra per tokens + per emmagatzematge de fitxers:

# Verificar ús d'un Run específic
run_details = client.beta.threads.runs.retrieve(
    thread_id=thread.id,
    run_id=run.id
)

usage = run_details.usage
print(f"Tokens usats: {usage.total_tokens}")
print(f"  - Prompt: {usage.prompt_tokens}")
print(f"  - Completion: {usage.completion_tokens}")

# Cost aproximat (gpt-4o-mini):
# Input:  $0.15 / 1M tokens
# Output: $0.60 / 1M tokens
# File Storage: $0.10 / GB / day

---

✅ Activitats

Exercici 5.3.1 — Assistant per a Documentació

Crea un assistant amb File Search que indexi el manual de l'assignatura (en PDF). Fes 10 preguntes i avalua la qualitat de les citations que proporciona.

Exercici 5.3.2 — Code Interpreter per a Anàlisi

Puja un fitxer CSV amb dades de xarxa (logs, estadístiques, etc.) i usa l'assistant amb Code Interpreter per: (a) generar estadístiques descriptives, (b) crear un gràfic, (c) detectar anomalies.

Exercici 5.3.3 — Comparativa de Costos

Per a una tasca específica (p.ex. respondre 100 preguntes sobre documentació):

1. Calcula el cost amb Assistants API (gpt-4o-mini)
2. Calcula el cost amb LangChain + ChromaDB local (gpt-4o-mini)
3. Calcula el cost amb LlamaIndex + Ollama local (sense cost de tokens)

Quina opció és millor per a quin cas d'ús?