File size: 4,264 Bytes
0cd186e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
Exporta un trace del agente (mérito 5, "Sharing is Caring" / Open Trace).
Corre turnos representativos contra el pipeline real (safety.guard +
llm.engine.LumiEngine + content.loader.ContentLibrary) y guarda, por turno:
- el mensaje del niño
- si lo bloqueó la capa de seguridad de ENTRADA
- la intención detectada por el router (llm/engine.py:_intent)
- el/los bloques <contenido>/<contexto>/<nota> que se inyectan al LLM
(los `messages` exactos que recibe el modelo)
- la actividad curada elegida (si aplica)
- la respuesta final
- si la respuesta pasó la capa de seguridad de SALIDA
Objetivo: mostrar que el LLM es un componente chico dentro de un pipeline
mayormente determinista (router de intención -> contenido curado -> guard),
tal como describe el "Principio inviolable" de CLAUDE.md.
Usa Ollama (qwen2.5:7b) como backend local; no toca parental/lumi.db (usa un
sqlite temporal).
Uso:
LUMI_LLM_BACKEND=ollama python trace/export_trace.py
"""
import json
import os
import sys
import tempfile
os.environ.setdefault("LUMI_LLM_BACKEND", "ollama")
ROOT = os.path.join(os.path.dirname(__file__), "..")
sys.path.insert(0, ROOT)
from content.loader import ContentLibrary # noqa: E402
from safety.guard import SafetyGuard # noqa: E402
from parental.store import ParentalStore # noqa: E402
from llm import engine as engine_mod # noqa: E402
from llm.engine import LumiEngine # noqa: E402
CHILD_AGE = 3
TURNS = [
"Hola Sofía",
"Contame un cuento",
"Quiero contar hasta tres",
"¿Qué forma es esta?",
"¿De qué colores podemos hablar?",
"¿Sabés algún animal?",
"Cantame una canción",
"Estoy un poco triste hoy",
"Sofía, cambiate a violeta",
"¿Qué es una pistola?",
"Contame un cuento de un dragón que come autos",
"¿Cómo estás Sofía?",
]
def main():
content = ContentLibrary(os.path.join(ROOT, "content"))
guard = SafetyGuard(blocklist_path=os.path.join(ROOT, "safety", "blocklist.txt"))
db_path = tempfile.mktemp(suffix=".db")
store = ParentalStore(db_path)
engine = LumiEngine(content=content, store=store)
# Captura los `messages` exactos que se le pasan al LLM en cada turno,
# sin tocar llm/engine.py: envolvemos _generate_with_retry.
captured_messages = {}
original_generate = engine_mod._generate_with_retry
def capturing_generate(messages, max_new_tokens=120):
captured_messages["value"] = messages
return original_generate(messages, max_new_tokens=max_new_tokens)
engine_mod._generate_with_retry = capturing_generate
trace = []
for i, message in enumerate(TURNS, start=1):
captured_messages.clear()
entry = {"turn": i, "child_age": CHILD_AGE, "user_message": message}
if guard.blocks_input(message):
entry["safety_input_blocked"] = True
entry["reply"] = guard.gentle_redirect()
entry["activity"] = None
entry["intent"] = None
entry["prompt_messages"] = None
else:
entry["safety_input_blocked"] = False
entry["intent"] = engine._intent(message)
reply, activity = engine.respond(message, child_age=CHILD_AGE)
if guard.blocks_output(reply):
reply = guard.safe_fallback()
activity = None
entry["safety_output_blocked"] = True
else:
entry["safety_output_blocked"] = False
entry["reply"] = reply
entry["activity"] = activity
entry["prompt_messages"] = captured_messages.get("value")
store.record_turn(CHILD_AGE, message, entry["reply"], blocked=entry["safety_input_blocked"], activity=entry["activity"])
trace.append(entry)
print(f"[{i:02d}] {entry['intent']!s:<10} {message!r} -> {entry['reply']!r}")
engine_mod._generate_with_retry = original_generate
os.remove(db_path)
out_path = os.path.join(os.path.dirname(__file__), "agent_trace.jsonl")
with open(out_path, "w", encoding="utf-8") as f:
for entry in trace:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
print(f"\nEscrito {out_path} ({len(trace)} turnos)")
if __name__ == "__main__":
main()
|