from __future__ import annotations from typing import ClassVar from .base import BaseAgent, AgentInput, AgentOutput class MomentumAgent(BaseAgent): """Characterises the user's recent engagement trend from profile features.""" agent_id: ClassVar[str] = "momentum" ttl_seconds: ClassVar[int] = 21600 # 6h version: ClassVar[str] = "1.0.0" def compute(self, inp: AgentInput) -> AgentOutput: completion = inp.profile.get("completion_rate_30d") dismiss = inp.profile.get("dismiss_rate_30d") volume = inp.profile.get("tip_volume_30d") parts: list[str] = [] if completion is not None: pct = round(completion * 100) if pct >= 50: parts.append(f"The user completes {pct}% of tips (strong engagement).") elif pct >= 25: parts.append(f"The user completes {pct}% of tips (moderate engagement).") else: parts.append( f"The user completes {pct}% of tips " f"(low engagement — prefer simple, immediately actionable tips)." ) else: parts.append("No completion-rate data yet (new user).") if dismiss is not None: dpct = round(dismiss * 100) if dpct >= 40: parts.append(f"Dismiss rate is high ({dpct}%) — avoid repetitive or irrelevant tips.") elif dpct <= 10: parts.append(f"Dismiss rate is low ({dpct}%).") if volume is not None and int(volume) < 5: parts.append("Very few tips served so far — this is an early-stage user.") prompt = " ".join(parts) if parts else "No engagement data available yet." snapshot = { "completion_rate_30d": completion, "dismiss_rate_30d": dismiss, "tip_volume_30d": volume, } return self._make_output(inp, prompt, snapshot)