Agentes con herramientas: tool calling, ReAct y planificación.

import ast
import json
import operator
import re

import numpy as np
import pandas as pd
from pydantic import BaseModel, Field, ValidationError

precios = pd.read_parquet("acciones6_precios.parquet")
noticias = pd.read_parquet("noticias6_corpus.parquet")
TICKERS = set(precios.ticker)
FECHA = r"^\d{4}-\d{2}-\d{2}$"


class ArgsPrecio(BaseModel):
    ticker: str = Field(pattern=r"^[A-Z]{4}$")
    fecha: str = Field(pattern=FECHA)


class ArgsNoticias(BaseModel):
    ticker: str = Field(pattern=r"^[A-Z]{4}$")
    desde: str = Field(pattern=FECHA)
    hasta: str = Field(pattern=FECHA)


class ArgsCalculadora(BaseModel):
    expresion: str


def precio(ticker, fecha):
    if ticker not in TICKERS:
        raise ValueError(f"ticker desconocido: {ticker}")
    s = precios[(precios.ticker == ticker) & (precios.fecha <= fecha)]
    if s.empty:
        raise ValueError(f"sin datos hasta {fecha}")
    r = s.iloc[-1]
    return {"ticker": ticker, "fecha": f"{r.fecha:%Y-%m-%d}",
            "cierre": round(float(r.cierre), 2)}


def buscar_noticias(ticker, desde, hasta):
    if ticker not in TICKERS:
        raise ValueError(f"ticker desconocido: {ticker}")
    ini = pd.Timestamp(desde, tz="UTC")
    fin = pd.Timestamp(hasta, tz="UTC") + pd.Timedelta(days=1)
    x = noticias[(noticias.ticker == ticker) & (noticias.ts >= ini)
                 & (noticias.ts < fin)].drop_duplicates("titular")
    return [{"fecha": f"{t:%Y-%m-%d}", "titular": h}
            for t, h in zip(x.ts[:5], x.titular[:5])]


OPS = {ast.Add: operator.add, ast.Sub: operator.sub,
       ast.Mult: operator.mul, ast.Div: operator.truediv,
       ast.USub: operator.neg}


def _evalua(n):
    if isinstance(n, ast.Constant) and type(n.value) in (int, float):
        return n.value
    if isinstance(n, ast.BinOp) and type(n.op) in OPS:
        return OPS[type(n.op)](_evalua(n.left), _evalua(n.right))
    if isinstance(n, ast.UnaryOp) and type(n.op) in OPS:
        return OPS[type(n.op)](_evalua(n.operand))
    raise ValueError("solo números, + - * / y paréntesis")


def calculadora(expresion):
    """Aritmética sin eval(): se recorre el árbol de la expresión."""
    return round(_evalua(ast.parse(expresion, mode="eval").body), 4)


HERRAMIENTAS = {
    "precio": (precio, ArgsPrecio, "Cierre de una acción en una "
               "sesión (la última sesión en o antes de la fecha)."),
    "buscar_noticias": (buscar_noticias, ArgsNoticias, "Titulares de "
                        "noticias de una acción entre dos fechas "
                        "(AAAA-MM-DD)."),
    "calculadora": (calculadora, ArgsCalculadora, "Evalúa una "
                    "expresión aritmética con + - * / y paréntesis."),
}


def ejecutar(nombre, args):
    """Valida y ejecuta; los errores vuelven como datos, no como
    excepciones, para que el modelo los vea."""
    if nombre not in HERRAMIENTAS:
        return {"error": f"herramienta desconocida: {nombre}"}
    f, Args, _ = HERRAMIENTAS[nombre]
    try:
        a = Args.model_validate(args)
    except ValidationError as e:
        return {"error": "argumentos no válidos: " + "; ".join(
            f"{x['loc'][0]}: {x['msg']}" for x in e.errors())}
    try:
        return {"resultado": f(**a.model_dump())}
    except Exception as e:
        return {"error": str(e)}


print(ejecutar("precio", {"ticker": "BUSI", "fecha": "2024-03-16"}))
print(ejecutar("precio", {"ticker": "BUSI", "fecha": "16/03/2024"}))
print(ejecutar("calculadora", {"expresion": "(27.88/26.19 - 1)*100"}))
print(ejecutar("calculadora",
               {"expresion": "__import__('os').getcwd()"}))
print(ejecutar("buscar_noticias", {"ticker": "BUSI",
               "desde": "2024-03-18", "hasta": "2024-03-20"}))

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

torch.manual_seed(0)
torch.use_deterministic_algorithms(True)
modelo = "Qwen/Qwen2.5-0.5B-Instruct"
tok = AutoTokenizer.from_pretrained(modelo)
llm = AutoModelForCausalLM.from_pretrained(modelo, dtype=torch.float32)


def esquema(nombres):
    """Herramientas en el formato de funciones de la plantilla."""
    return [{"type": "function", "function": {
        "name": n, "description": HERRAMIENTAS[n][2],
        "parameters": HERRAMIENTAS[n][1].model_json_schema()}}
        for n in nombres]


SISTEMA = ("Eres un asistente de research de renta variable. Usa las "
           "herramientas para obtener datos; no inventes cifras. Cuando "
           "tengas la respuesta, contesta en español en una o dos frases.")
mensajes = [{"role": "system", "content": SISTEMA},
            {"role": "user", "content": "¿Cierre de BUSI el 2024-03-15?"}]
texto = tok.apply_chat_template(mensajes, tools=esquema(["precio"]),
                                add_generation_prompt=True,
                                tokenize=False)
print(texto[texto.index("# Tools"):texto.index("<|im_start|>user")])

LLAMADA = re.compile(r"<tool_call>\s*(\{.*?\})\s*</tool_call>", re.S)


def agente_local(pregunta, nombres, max_pasos=5):
    msgs = [{"role": "system", "content": SISTEMA},
            {"role": "user", "content": pregunta}]
    traza, vistos = [], set()
    for paso in range(1, max_pasos + 1):
        e = tok.apply_chat_template(msgs, tools=esquema(nombres),
                                    add_generation_prompt=True,
                                    return_tensors="pt",
                                    return_dict=True)
        with torch.inference_mode():
            out = llm.generate(**e, max_new_tokens=80, do_sample=False,
                               pad_token_id=tok.eos_token_id)
        n = e["input_ids"].shape[1]
        txt = tok.decode(out[0, n:], skip_special_tokens=True).strip()
        llamadas = LLAMADA.findall(txt)
        print(f"  paso {paso} ({n} tokens de entrada)")
        if not llamadas:
            return txt, traza
        msgs.append({"role": "assistant", "content": txt})
        for crudo in llamadas:
            try:
                c = json.loads(crudo)
                nombre, args = c["name"], c.get("arguments", {})
            except (json.JSONDecodeError, KeyError, TypeError):
                nombre, args = None, None
            clave = json.dumps([nombre, args], sort_keys=True)
            if nombre is None:
                r = {"error": "llamada con JSON no válido"}
            elif clave in vistos:
                r = {"error": "llamada repetida: ya tienes el resultado"}
            else:
                r = ejecutar(nombre, args)
            vistos.add(clave)
            traza.append((nombre, args, r))
            print("    ", nombre, args, "=>",
                  json.dumps(r, ensure_ascii=False)[:70])
            msgs.append({"role": "tool",
                         "content": json.dumps(r, ensure_ascii=False)})
    return None, traza


BASICAS = ["precio", "buscar_noticias", "calculadora"]
TAREAS = [
    "¿Cuánto cambió en porcentaje el cierre de BUSI entre el "
    "2024-03-15 y el 2024-03-22?",
    "¿Cuál fue el cierre de ACME el 2024-03-15?"]
resultados = []
for pregunta in TAREAS:
    print(pregunta)
    final, traza = agente_local(pregunta, BASICAS)
    resultados.append((pregunta, final, traza))
    print("  RESPUESTA:", final)

NUM = re.compile(r"\d[\d.,]*\d|\d")


def cifras(s):
    out = []
    for t in NUM.findall(s or ""):
        if "," in t:
            t = t.replace(".", "").replace(",", ".")
        elif re.fullmatch(r"\d{1,3}(\.\d{3})+", t):
            t = t.replace(".", "")
        out.append(float(t.rstrip(".")))
    return out


def cerca(x, ys):
    return any(abs(x - y) <= 0.006 * max(1.0, abs(y)) for y in ys)


def verificar(pregunta, final, traza):
    if final is None:
        return ["sin respuesta: límite de pasos"]
    vistas, calculadas, errores = [], [], []
    for nombre, args, r in traza:
        if "resultado" in r:
            vistas += cifras(json.dumps(r["resultado"]))
            if nombre in ("calculadora", "variacion"):
                calculadas += cifras(json.dumps(r["resultado"]))
        else:
            errores.append(nombre)
    fallos = []
    for x in cifras(final):
        if not cerca(x, vistas + cifras(pregunta)):
            fallos.append(f"cifra sin fuente: {x:g}")
    for p in re.findall(r"(\d[\d.,]*)\s*%", final):
        if not cerca(cifras(p)[0], calculadas):
            fallos.append(f"porcentaje sin cálculo: {p}%")
    if errores and not vistas:
        fallos.append("error de herramienta sin resolver")
    return fallos


for pregunta, final, traza in resultados:
    print(pregunta[:45], "|", verificar(pregunta, final, traza) or "OK")

class ArgsVariacion(BaseModel):
    ticker: str = Field(pattern=r"^[A-Z]{4}$")
    desde: str = Field(pattern=FECHA)
    hasta: str = Field(pattern=FECHA)


def variacion(ticker, desde, hasta):
    a, b = precio(ticker, desde), precio(ticker, hasta)
    return {"desde": a, "hasta": b, "variacion_pct":
            round(100 * (b["cierre"] / a["cierre"] - 1), 2)}


HERRAMIENTAS["variacion"] = (
    variacion, ArgsVariacion, "Variación porcentual del cierre de una "
    "acción entre dos fechas (AAAA-MM-DD), calculada con los dos "
    "cierres.")
final, traza = agente_local(TAREAS[0], ["variacion", "buscar_noticias"])
print("  RESPUESTA:", final)
print("  verificación:", verificar(TAREAS[0], final, traza) or "OK")

from openai import OpenAI

cliente = OpenAI()                      # OPENAI_API_KEY del entorno
herramientas = [{"type": "function", "name": n,
                 "description": HERRAMIENTAS[n][2],
                 "parameters": HERRAMIENTAS[n][1].model_json_schema()}
                for n in BASICAS]
entrada = [{"role": "user", "content": TAREAS[0]}]
r = cliente.responses.create(model="gpt-5", input=entrada,
                             tools=herramientas)
for item in r.output:
    if item.type == "function_call":
        res = ejecutar(item.name, json.loads(item.arguments))
        entrada += [item, {"type": "function_call_output",
                           "call_id": item.call_id,
                           "output": json.dumps(res)}]

import anthropic

cliente = anthropic.Anthropic()         # ANTHROPIC_API_KEY del entorno
herramientas = [{"name": n, "description": HERRAMIENTAS[n][2],
                 "input_schema": HERRAMIENTAS[n][1].model_json_schema()}
                for n in BASICAS]
mensajes = [{"role": "user", "content": TAREAS[0]}]
r = cliente.messages.create(model="claude-opus-5-5", max_tokens=1024,
                            tools=herramientas, messages=mensajes)
if r.stop_reason == "tool_use":
    resultados = [{"type": "tool_result", "tool_use_id": b.id,
                   "content": json.dumps(ejecutar(b.name, b.input))}
                  for b in r.content if b.type == "tool_use"]
    mensajes += [{"role": "assistant", "content": r.content},
                 {"role": "user", "content": resultados}]

import anthropic
import httpx
import httpx2
import openai
import tiktoken

PREGUNTA = TAREAS[0]
DESDE, HASTA = "2024-03-15", "2024-03-22"


def guion(escenario, resultados):
    """Siguiente acción del «modelo» simulado: lista de llamadas
    (nombre, args) o el texto final. `resultados` = salidas de
    herramientas recibidas hasta ahora (diccionarios)."""
    n = len(resultados)
    if escenario == "bucle":
        return [("precio", {"ticker": "BUSI", "fecha": DESDE})]
    if escenario == "herramienta_equivocada":
        if n == 0:
            return [("buscar_noticias", {"ticker": "BUSI",
                     "desde": DESDE, "hasta": HASTA})]
        return "BUSI subió por la oferta de compra de un fondo."
    if escenario == "argumentos_inventados":
        if n == 0:
            return [("precio", {"ticker": "BUSMAR",
                                "fecha": "15/03/2024"})]
        resultados = resultados[1:]
        n -= 1
    if n == 0:
        return [("precio", {"ticker": "BUSI", "fecha": DESDE}),
                ("precio", {"ticker": "BUSI", "fecha": HASTA})]
    if n == 2:
        a, b = (r["resultado"]["cierre"] for r in resultados[:2])
        return [("calculadora", {"expresion": f"({b}/{a} - 1)*100"})]
    v = resultados[2]["resultado"]
    return f"El cierre de BUSI varió un {v:.2f} % entre {DESDE} y {HASTA}."


ENC = tiktoken.get_encoding("o200k_base")
REGISTRO = []                      # tokens de cada petición recibida


def sim_openai(request):
    cuerpo = json.loads(request.content)
    REGISTRO.append(len(ENC.encode(request.content.decode())))
    res = [json.loads(x["output"]) for x in cuerpo["input"]
           if isinstance(x, dict) and x.get("type") == "function_call_output"]
    accion = guion(cuerpo["model"], res)
    if isinstance(accion, str):
        salida = [{"type": "message", "id": "msg_1", "role": "assistant",
                   "status": "completed", "content": [
                       {"type": "output_text", "text": accion,
                        "annotations": []}]}]
    else:
        salida = [{"type": "function_call", "id": f"fc_{len(res)}_{i}",
                   "call_id": f"call_{len(res)}_{i}", "name": nom,
                   "arguments": json.dumps(args), "status": "completed"}
                  for i, (nom, args) in enumerate(accion)]
    return httpx.Response(200, json={
        "id": "resp_sim", "object": "response", "created_at": 0,
        "status": "completed", "model": cuerpo["model"],
        "output": salida, "parallel_tool_calls": True,
        "tool_choice": "auto", "tools": []})


def sim_anthropic(request):
    cuerpo = json.loads(request.content)
    REGISTRO.append(len(ENC.encode(request.content.decode())))
    res = [json.loads(b["content"]) for m in cuerpo["messages"]
           if m["role"] == "user" and isinstance(m["content"], list)
           for b in m["content"] if b.get("type") == "tool_result"]
    accion = guion(cuerpo["model"], res)
    if isinstance(accion, str):
        contenido, parada = [{"type": "text", "text": accion}], "end_turn"
    else:
        contenido = [{"type": "tool_use", "id": f"toolu_{len(res)}_{i}",
                      "name": nom, "input": args}
                     for i, (nom, args) in enumerate(accion)]
        parada = "tool_use"
    return httpx2.Response(200, json={
        "id": "msg_sim", "type": "message", "role": "assistant",
        "model": cuerpo["model"], "content": contenido,
        "stop_reason": parada, "stop_sequence": None,
        "usage": {"input_tokens": 0, "output_tokens": 0}})


oai = openai.OpenAI(
    api_key="sk-simulado", base_url="http://simulado/v1",
    http_client=httpx.Client(transport=httpx.MockTransport(sim_openai)))
ant = anthropic.Anthropic(
    api_key="sk-simulado", base_url="http://simulado",
    http_client=httpx2.Client(
        transport=httpx2.MockTransport(sim_anthropic)))
r = oai.responses.create(model="correcto", input=PREGUNTA)
print([(o.type, o.name, o.arguments) for o in r.output])
m = ant.messages.create(model="correcto", max_tokens=200,
                        messages=[{"role": "user", "content": PREGUNTA}])
print(m.stop_reason, [(b.type, b.name, b.input) for b in m.content])

def ejecutar_seguro(nombre, args, vistos):
    clave = json.dumps([nombre, args], sort_keys=True)
    if clave in vistos:
        return {"error": "llamada repetida: ya tienes el resultado"}
    vistos.add(clave)
    return ejecutar(nombre, args)


def agente_api(proveedor, escenario, max_pasos=6):
    REGISTRO.clear()
    historial = [{"role": "user", "content": PREGUNTA}]
    vistos, eventos = set(), []
    for paso in range(1, max_pasos + 1):
        if proveedor == "openai":
            r = oai.responses.create(model=escenario, input=historial)
            calls = [(o.call_id, o.name, json.loads(o.arguments))
                     for o in r.output if o.type == "function_call"]
            final = r.output_text
            historial += [o.model_dump(exclude_none=True)
                          for o in r.output]
        else:
            r = ant.messages.create(model=escenario, max_tokens=500,
                                    messages=historial)
            calls = [(b.id, b.name, b.input) for b in r.content
                     if b.type == "tool_use"]
            final = "".join(b.text for b in r.content
                            if b.type == "text")
            historial.append({"role": "assistant", "content": [
                b.model_dump(exclude_none=True) for b in r.content]})
        if not calls:
            eventos.append((paso, "respuesta"))
            return final, eventos, list(REGISTRO)
        bloques = []
        for cid, nombre, args in calls:
            res = ejecutar_seguro(nombre, args, vistos)
            tipo = "error" if "error" in res else "ok"
            eventos.append((paso, f"{nombre}:{tipo}"))
            if proveedor == "openai":
                historial.append({"type": "function_call_output",
                                  "call_id": cid,
                                  "output": json.dumps(res)})
            else:
                bloques.append({"type": "tool_result",
                                "tool_use_id": cid,
                                "content": json.dumps(res),
                                "is_error": "error" in res})
        if bloques:
            historial.append({"role": "user", "content": bloques})
    eventos.append((max_pasos, "corte"))
    return None, eventos, list(REGISTRO)


ESCENARIOS = ["correcto", "herramienta_equivocada",
              "argumentos_inventados", "bucle"]
for esc in ESCENARIOS:
    for prov in ("openai", "anthropic"):
        final, ev, tokens = agente_api(prov, esc)
        print(f"{esc:22s} {prov:9s} pasos={ev[-1][0]}"
              f" llamadas={sum(':' in e for _, e in ev)}"
              f" errores={sum(e.endswith('error') for _, e in ev)}"
              f" tokens={sum(tokens):5d} | {final}")

final, ev, tokens_bucle = agente_api("anthropic", "bucle", max_pasos=30)
_, _, tokens_ok = agente_api("anthropic", "correcto")
acum = np.cumsum(tokens_bucle)
print("tokens por petición, pasos 1-6:", tokens_bucle[:6])
print("pasos 25-30:", tokens_bucle[24:30])
print("acumulado tras 6, 12 y 30 pasos:", acum[5], acum[11], acum[29])
print("caso correcto, total:", sum(tokens_ok), "en", len(tokens_ok),
      "peticiones")