, key: string, fallback: string | number = "—") {
+ const raw = metadata[key];
+ return raw === undefined || raw === null || raw === "" ? String(fallback) : String(raw);
+}
+
+function duration(secondsValue: unknown): string {
+ const total = Math.max(0, Number(secondsValue) || 0);
+ const hours = Math.floor(total / 3600);
+ const minutes = Math.floor((total % 3600) / 60);
+ const seconds = Math.floor(total % 60);
+ return [hours, minutes, seconds].map((part) => String(part).padStart(2, "0")).join(":");
+}
+
+export function DigitadorLiveMetrics({ events }: Props) {
+ const latest = [...events].reverse().find((event) => {
+ const metadata = event.metadata ?? {};
+ return "estado" in metadata || "radicado_actual" in metadata;
+ });
+ if (!latest) return null;
+ const metadata = latest.metadata ?? {};
+
+ return (
+
+
Estado{value(metadata, "estado", latest.event_type)}
+
Radicado{value(metadata, "radicado_actual")}
+
Reinicios{value(metadata, "reinicios", 0)}
+
Velocidad{value(metadata, "speed_per_minute", 0)} rad/min
+
Tiempo{duration(metadata.elapsed_seconds)}
+
Último evento{value(metadata, "ultimo", latest.message)}
+
+ );
+}
diff --git a/services/desktop_app/src/renderer/modules/digitador/DigitadorResultsPanel.tsx b/services/desktop_app/src/renderer/modules/digitador/DigitadorResultsPanel.tsx
index 8236ec7..2a4f31c 100644
--- a/services/desktop_app/src/renderer/modules/digitador/DigitadorResultsPanel.tsx
+++ b/services/desktop_app/src/renderer/modules/digitador/DigitadorResultsPanel.tsx
@@ -17,6 +17,7 @@ import type {
} from "../../../shared/types";
import type { ApiClient } from "../../lib/apiClient";
import { ParallelBotProgress } from "../../components/ParallelBotProgress";
+import { DigitadorLiveMetrics } from "./DigitadorLiveMetrics";
declare const desktopApi: import("../../../shared/types").DesktopBridge | undefined;
@@ -284,6 +285,8 @@ export function DigitadorResultsPanel({ client, job, tasks, onStartDigitador }:
+
+
{selected.shard_count && Number(selected.shard_count) > 1
? `${selected.shard_count} bots (distribuido)`
diff --git a/services/desktop_app/src/renderer/modules/digitador/DigitadorWizardModule.tsx b/services/desktop_app/src/renderer/modules/digitador/DigitadorWizardModule.tsx
index b68e8b4..5cd75c2 100644
--- a/services/desktop_app/src/renderer/modules/digitador/DigitadorWizardModule.tsx
+++ b/services/desktop_app/src/renderer/modules/digitador/DigitadorWizardModule.tsx
@@ -15,6 +15,7 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import type { ApiClient } from "../../lib/apiClient";
import { formatDateTime } from "../../lib/time";
import { ParallelBotProgress } from "../../components/ParallelBotProgress";
+import { DigitadorLiveMetrics } from "./DigitadorLiveMetrics";
import type {
DigitadorJob,
DigitadorJobEvent,
@@ -627,6 +628,8 @@ export function DigitadorWizardModule({ client, job, initialSource = null }: Pro
+
+
diff --git a/services/desktop_app/src/renderer/styles.css b/services/desktop_app/src/renderer/styles.css
index 466d4bd..3d7502e 100644
--- a/services/desktop_app/src/renderer/styles.css
+++ b/services/desktop_app/src/renderer/styles.css
@@ -5635,3 +5635,65 @@ th {
margin-left: 28px !important;
margin-right: 28px !important;
}
+
+/* ── Costos de API por trabajo ── */
+.job-costs-view {
+ box-sizing: border-box;
+ height: 100%;
+ overflow: auto;
+ padding: 22px 28px 32px;
+}
+
+.job-costs-toolbar { align-items: flex-start; display: flex; gap: 20px; justify-content: space-between; }
+.job-costs-toolbar h2 { font-size: 20px; margin: 0 0 4px; }
+.job-costs-toolbar p { color: var(--text-muted); font-size: 13px; margin: 0; }
+.job-costs-summary { display: grid; gap: 12px; grid-template-columns: repeat(4, minmax(0, 1fr)); margin: 20px 0; }
+.job-costs-summary > div { background: var(--surface, #fff); border: 1px solid var(--border); border-radius: 10px; display: flex; flex-direction: column; gap: 6px; padding: 14px 16px; }
+.job-costs-summary span { color: var(--text-muted); font-size: 12px; }
+.job-costs-summary strong { font-size: 20px; }
+.job-cost-warning { background: #fffbeb; border: 1px solid #fde68a; border-radius: 8px; color: #92400e; font-size: 12px; margin: -8px 0 16px; padding: 9px 12px; }
+.job-costs-list { display: flex; flex-direction: column; gap: 10px; }
+.job-cost-card { background: var(--surface, #fff); border: 1px solid var(--border); border-radius: 10px; overflow: hidden; }
+.job-cost-card > summary { align-items: center; cursor: pointer; display: grid; gap: 16px; grid-template-columns: minmax(220px, 1.5fr) auto auto minmax(100px, auto); list-style-position: inside; padding: 14px 16px; }
+.job-cost-process { display: inline-flex; flex-direction: column; min-width: 0; }
+.job-cost-process span { color: var(--text-muted); font-size: 12px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
+.job-cost-api-badge { background: #f1f5f9; border-radius: 999px; color: #64748b; font-size: 11px; padding: 4px 8px; }
+.job-cost-api-badge.has-usage { background: #dcfce7; color: #15803d; }
+.job-cost-total { color: #0f766e; text-align: right; }
+.job-cost-detail { border-top: 1px solid var(--border); padding: 12px 16px 16px; }
+.job-cost-table { border-collapse: collapse; font-size: 12px; width: 100%; }
+.job-cost-table th, .job-cost-table td { border-bottom: 1px solid var(--border); padding: 9px 10px; text-align: right; }
+.job-cost-table th:first-child, .job-cost-table td:first-child { text-align: left; }
+.job-cost-table td small { color: var(--text-muted); display: block; font-size: 10px; margin-top: 2px; }
+.job-cost-calls { margin-top: 14px; }
+.job-cost-calls > summary { color: #2563eb; cursor: pointer; font-size: 12px; font-weight: 600; }
+.job-cost-calls-scroll { margin-top: 8px; max-height: 340px; overflow: auto; }
+.job-cost-no-api, .job-costs-empty { color: var(--text-muted); font-size: 13px; }
+.job-costs-empty { align-items: center; border: 1px dashed var(--border); border-radius: 10px; display: flex; flex-direction: column; padding: 40px; }
+@media (max-width: 900px) {
+ .job-costs-summary { grid-template-columns: repeat(2, minmax(0, 1fr)); }
+ .job-cost-card > summary { grid-template-columns: 1fr auto; }
+ .job-cost-card > summary > span:not(.job-cost-api-badge) { display: none; }
+}
+
+.digitador-live-metrics {
+ display: grid;
+ gap: 8px;
+ grid-template-columns: repeat(3, minmax(0, 1fr));
+ margin: 4px 0 16px;
+ width: 100%;
+}
+
+.digitador-live-metrics > div {
+ background: var(--surface, #fff);
+ border: 1px solid var(--border);
+ border-radius: 8px;
+ display: flex;
+ flex-direction: column;
+ gap: 3px;
+ min-width: 0;
+ padding: 9px 11px;
+}
+
+.digitador-live-metrics span { color: var(--text-muted); font-size: 10px; text-transform: uppercase; }
+.digitador-live-metrics strong { font-size: 12px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
diff --git a/services/desktop_app/src/renderer/types/jobSection.ts b/services/desktop_app/src/renderer/types/jobSection.ts
index 2aed4da..272b558 100644
--- a/services/desktop_app/src/renderer/types/jobSection.ts
+++ b/services/desktop_app/src/renderer/types/jobSection.ts
@@ -5,4 +5,5 @@ export type JobSection =
| "poblacion_apta"
| "downloads"
| "digitador"
+ | "costs"
| "results";
diff --git a/services/desktop_app/src/renderer/views/JobCostsView.test.tsx b/services/desktop_app/src/renderer/views/JobCostsView.test.tsx
new file mode 100644
index 0000000..a93720f
--- /dev/null
+++ b/services/desktop_app/src/renderer/views/JobCostsView.test.tsx
@@ -0,0 +1,50 @@
+import { render, screen } from "@testing-library/react";
+import type { Task } from "../../shared/types";
+import { JobCostsView } from "./JobCostsView";
+
+function task(result: Task["result"]): Task {
+ return {
+ id: "task-1",
+ type: "extraction",
+ status: "completed",
+ title: "Extracción facturas",
+ request: {},
+ result,
+ error: null,
+ owner_id: null,
+ job_id: "job-1",
+ created_at: "2026-08-30T12:00:00Z",
+ updated_at: "2026-08-30T12:01:00Z",
+ started_at: "2026-08-30T12:00:00Z",
+ finished_at: "2026-08-30T12:01:00Z",
+ };
+}
+
+describe("JobCostsView", () => {
+ it("shows real API usage totals for a process", () => {
+ render(
+
undefined}
+ tasks={[task({
+ api_usage: {
+ summary: { requests: 2, total_tokens: 1250, total_cost_usd: 0.00125 },
+ phases: {
+ extract: { requests: 2, prompt_tokens: 1000, completion_tokens: 250, total_cost_usd: 0.00125, model: "gpt-5.6-luna" },
+ },
+ },
+ })]}
+ />,
+ );
+
+ expect(screen.getByText("Extracción facturas")).toBeInTheDocument();
+ expect(screen.getByText("2 llamadas")).toBeInTheDocument();
+ expect(screen.getAllByText("1.250").length).toBeGreaterThan(0);
+ expect(screen.getAllByText("$0.00125").length).toBeGreaterThan(0);
+ });
+
+ it("shows processes that did not consume a priced API", () => {
+ render( undefined} tasks={[task({})]} />);
+ expect(screen.getByText("Sin consumo API")).toBeInTheDocument();
+ });
+});
diff --git a/services/desktop_app/src/renderer/views/JobCostsView.tsx b/services/desktop_app/src/renderer/views/JobCostsView.tsx
new file mode 100644
index 0000000..186ab97
--- /dev/null
+++ b/services/desktop_app/src/renderer/views/JobCostsView.tsx
@@ -0,0 +1,190 @@
+import { Coins, RefreshCw } from "lucide-react";
+import type { Task } from "../../shared/types";
+
+type UsageSummary = {
+ requests?: number;
+ prompt_tokens?: number;
+ cached_tokens?: number;
+ completion_tokens?: number;
+ total_tokens?: number;
+ input_usd?: number;
+ cached_input_usd?: number;
+ output_usd?: number;
+ total_cost_usd?: number;
+ unpriced_requests?: number;
+ model?: string;
+};
+
+type UsageDetail = UsageSummary & {
+ timestamp?: string;
+ phase?: string;
+ model?: string;
+ radicado?: string | null;
+ filename?: string | null;
+ cost?: { total_usd?: number };
+ pricing_known?: boolean;
+};
+
+type ApiUsage = {
+ summary?: UsageSummary;
+ phases?: Record;
+ details?: UsageDetail[];
+};
+
+interface Props {
+ tasks: Task[];
+ refreshing: boolean;
+ onRefresh: () => void;
+}
+
+const PROCESS_NAMES: Record = {
+ counting: "Conteo",
+ extraction: "Extracción",
+ cleaning: "Limpieza",
+ poblacion_apta: "Población apta",
+ downloads: "Bot de descarga",
+ digitador: "Bot digitador",
+};
+
+function usageOf(task: Task): ApiUsage {
+ const value = task.result?.api_usage;
+ return value && typeof value === "object" ? value as ApiUsage : {};
+}
+
+function n(value: unknown): number {
+ const parsed = Number(value ?? 0);
+ return Number.isFinite(parsed) ? parsed : 0;
+}
+
+function usd(value: unknown): string {
+ return new Intl.NumberFormat("en-US", {
+ style: "currency",
+ currency: "USD",
+ minimumFractionDigits: 4,
+ maximumFractionDigits: 6,
+ }).format(n(value));
+}
+
+function integer(value: unknown): string {
+ return new Intl.NumberFormat("es-CO").format(n(value));
+}
+
+function phaseName(phase: string): string {
+ if (phase === "extract") return "Extracción LLM";
+ if (phase === "normalize") return "Normalización LLM";
+ return phase;
+}
+
+export function JobCostsView({ tasks, refreshing, onRefresh }: Props) {
+ const ordered = [...tasks].sort((a, b) => new Date(b.created_at).getTime() - new Date(a.created_at).getTime());
+ const summaries = ordered.map((task) => usageOf(task).summary ?? {});
+ const totalCost = summaries.reduce((sum, item) => sum + n(item.total_cost_usd), 0);
+ const totalTokens = summaries.reduce((sum, item) => sum + n(item.total_tokens), 0);
+ const totalRequests = summaries.reduce((sum, item) => sum + n(item.requests), 0);
+ const unpricedRequests = summaries.reduce((sum, item) => sum + n(item.unpriced_requests), 0);
+
+ return (
+
+
+
+
Costos de API por proceso
+
Uso real reportado por el proveedor. Los valores históricos conservan la tarifa aplicada al procesar.
+
+
+
+
+
+
Costo total{usd(totalCost)}
+
Tokens{integer(totalTokens)}
+
Llamadas API{integer(totalRequests)}
+
Procesos{integer(tasks.length)}
+
+
+ {unpricedRequests > 0 && (
+
+ {integer(unpricedRequests)} llamadas tienen un modelo sin tarifa configurada; sus tokens se muestran, pero no se suman al costo.
+
+ )}
+
+ {ordered.length === 0 ? (
+
Aún no hay procesos en este trabajo.
+ ) : (
+
+ {ordered.map((task) => {
+ const apiUsage = usageOf(task);
+ const summary = apiUsage.summary ?? {};
+ const phases = Object.entries(apiUsage.phases ?? {});
+ const details = apiUsage.details ?? [];
+ const hasUsage = n(summary.requests) > 0;
+ return (
+
+
+
+ {PROCESS_NAMES[task.type] ?? task.type}
+ {task.title || `Proceso ${task.id.slice(0, 8)}`}
+
+
+ {hasUsage ? `${integer(summary.requests)} llamadas` : "Sin consumo API"}
+
+ {integer(summary.total_tokens)} tokens
+ {usd(summary.total_cost_usd)}
+
+
+
+ {hasUsage ? (
+ <>
+
+ | Fase / modelo | Llamadas | Entrada | En caché | Salida | Costo |
+
+ {phases.map(([phase, item]) => (
+
+ | {phaseName(phase)}{item.model || "Modelo no informado"} |
+ {integer(item.requests)} |
+ {integer(item.prompt_tokens)} |
+ {integer(item.cached_tokens)} |
+ {integer(item.completion_tokens)} |
+ {usd(item.total_cost_usd)} |
+
+ ))}
+
+
+
+ {details.length > 0 && (
+
+ Ver detalle de {details.length} llamadas
+
+
+ | Documento | Fase | Modelo | Entrada | Caché | Salida | Costo |
+
+ {details.map((item, index) => (
+
+ | {item.radicado || item.filename || "—"}{item.radicado && item.filename ? item.filename : ""} |
+ {phaseName(item.phase || "API")} |
+ {item.model || "—"} |
+ {integer(item.prompt_tokens)} |
+ {integer(item.cached_tokens)} |
+ {integer(item.completion_tokens)} |
+ {item.pricing_known === false ? "Sin tarifa" : usd(item.cost?.total_usd)} |
+
+ ))}
+
+
+
+
+ )}
+ >
+ ) : (
+
Este proceso no registró llamadas a una API LLM tarifada.
+ )}
+
+
+ );
+ })}
+
+ )}
+
+ );
+}
diff --git a/services/desktop_app/src/renderer/views/JobView.tsx b/services/desktop_app/src/renderer/views/JobView.tsx
index 09c7663..0fbd7ca 100644
--- a/services/desktop_app/src/renderer/views/JobView.tsx
+++ b/services/desktop_app/src/renderer/views/JobView.tsx
@@ -11,6 +11,7 @@ import { JobExtractionWizard } from "../modules/extraction/JobExtractionWizard";
import { DownloadsWizardModule } from "../modules/downloads/DownloadsWizardModule";
import { DigitadorWizardModule } from "../modules/digitador/DigitadorWizardModule";
import { JobResultsView } from "./JobResultsView";
+import { JobCostsView } from "./JobCostsView";
interface Props {
client: ApiClient;
@@ -30,6 +31,7 @@ export function JobView({ client, job, section, sectionContext, onActivateSectio
const [extractionProgress, setExtractionProgress] = useState(null);
const [cleaningProgress, setCleaningProgress] = useState(null);
const [poblacionAptaProgress, setPoblacionAptaProgress] = useState(null);
+ const [costsRefreshing, setCostsRefreshing] = useState(false);
const pollingRef = useRef | null>(null);
const backgroundRunning =
@@ -45,6 +47,11 @@ export function JobView({ client, job, section, sectionContext, onActivateSectio
} catch { /* silent */ }
}, [client, job.id]);
+ const refreshCosts = useCallback(async () => {
+ setCostsRefreshing(true);
+ try { await loadTasks(); } finally { setCostsRefreshing(false); }
+ }, [loadTasks]);
+
useEffect(() => {
loadTasks();
return () => { if (pollingRef.current) clearInterval(pollingRef.current); };
@@ -129,6 +136,7 @@ export function JobView({ client, job, section, sectionContext, onActivateSectio
{section === "poblacion_apta" && "Población apta"}
{section === "downloads" && "Bot de descarga"}
{section === "digitador" && "Bot digitador"}
+ {section === "costs" && "Costos de API"}
{section === "results" && "Resultados del trabajo"}
@@ -236,6 +244,10 @@ export function JobView({ client, job, section, sectionContext, onActivateSectio
onContinueToDigitador={(ctx) => onActivateSection("digitador", ctx)}
/>
+
+
+ void refreshCosts()} />
+
);
diff --git a/services/extraction/api/desktop_ingest.py b/services/extraction/api/desktop_ingest.py
index 2ba6fed..438f668 100644
--- a/services/extraction/api/desktop_ingest.py
+++ b/services/extraction/api/desktop_ingest.py
@@ -8,6 +8,8 @@ from domain.normalizer import normalize_to_schema
from domain.profiles import get_profile
from domain.prompts import PromptBuilder
from domain.service import InvoiceProcessingService
+from domain.config import settings
+from domain.usage_tracking import record_api_usage
from infrastructure.database import ExtractionRow, Task, utcnow
from infrastructure.serialization import flatten_extraction_result
from sqlalchemy.orm import Session
@@ -276,7 +278,7 @@ def process_desktop_ingest(
else:
prompt = instruction or PromptBuilder.build(selected_paths, {}, profile_name=prof.get("name", "invoice"))
- data, _usage = service.extract_full_from_pdf(
+ data, extraction_usage = service.extract_full_from_pdf(
temp_path,
instruction=prompt,
tiff_page_range=tiff_page_range,
@@ -292,8 +294,28 @@ def process_desktop_ingest(
profile_name=prof.get("name", "invoice"),
)
normalized_obj = data
+ normalizer_usage = None
if isinstance(norm, dict):
normalized_obj = norm.get("normalized") or data
+ normalizer_usage = norm.get("_usage")
+
+ usage_result = record_api_usage(
+ dict(task.result or {}),
+ phase="extract",
+ model=str(getattr(service.extractor, "model", model or "unknown")),
+ usage=extraction_usage,
+ radicado=radicado,
+ filename=filename,
+ )
+ usage_result = record_api_usage(
+ usage_result,
+ phase="normalize",
+ model=str(settings.api.normalizer_model),
+ usage=normalizer_usage,
+ radicado=radicado,
+ filename=filename,
+ )
+ task.result = usage_result
return persist_desktop_ingest_result(
session,
diff --git a/services/extraction/api/main.py b/services/extraction/api/main.py
index 0c9d186..268c2bd 100644
--- a/services/extraction/api/main.py
+++ b/services/extraction/api/main.py
@@ -228,6 +228,28 @@ def serialize_task(task: Task) -> dict[str, Any]:
def persist_extraction_request(request: ExtractionRequest) -> dict[str, Any]:
+ from domain.usage_tracking import record_api_usage
+
+ usage_result: dict[str, Any] = {}
+
+ def capture_usage(
+ phase: str,
+ radicado: str,
+ usage: dict[str, Any] | None,
+ _cost: dict[str, Any] | None,
+ ) -> None:
+ nonlocal usage_result
+ model = request.model or settings.api.extractor_model
+ if phase == "normalize":
+ model = settings.api.normalizer_model
+ usage_result = record_api_usage(
+ usage_result,
+ phase=phase,
+ model=str(model),
+ usage=usage,
+ radicado=radicado,
+ )
+
with session_scope() as session:
batch = ExtractionBatch(
status="running",
@@ -252,6 +274,7 @@ def persist_extraction_request(request: ExtractionRequest) -> dict[str, Any]:
structure_mode=request.structure_mode,
retry_on_empty=request.retry_on_empty,
radicado_filter=request.radicado_filter,
+ on_usage=capture_usage,
)
with session_scope() as session:
@@ -279,6 +302,7 @@ def persist_extraction_request(request: ExtractionRequest) -> dict[str, Any]:
session.flush()
serialized = serialize_batch(batch)
serialized["row_count"] = row_index
+ serialized["api_usage"] = usage_result.get("api_usage", {})
return serialized
except Exception as exc:
with session_scope() as session:
@@ -312,6 +336,7 @@ def run_extraction_task(task_id: str) -> None:
"extraction_batch_id": batch["id"],
"row_count": batch.get("row_count", 0),
"batch": batch,
+ "api_usage": batch.get("api_usage", {}),
}
task.error = None
task.finished_at = utcnow()
@@ -620,6 +645,8 @@ async def _process_sync(
from domain.profiles import get_profile
from domain.prompts import PromptBuilder
from domain.normalizer import normalize_to_schema
+ from domain.config import settings as _settings
+ from domain.usage_tracking import record_api_usage
_log = _logging.getLogger("extraction.ingest")
@@ -656,13 +683,13 @@ async def _process_sync(
tiff_range = req.get("tiff_page_range")
model_name = getattr(svc.extractor, "model", "?")
_log.info("→ OpenAI [%s] radicado=%s filename=%s", model_name, radicado, effective_name)
- data, _usage = svc.extract_full_from_pdf(
+ data, extraction_usage = svc.extract_full_from_pdf(
temp_path,
instruction=prompt,
tiff_page_range=tiff_range,
)
- tok_in = (_usage or {}).get("prompt_tokens", "?")
- tok_out = (_usage or {}).get("completion_tokens", "?")
+ tok_in = (extraction_usage or {}).get("prompt_tokens", "?")
+ tok_out = (extraction_usage or {}).get("completion_tokens", "?")
_log.info("← OpenAI [%s] radicado=%s tokens=%s→%s", model_name, radicado, tok_in, tok_out)
if isinstance(data, dict):
data["radicado"] = radicado
@@ -670,11 +697,13 @@ async def _process_sync(
data = {"radicado": radicado, "data": data}
norm = normalize_to_schema(data, paths=selected_paths, profile_name=prof.get("name", "invoice"))
normalized_obj = data
+ normalizer_usage = None
if isinstance(norm, dict):
normalized_obj = norm.get("normalized") or data
- return normalized_obj, selected_paths
+ normalizer_usage = norm.get("_usage")
+ return normalized_obj, selected_paths, extraction_usage, normalizer_usage, model_name
- normalized_obj, selected_paths = await _asyncio.to_thread(_call_llm)
+ normalized_obj, selected_paths, extraction_usage, normalizer_usage, model_name = await _asyncio.to_thread(_call_llm)
# ── Fase 2: persistir resultado (con FOR UPDATE para serializar) ───
def _persist():
@@ -682,6 +711,14 @@ async def _process_sync(
task = session.get(Task, task_id, with_for_update=True)
if task is None:
return {"error": "task_not_found"}
+ task.result = record_api_usage(
+ dict(task.result or {}), phase="extract", model=str(model_name),
+ usage=extraction_usage, radicado=radicado, filename=effective_name,
+ )
+ task.result = record_api_usage(
+ dict(task.result or {}), phase="normalize", model=str(_settings.api.normalizer_model),
+ usage=normalizer_usage, radicado=radicado, filename=effective_name,
+ )
payload = persist_desktop_ingest_result(
session,
task=task,
diff --git a/services/extraction/domain/usage_tracking.py b/services/extraction/domain/usage_tracking.py
new file mode 100644
index 0000000..c9f1ea1
--- /dev/null
+++ b/services/extraction/domain/usage_tracking.py
@@ -0,0 +1,90 @@
+"""Persistencia compacta del uso real de APIs LLM dentro del resultado de una tarea."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+from domain.costs import MODEL_PRICING, compute_cost_from_usage
+
+
+def _number(value: Any) -> float:
+ try:
+ return float(value or 0)
+ except (TypeError, ValueError):
+ return 0.0
+
+
+def _tokens(usage: dict[str, Any], key: str) -> int:
+ try:
+ return int(usage.get(key) or 0)
+ except (TypeError, ValueError):
+ return 0
+
+
+def record_api_usage(
+ result: dict[str, Any],
+ *,
+ phase: str,
+ model: str,
+ usage: dict[str, Any] | None,
+ radicado: str | None = None,
+ filename: str | None = None,
+) -> dict[str, Any]:
+ """Agrega una llamada facturable a ``result.api_usage`` y devuelve una copia.
+
+ El proveedor es la fuente de los tokens. El costo se calcula con la tabla de
+ precios configurada para el modelo y se conserva separado por fase y llamada.
+ """
+ if not isinstance(usage, dict):
+ return dict(result)
+
+ next_result = dict(result)
+ ledger = dict(next_result.get("api_usage") or {})
+ summary = dict(ledger.get("summary") or {})
+ phases = dict(ledger.get("phases") or {})
+ phase_summary = dict(phases.get(phase) or {})
+ details = list(ledger.get("details") or [])
+
+ prompt_tokens = _tokens(usage, "prompt_tokens")
+ completion_tokens = _tokens(usage, "completion_tokens")
+ token_details = usage.get("prompt_tokens_details") or {}
+ cached_tokens = _tokens(token_details, "cached_tokens") if isinstance(token_details, dict) else 0
+ total_tokens = _tokens(usage, "total_tokens") or prompt_tokens + completion_tokens
+ cost = compute_cost_from_usage(usage, phase=phase, model=model) or {}
+
+ values: dict[str, int | float] = {
+ "requests": 1,
+ "prompt_tokens": prompt_tokens,
+ "cached_tokens": cached_tokens,
+ "completion_tokens": completion_tokens,
+ "total_tokens": total_tokens,
+ "input_usd": _number(cost.get("input_usd")),
+ "cached_input_usd": _number(cost.get("cached_input_usd")),
+ "output_usd": _number(cost.get("output_usd")),
+ "total_cost_usd": _number(cost.get("total_usd")),
+ "unpriced_requests": 0 if model in MODEL_PRICING else 1,
+ }
+ for key, value in values.items():
+ precision = 6 if key.endswith("usd") else None
+ summary[key] = round(_number(summary.get(key)) + value, precision) if precision else int(_number(summary.get(key)) + value)
+ phase_summary[key] = round(_number(phase_summary.get(key)) + value, precision) if precision else int(_number(phase_summary.get(key)) + value)
+
+ phase_summary["model"] = model
+ phases[phase] = phase_summary
+ details.append({
+ "timestamp": datetime.now(timezone.utc).isoformat(),
+ "phase": phase,
+ "model": model,
+ "radicado": radicado,
+ "filename": filename,
+ "prompt_tokens": prompt_tokens,
+ "cached_tokens": cached_tokens,
+ "completion_tokens": completion_tokens,
+ "total_tokens": total_tokens,
+ "cost": cost,
+ "pricing_known": model in MODEL_PRICING,
+ })
+
+ ledger.update({"summary": summary, "phases": phases, "details": details})
+ next_result["api_usage"] = ledger
+ return next_result
diff --git a/services/extraction/tests/test_usage_tracking.py b/services/extraction/tests/test_usage_tracking.py
new file mode 100644
index 0000000..9a92e7c
--- /dev/null
+++ b/services/extraction/tests/test_usage_tracking.py
@@ -0,0 +1,51 @@
+from domain.usage_tracking import record_api_usage
+
+
+def test_record_api_usage_aggregates_summary_phases_and_details():
+ result = record_api_usage(
+ {},
+ phase="extract",
+ model="gpt-5.6-luna",
+ usage={
+ "prompt_tokens": 1_000,
+ "completion_tokens": 200,
+ "total_tokens": 1_200,
+ "prompt_tokens_details": {"cached_tokens": 100},
+ },
+ radicado="RAD-1",
+ filename="factura.pdf",
+ )
+ result = record_api_usage(
+ result,
+ phase="extract",
+ model="gpt-5.6-luna",
+ usage={"prompt_tokens": 500, "completion_tokens": 50},
+ radicado="RAD-2",
+ )
+
+ ledger = result["api_usage"]
+ assert ledger["summary"]["requests"] == 2
+ assert ledger["summary"]["prompt_tokens"] == 1_500
+ assert ledger["summary"]["completion_tokens"] == 250
+ assert ledger["summary"]["total_tokens"] == 1_750
+ assert ledger["summary"]["unpriced_requests"] == 0
+ assert ledger["phases"]["extract"]["requests"] == 2
+ assert ledger["details"][0]["filename"] == "factura.pdf"
+ assert ledger["details"][0]["pricing_known"] is True
+
+
+def test_record_api_usage_marks_unknown_model_as_unpriced():
+ result = record_api_usage(
+ {},
+ phase="extract",
+ model="provider/model-without-price",
+ usage={"prompt_tokens": 100, "completion_tokens": 10},
+ )
+
+ assert result["api_usage"]["summary"]["unpriced_requests"] == 1
+ assert result["api_usage"]["details"][0]["pricing_known"] is False
+
+
+def test_record_api_usage_ignores_missing_usage():
+ original = {"processed": 1}
+ assert record_api_usage(original, phase="extract", model="gpt-5.6-luna", usage=None) == original
diff --git a/services/extraction/worker/tasks.py b/services/extraction/worker/tasks.py
index ff00230..5e2b578 100644
--- a/services/extraction/worker/tasks.py
+++ b/services/extraction/worker/tasks.py
@@ -24,6 +24,8 @@ from domain.profiles import get_profile
from domain.prompts import PromptBuilder
from domain.normalizer import normalize_to_schema
from domain.rate_limiter import get_rate_limiter
+from domain.config import settings
+from domain.usage_tracking import record_api_usage
logger = logging.getLogger(__name__)
@@ -131,8 +133,8 @@ async def process_document(
await rate_limiter.acquire(estimated_tokens=estimated_tokens)
# Extracción + normalización (CPU + IO bound, en thread pool)
- def _extract_and_normalize() -> tuple[dict, dict]:
- data, _usage = service.extract_full_from_pdf(
+ def _extract_and_normalize() -> tuple[dict, dict, dict | None, dict | None]:
+ data, extraction_usage = service.extract_full_from_pdf(
queue_path,
instruction=prompt,
tiff_page_range=tiff_range,
@@ -149,11 +151,13 @@ async def process_document(
profile_name=prof.get("name", "invoice"),
)
normalized_obj = data
+ normalizer_usage = None
if isinstance(norm, dict):
normalized_obj = norm.get("normalized") or data
- return raw_data, normalized_obj
+ normalizer_usage = norm.get("_usage")
+ return raw_data, normalized_obj, extraction_usage, normalizer_usage
- _raw, normalized_obj = await asyncio.to_thread(_extract_and_normalize)
+ _raw, normalized_obj, extraction_usage, normalizer_usage = await asyncio.to_thread(_extract_and_normalize)
# Persistir resultado en BD
def _persist() -> dict[str, Any]:
@@ -164,6 +168,15 @@ async def process_document(
return {"skipped": True, "reason": "task_not_found"}
result = dict(task.result or {})
+ result = record_api_usage(
+ result, phase="extract", model=str(effective_model), usage=extraction_usage,
+ radicado=radicado, filename=filename,
+ )
+ result = record_api_usage(
+ result, phase="normalize", model=str(settings.api.normalizer_model), usage=normalizer_usage,
+ radicado=radicado, filename=filename,
+ )
+ task.result = result
if structure_mode == "radicados" and doc_types and doc_type:
partials = dict(result.get("partials") or {})