diff --git a/NOTES.md b/NOTES.md index 1b2daf9..0e453e3 100644 --- a/NOTES.md +++ b/NOTES.md @@ -91,6 +91,9 @@ Pipeline `print()` is teed for the job thread ([backend/job_log.py](backend/job_ - UI: “Run log” panel updates while running; stays visible after finish/fail/review - Failures append the **full traceback** to the log; each run starts with a header line (job id, mode, models, start time) - `outputs//job.json` (written at start) carries email/mode/models so the disk fallback can rebuild a job after restart +- **Verbose LLM observability** (`LLM_VERBOSE=true`, default on): one `[LLM] #NNNN stage | model (backend) | in/out sizes | cost | parsed-item counts` line per call in `job.log` — list-valued keys show counts (`conflicts[3]`) so empty (miss) or invented (hallucination) results stand out +- **Raw dumps** (`LLM_RAW_DUMP=true`, default on): full prompt + raw response per call in `outputs//llm_raw/NNNN_stage_model.json` (base64 images excluded, `n_images` recorded). This is the source for tracing *why* the model missed/invented an item +- **End-of-log cost**: every run ends with an `=== Estimated LLM cost (this run) ===` block (total, per-stage, models); failed runs append a cost-so-far line. Local/hybrid calls have no $ accounting — total covers OpenRouter only ## Vision vs text models diff --git a/backend/.env.example b/backend/.env.example index ff845fc..ec5335e 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -48,6 +48,13 @@ REASON_CONCURRENCY=4 APP_BASE_URL=https://conchecker.scoutitsystems.com # APP_BUILD is set by CI at image build time (sha-) - do not set manually. +# LLM observability (job-log verbosity + raw request/response dumps) +# LLM_VERBOSE: one line per LLM call in job.log (model, sizes, item counts, cost) +# LLM_RAW_DUMP: full prompt+response per call in outputs//llm_raw/ +# (base64 images excluded). Both default on; set false to quiet down. +LLM_VERBOSE=true +LLM_RAW_DUMP=true + # Email notifications (optional). Leave SMTP_HOST blank to disable. # Examples: # Gmail: SMTP_HOST=smtp.gmail.com SMTP_PORT=587 (use an App Password) diff --git a/backend/agents/runner.py b/backend/agents/runner.py index afb89f2..bacd107 100644 --- a/backend/agents/runner.py +++ b/backend/agents/runner.py @@ -20,6 +20,7 @@ from backend.agents.linker import LinkerAgent, build_link_scopes, build_object_g from backend.agents.memory import ProjectMemory from backend.agents.orchestrator import Orchestrator from backend.agents.rfi_writer import RFIWriterAgent +from backend.llm import reset_cost from backend.pipeline.pdf_processor import convert_pdf_to_images from backend.pipeline.report import build_report, to_markdown from backend.pipeline.sheet_index import derive_project_meta_from_cover @@ -39,6 +40,10 @@ def run_agent_pipeline( if not os.path.isfile(pdf_path): raise FileNotFoundError(pdf_path) + # Keep the llm module's counters job-local (matches Classic): the job + # log's failure-path cost estimate in jobs.py reads llm.get_cost(). + reset_cost() + agent_dir = os.path.join(out_dir, "agent") if out_dir else None memory = ProjectMemory(artifact_dir=agent_dir) orchestrator = Orchestrator(memory=memory, on_stage=on_stage) diff --git a/backend/config.py b/backend/config.py index 169b5a8..73c68a9 100644 --- a/backend/config.py +++ b/backend/config.py @@ -102,6 +102,14 @@ CLUSTER_MAX = int(os.getenv("CLUSTER_MAX", "120")) LLM_CACHE = os.getenv("LLM_CACHE", "false").strip().lower() in ("1", "true", "yes") LLM_CACHE_DIR = os.getenv("LLM_CACHE_DIR", os.path.join(_BASE_DIR, ".llm_cache")) +# Verbose LLM observability. Per call, one line lands in the job log (model, +# backend, prompt size, response size, parsed-item counts, per-call cost) and +# the full request/response is dumped to /llm_raw/ (base64 image +# payloads excluded; image count recorded instead) so missed or hallucinated +# items can be traced back to exactly what the model saw and returned. +LLM_VERBOSE = os.getenv("LLM_VERBOSE", "true").strip().lower() in ("1", "true", "yes") +LLM_RAW_DUMP = os.getenv("LLM_RAW_DUMP", "true").strip().lower() in ("1", "true", "yes") + # Parallelism (ThreadPoolExecutor workers) EXTRACT_CONCURRENCY = int(os.getenv("EXTRACT_CONCURRENCY", "4")) REASON_CONCURRENCY = int(os.getenv("REASON_CONCURRENCY", "4")) diff --git a/backend/jobs.py b/backend/jobs.py index d199541..93d572d 100644 --- a/backend/jobs.py +++ b/backend/jobs.py @@ -21,7 +21,8 @@ import traceback import uuid import shutil import threading -from typing import Dict, List, Optional +from contextlib import contextmanager +from typing import Dict, Iterator, List, Optional from backend import config from backend import llm @@ -139,6 +140,12 @@ def _run( _append_log(job_id, f"[Jobs] Job {job_id} failed: {e}", log_path) for ln in traceback.format_exc().rstrip().splitlines(): _append_log(job_id, ln, log_path) + # Cost-so-far for the failed run (counters are reset per job). + cost = llm.get_cost() + _append_log(job_id, + f"[Jobs] Estimated LLM cost before failure: " + f"${cost['usd']:.4f} over {cost['calls']} live calls " + f"({cost.get('cached', 0)} cached)", log_path) except Exception: pass print(f"[Jobs] Job {job_id} failed: {e}") @@ -170,6 +177,9 @@ def _run_pipeline(job_id: str, pdf_path: str, out_dir: str, }, f, indent=2) # Keep a copy of the source PDF so its sheets can be viewed later. shutil.copy2(pdf_path, os.path.join(out_dir, "source.pdf")) + # Raw per-call LLM request/response dumps land in /llm_raw/. + if config.LLM_RAW_DUMP: + llm.set_raw_dump_dir(os.path.join(out_dir, "llm_raw")) runner = run_agent_pipeline if pipeline_mode == "agent" else run_pipeline runner_kwargs = { "out_dir": out_dir, @@ -192,9 +202,11 @@ def _run_pipeline(job_id: str, pdf_path: str, out_dir: str, try: report = runner(pdf_path, **runner_kwargs) finally: + llm.set_raw_dump_dir(None) if pipeline_mode == "agent": llm.set_model_overrides(None, None) report.setdefault("summary", {})["pipeline_mode"] = pipeline_mode + _log_cost_summary(report.get("summary", {})) if report["summary"].get("agent_status") == "needs_review": # Human-review gate: hold the job, don't email the unreviewed report. _set(job_id, status="needs_review", report=report, @@ -208,6 +220,55 @@ def _run_pipeline(job_id: str, pdf_path: str, out_dir: str, _notify(job_id, report, out_dir) +def _log_cost_summary(summary: Dict, label: str = "this run") -> None: + """ + End-of-log estimated LLM cost block, printed inside the job's stdio tee so + it lands at the tail of job.log. Both runners populate the same summary + fields (cost_usd / llm_calls / cached_calls / cost_by_stage / models_used). + Hybrid note: local text calls carry no usage accounting, so the dollar + figure covers OpenRouter calls only (local call counts still appear). + """ + if summary.get("cost_usd") is None and not summary.get("llm_calls"): + return + print(f"\n=== Estimated LLM cost ({label}) ===") + print(f" Total: ${summary.get('cost_usd', 0.0):.4f} across " + f"{summary.get('llm_calls', 0)} live calls " + f"({summary.get('cached_calls', 0)} cached at $0)") + for name, bucket in (summary.get("cost_by_stage") or {}).items(): + print(f" {name}: ${bucket.get('usd', 0.0):.4f} " + f"({bucket.get('calls', 0)} live, {bucket.get('cached', 0)} cached)") + mu = summary.get("models_used") or {} + if mu.get("vision"): + print(f" Vision models: {', '.join(mu['vision'])}") + if mu.get("text_local"): + print(f" Text models (local): {', '.join(mu['text_local'])}") + if mu.get("text_cloud"): + print(f" Text models (cloud): {', '.join(mu['text_cloud'])}") + if mu.get("fallback_count"): + print(f" Local->cloud fallbacks: {mu['fallback_count']}") + if mu.get("text_local"): + print(" Note: local calls have no cost accounting; " + "the dollar total covers OpenRouter usage only.") + + +@contextmanager +def capture_job_output(job_id: str, out_dir: str) -> Iterator[None]: + """ + Re-open the stdio tee + raw LLM dump dir for post-run work that still + belongs to this job (review finalization): lines append to job.log and + the in-memory log, raw dumps resume under /llm_raw/. The tee is + process-global — same overlapping-job caveat as the main run. + """ + log_path = os.path.join(out_dir, "job.log") + if config.LLM_RAW_DUMP: + llm.set_raw_dump_dir(os.path.join(out_dir, "llm_raw")) + try: + with capture_stdio(lambda raw: _append_log(job_id, raw, log_path)): + yield + finally: + llm.set_raw_dump_dir(None) + + def _notify(job_id: str, report: Dict, out_dir: str) -> None: email = _jobs[job_id].get("email") if not email: diff --git a/backend/llm.py b/backend/llm.py index 7871b67..3a5cf98 100644 --- a/backend/llm.py +++ b/backend/llm.py @@ -8,6 +8,7 @@ with an image) and conflict-reasoning (Stage 3, with images) calls. """ import os +import re import json import hashlib import threading @@ -31,6 +32,104 @@ _text_local = False _vision_model_override: Optional[str] = None _text_model_override: Optional[str] = None +# Per-job raw request/response dumps (missed/hallucinated-item debugging). +# Set by the job runner to /llm_raw at job start, cleared after. +# Same module-global pattern as the model overrides (single-user tool). +_raw_dump_dir: Optional[str] = None +_seq_lock = threading.Lock() +_call_seq = 0 + + +def set_raw_dump_dir(path: Optional[str]) -> None: + """Point raw LLM request/response dumps at a directory. None disables.""" + global _raw_dump_dir, _call_seq + with _seq_lock: + _raw_dump_dir = path + _call_seq = 0 + + +def _next_seq() -> int: + global _call_seq + with _seq_lock: + _call_seq += 1 + return _call_seq + + +def _summarize_parsed(parsed: Any) -> str: + """ + Compact digest of a parsed response for the job log. List values become + item counts (e.g. conflicts[3]) so a stage that returned nothing (miss) + or invented items (hallucination) is visible without opening the raw dump. + """ + if isinstance(parsed, list): + return f"list[{len(parsed)}]" + if not isinstance(parsed, dict): + return type(parsed).__name__ + parts = [] + for k, v in parsed.items(): + if isinstance(v, list): + parts.append(f"{k}[{len(v)}]") + elif isinstance(v, dict): + parts.append(f"{k}{{{len(v)}}}") + else: + s = str(v) + parts.append(f"{k}={s[:40]!r}{'...' if len(s) > 40 else ''}") + out = ", ".join(parts) + return out[:300] + ("..." if len(out) > 300 else "") + + +def _dump_raw(seq: int, be: Dict[str, Any], usage_stage: str, + system_prompt: str, user_text: str, + images_b64: Optional[List[str]], max_tokens: int, + raw: str, parsed: Any) -> None: + """Write the full request/response for one call to the job's llm_raw dir.""" + if not _raw_dump_dir: + return + try: + os.makedirs(_raw_dump_dir, exist_ok=True) + safe_stage = re.sub(r"[^A-Za-z0-9_.-]+", "_", usage_stage)[:40] + safe_model = re.sub(r"[^A-Za-z0-9_.-]+", "_", be["model"]) + payload = { + "seq": seq, + "stage": usage_stage, + "model": be["model"], + "backend": "local" if be.get("local") else "cloud", + "max_tokens": max_tokens, + # base64 image payloads deliberately excluded (multi-MB each); + # the count + source.pdf in the job dir identify what was sent. + "n_images": len(images_b64 or []), + "system_prompt": system_prompt, + "user_text": user_text, + "raw_response": raw, + "parsed": parsed, + } + path = os.path.join(_raw_dump_dir, f"{seq:04d}_{safe_stage}_{safe_model}.json") + tmp = f"{path}.{threading.get_ident()}.tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump(payload, f, indent=2) + os.replace(tmp, path) # atomic so thread-pooled stages can't tear it + except OSError as e: + print(f"[LLM] raw dump failed: {e}") + + +def _log_call(seq: int, be: Dict[str, Any], usage_stage: str, + user_text: str, images_b64: Optional[List[str]], + raw: str, parsed: Any, usd: Optional[float], + cached: bool = False) -> None: + """One verbose per-call line for the job log (tee'd by job_log.py).""" + if not config.LLM_VERBOSE: + return + backend = "local" if be.get("local") else "cloud" + if cached: + cost_str = "cache hit" + elif usd is not None: + cost_str = f"${usd:.4f}" + else: + cost_str = "cost n/a" + print(f"[LLM] #{seq:04d} {usage_stage} | {be['model']} ({backend}) | " + f"in {len(user_text)}ch+{len(images_b64 or [])}img | " + f"out {len(raw)}ch | {cost_str} | {_summarize_parsed(parsed)}") + def set_text_backend(local: bool) -> None: """Choose whether text (no-image) calls go to the local endpoint this run.""" @@ -300,6 +399,7 @@ def call_json( """ has_images = bool(images_b64) be = _resolve_backend(has_images, model) + seq = _next_seq() cache_key = None if config.LLM_CACHE: @@ -312,6 +412,8 @@ def call_json( usage_tracker.record( usage_stage, be["model"], cached=True, has_images=has_images ) + _log_call(seq, be, usage_stage, user_text, images_b64, + "", hit, None, cached=True) return hit content: List[Dict[str, Any]] = [] @@ -352,6 +454,11 @@ def call_json( _record_model(be, has_images, fell_back) if cache_key: _cache_set(cache_key, parsed) + _log_call(seq, be, usage_stage, user_text, images_b64, + raw, parsed, usd) + if config.LLM_RAW_DUMP: + _dump_raw(seq, be, usage_stage, system_prompt, user_text, + images_b64, max_tokens, raw, parsed) return parsed if attempt == 0: print("[LLM] JSON parse error (retrying)") diff --git a/backend/main.py b/backend/main.py index d4232a1..8145c44 100644 --- a/backend/main.py +++ b/backend/main.py @@ -20,7 +20,7 @@ from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles import backend.jobs -from backend import config +from backend import config, llm from backend.jobs import PIPELINE_MODES, create_job, get_job, _set from backend.pipeline.pdf_processor import render_page_jpeg from backend.review.feedback import decision_to_label, write_label @@ -186,7 +186,20 @@ def save_review_decisions(job_id: str, payload: dict): def _finalize_job(job_id: str, out_dir: str) -> None: """Background finalization: the ONE place the final report email may fire.""" try: - report = finalize_review(job_id, out_dir) + # Re-open the job's log tee + raw dump dir so the finalization LLM + # calls (clarification reruns, RFI drafting) land in job.log / llm_raw. + with backend.jobs.capture_job_output(job_id, out_dir): + print("\n=== Review finalization ===") + llm.reset_cost() # finalization-only cost attribution + report = finalize_review(job_id, out_dir) + cost = llm.get_cost() + backend.jobs._log_cost_summary({ + "cost_usd": round(cost["usd"], 4), + "llm_calls": cost["calls"], + "cached_calls": cost.get("cached", 0), + "cost_by_stage": cost.get("by_stage", {}), + "models_used": cost.get("models", {}), + }, label="finalization") except Exception as e: try: _set(job_id, status="finalization_error", error=str(e),