AegisSight/AegisSight-Monitor
3
0
Fork 0
Code Issues Pull-Requests Actions Pakete Projekte Releases Wiki Aktivität

Analysepipeline: Visualisierung der Refresh-Schritte

Quellcode durchsuchen 
Neuer Tab "Analysepipeline" zwischen Faktencheck und Quellenuebersicht.
Zeigt 9 Verarbeitungsschritte als n8n-artige Blockkette: Quellen sichten,
Nachrichten sammeln, Doppeltes filtern, Relevanz bewerten, Orte erkennen,
Lagebild verfassen, Fakten pruefen, Qualitaetscheck, Benachrichtigen.

- Backend: refresh_pipeline_steps-Tabelle persistiert pro Refresh+Pass die
  Status- und Zahlen-Werte. pipeline_tracker.py kapselt Start/Done/Skip/Error
  inkl. WebSocket-Broadcast (Event-Typ pipeline_step). 9 Hooks im Orchestrator
  speisen die Anzeige.
- API: GET /api/incidents/{id}/pipeline liefert Definition + letzten Stand
  (Zahlen aus letztem Refresh, Multi-Pass-Konsolidierung).
- Frontend: pipeline.js rendert Vollbild-Blockkette mit pulsierendem Glow am
  aktiven Block, animierten Pfeilen bei Datenfluss, Haekchen am fertigen Block.
  Hover-Tooltip mit Erklaerung in Nutzersprache, Klick oeffnet Detail-Popup.
  Bei Research-Lagen leuchtet ein Schleifen-Pfeil pro Mehrfach-Durchlauf auf.
  Mini-Variante (nur Icons) im Refresh-Progress-Popup.
- CSS: Light/Dark-Theme-fest, dezenter Circuit-Hintergrund (5% Opacity),
  Mobile-vertikale Stapelung unter 900px, prefers-reduced-motion respektiert.
- Uebersprungene Schritte (z.B. Geoparsing ohne neue Artikel) werden
  ausgeblendet, brandneue Lagen ohne Refresh zeigen Hinweis.

Tooltips bewusst in normaler Sprache ohne Internas (keine Modellnamen,
keine Toolnamen, keine Phasen-Labels).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Dieser Commit ist enthalten in:
UserIsMH
2026-05-01 13:53:44 +02:00
Ursprung 2b51e49d0d
Commit 3a346ba2ec
10 geänderte Dateien mit 1326 neuen und 2 gelöschten Zeilen
Patch-Datei herunterladen Vergleichs-Datei herunterladen Alle Dateien ausklappen Alle Dateien einklappen

128
src/agents/orchestrator.py
Datei anzeigen

@@ -677,6 +677,7 @@ class AgentOrchestrator:
from agents.analyzer import AnalyzerAgent
from agents.factchecker import FactCheckerAgent
from feeds.rss_parser import RSSParser
from services import pipeline_tracker as _pipe
db = await get_db()
try:
@@ -719,6 +720,47 @@ class AgentOrchestrator:
log_id = cursor.lastrowid
usage_acc = UsageAccumulator()
# --- Pipeline-Tracking (Analysepipeline-Visualisierung) ---
_pass_nr = (_pass_info or {}).get("nr", 1)
_step_ids: dict[str, Optional[int]] = {}
async def _pipe_start(step_key: str):
try:
sid = await _pipe.start_step(
db, self._ws_manager,
refresh_log_id=log_id, incident_id=incident_id, step_key=step_key,
pass_number=_pass_nr, tenant_id=tenant_id,
visibility=visibility, created_by=created_by,
)
_step_ids[step_key] = sid
return sid
except Exception as _e:
logger.debug(f"_pipe_start({step_key}) ignoriert: {_e}")
return None
async def _pipe_done(step_key: str, count_value=None, count_secondary=None):
try:
sid = _step_ids.pop(step_key, None)
await _pipe.complete_step(
db, self._ws_manager, step_id=sid,
refresh_log_id=log_id, incident_id=incident_id, step_key=step_key,
pass_number=_pass_nr, count_value=count_value, count_secondary=count_secondary,
tenant_id=tenant_id, visibility=visibility, created_by=created_by,
)
except Exception as _e:
logger.debug(f"_pipe_done({step_key}) ignoriert: {_e}")
async def _pipe_skip(step_key: str):
try:
await _pipe.skip_step(
db, self._ws_manager,
refresh_log_id=log_id, incident_id=incident_id, step_key=step_key,
pass_number=_pass_nr, tenant_id=tenant_id,
visibility=visibility, created_by=created_by,
)
except Exception as _e:
logger.debug(f"_pipe_skip({step_key}) ignoriert: {_e}")
research_status = "deep_researching" if incident_type == "research" else "researching"
research_detail = "Hintergrundrecherche im Web läuft..." if incident_type == "research" else "RSS-Feeds und Web werden durchsucht..."
# Multi-Pass: Detail-Text mit Durchlauf-Info versehen
@@ -741,6 +783,23 @@ class AgentOrchestrator:
)
existing_db_articles_full = await cursor.fetchall()
# Pipeline-Schritt 1: Quellen sichten (vorbereitet)
await _pipe_start("sources_review")
try:
if incident_type == "adhoc":
_src_cursor = await db.execute(
"SELECT COUNT(*) AS cnt FROM sources WHERE tenant_id = ? AND status = 'active'",
(tenant_id,),
)
_src_row = await _src_cursor.fetchone()
_src_total = _src_row["cnt"] if _src_row else 0
else:
_src_total = None
except Exception:
_src_total = None
# secondary wird später mit der Anzahl tatsächlich liefernder Quellen ergänzt
await _pipe_done("sources_review", count_value=_src_total, count_secondary=None)
# Schritt 1+2: RSS-Feeds und Claude-Recherche parallel ausführen
async def _rss_pipeline():
"""RSS-Feed-Suche (Feed-Selektion + dynamische Keywords + Parsing)."""
@@ -880,6 +939,9 @@ class AgentOrchestrator:
logger.info(f"Telegram-Pipeline: {len(articles)} Nachrichten")
return articles, None
# Pipeline-Schritt 2: Nachrichten sammeln (Start)
await _pipe_start("collect")
# Pipelines parallel starten (RSS + WebSearch + Podcasts + optional Telegram)
pipelines = [_rss_pipeline(), _web_search_pipeline(), _podcast_pipeline()]
if include_telegram:
@@ -910,6 +972,15 @@ class AgentOrchestrator:
# Alle Ergebnisse zusammenführen
all_results = rss_articles + search_results + telegram_articles
# Pipeline-Schritt 2: Nachrichten sammeln (fertig)
try:
_delivering_sources = len({a.get("source", "") for a in all_results if a.get("source")})
except Exception:
_delivering_sources = None
await _pipe_done("collect", count_value=len(all_results), count_secondary=_delivering_sources)
# Pipeline-Schritt 3: Doppeltes filtern (Start)
await _pipe_start("dedup")
# Duplikate entfernen (normalisierte URL + Headline-Ähnlichkeit)
seen_urls = set()
@@ -922,6 +993,7 @@ class AgentOrchestrator:
dupes_removed = len(all_results) - len(unique_results)
if dupes_removed > 0:
logger.info(f"Deduplizierung: {dupes_removed} Duplikate entfernt, {len(unique_results)} verbleibend")
await _pipe_done("dedup", count_value=dupes_removed, count_secondary=len(unique_results))
# Relevanz-Scoring und Sortierung
for article in unique_results:
@@ -978,6 +1050,10 @@ class AgentOrchestrator:
new_candidates.append(article)
# Pipeline-Schritt 4: Relevanz bewerten (Start)
await _pipe_start("relevance")
_candidates_before_topic = len(new_candidates)
# --- Semantischer Topic-Filter (Haiku) ---
# Wirft Artikel raus, die zwar Keyword-Treffer hatten, aber das Kernthema
# der Lage nicht inhaltlich behandeln. Bei Fehler Fallback auf alle Kandidaten.
@@ -988,6 +1064,7 @@ class AgentOrchestrator:
)
if _tf_usage:
usage_acc.add(_tf_usage)
await _pipe_done("relevance", count_value=len(new_candidates), count_secondary=_candidates_before_topic)
# --- Neue (thematisch gefilterte) Artikel speichern und für Analyse tracken ---
new_count = 0
@@ -1019,6 +1096,8 @@ class AgentOrchestrator:
# Geoparsing: Orte aus neuen Artikeln extrahieren und speichern
if new_articles_for_analysis:
# Pipeline-Schritt 5: Orte erkennen (Start)
await _pipe_start("geoparsing")
try:
from agents.geoparsing import geoparse_articles
incident_context = f"{title} - {description}"
@@ -1049,8 +1128,12 @@ class AgentOrchestrator:
)
await db.commit()
logger.info(f"Category-Labels gespeichert fuer Incident {incident_id}: {category_labels}")
await _pipe_done("geoparsing", count_value=geo_count, count_secondary=len(geo_results) if geo_results else 0)
except Exception as e:
logger.warning(f"Geoparsing fehlgeschlagen (Pipeline laeuft weiter): {e}")
await _pipe_done("geoparsing", count_value=0, count_secondary=0)
else:
await _pipe_skip("geoparsing")
# Quellen-Statistiken aktualisieren
if new_count > 0:
@@ -1196,6 +1279,10 @@ class AgentOrchestrator:
articles_for_check = [dict(row) for row in await cursor.fetchall()]
return await factchecker.check(title, articles_for_check, incident_type)
# Pipeline-Schritte 6+7: Lagebild verfassen + Fakten prüfen (Start, parallel)
await _pipe_start("summary")
await _pipe_start("factcheck")
# Beide Tasks PARALLEL starten
logger.info("Starte Analyse und Faktencheck parallel...")
analysis_result, factcheck_result = await asyncio.gather(
@@ -1205,6 +1292,8 @@ class AgentOrchestrator:
analysis, analysis_usage = analysis_result
fact_checks, fc_usage = factcheck_result
# Pipeline-Schritt 6: Lagebild verfassen (fertig — keine Zahl, nur Status)
await _pipe_done("summary", count_value=None, count_secondary=None)
# --- Analyse-Ergebnisse verarbeiten ---
if analysis_usage:
@@ -1458,6 +1547,13 @@ class AgentOrchestrator:
await db.commit()
# Pipeline-Schritt 7: Fakten prüfen (fertig)
_new_facts_count = max(0, len(fact_checks) - len(existing_facts))
await _pipe_done("factcheck", count_value=_new_facts_count, count_secondary=len(fact_checks) if fact_checks else 0)
# Pipeline-Schritt 8: Qualitätscheck (Start, ohne Zahlen)
await _pipe_start("qc")
# Post-Refresh Quality Check: Duplikate und Karten-Kategorien pruefen
try:
from services.post_refresh_qc import run_post_refresh_qc
@@ -1469,6 +1565,12 @@ class AgentOrchestrator:
)
except Exception as qc_err:
logger.warning(f"Post-Refresh QC fehlgeschlagen: {qc_err}")
await _pipe_done("qc", count_value=None, count_secondary=None)
# Pipeline-Schritt 9: Benachrichtigen (Start)
await _pipe_start("notify")
_notify_count = 0
# Gebündelte Notification senden (nicht beim ersten Refresh)
if not is_first_refresh:
if self._ws_manager:
@@ -1525,6 +1627,32 @@ class AgentOrchestrator:
db, incident_id, title, visibility, created_by, tenant_id, db_notifications,
incident_type=incident_type,
)
_notify_count = len(db_notifications)
# Pipeline-Schritt 9: Benachrichtigen (fertig)
await _pipe_done("notify", count_value=_notify_count, count_secondary=None)
# Falls Analyse-Block uebersprungen wurde (kein neuer Artikel und Summary existiert),
# die noch offenen Pipeline-Schritte als uebersprungen markieren.
for _skipped_key in ("summary", "factcheck", "qc", "notify"):
if _skipped_key in _step_ids or _skipped_key not in {"summary", "factcheck", "qc", "notify"}:
pass
# Saubere Variante: alle noch offenen Steps am Ende skippen
for _open_key in list(_step_ids.keys()):
await _pipe_skip(_open_key)
# Auch Steps die nie gestartet wurden (bei uebersprungenem Outer-If)
_started_keys = set()
try:
_check_cursor = await db.execute(
"SELECT step_key FROM refresh_pipeline_steps WHERE refresh_log_id = ? AND pass_number = ?",
(log_id, _pass_nr),
)
_started_keys = {row[0] for row in await _check_cursor.fetchall()}
except Exception:
pass
for _missing_key in ("summary", "factcheck", "qc", "notify"):
if _missing_key not in _started_keys:
await _pipe_skip(_missing_key)
# Refresh-Log abschließen (mit Token-Statistiken)
await db.execute(