{“role”: “system”, “content”: system},
{“role”: “user”, “content”: user},
]temperatura = 0,3) devolver opciones respectivas[0].message.content def heuristic_responder(contexto: str, pregunta: str) -> str: lecciones = re.findall(r”Lecciones=(.*)”, contexto) evitar = re.findall(r”Evitar=(.*)”, contexto) ltm_lines = [ln for ln in context.splitlines() if ln.startswith(“[LTM:”)]
pasos = []
si lecciones: para fragmentos de lecciones[:2]: para s en [x.strip() for x in chunk.split(“;”) if x.strip()]: pasos.append(s) para ln en ltm_lines: si “[LTM:procedure]” en ln.lower(): proc = re.sub(r”^\[LTM:procedure\]\s*”, “”, ln, banderas=re.I) proc = proc.split(“(salience=”)[0].strip() para parte en [p.strip() for p in proc.split(“|”) if p.strip()]: pasos.append(parte) pasos = pasos[:8] si hay otros pasos [“Clarify the target outcome and constraints.”, “Use semantic recall + episodic lessons to propose a plan.”, “Execute, then store lessons learned.”]
trampas = []
si evitar: para trozo en evitar[:2]: para s en [x.strip() for x in chunk.split(“;”) if x.strip()]: trampas.append(s) trampas = trampas[:6]
preferencias = [ln for ln in ltm_lines if “[LTM:preference]”en ln.lower()]hechos = [ln for ln in ltm_lines if “[LTM:fact]” en ln.lower() o “[LTM:constraint]”en ln.lower()]fuera = []
out.append(“Respuesta (reserva sin conexión basada en la memoria)\n”) if prefs: out.append(“Preferencias/restricciones relevantes recordadas:”) for ln in (prefs + hechos)[:6]: out.append(” – ” + ln.split(“]”,1)[1].split(” (prominencia=”)[0].strip()) out.append(“”) out.append(“Enfoque recomendado:”) for i, s in enumerate(steps, 1): out.append(f” {i}. {s}”) if trampas: out.append(“\nTrampas a evitar (de rastros episódicos):”) for p in trampas: out.append(” – ” + p) out.append(“\n(Si agrega una clave API, el mismo contexto de memoria alimentará un LLM más potente para obtener respuestas de mayor calidad.)”) return “\n”.join(out).strip() class MemoryAugmentedAgent: def __init__(self, mem: MemoryEngine): self.mem = mem def respuesta(self, question: str) -> Dict[str, Any]: pack = self.mem.retrieve(question) context = self.mem.build_context(question, pack) system = ( “Eres un agente con memoria aumentada. Usa el contexto de memoria proporcionado.\n” “Priorizar:\n” “1) Lecciones episódicas (lo que funcionó antes)\n” “2) Hechos/preferencias/procedimientos a largo plazo\n” “3) Estado de conversación a corto plazo\n” “Sea concreto y gradual. Si la memoria entra en conflicto, indique la incertidumbre”. ) if USE_OPENAI: respuesta = openai_chat(sistema=sistema, usuario=contexto + “\n\nPregunta de usuario:\n” + pregunta) else: respuesta = heuristic_responder(context=contexto, pregunta=pregunta) self.mem.st_add(“usuario”, pregunta, tipo=”mensaje”) self.mem.st_add(“asistente”, respuesta, tipo=”mensaje”) return {“respuesta”: respuesta, “pack”: pack, “context”: context} mem = MemoryEngine() agent = MemoryAugmentedAgent(mem) mem.ltm_add(kind=”preference”, text=”Prefiere respuestas concisas y estructuradas con pasos y viñetas cuando sean útiles.”, tags=[“style”]pinned=True) mem.ltm_add(kind=”preference”, text=”Prefiere soluciones que se ejecutan en Google Colab sin configuración adicional.”, etiquetas=[“environment”]pinned=True) mem.ltm_add(kind=”procedure”, text=”Al crear memoria de agente: incruste elementos, guárdelos con una política de prominencia/novedad, recupérelos con semántica híbrida+episódica y decaiga el uso excesivo para evitar repeticiones.”, tags=[“agent-memory”]) mem.ltm_add(kind=”constraint”, text=”Si no hay ninguna clave API disponible, proporcione una alternativa ejecutable sin conexión en lugar de fallar.”, etiquetas=[“robustness”]pinned=True) mem.episode_add( task=”Crear una capa de memoria de agente para solucionar errores de Python en Colab”, constraints={“offline_ok”: True, “single_notebook”: True}, plan=[
“Capture short-term chat context”,
“Store durable constraints/preferences in long-term vector memory”,
“After solving, extract lessons into episodic traces”,
“On new tasks, retrieve top episodic lessons + semantic facts”
]acciones =[
{“type”:”analysis”, “detail”:”Identified recurring failure: missing installs and version mismatches.”},
{“type”:”action”, “detail”:”Added pip install block + minimal fallbacks.”},
{“type”:”action”, “detail”:”Added memory policy: pin constraints, drop low-salience items.”}
]result=”El portátil se volvió robusto: se ejecuta con o sin claves externas; la calidad de la resolución de problemas mejoró con lecciones episódicas.”, result_score=0.90, lecciones=[
“Always include a pip install cell for non-standard deps.”,
“Pin hard constraints (e.g., offline fallback) into long-term memory.”,
“Store a post-task ‘lesson list’ as an episodic trace for reuse.”
]modos_fallo=[
“Assuming an API key exists and crashing when absent.”,
“Storing too much noise into long-term memory causing irrelevant recall context.”
]etiquetas =[“colab”,”robustness”,”memory”]
) print(“✅ Motor de memoria inicializado.”) print(f” Elementos LTM: {len(mem.ltm)} | Episodios: {len(mem.episodes)} | Elementos ST: {len(mem.short_term)}”) q1 = “Quiero crear memoria para un agente en Colab. ¿Qué debo almacenar y cómo lo recupero?” salida1 = agente.respuesta(q1) print(“\n” + “=”*90) print(“RESPUESTA Q1\n”) print(salida1[“reply”][:1800]) q2 = “¿Cómo evito que mi agente repita el mismo recuerdo una y otra vez?” salida2 = agente.respuesta(q2) print(“\n” + “=”*90) print(“RESPUESTA Q2\n”) print(salida2[“reply”][:1800]) def simple_outcome_eval(texto: str) -> float: hits = 0 para kw en [“decay”, “usage”, “penalty”, “novelty”, “prune”, “retrieve”, “episodic”, “semantic”]: si kw en text.lower(): visitas += 1 return float(np.clip(hits/8.0, 0.0, 1.0)) puntuación2 = simple_outcome_eval(out2[“reply”]) mem.episode_add( task=”Evitar la recuperación repetitiva en un agente con memoria aumentada”, constraints={“must_be_simple”: True, “runs_in_colab”: True}, plan=[
“Track usage counts per memory item”,
“Apply usage-based penalty during ranking”,
“Boost novelty during storage to reduce duplicates”,
“Optionally prune low-salience memories”
]acciones =[
{“type”:”design”, “detail”:”Added usage-based penalty 1/(1+alpha*usage).”},
{“type”:”design”, “detail”:”Used novelty = 1 – max_similarity at store time.”}
]resultado=out2[“reply”][:600]puntuación_resultado=puntuación2, lecciones=[
“Penalize overused memories during ranking (usage decay).”,
“Enforce novelty threshold at storage time to prevent duplicates.”,
“Keep episodic lessons distilled to avoid bloated recall context.”
]modos_fallo=[
“No usage tracking, causing one high-similarity memory to dominate forever.”,
“Storing raw chat logs as LTM instead of distilled summaries.”
]etiquetas =[“ranking”,”decay”,”policy”]
) cons = mem.consolidate() print(“\n” + “=”*90) print(“RESULTADO DE LA CONSOLIDACIÓN:”, cons) print(“\n” + “=”*90) print(“LTM (filas superiores):”) display(mem.ltm_df().head(12)) print(“\n” + “=”*90) print(“EPISODIOS (filas superiores):”) display(mem.episodes_df().head(12)) def debug_retrieval(consulta: str): paquete = mem.retrieve(consulta) ctx = mem.build_context(consulta, paquete) sem = []
para medio, pb en paquete[“semantic_scored”]: it = mem.ltm[mid]
sem.append({“mem_id”: mid, “score”: sc, “kind”: it.kind, “salience”: it.salience, “usage”: it.usage, “text”: it.text[:160]}) ep = []
para eid, sc en paquete[“episodic_scored”]: e = mem.episodios[eid]
ep.append({“ep_id”: eid, “score”: sc, “resultado”: e.outcome_score, “tarea”: e.task[:140]”lecciones”: ” | “.join(e.lecciones[:4])}) return ctx, pd.DataFrame(sem), pd.DataFrame(ep) print(“\n” + “=”*90) ctx, sem_df, ep_df = debug_retrieval(“¿Cómo diseño una política de memoria del agente para almacenamiento y recuperación?”) print(ctx[:1600]) print(“\nPrincipales visitas semánticas:”) display(sem_df) print(“\nPrincipales visitas episódicas:”) display(ep_df) print(“\n✅ Listo. Ahora tiene memoria episódica y vectorial a corto y largo plazo en funcionamiento con políticas de almacenamiento/recuperación en un fragmento de Colab.”)