Spaces:
Sleeping
Sleeping
| """Browser-use integration — execute step, result type, and trace converter.""" | |
| from __future__ import annotations | |
| import logging | |
| from dataclasses import dataclass, field | |
| from typing import Any, List, Optional | |
| from ..core.context import ACEStepContext | |
| from ..implementations.prompts import wrap_skillbook_for_external_agent | |
| logger = logging.getLogger(__name__) | |
| try: | |
| from browser_use import Agent, Browser | |
| BROWSER_USE_AVAILABLE = True | |
| except ImportError: | |
| BROWSER_USE_AVAILABLE = False | |
| Agent = None # type: ignore[misc,assignment] | |
| Browser = None # type: ignore[misc,assignment] | |
| # --------------------------------------------------------------------------- | |
| # Input / Output types | |
| # --------------------------------------------------------------------------- | |
| class BrowserResult: | |
| """Output from a browser-use execution. | |
| This is the integration-specific result — not yet in ACE trace format. | |
| Use ``BrowserToTrace`` to convert to a standardised trace dict. | |
| """ | |
| task: str | |
| success: bool | |
| output: str = "" | |
| error: Optional[str] = None | |
| steps_count: int = 0 | |
| duration_seconds: Optional[float] = None | |
| cited_skill_ids: List[str] = field(default_factory=list) | |
| chronological_steps: List[dict] = field(default_factory=list) | |
| raw_history: Any = None | |
| # --------------------------------------------------------------------------- | |
| # Execute step | |
| # --------------------------------------------------------------------------- | |
| class BrowserExecuteStep: | |
| """INJECT skillbook context and EXECUTE via browser-use Agent. | |
| Reads a task string from ``ctx.sample``, writes a ``BrowserResult`` | |
| to ``ctx.trace``. | |
| This is an **async** step — ``__call__`` is a coroutine because | |
| browser-use is an async framework. | |
| """ | |
| requires = frozenset({"sample", "skillbook"}) | |
| provides = frozenset({"trace"}) | |
| def __init__( | |
| self, browser_llm: Any, browser: Any = None, **agent_kwargs: Any | |
| ) -> None: | |
| if not BROWSER_USE_AVAILABLE: | |
| raise ImportError( | |
| "browser-use is not installed. Install with: " "pip install browser-use" | |
| ) | |
| self.browser_llm = browser_llm | |
| self.browser = browser | |
| self.agent_kwargs = agent_kwargs | |
| async def __call__(self, ctx: ACEStepContext) -> ACEStepContext: | |
| task: str = ctx.sample | |
| # -- INJECT -- | |
| enhanced_task = self._inject(task, ctx.skillbook) | |
| # -- EXECUTE -- | |
| agent_params: dict[str, Any] = { | |
| **self.agent_kwargs, | |
| "task": enhanced_task, | |
| "llm": self.browser_llm, | |
| } | |
| if self.browser is not None: | |
| agent_params["browser"] = self.browser | |
| success = False | |
| error: Optional[str] = None | |
| history: Any = None | |
| try: | |
| agent = Agent(**agent_params) | |
| history = await agent.run() | |
| success = True | |
| except Exception as exc: | |
| error = str(exc) | |
| result = self._build_result(task, history, success, error) | |
| return ctx.replace(trace=result) | |
| # ------------------------------------------------------------------ | |
| # Helpers | |
| # ------------------------------------------------------------------ | |
| def _inject(task: str, skillbook: Any) -> str: | |
| if skillbook is None: | |
| return task | |
| context = wrap_skillbook_for_external_agent(skillbook) | |
| if not context: | |
| return task | |
| return f"{task}\n\n{context}" | |
| def _build_result( | |
| task: str, | |
| history: Any, | |
| success: bool, | |
| error: Optional[str], | |
| ) -> BrowserResult: | |
| if history is None: | |
| return BrowserResult(task=task, success=success, error=error) | |
| # Extract basic info | |
| try: | |
| output = ( | |
| history.final_result() if hasattr(history, "final_result") else "" | |
| ) or "" | |
| except Exception: | |
| output = "" | |
| try: | |
| steps_count = ( | |
| history.number_of_steps() if hasattr(history, "number_of_steps") else 0 | |
| ) | |
| except Exception: | |
| steps_count = 0 | |
| duration: Optional[float] = None | |
| try: | |
| if hasattr(history, "total_duration_seconds"): | |
| duration = round(history.total_duration_seconds(), 2) | |
| except Exception: | |
| pass | |
| # Extract chronological step data | |
| chronological: list[dict] = [] | |
| try: | |
| if hasattr(history, "history"): | |
| for step_idx, step in enumerate(history.history, 1): | |
| step_data: dict[str, Any] = {"step_number": step_idx} | |
| if step.model_output: | |
| step_data["thought"] = { | |
| "thinking": step.model_output.thinking, | |
| "evaluation": step.model_output.evaluation_previous_goal, | |
| "memory": step.model_output.memory, | |
| "next_goal": step.model_output.next_goal, | |
| } | |
| if step.model_output.action: | |
| step_data["actions"] = [ | |
| {k: v for k, v in a.model_dump().items()} | |
| for a in step.model_output.action | |
| ] | |
| if step.result: | |
| step_data["results"] = [ | |
| { | |
| "is_done": r.is_done, | |
| "success": r.success, | |
| "error": r.error, | |
| "extracted_content": r.extracted_content, | |
| } | |
| for r in step.result | |
| ] | |
| if step.state: | |
| step_data["url"] = step.state.url | |
| chronological.append(step_data) | |
| except Exception as exc: | |
| logger.debug("Trace extraction error: %s", exc) | |
| # Extract cited skill IDs from agent thoughts | |
| cited_ids: list[str] = [] | |
| try: | |
| if hasattr(history, "model_thoughts"): | |
| thoughts = history.model_thoughts() | |
| thoughts_text = "\n".join( | |
| t.thinking | |
| for t in thoughts | |
| if hasattr(t, "thinking") and t.thinking | |
| ) | |
| from ..implementations.helpers import extract_cited_skill_ids | |
| cited_ids = extract_cited_skill_ids(thoughts_text) | |
| except Exception: | |
| pass | |
| return BrowserResult( | |
| task=task, | |
| success=success, | |
| output=output, | |
| error=error, | |
| steps_count=steps_count, | |
| duration_seconds=duration, | |
| cited_skill_ids=cited_ids, | |
| chronological_steps=chronological, | |
| raw_history=history, | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Convert step — BrowserResult → standardised trace dict | |
| # --------------------------------------------------------------------------- | |
| class BrowserToTrace: | |
| """Convert a ``BrowserResult`` on ``ctx.trace`` to the standardised | |
| trace dict that the learning tail (``ReflectStep``) expects. | |
| """ | |
| requires = frozenset({"trace"}) | |
| provides = frozenset({"trace"}) | |
| def __call__(self, ctx: ACEStepContext) -> ACEStepContext: | |
| r: BrowserResult = ctx.trace # type: ignore[assignment] | |
| # Build human-readable reasoning from chronological steps | |
| parts: list[str] = [] | |
| status = "succeeded" if r.success else "failed" | |
| parts.append(f"Browser task {status} in {r.steps_count} steps") | |
| if r.duration_seconds is not None: | |
| parts.append(f"Duration: {r.duration_seconds}s") | |
| if r.output: | |
| preview = r.output[:150] + ("..." if len(r.output) > 150 else "") | |
| parts.append(f"\nFinal output: {preview}") | |
| if r.error: | |
| parts.append(f"\nFailure reason: {r.error}") | |
| if r.chronological_steps: | |
| parts.append("\n\n=== BROWSER EXECUTION TRACE (Chronological) ===") | |
| for step in r.chronological_steps: | |
| step_num = step["step_number"] | |
| parts.append(f"\n--- Step {step_num} ---") | |
| if "thought" in step: | |
| thought = step["thought"] | |
| if thought.get("thinking"): | |
| parts.append(f"Thinking: {thought['thinking']}") | |
| if thought.get("evaluation"): | |
| parts.append(f" Evaluation: {thought['evaluation']}") | |
| if thought.get("next_goal"): | |
| parts.append(f" Next Goal: {thought['next_goal']}") | |
| if "actions" in step: | |
| for action in step["actions"]: | |
| name = next(iter(action), "unknown") | |
| parts.append(f"Action: {name}({action.get(name, {})})") | |
| if "results" in step: | |
| for res in step["results"]: | |
| res_parts = [] | |
| if res.get("success") is not None: | |
| res_parts.append(f"success={res['success']}") | |
| if res.get("error"): | |
| res_parts.append(f"error={res['error']}") | |
| if res.get("extracted_content"): | |
| res_parts.append( | |
| f"content={str(res['extracted_content'])[:200]}" | |
| ) | |
| parts.append(f"Result: {', '.join(res_parts)}") | |
| if "url" in step: | |
| parts.append(f"URL: {step['url']}") | |
| parts.append("\n=== END EXECUTION TRACE ===") | |
| reasoning = "\n".join(parts) | |
| feedback = f"Browser task {status} in {r.steps_count} steps" | |
| if r.duration_seconds is not None: | |
| feedback += f" ({r.duration_seconds}s)" | |
| if r.error: | |
| feedback += f"\nError: {r.error}" | |
| trace: dict = { | |
| "question": r.task, | |
| "reasoning": reasoning, | |
| "answer": r.output, | |
| "skill_ids": r.cited_skill_ids, | |
| "feedback": feedback, | |
| "ground_truth": None, | |
| } | |
| return ctx.replace(trace=trace) | |
| __all__ = [ | |
| "BrowserExecuteStep", | |
| "BrowserResult", | |
| "BrowserToTrace", | |
| ] | |