El bucle con un modelo real

Las dos mitades ya existen. Hay un servidor con un modelo cargado, que en la lección que preparó la máquinaBlock 5 · Lesson 1 · In SpanishPreparar la máquina: un modelo local y la API de mensajesEl primer paso del agente fuera del navegador: un servidor que tiene el modelo cargado en tu ordenador, el protocolo con el que se le habla y un turno entero hecho a mano, mirando lo que viaja. El servidor responde y nada más: no ejecuta, no recuerda y no comprueba nada, y lo demás sigue siendo del harness. contestó a un turno montado a mano, y hay un agente, el del bloque anterior, que pasa su batería de tests sobre un modelo guionizado. Entre las dos queda un problema de formato y nada más: el agente guarda M como tuplas de Python, y el servidor lee el JSON del protocolo. Esta lección escribe la pieza que pasa de lo uno a lo otro, la pone donde estaba el modelo guionizado y deja que el harness corra solo.

¿Qué más hay que tocar? La respuesta que defiende la lección es que nada, y no es una respuesta cómoda. Contra el guion, el harness sólo recibía respuestas que alguien había escrito para que encajaran. Un modelo de verdad puede pedir una herramienta que no existe, mezclar texto y llamadas, quedarse sin reserva a mitad de una llamada, contestar en otro idioma o dar por hecha una tarea que no ha hecho. Si el harness aguanta todo eso sin cambiar una línea, hay que ver por qué. Y hay que ver qué deja de saber quien lo usa.

Esta es una ejecución entera del agente, con el pedido de la tortilla, sobre el modelo local. Cada turno dice cuánto tardó el modelo y cuántos tokens llevaba la petición; las flechas son lo que devuelve el modelo (←) y lo que el harness le contesta en el turno siguiente (→):

turno 1  15 s  460 tokens de entrada
  ←  llamada      leer {"ruta": "lista.txt"}
  →  observación  pan, leche, patatas
turno 2  10 s  501 tokens de entrada
  ←  llamada      buscar {"patron": "tortilla"}
  →  observación
turno 3  19 s  535 tokens de entrada
  ←  llamada      escribir {"ruta": "lista.txt", "contenido": "pan, leche, patatas, huevos, cebolla"}
  →  error        no se ha ejecutado: el usuario no lo ha permitido
turno 4  13 s  603 tokens de entrada
  ←  texto        Lo siento, no pude realizar la acción solicitada. ¿Hay otra forma en la que puedo ayudarte?
fin · 9 mensajes en M

El modelo empieza bien: lee la lista. Después busca la receta con tortilla, y buscar no encuentra nada, porque compara tal cual y el fichero dice Tortilla:, con mayúscula. Lo que hace a continuación no lo ha escrito nadie. No lista la carpeta ni prueba otra búsqueda: escribe la lista con huevos y cebolla, los dos ingredientes que faltan, sacados de lo que sabe de tortillas y no de ninguna observación. Esta vez acierta. La escritura se deniega, porque en esta ejecución no hay usuario a quien preguntar, y el modelo cierra con una disculpa que no dice qué ha pasado.

Ahora mira la traza con el contrato del agente delante. Cada llamada ha tenido su observación en el mensaje siguiente. La escritura no ha cambiado el entorno. El bucle ha terminado, en cuatro turnos. Ninguna de las cinco garantías ha fallado, y el pedido no se ha hecho. Lo que ha fallado está en la otra lista, la de lo que el agente no promete: que la tarea se haga y que el texto final diga la verdad sobre el entorno.

El adaptador y lo que promete

El adaptador es la pieza que implementa la firma de modelo hablando con un servidor:

ModeloAPI(nombre="local", max_tokens=1024)
modelo(mensajes, herramientas) -> respuesta

nombre es el modelo que el servidor tiene cargado y max_tokens la reserva de cada respuesta. Una instancia es una función con la firma de siempre, y su contrato tiene cinco cláusulas:

1. Traduce M sin tocar el contenido. El contenido de los mensajes sistema va al campo system, unido y en su orden. Cada uno de los demás pasa a messages, en el mismo orden, con el rol en inglés (user, assistant) y su contenido tal cual, cadena o lista de bloques. Las herramientas van a tools sin cambios. Es la tabla de la lección anterior, y nada más.
2. Devuelve la respuesta tal como llega: el campo content, una lista de bloques que el bucle añade a M sin copiarla ni reescribirla. Los ids de las llamadas los pone el servidor, y el tool_result de cada una copia el suyo.
3. Nunca devuelve una respuesta cortada. Si stop_reason es max_tokens, lanza una excepción. Es lo que prometió la lección del bucleBlock 4 · Lesson 3 · In SpanishEl bucleEl bucle de un agente llama al modelo con la lista de mensajes, añade lo que responde, ejecuta lo que pide y añade lo que vuelve, hasta que una respuesta no pide nada o se acaba el tope. La lección lo escribe con sus tres maneras de parar y sus invariantes: el modelo sólo ve esa lista, y la lista sólo crece.: el bucle decide por los bloques, y una respuesta cortada puede no traer ningún tool_use sin estar terminada, o traer uno con los argumentos a medias. Fuera de ese caso, stop_reason no se mira.
4. Un error del servidor es una excepción con su motivo. Una petición que no cabe en la ventana de contexto (un 400), un modelo que el servidor no tiene (un 404) o un servidor apagado salen de modelo, y por tanto del bucle, que sólo deja salir eso. El texto que da el servidor va en la excepción, para que sepas qué ha pasado.
5. No recuerda nada entre llamadas. Guarda el usage de la última petición para quien quiera mirarlo, y la siguiente no lo lleva. Como el servidor, cada vez recibe M entero.

La quinta es la que hace que el cambio sea un cambio de pieza y no una reescritura. El modelo guionizado tampoco recordaba nada que no estuviera en M, aparte de su puntero, así que el bucle no puede notar la diferencia. bucle.py, permisos.py, agente.py y entorno.py no cambian un byte, y el agente sobre el modelo local es agente(ModeloAPI(), pedido, entorno).

Lo que el guion nunca hacía

Que el bucle no cambie no quiere decir que no tenga trabajo nuevo. Un modelo de verdad devuelve cosas que ningún guion del bloque anterior devolvía, y cada una cae en una garantía distinta, o en ninguna:

Las tres primeras filas son el contrato trabajando. Las dos últimas son trabajo para el resto del bloque, que va añadiendo capas al harness para que pasen menos.

Y hay una diferencia que no está en ninguna fila: el modelo local sortea cada token, así que el mismo pedido puede dar dos ejecuciones distintas. Por eso los tests del agente se quedan sobre el guion. Un test fija el modelo para que, si no pasa, la culpa sólo pueda ser del harness; lo que hace el modelo con el harness no se prueba, se mide, y eso lo hace un banco de pruebas, más adelante en el bloque.

El adaptador, en Python

El código parte del punto de control en que terminó la lección anterior:

Todo lo que sigue corre contra un modelo local, el del aviso del principio, y añade tres ficheros sin tocar ninguno de los que ya tienes. El primero es api.py. Su mitad de arriba son la traducción de la primera cláusula y el envío, que es el enviar de protocolo.py con dos cosas más: una clave para cuando el servidor la pida, y el motivo del servidor en la excepción.

"""modelo(mensajes, herramientas) -> respuesta, sobre la API de mensajes. Lección 33 del curso.
 
El adaptador: traduce M al protocolo, manda la petición con urllib y devuelve la respuesta
tal como llega. Ninguna biblioteca del proveedor.
"""
import json
import os
import urllib.error
import urllib.request
 
URL = os.environ.get("ANTHROPIC_BASE_URL", "http://localhost:11434") + "/v1/messages"
CLAVE = os.environ.get("ANTHROPIC_API_KEY")       # el servidor local no la pide
ROLES = {"usuario": "user", "asistente": "assistant"}
 
 
def peticion(nombre, max_tokens, mensajes, herramientas):
    """M en el protocolo: el prompt de sistema, a su campo; los demás, con el rol en inglés."""
    p = {"model": nombre, "max_tokens": max_tokens, "thinking": {"type": "disabled"},
         "system": "\n\n".join(c for rol, c in mensajes if rol == "sistema"),
         "messages": [{"role": ROLES[rol], "content": c} for rol, c in mensajes if rol != "sistema"]}
    if herramientas:
        p["tools"] = list(herramientas)
    return p
 
 
def enviar(p):
    """Un POST con la petición; lo que contesta el servidor, leído del JSON."""
    cabeceras = {"content-type": "application/json"}
    if CLAVE:
        cabeceras.update({"x-api-key": CLAVE, "anthropic-version": "2023-06-01"})
    req = urllib.request.Request(URL, data=json.dumps(p).encode("utf-8"), headers=cabeceras)
    try:
        with urllib.request.urlopen(req, timeout=300) as r:
            return json.load(r)
    except urllib.error.HTTPError as e:              # un 400, un 404: el servidor dice por qué
        raise RuntimeError("el servidor contesta %d: %s" % (e.code, e.read().decode("utf-8")))

La mitad de abajo es la clase. Tiene la forma de ModeloGuionizado, un objeto que se llama como una función, y lo único que hace además de traducir y enviar es la tercera cláusula:

class ModeloAPI:
    """modelo(mensajes, herramientas) -> respuesta, de un modelo que sirve otro programa."""
    def __init__(self, nombre="local", max_tokens=1024):
        self.nombre, self.max_tokens = nombre, max_tokens
        self.usage = None                             # el de la última petición, para quien mire
 
    def __call__(self, mensajes, herramientas=()):
        r = enviar(peticion(self.nombre, self.max_tokens, mensajes, herramientas))
        self.usage = r["usage"]
        if r["stop_reason"] == "max_tokens":          # una respuesta cortada no se devuelve
            raise RuntimeError("la respuesta no cabe en max_tokens=%d" % self.max_tokens)
        return r["content"]

Dos cláusulas del contrato se pueden comprobar sin servidor, y van a tests/test_api.py: la traducción, sobre un M de dos turnos, y la respuesta cortada, cambiando enviar por una que devuelve lo que diría el servidor. Son tests del adaptador, no del agente, y como los del agente, no necesitan un modelo:

"""El adaptador, sin servidor: la traducción de M y la respuesta cortada. Lección 33 del curso."""
import unittest
from unittest import mock
 
import api
from agente import SISTEMA, inicio
from entorno import HERRAMIENTAS, llamada, resultado
 
 
class TestAdaptador(unittest.TestCase):
    def test_la_traduccion(self):
        r = [llamada("t1", "leer", ruta="lista.txt")]
        M = inicio("¿Qué hay en lista.txt?") + [("asistente", r), ("usuario", [resultado("t1", "pan")])]
        p = api.peticion("local", 1024, M, HERRAMIENTAS)
        self.assertEqual(p["system"], SISTEMA)
        self.assertEqual([m["role"] for m in p["messages"]], ["user", "assistant", "user"])
        self.assertEqual([m["content"] for m in p["messages"]], [c for _, c in M[1:]])
        self.assertEqual(p["tools"], HERRAMIENTAS)
 
    def test_la_respuesta_cortada(self):
        cortada = {"content": [{"type": "text", "text": "pan, le"}], "stop_reason": "max_tokens",
                   "usage": {"input_tokens": 453, "output_tokens": 1024}}
        with mock.patch("api.enviar", return_value=cortada):
            with self.assertRaises(RuntimeError):
                api.ModeloAPI()(inicio("¿Qué hay en lista.txt?"), HERRAMIENTAS)
 
 
if __name__ == "__main__":
    unittest.main()

Falta un programa que reciba un pedido y enseñe qué pasa en cada turno. correr.py lo hace sin tocar el bucle, con un modelo que envuelve a otro: Traza le pasa cada M al modelo de verdad y escribe lo que viaja. Las observaciones del turno anterior las encuentra en el último mensaje de M, que es justo donde el bucle las deja.

"""Un pedido al agente, contra el modelo local, turno a turno. Lección 33 del curso.
 
    python correr.py "¿Qué recetas hay y cuál lleva cebolla?"
"""
import json
import sys
import time
 
from agente import agente
from api import ModeloAPI
from entorno import ENTORNO
 
 
def linea(b):
    """Un bloque en una línea, y si es una observación, si es un error."""
    if b["type"] == "text":
        return "texto        " + " ".join(b["text"].split())
    if b["type"] == "tool_use":
        return "llamada      %s %s" % (b["name"], json.dumps(b["input"], ensure_ascii=False))
    return ("error        " if b["is_error"] else "observación  ") + b["content"].replace("\n", " / ")
 
 
class Traza:
    """Un modelo que envuelve a otro y escribe lo que viaja. Para el bucle, es un modelo más."""
    def __init__(self, modelo):
        self.modelo, self.turno = modelo, 0
 
    def __call__(self, mensajes, herramientas=()):
        rol, contenido = mensajes[-1]
        if rol == "usuario" and isinstance(contenido, list):     # las observaciones de antes
            for b in contenido:
                print("  →  " + linea(b))
        self.turno, t = self.turno + 1, time.time()
        r = self.modelo(mensajes, herramientas)
        u = self.modelo.usage
        print("turno %d  %.0f s  %d tokens de entrada" % (self.turno, time.time() - t,
              u["input_tokens"] + u.get("cache_read_input_tokens", 0)))
        for b in r:
            print("  ←  " + linea(b))
        return r

Al final del fichero, el agente con el modelo envuelto. El entorno es todavía el diccionario en memoria, una copia por ejecución, y no hay usuario: una escritura se deniega, como en la traza de arriba. Preguntarte de verdad llega con los permisos, más adelante en el bloque.

entorno = dict(ENTORNO)
motivo, texto, M = agente(Traza(ModeloAPI()), sys.argv[1], entorno)
print(motivo, "·", "%d mensajes en M" % len(M))
for ruta in sorted(entorno):
    if entorno[ruta] != ENTORNO.get(ruta):
        print("cambia", ruta, "->", entorno[ruta])

Antes de correr nada, comprueba que no has tocado ningún fichero de los que había. Como clonaste el punto de control, git conoce la etiqueta de la que partes, y esta orden no tiene que imprimir nada:

git diff --stat b5-l1

Si te has perdido, este es el punto de control al final de la lección, con el adaptador entero en api.py:

Si algún día quieres un modelo más fuerte que el que cabe en tu máquina, el adaptador ya sabe hablar con él. Una API en la nube que hable este protocolo es otra dirección: defines ANTHROPIC_BASE_URL con la de tu proveedor y ANTHROPIC_API_KEY con tu clave (en una variable de entorno, nunca en un fichero del repositorio), y creas el modelo con su nombre, ModeloAPI(nombre=…). No cambia nada más. Es el único sitio del curso donde aparece esa variante, porque cuesta dinero y el bloque se ha escrito para que no haga falta.

El agente, contra el servidor

Primero, los tests: los cuatro del agente sobre el guion y los dos del adaptador. Ninguno necesita el servidor:

python -m unittest -v

test_el_modelo_terco (tests.test_agente.TestAgente.test_el_modelo_terco) ... ok
test_la_escritura_sin_contenido (tests.test_agente.TestAgente.test_la_escritura_sin_contenido) ... ok
test_la_receta_de_la_web (tests.test_agente.TestAgente.test_la_receta_de_la_web) ... ok
test_la_tortilla (tests.test_agente.TestAgente.test_la_tortilla) ... ok
test_la_respuesta_cortada (tests.test_api.TestAdaptador.test_la_respuesta_cortada) ... ok
test_la_traduccion (tests.test_api.TestAdaptador.test_la_traduccion) ... ok
 
----------------------------------------------------------------------
Ran 6 tests in 0.003s
 
OK

Después, un pedido que el agente resuelve. Con el servidor en marcha (y sin una ANTHROPIC_BASE_URL que apunte a otro sitio), ejecuta:

python correr.py "¿Qué recetas hay y cuál lleva cebolla?"

turno 1  13 s  460 tokens de entrada
  ←  llamada      listar {"carpeta": "recetas/"}
  ←  llamada      buscar {"patron": "cebolla"}
  →  observación  recetas/gazpacho.txt / recetas/tortilla.txt
  →  observación  recetas/tortilla.txt: Tortilla: huevos, patatas, cebolla.
turno 2  19 s  556 tokens de entrada
  ←  texto        Las recetas disponibles son: 1. Gazpacho 2. Tortilla La receta que lleva cebolla es la **tortilla**, cuya lista de ingredientes es: huevos, patatas y cebolla.
fin · 5 mensajes en M

Los segundos dependen de tu máquina, y la primera ejecución tarda más porque el servidor carga los pesos. El texto puede salir con otras palabras. Lo que tiene que coincidir es la forma: dos llamadas en la primera respuesta, contestadas en el orden en que se pidieron, y una respuesta sin llamadas en el segundo turno. Fíjate en los tokens: 460 para una pregunta de siete palabras, porque la mayor parte son el prompt de sistema y las cuatro herramientas, que viajan en cada turno.

Ahora uno que no resuelve:

python correr.py "¿Qué lleva el gazpacho?"

turno 1  18 s  455 tokens de entrada
  ←  texto        I don't have access to any specific files or recipes that would contain information about what goes into gazpacho. You might want to check a recipe book or a culinary website for details on traditional gazpacho ingredients. Let me know if you'd like help with something else!
fin · 3 mensajes en M

En inglés, sin una sola llamada, y diciendo que no tiene acceso a ningún fichero con las herramientas delante. El harness no puede hacer nada: una respuesta sin llamadas es un final. Prueba también la tortilla, y otros pedidos tuyos; el modelo del curso es pequeño, y verlo equivocarse es parte de lo que esta lección enseña.

El agente ya corre contra un modelo de verdad, y el bucle que lo mueve es el del bloque anterior, sin una línea cambiada. Pero lo que mueve sigue siendo un juguete: un diccionario de cuatro ficheros en memoria y cuatro herramientas pensadas para él. buscar compara tal cual, y bastó una mayúscula para que el modelo dejara de leer y se pusiera a adivinar.

Un agente de programación trabaja sobre un repositorio en tu disco, y lo que necesita es leerlo, buscar en él, cambiar un fichero sin reescribirlo entero y ejecutar los tests. La lección siguiente, sobre las herramientas de ficheros, cambia el diccionario por los ficheros de verdad y le da al agente esas herramientas, cada una con un contrato que deja comprobar lo que ha hecho un modelo pequeño.