Blablablab/audio-classification
0
1"""2Live Agent Runner3 4Manages an AI agent that browses the web via Playwright, controlled by an LLM.5Annotators can observe, pause, instruct, or take over the agent in real time.6 7The agent loop runs in a background thread with its own asyncio event loop.8Communication with Flask routes happens through thread-safe state and queues.9"""10 11import asyncio12import base6413import json14import logging15import os16import threading17import time18import uuid19from dataclasses import dataclass, field20from enum import Enum21from queue import Queue, Empty22from typing import Any, Callable, Dict, List, Optional23 24logger = logging.getLogger(__name__)25 26 27class AgentState(Enum):28 """States of the agent lifecycle."""29 IDLE = "idle"30 RUNNING = "running"31 PAUSED = "paused"32 TAKEOVER = "takeover"33 COMPLETED = "completed"34 ERROR = "error"35 36 37@dataclass38class AgentStep:39 """A single step in the agent's execution."""40 step_index: int41 screenshot_path: str42 action: Dict[str, Any]43 thought: str44 observation: str45 timestamp: float46 url: str = ""47 viewport: Optional[Dict[str, int]] = None48 coordinates: Optional[Dict[str, int]] = None49 element: Optional[Dict[str, Any]] = None50 annotator_instruction: Optional[str] = None51 52 def to_dict(self) -> Dict[str, Any]:53 d = {54 "step_index": self.step_index,55 "screenshot_url": self.screenshot_path,56 "action_type": self.action.get("type", "unknown"),57 "action": self.action,58 "thought": self.thought,59 "observation": self.observation,60 "timestamp": self.timestamp,61 "url": self.url,62 }63 if self.viewport:64 d["viewport"] = self.viewport65 if self.coordinates:66 d["coordinates"] = self.coordinates67 if self.element:68 d["element"] = self.element69 if self.annotator_instruction:70 d["annotator_instruction"] = self.annotator_instruction71 return d72 73 74@dataclass75class AgentConfig:76 """Configuration for the agent runner."""77 max_steps: int = 3078 step_delay: float = 1.079 viewport_width: int = 128080 viewport_height: int = 72081 system_prompt: str = ""82 model: str = "claude-sonnet-4-20250514"83 api_key: str = ""84 max_tokens: int = 409685 temperature: float = 0.386 endpoint_type: str = "anthropic_vision"87 history_window: int = 5 # Number of recent steps to include in LLM context88 timeout: int = 60 # Per-request timeout in seconds89 90 base_url: str = "" # For Ollama: server URL91 92 @classmethod93 def from_config(cls, config: Dict[str, Any]) -> "AgentConfig":94 """Create AgentConfig from a live_agent YAML config dict."""95 ai_config = config.get("ai_config", {})96 viewport = config.get("viewport", {})97 endpoint_type = config.get("endpoint_type", "anthropic_vision")98 99 # API key: Ollama doesn't need one; OpenAI-compatible servers100 # (e.g. vLLM) ignore it but the SDK requires a non-empty string.101 if endpoint_type == "ollama_vision":102 api_key = ai_config.get("api_key", "")103 default_model = "gemma3:4b"104 elif endpoint_type == "openai_vision":105 api_key = ai_config.get("api_key", os.environ.get("OPENAI_API_KEY", "EMPTY"))106 default_model = "" # must be set explicitly (e.g. served model id)107 else:108 api_key = ai_config.get("api_key", os.environ.get("ANTHROPIC_API_KEY", ""))109 default_model = "claude-sonnet-4-20250514"110 111 return cls(112 max_steps=config.get("max_steps", 30),113 step_delay=config.get("step_delay", 1.0),114 viewport_width=viewport.get("width", 1280),115 viewport_height=viewport.get("height", 720),116 system_prompt=config.get("system_prompt", DEFAULT_SYSTEM_PROMPT),117 model=ai_config.get("model", default_model),118 api_key=api_key,119 max_tokens=ai_config.get("max_tokens", 4096),120 temperature=ai_config.get("temperature", 0.3),121 endpoint_type=endpoint_type,122 history_window=config.get("history_window", 5),123 timeout=ai_config.get("timeout", 60),124 base_url=ai_config.get("base_url", "http://localhost:11434"),125 )126 127 128DEFAULT_SYSTEM_PROMPT = """You are a web browsing agent. You can see screenshots of web pages and take actions to complete tasks.129 130For each step, analyze the current screenshot and respond with a JSON object:131{132 "thought": "Your reasoning about what you see and what to do next",133 "action": {134 "type": "click|type|scroll|navigate|wait|done",135 // For click: "x": 100, "y": 200136 // For type: "text": "hello world"137 // For scroll: "direction": "up|down", "amount": 300138 // For navigate: "url": "https://..."139 // For wait: (no extra fields)140 // For done: "summary": "Task completed because..."141 }142}143 144Always respond with valid JSON only. No markdown, no extra text."""145 146 147class AgentRunner:148 """149 Runs an AI agent that browses the web via Playwright.150 151 The agent loop:152 1. Takes a screenshot153 2. Sends it to the LLM with context/history154 3. Parses the LLM response for an action155 4. Executes the action via Playwright156 5. Emits events to all listeners (for SSE)157 6. Repeats until done, error, or max_steps158 159 Thread-safe control methods allow pause/resume/instruct/takeover.160 """161 162 def __init__(self, session_id: str, config: AgentConfig, screenshot_dir: str):163 self.session_id = session_id164 self.config = config165 self.screenshot_dir = screenshot_dir166 167 # State168 self._state = AgentState.IDLE169 self._state_lock = threading.Lock()170 self._steps: List[AgentStep] = []171 self._error: Optional[str] = None172 173 # Control174 self._pause_event = threading.Event()175 self._pause_event.set() # Not paused initially176 self._stop_flag = threading.Event()177 self._instruction_queue: Queue = Queue()178 self._takeover_actions: Queue = Queue()179 180 # Listeners for SSE181 self._listeners: List[Callable] = []182 self._listeners_lock = threading.Lock()183 184 # Annotator interactions log185 self._interactions: List[Dict[str, Any]] = []186 187 # Playwright session (set during run)188 self._playwright_session = None189 self._llm_client = None190 191 # Background thread192 self._thread: Optional[threading.Thread] = None193 194 @property195 def state(self) -> AgentState:196 with self._state_lock:197 return self._state198 199 @state.setter200 def state(self, new_state: AgentState):201 with self._state_lock:202 old_state = self._state203 self._state = new_state204 self._emit_event("state_change", {205 "old_state": old_state.value,206 "new_state": new_state.value,207 "timestamp": time.time(),208 })209 210 @property211 def steps(self) -> List[AgentStep]:212 return list(self._steps)213 214 @property215 def step_count(self) -> int:216 return len(self._steps)217 218 @property219 def error(self) -> Optional[str]:220 return self._error221 222 # --- Control methods (thread-safe) ---223 224 def pause(self):225 """Pause the agent loop after the current step completes."""226 if self.state == AgentState.RUNNING:227 self._pause_event.clear()228 self.state = AgentState.PAUSED229 logger.info(f"[{self.session_id}] Agent paused")230 231 def resume(self):232 """Resume a paused agent."""233 if self.state == AgentState.PAUSED:234 self.state = AgentState.RUNNING235 self._pause_event.set()236 logger.info(f"[{self.session_id}] Agent resumed")237 238 def inject_instruction(self, instruction: str):239 """Send an instruction to the agent (processed at next step)."""240 self._instruction_queue.put(instruction)241 self._interactions.append({242 "type": "instruction",243 "text": instruction,244 "timestamp": time.time(),245 "step_index": self.step_count,246 })247 self._emit_event("instruction_received", {"instruction": instruction})248 logger.info(f"[{self.session_id}] Instruction injected: {instruction[:100]}")249 250 def enter_takeover(self):251 """Switch to manual takeover mode."""252 if self.state in (AgentState.RUNNING, AgentState.PAUSED):253 self._pause_event.clear() # Pause the agent loop254 self.state = AgentState.TAKEOVER255 self._interactions.append({256 "type": "takeover_start",257 "timestamp": time.time(),258 "step_index": self.step_count,259 })260 logger.info(f"[{self.session_id}] Takeover mode entered")261 262 def exit_takeover(self):263 """Exit manual takeover and resume the agent."""264 if self.state == AgentState.TAKEOVER:265 self._interactions.append({266 "type": "takeover_end",267 "timestamp": time.time(),268 "step_index": self.step_count,269 })270 self.state = AgentState.RUNNING271 self._pause_event.set()272 logger.info(f"[{self.session_id}] Takeover mode exited")273 274 def submit_manual_action(self, action: Dict[str, Any]):275 """Submit a manual action during takeover mode."""276 if self.state == AgentState.TAKEOVER:277 self._takeover_actions.put(action)278 279 def stop(self):280 """Stop the agent loop."""281 self._stop_flag.set()282 self._pause_event.set() # Unblock if paused283 logger.info(f"[{self.session_id}] Stop requested")284 285 # --- Listener management ---286 287 def add_listener(self, callback: Callable):288 """Add an SSE listener callback."""289 with self._listeners_lock:290 self._listeners.append(callback)291 292 def remove_listener(self, callback: Callable):293 """Remove an SSE listener callback."""294 with self._listeners_lock:295 self._listeners = [l for l in self._listeners if l is not callback]296 297 def _emit_event(self, event_type: str, data: Dict[str, Any]):298 """Emit an event to all listeners."""299 event = {"type": event_type, "data": data, "session_id": self.session_id}300 with self._listeners_lock:301 for listener in self._listeners:302 try:303 listener(event)304 except Exception as e:305 logger.warning(f"Listener error: {e}")306 307 # --- Main agent loop ---308 309 def start(self, task_description: str, start_url: str):310 """Start the agent in a background thread."""311 if self.state != AgentState.IDLE:312 raise RuntimeError(f"Cannot start agent in state {self.state}")313 314 self._thread = threading.Thread(315 target=self._run_thread,316 args=(task_description, start_url),317 daemon=True,318 name=f"agent-{self.session_id}",319 )320 self._thread.start()321 322 def _run_thread(self, task_description: str, start_url: str):323 """Thread target: runs the async agent loop."""324 loop = asyncio.new_event_loop()325 asyncio.set_event_loop(loop)326 try:327 loop.run_until_complete(self._run_async(task_description, start_url))328 except Exception as e:329 logger.error(f"[{self.session_id}] Agent thread error: {e}")330 self._error = str(e)331 self.state = AgentState.ERROR332 self._emit_event("error", {"message": str(e)})333 finally:334 loop.close()335 336 async def _run_async(self, task_description: str, start_url: str):337 """Async agent loop."""338 from potato.web_playwright import PlaywrightSession339 340 self.state = AgentState.RUNNING341 342 # Initialize Playwright343 self._playwright_session = PlaywrightSession(344 width=self.config.viewport_width,345 height=self.config.viewport_height,346 )347 started = await self._playwright_session.start(start_url)348 if not started:349 raise RuntimeError("Failed to start Playwright browser session")350 351 # Initialize LLM client352 self._init_llm_client()353 354 self._emit_event("started", {355 "task": task_description,356 "start_url": start_url,357 "max_steps": self.config.max_steps,358 })359 360 try:361 for step_index in range(self.config.max_steps):362 # Check stop flag363 if self._stop_flag.is_set():364 logger.info(f"[{self.session_id}] Stopped by user")365 break366 367 # Wait if paused (blocks until resume/stop)368 while not self._pause_event.is_set():369 if self._stop_flag.is_set():370 break371 # Handle takeover actions while paused in takeover mode372 if self.state == AgentState.TAKEOVER:373 await self._process_takeover_actions()374 await asyncio.sleep(0.1)375 376 if self._stop_flag.is_set():377 break378 379 # Check for injected instructions380 instruction = None381 try:382 instruction = self._instruction_queue.get_nowait()383 except Empty:384 pass385 386 # Execute one agent step387 step = await self._agent_step(388 step_index, task_description, instruction389 )390 self._steps.append(step)391 392 # Check if agent decided it's done393 if step.action.get("type") == "done":394 logger.info(f"[{self.session_id}] Agent completed task")395 break396 397 # Step delay398 if self.config.step_delay > 0:399 await asyncio.sleep(self.config.step_delay)400 401 self.state = AgentState.COMPLETED402 self._emit_event("complete", {403 "total_steps": len(self._steps),404 "final_url": (await self._playwright_session.get_state()).get("url", ""),405 })406 407 finally:408 await self._playwright_session.stop()409 self._playwright_session = None410 411 async def _agent_step(412 self,413 step_index: int,414 task_description: str,415 instruction: Optional[str] = None,416 ) -> AgentStep:417 """Execute a single agent step: screenshot → LLM → action → emit."""418 419 # 1. Take screenshot420 screenshot_bytes = await self._playwright_session.screenshot()421 if not screenshot_bytes:422 raise RuntimeError("Failed to capture screenshot")423 424 screenshot_path = os.path.join(425 self.screenshot_dir, f"step_{step_index:03d}.png"426 )427 os.makedirs(os.path.dirname(screenshot_path), exist_ok=True)428 with open(screenshot_path, "wb") as f:429 f.write(screenshot_bytes)430 431 # 2. Get page state432 page_state = await self._playwright_session.get_state()433 434 # 3. Emit thinking event435 self._emit_event("thinking", {436 "step_index": step_index,437 "screenshot_url": screenshot_path,438 "url": page_state.get("url", ""),439 })440 441 # 4. Build messages and query LLM442 screenshot_b64 = base64.b64encode(screenshot_bytes).decode("utf-8")443 messages = self._build_llm_messages(444 screenshot_b64, task_description, instruction445 )446 llm_response = self._query_llm(messages)447 448 # 5. Parse action from response449 thought, action = self._parse_action(llm_response)450 451 # 6. Execute action452 observation = await self._execute_action(action)453 454 # 7. Build step455 step = AgentStep(456 step_index=step_index,457 screenshot_path=screenshot_path,458 action=action,459 thought=thought,460 observation=observation,461 timestamp=time.time(),462 url=page_state.get("url", ""),463 viewport=page_state.get("viewport"),464 coordinates=_extract_coordinates(action),465 annotator_instruction=instruction,466 )467 468 # 8. Emit step event469 self._emit_event("step", step.to_dict())470 471 return step472 473 def _build_llm_messages(474 self,475 screenshot_b64: str,476 task_description: str,477 instruction: Optional[str] = None,478 ) -> List[Dict[str, Any]]:479 """Build message list for the LLM vision API."""480 messages = []481 482 # System message483 system_prompt = self.config.system_prompt or DEFAULT_SYSTEM_PROMPT484 messages.append({"role": "system", "content": system_prompt})485 486 # Task description487 task_msg = f"Task: {task_description}"488 if instruction:489 task_msg += f"\n\nAnnotator instruction: {instruction}"490 491 # Include recent step history492 history_steps = self._steps[-self.config.history_window:]493 if history_steps:494 history_parts = []495 for s in history_steps:496 entry = f"Step {s.step_index}: thought='{s.thought}', action={json.dumps(s.action)}, observation='{s.observation}'"497 history_parts.append(entry)498 task_msg += "\n\nRecent history:\n" + "\n".join(history_parts)499 500 messages.append({"role": "user", "content": task_msg})501 502 # Current screenshot (as a separate user message with image)503 messages.append({504 "role": "user",505 "content": [506 {507 "type": "image",508 "source": {509 "type": "base64",510 "media_type": "image/png",511 "data": screenshot_b64,512 },513 },514 {515 "type": "text",516 "text": f"Current page screenshot (step {len(self._steps)}). What action should I take next?",517 },518 ],519 })520 521 return messages522 523 def _init_llm_client(self):524 """Initialize the LLM client based on endpoint_type."""525 if self.config.endpoint_type == "anthropic_vision":526 try:527 import anthropic528 except ImportError:529 raise RuntimeError(530 "anthropic package required. Install with: pip install anthropic"531 )532 api_key = self.config.api_key or os.environ.get("ANTHROPIC_API_KEY")533 if not api_key:534 raise RuntimeError(535 "Anthropic API key required. Set in config or ANTHROPIC_API_KEY env var."536 )537 self._llm_client = anthropic.Anthropic(538 api_key=api_key, timeout=self.config.timeout539 )540 elif self.config.endpoint_type == "ollama_vision":541 try:542 import ollama543 except ImportError:544 raise RuntimeError(545 "ollama package required. Install with: pip install ollama"546 )547 host = self.config.base_url or "http://localhost:11434"548 self._llm_client = ollama.Client(549 host=host, timeout=self.config.timeout550 )551 # Verify connectivity552 try:553 self._llm_client.list()554 logger.info(f"Connected to Ollama at {host}, model: {self.config.model}")555 except Exception as e:556 raise RuntimeError(f"Failed to connect to Ollama at {host}: {e}")557 elif self.config.endpoint_type == "openai_vision":558 try:559 from openai import OpenAI560 except ImportError:561 raise RuntimeError(562 "openai package required. Install with: pip install openai"563 )564 base_url = self.config.base_url or "https://api.openai.com/v1"565 self._llm_client = OpenAI(566 base_url=base_url,567 api_key=self.config.api_key or "EMPTY",568 timeout=self.config.timeout,569 )570 try:571 self._llm_client.models.list()572 logger.info(573 f"Connected to OpenAI-compatible endpoint at {base_url}, "574 f"model: {self.config.model}"575 )576 except Exception as e:577 # Non-fatal: some servers gate /models; the chat call will578 # surface a real error if the endpoint is truly unreachable.579 logger.warning(580 f"Could not list models at {base_url} ({e}); continuing."581 )582 else:583 raise RuntimeError(584 f"Unsupported endpoint_type: {self.config.endpoint_type}. "585 f"Supported: 'anthropic_vision', 'ollama_vision', 'openai_vision'."586 )587 588 def _query_llm(self, messages: List[Dict[str, Any]]) -> str:589 """Send messages to the LLM and return the text response."""590 if self.config.endpoint_type == "anthropic_vision":591 return self._query_anthropic(messages)592 elif self.config.endpoint_type == "ollama_vision":593 return self._query_ollama(messages)594 elif self.config.endpoint_type == "openai_vision":595 return self._query_openai(messages)596 raise RuntimeError(f"Unsupported endpoint type: {self.config.endpoint_type}")597 598 def _query_openai(self, messages: List[Dict[str, Any]]) -> str:599 """Query an OpenAI-compatible vision endpoint (OpenAI, vLLM, etc.).600 601 Converts the internal Anthropic-style message blocks into OpenAI602 chat-completions format (image blocks become ``image_url`` data603 URIs). Requests a JSON object response when the server supports it,604 falling back gracefully if it does not.605 """606 oai_messages = []607 for msg in messages:608 role = msg["role"]609 content = msg.get("content", "")610 if isinstance(content, str):611 oai_messages.append({"role": role, "content": content})612 continue613 parts = []614 for block in content:615 if not isinstance(block, dict):616 continue617 if block.get("type") == "text":618 parts.append({"type": "text", "text": block.get("text", "")})619 elif block.get("type") == "image":620 src = block.get("source", {})621 if src.get("type") == "base64":622 media = src.get("media_type", "image/png")623 parts.append({624 "type": "image_url",625 "image_url": {626 "url": f"data:{media};base64,{src['data']}"627 },628 })629 oai_messages.append({"role": role, "content": parts})630 631 kwargs = {632 "model": self.config.model,633 "messages": oai_messages,634 "max_tokens": self.config.max_tokens,635 "temperature": self.config.temperature,636 }637 638 def _is_rate_limit(exc) -> bool:639 if getattr(exc, "status_code", None) == 429:640 return True641 s = str(exc).lower()642 return ("429" in s or "rate limit" in s or "quota" in s643 or "resource_exhausted" in s)644 645 def _create(use_rf: bool):646 if use_rf:647 return self._llm_client.chat.completions.create(648 response_format={"type": "json_object"}, **kwargs)649 return self._llm_client.chat.completions.create(**kwargs)650 651 # Transient 429s (per-minute rate/token bursts) are common mid-run652 # even on paid tiers; back off and retry instead of failing the653 # whole agent session.654 backoffs = [5, 15, 30, 30, 30]655 use_rf = True656 attempt = 0657 while True:658 try:659 resp = _create(use_rf)660 break661 except Exception as e:662 if _is_rate_limit(e):663 if attempt >= len(backoffs):664 raise665 wait = backoffs[attempt]666 attempt += 1667 logger.warning(668 f"[{self.session_id}] LLM 429/rate-limited; "669 f"retry {attempt}/{len(backoffs)} in {wait}s"670 )671 self._emit_event("thinking", {672 "text": f"Rate-limited by the model API; "673 f"waiting {wait}s before retrying…"674 })675 time.sleep(wait)676 continue677 if use_rf:678 # Server may not support response_format; drop it once.679 use_rf = False680 continue681 raise682 return resp.choices[0].message.content or ""683 684 def _query_anthropic(self, messages: List[Dict[str, Any]]) -> str:685 """Query Anthropic Claude with vision support."""686 # Separate system message687 system = ""688 api_messages = []689 for msg in messages:690 if msg["role"] == "system":691 system = msg["content"]692 else:693 api_messages.append(msg)694 695 kwargs = {696 "model": self.config.model,697 "max_tokens": self.config.max_tokens,698 "temperature": self.config.temperature,699 "messages": api_messages,700 }701 if system:702 kwargs["system"] = system703 704 response = self._llm_client.messages.create(**kwargs)705 return response.content[0].text706 707 def _query_ollama(self, messages: List[Dict[str, Any]]) -> str:708 """Query Ollama vision model.709 710 Converts Anthropic-format messages to Ollama format:711 - System messages are prepended to the prompt text712 - Multiple user messages are merged into a single message713 - Content blocks with images use Ollama's 'images' key714 """715 # Extract text and images from Anthropic-format messages716 all_text_parts = []717 all_images = []718 for msg in messages:719 content = msg.get("content", "")720 if msg["role"] == "system":721 if isinstance(content, str) and content:722 all_text_parts.insert(0, content)723 continue724 if isinstance(content, list):725 for block in content:726 if isinstance(block, dict):727 if block.get("type") == "text":728 all_text_parts.append(block["text"])729 elif block.get("type") == "image":730 source = block.get("source", {})731 if source.get("type") == "base64":732 all_images.append(source["data"])733 elif isinstance(content, str) and content:734 all_text_parts.append(content)735 736 ollama_msg = {737 "role": "user",738 "content": "\n\n".join(all_text_parts),739 }740 if all_images:741 ollama_msg["images"] = all_images742 743 options = {744 "temperature": self.config.temperature,745 "num_predict": self.config.max_tokens,746 }747 748 # Use Ollama's format schema to force structured JSON output749 agent_schema = {750 "type": "object",751 "properties": {752 "thought": {"type": "string"},753 "action": {754 "type": "object",755 "properties": {756 "type": {"type": "string"},757 "x": {"type": "integer"},758 "y": {"type": "integer"},759 "text": {"type": "string"},760 "url": {"type": "string"},761 "direction": {"type": "string"},762 "amount": {"type": "integer"},763 "summary": {"type": "string"},764 },765 "required": ["type"],766 },767 },768 "required": ["thought", "action"],769 }770 771 response = self._llm_client.chat(772 model=self.config.model,773 messages=[ollama_msg],774 options=options,775 format=agent_schema,776 )777 778 # Extract content from response (handle both dict and Pydantic model)779 message = (780 response.get("message")781 if hasattr(response, "get")782 else getattr(response, "message", None)783 )784 if message is None:785 raise RuntimeError("No message in Ollama response")786 787 content = (788 message.get("content")789 if hasattr(message, "get")790 else getattr(message, "content", None)791 )792 793 # Some models (e.g. qwen3-vl) put responses in 'thinking' field794 # and leave content empty. Extract the agent JSON from thinking.795 if not content:796 thinking = (797 message.get("thinking")798 if hasattr(message, "get")799 else getattr(message, "thinking", None)800 )801 if thinking:802 content = _extract_agent_json(thinking)803 804 return content or ""805 806 def _parse_action(self, llm_response: str) -> tuple:807 """Parse thought and action from LLM JSON response.808 809 Returns:810 (thought, action_dict)811 """812 # Try to extract JSON from response813 text = llm_response.strip()814 815 # Handle markdown code blocks816 if "```json" in text:817 import re818 match = re.search(r"```json\s*([\s\S]*?)\s*```", text)819 if match:820 text = match.group(1).strip()821 elif "```" in text:822 import re823 match = re.search(r"```\s*([\s\S]*?)\s*```", text)824 if match:825 text = match.group(1).strip()826 827 try:828 parsed = json.loads(text)829 except json.JSONDecodeError:830 logger.warning(f"Failed to parse LLM response as JSON: {text[:200]}")831 return text, {"type": "wait"}832 833 thought = parsed.get("thought", "")834 action = parsed.get("action", {"type": "wait"})835 836 # Validate action has a type837 if "type" not in action:838 action["type"] = "wait"839 840 return thought, action841 842 async def _execute_action(self, action: Dict[str, Any]) -> str:843 """Execute an action via Playwright and return observation."""844 action_type = action.get("type", "wait")845 pw = self._playwright_session846 847 try:848 if action_type == "click":849 x = int(action.get("x", 0))850 y = int(action.get("y", 0))851 success = await pw.click(x, y)852 return f"Clicked at ({x}, {y})" if success else f"Click failed at ({x}, {y})"853 854 elif action_type == "type":855 text = action.get("text", "")856 # Handle control characters via keyboard.press857 if text == "\b":858 success = await pw.page.keyboard.press("Backspace") or True859 return "Pressed Backspace"860 elif text == "\n":861 success = await pw.page.keyboard.press("Enter") or True862 return "Pressed Enter"863 elif text == "\t":864 success = await pw.page.keyboard.press("Tab") or True865 return "Pressed Tab"866 else:867 success = await pw.type_text(text)868 return f"Typed '{text}'" if success else f"Type failed: '{text}'"869 870 elif action_type == "scroll":871 direction = action.get("direction", "down")872 amount = int(action.get("amount", 300))873 dy = amount if direction == "down" else -amount874 success = await pw.scroll(0, dy)875 return f"Scrolled {direction} by {amount}px" if success else "Scroll failed"876 877 elif action_type == "navigate":878 url = action.get("url", "")879 success = await pw.navigate(url)880 return f"Navigated to {url}" if success else f"Navigation failed: {url}"881 882 elif action_type == "wait":883 await asyncio.sleep(1)884 return "Waited 1 second"885 886 elif action_type == "done":887 summary = action.get("summary", "Task completed")888 return summary889 890 else:891 logger.warning(f"Unknown action type: {action_type}")892 return f"Unknown action: {action_type}"893 894 except Exception as e:895 logger.error(f"Action execution error: {e}")896 return f"Error executing {action_type}: {e}"897 898 async def _process_takeover_actions(self):899 """Process manual actions submitted during takeover mode."""900 try:901 action = self._takeover_actions.get_nowait()902 except Empty:903 return904 905 pw = self._playwright_session906 if not pw:907 return908 909 observation = await self._execute_action(action)910 911 # Take screenshot after manual action912 screenshot_bytes = await pw.screenshot()913 step_index = len(self._steps)914 screenshot_path = os.path.join(915 self.screenshot_dir, f"step_{step_index:03d}_manual.png"916 )917 if screenshot_bytes:918 with open(screenshot_path, "wb") as f:919 f.write(screenshot_bytes)920 921 page_state = await pw.get_state()922 923 step = AgentStep(924 step_index=step_index,925 screenshot_path=screenshot_path,926 action={**action, "_manual": True},927 thought="[Manual takeover action]",928 observation=observation,929 timestamp=time.time(),930 url=page_state.get("url", ""),931 viewport=page_state.get("viewport"),932 coordinates=_extract_coordinates(action),933 )934 self._steps.append(step)935 self._emit_event("step", step.to_dict())936 937 # --- Trace export ---938 939 def get_trace(self) -> Dict[str, Any]:940 """Export the session as a web_agent_trace-compatible dict."""941 return {942 "steps": [s.to_dict() for s in self._steps],943 "task_description": "", # Set by caller944 "session_id": self.session_id,945 "agent_config": {946 "model": self.config.model,947 "endpoint_type": self.config.endpoint_type,948 "max_steps": self.config.max_steps,949 },950 "annotator_interactions": self._interactions,951 "state": self.state.value,952 "total_steps": len(self._steps),953 }954 955 def get_state_summary(self) -> Dict[str, Any]:956 """Get a summary of current state for API responses."""957 return {958 "session_id": self.session_id,959 "state": self.state.value,960 "step_count": len(self._steps),961 "error": self._error,962 "has_instructions_pending": not self._instruction_queue.empty(),963 }964 965 966def _extract_agent_json(text: str) -> str:967 """Extract the last valid JSON object containing 'thought' or 'action' from text.968 969 Some models (qwen3-vl) put their chain-of-thought in the thinking field970 with the actual JSON answer embedded in the text. This function finds971 that JSON, skipping any example/template JSON from the prompt.972 """973 import re974 975 # Find all JSON-like blocks (balanced braces)976 candidates = []977 depth = 0978 start = None979 for i, ch in enumerate(text):980 if ch == "{":981 if depth == 0:982 start = i983 depth += 1984 elif ch == "}":985 depth -= 1986 if depth == 0 and start is not None:987 candidates.append(text[start : i + 1])988 start = None989 990 # Try each candidate (last first — most likely to be the final answer)991 for candidate in reversed(candidates):992 try:993 parsed = json.loads(candidate)994 if isinstance(parsed, dict) and ("thought" in parsed or "action" in parsed):995 return candidate996 except (json.JSONDecodeError, ValueError):997 continue998 999 # Fallback: try greedy regex for any JSON1000 match = re.search(r"\{[^{}]*\}", text)1001 return match.group(0) if match else ""1002 1003 1004def _extract_coordinates(action: Dict[str, Any]) -> Optional[Dict[str, int]]:1005 """Extract x, y coordinates from an action if present."""1006 if "x" in action and "y" in action:1007 return {"x": int(action["x"]), "y": int(action["y"])}1008 return None1009