Agent sull'API LLM senza censura: Chiamata di funzioni, Framework e Guardrail

Anatomia del ciclo di chiamata di funzioni

Rimuovi i framework e ogni agent fa le stesse quattro cose. Invii la conversazione più una lista tools. Il modello risponde con testo, il che significa che ha finito, oppure con tool_calls, il che significa che vuole qualcosa di eseguito. Eseguisci ogni chiamata e aggiungi un messaggio tool per ogni ID chiamata, citando l'id della chiamata. Poi chiedi di nuovo al modello con la conversazione più lunga.

Il segnale per il ramo è finish_reason. Quando è uguale a tool_calls, il messaggio dell'assistant ha content impostato su null e un array tool_calls, ogni voce contiene un id, un nome di funzione e una stringa arguments in JSON. La pagina di riferimento mostra la forma esatta. Due dettagli creano confusione: gli arguments sono una stringa che deve essere analizzata, e il messaggio dell'assistant contenente le chiamate deve rimanere nella cronologia prima dei risultati dei tool, in ordine.

Un ciclo completo da zero

Lo script qui sotto definisce una funzione, la descrive in schema JSON e fa un ciclo finché il modello produce testo normale. Usa l'SDK Python ufficiale puntato all'API. Ogni passaggio accoda il messaggio dell'assistant, poi un messaggio tool per ogni chiamata, e un contatore di passaggi ferma i comportamenti incontrollati.

Copyimport json, os
from openai import OpenAI

client = OpenAI(base_url="https://api.uncensoredapis.com/v1", api_key=os.environ["API_KEY"])

TICKETS = {"T-101": "open", "T-102": "closed"}

def ticket_status(ticket_id: str) -> str:
    return TICKETS.get(ticket_id, "unknown ticket")

TOOLS = [{
    "type": "function",
    "function": {
        "name": "ticket_status",
        "description": "Look up the status of a support ticket by id.",
        "parameters": {
            "type": "object",
            "properties": {"ticket_id": {"type": "string"}},
            "required": ["ticket_id"],
        },
    },
}]
REGISTRY = {"ticket_status": ticket_status}

messages = [{"role": "user", "content": "Are tickets T-101 and T-102 still open?"}]

for step in range(6):                      # hard cap on loop turns
    r = client.chat.completions.create(
        model="uncensored", messages=messages, tools=TOOLS, max_tokens=400)
    msg = r.choices[0].message
    messages.append(msg.model_dump(exclude_none=True))
    if r.choices[0].finish_reason != "tool_calls":
        print(msg.content)
        break
    for call in msg.tool_calls:
        fn = REGISTRY.get(call.function.name)
        try:
            args = json.loads(call.function.arguments)
            result = fn(**args) if fn else f"error: unknown tool {call.function.name}"
        except Exception as exc:
            result = f"error: {exc}"
        messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
else:
    print("stopped: step limit reached")

Nota cosa fa il ciclo in caso di errore: gli errori diventano risultati dei tool come stringhe. Questo permette al modello di leggere il problema e riprovare, ad esempio correggendo un ID ticket errato, invece di far crashare il tuo programma. La clausola for ... else si attiva solo se il cap viene raggiunto senza una risposta finale, che è così che rilevi un modello bloccato in loop.

Mantieni i prompt focalizzati. Un breve system prompt che indica quando chiamare uno strumento e quando rispondere direttamente riduce le chiamate inutili, e ogni chiamata saltata salva un giro di andata e ritorno, token e uno slot del tuo limite di richieste.

Protezione dell'esecuzione degli strumenti

L'output del modello è un input non attendibile. Un nome di strumento o un argomento scelto dal modello potrebbe essere errato, strano o influenzato dal testo che il modello ha letto da una pagina web o da un utente. Tratta ogni chiamata come una richiesta che arriva da internet pubblico.

• Allowlist per nome. Cerca la funzione in un registro che hai creato. Non usare mai la stringa per importare o valutare codice.
• Separa lettura da scrittura. I tool solo lettura sono liberi. Qualsiasi cosa che invia email, spende denaro, cancella dati o tocca file richiede approvazione umana esplicita.
• Convalida gli arguments rispetto allo schema che hai annunciato: chiavi richieste, nessuna chiave extra, tipi e limiti di lunghezza.
• Limita tempo e output. Applica timeout e tronca i risultati prima di reinserirli così un output enorme di tool non consuma il tuo contesto di 100.000 token.
• Cap le iterazioni e il costo. Limita i passaggi per attività e i token per sessione, usando il blocco usage.
• Registra ogni chiamata con nome, argomenti e esito così puoi fare audit del comportamento.

Copyimport json
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutTimeout

ALLOWED = {"ticket_status", "lookup_invoice"}      # read-only tools only
NEEDS_APPROVAL = {"refund_order"}                  # side effects wait for a human
POOL = ThreadPoolExecutor(max_workers=4)

def validate(args: dict, schema: dict) -> None:
    props = schema["properties"]
    for key in schema.get("required", []):
        if key not in args:
            raise ValueError(f"missing argument: {key}")
    for key, val in args.items():
        if key not in props:
            raise ValueError(f"unexpected argument: {key}")
        if props[key]["type"] == "string" and not isinstance(val, str):
            raise ValueError(f"{key} must be a string")
        if isinstance(val, str) and len(val) > 200:
            raise ValueError(f"{key} too long")

def run_tool(name, raw_args, registry, schemas, approve=lambda n, a: False):
    if name in NEEDS_APPROVAL:
        args = json.loads(raw_args)
        if not approve(name, args):
            return "denied: this action needs human approval"
    elif name not in ALLOWED:
        return f"error: tool {name} is not permitted"
    args = json.loads(raw_args)
    validate(args, schemas[name]["parameters"])
    future = POOL.submit(registry[name], **args)
    try:
        return str(future.result(timeout=5))[:2000]    # time and size limits
    except FutTimeout:
        return "error: tool timed out"

Tratta anche i risultati degli strumenti come non attendibili. Se uno strumento restituisce testo scrapato contenente istruzioni, il modello potrebbe seguirle. Avvolgi tali contenuti in delimitatori chiari nel messaggio tool e dichiara nel tuo system prompt che l'output degli strumenti è dati, non comandi.

Scrittura degli schemi degli strumenti utilizzabili dal modello

Lo schema è l'unica documentazione che il modello vede, quindi merita la stessa cura di un'API pubblica. Nomina le funzioni con verbi e sostantivi che corrispondono all'attività, come ticket_status o lookup_invoice, invece di etichette generiche come run o helper. Scrivi la description come istruzione: cosa fa la funzione, cosa restituisce e quando non usarla.

• Mantieni pochi parametri piatti. Due o tre campi semplici vengono chiamati correttamente molto più spesso di un oggetto annidato con dieci.
• Usa enums per insiemi chiusi, ad esempio "enum": ["open", "closed"], così il modello non può inventare un valore.
• Descrivi ogni proprietà con un'unità o esempio: "ID ticket come T-101" è meglio di una stringa nuda.
• Restituisci risultati compatti. Stringhe o piccoli oggetti JSON funzionano meglio; dump lunghi consumano contesto per poco guadagno.
• Preferisci diversi strumenti stretti rispetto a un unico strumento omnibus con un flag modalità.

Quando una chiamata fallisce la convalida, restituisci il messaggio di convalida come risultato dello strumento invece di sollevare un'eccezione. Il modello di solito ripara la chiamata al turno successivo, poiché può leggere cosa c'era che non andava. Se fallisce nello stesso modo due volte, riscrivi la descrizione, non il ciclo.

Configurazione LangChain

La classe ChatOpenAI di LangChain accetta un base_url, quindi non serve un pacchetto provider personalizzato. Imposta model su uncensored, passa la chiave dal tuo ambiente e lega gli strumenti con bind_tools. Installa langchain-openai e langchain-core.

Copyimport os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = ChatOpenAI(
    model="uncensored",
    base_url="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    temperature=0.3,
    max_tokens=400,
)
llm_with_tools = llm.bind_tools([ticket_status])

ai = llm_with_tools.invoke("What is the status of ticket T-101?")
for call in ai.tool_calls:
    print(call["name"], call["args"], "->", ticket_status.invoke(call["args"]))

Il risultato di invoke è un AIMessage la cui lista tool_calls contiene nomi e arguments analizzati, così salti il passaggio JSON manuale. Per ottenere un ciclo autonomo completo, passa lo stesso oggetto llm ai costruttori agent di LangChain; l'endpoint deve solo onorare il tool calling, che fa. Se un agent sembra ignorare i suoi tool, controlla che il docstring e i type hints siano presenti, poiché diventano lo schema che il modello vede.

Configurazione LlamaIndex

LlamaIndex fornisce OpenAILike per server che seguono il formato delle chat completions. Tre impostazioni sono importanti. api_base punta all'API, is_chat_model=True seleziona il route chat, e is_function_calling_model=True dice al framework che può inviare tool. Imposta context_window a 100000 così i suoi splitter di testo e buffer di memoria si dimensionano correttamente. Installa llama-index-llms-openai-like.

Copyimport os
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.tools import FunctionTool

def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = OpenAILike(
    model="uncensored",
    api_base="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    is_chat_model=True,
    is_function_calling_model=True,
    context_window=100000,
    max_tokens=400,
)

tool = FunctionTool.from_defaults(fn=ticket_status)
response = llm.predict_and_call([tool], "Check ticket T-102 for me.")
print(response)

Da qui puoi passare lo stesso llm a un worker agent o a un motore di query. Ricorda che i pipeline di retrieval possono riempire molti chunk in un prompt; con una finestra di 100.000 token condivisa tra prompt e completion, limita il numero di chunk recuperati e mantieni max_tokens moderato.

Gestione degli agent con budget limitato

Un agent trasforma un messaggio utente in diverse richieste, quindi pianifica la capacità. Un'attività con cinque round di tool costa sei chiamate al modello. Dieci attività simultanee producono quindi sessanta richieste in breve spazio, e il limite di 300 richieste al minuto arriva prima di quanto un'app chat lo raggiungerebbe. Metti il throttle a secchio token dal playbook errori e limiti di richiesta davanti al tuo client.

Il contesto cresce con ogni round perché ogni risultato di tool viene aggiunto. Riassumi o rimuovi vecchi output di tool una volta che sono stati usati, e tieni d'occhio usage.prompt_tokens; quando si avvicina a 50.000, comprimi la cronologia. Per il costo, assumi un'attività illustrativa di 6 chiamate in media 3.000 prompt token e 200 completion token: sono 18.000 input token (circa $0,0045 a $0,25 per milione) e 1.200 output token (circa $0,0012 a $1,00 per milione). Questi conteggi di token sono assunzioni per le somme, non misurazioni. Se sei nuovo all'endpoint, inizia con la documentazione e il credito di prova gratuito.

Test e debug di un agent

Gli agent falliscono in modi che i test unitari su una singola funzione mancano, quindi testa la conversazione. Registra la lista completa dei messaggi per ogni esecuzione, inclusi argomenti e risultati degli strumenti, e riproduci i fallimenti dal log. Tre abitudini pagano rapidamente.

1. Simula il modello. Crea uno stub del client in modo che restituisca messaggi tool_calls scriptati. Questo testa il tuo registro, la convalida e la terminazione del loop senza rete e senza costi.
2. Blocca il campionamento. Usa un temperature basso, ad esempio 0,2, per le attività guidate dagli strumenti. Una minore casualità fornisce JSON degli argomenti più ripetibili; puoi aumentarlo in seguito per le fasi più orientate alla scrittura.
3. Asserisci sulle traiettorie. Per un piccolo numero di prompt fissi, controlla quali tool sono stati chiamati e in che ordine, non solo il testo finale.

Osserva anche finish_reason. Un valore di length nel mezzo di una chiamata di funzione indica che gli argomenti sono stati troncati; aumenta max_tokens (fino a 32.000) e il JSON verrà completato. Lo streaming funziona anche con gli strumenti, ma accumula i frammenti di argomento per indice di chiamata ed esegui il parsing solo dopo la chiusura dello stream. Per un ripasso di questi frammenti, consulta la sezione sullo streaming della documentazione.