diff --git a/app/analyzer.py b/app/analyzer.py index cf76a4b..ef3d7aa 100644 --- a/app/analyzer.py +++ b/app/analyzer.py @@ -8,7 +8,12 @@ from openai import AsyncOpenAI from .config import settings from .models import Assessment -from .wahlprogramme import find_relevant_quotes, format_quote_for_prompt, WAHLPROGRAMME +from .bundeslaender import BUNDESLAENDER +from .wahlprogramme import ( + find_relevant_quotes, + format_quote_for_prompt, + WAHLPROGRAMM_KONTEXT_FILES, +) from .embeddings import get_relevant_quotes_for_antrag, format_quotes_for_prompt, EMBEDDINGS_DB # Load context files @@ -144,32 +149,52 @@ Antworte NUR mit einem JSON-Objekt im folgenden Format (keine Markdown-Codeblöc def get_bundesland_context(bundesland: str) -> str: - """Get context for a specific state.""" - contexts = { - "NRW": { - "wahlprogramme": "wahlprogramme-nrw-2022.md", - "parteiprogramme": "parteiprogramme.md", - "regierungsfraktionen": ["CDU", "GRÜNE"], - } - } - - ctx = contexts.get(bundesland, contexts["NRW"]) - - wahlprogramme = load_context_file(ctx["wahlprogramme"]) - parteiprogramme = load_context_file(ctx["parteiprogramme"]) - - return f""" -## Wahlprogramme {bundesland} 2022 + """Build the LLM context block for a specific state. -{wahlprogramme} + Liest Regierungsfraktionen und Parlamentsname aus ``BUNDESLAENDER`` und + die optionale Wahlprogramm-Übersichtsdatei aus ``WAHLPROGRAMM_KONTEXT_FILES``. + Federal-level Grundsatzprogramme (parteiprogramme.md) sind bundesländer- + übergreifend. + + Raises: + ValueError: bei unbekanntem oder inaktivem Bundesland. Pre-#5 + existierte hier ein silent fallback auf NRW — bewusst entfernt, + damit Konfigurationslücken früh sichtbar werden. + """ + bl = BUNDESLAENDER.get(bundesland) + if bl is None: + raise ValueError(f"Unbekanntes Bundesland: {bundesland}") + if not bl.aktiv: + raise ValueError( + f"Bundesland {bundesland} ist nicht aktiv (siehe bundeslaender.py)" + ) + + wahlprogramm_kontext_file = WAHLPROGRAMM_KONTEXT_FILES.get(bundesland) + wahlprogramme_text = ( + load_context_file(wahlprogramm_kontext_file) if wahlprogramm_kontext_file else "" + ) + parteiprogramme_text = load_context_file("parteiprogramme.md") + + return f""" +## Parlament + +{bl.parlament_name} (Wahlperiode {bl.wahlperiode}, seit {bl.wahlperiode_start}) + +## Wahlprogramme {bl.name} + +{wahlprogramme_text or '(keine Übersichtsdatei hinterlegt)'} ## Grundsatzprogramme der Parteien -{parteiprogramme} +{parteiprogramme_text} -## Regierungsfraktionen in {bundesland} +## Regierungsfraktionen in {bl.name} -{', '.join(ctx['regierungsfraktionen'])} +{', '.join(bl.regierungsfraktionen)} + +## Im Landtag vertretene Fraktionen + +{', '.join(bl.landtagsfraktionen)} Bei Oppositionsanträgen: Bewerte zusätzlich, ob die Regierungsfraktionen zustimmen würden. """ @@ -185,26 +210,34 @@ async def analyze_antrag(text: str, bundesland: str = "NRW", model: str = "qwen- system_prompt = get_system_prompt() bundesland_context = get_bundesland_context(bundesland) - - # Extrahiere Fraktionen aus Text (einfache Heuristik) - fraktionen = [] - for partei in WAHLPROGRAMME.keys(): - if partei in text or partei.lower() in text.lower(): - fraktionen.append(partei) - + + # Extrahiere Fraktionen aus Text (einfache Heuristik): Welche der im + # Landtag vertretenen Parteien werden im Antrag genannt? Quelle ist + # BUNDESLAENDER.landtagsfraktionen — nicht WAHLPROGRAMME, weil wir + # auch Fraktionen erkennen wollen, für die wir (noch) kein Wahlprogramm + # hinterlegt haben. + landtagsfraktionen = BUNDESLAENDER[bundesland].landtagsfraktionen + text_lower = text.lower() + fraktionen = [ + partei for partei in landtagsfraktionen + if partei in text or partei.lower() in text_lower + ] + # Suche relevante Zitate via semantische Suche (Embeddings) quotes_context = "" if EMBEDDINGS_DB.exists(): try: - semantic_quotes = get_relevant_quotes_for_antrag(text, fraktionen, top_k_per_partei=2) + semantic_quotes = get_relevant_quotes_for_antrag( + text, fraktionen, bundesland=bundesland, top_k_per_partei=2, + ) quotes_context = format_quotes_for_prompt(semantic_quotes) except Exception as e: print(f"Semantic search failed: {e}, falling back to keyword search") - quotes = find_relevant_quotes(text, fraktionen) + quotes = find_relevant_quotes(text, fraktionen, bundesland=bundesland) quotes_context = format_quote_for_prompt(quotes) else: # Fallback to keyword search - quotes = find_relevant_quotes(text, fraktionen) + quotes = find_relevant_quotes(text, fraktionen, bundesland=bundesland) quotes_context = format_quote_for_prompt(quotes) user_prompt = f"""Analysiere den folgenden Antrag: diff --git a/app/embeddings.py b/app/embeddings.py index 2fea310..f2914cf 100644 --- a/app/embeddings.py +++ b/app/embeddings.py @@ -84,7 +84,14 @@ PROGRAMME = { def init_embeddings_db(): - """Initialize the embeddings database.""" + """Initialize the embeddings database. + + Includes a forward-only migration step (Issue #5): adds the + ``bundesland`` column if missing and backfills existing rows from the + ``PROGRAMME`` registry. Grundsatzprogramme (federal level) keep + ``bundesland = NULL``; the ``find_relevant_chunks`` query treats NULL + as "matches any state". + """ conn = sqlite3.connect(EMBEDDINGS_DB) conn.execute(""" CREATE TABLE IF NOT EXISTS chunks ( @@ -100,6 +107,23 @@ def init_embeddings_db(): """) conn.execute("CREATE INDEX IF NOT EXISTS idx_chunks_partei ON chunks(partei)") conn.execute("CREATE INDEX IF NOT EXISTS idx_chunks_typ ON chunks(typ)") + + # Migration: bundesland-Spalte ergänzen, falls Tabelle aus Pre-#5-Zeit + cols = {row[1] for row in conn.execute("PRAGMA table_info(chunks)").fetchall()} + if "bundesland" not in cols: + conn.execute("ALTER TABLE chunks ADD COLUMN bundesland TEXT") + conn.execute("CREATE INDEX IF NOT EXISTS idx_chunks_bundesland ON chunks(bundesland)") + + # Backfill: Bundesland aus PROGRAMME-Registry für bestehende Zeilen + # nachtragen. Grundsatzprogramme bleiben NULL. + for prog_id, info in PROGRAMME.items(): + bl = info.get("bundesland") + if bl is not None: + conn.execute( + "UPDATE chunks SET bundesland = ? WHERE programm_id = ? AND bundesland IS NULL", + (bl, prog_id), + ) + conn.commit() conn.close() @@ -187,8 +211,8 @@ def index_programm(programm_id: str, pdf_dir: Path) -> int: embedding_blob = json.dumps(embedding).encode() conn.execute(""" - INSERT INTO chunks (programm_id, partei, typ, seite, text, embedding) - VALUES (?, ?, ?, ?, ?, ?) + INSERT INTO chunks (programm_id, partei, typ, seite, text, embedding, bundesland) + VALUES (?, ?, ?, ?, ?, ?, ?) """, ( programm_id, info["partei"], @@ -196,6 +220,7 @@ def index_programm(programm_id: str, pdf_dir: Path) -> int: page_num, chunk_text_content, embedding_blob, + info.get("bundesland"), # NULL für Grundsatzprogramme )) total_chunks += 1 except Exception as e: @@ -223,29 +248,41 @@ def find_relevant_chunks( query: str, parteien: list[str] = None, typ: str = None, + bundesland: str = None, top_k: int = 3, min_similarity: float = 0.5, ) -> list[dict]: - """Find most relevant chunks for a query.""" - + """Find most relevant chunks for a query. + + Args: + bundesland: Wenn gesetzt, werden nur Chunks dieses Bundeslands ODER + globale Chunks (bundesland IS NULL, z.B. Grundsatzprogramme) + berücksichtigt. Wenn None, kein Filter. + """ + query_embedding = create_embedding(query) - + conn = sqlite3.connect(EMBEDDINGS_DB) conn.row_factory = sqlite3.Row - + # Build query sql = "SELECT * FROM chunks WHERE 1=1" params = [] - + if parteien: placeholders = ",".join("?" * len(parteien)) sql += f" AND partei IN ({placeholders})" params.extend(parteien) - + if typ: sql += " AND typ = ?" params.append(typ) - + + if bundesland: + # Bundesland-spezifische ODER globale Chunks (Grundsatzprogramme). + sql += " AND (bundesland = ? OR bundesland IS NULL)" + params.append(bundesland) + rows = conn.execute(sql, params).fetchall() conn.close() @@ -273,39 +310,57 @@ def find_relevant_chunks( def get_relevant_quotes_for_antrag( antrag_text: str, fraktionen: list[str], + bundesland: str, top_k_per_partei: int = 2, ) -> dict[str, list[dict]]: - """Get relevant quotes from Wahl- and Parteiprogramme for an Antrag.""" - + """Get relevant quotes from Wahl- and Parteiprogramme for an Antrag. + + Args: + bundesland: Pflicht. Bestimmt, welche Wahlprogramme durchsucht werden + und welche Regierungsfraktionen zusätzlich zu den Antragstellern + einbezogen werden. + """ + # Lokaler Import vermeidet Zirkularität: bundeslaender.py importiert nichts + # aus diesem Modul, aber der saubere Trennstrich bleibt erhalten. + from .bundeslaender import BUNDESLAENDER + + if bundesland not in BUNDESLAENDER: + raise ValueError(f"Unbekanntes Bundesland: {bundesland}") + + regierungsfraktionen = BUNDESLAENDER[bundesland].regierungsfraktionen + parteien_to_search = list(dict.fromkeys(fraktionen + regierungsfraktionen)) # dedupe, Reihenfolge stabil + results = {} - - for partei in fraktionen + ["CDU", "GRÜNE"]: # Include Regierungsfraktionen + + for partei in parteien_to_search: partei_upper = partei.upper() if partei != "GRÜNE" else "GRÜNE" - - # Wahlprogramm + + # Wahlprogramm — bundesland-gefiltert wahl_chunks = find_relevant_chunks( antrag_text, parteien=[partei_upper], typ="wahlprogramm", + bundesland=bundesland, top_k=top_k_per_partei, min_similarity=0.45, ) - - # Parteiprogramm + + # Parteiprogramm (Grundsatz, federal — bundesland=NULL matched implizit) partei_chunks = find_relevant_chunks( antrag_text, parteien=[partei_upper], typ="parteiprogramm", + bundesland=bundesland, top_k=top_k_per_partei, min_similarity=0.45, ) - + if wahl_chunks or partei_chunks: results[partei_upper] = { "wahlprogramm": wahl_chunks, "parteiprogramm": partei_chunks, } - + return results @@ -320,7 +375,7 @@ def format_quotes_for_prompt(quotes: dict) -> str: lines.append(f"\n### {partei}\n") if data.get("wahlprogramm"): - lines.append("**Wahlprogramm NRW 2022:**") + lines.append("**Wahlprogramm:**") for chunk in data["wahlprogramm"]: text = chunk["text"][:500] + "..." if len(chunk["text"]) > 500 else chunk["text"] lines.append(f'- S. {chunk["seite"]}: "{text}"') diff --git a/app/wahlprogramme.py b/app/wahlprogramme.py index 55a35fa..68bc051 100644 --- a/app/wahlprogramme.py +++ b/app/wahlprogramme.py @@ -1,126 +1,162 @@ -"""Wahlprogramm-Referenzsystem mit Zitaten und Seitenreferenzen.""" +"""Wahlprogramm-Referenzsystem mit Zitaten und Seitenreferenzen. + +Bundesland-bewusst seit Issue #5: ``WAHLPROGRAMME[bundesland][partei]`` statt +flach. Konsumiert ``BUNDESLAENDER`` aus ``bundeslaender.py`` für die +Regierungsfraktionen-Lookup und für Plausibilitätsprüfungen. + +Verantwortlich für die schlüsselwortbasierte Fallback-Suche in den +paged-Textversionen der Wahlprogramme. Die semantische Suche lebt in +``embeddings.py``. +""" -import json import re from pathlib import Path from typing import Optional -# Wahlprogramm-Metadaten -WAHLPROGRAMME = { - "CDU": { - "file": "cdu-nrw-2022.pdf", - "titel": "Machen, worauf es ankommt", - "partei": "CDU NRW", - "jahr": 2022, - "seiten": 109, - }, - "SPD": { - "file": "spd-nrw-2022.pdf", - "titel": "Unser Land von morgen", - "partei": "SPD NRW", - "jahr": 2022, - "seiten": 116, - }, - "GRÜNE": { - "file": "gruene-nrw-2022.pdf", - "titel": "Von hier an Zukunft", - "partei": "BÜNDNIS 90/DIE GRÜNEN NRW", - "jahr": 2022, - "seiten": 100, - }, - "FDP": { - "file": "fdp-nrw-2022.pdf", - "titel": "Nie gab es mehr zu tun", - "partei": "FDP NRW", - "jahr": 2022, - "seiten": 96, - }, - "AfD": { - "file": "afd-nrw-2022.pdf", - "titel": "Wer sonst.", - "partei": "AfD NRW", - "jahr": 2022, - "seiten": 68, +from .bundeslaender import BUNDESLAENDER + + +# WAHLPROGRAMME[bundesland][partei] -> Metadaten +# Beim Hinzufügen eines neuen Bundeslands: Eintrag hier UND parallel +# in WAHLPROGRAMM_KONTEXT_FILES. +WAHLPROGRAMME: dict[str, dict[str, dict]] = { + "NRW": { + "CDU": { + "file": "cdu-nrw-2022.pdf", + "titel": "Machen, worauf es ankommt", + "partei": "CDU NRW", + "jahr": 2022, + "seiten": 109, + }, + "SPD": { + "file": "spd-nrw-2022.pdf", + "titel": "Unser Land von morgen", + "partei": "SPD NRW", + "jahr": 2022, + "seiten": 116, + }, + "GRÜNE": { + "file": "gruene-nrw-2022.pdf", + "titel": "Von hier an Zukunft", + "partei": "BÜNDNIS 90/DIE GRÜNEN NRW", + "jahr": 2022, + "seiten": 100, + }, + "FDP": { + "file": "fdp-nrw-2022.pdf", + "titel": "Nie gab es mehr zu tun", + "partei": "FDP NRW", + "jahr": 2022, + "seiten": 96, + }, + "AfD": { + "file": "afd-nrw-2022.pdf", + "titel": "Wer sonst.", + "partei": "AfD NRW", + "jahr": 2022, + "seiten": 68, + }, }, } -# Basis-Pfad für Referenzdokumente +# Pro Bundesland: Markdown-Übersichtsdatei mit Wahlprogramm-Zusammenfassungen, +# wird als Kontext in den LLM-Prompt geladen (nicht für die Suche). +WAHLPROGRAMM_KONTEXT_FILES: dict[str, str] = { + "NRW": "wahlprogramme-nrw-2022.md", +} + REFERENZEN_PATH = Path(__file__).parent / "static" / "referenzen" KONTEXT_PATH = Path(__file__).parent / "kontext" -def load_wahlprogramm_text(partei: str) -> dict[int, str]: +def get_wahlprogramm(bundesland: str, partei: str) -> Optional[dict]: + """Liefert die Wahlprogramm-Metadaten oder None, wenn keins vorliegt.""" + return WAHLPROGRAMME.get(bundesland, {}).get(partei) + + +def parteien_mit_wahlprogramm(bundesland: str) -> list[str]: + """Liste der Parteien, für die im gegebenen Bundesland ein Wahlprogramm vorliegt.""" + return list(WAHLPROGRAMME.get(bundesland, {}).keys()) + + +def load_wahlprogramm_text(bundesland: str, partei: str) -> dict[int, str]: """Lädt Wahlprogramm-Text mit Seitenzuordnung. - + Returns: - Dict mit Seitennummer -> Text + Dict mit Seitennummer -> Text. Leer, wenn kein Wahlprogramm hinterlegt + oder die paged-Textdatei fehlt. """ - if partei not in WAHLPROGRAMME: + info = get_wahlprogramm(bundesland, partei) + if not info: return {} - + # Versuche paged-Textdatei zu laden - paged_file = KONTEXT_PATH / f"{WAHLPROGRAMME[partei]['file'].replace('.pdf', '-paged.txt')}" + paged_file = KONTEXT_PATH / info['file'].replace('.pdf', '-paged.txt') if not paged_file.exists(): # Fallback: Normale Textdatei - txt_file = KONTEXT_PATH / f"{WAHLPROGRAMME[partei]['file'].replace('.pdf', '.txt')}" + txt_file = KONTEXT_PATH / info['file'].replace('.pdf', '.txt') if txt_file.exists(): return {1: txt_file.read_text()} return {} - + text = paged_file.read_text() pages = {} current_page = 1 current_text = [] - + for line in text.split('\n'): if line.startswith('--- PAGE '): - # Speichere vorherige Seite if current_text: pages[current_page] = '\n'.join(current_text) - # Extrahiere neue Seitenzahl match = re.search(r'PAGE (\d+)', line) if match: current_page = int(match.group(1)) current_text = [] else: current_text.append(line) - - # Letzte Seite speichern + if current_text: pages[current_page] = '\n'.join(current_text) - + return pages -def search_wahlprogramm(partei: str, keywords: list[str], max_results: int = 3) -> list[dict]: +def search_wahlprogramm( + bundesland: str, + partei: str, + keywords: list[str], + max_results: int = 3, +) -> list[dict]: """Sucht relevante Passagen in einem Wahlprogramm. - + Args: - partei: Partei-Kürzel (CDU, SPD, GRÜNE, FDP, AfD) + bundesland: Bundesland-Code (NRW, LSA, …) + partei: Partei-Kürzel (CDU, SPD, GRÜNE, FDP, AfD, …) keywords: Suchbegriffe max_results: Maximale Anzahl Ergebnisse - + Returns: - Liste von {seite, text, score, url} + Liste von {bundesland, partei, seite, text, score, url, quelle} """ - pages = load_wahlprogramm_text(partei) + info = get_wahlprogramm(bundesland, partei) + if not info: + return [] + + pages = load_wahlprogramm_text(bundesland, partei) if not pages: return [] - + results = [] keywords_lower = [k.lower() for k in keywords] - + for page_num, text in pages.items(): text_lower = text.lower() - - # Zähle Keyword-Treffer score = sum(1 for kw in keywords_lower if kw in text_lower) - + if score > 0: - # Finde relevante Absätze (mit Keyword) paragraphs = text.split('\n\n') relevant_paragraphs = [] - + for para in paragraphs: para_clean = para.strip() if len(para_clean) < 50: @@ -128,72 +164,79 @@ def search_wahlprogramm(partei: str, keywords: list[str], max_results: int = 3) para_lower = para_clean.lower() if any(kw in para_lower for kw in keywords_lower): relevant_paragraphs.append(para_clean) - + if relevant_paragraphs: - # Nimm den relevantesten Absatz (mit meisten Keywords) - best_para = max(relevant_paragraphs, - key=lambda p: sum(1 for kw in keywords_lower if kw in p.lower())) - - # Kürze auf ~300 Zeichen + best_para = max( + relevant_paragraphs, + key=lambda p: sum(1 for kw in keywords_lower if kw in p.lower()), + ) if len(best_para) > 300: best_para = best_para[:297] + "..." - + results.append({ "partei": partei, + "bundesland": bundesland, "seite": page_num, "text": best_para, "score": score, - "url": f"/static/referenzen/{WAHLPROGRAMME[partei]['file']}#page={page_num}", - "quelle": f"{WAHLPROGRAMME[partei]['partei']} Wahlprogramm {WAHLPROGRAMME[partei]['jahr']}, S. {page_num}" + "url": f"/static/referenzen/{info['file']}#page={page_num}", + "quelle": f"{info['partei']} Wahlprogramm {info['jahr']}, S. {page_num}", }) - - # Sortiere nach Score, nimm Top-Ergebnisse + results.sort(key=lambda x: x['score'], reverse=True) return results[:max_results] -def find_relevant_quotes(antrag_text: str, fraktionen: list[str]) -> dict[str, list[dict]]: +def find_relevant_quotes( + antrag_text: str, + fraktionen: list[str], + bundesland: str, +) -> dict[str, list[dict]]: """Findet relevante Zitate aus Wahlprogrammen für einen Antrag. - + Args: antrag_text: Volltext des Antrags - fraktionen: Liste der Fraktionen (Antragsteller + Regierung) - + fraktionen: Liste der einreichenden Fraktionen + bundesland: Bundesland-Code (Pflichtparameter; bestimmt, welche + Wahlprogramme durchsucht werden und welche Regierungsfraktionen + zusätzlich einbezogen werden). + Returns: Dict mit Partei -> Liste von Zitaten """ + if bundesland not in BUNDESLAENDER: + raise ValueError(f"Unbekanntes Bundesland: {bundesland}") + # Extrahiere Keywords aus Antrag (einfache Heuristik) - # Entferne Stoppwörter und kurze Wörter - stopwords = {'der', 'die', 'das', 'und', 'oder', 'für', 'mit', 'von', 'zu', 'auf', - 'ist', 'sind', 'wird', 'werden', 'hat', 'haben', 'ein', 'eine', 'einer', - 'den', 'dem', 'des', 'im', 'in', 'an', 'bei', 'nach', 'über', 'unter', - 'durch', 'als', 'auch', 'nur', 'noch', 'aber', 'wenn', 'dass', 'sich', - 'nicht', 'wie', 'so', 'aus', 'zum', 'zur', 'vom', 'beim', 'seit', 'bis'} - + stopwords = { + 'der', 'die', 'das', 'und', 'oder', 'für', 'mit', 'von', 'zu', 'auf', + 'ist', 'sind', 'wird', 'werden', 'hat', 'haben', 'ein', 'eine', 'einer', + 'den', 'dem', 'des', 'im', 'in', 'an', 'bei', 'nach', 'über', 'unter', + 'durch', 'als', 'auch', 'nur', 'noch', 'aber', 'wenn', 'dass', 'sich', + 'nicht', 'wie', 'so', 'aus', 'zum', 'zur', 'vom', 'beim', 'seit', 'bis', + } + words = re.findall(r'\b[A-Za-zäöüÄÖÜß]{4,}\b', antrag_text) keywords = [w for w in words if w.lower() not in stopwords] - - # Zähle Worthäufigkeit - word_freq = {} + + word_freq: dict[str, int] = {} for w in keywords: w_lower = w.lower() word_freq[w_lower] = word_freq.get(w_lower, 0) + 1 - - # Top-Keywords (häufigste) + top_keywords = sorted(word_freq.keys(), key=lambda x: word_freq[x], reverse=True)[:15] - - # Suche in relevanten Wahlprogrammen - quotes = {} - - # Immer Regierungsfraktionen einbeziehen - parteien_to_search = set(fraktionen) | {"CDU", "GRÜNE"} - + + # Antragsteller + Regierungsfraktionen des Bundeslands + regierungsfraktionen = BUNDESLAENDER[bundesland].regierungsfraktionen + parteien_to_search = set(fraktionen) | set(regierungsfraktionen) + + quotes: dict[str, list[dict]] = {} for partei in parteien_to_search: - if partei in WAHLPROGRAMME: - found = search_wahlprogramm(partei, top_keywords, max_results=2) + if get_wahlprogramm(bundesland, partei): + found = search_wahlprogramm(bundesland, partei, top_keywords, max_results=2) if found: quotes[partei] = found - + return quotes @@ -201,14 +244,14 @@ def format_quote_for_prompt(quotes: dict[str, list[dict]]) -> str: """Formatiert Zitate für den LLM-Prompt.""" if not quotes: return "" - + lines = ["\n## Relevante Passagen aus Wahlprogrammen\n"] lines.append("Nutze diese Originalzitate als Belege in deiner Bewertung:\n") - + for partei, zitate in quotes.items(): for z in zitate: lines.append(f"### {z['quelle']}") lines.append(f'> "{z["text"]}"') lines.append("") - + return "\n".join(lines)