Asistente de laboratorio con LLM

# Protocolo: extracción de ADN genómico por columna de sílice

Material de partida: 200 µL de sangre total con EDTA, o hasta 25 mg de tejido.

## Lisis
- Paso 3. Pipetear 20 µL de proteinasa K en un tubo de 1,5 mL.
- Paso 4. Agregar 200 µL de muestra y 200 µL de buffer AL. Mezclar con
  vórtex durante 15 segundos. No agregar la proteinasa K directamente al
  buffer AL.
- Paso 5. Incubar a 56 °C durante 10 minutos en el baño seco.
…
## Elución
- Paso 14. Agregar 100 µL de buffer AE al centro de la membrana. Incubar a
  temperatura ambiente durante 5 minutos.
…
- Paso 18. Almacenar el ADN a 4 °C si se usará en la semana, o a -20 °C
  para almacenamiento prolongado.

Eres el asistente de un laboratorio de biología molecular. Respondes preguntas
sobre el protocolo que se te entrega dentro de la etiqueta <protocolo>.

Reglas:
- Responde SOLO con información que esté en el protocolo. No inventes pasos,
  tiempos, volúmenes ni temperaturas.
- Si la respuesta no está en el protocolo, dilo explícitamente y usa
  confianza "baja" y fuente null.
- "fuente" es el número o título del paso del protocolo del que sale la
  respuesta (por ejemplo "Paso 4").
- "confianza" es "alta" si la respuesta está literal en el protocolo,
  "media" si requiere interpretación, "baja" si no está.
- Sé breve: una o dos frases.

"""Asistente de laboratorio: responde preguntas sobre un protocolo usando la API de Claude."""

import json
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, ValidationError

# Precio de lista de claude-sonnet-5 (USD por token), agosto 2026.
# Si cambias de modelo, cambia también esto: el costo es parte del contrato.
PRECIO_INPUT_USD = 3.00 / 1_000_000
PRECIO_OUTPUT_USD = 15.00 / 1_000_000

MODELO_DEFAULT = "claude-sonnet-5"
MAX_TOKENS = 500

_PROMPT_PATH = Path(__file__).parent / "prompts" / "asistente.md"

# El esquema que el modelo DEBE cumplir (salida estructurada).
SCHEMA = {
    "type": "object",
    "properties": {
        "respuesta": {"type": "string"},
        "confianza": {"type": "string", "enum": ["alta", "media", "baja"]},
        "fuente": {"type": ["string", "null"]},
    },
    "required": ["respuesta", "confianza", "fuente"],
    "additionalProperties": False,
}


class RespuestaJSON(BaseModel):
    """Validación del JSON que devuelve el modelo."""

    respuesta: str
    confianza: Literal["alta", "media", "baja"]
    fuente: str | None


@dataclass(frozen=True)
class Respuesta:
    """Lo que devuelve el asistente: la respuesta y su costo."""

    texto: str
    confianza: str
    fuente: str | None
    tokens_in: int
    tokens_out: int
    costo_usd: float


class AsistenteError(Exception):
    """Error base del asistente."""


class FormatoInvalido(AsistenteError):
    """El modelo no devolvió el JSON acordado."""


class RespuestaIncompleta(AsistenteError):
    """El modelo se quedó sin tokens antes de terminar (stop_reason=max_tokens)."""


def cargar_prompt() -> str:
    """El prompt vive en un archivo versionado, no en el código."""
    return _PROMPT_PATH.read_text(encoding="utf-8")


def responder(pregunta: str, contexto: str, cliente=None) -> Respuesta:
    """Pregunta al modelo sobre el protocolo dado.

    `cliente` se inyecta en los tests (cliente falso). En producción se crea
    el cliente real de Anthropic, que lee ANTHROPIC_API_KEY del entorno.
    """
    if not pregunta.strip():
        raise ValueError("la pregunta está vacía")
    if cliente is None:
        import anthropic

        cliente = anthropic.Anthropic(timeout=30.0)

    modelo = os.environ.get("ASISTENTE_MODEL", MODELO_DEFAULT)
    respuesta_api = cliente.messages.create(
        model=modelo,
        max_tokens=MAX_TOKENS,
        system=cargar_prompt(),
        output_config={"format": {"type": "json_schema", "schema": SCHEMA}},
        messages=[
            {
                "role": "user",
                "content": f"<protocolo>\n{contexto}\n</protocolo>\n\nPregunta: {pregunta}",
            }
        ],
    )

    if respuesta_api.stop_reason == "max_tokens":
        raise RespuestaIncompleta(
            f"el modelo agotó los {MAX_TOKENS} tokens; la respuesta llegó cortada"
        )

    texto_crudo = next(
        (bloque.text for bloque in respuesta_api.content if bloque.type == "text"), ""
    )
    try:
        datos = RespuestaJSON.model_validate(json.loads(texto_crudo))
    except (json.JSONDecodeError, ValidationError) as err:
        raise FormatoInvalido(f"el modelo no devolvió el JSON acordado: {texto_crudo!r}") from err

    tokens_in = respuesta_api.usage.input_tokens
    tokens_out = respuesta_api.usage.output_tokens
    return Respuesta(
        texto=datos.respuesta,
        confianza=datos.confianza,
        fuente=datos.fuente,
        tokens_in=tokens_in,
        tokens_out=tokens_out,
        costo_usd=tokens_in * PRECIO_INPUT_USD + tokens_out * PRECIO_OUTPUT_USD,
    )

import json
from types import SimpleNamespace

import pytest


def hacer_cliente_falso(cuerpo: dict | str, stop_reason: str = "end_turn"):
    """Cliente que devuelve siempre el mismo cuerpo. Registra la última llamada."""
    texto = cuerpo if isinstance(cuerpo, str) else json.dumps(cuerpo, ensure_ascii=False)
    llamadas = []

    def create(**kwargs):
        llamadas.append(kwargs)
        return SimpleNamespace(
            stop_reason=stop_reason,
            content=[SimpleNamespace(type="text", text=texto)],
            usage=SimpleNamespace(input_tokens=1000, output_tokens=100),
        )

    cliente = SimpleNamespace(messages=SimpleNamespace(create=create))
    cliente.llamadas = llamadas
    return cliente


@pytest.fixture
def cliente_ok():
    return hacer_cliente_falso(
        {"respuesta": "Se incuba a 56 °C.", "confianza": "alta", "fuente": "Paso 5"}
    )

import pytest

from inventario.asistente import (
    PRECIO_INPUT_USD,
    PRECIO_OUTPUT_USD,
    FormatoInvalido,
    RespuestaIncompleta,
    responder,
)
from tests.conftest import hacer_cliente_falso

CONTEXTO = "Paso 5. Incubar a 56 °C durante 10 minutos."


def test_responder_devuelve_respuesta_estructurada(cliente_ok):
    r = responder("¿A qué temperatura se incuba?", CONTEXTO, cliente=cliente_ok)
    assert r.texto == "Se incuba a 56 °C."
    assert r.confianza == "alta"
    assert r.fuente == "Paso 5"


def test_responder_calcula_el_costo(cliente_ok):
    r = responder("¿A qué temperatura se incuba?", CONTEXTO, cliente=cliente_ok)
    assert r.tokens_in == 1000
    assert r.tokens_out == 100
    assert r.costo_usd == pytest.approx(1000 * PRECIO_INPUT_USD + 100 * PRECIO_OUTPUT_USD)


def test_responder_manda_el_protocolo_y_el_prompt(cliente_ok):
    responder("¿Cuánto dura?", CONTEXTO, cliente=cliente_ok)
    llamada = cliente_ok.llamadas[0]
    assert CONTEXTO in llamada["messages"][0]["content"]
    assert "No inventes" in llamada["system"]
    assert llamada["output_config"]["format"]["type"] == "json_schema"


def test_pregunta_vacia_lanza_error(cliente_ok):
    with pytest.raises(ValueError):
        responder("   ", CONTEXTO, cliente=cliente_ok)


def test_json_invalido_lanza_formato_invalido():
    cliente = hacer_cliente_falso("esto no es JSON")
    with pytest.raises(FormatoInvalido):
        responder("¿Cuánto dura?", CONTEXTO, cliente=cliente)


def test_confianza_fuera_del_enum_lanza_formato_invalido():
    cliente = hacer_cliente_falso(
        {"respuesta": "56 °C", "confianza": "altisima", "fuente": None}
    )
    with pytest.raises(FormatoInvalido):
        responder("¿Cuánto dura?", CONTEXTO, cliente=cliente)


def test_max_tokens_lanza_respuesta_incompleta():
    cliente = hacer_cliente_falso(
        {"respuesta": "cortada", "confianza": "alta", "fuente": None},
        stop_reason="max_tokens",
    )
    with pytest.raises(RespuestaIncompleta):
        responder("¿Cuánto dura?", CONTEXTO, cliente=cliente)

from pathlib import Path
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel

from inventario.asistente import AsistenteError, Respuesta, responder

RAIZ = Path(__file__).resolve().parent.parent.parent.parent
PROTOCOLO = RAIZ / "data" / "protocolo-extraccion-adn.md"


def get_cliente_llm():
    """En producción devuelve None (asistente crea el cliente real).

    En tests se sobreescribe con dependency_overrides para inyectar el falso.
    """
    return


class PreguntaIn(BaseModel):
    pregunta: str


class PreguntaOut(BaseModel):
    respuesta: str
    confianza: str
    fuente: str | None
    tokens_in: int
    tokens_out: int
    costo_usd: float


@app.post("/preguntar", response_model=PreguntaOut)
def preguntar(
    datos: PreguntaIn, cliente: Annotated[object, Depends(get_cliente_llm)]
) -> PreguntaOut:
    contexto = PROTOCOLO.read_text(encoding="utf-8")
    try:
        r: Respuesta = responder(datos.pregunta, contexto, cliente=cliente)
    except ValueError as err:
        raise HTTPException(status_code=422, detail=str(err)) from err
    except AsistenteError as err:
        # El modelo falló (formato, tokens): no es culpa del cliente -> 502
        raise HTTPException(status_code=502, detail=str(err)) from err
    return PreguntaOut(
        respuesta=r.texto,
        confianza=r.confianza,
        fuente=r.fuente,
        tokens_in=r.tokens_in,
        tokens_out=r.tokens_out,
        costo_usd=r.costo_usd,
    )

from fastapi.testclient import TestClient

from inventario.api.main import app, get_cliente_llm
from tests.conftest import hacer_cliente_falso


def cliente_de_prueba(cuerpo):
    falso = hacer_cliente_falso(cuerpo)
    app.dependency_overrides[get_cliente_llm] = lambda: falso
    return TestClient(app)


def test_preguntar_feliz():
    tc = cliente_de_prueba(
        {"respuesta": "Se incuba a 56 °C.", "confianza": "alta", "fuente": "Paso 5"}
    )
    r = tc.post("/preguntar", json={"pregunta": "¿A qué temperatura se incuba la lisis?"})
    assert r.status_code == 200
    datos = r.json()
    assert datos["respuesta"] == "Se incuba a 56 °C."
    assert datos["fuente"] == "Paso 5"
    assert datos["costo_usd"] > 0


def test_preguntar_cuerpo_invalido_es_422():
    tc = cliente_de_prueba({"respuesta": "x", "confianza": "alta", "fuente": None})
    r = tc.post("/preguntar", json={})
    assert r.status_code == 422


def test_modelo_rompe_el_formato_es_502():
    tc = cliente_de_prueba("no soy JSON")
    r = tc.post("/preguntar", json={"pregunta": "¿Cuánto dura la lisis?"})
    assert r.status_code == 502

[
  {"id": 1,  "pregunta": "¿A qué temperatura se incuba la lisis?",
   "esperado": ["56"], "fuente_esperada": "Paso 5"},
  {"id": 2,  "pregunta": "¿Cuánto tiempo dura la incubación de lisis?",
   "esperado": ["10 min"], "fuente_esperada": "Paso 5"},
  {"id": 5,  "pregunta": "¿Se puede agregar la proteinasa K directamente al buffer AL?",
   "esperado": ["no"], "fuente_esperada": "Paso 4"},
  {"id": 16, "pregunta": "¿Qué relación A260/A280 indica ADN limpio?",
   "esperado": ["1,7", "1.7"], "fuente_esperada": "Paso 17"},
  {"id": 20, "pregunta": "¿Qué primers uso para amplificar el gen 16S?",
   "esperado": ["no está", "no aparece", "no lo indica", "no se indica", "no figura"],
   "fuente_esperada": null}
]

"""Evaluación del asistente: 20 preguntas con respuesta conocida.

Uso:
    uv run python -m inventario.evals          # contra la API real (necesita ANTHROPIC_API_KEY)
    uv run python -m inventario.evals --fake   # cliente falso: prueba la tubería sin gastar

Imprime el % de respuestas correctas y el costo total. Antes de cambiar el
prompt, corre esto y anota los números; después de cambiarlo, corre de nuevo.
Se decide con números, no con una pregunta de ejemplo.
"""

import argparse
import json
import sys
from pathlib import Path
from types import SimpleNamespace

from inventario.asistente import AsistenteError, responder

RAIZ = Path(__file__).resolve().parent.parent.parent
CASOS = RAIZ / "evals" / "casos.json"
PROTOCOLO = RAIZ / "data" / "protocolo-extraccion-adn.md"


class ClienteFalso:
    """Responde desde el propio caso: sirve para probar la tubería, no el modelo."""

    def __init__(self, casos: list[dict]):
        self._por_pregunta = {c["pregunta"]: c for c in casos}
        self.messages = SimpleNamespace(create=self._create)

    def _create(self, **kwargs):
        contenido_usuario = kwargs["messages"][0]["content"]
        caso = next(
            (c for p, c in self._por_pregunta.items() if p in contenido_usuario),
            None,
        )
        if caso is None:
            cuerpo = {"respuesta": "No está en el protocolo.", "confianza": "baja", "fuente": None}
        else:
            cuerpo = {
                "respuesta": f"Según el protocolo: {caso['esperado'][0]}.",
                "confianza": "alta" if caso["fuente_esperada"] else "baja",
                "fuente": caso["fuente_esperada"],
            }
        return SimpleNamespace(
            stop_reason="end_turn",
            content=[SimpleNamespace(type="text", text=json.dumps(cuerpo, ensure_ascii=False))],
            usage=SimpleNamespace(input_tokens=650, output_tokens=60),
        )


def es_correcta(texto: str, esperado: list[str]) -> bool:
    """Correcta si alguna de las palabras clave esperadas aparece en la respuesta."""
    texto = texto.lower()
    return any(clave.lower() in texto for clave in esperado)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--fake", action="store_true", help="usa el cliente falso (sin costo)")
    args = parser.parse_args()

    casos = json.loads(CASOS.read_text(encoding="utf-8"))
    contexto = PROTOCOLO.read_text(encoding="utf-8")
    cliente = ClienteFalso(casos) if args.fake else None

    correctas = 0
    costo_total = 0.0
    for caso in casos:
        try:
            r = responder(caso["pregunta"], contexto, cliente=cliente)
        except AsistenteError as err:
            print(f"[{caso['id']:>2}] ERROR  {caso['pregunta']}  ({err})")
            continue
        ok = es_correcta(r.texto, caso["esperado"])
        correctas += ok
        costo_total += r.costo_usd
        marca = "ok " if ok else "MAL"
        print(f"[{caso['id']:>2}] {marca}  {caso['pregunta']}")
        if not ok:
            print(f"        esperaba {caso['esperado']!r}, respondió: {r.texto!r}")

    n = len(casos)
    print("-" * 60)
    print(f"correctas: {correctas}/{n} ({100 * correctas / n:.0f}%)")
    print(f"costo total: ${costo_total:.4f} USD ({'cliente falso' if args.fake else 'API real'})")
    return 0 if correctas == n else 1


if __name__ == "__main__":
    sys.exit(main())

## 2026-08-xx · Prompt del asistente v2
Hipótesis: fallaba tiempos por responder en palabras ("tres minutos").
Cambio: regla "escribe cantidades con número y unidad (3 min, 56 °C)".
Antes:  18/20 (90%), $0.0980 por corrida.
Después: 20/20 (100%), $0.0995 por corrida.
Se queda. Nota: +$0.0015 por respuestas un poco más largas.

# Por defecto: tests + evaluación con cliente falso (sin API key, sin costo)
CMD ["sh", "-c", "uv run pytest -q && uv run python -m inventario.evals --fake"]