import asyncio
import importlib
import logging
from collections import deque
from datetime import datetime, timedelta, timezone
from devplacepy.services.base import BaseService, ConfigField
from devplacepy.services.bot import config
logger = logging.getLogger(__name__)
class BotsService(BaseService):
default_enabled = False
min_interval = 5
title = "Bots"
description = (
"Runs a fleet of Playwright-driven AI personas that browse a DevPlace "
"instance and post, comment, vote, and react like real users. Fleet size, "
"target site, and LLM settings are configurable, with live cost and usage "
"metrics per bot."
)
config_fields = [
ConfigField("bot_fleet_size", "Fleet size (bots)", type="int", default=1, minimum=0, maximum=20,
help="Number of concurrent bot accounts to run.", group="Fleet"),
ConfigField("bot_headless", "Headless browser", type="bool", default=True,
help="Run Chromium headless. Disable only for local debugging on a desktop.",
group="Fleet"),
ConfigField("bot_max_actions", "Max actions per bot", type="int", default=0, minimum=0,
help="Stop a bot after this many actions (0 = run indefinitely).", group="Fleet"),
ConfigField("bot_base_url", "Target site URL", type="url", default=config.BASE_URL_DEFAULT,
help="DevPlace instance the bots browse and post to.", group="Target"),
ConfigField("bot_news_api", "News API URL", type="url", default=config.NEWS_API_DEFAULT,
help="Source of articles the bots react to.", group="Target"),
ConfigField("bot_api_url", "LLM API URL", type="url", default=config.API_URL_DEFAULT,
help="OpenAI-compatible chat-completions endpoint.", group="LLM"),
ConfigField("bot_model", "LLM model", type="str", default=config.MODEL_DEFAULT,
help="Model name sent to the LLM API.", group="LLM"),
ConfigField("bot_api_key", "LLM API key", type="password", default="", secret=True,
help="Defaults to the gateway's internal key (the bots use the local AI gateway).",
group="LLM"),
ConfigField("bot_input_cost_per_1m", "Input cost per 1M tokens ($)", type="float",
default=config.INPUT_COST_PER_1M_DEFAULT, minimum=0,
help="Used to compute live spend per bot and across the fleet.", group="LLM"),
ConfigField("bot_output_cost_per_1m", "Output cost per 1M tokens ($)", type="float",
default=config.OUTPUT_COST_PER_1M_DEFAULT, minimum=0,
help="Used to compute live spend per bot and across the fleet.", group="LLM"),
]
def __init__(self):
super().__init__(name="bots", interval_seconds=15)
self._fleet = {}
self._state_dir = config.STATE_DIR
self._cost_samples = deque()
self._cost_anchor = None
def _resolve_api_key(self, cfg: dict) -> str:
from devplacepy.database import internal_gateway_key
return cfg["bot_api_key"] or internal_gateway_key()
@staticmethod
def _ensure_importable() -> None:
importlib.import_module("devplacepy.services.bot.bot")
async def run_once(self) -> None:
cfg = self.get_config()
desired = cfg["bot_fleet_size"]
api_key = self._resolve_api_key(cfg)
if not api_key:
if self._fleet:
await self._stop_fleet()
self.log("No LLM API key available (bot_api_key / gateway internal key); fleet idle")
return
try:
self._ensure_importable()
except ImportError as e:
if self._fleet:
await self._stop_fleet()
self.log(f"Bot dependencies unavailable ({e}); install the 'bots' extra")
return
for slot in list(self._fleet):
task = self._fleet[slot]["task"]
if task.done():
self._report_exit(slot, task)
del self._fleet[slot]
while len(self._fleet) > desired:
await self._stop_slot(max(self._fleet))
for slot in range(desired):
if slot not in self._fleet:
self._launch_slot(slot, cfg, api_key)
self.log(f"Fleet: {len(self._fleet)}/{desired} bots active")
def _report_exit(self, slot: int, task) -> None:
if task.cancelled():
return
exc = task.exception()
if exc:
self.log(f"bot{slot} exited with error: {exc}; relaunching")
else:
self.log(f"bot{slot} finished; relaunching")
def _launch_slot(self, slot: int, cfg: dict, api_key: str) -> None:
from devplacepy.services.bot.bot import DevPlaceBot
from devplacepy.services.bot.runtime import BotRuntimeConfig
rc = BotRuntimeConfig(
state_path=self._state_dir / f"state_slot{slot}.json",
base_url=cfg["bot_base_url"],
api_url=cfg["bot_api_url"],
news_api=cfg["bot_news_api"],
model=cfg["bot_model"],
api_key=api_key,
input_cost_per_1m=cfg["bot_input_cost_per_1m"],
output_cost_per_1m=cfg["bot_output_cost_per_1m"],
headless=cfg["bot_headless"],
registry_path=config.ARTICLE_REGISTRY_PATH,
)
bot = DevPlaceBot(rc, on_event=self.log)
task = asyncio.create_task(self._run_slot(slot, bot, cfg["bot_max_actions"]))
self._fleet[slot] = {"bot": bot, "task": task}
self.log(f"Launched bot{slot}")
async def _run_slot(self, slot: int, bot, max_actions: int) -> None:
try:
await bot.run_forever(action_limit=max_actions)
except asyncio.CancelledError:
raise
except Exception as e:
self.log(f"bot{slot} crashed: {e}")
async def on_disable(self) -> None:
await self._stop_fleet()
async def _stop_fleet(self) -> None:
for slot in list(self._fleet):
await self._stop_slot(slot)
async def _stop_slot(self, slot: int) -> None:
entry = self._fleet.pop(slot, None)
if not entry:
return
task = entry["task"]
task.cancel()
try:
await asyncio.wait_for(task, timeout=30)
except (asyncio.CancelledError, asyncio.TimeoutError):
pass
except Exception as e:
self.log(f"bot{slot} stop error: {e}")
self.log(f"Stopped bot{slot}")
def collect_metrics(self) -> dict:
rows = []
total_cost = 0.0
total_calls = total_in = total_out = 0
posts = comments = votes = running = 0
for slot in sorted(self._fleet):
entry = self._fleet[slot]
bot = entry["bot"]
alive = not entry["task"].done()
running += 1 if alive else 0
llm = getattr(bot, "llm", None)
st = getattr(bot, "state", None)
cost = llm.total_cost if llm else 0.0
calls = llm.total_calls if llm else 0
total_cost += cost
total_calls += calls
total_in += llm.total_in_tokens if llm else 0
total_out += llm.total_out_tokens if llm else 0
if st:
posts += st.created_posts
comments += st.comments_posted
votes += st.votes_cast
rows.append([
f"bot{slot}",
(st.username if st else "") or "-",
(st.persona if st else "") or "-",
"running" if alive else "stopped",
st.created_posts if st else 0,
st.comments_posted if st else 0,
st.votes_cast if st else 0,
calls,
f"${cost:.4f}",
])
window, observed_cost, ready = self._sample_cost(datetime.now(timezone.utc), total_cost)
cost_per_hour = observed_cost / window * 3600 if ready else 0.0
projected_24h = observed_cost / window * 86400 if ready else 0.0
rate_value = f"${cost_per_hour:.4f}/h" if ready else "warming up"
projected_value = f"${projected_24h:.2f}" if ready else "warming up"
stats = [
{"label": "Bots running", "value": running},
{"label": "Fleet cost", "value": f"${total_cost:.4f}"},
{"label": "Observed window", "value": f"{window:.0f}s"},
{"label": "Cost rate", "value": rate_value},
{"label": "Projected 24h cost", "value": projected_value},
{"label": "LLM calls", "value": total_calls},
{"label": "Tokens in", "value": total_in},
{"label": "Tokens out", "value": total_out},
{"label": "Posts", "value": posts},
{"label": "Comments", "value": comments},
{"label": "Votes", "value": votes},
]
table = {
"columns": ["Bot", "User", "Persona", "Status", "Posts", "Comments", "Votes", "Calls", "Cost"],
"rows": rows,
}
return {"stats": stats, "table": table}
def _sample_cost(self, now: datetime, total_cost: float) -> tuple[float, float, bool]:
if self._started_at is None:
self._cost_samples.clear()
self._cost_anchor = None
return 0.0, 0.0, False
if self._cost_anchor != self._started_at:
self._cost_samples.clear()
self._cost_anchor = self._started_at
self._cost_samples.append((now, total_cost))
cutoff = now - timedelta(seconds=config.COST_WINDOW_SECONDS)
while len(self._cost_samples) > 2 and self._cost_samples[0][0] < cutoff:
self._cost_samples.popleft()
start_at, start_cost = self._cost_samples[0]
window = (now - start_at).total_seconds()
observed_cost = max(0.0, total_cost - start_cost)
ready = window >= config.COST_WARMUP_SECONDS
return window, observed_cost, ready
Reference in New Issue View Git Blame Copy Permalink