Constructing a Context-Folding LLM Agent for Lengthy-Horizon Reasoning with Reminiscence Compression and Software Use

import os, re, sys, math, random, json, textwrap, subprocess, shutil, time
from typing import Record, Dict, Tuple
strive:
   import transformers
besides:
   subprocess.run([sys.executable, "-m", "pip", "install", "-q", "transformers", "accelerate", "sentencepiece"], verify=True)
from transformers import AutoTokenizer, AutoModelForSeq2SeqLM, pipeline
MODEL_NAME = os.environ.get("CF_MODEL", "google/flan-t5-small")
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
mannequin = AutoModelForSeq2SeqLM.from_pretrained(MODEL_NAME)
llm = pipeline("text2text-generation", mannequin=mannequin, tokenizer=tokenizer, device_map="auto")
def llm_gen(immediate: str, max_new_tokens=160, temperature=0.0) -> str:
   out = llm(immediate, 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 sort(n.op) in OPS: return OPS[type(n.op)](_eval_node(n.operand))
   if isinstance(n, ast.BinOp) and sort(n.op) in OPS: return OPS[type(n.op)](_eval_node(n.left), _eval_node(n.proper))
   increase 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:str):
       self.energetic.append(textual content.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,abstract:str): self.folds.append(abstract.strip())
   def active_text(self)->str: return "n".be part of(self.energetic)
   def folded_text(self)->str: return "n".be part of(self.folds)
   def snapshot(self)->Dict: return {"active_chars":len(self.active_text()),"n_folds":len(self.folds)}

SUBTASK_DECOMP_PROMPT="""You might be an knowledgeable planner. Decompose the duty under into 2-4 crisp subtasks.
Return every subtask as a bullet beginning with '- ' in precedence order.
Process: "{activity}" """
SUBTASK_SOLVER_PROMPT="""You're a exact drawback solver with minimal steps.
If a calculation is required, write one line 'CALC(expr)'.
In any other case write 'ANSWER: '.
Suppose briefly; keep away from chit-chat.


Process: {activity}
Subtask: {subtask}
Notes (folded context):
{notes}


Now reply with both CALC(...) or ANSWER: ..."""
SUBTASK_SUMMARY_PROMPT="""Summarize the subtask end result in <=3 bullets, whole <=50 tokens.
Subtask: {identify}
Steps:
{hint}
Remaining: {remaining}
Return solely bullets beginning with '- '."""
FINAL_SYNTH_PROMPT="""You're a senior agent. Synthesize a remaining, coherent resolution utilizing ONLY:
- The unique activity
- Folded summaries (under)
Keep away from repeating steps. Be concise and actionable.


Process: {activity}
Folded summaries:
{folds}


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

def run_subtask(activity:str, subtask:str, reminiscence:FoldingMemory, max_tool_iters:int=3)->Tuple[str,str,List[str]]:
   notes=(reminiscence.folded_text() or "(none)")
   hint=[]; remaining=""
   for _ in vary(max_tool_iters):
       immediate=SUBTASK_SOLVER_PROMPT.format(activity=activity,subtask=subtask,notes=notes)
       out=llm_gen(immediate,max_new_tokens=96); hint.append(out)
       m=re.search(r"CALC((.+?))",out)
       if m:
           strive:
               val=calc(m.group(1))
               hint.append(f"TOOL:CALC -> {val}")
               out2=llm_gen(immediate+f"nTool end result: {val}nNow produce 'ANSWER: ...' solely.",max_new_tokens=64)
               hint.append(out2)
               if out2.strip().startswith("ANSWER:"):
                   remaining=out2.break up("ANSWER:",1)[1].strip(); break
           besides Exception as e:
               hint.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 part of(hint[-2:])
   summ=llm_gen(SUBTASK_SUMMARY_PROMPT.format(identify=subtask,hint="n".be part of(hint),remaining=remaining),max_new_tokens=80)
   summary_bullets="n".be part of(parse_bullets(summ)[:3]) or f"- {subtask}: {remaining[:60]}..."
   return remaining, summary_bullets, hint
class ContextFoldingAgent:
   def __init__(self,max_active_chars:int=800):
       self.reminiscence=FoldingMemory(max_chars=max_active_chars)
       self.metrics={"subtasks":0,"tool_calls":0,"chars_saved_est":0}
   def decompose(self,activity:str)->Record[str]:
       plan=llm_gen(SUBTASK_DECOMP_PROMPT.format(activity=activity),max_new_tokens=96)
       subs=parse_bullets(plan)
       return subs[:4] if subs else ["Main solution"]
   def run(self,activity:str)->Dict:
       t0=time.time()
       self.reminiscence.add(f"TASK: {activity}")
       subtasks=self.decompose(activity)
       self.metrics["subtasks"]=len(subtasks)
       folded=[]
       for st in subtasks:
           self.reminiscence.add(f"SUBTASK: {st}")
           remaining,fold_summary,hint=run_subtask(activity,st,self.reminiscence)
           self.reminiscence.fold_in(fold_summary)
           folded.append(f"- {st}: {remaining}")
           self.reminiscence.add(f"SUBTASK_DONE: {st}")
       remaining=llm_gen(FINAL_SYNTH_PROMPT.format(activity=activity,folds=self.reminiscence.folded_text()),max_new_tokens=200)
       t1=time.time()
       return {"activity":activity,"remaining":remaining.strip(),"folded_summaries":self.reminiscence.folded_text(),
               "active_context_chars":len(self.reminiscence.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 fairly(d): return json.dumps(d, indent=2, ensure_ascii=False)
if __name__=="__main__":
   agent=ContextFoldingAgent(max_active_chars=700)
   for i,activity in enumerate(DEMO_TASKS,1):
       print("="*70)
       print(f"DEMO #{i}: {activity}")
       res=agent.run(activity)
       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(activity))
       print(fairly(diag))