Agents sur une API LLM sans censure : appel de fonctions, frameworks et garde-fous

Anatomie de la boucle d'appel d'outils

Débarrassez-vous des frameworks et tout agent fait les quatre mêmes choses. Vous envoyez la conversation plus une liste tools. Le modèle répond soit par du texte, ce qui signifie qu'il a terminé, soit par tool_calls, ce qui signifie qu'il veut que quelque chose soit exécuté. Vous exécutez chaque appel et ajoutez un message tool par appel, en citant l'id de l'appel. Ensuite, vous demandez à nouveau au modèle avec la conversation plus longue.

Le signal pour la branche est finish_reason. Lorsqu'il est égal à tool_calls, le message de l'assistant a content défini à null et un tableau tool_calls, chaque entrée contenant un id, un nom de fonction et une chaîne d'arguments JSON. La page de référence montre la forme exacte. Deux détails piègent les gens : les arguments sont une chaîne à analyser, et le message de l'assistant contenant les appels doit rester dans l'historique avant vos résultats d'outil, dans l'ordre.

Une boucle complète de zéro

Le script ci-dessous définit une fonction, la décrit dans un schéma JSON et boucle jusqu'à ce que le modèle produise du texte brut. Il utilise le SDK Python officiel pointé vers l'API. Chaque itération ajoute le message de l'assistant, puis un message tool par appel, et un compteur d'étapes arrête les comportements incontrôlés.

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")

Remarquez ce que fait la boucle en cas d'échec : les erreurs deviennent des résultats tool sous forme de chaînes. Cela permet au modèle de lire le problème et d'essayer à nouveau, par exemple en corrigeant un mauvais id de ticket, au lieu de faire planter votre programme. La clause for ... else ne se déclenche que si le plafond est atteint sans réponse finale, ce qui est ainsi que vous détectez un modèle bloqué dans des boucles.

Gardez les prompts ciblés. Un court message système indiquant quand appeler un outil et quand répondre directement réduit les appels inutiles, et chaque appel sauté économise un aller-retour, des tokens et un slot de votre limite de débit.

Sécuriser l'exécution des outils

La sortie du modèle est une entrée non fiable. Un nom d'outil ou un argument choisi par le modèle peut être incorrect, étrange ou influencé par du texte que le modèle a lu sur une page web ou par un utilisateur. Traitez chaque appel comme une requête provenant d'Internet public.

• Liste blanche par nom. Recherchez la fonction dans un registre que vous avez créé. N'utilisez jamais la chaîne pour importer ou évaluer du code.
• Séparez lecture et écriture. Les outils en lecture seule s'exécutent librement. Tout ce qui envoie des e-mails, dépense de l'argent, supprime des données ou touche des fichiers nécessite une approbation humaine explicite.
• Validez les arguments contre le schéma que vous avez annoncé : clés requises, aucune clé supplémentaire, types et limites de longueur.
• Limitez le temps et la sortie. Appliquez des délais d'attente, et tronquez les résultats avant de les renvoyer pour qu'une énorme sortie d'outil ne consomme pas votre fenêtre de contexte de 100 000 tokens.
• Limitez les itérations et le coût. Limitez les étapes par tâche et les tokens par session, en utilisant le bloc usage.
• Journalisez chaque appel avec le nom, les arguments et le résultat afin que vous puissiez auditer le comportement.

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"

Traitez également les résultats tool comme non fiables. Si un tool retourne du texte extrait contenant des instructions, le modèle peut les suivre. Entourez ce contenu de délimiteurs clairs dans le message tool et indiquez dans votre prompt système que la sortie du tool est des données, pas des commandes.

Écrire des schémas d'outils exploitables par le modèle

Le schéma est la seule documentation que le modèle voit, il mérite donc le même soin qu'une API publique. Nommez les fonctions avec des verbes et des noms correspondant à la tâche, comme ticket_status ou lookup_invoice, plutôt que des étiquettes génériques comme run ou helper. Rédigez la description comme une instruction : ce que fait la fonction, ce qu'elle retourne, et quand ne pas l'utiliser.

• Gardez les paramètres peu nombreux et plats. Deux ou trois champs simples sont appelés correctement beaucoup plus souvent qu'un objet imbriqué avec dix.
• Utilisez des énumérations pour les ensembles fermés, par exemple "enum": ["open", "closed"], afin que le modèle ne puisse pas inventer une valeur.
• Décrivez chaque propriété avec une unité ou un exemple : "Id de ticket tel que T-101" vaut mieux qu'une chaîne nue.
• Retournez des résultats compacts. Les chaînes ou les petits objets JSON fonctionnent le mieux ; les longs dumps consomment du contexte pour peu de gain.
• Préférez plusieurs outils étroits à un outil omnibus avec un indicateur de mode.

Lorsqu'un appel échoue à la validation, retournez le message de validation en tant que résultat tool au lieu de lever une exception. Le modèle répare généralement l'appel au tour suivant, car il peut lire ce qui n'allait pas. S'il échoue de la même manière deux fois, réécrivez la description, pas la boucle.

Configuration LangChain

La classe ChatOpenAI de LangChain accepte un base_url, donc aucun package de fournisseur personnalisé n'est nécessaire. Définissez model sur uncensored, passez la clé depuis votre environnement, et liez les outils avec bind_tools. Installez langchain-openai et 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"]))

Le résultat de invoke est un AIMessage dont la liste tool_calls contient des noms et arguments analysés, donc vous sautez l'étape JSON manuelle. Pour obtenir une boucle autonome complète, remettez le même objet llm aux constructeurs d'agents de LangChain ; l'endpoint doit seulement honorer l'appel d'outils, ce qu'il fait. Si un agent semble ignorer ses outils, vérifiez que le docstring et les hints de type sont présents, car ils deviennent le schéma que le modèle voit.

Configuration LlamaIndex

LlamaIndex fournit OpenAILike pour les serveurs qui suivent le format des complétions de chat. Trois paramètres sont importants. api_base pointe vers l'API, is_chat_model=True sélectionne la route de chat, et is_function_calling_model=True indique au framework qu'il peut envoyer des outils. Définissez context_window à 100000 pour que ses séparateurs de texte et buffers de mémoire se dimensionnent correctement. Installez 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)

À partir de là, vous pouvez passer le même llm à un agent worker ou un moteur de requête. Rappelez-vous que les pipelines de récupération peuvent bourrer beaucoup de chunks dans un seul prompt ; avec une fenêtre de 100 000 tokens partagée entre prompt et complétion, limitez le nombre de chunks récupérés et gardez max_tokens modeste.

Gérer les agents avec un budget

Un agent transforme un message utilisateur en plusieurs requêtes, donc planifiez la capacité. Une tâche avec cinq tours d'outils coûte six appels de modèle. Dix tâches simultanées produisent donc soixante requêtes en peu de temps, et le plafond de 300 requêtes par minute arrive plus vite qu'une application de chat ne l'atteindrait. Placez le throttle à seau de jetons du guide erreurs et limites de débit devant votre client.

Le contexte augmente à chaque tour car chaque résultat tool est ajouté. Résumez ou supprimez les anciennes sorties tool une fois qu'elles ont été utilisées, et surveillez usage.prompt_tokens ; lorsqu'il approche 50 000, compressez l'historique. Pour le coût, supposez une tâche illustrative de 6 appels en moyenne 3 000 tokens de prompt et 200 tokens de complétion : cela fait 18 000 tokens d'entrée (environ 0,0045 $ à 0,25 $ par million) et 1 200 tokens de sortie (environ 0,0012 $ à 1,00 $ par million). Ces comptes de tokens sont des hypothèses pour les sommes, pas des mesures. Si vous débutez sur l'endpoint, commencez par les docs et le crédit d'essai gratuit.

Tester et déboguer un agent

Les agents échouent de manières que les tests unitaires sur une seule fonction manquent, donc testez la conversation. Enregistrez la liste complète des messages pour chaque exécution, y compris les arguments et résultats d'outil, et rejouez les échecs depuis le journal. Trois habitudes rapportent rapidement.

1. Simulez le modèle. Bouchonnez le client pour qu'il retourne des messages tool_calls scriptés. Cela teste votre registre, la validation et la terminaison de boucle sans réseau et sans dépense.
2. Échantillonnage par pointage. Utilisez une temperature faible, telle que 0,2, pour les tâches pilotées par des outils. Une moindre variance permet d'obtenir un JSON d'arguments plus reproductible, et vous pouvez l'augmenter ultérieurement pour les étapes plus axées sur la rédaction.
3. Vérifiez les trajectoires. Pour un petit nombre de prompts fixes, vérifiez quels outils ont été appelés et dans quel ordre, pas seulement le texte final.

Surveillez également finish_reason. Une valeur de length au milieu d'un appel d'outil signifie que les arguments ont été tronqués ; augmentez max_tokens (jusqu'à 32 000) et le JSON sera complet. Le streaming fonctionne aussi avec les outils, mais accumulez les fragments d'arguments par index d'appel et ne parsez qu'après la fin du flux. Pour un rappel sur ces chunks, consultez la section streaming de la référence.