Spaces:
Sleeping
Sleeping
File size: 6,483 Bytes
71f36c6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | """
Strategy Reflector — Self-Improvement Engine
==============================================
Analyzes agent performance and generates improvement strategies.
"""
import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
logger = logging.getLogger("openclaw.strategy")
class StrategyReflector:
"""Analyzes performance and suggests improvements."""
def __init__(self, state_dir: str = "state"):
self.state_dir = Path(state_dir)
def analyze(self) -> dict:
"""Run full analysis of agent performance."""
metrics = self._gather_metrics()
insights = self._derive_insights(metrics)
strategy = self._generate_strategy(insights)
report = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"metrics": metrics,
"insights": insights,
"strategy": strategy,
}
# Save report
self.state_dir.mkdir(parents=True, exist_ok=True)
with open(self.state_dir / "strategy_report.json", "w") as f:
json.dump(report, f, indent=2)
return report
def _gather_metrics(self) -> dict:
"""Gather all available metrics."""
metrics = {
"total_cycles": 0,
"total_posts": 0,
"total_engagements": 0,
"papers_shared": 0,
"errors": 0,
"uptime_hours": 0,
"post_frequency": 0,
"services_available": 0,
}
# Load from agent state
state_file = self.state_dir / "agent_state.json"
if state_file.exists():
try:
with open(state_file) as f:
state = json.load(f)
metrics["total_cycles"] = state.get("cycle_count", 0)
metrics["total_posts"] = state.get("posts_created", 0)
metrics["total_engagements"] = state.get("engagement_count", 0)
metrics["papers_shared"] = len(state.get("posted_paper_ids", []))
metrics["errors"] = len(state.get("errors", []))
# Calculate uptime
started = state.get("started_at", "")
if started:
try:
start_dt = datetime.fromisoformat(started)
if start_dt.tzinfo is None:
start_dt = start_dt.replace(tzinfo=timezone.utc)
delta = datetime.now(timezone.utc) - start_dt
metrics["uptime_hours"] = round(delta.total_seconds() / 3600, 1)
except Exception:
pass
except Exception:
pass
# Load from post history
history_file = self.state_dir / "post_history.json"
if history_file.exists():
try:
with open(history_file) as f:
history = json.load(f)
if history and metrics["uptime_hours"] > 0:
metrics["post_frequency"] = round(
len(history) / max(metrics["uptime_hours"] / 24, 1), 2
)
except Exception:
pass
return metrics
def _derive_insights(self, metrics: dict) -> list[str]:
"""Derive actionable insights from metrics."""
insights = []
if metrics["total_cycles"] == 0:
insights.append("Agent has not completed any cycles yet. First run pending.")
return insights
# Post frequency analysis
if metrics["total_posts"] == 0:
insights.append("CRITICAL: No posts created. Check Moltbook API connection and account status.")
elif metrics["post_frequency"] < 2:
insights.append("Low post frequency. Consider increasing research post rate or adding more platforms.")
elif metrics["post_frequency"] > 10:
insights.append("High post frequency. Risk of appearing spammy. Consider quality over quantity.")
# Engagement analysis
if metrics["total_engagements"] == 0 and metrics["total_cycles"] > 5:
insights.append("No engagements despite multiple cycles. Review engagement strategy and keyword matching.")
# Error rate
if metrics["errors"] > metrics["total_cycles"] * 0.3:
insights.append(f"High error rate ({metrics['errors']}/{metrics['total_cycles']}). Investigate API failures.")
# Paper sharing
if metrics["papers_shared"] < 3 and metrics["total_cycles"] > 10:
insights.append("Few papers shared. Ensure ArXiv fetcher is working and paper cache is populated.")
# Platform diversity
insights.append("Currently using Moltbook only. Consider adding: Chirper.ai, Reddit, Twitter for wider reach.")
if not insights:
insights.append("Agent operating within normal parameters.")
return insights
def _generate_strategy(self, insights: list[str]) -> dict:
"""Generate improvement strategy from insights."""
actions = []
priorities = []
for insight in insights:
if "CRITICAL" in insight:
priorities.append("HIGH: " + insight)
actions.append("Diagnose and fix API connection immediately")
elif "No posts" in insight or "No engagements" in insight:
priorities.append("MEDIUM: " + insight)
actions.append("Review API keys and platform access")
elif "platform" in insight.lower() or "wider reach" in insight.lower():
actions.append("Implement multi-platform support (Chirper.ai, Reddit)")
elif "error rate" in insight.lower():
actions.append("Add retry logic and circuit breaker patterns")
# Default strategic actions
actions.extend([
"Scan ArXiv weekly for papers citing our work",
"Track which topics generate most engagement",
"Build keyword database from successful interactions",
"Monitor new AI agent platforms for early adoption",
])
return {
"priorities": priorities,
"actions": actions[:10],
"next_review": "24 hours",
}
|