Cameron Grant
047c1ac8ff
All checks were successful
continuous-integration/drone/push Build is passing
43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
import json, time
|
|
from typing import List, Dict, Any
|
|
from . import config
|
|
|
|
def append(role: str, content: str, ts: float | None = None):
|
|
rec = {"ts": ts if ts is not None else time.time(), "role": role, "content": content}
|
|
with config.HISTORY_FILE.open("a", encoding="utf-8") as f:
|
|
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
|
|
|
def load() -> List[Dict[str, Any]]:
|
|
if not config.HISTORY_FILE.exists(): return []
|
|
out = []
|
|
with config.HISTORY_FILE.open("r", encoding="utf-8") as f:
|
|
for line in f:
|
|
try: out.append(json.loads(line))
|
|
except Exception: pass
|
|
return out
|
|
|
|
def prune(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
if len(records) > config.MAX_MESSAGES_HISTORY:
|
|
records = records[-config.MAX_MESSAGES_HISTORY:]
|
|
total_tokens_est = sum(max(1, len(r.get("content","")) // 4) for r in records)
|
|
while total_tokens_est > config.MAX_TOKENS_HISTORY and len(records) > 10:
|
|
records = records[1:]
|
|
total_tokens_est = sum(max(1, len(r.get("content","")) // 4) for r in records)
|
|
return records
|
|
|
|
def last_record():
|
|
recs = load()
|
|
return recs[-1] if recs else None
|
|
|
|
def last_role():
|
|
r = last_record()
|
|
return r.get("role") if r else None
|
|
|
|
def build_messages(system_prompt: str):
|
|
recs = prune(load())
|
|
msgs = [{"role": "system", "content": system_prompt}]
|
|
for r in recs:
|
|
if r["role"] in ("user", "assistant"):
|
|
msgs.append({"role": r["role"], "content": r["content"]})
|
|
return msgs
|