feat: kontinuierliche LXC↔Mac-Diagnose + Auto-Retry
Diagnose-Sichtbarkeit (Plan A):
- Job.step_started_at + last_heartbeat_at, Heartbeat-Task pingt alle 3 s
während laufender Mac-Calls
- Mac-Worker hält per X-Job-Id Header einen Log-Ringbuffer pro Job
(200 Zeilen, 1 h TTL); GET /api/jobs/{id}/log liefert ihn aus
- LXC-Endpoint GET /api/jobs/{id}/diag bündelt Job-Status, Stage-Sekunden,
Heartbeat-Alter und Mac-Worker-Log
- Frontend: Live-Timing im Status-Pill ("läuft seit Xs · letzter Ping vor Ys"),
klappbares Diagnose-Panel mit Mac-Log für FAILED + langlaufende Jobs
Robustheit (Plan B):
- MacClient: einmaliger Retry mit 3 s Backoff bei RemoteProtocolError /
ConnectError / ReadError für summarize/protocol/export
- Mac-Worker /api/preload heizt Ollama vor (keep_alive 30 m, num_predict 1);
Pipeline ruft Preload vor summarize, damit Modell-Reload nicht mitten
in der Generierung passiert (Hauptursache "Server disconnected")
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
@@ -1,7 +1,9 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Awaitable, TypeVar
|
||||
|
||||
from sqlmodel import Session
|
||||
|
||||
@@ -12,6 +14,10 @@ from app.models.job import Job, JobStatus
|
||||
|
||||
log = logging.getLogger("voice-agent.pipeline")
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
HEARTBEAT_INTERVAL = 3.0 # Sekunden
|
||||
|
||||
|
||||
def _update(job_id: int, **fields) -> None:
|
||||
with Session(engine) as s:
|
||||
@@ -32,8 +38,51 @@ def _existing(path_str: str) -> Path | None:
|
||||
return p if p.exists() and p.stat().st_size > 0 else None
|
||||
|
||||
|
||||
def _start_step(job_id: int, status: str, progress: int) -> None:
|
||||
"""Setzt status/progress + step_started_at + initialen Heartbeat."""
|
||||
now = datetime.now(timezone.utc)
|
||||
_update(
|
||||
job_id,
|
||||
status=status,
|
||||
progress=progress,
|
||||
step_started_at=now,
|
||||
last_heartbeat_at=now,
|
||||
error="",
|
||||
)
|
||||
|
||||
|
||||
async def _with_heartbeat(job_id: int, coro: Awaitable[T]) -> T:
|
||||
"""Führt `coro` aus und pingt parallel alle paar Sekunden last_heartbeat_at.
|
||||
|
||||
Damit sieht die UI live, ob ein Schritt noch lebt — auch wenn der Mac
|
||||
minutenlang an einer Ollama-Generierung arbeitet.
|
||||
"""
|
||||
stop = asyncio.Event()
|
||||
|
||||
async def beat() -> None:
|
||||
while not stop.is_set():
|
||||
try:
|
||||
_update(job_id, last_heartbeat_at=datetime.now(timezone.utc))
|
||||
except Exception: # noqa: BLE001
|
||||
pass # Heartbeat darf den Job nie killen
|
||||
try:
|
||||
await asyncio.wait_for(stop.wait(), timeout=HEARTBEAT_INTERVAL)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
|
||||
beat_task = asyncio.create_task(beat())
|
||||
try:
|
||||
return await coro
|
||||
finally:
|
||||
stop.set()
|
||||
try:
|
||||
await beat_task
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
|
||||
async def run_pipeline(job_id: int) -> None:
|
||||
"""Pipeline: transcribe → summarize → protocol → DOCX.
|
||||
"""Pipeline: transcribe → preload → summarize → protocol → DOCX → PDF.
|
||||
|
||||
Idempotent: bereits vorhandene Zwischenergebnisse werden weiterverwendet,
|
||||
damit ein 'Erneut versuchen' nicht die teure Transkription wiederholt.
|
||||
@@ -64,9 +113,9 @@ async def run_pipeline(job_id: int) -> None:
|
||||
tx = json.loads(existing_tx_path.read_text(encoding="utf-8"))
|
||||
_update(job_id, status=JobStatus.TRANSCRIBING, progress=45)
|
||||
else:
|
||||
_update(job_id, status=JobStatus.TRANSCRIBING, progress=10, error="")
|
||||
_start_step(job_id, JobStatus.TRANSCRIBING, 10)
|
||||
log.info("[job %s] transcribe (profile=%s)", job_id, profile)
|
||||
tx = await client.transcribe(audio_path, language="de")
|
||||
tx = await _with_heartbeat(job_id, client.transcribe(audio_path, language="de", job_id=job_id))
|
||||
transcript_path = job_dir / "transcript.json"
|
||||
transcript_path.write_text(json.dumps(tx, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
_update(job_id, transcript_path=str(transcript_path), progress=45)
|
||||
@@ -75,15 +124,24 @@ async def run_pipeline(job_id: int) -> None:
|
||||
s.get("text", "") for s in tx.get("segments", [])
|
||||
)
|
||||
|
||||
# 2) SUMMARY
|
||||
# 2) SUMMARY — vorher Preload, damit Ollama nicht mitten drin lädt
|
||||
if existing_sum_path:
|
||||
log.info("[job %s] reuse summary %s", job_id, existing_sum_path.name)
|
||||
summary = json.loads(existing_sum_path.read_text(encoding="utf-8"))
|
||||
_update(job_id, status=JobStatus.SUMMARIZING, progress=70)
|
||||
else:
|
||||
_update(job_id, status=JobStatus.SUMMARIZING, progress=55)
|
||||
_start_step(job_id, JobStatus.SUMMARIZING, 50)
|
||||
log.info("[job %s] preload model", job_id)
|
||||
try:
|
||||
await _with_heartbeat(job_id, client.preload(profile=profile, job_id=job_id))
|
||||
except Exception as e: # noqa: BLE001
|
||||
log.warning("[job %s] preload failed (non-fatal): %s", job_id, e)
|
||||
_update(job_id, progress=55)
|
||||
log.info("[job %s] summarize", job_id)
|
||||
summary = await client.summarize(transcript_text, title=title, profile=profile)
|
||||
summary = await _with_heartbeat(
|
||||
job_id,
|
||||
client.summarize(transcript_text, title=title, profile=profile, job_id=job_id),
|
||||
)
|
||||
summary_path = job_dir / "summary.json"
|
||||
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
_update(job_id, summary_path=str(summary_path), progress=70)
|
||||
@@ -94,9 +152,12 @@ async def run_pipeline(job_id: int) -> None:
|
||||
protocol = json.loads(existing_pro_path.read_text(encoding="utf-8"))
|
||||
_update(job_id, status=JobStatus.PROTOCOLLING, progress=85)
|
||||
else:
|
||||
_update(job_id, status=JobStatus.PROTOCOLLING, progress=75)
|
||||
_start_step(job_id, JobStatus.PROTOCOLLING, 75)
|
||||
log.info("[job %s] protocol", job_id)
|
||||
protocol = await client.protocol(transcript_text, summary, title=title, profile=profile)
|
||||
protocol = await _with_heartbeat(
|
||||
job_id,
|
||||
client.protocol(transcript_text, summary, title=title, profile=profile, job_id=job_id),
|
||||
)
|
||||
protocol_path = job_dir / "protocol.json"
|
||||
protocol_path.write_text(json.dumps(protocol, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
_update(job_id, protocol_path=str(protocol_path), progress=85)
|
||||
@@ -105,9 +166,12 @@ async def run_pipeline(job_id: int) -> None:
|
||||
if existing_dx_path:
|
||||
log.info("[job %s] reuse docx %s", job_id, existing_dx_path.name)
|
||||
else:
|
||||
_update(job_id, status=JobStatus.EXPORTING, progress=90)
|
||||
_start_step(job_id, JobStatus.EXPORTING, 90)
|
||||
log.info("[job %s] export docx", job_id)
|
||||
docx_bytes = await client.export_docx(protocol, profile=profile)
|
||||
docx_bytes = await _with_heartbeat(
|
||||
job_id,
|
||||
client.export_docx(protocol, profile=profile, job_id=job_id),
|
||||
)
|
||||
docx_path = job_dir / "protocol.docx"
|
||||
docx_path.write_bytes(docx_bytes)
|
||||
_update(job_id, docx_path=str(docx_path), progress=95)
|
||||
@@ -119,7 +183,10 @@ async def run_pipeline(job_id: int) -> None:
|
||||
else:
|
||||
_update(job_id, status=JobStatus.EXPORTING, progress=96)
|
||||
log.info("[job %s] export pdf", job_id)
|
||||
pdf_bytes = await client.export_pdf(protocol, profile=profile)
|
||||
pdf_bytes = await _with_heartbeat(
|
||||
job_id,
|
||||
client.export_pdf(protocol, profile=profile, job_id=job_id),
|
||||
)
|
||||
pdf_path = job_dir / "protocol.pdf"
|
||||
pdf_path.write_bytes(pdf_bytes)
|
||||
_update(job_id, pdf_path=str(pdf_path), status=JobStatus.DONE, progress=100, error="")
|
||||
|
||||
Reference in New Issue
Block a user