Establishing A Context-Folding LLM Agent For Prolonged-Horizon Reasoning With Memory Compression And Software program Use

import os, re, sys, math, random, json, textwrap, subprocess, shutil, time
from typing import File, Dict, Tuple
attempt:
   import transformers
apart from:
   subprocess.run([sys.executable, "-m", "pip", "install", "-q", "transformers", "accelerate", "sentencepiece"], confirm=True)
from transformers import AutoTokenizer, AutoModelForSeq2SeqLM, pipeline
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-generation", model=model, tokenizer=tokenizer, device_map="auto")
def llm_gen(speedy: str, max_new_tokens=160, temperature=0.0) -> str:
   out = llm(speedy, max_new_tokens=max_new_tokens, do_sample=temperature>0.0, temperature=temperature)[0]["generated_text"]
   return out.strip()

import ast, operator as 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):
   if isinstance(n, ast.Num): return n.n
   if isinstance(n, ast.UnaryOp) and type(n.op) in OPS: return OPS[type(n.op)](_eval_node(n.operand))
   if isinstance(n, ast.BinOp) and type(n.op) in OPS: return OPS[type(n.op)](_eval_node(n.left), _eval_node(n.correct))
   improve ValueError("Unsafe expression")
def calc(expr: str):
   node = ast.parse(expr, mode="eval").physique
   return _eval_node(node)
class FoldingMemory:
   def __init__(self, max_chars:int=800):
       self.energetic=[]; self.folds=[]; self.max_chars=max_chars
   def add(self,textual content material:str):
       self.energetic.append(textual content material.strip())
       whereas len(self.active_text())>self.max_chars and len(self.energetic)>1:
           popped=self.energetic.pop(0)
           fold=f"- Folded: {popped[:120]}..."
           self.folds.append(fold)
   def fold_in(self,summary:str): self.folds.append(summary.strip())
   def active_text(self)->str: return "n".be a part of(self.energetic)
   def folded_text(self)->str: return "n".be a part of(self.folds)
   def snapshot(self)->Dict: return {"active_chars":len(self.active_text()),"n_folds":len(self.folds)}

SUBTASK_DECOMP_PROMPT="""You may be an educated planner. Decompose the obligation below into 2-4 crisp subtasks.
Return each subtask as a bullet starting with '- ' in priority order.
Course of: "{exercise}" """
SUBTASK_SOLVER_PROMPT="""You are a actual downside solver with minimal steps.
If a calculation is required, write one line 'CALC(expr)'.
In some other case write 'ANSWER: '.
Suppose briefly; steer clear of chit-chat.


Course of: {exercise}
Subtask: {subtask}
Notes (folded context):
{notes}


Now reply with each CALC(...) or ANSWER: ..."""
SUBTASK_SUMMARY_PROMPT="""Summarize the subtask finish end in <=3 bullets, entire <=50 tokens.
Subtask: {establish}
Steps:
{trace}
Remaining: {remaining}
Return solely bullets starting with '- '."""
FINAL_SYNTH_PROMPT="""You are a senior agent. Synthesize a remaining, coherent decision using ONLY:
- The distinctive exercise
- Folded summaries (below)
Stay away from repeating steps. Be concise and actionable.


Course of: {exercise}
Folded summaries:
{folds}


Remaining reply:"""
def parse_bullets(textual content material:str)->File[str]:
   return [ln[2:].strip() for ln in textual content material.splitlines() if ln.strip().startswith("- ")]

def run_subtask(exercise:str, subtask:str, memory:FoldingMemory, max_tool_iters:int=3)->Tuple[str,str,List[str]]:
   notes=(memory.folded_text() or "(none)")
   trace=[]; remaining=""
   for _ in range(max_tool_iters):
       speedy=SUBTASK_SOLVER_PROMPT.format(exercise=exercise,subtask=subtask,notes=notes)
       out=llm_gen(speedy,max_new_tokens=96); trace.append(out)
       m=re.search(r"CALC((.+?))",out)
       if m:
           attempt:
               val=calc(m.group(1))
               trace.append(f"TOOL:CALC -> {val}")
               out2=llm_gen(speedy+f"nTool finish consequence: {val}nNow produce 'ANSWER: ...' solely.",max_new_tokens=64)
               trace.append(out2)
               if out2.strip().startswith("ANSWER:"):
                   remaining=out2.break up("ANSWER:",1)[1].strip(); break
           apart from Exception as e:
               trace.append(f"TOOL:CALC ERROR -> {e}")
       if out.strip().startswith("ANSWER:"):
           remaining=out.break up("ANSWER:",1)[1].strip(); break
   if not remaining:
       remaining="No definitive reply; partial reasoning:n"+"n".be a part of(trace[-2:])
   summ=llm_gen(SUBTASK_SUMMARY_PROMPT.format(establish=subtask,trace="n".be a part of(trace),remaining=remaining),max_new_tokens=80)
   summary_bullets="n".be a part of(parse_bullets(summ)[:3]) or f"- {subtask}: {remaining[:60]}..."
   return remaining, summary_bullets, trace
class 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 decompose(self,exercise:str)->File[str]:
       plan=llm_gen(SUBTASK_DECOMP_PROMPT.format(exercise=exercise),max_new_tokens=96)
       subs=parse_bullets(plan)
       return subs[:4] if subs else ["Main solution"]
   def run(self,exercise:str)->Dict:
       t0=time.time()
       self.memory.add(f"TASK: {exercise}")
       subtasks=self.decompose(exercise)
       self.metrics["subtasks"]=len(subtasks)
       folded=[]
       for st in subtasks:
           self.memory.add(f"SUBTASK: {st}")
           remaining,fold_summary,trace=run_subtask(exercise,st,self.memory)
           self.memory.fold_in(fold_summary)
           folded.append(f"- {st}: {remaining}")
           self.memory.add(f"SUBTASK_DONE: {st}")
       remaining=llm_gen(FINAL_SYNTH_PROMPT.format(exercise=exercise,folds=self.memory.folded_text()),max_new_tokens=200)
       t1=time.time()
       return {"exercise":exercise,"remaining":remaining.strip(),"folded_summaries":self.memory.folded_text(),
               "active_context_chars":len(self.memory.active_text()),
               "subtask_finals":folded,"runtime_sec":spherical(t1-t0,2)}

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 pretty(d): return json.dumps(d, indent=2, ensure_ascii=False)
if __name__=="__main__":
   agent=ContextFoldingAgent(max_active_chars=700)
   for i,exercise in enumerate(DEMO_TASKS,1):
       print("="*70)
       print(f"DEMO #{i}: {exercise}")
       res=agent.run(exercise)
       print("n--- Folded Summaries ---n"+(res["folded_summaries"] or "(none)"))
       print("n--- Remaining Reply ---n"+res["final"])
       print("n--- Diagnostics ---")
       diag={okay:res[k] for okay in ["active_context_chars","runtime_sec"]}
       diag["n_subtasks"]=len(agent.decompose(exercise))
       print(pretty(diag))