Creación de un agente LLM que pliega el contexto para el razonamiento a largo plazo con compresión de memoria y uso de herramientas

En este tutorial, exploramos cómo crear un agente LLM de plegado de contexto que resuelva de manera eficiente tareas largas y complejas mediante la gestión inteligente de un contexto limitado. Diseñamos el agente para dividir una tarea grande en subtareas más pequeñas, realizar razonamientos o cálculos cuando sea necesario y luego plegar cada subtrayectoria completa en resúmenes concisos. Al hacer esto, preservamos el conocimiento esencial manteniendo pequeña la memoria activa. Consulta los CÓDIGOS COMPLETOS aquí.

import os, re, sys, math, random, json, textwrap, subprocess, Shutil, tiempo desde que se escribe import List, Dict, Tuple try: importar transformadores excepto: subprocess.run([sys.executable, “-m”, “pip”, “install”, “-q”, “transformers”, “accelerate”, “sentencepiece”]check=True) de transformadores importa AutoTokenizer, AutoModelForSeq2SeqLM, canalización MODEL_NAME = os.environ.get(“CF_MODEL”, “google/flan-t5-small”) tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) model = AutoModelForSeq2SeqLM.from_pretrained(MODEL_NAME) llm = pipeline(“text2text-generación”, modelo=modelo, tokenizer=tokenizer, dispositivo_map=”auto”) def llm_gen(mensaje: str, max_new_tokens=160, temperatura=0.0) -> str: out = llm(mensaje, max_new_tokens=max_new_tokens, do_sample=temperatura>0.0, temperatura=temperatura)[0][“generated_text”]

volver fuera.strip()

Comenzamos configurando nuestro entorno y cargando un modelo ligero de Hugging Face. Usamos este modelo para generar y procesar texto localmente, asegurando que el agente funcione sin problemas en Google Colab sin dependencias de API. Consulta los CÓDIGOS COMPLETOS aquí.

importar ast, operador como op OPS = {ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv, ast.Pow: op.pow, ast.USub: op.neg, ast.FloorDiv: op.floordiv, ast.Mod: op.mod} def _eval_node(n): si isinstance(n, ast.Num): devuelve nn si isinstance(n, ast.UnaryOp) y escribe (n.op) en OPS: devuelve OPS[type(n.op)](_eval_node(n.operand)) si esinstance(n, ast.BinOp) y tipo(n.op) en OPS: devuelve OPS[type(n.op)](_eval_node(n.left), _eval_node(n.right)) rise ValueError(“Expresión insegura”) def calc(expr: str): node = ast.parse(expr, mode=”eval”).body return _eval_node(node) class FoldingMemory: def __init__(self, max_chars:int=800): self.active=[]; self.pliegues=[]; self.max_chars=max_chars def add(self,text:str): self.active.append(text.strip()) mientras len(self.active_text())>self.max_chars y len(self.active)>1: popped=self.active.pop(0) fold=f”- Plegado: {apareció[:120]}…” self.folds.append(fold) def fold_in(self,summary:str): self.folds.append(summary.strip()) def active_text(self)->str: return “\n”.join(self.active) def fold_text(self)->str: return “\n”.join(self.folds) def snapshot(self)->Dict: return {“active_chars”:len(self.active_text()),”n_folds”:len(self.folds)}

Definimos una herramienta de calculadora simple para aritmética básica y creamos un sistema de memoria que pliega dinámicamente el contexto pasado en resúmenes concisos. Esto nos ayuda a mantener una memoria activa manejable mientras retenemos información esencial. Consulta los CÓDIGOS COMPLETOS aquí.

SUBTASK_DECOMP_PROMPT=”””Eres un planificador experto. Descompone la siguiente tarea en 2 a 4 subtareas claras. Devuelve cada subtarea como una viñeta que comienza con ‘- ‘ en orden de prioridad. Tarea: “{task}” “”” SUBTASK_SOLVER_PROMPT=”””Eres un solucionador de problemas preciso con pasos mínimos. Si se necesita un cálculo, escribe una línea ‘CALC(expr)’. De lo contrario, escribe ‘RESPUESTA: ‘. Piense brevemente; Evite la charla. Tarea: {tarea} Subtarea: {subtarea} Notas (contexto plegado): {notas} Ahora responda con CALC(…) o RESPUESTA: …””” SUBTASK_SUMMARY_PROMPT=”””Resuma el resultado de la subtarea en <=3 viñetas, en total <=50 tokens. Subtarea: {nombre} Pasos: {trace} Final: {final} Devuelve solo viñetas que comienzan con '- '.""" FINAL_SYNTH_PROMPT="""Eres un agente senior. Sintetizar una solución final coherente utilizando SÓLO: - La tarea original - Resúmenes doblados (abajo) Evite repetir pasos. Sea conciso y práctico. Tarea: {task} Resúmenes plegados: {folds} Respuesta final:""" def parse_bullets(text:str)->List[str]: devolver [ln[2:].strip() para ln en text.splitlines() si ln.strip().startswith(“- “)]

Diseñamos plantillas de indicaciones que guían al agente a descomponer tareas, resolver subtareas y resumir resultados. Estas indicaciones estructuradas permiten una comunicación clara entre los pasos de razonamiento y las respuestas del modelo. Consulta los CÓDIGOS COMPLETOS aquí.

def run_subtask(tarea:cadena, subtarea:cadena, memoria:FoldingMemory, max_tool_iters:int=3)->Tupla[str,str,List[str]]: notas=(memoria.folded_text() o “(ninguno)”) trace=[]; final=”” para _ en rango(max_tool_iters): aviso=SUBTASK_SOLVER_PROMPT.format(tarea=tarea,subtarea=subtarea,notas=notas) out=llm_gen(prompt,max_new_tokens=96); trace.append(out) m=re.search(r”CALC\((.+?)\)”,out) if m: intente: val=calc(m.group(1)) trace.append(f”TOOL:CALC -> {val}”) out2=llm_gen(prompt+f”\nResultado de la herramienta: {val}\nAhora produzca ‘RESPUESTA: …’ solamente.”,max_new_tokens=64) trace.append(out2) if out2.strip().startswith(“RESPUESTA:”): final=out2.split(“RESPUESTA:”,1)[1].banda(); romper excepto Excepción como e: trace.append(f”HERRAMIENTA: ERROR DE CALC -> {e}”) if out.strip().startswith(“RESPUESTA:”): final=out.split(“RESPUESTA:”,1)[1].banda(); romper si no es final: final=”Sin respuesta definitiva; razonamiento parcial:\n”+”\n”.join(trace[-2:]) summ=llm_gen(SUBTASK_SUMMARY_PROMPT.format(nombre=subtarea,trace=”\n”.join(trace),final=final),max_new_tokens=80) resumen_bullets=”\n”.join(parse_bullets(summ)[:3]) o f”- {subtarea}: {final[:60]}…” return final, resumen_bullets, clase de seguimiento ContextFoldingAgent: def __init__(self,max_active_chars:int=800): self.memory=FoldingMemory(max_chars=max_active_chars) self.metrics={“subtasks”:0,”tool_calls”:0,”chars_saved_est”:0} def descomponer(self,tarea:str)->Lista[str]: plan=llm_gen(SUBTASK_DECOMP_PROMPT.format(tarea=tarea),max_new_tokens=96) subs=parse_bullets(plan) devolver subs[:4] si subs más [“Main solution”]
def ejecutar(self,tarea:str)->Dict: t0=time.time() self.memory.add(f”TASK: {tarea}”) subtasks=self.decompose(tarea) self.metrics[“subtasks”]=len(subtareas) plegadas=[]
para st en subtareas: self.memory.add(f”SUBTASK: {st}”) final,fold_summary,trace=run_subtask(task,st,self.memory) self.memory.fold_in(fold_summary) fold.append(f”- {st}: {final}”) self.memory.add(f”SUBTASK_DONE: {st}”) final=llm_gen(FINAL_SYNTH_PROMPT.format(task=task,folds=self.memory.folded_text()),max_new_tokens=200) t1=time.time() return {“task”:task,”final”:final.strip(),”folded_summaries”:self.memory.folded_text(), “active_context_chars”:len(self.memory.active_text()), “subtask_finals”:folded,”runtime_sec”:round(t1-t0,2)}

Implementamos la lógica central del agente, en la que cada subtarea se ejecuta, se resume y se vuelve a plegar en la memoria. Este paso demuestra cómo el plegado del contexto permite al agente razonar de forma iterativa sin perder de vista el razonamiento previo. Consulta los CÓDIGOS COMPLETOS aquí.

DEMO_TASKS=[
“Plan a 3-day study schedule for ML with daily workouts and simple meals; include time blocks.”,
“Compute a small project budget with 3 items (laptop 799.99, course 149.5, snacks 23.75), add 8% tax and 5% buffer, and present a one-paragraph recommendation.”
]
def bonita(d): devuelve json.dumps(d, sangría=2, asegurar_ascii=False) si __name__==”__main__”: agente=ContextFoldingAgent(max_active_chars=700) para i, tarea en enumerar(DEMO_TASKS,1): print(“=”*70) print(f”DEMO #{i}: {tarea}”) res=agent.run(tarea) print(“\n— Resúmenes plegados —\n”+(res[“folded_summaries”] o “(ninguno)”)) print(“\n— Respuesta final —\n”+res[“final”]) print(“\n— Diagnóstico —“) diag={k:res[k] para k en [“active_context_chars”,”runtime_sec”]} diagnóstico[“n_subtasks”]=len(agente.descomponer(tarea)) imprimir(bonita(diag))

Ejecutamos el agente en tareas de muestra para observar cómo planifica, ejecuta y sintetiza los resultados finales. A través de estos ejemplos, vemos el proceso completo de plegado del contexto en acción, produciendo resultados concisos y coherentes.

En conclusión, demostramos cómo el plegamiento del contexto permite el razonamiento a largo plazo evitando al mismo tiempo la sobrecarga de memoria. Vemos cómo se planifica, ejecuta, resume y destila cada subtarea en conocimiento compacto, imitando cómo un agente inteligente manejaría flujos de trabajo complejos a lo largo del tiempo. Al combinar descomposición, uso de herramientas y compresión de contexto, creamos un sistema agente liviano pero poderoso que escala el razonamiento de manera eficiente.

Consulte los CÓDIGOS COMPLETOS aquí y en papel. No dude en consultar nuestra página de GitHub para tutoriales, códigos y cuadernos. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.

Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como empresario e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.

🙌 Siga MARKTECHPOST: agréguenos como fuente preferida en Google.