#!/usr/bin/env python3 """ Intel Arc-Optimized Legislative Analysis System Uses DuckDB + VSS for fast context injection into LLMs Hardware: Intel Core Ultra 7 165H with Arc Graphics + NPU + 64GB RAM Features: - Fast DuckDB queries for legislative history - Vector similarity search for relevant testimony - Intel-optimized inference (IPEX-LLM or OpenVINO) - Structured extraction: interest groups, lobbyists, positions, tradeoffs Author: CommunityOne Date: 2026-04-30 """ import os import json from pathlib import Path from typing import List, Dict, Any, Optional from dataclasses import dataclass, asdict import duckdb from loguru import logger import sys # Configure logging logger.remove() logger.add(sys.stderr, level="INFO") # Paths PROJECT_ROOT = Path(__file__).parent.parent.parent # Go up to project root from scripts/enrichment_ai/ DATA_DIR = PROJECT_ROOT / "data" GOLD_DIR = DATA_DIR / "gold" ANALYSIS_DIR = DATA_DIR / "gold" / "analysis" # Store analysis results here DUCKDB_PATH = DATA_DIR / "legislative.duckdb" # Ensure output directories exist ANALYSIS_DIR.mkdir(parents=True, exist_ok=True) @dataclass class InterestGroup: """Structured schema for interest group extraction""" group_name: str lobbyist: Optional[str] stance: str # support, oppose, neutral, conditional stance_score: float # -1.0 (oppose) to +1.0 (support) tradeoff_notes: Optional[str] testimony_excerpt: str bill_id: str confidence: float # 0.0 to 1.0 def to_dict(self) -> Dict[str, Any]: return asdict(self) class DuckDBLegislativeAnalyzer: """ DuckDB-powered legislative analysis optimized for Intel Arc Why DuckDB? - 10-100x faster than Postgres for analytical queries - Native Parquet support (your Hugging Face datasets) - Embedded (no server needed) - Fast context injection for LLMs (thousands of rows in <100ms) """ def __init__(self, db_path: Path = DUCKDB_PATH): self.db_path = db_path self.conn: Optional[duckdb.DuckDBPyConnection] = None def __enter__(self): self.connect() return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() def connect(self): """Connect to DuckDB and install extensions""" logger.info(f"šŸ“Š Connecting to DuckDB: {self.db_path}") self.conn = duckdb.connect(str(self.db_path)) # Install VSS extension for vector similarity search try: self.conn.execute("INSTALL vss") self.conn.execute("LOAD vss") logger.info("āœ… VSS extension loaded") except Exception as e: logger.warning(f"āš ļø VSS extension not available: {e}") # Install Parquet extension self.conn.execute("INSTALL parquet") self.conn.execute("LOAD parquet") logger.info("āœ… Parquet extension loaded") def close(self): """Close connection""" if self.conn: self.conn.close() logger.info("šŸ”Œ DuckDB connection closed") def create_bills_table(self): """Create bills table from Parquet files""" logger.info("šŸ“‹ Creating bills table...") # Read from OpenStates bulk data if available bills_parquet = DATA_DIR / "gold" / "bills_bills.parquet" if not bills_parquet.exists(): logger.warning(f"āš ļø Bills parquet not found: {bills_parquet}") logger.info(" Creating demo bills table instead...") # Create demo table with sample data self.conn.execute(""" CREATE TABLE IF NOT EXISTS bills ( identifier VARCHAR, title TEXT, abstract TEXT, classification VARCHAR, subject VARCHAR, from_organization_name VARCHAR, from_organization_state VARCHAR(2), updated_at TIMESTAMP ) """) # Insert demo data demo_bills = [ ('HB1234', 'Water Fluoridation Act', 'Requires community water fluoridation', 'bill', 'Health', 'Alabama House', 'AL', '2026-04-01'), ('SB5678', 'Dental Care Access', 'Expands dental coverage for children', 'bill', 'Health', 'Georgia Senate', 'GA', '2026-04-15'), ('HB9012', 'School Health Programs', 'Funds oral health screenings in schools', 'bill', 'Education', 'Massachusetts House', 'MA', '2026-03-20'), ] self.conn.executemany(""" INSERT INTO bills VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, demo_bills) logger.info("āœ… Demo bills table created (3 sample bills)") return # Create table directly from Parquet self.conn.execute(f""" CREATE OR REPLACE TABLE bills AS SELECT * FROM read_parquet('{bills_parquet}') """) logger.info("āœ… Bills table created") def create_testimony_table(self): """Create testimony table with vector embeddings""" logger.info("šŸ“ Creating testimony table...") # This would be populated from meeting transcripts self.conn.execute(""" CREATE TABLE IF NOT EXISTS testimony ( id INTEGER PRIMARY KEY, bill_id VARCHAR, speaker_name VARCHAR, organization VARCHAR, testimony_text TEXT, stance VARCHAR, -- support, oppose, neutral timestamp TIMESTAMP, embedding FLOAT[384] -- Sentence transformer embeddings ) """) logger.info("āœ… Testimony table created") def create_vector_index(self): """Create HNSW index for fast vector similarity search""" try: self.conn.execute(""" CREATE INDEX IF NOT EXISTS testimony_vss_idx ON testimony USING HNSW (embedding) """) logger.info("āœ… Vector index created (HNSW)") except Exception as e: logger.warning(f"āš ļø Vector index creation failed: {e}") def search_similar_testimony( self, query_embedding: List[float], limit: int = 10 ) -> List[Dict[str, Any]]: """ Fast vector similarity search using VSS extension This is 100-1000x faster than computing similarity in Python """ try: result = self.conn.execute(f""" SELECT id, bill_id, speaker_name, organization, testimony_text, stance, array_distance(embedding, ?::FLOAT[384]) as distance FROM testimony ORDER BY distance ASC LIMIT {limit} """, [query_embedding]).fetchall() return [ { 'id': row[0], 'bill_id': row[1], 'speaker': row[2], 'organization': row[3], 'text': row[4], 'stance': row[5], 'similarity': 1.0 - row[6] # Convert distance to similarity } for row in result ] except Exception as e: logger.error(f"āŒ Vector search failed: {e}") return [] def get_bill_context(self, bill_id: str) -> Dict[str, Any]: """ Fast context retrieval for LLM injection On Intel Arc + 64GB RAM, this can pull 100+ page bills in <50ms """ result = self.conn.execute(""" SELECT identifier, title, abstract, classification, subject, from_organization_name, updated_at FROM bills WHERE identifier = ? """, [bill_id]).fetchone() if not result: return {} return { 'id': result[0], 'title': result[1], 'abstract': result[2], 'classification': result[3], 'subject': result[4], 'sponsor': result[5], 'updated': result[6] } def get_all_testimony_for_bill(self, bill_id: str) -> List[Dict[str, Any]]: """Get all testimony for a bill (for full context window)""" result = self.conn.execute(""" SELECT speaker_name, organization, testimony_text, stance, timestamp FROM testimony WHERE bill_id = ? ORDER BY timestamp """, [bill_id]).fetchall() return [ { 'speaker': row[0], 'organization': row[1], 'text': row[2], 'stance': row[3], 'timestamp': row[4] } for row in result ] def is_bill_analyzed(self, bill_id: str) -> bool: """ Check if a bill has already been analyzed (incremental processing) Returns: True if analysis exists in Parquet, False otherwise """ analysis_file = ANALYSIS_DIR / "interest_groups_analysis.parquet" if not analysis_file.exists(): return False try: result = self.conn.execute(f""" SELECT COUNT(*) FROM read_parquet('{analysis_file}') WHERE bill_id = ? """, [bill_id]).fetchone() return result[0] > 0 if result else False except: return False def save_analysis_results( self, results: List[InterestGroup], append: bool = True ) -> Path: """ Save analysis results to Parquet file (proper data pipeline!) Args: results: List of InterestGroup analysis results append: If True, append to existing file; if False, overwrite Returns: Path to saved Parquet file """ if not results: logger.warning("No results to save") return None import pandas as pd # Convert to DataFrame df = pd.DataFrame([r.to_dict() for r in results]) # Add metadata df['analyzed_at'] = pd.Timestamp.now() df['model'] = 'llama-3.2-3b' # or from config output_file = ANALYSIS_DIR / "interest_groups_analysis.parquet" if append and output_file.exists(): # Append to existing Parquet logger.info(f"šŸ“Š Appending {len(results)} results to {output_file}") # Read existing existing_df = pd.read_parquet(output_file) # Remove duplicates (keep newest) df_combined = pd.concat([existing_df, df], ignore_index=True) df_combined = df_combined.drop_duplicates( subset=['bill_id', 'group_name'], keep='last' ) # Save df_combined.to_parquet(output_file, index=False) logger.info(f"āœ… Saved {len(df_combined)} total records") else: # Create new file logger.info(f"šŸ“Š Saving {len(results)} results to {output_file}") df.to_parquet(output_file, index=False) logger.info(f"āœ… Created new analysis file") return output_file def get_bills_to_analyze( self, state: Optional[str] = None, topic_filter: Optional[str] = None, limit: int = 100, skip_analyzed: bool = True ) -> List[Dict[str, Any]]: """ Get bills that need analysis (incremental processing support) Args: state: Filter by state code (e.g., 'GA', 'AL') topic_filter: Search term in title (e.g., 'fluorid') limit: Maximum bills to return skip_analyzed: Skip bills already in analysis Parquet Returns: List of bill dicts ready for analysis """ # Build query where_clauses = [] params = [] if state: where_clauses.append("state = ?") params.append(state) if topic_filter: where_clauses.append("LOWER(title) LIKE ?") params.append(f"%{topic_filter.lower()}%") where_sql = " AND ".join(where_clauses) if where_clauses else "1=1" # Get bills query = f""" SELECT bill_id, bill_number, title, abstract, state, jurisdiction_name FROM bills WHERE {where_sql} LIMIT ? """ params.append(limit * 2) # Get extra in case we filter some out bills = self.conn.execute(query, params).fetchall() # Convert to dicts result = [] for row in bills: bill = { 'bill_id': row[0], 'bill_number': row[1], 'title': row[2], 'abstract': row[3], 'state': row[4], 'jurisdiction': row[5] } # Skip if already analyzed (incremental!) if skip_analyzed and self.is_bill_analyzed(bill['bill_id']): logger.debug(f"Skipping {bill['bill_number']} - already analyzed") continue result.append(bill) if len(result) >= limit: break return result def analyze_bill_statistics(self): """Fast analytical queries on bill data""" stats = {} # Check if bills table exists tables = self.conn.execute(""" SELECT table_name FROM information_schema.tables WHERE table_schema = 'main' AND table_name = 'bills' """).fetchall() if not tables: logger.warning("āš ļø Bills table not found, skipping statistics") return {'top_states': [], 'top_topics': []} # Check what columns exist columns = self.conn.execute("DESCRIBE bills").fetchall() col_names = [col[0] for col in columns] # Adapt query based on available columns if 'state' in col_names and 'topic' in col_names and 'total_bills' in col_names: # This is bill_map_aggregate format (aggregated data) logger.info(" Using aggregated bills format (bill_map_aggregate)") # Bills by state result = self.conn.execute(""" SELECT state, SUM(total_bills) as count FROM bills WHERE state IS NOT NULL GROUP BY state ORDER BY count DESC LIMIT 10 """).fetchall() stats['top_states'] = [{'state': r[0], 'count': r[1]} for r in result] # Bills by topic result = self.conn.execute(""" SELECT topic, SUM(total_bills) as count FROM bills WHERE topic IS NOT NULL GROUP BY topic ORDER BY count DESC LIMIT 10 """).fetchall() stats['top_topics'] = [{'topic': r[0], 'count': r[1]} for r in result] elif 'state' in col_names and 'jurisdiction_name' in col_names: # This is OpenStates bills format (current data) logger.info(" Using OpenStates bills format") # Bills by state result = self.conn.execute(""" SELECT state, jurisdiction_name, COUNT(*) as count FROM bills WHERE state IS NOT NULL GROUP BY state, jurisdiction_name ORDER BY count DESC LIMIT 10 """).fetchall() stats['top_states'] = [{'state': r[0], 'jurisdiction': r[1], 'count': r[2]} for r in result] # Bills by session result = self.conn.execute(""" SELECT session_name, COUNT(*) as count FROM bills WHERE session_name IS NOT NULL GROUP BY session_name ORDER BY count DESC LIMIT 10 """).fetchall() stats['top_sessions'] = [{'session': r[0], 'count': r[1]} for r in result] elif 'from_organization_state' in col_names and 'subject' in col_names: # This is individual bills format (OpenStates schema) logger.info(" Using individual bills format (OpenStates schema)") # Bills by state result = self.conn.execute(""" SELECT from_organization_state, COUNT(*) as count FROM bills WHERE from_organization_state IS NOT NULL GROUP BY from_organization_state ORDER BY count DESC LIMIT 10 """).fetchall() stats['top_states'] = [{'state': r[0], 'count': r[1]} for r in result] # Bills by subject result = self.conn.execute(""" SELECT subject, COUNT(*) as count FROM bills WHERE subject IS NOT NULL GROUP BY subject ORDER BY count DESC LIMIT 10 """).fetchall() stats['top_subjects'] = [{'subject': r[0], 'count': r[1]} for r in result] else: logger.warning(f"āš ļø Unknown bills table schema, columns: {col_names[:5]}") return {'top_states': [], 'top_topics': []} return stats class IntelOptimizedLLM: """ Intel Arc-optimized LLM inference Uses IPEX-LLM or OpenVINO for maximum performance on Arc GPU + NPU """ def __init__(self, model_name: str = "meta-llama/Llama-3.2-3B-Instruct"): self.model_name = model_name self.model = None self.tokenizer = None # Detect Intel hardware self.has_arc = self._detect_arc_gpu() logger.info(f"šŸŽ® Intel Arc GPU detected: {self.has_arc}") def _detect_arc_gpu(self) -> bool: """Detect Intel Arc graphics via XPU availability""" try: import torch # Check if XPU module exists and has devices if hasattr(torch, 'xpu'): try: return torch.xpu.is_available() and torch.xpu.device_count() > 0 except: pass # Fallback: check lspci for Arc GPU import subprocess result = subprocess.run( ['lspci'], capture_output=True, text=True, timeout=2 ) return 'Intel' in result.stdout and 'Arc' in result.stdout except: return False def load_model(self, use_openvino: bool = True): """ Load model with Intel optimizations Options: 1. OpenVINO: Best for Arc GPU (recommended) 2. IPEX CPU: Good for CPU inference with Intel optimizations 3. Transformers: Fallback (slower) """ if use_openvino and self.has_arc: logger.info("šŸš€ Loading model with OpenVINO (Arc GPU optimized)...") try: from optimum.intel import OVModelForCausalLM from transformers import AutoTokenizer self.model = OVModelForCausalLM.from_pretrained( self.model_name, export=True, device="GPU" # Use Arc GPU ) self.tokenizer = AutoTokenizer.from_pretrained(self.model_name) logger.info("āœ… Model loaded with OpenVINO (GPU)") return except Exception as e: logger.warning(f"āš ļø OpenVINO failed: {e}, falling back...") # Try Intel CPU optimizations logger.info("šŸ“¦ Loading model with Intel CPU optimizations...") from transformers import AutoModelForCausalLM, AutoTokenizer import torch import os # Get HF token from environment or .env file hf_token = os.getenv('HF_TOKEN') self.model = AutoModelForCausalLM.from_pretrained( self.model_name, device_map="cpu", torch_dtype=torch.bfloat16, # Use bfloat16 for better CPU performance token=hf_token # Pass token for gated models ) self.tokenizer = AutoTokenizer.from_pretrained( self.model_name, token=hf_token # Pass token for gated models ) # Apply Intel optimizations try: import intel_extension_for_pytorch as ipex logger.info("šŸš€ Applying Intel CPU optimizations...") self.model = ipex.optimize(self.model, dtype=torch.bfloat16) logger.info("āœ… Model loaded with Intel CPU optimizations") except Exception as e: logger.warning(f"āš ļø Intel optimizations failed: {e}") logger.info("āœ… Model loaded (standard PyTorch)") def extract_interest_groups( self, bill_context: Dict[str, Any], testimony: List[Dict[str, Any]] ) -> List[InterestGroup]: """ Extract structured interest group data using LLM On 64GB RAM, we can fit the entire bill + all testimony in one prompt """ if not self.model or not self.tokenizer: self.load_model() # Build prompt prompt = self._build_extraction_prompt(bill_context, testimony) # Run inference inputs = self.tokenizer(prompt, return_tensors="pt") outputs = self.model.generate( **inputs, max_new_tokens=2048, temperature=0.3, # Lower for structured extraction do_sample=True ) response = self.tokenizer.decode(outputs[0], skip_special_tokens=True) # Parse JSON response try: groups_data = json.loads(response.split("```json")[1].split("```")[0]) return [InterestGroup(**g) for g in groups_data.get('groups', [])] except: logger.error("āŒ Failed to parse LLM response") return [] def _build_extraction_prompt( self, bill: Dict[str, Any], testimony: List[Dict[str, Any]] ) -> str: """Build structured extraction prompt""" return f"""You are a legislative analyst. Extract interest group positions from testimony. BILL: {bill['id']} - {bill['title']} {bill.get('abstract', '')} TESTIMONY: {chr(10).join([f"- {t['speaker']} ({t['organization']}): {t['text'][:200]}..." for t in testimony])} Extract each group's position in JSON format: ```json {{ "groups": [ {{ "group_name": "Organization name", "lobbyist": "Name if mentioned, else null", "stance": "support|oppose|neutral|conditional", "stance_score": -1.0 to 1.0, "tradeoff_notes": "Any concessions or compromises mentioned", "testimony_excerpt": "Key quote showing their position", "bill_id": "{bill['id']}", "confidence": 0.0 to 1.0 }} ] }} ``` Focus on: 1. Named organizations and their representatives 2. Explicit support/opposition statements 3. Conditional support ("we support IF...") 4. Tradeoffs or compromises mentioned Return only valid JSON.""" def main(): """Demo: Intel-optimized legislative analysis with incremental processing""" logger.info("šŸš€ Intel Arc-Optimized Legislative Analysis Demo") logger.info("=" * 60) logger.info("") logger.info("šŸ“ Data Pipeline Architecture:") logger.info(" Source: Parquet files (bills, sponsors, officials)") logger.info(" Processing: DuckDB (fast queries) + Llama (AI analysis)") logger.info(" Results: Parquet files (for reuse & sharing)") logger.info(" Incremental: Skip already-analyzed bills") logger.info("") # Initialize DuckDB analyzer with DuckDBLegislativeAnalyzer() as analyzer: # Create tables (DuckDB queries Parquet directly - no copying!) analyzer.create_bills_table() analyzer.create_testimony_table() # Show statistics logger.info("šŸ“Š Bill Statistics:") stats = analyzer.analyze_bill_statistics() logger.info(f" Top states: {stats.get('top_states', [])[:5]}") if 'top_subjects' in stats: logger.info(f" Top subjects: {stats.get('top_subjects', [])[:5]}") elif 'top_sessions' in stats: logger.info(f" Top sessions: {stats.get('top_sessions', [])[:5]}") elif 'top_topics' in stats: logger.info(f" Top topics: {stats.get('top_topics', [])[:5]}") # Demonstrate incremental processing logger.info("\nšŸ”„ Incremental Processing Demo:") # Check for existing analysis results analysis_file = ANALYSIS_DIR / "interest_groups_analysis.parquet" if analysis_file.exists(): # Show what's already analyzed result = analyzer.conn.execute(f""" SELECT COUNT(DISTINCT bill_id) as bills_analyzed, COUNT(*) as total_groups, COUNT(DISTINCT group_name) as unique_groups FROM read_parquet('{analysis_file}') """).fetchone() logger.info(f" āœ… Found existing analysis:") logger.info(f" - {result[0]} bills analyzed") logger.info(f" - {result[1]} interest groups extracted") logger.info(f" - {result[2]} unique organizations") else: logger.info(f" šŸ“ No existing analysis found") logger.info(f" šŸ’¾ Results will be saved to: {analysis_file}") # Get bills that need analysis (skips already-analyzed) logger.info("\nšŸ” Finding bills to analyze...") bills_to_analyze = analyzer.get_bills_to_analyze( state='GA', # Georgia bills topic_filter='fluorid', # Fluoride-related limit=5, skip_analyzed=True # Incremental processing! ) if bills_to_analyze: logger.info(f" šŸ“‹ Found {len(bills_to_analyze)} unanalyzed bills:") for bill in bills_to_analyze[:3]: logger.info(f" - {bill['bill_number']}: {bill['title'][:60]}...") else: logger.info(f" āœ… All matching bills already analyzed!") logger.info(f" šŸ’” To re-analyze, delete {analysis_file}") logger.info("\n" + "=" * 60) logger.info("āœ… Demo complete!") logger.info("") logger.info("šŸŽÆ Why this architecture?") logger.info("") logger.info(" Parquet Storage:") logger.info(" āœ… Portable - share with anyone") logger.info(" āœ… Fast - columnar format") logger.info(" āœ… Compatible - works with Pandas, Spark, DuckDB") logger.info(" āœ… Versioned - track in git (if small) or DVC") logger.info("") logger.info(" DuckDB Query Engine:") logger.info(" āœ… 10-100x faster than Postgres for analytics") logger.info(" āœ… Queries Parquet directly (no import needed)") logger.info(" āœ… Embedded - no server to manage") logger.info(" āœ… SQL interface - easy to use") logger.info("") logger.info(" Incremental Processing:") logger.info(" āœ… Skip already-analyzed bills") logger.info(" āœ… Resume after failures") logger.info(" āœ… Append new results to Parquet") logger.info("") logger.info("šŸ“– Next: Run batch analysis with:") logger.info(" python scripts/enrichment_ai/batch_analyze_bills.py") if __name__ == "__main__": main()