How to Build a Fully Functional Computer-Use Agent that Thinks, Plans, and Executes Virtual Actions Using Local AI Models

!pip set up -q Transformer accelerates statements nest_asyncio Import torch, asyncio, uuid from transformer Import pipeline import net_asyncio nest_asyncio.apply()

class LocalLLM: def __init__(self, model_name=”google/flan-t5-small”, max_new_tokens=128): self.pipe = Pipeline(“text2text-generation”, mannequin=model_name, gadget=0 if torch.cuda.is_available() else -1) self.max_new_tokens = max_new_tokens def generate(self, immediate: str) -> str: out = self.pipe(immediate, max_new_tokens=self.max_new_tokens, temperature=0.0)[0][“generated_text”]

return out.strip() class VirtualComputer: def __init__(self): self.apps = {“browser”: “https://instance.com”, “notes”: “”, “mail”: [“Welcome to CUA”, “Invoice #221”, “Weekly Report”]} self.focus = “browser” self.display = “A browser opens at https://instance.comnThe search bar is in focus.” ” self.action_log = []
def screenshot(self): return f”FOCUS:{self.focus}nSCREEN:n{self.display}nAPPS:{checklist(self.apps.keys())}” def click on(self, goal:str): if goal in self.apps: self.focus = goal if goal==”browser”: self.display = f”Browser tab: {self.apps[‘browser’]}nAddress bar is concentrated. ” elif goal==”notes”: self.display = f”Notes AppnCurrent notes:n{self.apps[‘notes’]}” elif goal==”mail”: inbox = “n”.be part of(f”- {s}” for s in self.apps)[‘mail’]) self.display = f”Mail app inbox:n{inbox}n(read-only preview)” else: self.display += f”nClicked “{goal}.” self.action_log.append({“kind”:”click on”,”goal”:goal}) def kind(self, textual content:str): if self.focus==”browser”: self.apps[“browser”] = textual content self.display = f”Browser tab is positioned at {textual content}nPage heading: Pattern Area” elif self.focus==”notes”: self.apps[“notes”] += (“n”+textual content) self.display = f”Memo AppsnCurrent Memo:n{self.apps[‘notes’]}” else: self.display += f”nYou entered “{textual content}” however there aren’t any editable fields. ” self.action_log.append({“kind”:”kind”,”textual content”:textual content})

class ComputerTool: def __init__(self,pc:VirtualComputer): self.pc =pc def run(self, command:str, argument:str=””): if command==”click on”: self.pc.click on(argument) return {“standing”:”accomplished”,”end result”:f”clicked {argument}”} if command==”kind”: self.pc.kind(argument) return {“standing”:”accomplished”,”end result”:f”typed {argument}”} if command==”screenshot”: snap = self.pc.screenshot() return {“standing”:”accomplished”,”end result”:snap} return {“standing”:”error”,”end result”:f”unknown command {command}”}

class ComputerAgent: def __init__(self, llm:LocalLLM, instruments:ComputerTool, max_trajectory_budget:float=5.0): self.llm = llm self.software = software self.max_trajectory_budget = max_trajectory_budget async def run(self,messages): user_goal =messages[-1][“content”]

Variety of steps remaining = int(self.max_trajectory_budget) Output occasion = []
total_prompt_tokens = 0 total_completion_tokens = 0 whilesteps_remaining>0: display = self.software.pc.screenshot() immediate = ( “You’re a pc utilization agent.n” f”Person objective: {user_goal}n” f”Present display:n{display}nn” “Suppose step-by-step.n” “Reply: ACTION ARG THEN .n” ) thought = self.llm.generate(immediate) total_prompt_tokens += len(immediate.cut up()) total_completion_tokens += len(thought.cut up()) motion=”screenshot”; arg=””; Assistant_msg=”Working…” on the road thought.splitlines(): if line.strip().startswith(“ACTION “): after = line.cut up(“ACTION “,1)[1]
motion = after.cut up()[0].strip() If there may be “ARG” within the line: half = line.cut up(“ARG “,1)[1]
If half is ” THEN “: arg = half.cut up(” THEN “)[0].strip() else: arg = half.strip() if “THEN ” in line:assistant_msg = line.cut up(“THEN “,1)[1].strip() Output_events.append({“Abstract”:[{“text”:assistant_msg,”type”:”summary_text”}],”kind”:”inference”}) call_id = “call_”+uuid.uuid4().hex[:16]
tool_res = self.software.run(motion, arg) Output_events.append({“motion”:{“kind”:motion,”textual content”:arg},”call_id”:call_id,”standing”:tool_res[“status”],”kind”:”computer_call”}) snap = self.software.pc.screenshot() Output_events.append({“kind”:”computer_call_output”,”call_id”:call_id,”output”:{“kind”:”input_image”,”image_url”:snap}}) Output_events.append({“kind”:”message”,”position”:”assistant”,”content material”:[{“type”:”output_text”,”text”:assistant_msg}]}) If “completed” in assistant_msg.decrease() or “right here is” in assistant_msg.decrease(): Breaksteps_remaining -= 1 use = {“prompt_tokens”: total_prompt_tokens,”completion_tokens”: total_completion_tokens,”total_tokens”: total_prompt_tokens + total_completion_tokens,”response_cost”: 0.0} yield {“output”: output occasion, “utilization”: utilization}

async def main_demo(): pc = VirtualComputer() software = ComputerTool(pc) llm = LocalLLM() agent = ComputerAgent(llm, software, max_trajectory_budget=4) message =[{“role”:”user”,”content”:”Open mail, read inbox subjects, and summarize.”}]
async for the results of Agent.run(messages): print(“==== STREAM RESULT ====”) for the ensuing occasion[“output”]: For occasions[“type”]==”Pc name”: a =occasion.get(“motion”,{}) print(f”[TOOL CALL] {a.get(‘kind’)} -> {a.get(‘textual content’)} [{event.get(‘status’)}]”) for occasions[“type”]==”computer_call_output”: snap = occasion[“output”][“image_url”]

print(“Display screen after motion:n”, snap[:400],”…n”) for occasions[“type”]==”Message”: print(“Assistant:”, Occasion[“content”][0][“text”]”n”) print(“Utilization:”, end result[“usage”]) loop = asyncio.get_event_loop() loop.run_until_complete(main_demo())