Verbose per-call LLM logging, raw request/response dumps, and end-of-log cost summary.
- [LLM] line per call: stage, model, prompt size, output size, cost, parsed item counts - LLM_RAW_DUMP: full prompt/response JSON per call under outputs/<job>/llm_raw/ - Cost block at tail of job.log (per-stage, per-model, cached vs live) - Agent mode: reset llm cost counters per job; review finalization now teed into job.log + dumps
This commit is contained in:
@@ -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_id>/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/<job_id>/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
|
||||
|
||||
|
||||
@@ -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-<short_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/<job_id>/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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 <out_dir>/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"))
|
||||
|
||||
+62
-1
@@ -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 <out_dir>/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 <out_dir>/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:
|
||||
|
||||
+107
@@ -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 <out_dir>/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)")
|
||||
|
||||
+15
-2
@@ -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),
|
||||
|
||||
Reference in New Issue
Block a user