feat(logic): generator promptow do LLM + budzetowanie (LOG-29, LOG-30)
Pierwszy krok ustalonej kolejnosci: prompt z podgladem, uzyteczny od razu BEZ integracji API (ta przyjdzie w LOG-31). LOG-29 — app/prompt.py: - profil natal (ekran Interpretacje) i period (ekran Kalendarz), - prompt po polsku: zadanie -> dane horoskopu (pozycje, osie, Lots, aspekty, sekta, zodiak) -> wskazania z baz wg wagi -> instrukcje -> zastrzezenie, - twarde reguly: kazda teza musi cytowac konkretny sygnifikator, zakaz wychodzenia poza dostarczone dane, jawne wskazanie sprzecznosci, - deterministyczny: ten sam horoskop + budzet = ten sam prompt. LOG-30 — redukcja do budzetu (concise/medium/extensive): dedup -> grupowanie z licznikiem -> sortowanie wg punktacji sily (LOG-21) -> obciecie ogona (jednostka = CALE wskazanie) -> skracanie dlugich opisow. Statystyki zwracaja ile weszlo/pominieto i jaki byl prog — takze w tresci promptu, zeby model wiedzial, ze widzi wybor. POST /chart/prompt — zwraca sam prompt + statystyki, bez wolania modelu. Prezentacja: przycisk „Generuj prompt (AI)" na obu ekranach, wybor budzetu, pole z promptem + kopiowanie (z fallbackiem dla http bez secure context). WAZNE (znalezione przy tescie e2e): padnieta warstwa danych zabiera tylko wskazania — horoskop i OS CZASU zostaja, bo sa czysto obliczeniowe. Wczesniej blad bazy gubil cala osie czasu, czyniac prognoze okresowa bezuzyteczna. Zabezpieczone testem. Testy: 19 nowych, calosc 118 passed / 1 skipped. Zweryfikowane e2e w przegladarce: profil natal (3340 znakow) i period (15 zdarzen, 4728 znakow). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit was merged in pull request #9.
This commit is contained in:
@@ -0,0 +1,339 @@
|
||||
"""Generator promptów do LLM (LOG-29) + budżetowanie rozmiaru (LOG-30).
|
||||
|
||||
Składa z naszych wyliczeń gotowe, profesjonalne polecenie po polsku:
|
||||
- profil **natal** — horoskop urodzeniowy (na bazie /chart/report),
|
||||
- profil **period** — horoskop na wybrany okres (na bazie /chart/timeline).
|
||||
|
||||
Zasada naczelna promptu: model ma pisać WYŁĄCZNIE na podstawie dostarczonych danych
|
||||
i przy każdej tezie wskazać konkretny sygnifikator, z którego ona wynika. To odróżnia
|
||||
wynik od ogólnikowej wróżby i pozwala go zweryfikować.
|
||||
|
||||
Budżetowanie (LOG-30) — kolejność redukcji:
|
||||
1. deduplikacja (ten sam sygnifikator ORAZ ten sam opis),
|
||||
2. grupowanie identycznych opisów z licznikiem wystąpień,
|
||||
3. sortowanie malejąco wg punktacji siły (LOG-21),
|
||||
4. obcięcie ogona do budżetu — jednostką obcięcia jest CAŁE wskazanie,
|
||||
5. skracanie nadmiernie długich opisów (z jawnym oznaczeniem).
|
||||
Zawsze raportujemy, ile wskazań weszło i ile pominięto — użytkownik ma wiedzieć,
|
||||
czego brakuje, i móc zwiększyć budżet.
|
||||
|
||||
Generowanie jest deterministyczne: ten sam horoskop + ten sam budżet = ten sam prompt.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
# budżety w znakach (całego promptu); dobrane pod wklejanie do ChatGPT/Claude
|
||||
BUDGETS: dict[str, int] = {
|
||||
"concise": 4_000,
|
||||
"medium": 12_000,
|
||||
"extensive": 30_000,
|
||||
}
|
||||
DEFAULT_BUDGET = "medium"
|
||||
|
||||
MAX_EFFECT_CHARS = 320 # dłuższe opisy skracamy (krok 5 redukcji)
|
||||
CHARS_PER_TOKEN = 4.0 # zgrubny szacunek tokenów do podglądu w UI
|
||||
|
||||
DISCLAIMER = (
|
||||
"Zakończ krótką notą: treść jest interpretacją astrologiczną i nie stanowi porady "
|
||||
"medycznej, prawnej ani finansowej."
|
||||
)
|
||||
|
||||
|
||||
def _norm(text: str) -> str:
|
||||
return re.sub(r"\s+", " ", (text or "").strip().lower())
|
||||
|
||||
|
||||
def est_tokens(text: str) -> int:
|
||||
return int(len(text) / CHARS_PER_TOKEN)
|
||||
|
||||
|
||||
def _shorten(text: str, limit: int = MAX_EFFECT_CHARS) -> tuple[str, bool]:
|
||||
text = (text or "").strip()
|
||||
if len(text) <= limit:
|
||||
return text, False
|
||||
return text[:limit].rstrip() + " […skrócono]", True
|
||||
|
||||
|
||||
# ----------------------------------------------------------------- wskazania
|
||||
|
||||
def natal_indications(report: dict) -> list[dict]:
|
||||
"""Płaska lista wskazań z raportu natalnego: obiekt + faseta + opis + waga."""
|
||||
out: list[dict] = []
|
||||
for obj in report.get("objects") or []:
|
||||
name = obj.get("object")
|
||||
for facet in obj.get("facets") or []:
|
||||
score = float(facet.get("score") or 0.0)
|
||||
for s in facet.get("samples") or []:
|
||||
out.append({
|
||||
"context": f"{name} — {facet.get('label')}",
|
||||
"sort_key": (name or "", str(facet.get("label") or "")),
|
||||
"score": score,
|
||||
"significator": s.get("expanded") or s.get("significator") or "",
|
||||
"effect": s.get("effect") or "",
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def period_indications(events: list[dict]) -> list[dict]:
|
||||
"""Wskazania z osi czasu — waga rośnie z liczbą trafień w bazie."""
|
||||
out: list[dict] = []
|
||||
for ev in events or []:
|
||||
head = f"{ev.get('technique')} · {ev.get('significator')} · {ev.get('exact')}"
|
||||
total = float(ev.get("interpretations_count") or 0)
|
||||
for s in ev.get("interpretations") or []:
|
||||
out.append({
|
||||
"context": head,
|
||||
"sort_key": (str(ev.get("exact") or ""), str(ev.get("technique") or "")),
|
||||
"score": total,
|
||||
"significator": s.get("expanded") or s.get("significator") or "",
|
||||
"effect": s.get("effect") or "",
|
||||
"date": ev.get("exact"),
|
||||
"start": ev.get("start"),
|
||||
"end": ev.get("end"),
|
||||
"technique": ev.get("technique"),
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def reduce_indications(items: list[dict], budget_chars: int) -> tuple[list[dict], dict]:
|
||||
"""Dedup → grupowanie → sortowanie wg wagi → obcięcie do budżetu.
|
||||
|
||||
Zwraca (wybrane wskazania, statystyki). Jednostką obcięcia jest całe wskazanie.
|
||||
"""
|
||||
# 1+2: dedup i grupowanie identycznych (kontekst, opis) z licznikiem
|
||||
grouped: dict[tuple[str, str], dict] = {}
|
||||
order: list[tuple[str, str]] = []
|
||||
for it in items:
|
||||
key = (_norm(it["context"]), _norm(it["effect"]))
|
||||
g = grouped.get(key)
|
||||
if g is None:
|
||||
g = {**it, "count": 0, "significators": []}
|
||||
grouped[key] = g
|
||||
order.append(key)
|
||||
g["count"] += 1
|
||||
if it["significator"] not in g["significators"]:
|
||||
g["significators"].append(it["significator"])
|
||||
merged = [grouped[k] for k in order]
|
||||
deduped = len(items) - len(merged)
|
||||
|
||||
# 3: sortowanie malejąco wg wagi; remis rozstrzygany deterministycznie
|
||||
merged.sort(key=lambda g: (-g["score"], -g["count"], g["sort_key"], g["significator"]))
|
||||
|
||||
# 5 (przed obcięciem, bo wpływa na rozmiar): skracanie długich opisów
|
||||
shortened = 0
|
||||
for g in merged:
|
||||
g["effect"], was = _shorten(g["effect"])
|
||||
shortened += 1 if was else 0
|
||||
|
||||
# 4: obcięcie ogona do budżetu — cała pozycja albo nic
|
||||
chosen: list[dict] = []
|
||||
used = 0
|
||||
for g in merged:
|
||||
cost = len(_render_indication(g)) + 1
|
||||
if used + cost > budget_chars and chosen:
|
||||
break
|
||||
chosen.append(g)
|
||||
used += cost
|
||||
|
||||
omitted = len(merged) - len(chosen)
|
||||
stats = {
|
||||
"source_rows": len(items),
|
||||
"after_grouping": len(merged),
|
||||
"deduplicated": deduped,
|
||||
"included": len(chosen),
|
||||
"omitted": omitted,
|
||||
"shortened": shortened,
|
||||
"min_score_included": round(chosen[-1]["score"], 3) if chosen else None,
|
||||
"max_score_omitted": round(merged[len(chosen)]["score"], 3) if omitted else None,
|
||||
}
|
||||
return chosen, stats
|
||||
|
||||
|
||||
def _render_indication(g: dict) -> str:
|
||||
times = f" (×{g['count']})" if g.get("count", 1) > 1 else ""
|
||||
sig = g.get("significator") or ""
|
||||
return f"- {sig} → {g.get('effect')}{times}"
|
||||
|
||||
|
||||
def _render_grouped(chosen: list[dict]) -> str:
|
||||
"""Renderuje wskazania pogrupowane po kontekście, zachowując kolejność wagi."""
|
||||
blocks: list[str] = []
|
||||
seen: dict[str, list[dict]] = {}
|
||||
order: list[str] = []
|
||||
for g in chosen:
|
||||
ctx = g["context"]
|
||||
if ctx not in seen:
|
||||
seen[ctx] = []
|
||||
order.append(ctx)
|
||||
seen[ctx].append(g)
|
||||
for ctx in order:
|
||||
rows = seen[ctx]
|
||||
blocks.append(f"## {ctx} [waga {rows[0]['score']:.2f}]")
|
||||
blocks.extend(_render_indication(g) for g in rows)
|
||||
blocks.append("")
|
||||
return "\n".join(blocks).rstrip()
|
||||
|
||||
|
||||
# ------------------------------------------------------------- sekcje danych
|
||||
|
||||
def _chart_section(chart: dict, moment_label: str | None) -> str:
|
||||
lines: list[str] = ["# DANE HOROSKOPU"]
|
||||
meta = []
|
||||
if moment_label:
|
||||
meta.append(f"moment: {moment_label}")
|
||||
if chart.get("zodiac"):
|
||||
z = chart["zodiac"]
|
||||
if chart.get("ayanamsha") is not None:
|
||||
z += f" (ayanamsa {chart['ayanamsha']:.4f}°)"
|
||||
meta.append(f"zodiak: {z}")
|
||||
if chart.get("house_system"):
|
||||
meta.append(f"system domów: {chart['house_system']}")
|
||||
if chart.get("sect"):
|
||||
meta.append(f"sekta: {'dzienna' if chart['sect'] == 'day' else 'nocna'}")
|
||||
if meta:
|
||||
lines.append(" · ".join(meta))
|
||||
|
||||
if chart.get("positions"):
|
||||
lines.append("\n## Pozycje")
|
||||
for p in chart["positions"]:
|
||||
house = f"dom {p['house']}" if p.get("house") else "—"
|
||||
lines.append(f"{p['name']:<12} {p.get('in_sign',''):<16} {house:<8} {p.get('direction','')}")
|
||||
|
||||
angles = chart.get("angles") or {}
|
||||
if angles:
|
||||
lines.append("\n## Osie")
|
||||
for key in ("Asc", "MC", "Dsc", "IC"):
|
||||
a = angles.get(key)
|
||||
if a:
|
||||
lines.append(f"{a['name']:<5} {a.get('in_sign','')}")
|
||||
|
||||
if chart.get("lots"):
|
||||
lines.append("\n## Lots (punkty arabskie)")
|
||||
for lot in chart["lots"]:
|
||||
lines.append(
|
||||
f"{lot['name']:<10} {lot.get('in_sign',''):<16} dom {lot.get('house','—')}"
|
||||
f" ({lot.get('formula','')})"
|
||||
)
|
||||
|
||||
if chart.get("aspects"):
|
||||
lines.append("\n## Aspekty (orb; A = aplikacyjny, S = separacyjny)")
|
||||
for a in chart["aspects"]:
|
||||
mark = a.get("as") or ""
|
||||
lines.append(
|
||||
f"{a['obj1']} {a['aspect']} {a['obj2']} orb {a.get('orb', 0):.2f}° {mark}"
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _indications_section(chosen: list[dict], stats: dict) -> str:
|
||||
if not chosen:
|
||||
return (
|
||||
"# WSKAZANIA Z BAZ\n"
|
||||
"(brak trafień w bazach dla tego horoskopu — oprzyj interpretację wyłącznie "
|
||||
"na danych horoskopu powyżej)"
|
||||
)
|
||||
head = [
|
||||
"# WSKAZANIA Z BAZ INTERPRETACYJNYCH",
|
||||
"Wskazania dopasowane do tego horoskopu, uporządkowane od najsilniejszych.",
|
||||
"Zapis: `- sygnifikator → opis (×ile razy wystąpiło w bazach)`.",
|
||||
]
|
||||
if stats.get("omitted"):
|
||||
head.append(
|
||||
f"UWAGA: pokazano {stats['included']} najsilniejszych wskazań, pominięto "
|
||||
f"{stats['omitted']} słabszych (limit długości)."
|
||||
)
|
||||
return "\n".join(head) + "\n\n" + _render_grouped(chosen)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ prompty
|
||||
|
||||
_NATAL_TASK = """# ZADANIE
|
||||
Jesteś doświadczonym astrologiem. Napisz profesjonalną interpretację HOROSKOPU
|
||||
URODZENIOWEGO po polsku, opierając się wyłącznie na danych podanych niżej."""
|
||||
|
||||
_PERIOD_TASK = """# ZADANIE
|
||||
Jesteś doświadczonym astrologiem. Napisz profesjonalną prognozę astrologiczną
|
||||
NA WYBRANY OKRES po polsku, opierając się wyłącznie na danych podanych niżej."""
|
||||
|
||||
_NATAL_OUTPUT = """# JAK MA WYGLĄDAĆ ODPOWIEDŹ
|
||||
1. Struktura: (a) portret ogólny, (b) temperament i sekta, (c) obszary życia według domów,
|
||||
(d) napięcia i wyzwania, (e) zasoby i mocne strony, (f) zwięzłe podsumowanie.
|
||||
2. KAŻDĄ tezę oprzyj na konkretnym wskazaniu i podaj je w nawiasie, np. „(Moon in 12th house)".
|
||||
Teza bez wskazania jest niedopuszczalna.
|
||||
3. Nie dodawaj twierdzeń, których nie da się wywieść z powyższych danych. Nie zmyślaj
|
||||
pozycji, aspektów ani wskazań; nie korzystaj z wiedzy spoza tego promptu.
|
||||
4. Gdy wskazania są sprzeczne, powiedz to wprost i wskaż obie strony, zamiast wybierać jedną.
|
||||
5. Wagę wskazania traktuj jako siłę świadectwa — mocniejsze mają pierwszeństwo w syntezie.
|
||||
6. Ton rzeczowy i profesjonalny: bez wróżbiarstwa, bez straszenia, bez diagnoz medycznych."""
|
||||
|
||||
_PERIOD_OUTPUT = """# JAK MA WYGLĄDAĆ ODPOWIEDŹ
|
||||
1. Uporządkuj prognozę CHRONOLOGICZNIE; przy każdym okresie podaj daty (start / dokładna / koniec).
|
||||
2. Dla każdej daty napisz, która technika ją wyznacza (profekcja, solariusz, dyrekcja solar-arc,
|
||||
Firdaria) i co z niej wynika.
|
||||
3. KAŻDĄ tezę oprzyj na konkretnym wskazaniu i podaj je w nawiasie. Teza bez wskazania jest
|
||||
niedopuszczalna.
|
||||
4. Nie zmyślaj dat ani zdarzeń spoza podanych. Nie korzystaj z wiedzy spoza tego promptu.
|
||||
5. Rozróżniaj okresy o mocnym świadectwie (wysoka waga, kilka technik zbieżnych w czasie)
|
||||
od słabych — i powiedz wprost, które są które.
|
||||
6. Na końcu dodaj krótkie zestawienie: najważniejsze okresy w kolejności ważności.
|
||||
7. Ton rzeczowy i profesjonalny: bez wróżbiarstwa, bez straszenia, bez diagnoz medycznych."""
|
||||
|
||||
|
||||
def _assemble(task: str, chart_sec: str, ind_sec: str, output: str) -> str:
|
||||
return "\n\n".join([task, chart_sec, ind_sec, output, f"# ZASTRZEŻENIE\n{DISCLAIMER}"])
|
||||
|
||||
|
||||
def _budget_chars(budget: str) -> int:
|
||||
if budget not in BUDGETS:
|
||||
raise ValueError(
|
||||
f"Nieznany budżet: {budget!r} (dostępne: {', '.join(BUDGETS)})"
|
||||
)
|
||||
return BUDGETS[budget]
|
||||
|
||||
|
||||
def _finish(prompt: str, budget: str, limit: int, stats: dict, profile: str) -> dict:
|
||||
stats = {
|
||||
**stats,
|
||||
"profile": profile,
|
||||
"budget": budget,
|
||||
"limit_chars": limit,
|
||||
"chars": len(prompt),
|
||||
"est_tokens": est_tokens(prompt),
|
||||
}
|
||||
return {"profile": profile, "prompt": prompt, "stats": stats}
|
||||
|
||||
|
||||
def build_natal_prompt(chart: dict, report: dict, budget: str = DEFAULT_BUDGET,
|
||||
moment_label: str | None = None) -> dict:
|
||||
"""Prompt na horoskop urodzeniowy (ekran „Interpretacje")."""
|
||||
limit = _budget_chars(budget)
|
||||
chart_sec = _chart_section(chart, moment_label)
|
||||
fixed = len(_NATAL_TASK) + len(chart_sec) + len(_NATAL_OUTPUT) + len(DISCLAIMER) + 200
|
||||
chosen, stats = reduce_indications(natal_indications(report), max(limit - fixed, 500))
|
||||
prompt = _assemble(_NATAL_TASK, chart_sec, _indications_section(chosen, stats), _NATAL_OUTPUT)
|
||||
return _finish(prompt, budget, limit, stats, "natal")
|
||||
|
||||
|
||||
def build_period_prompt(chart: dict, events: list[dict], from_date: str, to_date: str,
|
||||
budget: str = DEFAULT_BUDGET, moment_label: str | None = None) -> dict:
|
||||
"""Prompt na horoskop okresowy (ekran „Kalendarz")."""
|
||||
limit = _budget_chars(budget)
|
||||
chart_sec = _chart_section(chart, moment_label)
|
||||
task = f"{_PERIOD_TASK}\nZakres prognozy: **{from_date} — {to_date}**."
|
||||
|
||||
ev_lines = ["# OŚ CZASU (techniki predykcyjne)",
|
||||
"Zapis: technika · sygnifikator · start → dokładna → koniec."]
|
||||
for ev in events or []:
|
||||
ev_lines.append(
|
||||
f"- {ev.get('technique')} · {ev.get('significator')} · "
|
||||
f"{ev.get('start')} → {ev.get('exact')} → {ev.get('end')}"
|
||||
)
|
||||
events_sec = "\n".join(ev_lines)
|
||||
|
||||
fixed = len(task) + len(chart_sec) + len(events_sec) + len(_PERIOD_OUTPUT) + len(DISCLAIMER) + 200
|
||||
chosen, stats = reduce_indications(period_indications(events), max(limit - fixed, 500))
|
||||
body = events_sec + "\n\n" + _indications_section(chosen, stats)
|
||||
prompt = _assemble(task, chart_sec, body, _PERIOD_OUTPUT)
|
||||
stats["events"] = len(events or [])
|
||||
return _finish(prompt, budget, limit, stats, "period")
|
||||
Reference in New Issue
Block a user