diff --git a/sentience/__init__.py b/sentience/__init__.py index 61526a6..76458db 100644 --- a/sentience/__init__.py +++ b/sentience/__init__.py @@ -14,9 +14,6 @@ from .cloud_tracing import CloudTraceSink, SentienceLogger from .conversational_agent import ConversationalAgent from .expect import expect - -# Formatting (v0.12.0+) -from .formatting import format_snapshot_for_llm from .generator import ScriptGenerator, generate from .inspector import Inspector, inspect from .llm_provider import ( @@ -62,6 +59,7 @@ from .tracing import JsonlTraceSink, TraceEvent, Tracer, TraceSink # Utilities (v0.12.0+) +# Import from utils package (re-exports from submodules for backward compatibility) from .utils import ( canonical_snapshot_loose, canonical_snapshot_strict, @@ -69,6 +67,9 @@ save_storage_state, sha256_digest, ) + +# Formatting (v0.12.0+) +from .utils.formatting import format_snapshot_for_llm from .wait import wait_for __version__ = "0.91.1" diff --git a/sentience/action_executor.py b/sentience/action_executor.py new file mode 100644 index 0000000..104e255 --- /dev/null +++ b/sentience/action_executor.py @@ -0,0 +1,191 @@ +""" +Action Executor for Sentience Agent. + +Handles parsing and execution of action commands (CLICK, TYPE, PRESS, FINISH). +This separates action execution concerns from LLM interaction. +""" + +import re +from typing import Any + +from .actions import click, click_async, press, press_async, type_text, type_text_async +from .browser import AsyncSentienceBrowser, SentienceBrowser +from .models import Snapshot + + +class ActionExecutor: + """ + Executes actions and handles parsing of action command strings. + + This class encapsulates all action execution logic, making it easier to: + - Test action execution independently + - Add new action types in one place + - Handle action parsing errors consistently + """ + + def __init__(self, browser: SentienceBrowser | AsyncSentienceBrowser): + """ + Initialize action executor. + + Args: + browser: SentienceBrowser or AsyncSentienceBrowser instance + """ + self.browser = browser + self._is_async = isinstance(browser, AsyncSentienceBrowser) + + def execute(self, action_str: str, snap: Snapshot) -> dict[str, Any]: + """ + Parse action string and execute SDK call (synchronous). + + Args: + action_str: Action string from LLM (e.g., "CLICK(42)", "TYPE(15, \"text\")") + snap: Current snapshot (for context, currently unused but kept for API consistency) + + Returns: + Execution result dictionary with keys: + - success: bool + - action: str (e.g., "click", "type", "press", "finish") + - element_id: Optional[int] (for click/type actions) + - text: Optional[str] (for type actions) + - key: Optional[str] (for press actions) + - outcome: Optional[str] (action outcome) + - url_changed: Optional[bool] (for click actions) + - error: Optional[str] (if action failed) + - message: Optional[str] (for finish action) + + Raises: + ValueError: If action format is unknown + RuntimeError: If called on async browser (use execute_async instead) + """ + if self._is_async: + raise RuntimeError( + "ActionExecutor.execute() called on async browser. Use execute_async() instead." + ) + + # Parse CLICK(42) + if match := re.match(r"CLICK\s*\(\s*(\d+)\s*\)", action_str, re.IGNORECASE): + element_id = int(match.group(1)) + result = click(self.browser, element_id) # type: ignore + return { + "success": result.success, + "action": "click", + "element_id": element_id, + "outcome": result.outcome, + "url_changed": result.url_changed, + } + + # Parse TYPE(42, "hello world") + elif match := re.match( + r'TYPE\s*\(\s*(\d+)\s*,\s*["\']([^"\']*)["\']\s*\)', + action_str, + re.IGNORECASE, + ): + element_id = int(match.group(1)) + text = match.group(2) + result = type_text(self.browser, element_id, text) # type: ignore + return { + "success": result.success, + "action": "type", + "element_id": element_id, + "text": text, + "outcome": result.outcome, + } + + # Parse PRESS("Enter") + elif match := re.match(r'PRESS\s*\(\s*["\']([^"\']+)["\']\s*\)', action_str, re.IGNORECASE): + key = match.group(1) + result = press(self.browser, key) # type: ignore + return { + "success": result.success, + "action": "press", + "key": key, + "outcome": result.outcome, + } + + # Parse FINISH() + elif re.match(r"FINISH\s*\(\s*\)", action_str, re.IGNORECASE): + return { + "success": True, + "action": "finish", + "message": "Task marked as complete", + } + + else: + raise ValueError( + f"Unknown action format: {action_str}\n" + f'Expected: CLICK(id), TYPE(id, "text"), PRESS("key"), or FINISH()' + ) + + async def execute_async(self, action_str: str, snap: Snapshot) -> dict[str, Any]: + """ + Parse action string and execute SDK call (asynchronous). + + Args: + action_str: Action string from LLM (e.g., "CLICK(42)", "TYPE(15, \"text\")") + snap: Current snapshot (for context, currently unused but kept for API consistency) + + Returns: + Execution result dictionary (same format as execute()) + + Raises: + ValueError: If action format is unknown + RuntimeError: If called on sync browser (use execute() instead) + """ + if not self._is_async: + raise RuntimeError( + "ActionExecutor.execute_async() called on sync browser. Use execute() instead." + ) + + # Parse CLICK(42) + if match := re.match(r"CLICK\s*\(\s*(\d+)\s*\)", action_str, re.IGNORECASE): + element_id = int(match.group(1)) + result = await click_async(self.browser, element_id) # type: ignore + return { + "success": result.success, + "action": "click", + "element_id": element_id, + "outcome": result.outcome, + "url_changed": result.url_changed, + } + + # Parse TYPE(42, "hello world") + elif match := re.match( + r'TYPE\s*\(\s*(\d+)\s*,\s*["\']([^"\']*)["\']\s*\)', + action_str, + re.IGNORECASE, + ): + element_id = int(match.group(1)) + text = match.group(2) + result = await type_text_async(self.browser, element_id, text) # type: ignore + return { + "success": result.success, + "action": "type", + "element_id": element_id, + "text": text, + "outcome": result.outcome, + } + + # Parse PRESS("Enter") + elif match := re.match(r'PRESS\s*\(\s*["\']([^"\']+)["\']\s*\)', action_str, re.IGNORECASE): + key = match.group(1) + result = await press_async(self.browser, key) # type: ignore + return { + "success": result.success, + "action": "press", + "key": key, + "outcome": result.outcome, + } + + # Parse FINISH() + elif re.match(r"FINISH\s*\(\s*\)", action_str, re.IGNORECASE): + return { + "success": True, + "action": "finish", + "message": "Task marked as complete", + } + + else: + raise ValueError( + f"Unknown action format: {action_str}\n" + f'Expected: CLICK(id), TYPE(id, "text"), PRESS("key"), or FINISH()' + ) diff --git a/sentience/agent.py b/sentience/agent.py index fec23d5..3238f30 100644 --- a/sentience/agent.py +++ b/sentience/agent.py @@ -5,14 +5,15 @@ import asyncio import hashlib -import re import time from typing import TYPE_CHECKING, Any, Optional -from .actions import click, click_async, press, press_async, type_text, type_text_async +from .action_executor import ActionExecutor from .agent_config import AgentConfig from .base_agent import BaseAgent, BaseAgentAsync from .browser import AsyncSentienceBrowser, SentienceBrowser +from .element_filter import ElementFilter +from .llm_interaction_handler import LLMInteractionHandler from .llm_provider import LLMProvider, LLMResponse from .models import ( ActionHistory, @@ -25,6 +26,7 @@ TokenStats, ) from .snapshot import snapshot, snapshot_async +from .trace_event_builder import TraceEventBuilder if TYPE_CHECKING: from .tracing import Tracer @@ -81,6 +83,10 @@ def __init__( self.tracer = tracer self.config = config or AgentConfig() + # Initialize handlers + self.llm_handler = LLMInteractionHandler(llm) + self.action_executor = ActionExecutor(browser) + # Screenshot sequence counter # Execution history self.history: list[dict[str, Any]] = [] @@ -100,9 +106,7 @@ def _compute_hash(self, text: str) -> str: """Compute SHA256 hash of text.""" return hashlib.sha256(text.encode("utf-8")).hexdigest() - def _get_element_bbox( - self, element_id: int | None, snap: Snapshot - ) -> dict[str, float] | None: + def _get_element_bbox(self, element_id: int | None, snap: Snapshot) -> dict[str, float] | None: """Get bounding box for an element from snapshot.""" if element_id is None: return None @@ -200,17 +204,8 @@ def act( # noqa: C901 # Emit snapshot trace event if tracer is enabled if self.tracer: - # Include ALL elements with full data for DOM tree display - # Use snap.elements (all elements) not filtered_elements - elements_data = [el.model_dump() for el in snap.elements] - # Build snapshot event data - snapshot_data = { - "url": snap.url, - "element_count": len(snap.elements), - "timestamp": snap.timestamp, - "elements": elements_data, # Full element data for DOM tree - } + snapshot_data = TraceEventBuilder.build_snapshot_event(snap) # Always include screenshot in trace event for studio viewer compatibility # CloudTraceSink will extract and upload screenshots separately, then remove @@ -250,10 +245,10 @@ def act( # noqa: C901 ) # 2. GROUND: Format elements for LLM context - context = self._build_context(filtered_snap, goal) + context = self.llm_handler.build_context(filtered_snap, goal) # 3. THINK: Query LLM for next action - llm_response = self._query_llm(context, goal) + llm_response = self.llm_handler.query_llm(context, goal) # Emit LLM query trace event if tracer is enabled if self.tracer: @@ -275,10 +270,10 @@ def act( # noqa: C901 self._track_tokens(goal, llm_response) # Parse action from LLM response - action_str = self._extract_action_from_response(llm_response.content) + action_str = self.llm_handler.extract_action(llm_response.content) # 4. EXECUTE: Parse and run action - result_dict = self._execute_action(action_str, filtered_snap) + result_dict = self.action_executor.execute(action_str, filtered_snap) duration_ms = int((time.time() - start_time) * 1000) @@ -425,23 +420,18 @@ def act( # noqa: C901 } # Build complete step_end event - step_end_data = { - "v": 1, - "step_id": step_id, - "step_index": self._step_count, - "goal": goal, - "attempt": attempt, - "pre": { - "url": pre_url, - "snapshot_digest": snapshot_digest, - }, - "llm": llm_data, - "exec": exec_data, - "post": { - "url": post_url, - }, - "verify": verify_data, - } + step_end_data = TraceEventBuilder.build_step_end_event( + step_id=step_id, + step_index=self._step_count, + goal=goal, + attempt=attempt, + pre_url=pre_url, + post_url=post_url, + snapshot_digest=snapshot_digest, + llm_data=llm_data, + exec_data=exec_data, + verify_data=verify_data, + ) self.tracer.emit("step_end", step_end_data, step_id=step_id) @@ -479,187 +469,6 @@ def act( # noqa: C901 ) raise RuntimeError(f"Failed after {max_retries} retries: {e}") - def _build_context(self, snap: Snapshot, goal: str) -> str: - """ - Convert snapshot elements to token-efficient prompt string - - Format: [ID] "text" {cues} @ (x,y) (Imp:score) - - Args: - snap: Snapshot object - goal: User goal (for context) - - Returns: - Formatted element context string - """ - lines = [] - # Note: elements are already filtered by filter_elements() in act() - for el in snap.elements: - # Extract visual cues - cues = [] - if el.visual_cues.is_primary: - cues.append("PRIMARY") - if el.visual_cues.is_clickable: - cues.append("CLICKABLE") - if el.visual_cues.background_color_name: - cues.append(f"color:{el.visual_cues.background_color_name}") - - # Format element line - cues_str = f" {{{','.join(cues)}}}" if cues else "" - text_preview = ( - (el.text[:50] + "...") if el.text and len(el.text) > 50 else (el.text or "") - ) - - lines.append( - f'[{el.id}] <{el.role}> "{text_preview}"{cues_str} ' - f"@ ({int(el.bbox.x)},{int(el.bbox.y)}) (Imp:{el.importance})" - ) - - return "\n".join(lines) - - def _extract_action_from_response(self, response: str) -> str: - """ - Extract action command from LLM response, handling cases where - the LLM adds extra explanation despite instructions. - - Args: - response: Raw LLM response text - - Returns: - Cleaned action command string - """ - import re - - # Remove markdown code blocks if present - response = re.sub(r"```[\w]*\n?", "", response) - response = response.strip() - - # Try to find action patterns in the response - # Pattern matches: CLICK(123), TYPE(123, "text"), PRESS("key"), FINISH() - action_pattern = r'(CLICK\s*\(\s*\d+\s*\)|TYPE\s*\(\s*\d+\s*,\s*["\'].*?["\']\s*\)|PRESS\s*\(\s*["\'].*?["\']\s*\)|FINISH\s*\(\s*\))' - - match = re.search(action_pattern, response, re.IGNORECASE) - if match: - return match.group(1) - - # If no pattern match, return the original response (will likely fail parsing) - return response - - def _query_llm(self, dom_context: str, goal: str) -> LLMResponse: - """ - Query LLM with standardized prompt template - - Args: - dom_context: Formatted element context - goal: User goal - - Returns: - LLMResponse from LLM provider - """ - system_prompt = f"""You are an AI web automation agent. - -GOAL: {goal} - -VISIBLE ELEMENTS (sorted by importance): -{dom_context} - -VISUAL CUES EXPLAINED: -- {{PRIMARY}}: Main call-to-action element on the page -- {{CLICKABLE}}: Element is clickable -- {{color:X}}: Background color name - -CRITICAL RESPONSE FORMAT: -You MUST respond with ONLY ONE of these exact action formats: -- CLICK(id) - Click element by ID -- TYPE(id, "text") - Type text into element -- PRESS("key") - Press keyboard key (Enter, Escape, Tab, ArrowDown, etc) -- FINISH() - Task complete - -DO NOT include any explanation, reasoning, or natural language. -DO NOT use markdown formatting or code blocks. -DO NOT say "The next step is..." or anything similar. - -CORRECT Examples: -CLICK(42) -TYPE(15, "magic mouse") -PRESS("Enter") -FINISH() - -INCORRECT Examples (DO NOT DO THIS): -"The next step is to click..." -"I will type..." -```CLICK(42)``` -""" - - user_prompt = "Return the single action command:" - - return self.llm.generate(system_prompt, user_prompt, temperature=0.0) - - def _execute_action(self, action_str: str, snap: Snapshot) -> dict[str, Any]: - """ - Parse action string and execute SDK call - - Args: - action_str: Action string from LLM (e.g., "CLICK(42)") - snap: Current snapshot (for context) - - Returns: - Execution result dictionary - """ - # Parse CLICK(42) - if match := re.match(r"CLICK\s*\(\s*(\d+)\s*\)", action_str, re.IGNORECASE): - element_id = int(match.group(1)) - result = click(self.browser, element_id) - return { - "success": result.success, - "action": "click", - "element_id": element_id, - "outcome": result.outcome, - "url_changed": result.url_changed, - } - - # Parse TYPE(42, "hello world") - elif match := re.match( - r'TYPE\s*\(\s*(\d+)\s*,\s*["\']([^"\']*)["\']\s*\)', - action_str, - re.IGNORECASE, - ): - element_id = int(match.group(1)) - text = match.group(2) - result = type_text(self.browser, element_id, text) - return { - "success": result.success, - "action": "type", - "element_id": element_id, - "text": text, - "outcome": result.outcome, - } - - # Parse PRESS("Enter") - elif match := re.match(r'PRESS\s*\(\s*["\']([^"\']+)["\']\s*\)', action_str, re.IGNORECASE): - key = match.group(1) - result = press(self.browser, key) - return { - "success": result.success, - "action": "press", - "key": key, - "outcome": result.outcome, - } - - # Parse FINISH() - elif re.match(r"FINISH\s*\(\s*\)", action_str, re.IGNORECASE): - return { - "success": True, - "action": "finish", - "message": "Task marked as complete", - } - - else: - raise ValueError( - f"Unknown action format: {action_str}\n" - f'Expected: CLICK(id), TYPE(id, "text"), PRESS("key"), or FINISH()' - ) - def _track_tokens(self, goal: str, llm_response: LLMResponse): """ Track token usage for analytics @@ -723,8 +532,8 @@ def filter_elements(self, snapshot: Snapshot, goal: str | None = None) -> list[E """ Filter elements from snapshot based on goal context. - This default implementation applies goal-based keyword matching to boost - relevant elements and filters out irrelevant ones. + This implementation uses ElementFilter to apply goal-based keyword matching + to boost relevant elements and filters out irrelevant ones. Args: snapshot: Current page snapshot @@ -733,76 +542,7 @@ def filter_elements(self, snapshot: Snapshot, goal: str | None = None) -> list[E Returns: Filtered list of elements """ - elements = snapshot.elements - - # If no goal provided, return all elements (up to limit) - if not goal: - return elements[: self.default_snapshot_limit] - - goal_lower = goal.lower() - - # Extract keywords from goal - keywords = self._extract_keywords(goal_lower) - - # Boost elements matching goal keywords - scored_elements = [] - for el in elements: - score = el.importance - - # Boost if element text matches goal - if el.text and any(kw in el.text.lower() for kw in keywords): - score += 0.3 - - # Boost if role matches goal intent - if "click" in goal_lower and el.visual_cues.is_clickable: - score += 0.2 - if "type" in goal_lower and el.role in ["textbox", "searchbox"]: - score += 0.2 - if "search" in goal_lower: - # Filter out non-interactive elements for search tasks - if el.role in ["link", "img"] and not el.visual_cues.is_primary: - score -= 0.5 - - scored_elements.append((score, el)) - - # Re-sort by boosted score - scored_elements.sort(key=lambda x: x[0], reverse=True) - elements = [el for _, el in scored_elements] - - return elements[: self.default_snapshot_limit] - - def _extract_keywords(self, text: str) -> list[str]: - """ - Extract meaningful keywords from goal text - - Args: - text: Text to extract keywords from - - Returns: - List of keywords - """ - stopwords = { - "the", - "a", - "an", - "and", - "or", - "but", - "in", - "on", - "at", - "to", - "for", - "of", - "with", - "by", - "from", - "as", - "is", - "was", - } - words = text.split() - return [w for w in words if w not in stopwords and len(w) > 2] + return ElementFilter.filter_by_goal(snapshot, goal, self.default_snapshot_limit) class SentienceAgentAsync(BaseAgentAsync): @@ -855,6 +595,10 @@ def __init__( self.tracer = tracer self.config = config or AgentConfig() + # Initialize handlers + self.llm_handler = LLMInteractionHandler(llm) + self.action_executor = ActionExecutor(browser) + # Screenshot sequence counter # Execution history self.history: list[dict[str, Any]] = [] @@ -874,9 +618,7 @@ def _compute_hash(self, text: str) -> str: """Compute SHA256 hash of text.""" return hashlib.sha256(text.encode("utf-8")).hexdigest() - def _get_element_bbox( - self, element_id: int | None, snap: Snapshot - ) -> dict[str, float] | None: + def _get_element_bbox(self, element_id: int | None, snap: Snapshot) -> dict[str, float] | None: """Get bounding box for an element from snapshot.""" if element_id is None: return None @@ -974,17 +716,8 @@ async def act( # noqa: C901 # Emit snapshot trace event if tracer is enabled if self.tracer: - # Include ALL elements with full data for DOM tree display - # Use snap.elements (all elements) not filtered_elements - elements_data = [el.model_dump() for el in snap.elements] - # Build snapshot event data - snapshot_data = { - "url": snap.url, - "element_count": len(snap.elements), - "timestamp": snap.timestamp, - "elements": elements_data, # Full element data for DOM tree - } + snapshot_data = TraceEventBuilder.build_snapshot_event(snap) # Always include screenshot in trace event for studio viewer compatibility # CloudTraceSink will extract and upload screenshots separately, then remove @@ -1024,10 +757,10 @@ async def act( # noqa: C901 ) # 2. GROUND: Format elements for LLM context - context = self._build_context(filtered_snap, goal) + context = self.llm_handler.build_context(filtered_snap, goal) # 3. THINK: Query LLM for next action - llm_response = self._query_llm(context, goal) + llm_response = self.llm_handler.query_llm(context, goal) # Emit LLM query trace event if tracer is enabled if self.tracer: @@ -1049,10 +782,10 @@ async def act( # noqa: C901 self._track_tokens(goal, llm_response) # Parse action from LLM response - action_str = self._extract_action_from_response(llm_response.content) + action_str = self.llm_handler.extract_action(llm_response.content) # 4. EXECUTE: Parse and run action - result_dict = await self._execute_action(action_str, filtered_snap) + result_dict = await self.action_executor.execute_async(action_str, filtered_snap) duration_ms = int((time.time() - start_time) * 1000) @@ -1199,23 +932,18 @@ async def act( # noqa: C901 } # Build complete step_end event - step_end_data = { - "v": 1, - "step_id": step_id, - "step_index": self._step_count, - "goal": goal, - "attempt": attempt, - "pre": { - "url": pre_url, - "snapshot_digest": snapshot_digest, - }, - "llm": llm_data, - "exec": exec_data, - "post": { - "url": post_url, - }, - "verify": verify_data, - } + step_end_data = TraceEventBuilder.build_step_end_event( + step_id=step_id, + step_index=self._step_count, + goal=goal, + attempt=attempt, + pre_url=pre_url, + post_url=post_url, + snapshot_digest=snapshot_digest, + llm_data=llm_data, + exec_data=exec_data, + verify_data=verify_data, + ) self.tracer.emit("step_end", step_end_data, step_id=step_id) @@ -1253,156 +981,6 @@ async def act( # noqa: C901 ) raise RuntimeError(f"Failed after {max_retries} retries: {e}") - def _build_context(self, snap: Snapshot, goal: str) -> str: - """Convert snapshot elements to token-efficient prompt string (same as sync version)""" - lines = [] - # Note: elements are already filtered by filter_elements() in act() - for el in snap.elements: - # Extract visual cues - cues = [] - if el.visual_cues.is_primary: - cues.append("PRIMARY") - if el.visual_cues.is_clickable: - cues.append("CLICKABLE") - if el.visual_cues.background_color_name: - cues.append(f"color:{el.visual_cues.background_color_name}") - - # Format element line - cues_str = f" {{{','.join(cues)}}}" if cues else "" - text_preview = ( - (el.text[:50] + "...") if el.text and len(el.text) > 50 else (el.text or "") - ) - - lines.append( - f'[{el.id}] <{el.role}> "{text_preview}"{cues_str} ' - f"@ ({int(el.bbox.x)},{int(el.bbox.y)}) (Imp:{el.importance})" - ) - - return "\n".join(lines) - - def _extract_action_from_response(self, response: str) -> str: - """Extract action command from LLM response (same as sync version)""" - # Remove markdown code blocks if present - response = re.sub(r"```[\w]*\n?", "", response) - response = response.strip() - - # Try to find action patterns in the response - # Pattern matches: CLICK(123), TYPE(123, "text"), PRESS("key"), FINISH() - action_pattern = r'(CLICK\s*\(\s*\d+\s*\)|TYPE\s*\(\s*\d+\s*,\s*["\'].*?["\']\s*\)|PRESS\s*\(\s*["\'].*?["\']\s*\)|FINISH\s*\(\s*\))' - - match = re.search(action_pattern, response, re.IGNORECASE) - if match: - return match.group(1) - - # If no pattern match, return the original response (will likely fail parsing) - return response - - def _query_llm(self, dom_context: str, goal: str) -> LLMResponse: - """Query LLM with standardized prompt template (same as sync version)""" - system_prompt = f"""You are an AI web automation agent. - -GOAL: {goal} - -VISIBLE ELEMENTS (sorted by importance): -{dom_context} - -VISUAL CUES EXPLAINED: -- {{PRIMARY}}: Main call-to-action element on the page -- {{CLICKABLE}}: Element is clickable -- {{color:X}}: Background color name - -CRITICAL RESPONSE FORMAT: -You MUST respond with ONLY ONE of these exact action formats: -- CLICK(id) - Click element by ID -- TYPE(id, "text") - Type text into element -- PRESS("key") - Press keyboard key (Enter, Escape, Tab, ArrowDown, etc) -- FINISH() - Task complete - -DO NOT include any explanation, reasoning, or natural language. -DO NOT use markdown formatting or code blocks. -DO NOT say "The next step is..." or anything similar. - -CORRECT Examples: -CLICK(42) -TYPE(15, "magic mouse") -PRESS("Enter") -FINISH() - -INCORRECT Examples (DO NOT DO THIS): -"The next step is to click..." -"I will type..." -```CLICK(42)``` -""" - - user_prompt = "Return the single action command:" - - return self.llm.generate(system_prompt, user_prompt, temperature=0.0) - - async def _execute_action(self, action_str: str, snap: Snapshot) -> dict[str, Any]: - """ - Parse action string and execute SDK call (async) - - Args: - action_str: Action string from LLM (e.g., "CLICK(42)") - snap: Current snapshot (for context) - - Returns: - Execution result dictionary - """ - # Parse CLICK(42) - if match := re.match(r"CLICK\s*\(\s*(\d+)\s*\)", action_str, re.IGNORECASE): - element_id = int(match.group(1)) - result = await click_async(self.browser, element_id) - return { - "success": result.success, - "action": "click", - "element_id": element_id, - "outcome": result.outcome, - "url_changed": result.url_changed, - } - - # Parse TYPE(42, "hello world") - elif match := re.match( - r'TYPE\s*\(\s*(\d+)\s*,\s*["\']([^"\']*)["\']\s*\)', - action_str, - re.IGNORECASE, - ): - element_id = int(match.group(1)) - text = match.group(2) - result = await type_text_async(self.browser, element_id, text) - return { - "success": result.success, - "action": "type", - "element_id": element_id, - "text": text, - "outcome": result.outcome, - } - - # Parse PRESS("Enter") - elif match := re.match(r'PRESS\s*\(\s*["\']([^"\']+)["\']\s*\)', action_str, re.IGNORECASE): - key = match.group(1) - result = await press_async(self.browser, key) - return { - "success": result.success, - "action": "press", - "key": key, - "outcome": result.outcome, - } - - # Parse FINISH() - elif re.match(r"FINISH\s*\(\s*\)", action_str, re.IGNORECASE): - return { - "success": True, - "action": "finish", - "message": "Task marked as complete", - } - - else: - raise ValueError( - f"Unknown action format: {action_str}\n" - f'Expected: CLICK(id), TYPE(id, "text"), PRESS("key"), or FINISH()' - ) - def _track_tokens(self, goal: str, llm_response: LLMResponse): """Track token usage for analytics (same as sync version)""" if llm_response.prompt_tokens: @@ -1447,66 +1025,17 @@ def clear_history(self) -> None: } def filter_elements(self, snapshot: Snapshot, goal: str | None = None) -> list[Element]: - """Filter elements from snapshot based on goal context (same as sync version)""" - elements = snapshot.elements - - # If no goal provided, return all elements (up to limit) - if not goal: - return elements[: self.default_snapshot_limit] - - goal_lower = goal.lower() - - # Extract keywords from goal - keywords = self._extract_keywords(goal_lower) - - # Boost elements matching goal keywords - scored_elements = [] - for el in elements: - score = el.importance - - # Boost if element text matches goal - if el.text and any(kw in el.text.lower() for kw in keywords): - score += 0.3 - - # Boost if role matches goal intent - if "click" in goal_lower and el.visual_cues.is_clickable: - score += 0.2 - if "type" in goal_lower and el.role in ["textbox", "searchbox"]: - score += 0.2 - if "search" in goal_lower: - # Filter out non-interactive elements for search tasks - if el.role in ["link", "img"] and not el.visual_cues.is_primary: - score -= 0.5 - - scored_elements.append((score, el)) - - # Re-sort by boosted score - scored_elements.sort(key=lambda x: x[0], reverse=True) - elements = [el for _, el in scored_elements] - - return elements[: self.default_snapshot_limit] - - def _extract_keywords(self, text: str) -> list[str]: - """Extract meaningful keywords from goal text (same as sync version)""" - stopwords = { - "the", - "a", - "an", - "and", - "or", - "but", - "in", - "on", - "at", - "to", - "for", - "of", - "with", - "by", - "from", - "as", - "is", - "was", - } - words = text.split() - return [w for w in words if w not in stopwords and len(w) > 2] + """ + Filter elements from snapshot based on goal context. + + This implementation uses ElementFilter to apply goal-based keyword matching + to boost relevant elements and filters out irrelevant ones. + + Args: + snapshot: Current page snapshot + goal: User's goal (can inform filtering) + + Returns: + Filtered list of elements + """ + return ElementFilter.filter_by_goal(snapshot, goal, self.default_snapshot_limit) diff --git a/sentience/browser_evaluator.py b/sentience/browser_evaluator.py index 79238a9..3cae2b4 100644 --- a/sentience/browser_evaluator.py +++ b/sentience/browser_evaluator.py @@ -21,7 +21,7 @@ class BrowserEvaluator: @staticmethod def wait_for_extension( - page: Union[Page, AsyncPage], + page: Page | AsyncPage, timeout_ms: int = 5000, ) -> None: """ @@ -79,7 +79,7 @@ async def wait_for_extension_async( ) from e @staticmethod - def _gather_diagnostics(page: Union[Page, AsyncPage]) -> dict[str, Any]: + def _gather_diagnostics(page: Page | AsyncPage) -> dict[str, Any]: """ Gather diagnostics about extension state. @@ -297,4 +297,3 @@ async def verify_method_exists_async( return await page.evaluate(f"typeof window.sentience.{method_name} !== 'undefined'") except Exception: return False - diff --git a/sentience/cloud_tracing.py b/sentience/cloud_tracing.py index 0631718..7c55c54 100644 --- a/sentience/cloud_tracing.py +++ b/sentience/cloud_tracing.py @@ -13,11 +13,11 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Optional, Protocol, Union -from collections.abc import Callable import requests from sentience.models import TraceStats +from sentience.trace_file_manager import TraceFileManager from sentience.tracing import TraceSink @@ -99,7 +99,7 @@ def __init__( # Use persistent cache directory instead of temp file # This ensures traces survive process crashes cache_dir = Path.home() / ".sentience" / "traces" / "pending" - cache_dir.mkdir(parents=True, exist_ok=True) + TraceFileManager.ensure_directory(cache_dir) # Persistent file (survives process crash) self._path = cache_dir / f"{run_id}.jsonl" @@ -125,9 +125,7 @@ def emit(self, event: dict[str, Any]) -> None: if self._closed: raise RuntimeError("CloudTraceSink is closed") - json_str = json.dumps(event, ensure_ascii=False) - self._trace_file.write(json_str + "\n") - self._trace_file.flush() # Ensure written to disk + TraceFileManager.write_event(self._trace_file, event) def close( self, @@ -386,7 +384,9 @@ def _upload_index(self) -> None: if self.logger: self.logger.warning(f"Error uploading trace index: {e}") - def _infer_final_status_from_trace(self) -> str: + def _infer_final_status_from_trace( + self, events: list[dict[str, Any]], run_end: dict[str, Any] | None + ) -> str: """ Infer final status from trace events by reading the trace file. @@ -437,92 +437,20 @@ def _infer_final_status_from_trace(self) -> str: # If we can't read the trace, default to unknown return "unknown" - def _extract_stats_from_trace(self) -> dict[str, Any]: + def _extract_stats_from_trace(self) -> TraceStats: """ Extract execution statistics from trace file. Returns: - Dictionary with stats fields for /v1/traces/complete + TraceStats with stats fields for /v1/traces/complete """ try: # Read trace file to extract stats - with open(self._path, encoding="utf-8") as f: - events = [] - for line in f: - line = line.strip() - if not line: - continue - try: - event = json.loads(line) - events.append(event) - except json.JSONDecodeError: - continue - - if not events: - return TraceStats( - total_steps=0, - total_events=0, - duration_ms=None, - final_status="unknown", - started_at=None, - ended_at=None, - ) - - # Find run_start and run_end events - run_start = next((e for e in events if e.get("type") == "run_start"), None) - run_end = next((e for e in events if e.get("type") == "run_end"), None) - - # Extract timestamps - started_at: str | None = None - ended_at: str | None = None - if run_start: - started_at = run_start.get("ts") - if run_end: - ended_at = run_end.get("ts") - - # Calculate duration - duration_ms: int | None = None - if started_at and ended_at: - try: - from datetime import datetime - - start_dt = datetime.fromisoformat(started_at.replace("Z", "+00:00")) - end_dt = datetime.fromisoformat(ended_at.replace("Z", "+00:00")) - delta = end_dt - start_dt - duration_ms = int(delta.total_seconds() * 1000) - except Exception: - pass - - # Count steps (from step_start events, only first attempt) - step_indices = set() - for event in events: - if event.get("type") == "step_start": - step_index = event.get("data", {}).get("step_index") - if step_index is not None: - step_indices.add(step_index) - total_steps = len(step_indices) if step_indices else 0 - - # If run_end has steps count, use that (more accurate) - if run_end: - steps_from_end = run_end.get("data", {}).get("steps") - if steps_from_end is not None: - total_steps = max(total_steps, steps_from_end) - - # Count total events - total_events = len(events) - - # Infer final status - final_status = self._infer_final_status_from_trace() - - return TraceStats( - total_steps=total_steps, - total_events=total_events, - duration_ms=duration_ms, - final_status=final_status, - started_at=started_at, - ended_at=ended_at, + events = TraceFileManager.read_events(self._path) + # Use TraceFileManager to extract stats (with custom status inference) + return TraceFileManager.extract_stats( + events, infer_status_func=self._infer_final_status_from_trace ) - except Exception as e: if self.logger: self.logger.warning(f"Error extracting stats from trace: {e}") @@ -594,28 +522,20 @@ def _extract_screenshots_from_trace(self) -> dict[int, dict[str, Any]]: sequence = 0 try: - with open(self._path, encoding="utf-8") as f: - for line in f: - line = line.strip() - if not line: - continue - - try: - event = json.loads(line) - # Check if this is a snapshot event with screenshot - if event.get("type") == "snapshot": - data = event.get("data", {}) - screenshot_base64 = data.get("screenshot_base64") - - if screenshot_base64: - sequence += 1 - screenshots[sequence] = { - "base64": screenshot_base64, - "format": data.get("screenshot_format", "jpeg"), - "step_id": event.get("step_id"), - } - except json.JSONDecodeError: - continue + events = TraceFileManager.read_events(self._path) + for event in events: + # Check if this is a snapshot event with screenshot + if event.get("type") == "snapshot": + data = event.get("data", {}) + screenshot_base64 = data.get("screenshot_base64") + + if screenshot_base64: + sequence += 1 + screenshots[sequence] = { + "base64": screenshot_base64, + "format": data.get("screenshot_format", "jpeg"), + "step_id": event.get("step_id"), + } except Exception as e: if self.logger: self.logger.error(f"Error extracting screenshots: {e}") @@ -630,34 +550,23 @@ def _create_cleaned_trace(self, output_path: Path) -> None: output_path: Path to write cleaned trace file """ try: - with ( - open(self._path, encoding="utf-8") as infile, - open(output_path, "w", encoding="utf-8") as outfile, - ): - for line in infile: - line = line.strip() - if not line: - continue - - try: - event = json.loads(line) - # Remove screenshot_base64 from snapshot events - if event.get("type") == "snapshot": - data = event.get("data", {}) - if "screenshot_base64" in data: - # Create copy without screenshot fields - cleaned_data = { - k: v - for k, v in data.items() - if k not in ("screenshot_base64", "screenshot_format") - } - event["data"] = cleaned_data - - # Write cleaned event - outfile.write(json.dumps(event, ensure_ascii=False) + "\n") - except json.JSONDecodeError: - # Skip invalid lines - continue + events = TraceFileManager.read_events(self._path) + with open(output_path, "w", encoding="utf-8") as outfile: + for event in events: + # Remove screenshot_base64 from snapshot events + if event.get("type") == "snapshot": + data = event.get("data", {}) + if "screenshot_base64" in data: + # Create copy without screenshot fields + cleaned_data = { + k: v + for k, v in data.items() + if k not in ("screenshot_base64", "screenshot_format") + } + event["data"] = cleaned_data + + # Write cleaned event + TraceFileManager.write_event(outfile, event) except Exception as e: if self.logger: self.logger.error(f"Error creating cleaned trace: {e}") diff --git a/sentience/element_filter.py b/sentience/element_filter.py new file mode 100644 index 0000000..df117b9 --- /dev/null +++ b/sentience/element_filter.py @@ -0,0 +1,134 @@ +""" +Element filtering utilities for agent-based element selection. + +This module provides centralized element filtering logic to reduce duplication +across agent implementations. +""" + +from typing import Optional + +from .models import Element, Snapshot + + +class ElementFilter: + """ + Centralized element filtering logic for agent-based element selection. + + Provides static methods for filtering elements based on: + - Importance scores + - Goal-based keyword matching + - Role and visual properties + """ + + # Common stopwords for keyword extraction + STOPWORDS = { + "the", + "a", + "an", + "and", + "or", + "but", + "in", + "on", + "at", + "to", + "for", + "of", + "with", + "by", + "from", + "as", + "is", + "was", + } + + @staticmethod + def filter_by_importance( + snapshot: Snapshot, + max_elements: int = 50, + ) -> list[Element]: + """ + Filter elements by importance score (simple top-N selection). + + Args: + snapshot: Current page snapshot + max_elements: Maximum number of elements to return + + Returns: + Top N elements sorted by importance score + """ + elements = snapshot.elements + # Elements are already sorted by importance in snapshot + return elements[:max_elements] + + @staticmethod + def filter_by_goal( + snapshot: Snapshot, + goal: str | None, + max_elements: int = 50, + ) -> list[Element]: + """ + Filter elements from snapshot based on goal context. + + Applies goal-based keyword matching to boost relevant elements + and filters out irrelevant ones. + + Args: + snapshot: Current page snapshot + goal: User's goal (can inform filtering) + max_elements: Maximum number of elements to return + + Returns: + Filtered list of elements sorted by boosted importance score + """ + elements = snapshot.elements + + # If no goal provided, return all elements (up to limit) + if not goal: + return elements[:max_elements] + + goal_lower = goal.lower() + + # Extract keywords from goal + keywords = ElementFilter._extract_keywords(goal_lower) + + # Boost elements matching goal keywords + scored_elements = [] + for el in elements: + score = el.importance + + # Boost if element text matches goal + if el.text and any(kw in el.text.lower() for kw in keywords): + score += 0.3 + + # Boost if role matches goal intent + if "click" in goal_lower and el.visual_cues.is_clickable: + score += 0.2 + if "type" in goal_lower and el.role in ["textbox", "searchbox"]: + score += 0.2 + if "search" in goal_lower: + # Filter out non-interactive elements for search tasks + if el.role in ["link", "img"] and not el.visual_cues.is_primary: + score -= 0.5 + + scored_elements.append((score, el)) + + # Re-sort by boosted score + scored_elements.sort(key=lambda x: x[0], reverse=True) + elements = [el for _, el in scored_elements] + + return elements[:max_elements] + + @staticmethod + def _extract_keywords(text: str) -> list[str]: + """ + Extract meaningful keywords from goal text. + + Args: + text: Text to extract keywords from + + Returns: + List of keywords (non-stopwords, length > 2) + """ + words = text.split() + return [w for w in words if w not in ElementFilter.STOPWORDS and len(w) > 2] diff --git a/sentience/formatting.py b/sentience/formatting.py index f8961c5..b8dd653 100644 --- a/sentience/formatting.py +++ b/sentience/formatting.py @@ -1,59 +1,15 @@ """ Snapshot formatting utilities for LLM prompts. -Provides functions to convert Sentience snapshots into text format suitable -for LLM consumption. -""" - -from typing import List - -from .models import Snapshot - - -def format_snapshot_for_llm(snap: Snapshot, limit: int = 50) -> str: - """ - Convert snapshot elements to text format for LLM consumption. - - This is the canonical way Sentience formats DOM state for LLMs. - The format includes element ID, role, text preview, visual cues, - position, and importance score. +DEPRECATED: This module is maintained for backward compatibility only. +New code should import from sentience.utils.formatting or sentience directly: - Args: - snap: Snapshot object with elements - limit: Maximum number of elements to include (default: 50) - - Returns: - Formatted string with one element per line - - Example: - >>> snap = snapshot(browser) - >>> formatted = format_snapshot_for_llm(snap, limit=10) - >>> print(formatted) - [1]