token usage & video recording

SentienceDEV · SentienceDEV · commit 810391697f27 · 2026-02-14T21:34:05.000-08:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -84,6 +84,28 @@ from predicate.llm_provider import LocalLLMProvider
 llm = LocalLLMProvider(model_name="Qwen/Qwen2.5-3B-Instruct", device="auto", load_in_4bit=True)
 ```
 
+##### Opt-in token usage accounting (best-effort)
+
+If you want to measure token spend, you can enable best-effort accounting (depends on provider reporting `prompt_tokens` / `completion_tokens` / `total_tokens` in `LLMResponse`):
+
+```python
+from predicate import PredicateBrowserAgentConfig
+
+config = PredicateBrowserAgentConfig(token_usage_enabled=True)
+
+# Later:
+usage = agent.get_token_usage()
+agent.reset_token_usage()
+```
+
+##### RuntimeAgent: act once without step lifecycle (orchestrators)
+
+`RuntimeAgent` now exposes `act_once(...)` helpers that execute exactly one action **without** calling `runtime.begin_step()` / `runtime.emit_step_end()`. This is intended for external orchestrators (e.g. WebBench) that already own step lifecycle and just want the SDK’s snapshot-first propose+execute block.
+
+- `await agent.act_once(...) -> str`
+- `await agent.act_once_with_snapshot(...) -> (action, snap)`
+- `await agent.act_once_result(...) -> { action, snap, used_vision }`
+
 ### 2026-02-13
 
 #### Expanded deterministic verifications (adaptive resnapshotting)
diff --git a/examples/agent/README.md b/examples/agent/README.md
@@ -2,4 +2,5 @@ Predicate agent examples.
 
 - `predicate_browser_agent_minimal.py`: minimal `PredicateBrowserAgent` usage.
 - `predicate_browser_agent_custom_prompt.py`: customize the compact prompt builder.
+- `predicate_browser_agent_video_recording_playwright.py`: enable Playwright video recording via context options (recommended).
 
diff --git a/examples/agent/predicate_browser_agent_video_recording_playwright.py b/examples/agent/predicate_browser_agent_video_recording_playwright.py
@@ -0,0 +1,92 @@
+"""
+Example: PredicateBrowserAgent + Playwright video recording (recommended approach).
+
+Video recording is a *Playwright context feature* (record_video_dir), not a PredicateBrowserAgent knob.
+This example shows how to:
+1) create a Playwright context with video recording enabled
+2) wrap the existing page with AsyncSentienceBrowser.from_page(...)
+3) use AgentRuntime + PredicateBrowserAgent normally
+
+Usage:
+  python examples/agent/predicate_browser_agent_video_recording_playwright.py
+"""
+
+import asyncio
+import os
+from pathlib import Path
+
+from playwright.async_api import async_playwright
+
+from predicate import AsyncSentienceBrowser, PredicateBrowserAgent, PredicateBrowserAgentConfig
+from predicate.agent_runtime import AgentRuntime
+from predicate.llm_provider import LLMProvider, LLMResponse
+from predicate.runtime_agent import RuntimeStep
+
+
+class FixedActionProvider(LLMProvider):
+    def __init__(self, action: str):
+        super().__init__(model="fixed-action")
+        self._action = action
+
+    def generate(self, system_prompt: str, user_prompt: str, **kwargs) -> LLMResponse:
+        _ = system_prompt, user_prompt, kwargs
+        return LLMResponse(content=self._action, model_name=self.model_name)
+
+    def supports_json_mode(self) -> bool:
+        return False
+
+    @property
+    def model_name(self) -> str:
+        return "fixed-action"
+
+
+async def main() -> None:
+    api_key = os.environ.get("PREDICATE_API_KEY") or os.environ.get("SENTIENCE_API_KEY")
+
+    recordings_dir = Path("recordings")
+    recordings_dir.mkdir(parents=True, exist_ok=True)
+
+    async with async_playwright() as p:
+        browser = await p.chromium.launch(headless=False)
+        context = await browser.new_context(
+            record_video_dir=str(recordings_dir),
+            record_video_size={"width": 1280, "height": 720},
+        )
+        page = await context.new_page()
+
+        # Wrap existing Playwright page.
+        sentience_browser = await AsyncSentienceBrowser.from_page(
+            page, api_key=api_key
+        )
+
+        try:
+            await page.goto("https://example.com")
+            await page.wait_for_load_state("networkidle")
+
+            runtime = await AgentRuntime.from_sentience_browser(
+                browser=sentience_browser, page=page, tracer=None
+            )
+
+            agent = PredicateBrowserAgent(
+                runtime=runtime,
+                executor=FixedActionProvider("FINISH()"),
+                config=PredicateBrowserAgentConfig(history_last_n=0),
+            )
+
+            out = await agent.step(
+                task_goal="Open example.com",
+                step=RuntimeStep(goal="Finish immediately"),
+            )
+            print(f"step ok: {out.ok}")
+            print(f"videos will be saved under: {recordings_dir.resolve()}")
+        finally:
+            # Close the Playwright context to flush the video.
+            try:
+                await context.close()
+            finally:
+                await browser.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
diff --git a/predicate/agents/browser_agent.py b/predicate/agents/browser_agent.py
@@ -10,7 +10,7 @@
 from ..captcha import CaptchaHandler, CaptchaOptions
 from ..captcha_strategies import ExternalSolver, HumanHandoffSolver, VisionSolver
 from ..llm_interaction_handler import LLMInteractionHandler
-from ..llm_provider import LLMProvider
+from ..llm_provider import LLMProvider, LLMResponse
 from ..models import Snapshot, StepHookContext
 from ..permissions import PermissionPolicy
 from ..runtime_agent import RuntimeAgent, RuntimeStep
@@ -84,6 +84,9 @@ class PredicateBrowserAgentConfig:
     # Prompt / token controls
     history_last_n: int = 0  # 0 disables LLM-facing step history (lowest token usage)
 
+    # Opt-in: track token usage from LLM provider responses (best-effort; depends on provider reporting).
+    token_usage_enabled: bool = False
+
     # Compact prompt customization
     # Signature: builder(task_goal, step_goal, dom_context, snapshot, history_summary) -> (system, user)
     compact_prompt_builder: Callable[
@@ -146,6 +149,112 @@ def apply_captcha_config_to_runtime(
     )
 
 
+@dataclass
+class TokenUsageTotals:
+    calls: int = 0
+    prompt_tokens: int = 0
+    completion_tokens: int = 0
+    total_tokens: int = 0
+
+    def add(self, resp: LLMResponse) -> None:
+        self.calls += 1
+        pt = resp.prompt_tokens if isinstance(resp.prompt_tokens, int) else 0
+        ct = resp.completion_tokens if isinstance(resp.completion_tokens, int) else 0
+        tt = resp.total_tokens if isinstance(resp.total_tokens, int) else (pt + ct)
+        self.prompt_tokens += max(0, int(pt))
+        self.completion_tokens += max(0, int(ct))
+        self.total_tokens += max(0, int(tt))
+
+
+class _TokenUsageCollector:
+    def __init__(self) -> None:
+        self._by_role: dict[str, TokenUsageTotals] = {}
+        self._by_model: dict[str, TokenUsageTotals] = {}
+
+    def record(self, *, role: str, resp: LLMResponse) -> None:
+        self._by_role.setdefault(role, TokenUsageTotals()).add(resp)
+        m = str(resp.model_name or "").strip() or "unknown"
+        self._by_model.setdefault(m, TokenUsageTotals()).add(resp)
+
+    def reset(self) -> None:
+        self._by_role.clear()
+        self._by_model.clear()
+
+    def summary(self) -> dict[str, Any]:
+        def _sum(items: dict[str, TokenUsageTotals]) -> TokenUsageTotals:
+            out = TokenUsageTotals()
+            for t in items.values():
+                out.calls += t.calls
+                out.prompt_tokens += t.prompt_tokens
+                out.completion_tokens += t.completion_tokens
+                out.total_tokens += t.total_tokens
+            return out
+
+        total = _sum(self._by_role)
+        return {
+            "total": {
+                "calls": total.calls,
+                "prompt_tokens": total.prompt_tokens,
+                "completion_tokens": total.completion_tokens,
+                "total_tokens": total.total_tokens,
+            },
+            "by_role": {
+                k: {
+                    "calls": v.calls,
+                    "prompt_tokens": v.prompt_tokens,
+                    "completion_tokens": v.completion_tokens,
+                    "total_tokens": v.total_tokens,
+                }
+                for k, v in self._by_role.items()
+            },
+            "by_model": {
+                k: {
+                    "calls": v.calls,
+                    "prompt_tokens": v.prompt_tokens,
+                    "completion_tokens": v.completion_tokens,
+                    "total_tokens": v.total_tokens,
+                }
+                for k, v in self._by_model.items()
+            },
+        }
+
+
+class _TokenAccountingProvider(LLMProvider):
+    def __init__(self, *, inner: LLMProvider, collector: _TokenUsageCollector, role: str):
+        super().__init__(model=getattr(inner, "model_name", "wrapped"))
+        self._inner = inner
+        self._collector = collector
+        self._role = role
+
+    def generate(self, system_prompt: str, user_prompt: str, **kwargs) -> LLMResponse:
+        resp = self._inner.generate(system_prompt, user_prompt, **kwargs)
+        try:
+            self._collector.record(role=self._role, resp=resp)
+        except Exception:
+            pass
+        return resp
+
+    def supports_json_mode(self) -> bool:
+        return self._inner.supports_json_mode()
+
+    def supports_vision(self) -> bool:
+        return self._inner.supports_vision()
+
+    def generate_with_image(
+        self, system_prompt: str, user_prompt: str, image_base64: str, **kwargs
+    ) -> LLMResponse:
+        resp = self._inner.generate_with_image(system_prompt, user_prompt, image_base64, **kwargs)
+        try:
+            self._collector.record(role=self._role, resp=resp)
+        except Exception:
+            pass
+        return resp
+
+    @property
+    def model_name(self) -> str:
+        return self._inner.model_name
+
+
 class _RuntimeAgentWithPromptOverrides(RuntimeAgent):
     def __init__(
         self,
@@ -227,9 +336,33 @@ def __init__(
         config: PredicateBrowserAgentConfig = PredicateBrowserAgentConfig(),
     ) -> None:
         self.runtime = runtime
-        self.executor = executor
-        self.vision_executor = vision_executor
-        self.vision_verifier = vision_verifier
+        self._token_usage: _TokenUsageCollector | None = (
+            _TokenUsageCollector() if bool(config.token_usage_enabled) else None
+        )
+
+        # Optionally wrap providers for best-effort token usage accounting.
+        if self._token_usage is not None:
+            self.executor = _TokenAccountingProvider(
+                inner=executor, collector=self._token_usage, role="executor"
+            )
+            self.vision_executor = (
+                _TokenAccountingProvider(
+                    inner=vision_executor, collector=self._token_usage, role="vision_executor"
+                )
+                if vision_executor is not None
+                else None
+            )
+            self.vision_verifier = (
+                _TokenAccountingProvider(
+                    inner=vision_verifier, collector=self._token_usage, role="vision_verifier"
+                )
+                if vision_verifier is not None
+                else None
+            )
+        else:
+            self.executor = executor
+            self.vision_executor = vision_executor
+            self.vision_verifier = vision_verifier
         self.config = config
 
         # LLM-facing step history summaries (bounded)
@@ -252,6 +385,23 @@ def __init__(
             history_summary_provider=self._get_history_summary,
         )
 
+    def get_token_usage(self) -> dict[str, Any]:
+        """
+        Best-effort token usage summary.
+
+        Only available when `PredicateBrowserAgentConfig.token_usage_enabled=True`.
+        """
+        if self._token_usage is None:
+            return {"enabled": False, "reason": "token_usage_enabled is False"}
+        out = self._token_usage.summary()
+        out["enabled"] = True
+        return out
+
+    def reset_token_usage(self) -> None:
+        if self._token_usage is None:
+            return
+        self._token_usage.reset()
+
     def _get_history_summary(self) -> str:
         if int(self.config.history_last_n) <= 0:
             return ""
diff --git a/tests/unit/test_predicate_browser_agent.py b/tests/unit/test_predicate_browser_agent.py
@@ -95,6 +95,29 @@ def model_name(self) -> str:
         return self._model_name
 
 
+class TokenProviderStub(LLMProvider):
+    def __init__(self, *, model: str = "stub", response: str = "FINISH()"):
+        super().__init__(model)
+        self._response = response
+
+    def generate(self, system_prompt: str, user_prompt: str, **kwargs) -> LLMResponse:
+        _ = system_prompt, user_prompt, kwargs
+        return LLMResponse(
+            content=self._response,
+            model_name=self.model_name,
+            prompt_tokens=11,
+            completion_tokens=7,
+            total_tokens=18,
+        )
+
+    def supports_json_mode(self) -> bool:
+        return True
+
+    @property
+    def model_name(self) -> str:
+        return self._model_name
+
+
 def make_snapshot(*, url: str, elements: list[Element], confidence: float | None = None) -> Snapshot:
     diagnostics = SnapshotDiagnostics(confidence=confidence) if confidence is not None else None
     return Snapshot(
@@ -162,3 +185,39 @@ def builder(
 
     asyncio.run(_run())
 
+
+def test_predicate_browser_agent_token_usage_is_opt_in_and_best_effort() -> None:
+    async def _run() -> None:
+        backend = MockBackend()
+        tracer = MockTracer()
+        runtime = AgentRuntime(backend=backend, tracer=tracer)
+
+        s0 = make_snapshot(url="https://example.com/start", elements=[make_clickable_element(1)])
+        async def fake_snapshot(**_kwargs):
+            runtime.last_snapshot = s0
+            return runtime.last_snapshot
+        runtime.snapshot = AsyncMock(side_effect=fake_snapshot)  # type: ignore[method-assign]
+
+        step = RuntimeStep(goal="No-op", verifications=[])
+        executor = TokenProviderStub(response="FINISH()")
+
+        agent = PredicateBrowserAgent(
+            runtime=runtime,
+            executor=executor,
+            config=PredicateBrowserAgentConfig(token_usage_enabled=True),
+        )
+
+        out = await agent.step(task_goal="test", step=step)
+        assert out.ok is True
+
+        usage = agent.get_token_usage()
+        assert usage["enabled"] is True
+        assert usage["total"]["total_tokens"] >= 18
+        assert usage["by_role"]["executor"]["calls"] >= 1
+
+        agent.reset_token_usage()
+        usage2 = agent.get_token_usage()
+        assert usage2["total"]["total_tokens"] == 0
+
+    asyncio.run(_run())
+

Original file line number	Diff line number	Diff line change
`@@ -2,4 +2,5 @@ Predicate agent examples.`
`2`	`2`
`3`	`3`	- `predicate_browser_agent_minimal.py`: minimal `PredicateBrowserAgent` usage.
`4`	`4`	- `predicate_browser_agent_custom_prompt.py`: customize the compact prompt builder.
	`5`	+- `predicate_browser_agent_video_recording_playwright.py`: enable Playwright video recording via context options (recommended).
`5`	`6`