diff --git a/services/logic/app/llm/providers.py b/services/logic/app/llm/providers.py index 1eb5f6a..53bf049 100644 --- a/services/logic/app/llm/providers.py +++ b/services/logic/app/llm/providers.py @@ -124,7 +124,14 @@ class _Driver: """(tekst, czy_ucięta, zużycie, nazwa_modelu, powód_zakończenia).""" raise NotImplementedError - def generate(self, prompt: str, max_tokens: int) -> Completion: + def generate(self, prompt: str, max_tokens: int, on_event=None) -> Completion: + """`on_event(dict)` dostaje zdarzenia postępu — UI pokazuje z nich log. + Raportujemy KAŻDĄ turę, bo to ona trwa; bez tego pasek postępu byłby + ozdobnikiem, a nie informacją.""" + def emit(kind: str, message: str, **extra): + if on_event: + on_event({"type": kind, "message": message, **extra}) + messages: list[dict] = [{"role": "user", "content": prompt}] parts: list[str] = [] usage: dict = {} @@ -137,9 +144,29 @@ class _Driver: while turns <= MAX_CONTINUATIONS: turns += 1 budget = max(256, min(remaining, TURN_TOKENS_CAP)) + emit("turn_start", + f"Tura {turns}: wysyłam do modelu {self.model} (limit {budget} tokenów)…", + turn=turns) + started = time.monotonic() text, truncated, turn_usage, model_name, stop = self._turn(messages, budget) + took = time.monotonic() - started _merge_usage(usage, turn_usage) - remaining -= budget + + # Odejmujemy tokeny FAKTYCZNIE wyprodukowane, nie zamówiony limit tury. + # Inaczej pierwsza tura zjadałaby cały budżet i urwana odpowiedź nigdy + # nie doczekałaby się kontynuacji — wracałby do użytkownika fragment + # udający całość. + produced = (turn_usage or {}).get("completion_tokens") + if produced is None: + produced = (turn_usage or {}).get("output_tokens") + if produced is None: + produced = max(1, int(len(text) / 3.6)) + remaining -= max(1, int(produced)) + + emit("turn_end", + f"Tura {turns}: odebrano {len(text.strip())} znaków w {took:.1f}s" + + (" — odpowiedź urwana, poproszę o dokończenie" if truncated else ""), + turn=turns, chars=len(text.strip()), truncated=truncated) chunk = text.strip() if chunk: @@ -172,6 +199,8 @@ class _Driver: final = _join(parts) if not final: raise LLMError(_explain_empty(turns, usage, stop)) + emit("generated", f"Gotowe: {len(final)} znaków w {turns} turach.", + chars=len(final), turns=turns) usage["turns"] = turns return Completion(text=final, model=model_name, provider=self.name, diff --git a/services/logic/app/main.py b/services/logic/app/main.py index 9f88a5a..04824cc 100644 --- a/services/logic/app/main.py +++ b/services/logic/app/main.py @@ -265,6 +265,72 @@ def chart_horoscope(req: HoroscopeRequest) -> dict: return out +@app.post("/chart/horoscope/stream") +def chart_horoscope_stream(req: HoroscopeRequest): + """To samo co /chart/horoscope, ale strumieniuje POSTĘP w trakcie pracy. + + Pisanie horoskopu trwa minutami — bez sygnału aplikacja wygląda na zawieszoną. + Strumień (NDJSON, jedna linia = jedno zdarzenie) niesie RZECZYWISTE etapy: + budowę promptu, limity modelu i każdą turę generowania. Ostatnie zdarzenie + (`result`) ma identyczny kształt co odpowiedź zwykłego endpointu. + """ + from fastapi.responses import StreamingResponse + + from app.progress import stream + + def work(emit) -> dict: + from app.llm.base import LLMError + from app.llm.factory import build_provider + from app.llm.limits import plan + + emit({"type": "stage", "message": "Liczę horoskop i szukam wskazań w bazach…"}) + out = chart_prompt(req) + st = out.get("stats", {}) + emit({"type": "stage", "message": + f"Prompt gotowy: {st.get('chars', 0)} znaków, " + f"wskazań {st.get('included', 0)}" + + (f", pominięto {st['omitted']}" if st.get("omitted") else "")}) + if out.get("data_error"): + emit({"type": "warn", "message": out["data_error"]}) + + try: + provider = build_provider(req.provider, req.model) + emit({"type": "stage", "message": + f"Dostawca: {provider.name}, model: {provider.model}" + + ("" if not provider.leaves_lan else " — dane opuszczają sieć")}) + + emit({"type": "stage", "message": "Liczę tokeny promptu…"}) + prompt_tokens = provider.count_tokens(out["prompt"]) + budget = plan(provider.name, provider.model, prompt_tokens, req.max_tokens) + out["token_plan"] = budget + emit({"type": "stage", "message": + f"Prompt {prompt_tokens} tok. · okno modelu {budget['context_window']} · " + f"na odpowiedź {budget['max_output']}"}) + for warning in budget["warnings"]: + emit({"type": "warn", "message": warning}) + if budget["warnings"]: + out["warnings"] = budget["warnings"] + if not budget["fits"]: + out["llm_error"] = " ".join(budget["warnings"]) + return out + + result = provider.generate(out["prompt"], budget["max_output"], on_event=emit) + except LLMError as e: + emit({"type": "warn", "message": f"Model zawiódł: {e}"}) + out["llm_error"] = str(e) + return out + + out.update(horoscope=result.text, provider=result.provider, model=result.model, + leaves_lan=result.leaves_lan, usage=result.usage) + return out + + return StreamingResponse( + stream(work), + media_type="application/x-ndjson", + headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}, + ) + + @app.get("/llm/models") def llm_models() -> dict: """Podpowiedzi modeli per dostawca — UI buduje z tego listę wyboru. diff --git a/services/logic/app/progress.py b/services/logic/app/progress.py new file mode 100644 index 0000000..cc4d16e --- /dev/null +++ b/services/logic/app/progress.py @@ -0,0 +1,70 @@ +"""Strumień postępu długiej operacji (NDJSON). + +Po co: pisanie horoskopu trwa — czasem minuty. Bez sygnału aplikacja wygląda na +zawieszoną. Zamiast udawanego paska postępu strumieniujemy **rzeczywiste** +zdarzenia z kolejnych etapów, żeby log pokazywał to, co faktycznie się dzieje. + +Dlaczego NDJSON, a nie SSE: `EventSource` w przeglądarce obsługuje wyłącznie GET, +a to jest POST z ciałem. Strumień „jedna linia = jeden obiekt JSON" czyta się +zwykłym `fetch()` i jest trywialny do sparsowania. + +Dlaczego wątek: właściwa praca (silnik, baza, model) jest synchroniczna. Puszczamy +ją w wątku roboczym, a generator odpompowuje kolejkę zdarzeń — dzięki temu +zdarzenia docierają w trakcie pracy, a nie dopiero na końcu. +""" +from __future__ import annotations + +import json +import queue +import threading +import traceback +from collections.abc import Iterator +from typing import Any, Callable + +_HEARTBEAT_SECONDS = 10.0 +_DONE = object() + + +def line(kind: str, message: str, **extra: Any) -> str: + return json.dumps({"type": kind, "message": message, **extra}, ensure_ascii=False) + "\n" + + +def stream(work: Callable[[Callable[[dict], None]], dict]) -> Iterator[str]: + """Uruchamia `work(emit)` w wątku i strumieniuje zdarzenia w czasie rzeczywistym. + + `work` dostaje funkcję `emit(zdarzenie)` i zwraca końcowy wynik, który leci + jako ostatnie zdarzenie typu `result`. Wyjątek zamienia się w zdarzenie `error` + — połączenie nigdy nie urywa się bez wyjaśnienia. + """ + events: queue.Queue = queue.Queue() + + def emit(event: dict) -> None: + events.put(event) + + def run() -> None: + try: + result = work(emit) + events.put({"type": "result", "message": "Gotowe.", "result": result}) + except Exception as e: # noqa: BLE001 — zgłaszamy KAŻDY błąd + events.put({ + "type": "error", + "message": f"{type(e).__name__}: {e}", + "detail": traceback.format_exc(limit=3), + }) + finally: + events.put(_DONE) + + worker = threading.Thread(target=run, daemon=True) + worker.start() + + while True: + try: + event = events.get(timeout=_HEARTBEAT_SECONDS) + except queue.Empty: + # cisza dłuższa niż heartbeat: dajemy znak życia, żeby pośredniki + # (proxy, load balancer) nie uznały połączenia za martwe + yield line("ping", "…") + continue + if event is _DONE: + break + yield json.dumps(event, ensure_ascii=False) + "\n" diff --git a/services/logic/tests/test_llm.py b/services/logic/tests/test_llm.py index 25659b5..1041ca5 100644 --- a/services/logic/tests/test_llm.py +++ b/services/logic/tests/test_llm.py @@ -405,3 +405,33 @@ def test_max_budget_differs_between_models(monkeypatch): haiku = prompt_token_budget(*factory.resolve_model("anthropic", "claude-haiku-4-5")) local = prompt_token_budget(*factory.resolve_model("local", "llama3.1:8b")) assert opus > haiku > local > 0 + + +def test_turn_budget_counts_produced_not_requested(monkeypatch): + """Regresja: odejmowanie ZAMOWIONEGO limitu tury zamiast wyprodukowanych + tokenow konczylo petle po jednej turze — urwany fragment wracal jako calosc.""" + seq = [_chat("Fragment 1. ", "length"), _chat("Fragment 2. ", "length"), + _chat("Zakonczenie. KONIEC", "stop")] + + def handler(request): + return httpx.Response(200, json=seq.pop(0) if seq else seq[-1]) + + monkeypatch.setattr(httpx, "Client", _mock_client(handler)) + # budzet 8000 < TURN_TOKENS_CAP: przy starej logice byla dokladnie jedna tura + out = ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 8000) + assert out.usage["turns"] == 3, "urwana odpowiedz musi byc kontynuowana" + assert "Fragment 1." in out.text and "Zakonczenie." in out.text + + +def test_generate_reports_progress_events(monkeypatch): + """Log w UI ma pokazywac RZECZYWISTE tury, nie udawany pasek postepu.""" + seq = [_chat("Czesc. ", "length"), _chat("Reszta. KONIEC", "stop")] + monkeypatch.setattr(httpx, "Client", _mock_client( + lambda r: httpx.Response(200, json=seq.pop(0) if seq else seq[-1]))) + events = [] + ChatCompletionsProvider("local", "http://x/v1", "m").generate( + "p", 8000, on_event=events.append) + kinds = [e["type"] for e in events] + assert kinds.count("turn_start") == 2 and kinds.count("turn_end") == 2 + assert kinds[-1] == "generated" + assert any("urwana" in e["message"] for e in events if e["type"] == "turn_end") diff --git a/services/presentation/app/clients/logic_client.py b/services/presentation/app/clients/logic_client.py index 7df6731..9e5ed1e 100644 --- a/services/presentation/app/clients/logic_client.py +++ b/services/presentation/app/clients/logic_client.py @@ -111,6 +111,20 @@ class LogicClient: r.raise_for_status() return r.json() + def horoscope_stream(self, payload: dict[str, Any]): + """Strumień postępu pisania horoskopu (NDJSON) — przekazywany do przeglądarki. + + Timeout jest długi, bo generowanie trwa; strumień i tak niesie heartbeat, + więc cisza na łączu nie zostanie wzięta za zerwanie. + """ + with httpx.Client(timeout=httpx.Timeout(None, connect=15.0)) as client: + with client.stream("POST", f"{self.base_url}/chart/horoscope/stream", + json=payload, headers=_auth_headers()) as r: + r.raise_for_status() + for chunk in r.iter_lines(): + if chunk: + yield chunk + def llm_models(self) -> dict[str, Any]: """Katalog modeli per dostawca (podpowiedzi do pola wyboru w UI).""" with httpx.Client(timeout=settings.http_timeout) as client: diff --git a/services/presentation/app/main.py b/services/presentation/app/main.py index 182b548..3243491 100644 --- a/services/presentation/app/main.py +++ b/services/presentation/app/main.py @@ -9,11 +9,12 @@ Strona główna „/" = wprowadzenie danych horoskopu i podgląd policzonych poz """ from __future__ import annotations +import json from datetime import datetime, timedelta, timezone import httpx from fastapi import FastAPI, Form, HTTPException, Query, Request -from fastapi.responses import HTMLResponse +from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates @@ -228,6 +229,61 @@ def timeline_run( return templates.TemplateResponse(request, "timeline.html", ctx) +# ---------------- Postęp pisania horoskopu (strumień do okna z logiem) ---------------- +@app.post("/horoscope/stream") +def horoscope_stream( + profile: str = Form("natal"), + date: str = Form(...), + time: str = Form(...), + tz_offset: float = Form(0.0), + lat: float = Form(0.0), + lon: float = Form(0.0), + prompt_budget: str = Form("medium"), + llm_provider: str = Form("local"), + llm_model: str = Form(""), + from_date: str = Form(""), + to_date: str = Form(""), +): + """Przekazuje strumień postępu z logiki i DOKLEJA gotowy HTML wyniku. + + Dzięki temu okno postępu wstawia dokładnie ten sam widok, który wyrenderowałoby + przeładowanie strony — jedno źródło prawdy dla wyglądu wyniku. + """ + from fastapi.responses import StreamingResponse + + try: + iso_utc, _ = _build_utc(date, time, tz_offset) + except ValueError as e: + return JSONResponse({"detail": f"Niepoprawne dane wejściowe: {e}"}, status_code=422) + + payload: dict = { + "profile": profile, "when_utc": iso_utc, "lat": lat, "lon": lon, + "budget": prompt_budget, "provider": llm_provider, "model": llm_model, + } + if profile == "period" and from_date and to_date: + payload["from_date"], payload["to_date"] = from_date, to_date + + def relay(): + try: + for raw in logic.horoscope_stream(payload): + try: + event = json.loads(raw) + except ValueError: + continue + if event.get("type") == "result": + html = templates.get_template("_prompt_result.html").render( + prompt_result=event.get("result") or {} + ) + event["html"] = html + yield json.dumps(event, ensure_ascii=False) + "\n" + except httpx.HTTPError as e: + yield json.dumps({"type": "error", "message": _logic_error(e)}, + ensure_ascii=False) + "\n" + + return StreamingResponse(relay(), media_type="application/x-ndjson", + headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}) + + # ---------------- Geokoder (proxy OSM/Nominatim dla wyszukiwarki lokalizacji) ---------------- @app.get("/geocode") def geocode_search(q: str = Query("", description="Nazwa / adres / POI do wyszukania")): diff --git a/services/presentation/app/static/progress.js b/services/presentation/app/static/progress.js new file mode 100644 index 0000000..197a55a --- /dev/null +++ b/services/presentation/app/static/progress.js @@ -0,0 +1,133 @@ +// Okno postępu przy pisaniu horoskopu. +// +// Problem: generowanie trwa minutami, a zwykły POST formularza nie daje żadnego +// sygnału — aplikacja wygląda na zawieszoną. Zamiast udawanego paska postępu +// czytamy strumień RZECZYWISTYCH zdarzeń z serwera (NDJSON) i wypisujemy je +// jako log: budowa promptu, limity modelu, każda tura generowania. +// +// Degradacja: jeśli przeglądarka nie umie strumieniować `fetch`, nie przechwytujemy +// wysyłki — formularz idzie klasycznie i wszystko działa jak wcześniej, tylko bez okna. +document.addEventListener('DOMContentLoaded', function () { + const canStream = typeof fetch === 'function' && typeof ReadableStream === 'function' && + typeof TextDecoder === 'function'; + const form = document.querySelector('form[action="/interpret"], form[action="/timeline"]'); + if (!canStream || !form) return; + + const profile = form.getAttribute('action') === '/timeline' ? 'period' : 'natal'; + const btn = form.querySelector('button[value="horoscope"]'); + if (!btn) return; + + // --- okno --------------------------------------------------------------- + const overlay = document.createElement('div'); + overlay.className = 'progress-overlay'; + overlay.hidden = true; + overlay.innerHTML = + '
Nie zamykaj tej karty — generowanie trwa na serwerze.
' + + '' + + '- Treść wygenerował model językowy na podstawie {{ st.included }} wskazań z baz. - Nie stanowi porady medycznej, prawnej ani finansowej. -
- {% endif %} - - {% if prompt_result.warnings %} - {% for w in prompt_result.warnings %} -Uwaga: {{ w }}
- {% endfor %} - {% endif %} - - {% if prompt_result.token_plan %} - {% set tp = prompt_result.token_plan %} -- Tokeny: prompt {{ tp.prompt_tokens }} · okno modelu {{ tp.context_window }} · - zarezerwowane na odpowiedź {{ tp.max_output }} - {% if prompt_result.usage and prompt_result.usage.turns and prompt_result.usage.turns > 1 %} - · odpowiedź złożona z {{ prompt_result.usage.turns }} tur (model dokańczał urwany tekst) - {% endif %} -
- {% endif %} - - {% if prompt_result.llm_error %} -- Pominięto {{ st.omitted }} najsłabszych wskazań (próg wagi {{ st.min_score_included }}). - Chcesz komplet — wybierz obszerniejszy budżet i wygeneruj ponownie. -
- {% endif %} - - {% if prompt_result.data_error %} -+ Treść wygenerował model językowy na podstawie {{ st.included }} wskazań z baz. + Nie stanowi porady medycznej, prawnej ani finansowej. +
+ {% endif %} + + {% if prompt_result.warnings %} + {% for w in prompt_result.warnings %} +Uwaga: {{ w }}
+ {% endfor %} + {% endif %} + + {% if prompt_result.token_plan %} + {% set tp = prompt_result.token_plan %} ++ Tokeny: prompt {{ tp.prompt_tokens }} · okno modelu {{ tp.context_window }} · + zarezerwowane na odpowiedź {{ tp.max_output }} + {% if prompt_result.usage and prompt_result.usage.turns and prompt_result.usage.turns > 1 %} + · odpowiedź złożona z {{ prompt_result.usage.turns }} tur (model dokańczał urwany tekst) + {% endif %} +
+ {% endif %} + + {% if prompt_result.llm_error %} ++ Pominięto {{ st.omitted }} najsłabszych wskazań (próg wagi {{ st.min_score_included }}). + Chcesz komplet — wybierz obszerniejszy budżet i wygeneruj ponownie. +
+ {% endif %} + + {% if prompt_result.data_error %} +