forked from retoor/devplacepy
feat: add soft delete and media tab for user attachments with admin restore
This commit is contained in:
@@ -5,12 +5,14 @@ import re
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from typing import Any
|
||||
|
||||
from devplacepy.services.bot.config import (
|
||||
GIST_LANGUAGES,
|
||||
GIST_MIN_LINES,
|
||||
PERSONA_GIST_FLAVOR,
|
||||
PERSONA_LANGUAGES,
|
||||
SEARCH_TERMS,
|
||||
TRIVIAL_GIST_TERMS,
|
||||
)
|
||||
|
||||
@@ -40,7 +42,7 @@ class LLMClient:
|
||||
self.total_in_tokens = 0
|
||||
self.total_out_tokens = 0
|
||||
|
||||
def _call(self, system: str, prompt: str, temperature: float = 0.7) -> str:
|
||||
def _raw_call(self, system: str, prompt: str, temperature: float = 0.7) -> str:
|
||||
for attempt in range(3):
|
||||
try:
|
||||
payload = {
|
||||
@@ -80,7 +82,7 @@ class LLMClient:
|
||||
self.total_calls += 1
|
||||
self.total_in_tokens += in_tokens
|
||||
self.total_out_tokens += out_tokens
|
||||
return self.clean(result["choices"][0]["message"]["content"])
|
||||
return result["choices"][0]["message"]["content"]
|
||||
except (
|
||||
urllib.error.HTTPError,
|
||||
urllib.error.URLError,
|
||||
@@ -93,6 +95,9 @@ class LLMClient:
|
||||
time.sleep(2**attempt)
|
||||
return ""
|
||||
|
||||
def _call(self, system: str, prompt: str, temperature: float = 0.7) -> str:
|
||||
return self.clean(self._raw_call(system, prompt, temperature))
|
||||
|
||||
@staticmethod
|
||||
def clean(text: str, preserve_md: bool = False) -> str:
|
||||
if not preserve_md:
|
||||
@@ -341,6 +346,183 @@ class LLMClient:
|
||||
return True, "ok"
|
||||
return False, verdict.strip()[:120] or "judge rejected"
|
||||
|
||||
@staticmethod
|
||||
def _parse_json(text: str) -> dict:
|
||||
text = LLMClient.strip_code_fences(text or "").strip()
|
||||
start = text.find("{")
|
||||
end = text.rfind("}")
|
||||
if start >= 0 and end > start:
|
||||
text = text[start : end + 1]
|
||||
parsed = json.loads(text)
|
||||
if not isinstance(parsed, dict):
|
||||
raise ValueError("expected a JSON object")
|
||||
return parsed
|
||||
|
||||
@staticmethod
|
||||
def _clamp01(value: Any, default: float) -> float:
|
||||
try:
|
||||
return max(0.0, min(1.0, float(value)))
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
@staticmethod
|
||||
def _as_int(value: Any, default: int) -> int:
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
@staticmethod
|
||||
def _as_terms(value: Any) -> list[str]:
|
||||
if isinstance(value, list):
|
||||
return [str(item).strip() for item in value if str(item).strip()][:8]
|
||||
if isinstance(value, str) and value.strip():
|
||||
return [part.strip() for part in value.split(",") if part.strip()][:8]
|
||||
return []
|
||||
|
||||
def _normalize_identity(self, data: dict, archetype: str) -> dict:
|
||||
interests = self._as_terms(data.get("interests")) or list(
|
||||
SEARCH_TERMS.get(archetype, [])
|
||||
)
|
||||
return {
|
||||
"archetype": archetype,
|
||||
"name": str(data.get("name", "")).strip()[:60],
|
||||
"backstory": str(data.get("backstory", "")).strip()[:300],
|
||||
"interests": interests,
|
||||
"dislikes": self._as_terms(data.get("dislikes")),
|
||||
"temperament": str(data.get("temperament", "")).strip()[:120],
|
||||
"verbosity": self._clamp01(data.get("verbosity"), 0.5),
|
||||
"contrarianness": self._clamp01(data.get("contrarianness"), 0.3),
|
||||
"generosity": self._clamp01(data.get("generosity"), 0.5),
|
||||
"curiosity": self._clamp01(data.get("curiosity"), 0.5),
|
||||
"rhythm": str(data.get("rhythm", "")).strip()[:120],
|
||||
}
|
||||
|
||||
def generate_identity(self, archetype: str) -> dict:
|
||||
interests = ", ".join(SEARCH_TERMS.get(archetype, []))
|
||||
system = (
|
||||
"You invent a unique, believable individual developer for a social coding "
|
||||
"platform, seeded from a broad archetype but distinct from anyone else who "
|
||||
"shares it. Return ONLY a JSON object, no prose and no markdown fences, with "
|
||||
"these keys: name (a short handle-like display name), backstory (one concrete "
|
||||
"sentence), interests (array of 4 to 6 short topics), dislikes (array of 2 to 4 "
|
||||
"short topics), temperament (a few words), verbosity (number 0.0 to 1.0), "
|
||||
"contrarianness (number 0.0 to 1.0), generosity (number 0.0 to 1.0), curiosity "
|
||||
"(number 0.0 to 1.0), rhythm (a few words on when and how they show up). Make it "
|
||||
"specific and memorable, never generic. No em dashes."
|
||||
)
|
||||
prompt = (
|
||||
f"Archetype: {archetype}\n"
|
||||
f"Typical interests for this archetype: {interests}\n"
|
||||
"Create the persona JSON:"
|
||||
)
|
||||
data = self._parse_json(self._raw_call(system, prompt, temperature=0.9))
|
||||
return self._normalize_identity(data, archetype)
|
||||
|
||||
@staticmethod
|
||||
def _identity_card(identity: dict) -> str:
|
||||
lines = [
|
||||
f"name: {identity.get('name', '')}",
|
||||
f"archetype: {identity.get('archetype', '')}",
|
||||
f"backstory: {identity.get('backstory', '')}",
|
||||
f"interests: {', '.join(identity.get('interests', []))}",
|
||||
f"dislikes: {', '.join(identity.get('dislikes', []))}",
|
||||
f"temperament: {identity.get('temperament', '')}",
|
||||
f"verbosity: {identity.get('verbosity', 0.5)}",
|
||||
f"contrarianness: {identity.get('contrarianness', 0.3)}",
|
||||
f"generosity: {identity.get('generosity', 0.5)}",
|
||||
f"curiosity: {identity.get('curiosity', 0.5)}",
|
||||
f"rhythm: {identity.get('rhythm', '')}",
|
||||
]
|
||||
return "\n".join(lines)
|
||||
|
||||
def _decision_system(self, identity: dict) -> str:
|
||||
return (
|
||||
"You are role-playing a single developer on a social coding platform. Stay in "
|
||||
"character at all times.\n\n"
|
||||
"IDENTITY\n"
|
||||
f"{self._identity_card(identity)}\n\n"
|
||||
"HOW TO DECIDE\n"
|
||||
"- Choose the actions that THIS person, with these interests and this "
|
||||
"temperament, would take on the page described. Behaviour must follow the "
|
||||
"identity, not chance.\n"
|
||||
"- A generous person votes and reacts readily; a contrarian one comments to push "
|
||||
"back; a terse, low-verbosity one acts rarely and briefly; a curious one explores "
|
||||
"and navigates. Let the numbers and interests drive the plan.\n"
|
||||
"- Only choose actions whose name appears in the MENU. Never invent an action or "
|
||||
"a target.\n"
|
||||
"- Order the plan the way this person would actually act, and include only what "
|
||||
"they would genuinely do (an empty plan is fine if nothing here interests them).\n"
|
||||
"- energy is how engaged this person is right now, 0.0 (barely present) to 1.0 "
|
||||
"(highly active), following their rhythm.\n"
|
||||
"- stop_after is roughly how many more actions this whole visit warrants before "
|
||||
"they would naturally drift away.\n"
|
||||
"- Output ONLY the JSON object below. No prose, no markdown fences.\n\n"
|
||||
"OUTPUT SCHEMA\n"
|
||||
'{"plan":[{"action":"<menu action>","target":"<id or empty>",'
|
||||
'"rationale":"<short>"}],"energy":<0.0-1.0>,"stop_after":<integer>}'
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _decision_prompt(page_state: dict, menu: list[dict], history: list[str]) -> str:
|
||||
menu_lines = "\n".join(f"- {item['action']}: {item['desc']}" for item in menu)
|
||||
recent = "; ".join(history[-8:]) if history else "nothing yet this visit"
|
||||
return (
|
||||
f"PAGE: {page_state.get('page', 'unknown')}\n"
|
||||
f"VISIBLE: {page_state.get('visible', '')}\n"
|
||||
f"MENU:\n{menu_lines}\n"
|
||||
f"RECENT (this visit): {recent}"
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _filter_plan(plan: Any, allowed: set) -> list[dict]:
|
||||
if not isinstance(plan, list):
|
||||
return []
|
||||
result = []
|
||||
for entry in plan:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
action = str(entry.get("action", "")).strip()
|
||||
if action not in allowed:
|
||||
continue
|
||||
result.append(
|
||||
{
|
||||
"action": action,
|
||||
"target": str(entry.get("target", "")).strip()[:120],
|
||||
"rationale": str(entry.get("rationale", "")).strip()[:200],
|
||||
}
|
||||
)
|
||||
return result[:12]
|
||||
|
||||
def decide(
|
||||
self,
|
||||
identity: dict,
|
||||
page_state: dict,
|
||||
menu: list[dict],
|
||||
history: list[str],
|
||||
temperature: float = 0.4,
|
||||
) -> dict:
|
||||
allowed = {item["action"] for item in menu}
|
||||
system = self._decision_system(identity)
|
||||
base = self._decision_prompt(page_state, menu, history)
|
||||
result = {"plan": [], "energy": 0.5, "stop_after": 0}
|
||||
for attempt in range(2):
|
||||
prompt = base
|
||||
if attempt:
|
||||
prompt += "\n\nReturn ONLY valid JSON matching the schema. No prose."
|
||||
try:
|
||||
data = self._parse_json(self._raw_call(system, prompt, temperature))
|
||||
except (ValueError, KeyError, json.JSONDecodeError):
|
||||
continue
|
||||
result = {
|
||||
"plan": self._filter_plan(data.get("plan"), allowed),
|
||||
"energy": self._clamp01(data.get("energy"), 0.5),
|
||||
"stop_after": self._as_int(data.get("stop_after"), 0),
|
||||
}
|
||||
if result["plan"]:
|
||||
break
|
||||
return result
|
||||
|
||||
def generate_bug(self, topic: str) -> tuple[str, str]:
|
||||
text = self._call(
|
||||
"Write a bug report. One line title. Then 2-3 sentences describing what happened and what should have happened. Like a real dev reporting a bug. No em dashes.",
|
||||
|
||||
Reference in New Issue
Block a user