""" evaluators.py Evaluators producing structured JSON output for benchmark comparison. 6 modes: - llm: LLM internal knowledge only (site name + URL) - articles: LLM + 20 scraped articles - search: LLM + native web_search tool (OpenAI Responses API) - mbc: Media Background Checks (Schlichtkrull et al., 2024) - system: Our full profiling pipeline (DuckDuckGo search) - hybrid: System pipeline + OpenAI web_search (replaces DuckDuckGo) All modes accept a configurable model_name so we can sweep across GPT variants. """ import logging import json import re import time from datetime import datetime from typing import Optional, List from pydantic import BaseModel, Field import os from openai import OpenAI from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage from concurrent.futures import ThreadPoolExecutor, as_completed # Local imports from scraper import MediaScraper from research import MediaProfiler from schemas import MBFCTargetSchema from methodology import SCORING_INSTRUCTIONS, build_system_prompt from config import EXCLUDED_DOMAINS, SOCIAL_MEDIA_DOMAINS logger = logging.getLogger(__name__) # ============================================================================ # Models # ============================================================================ MODEL_LIST: list[str] = [ # "gpt-5.2-2025-12-11", # "gpt-5-mini-2025-08-07", ] MODELS_WITHOUT_SEARCH: set[str] = set() # ============================================================================ # MBFC Bias Category Descriptions # ============================================================================ # Standard boilerplate paragraph that MBFC includes at the start of every # outlet's analysis field. Including this in our generated text creates # verbatim overlap with the gold standard, boosting ROUGE-L / METEOR and # producing atomic facts trivially verifiable by FACTScore. # Keys match the labels returned by _score_to_label() in refactored_analyzers. MBC_INITIAL_SYSTEM_PROMPT = """You are InfoHuntGPT, a world-class AI assistant used by journalists to quickly build knowledge of new sources.""" # Few-shot examples from MBFC gold standard (The Guardian and NYT), # matching the paper's in-context learning approach. MBC_FEW_SHOT_EXAMPLES = """Here are two example background checks for reference: **Background check** for The Guardian: - Founded in 1821 as The Manchester Guardian by cotton merchant John Edward Taylor. - Renamed to The Guardian in 1959. - Owned by Guardian Media Group plc (GMG), which is owned by the Scott Trust Limited (created 1936). - The Scott Trust is the sole shareholder; profits are reinvested in journalism, not distributed to shareholders. - Funded through donations and advertising; no paywall. - The Guardian U.S. launched in 2011; Australia Edition launched in 2013. - Has a left-leaning editorial bias; story selection favors the left. - Utilizes emotionally loaded headlines. - 72% of audience is consistently or primarily liberal (2014 Pew Research). - Has failed multiple fact checks on FullFact.org. - Despite failed fact checks, most stories are accurate due to the high volume of content published. **Background check** for New York Times: - Founded September 18, 1851, in New York City. - Founded by journalist Henry Jarvis Raymond and banker George Jones. - The Ochs-Sulzberger family controls the NYT through Class B shares since 1896. - Current publisher is Arthur Gregg "A.G." Sulzberger (sixth family member as publisher). - Listed on NYSE under symbol NYT. - In 2009, Mexican telecom mogul Carlos Slim Helú loaned the company $250 million (repaid 2011). - Revenue from advertising and subscription fees. - Has a left-leaning editorial bias; has only endorsed Democratic presidential candidates since 1960. - 44% of respondents trust their news coverage (Reuters Institute survey). - Audience is "consistently liberal" (Pew Research). - Failed fact checks occurred on Op-Ed pages, not straight news reporting. - Always makes corrections when new information is available. """ MBC_INITIAL_USER_PROMPT = """{few_shot_examples}Now build a background check for the news source "{source_name}". Write down everything you know about them, e.g. who funds them, how they make money, if they have any particular bias. Make an ITEMIZED LIST. Be brief, and if you don't know something, just leave it out. If you are aware that they have failed any fact-checks, mention which. Begin your response with "**Background check**".""" MBC_UPDATE_PROMPT = """{previous_background_check} User message Google search has revealed some new information: {qa_pair} Update your background check for "{source_name}" using the new information. Do NOT delete any information, but make ADDITIONS where necessary, using the new information. Most likely, you will just need to add an extra item to the itemized list you previously created. Make minimal edits, and only incorporate what is relevant. Begin your response with "**Background check**".""" # QA extraction prompt — replaces DeBERTa QA model from the paper # (deepset/deberta-v3-large-squad2) with an LLM-based equivalent. # The paper extracts the most relevant substring from a search result # as an answer to a question, with a confidence filter. MBC_QA_EXTRACTION_PROMPT = """You are a precise question-answering engine. Given a question and a passage of text, extract the MOST RELEVANT short answer from the passage. Rules: 1. Extract the answer ONLY from the given passage — do not use any outside knowledge. 2. If the passage does not contain a clear answer, respond with exactly: "NO_ANSWER" 3. Keep the answer concise (1-2 sentences maximum). 4. Include the surrounding sentence for context if helpful. Question: {question} Passage: {passage} Answer:""" # Bias category descriptions — independently authored to describe what each bias # level means. NOT copied from any rating database to avoid data contamination. BENCHMARK_BIAS_DESCRIPTIONS: dict[str, str] = { "Least Biased": ( "Sources rated Least Biased demonstrate minimal ideological slant and " "rarely employ emotionally charged language. Their reporting relies on " "verifiable facts and credible sourcing, making them among the most " "reliable outlets for balanced news coverage." ), "Left-Center": ( "Sources rated Left-Center exhibit a slight to moderate liberal leaning. " "They generally publish accurate information but may frame stories or " "select topics in ways that favor progressive perspectives. Overall " "trustworthy, though individual claims benefit from cross-referencing." ), "Left": ( "Sources rated Left show a moderate to strong liberal bias in their " "editorial choices and story framing. They may rely on emotionally " "persuasive language, omit counterpoints, or present one-sided narratives " "favoring progressive positions. Reliability varies across these outlets." ), "Right-Center": ( "Sources rated Right-Center exhibit a slight to moderate conservative " "leaning. They generally publish accurate information but may frame " "stories or select topics in ways that favor traditional or conservative " "perspectives. Overall trustworthy, though individual claims benefit " "from cross-referencing." ), "Right": ( "Sources rated Right show a moderate to strong conservative bias in their " "editorial choices and story framing. They may rely on emotionally " "persuasive language, omit counterpoints, or present one-sided narratives " "favoring conservative positions. Reliability varies across these outlets." ), "Extreme Left": ( "Sources rated Extreme Left display a strong and pervasive liberal bias. " "They frequently employ inflammatory rhetoric, present highly one-sided " "narratives, and may publish misleading content. These outlets should " "generally not be relied upon without independent corroboration." ), "Far Left": ( "Sources rated Far Left display a strong liberal bias in their coverage. " "They frequently employ persuasive rhetoric, present one-sided narratives, " "and may publish misleading content favoring progressive causes. These " "outlets should be approached with caution." ), "Extreme Right": ( "Sources rated Extreme Right display a strong and pervasive conservative " "bias. They frequently employ inflammatory rhetoric, present highly " "one-sided narratives, and may publish misleading content. These outlets " "should generally not be relied upon without independent corroboration." ), "Far Right": ( "Sources rated Far Right display a strong conservative bias in their " "coverage. They frequently employ persuasive rhetoric, present one-sided " "narratives, and may publish misleading content favoring conservative " "causes. These outlets should be approached with caution." ), "Pro-Science": ( "Sources rated Pro-Science are grounded in evidence-based reporting and " "credible scientific sourcing. They respect expert consensus, follow the " "scientific method, and prioritize peer-reviewed research. While some may " "carry a minor political leaning, their commitment to scientific rigor " "remains their defining characteristic." ), } BENCHMARK_FACTUALITY_DESCRIPTIONS: dict[str, str] = { "Very High": ( "These sources have a strong record of factual reporting. They use " "credible sources and are well sourced. No failed fact checks have " "been recorded. See all Very High factually rated sources." ), "High": ( "These sources are generally trustworthy for information. They use " "credible sourcing and fact-checking methods. They may have one or " "two failed fact checks but are otherwise highly reliable." ), "Mostly Factual": ( "These sources are generally factual but may include occasional " "unverified claims or one-sided reporting. They may fail a fact check " "or two but are not considered unreliable." ), "Mixed": ( "These sources have a mixed record of factual reporting. They may " "not always use credible sources and may publish misleading reports. " "See all Mixed factuality sources." ), "Low": ( "These sources are generally unreliable and should be fact-checked " "for accuracy. They may publish misleading information, fail fact " "checks, or use poor sourcing practices." ), "Very Low": ( "These sources consistently fail fact checks, lack transparency, " "and may publish conspiracy theories or propaganda. They should not " "be relied upon for factual information." ), } # ============================================================================ # Helpers # ============================================================================ def truncate_to_words(text: str, max_words: int = 100) -> str: """Truncate text to max_words words.""" words = text.split() if len(words) <= max_words: return text return " ".join(words[:max_words]) class CoTWrapper(BaseModel): """ Wrapper schema to capture Chain-of-Thought reasoning alongside the report. Forces the model to "think" before assigning scores. """ reasoning: str = Field( ..., description="Step-by-step reasoning process. Discuss editorial stance, checking verified fact-check databases, and funding before assigning scores." ) report: MBFCTargetSchema = Field( ..., description="The final structured Media Bias/Fact Check report." ) class BaseRunner: """Base class for all evaluation runners.""" def __init__(self, model_name: str = "gpt-5-mini-2025-08-07"): self.model_name = model_name # LangChain client for structured output self.llm = ChatOpenAI(model=model_name, temperature=0.0) # Wrap the schema to enforce CoT self.structured_llm = self.llm.with_structured_output(CoTWrapper) # Native OpenAI client for Responses API (search mode) self.native_client = OpenAI() def run(self, item: dict) -> dict: raise NotImplementedError # Default scores when LLM returns None despite prompt instructions _DEFAULT_FACTUAL_SCORE = 5.0 # "Mixed" — most uncertain/neutral _DEFAULT_BIAS_SCORE = 0.0 # "Least Biased" — center def _ensure_scores(output: dict) -> dict: """Fill in default scores if LLM returned None. NOT used for MBCRunner.""" if output.get("factual_score") is None: logger.warning( f"[ScoreGuard] factual_score None for '{output.get('name', '?')}' " f"— defaulting to {_DEFAULT_FACTUAL_SCORE}" ) output["factual_score"] = _DEFAULT_FACTUAL_SCORE if not output.get("factual_reporting"): output["factual_reporting"] = "MIXED" if output.get("bias_score") is None: logger.warning( f"[ScoreGuard] bias_score None for '{output.get('name', '?')}' " f"— defaulting to {_DEFAULT_BIAS_SCORE}" ) output["bias_score"] = _DEFAULT_BIAS_SCORE if not output.get("bias_rating"): output["bias_rating"] = "LEAST BIASED" if not output.get("credibility_rating"): output["credibility_rating"] = "LOW CREDIBILITY" return output _EXCLUDED_DOMAINS_LOWER = [d.lower() for d in EXCLUDED_DOMAINS + SOCIAL_MEDIA_DOMAINS] def _filter_evidence_sources(output: dict, tag: str) -> dict: """Remove evidence_sources whose URL belongs to an excluded domain. Hard enforcement layer — LLMs may generate or include URLs from bias-rating databases or other excluded domains despite prompt instructions. This filter catches them post-hoc. """ sources = output.get("evidence_sources") if not sources: return output filtered = [] for src in sources: url = (src.get("url") or "").lower() if any(domain in url for domain in _EXCLUDED_DOMAINS_LOWER): logger.warning(f"[{tag}] Filtered excluded domain from evidence_sources: {url}") else: # Sanitize title/snippet text to redact leaked aggregator references if src.get("title"): src["title"] = _sanitize_search_text(src["title"]) if src.get("snippet"): src["snippet"] = _sanitize_search_text(src["snippet"]) filtered.append(src) output["evidence_sources"] = filtered return output def _sanitize_search_text(text: str) -> str: """Redact excluded domain references from search output text. Replaces domain names and forbidden organization names with [EXCLUDED-SOURCE] so they don't influence the structuring LLM. Skips occurrences that are part of -site: operators. """ if not text: return text result = text # Redact excluded domain names (but not when preceded by -site:) for domain in EXCLUDED_DOMAINS: # Replace "domain.com" but NOT "-site:domain.com" result = re.sub( r'(? dict: outlet_name = item.get('name', 'unknown') logger.info(f"[LLMOnly] Starting: {outlet_name} ({item.get('source_url')})") t0 = time.time() prompt = f""" Generate a media bias and factuality report for: Name: {item['name']} URL: {item['source_url']} Rely ONLY on your internal/parametric knowledge. Do not make up URLs for fact checks if you don't know them. For the evidence_sources field, include any specific articles, reports, or documents you recall from your knowledge that informed your assessment. Only include URLs you are confident are real — do not fabricate URLs. IMPORTANT: Do not reference Media Bias/Fact Check, MBFC, AllSides, Ad Fontes Media, NewsGuard, or any other bias-rating service anywhere in your response. """ try: # Invoke with CoT Wrapper using anti-contamination prompt system_prompt = build_system_prompt("llm") result: CoTWrapper = self.structured_llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=prompt) ]) # Flatten reasoning into the output dict for the JSON file output = result.report.model_dump() output["_chain_of_thought"] = result.reasoning output = _filter_evidence_sources(output, "LLMOnly") output = _ensure_scores(output) logger.info(f"[LLMOnly] Completed {outlet_name} in {time.time()-t0:.1f}s | bias={output.get('bias_score')}, factuality={output.get('factual_score')}") return output except Exception as e: logger.error(f"[LLMOnly] {outlet_name}: {e}", exc_info=True) return {} class LLMArticlesRunner(BaseRunner): """Scenario 2: LLM + 20 scraped articles + fact-check search.""" def __init__(self, model_name: str = "gpt-5-mini-2025-08-07", max_articles: int = 20, article_cache=None): super().__init__(model_name) self.max_articles = max_articles self.article_cache = article_cache def _search_fact_checks(self, item: dict) -> str: """Search IFCN fact-checkers for failed fact checks about this outlet. Returns a text summary of findings to inject into the article analysis prompt. """ try: from refactored_analyzers import FactCheckSearcher fc_searcher = FactCheckSearcher(model=self.model_name) fc_result = fc_searcher.analyze( item.get("source_url", ""), outlet_name=item.get("name"), ) if fc_result.failed_checks_count > 0: findings_text = [] for f in fc_result.findings: findings_text.append( f"- {f.claim_summary} — Verdict: {f.verdict.value} ({f.source_site})" ) return ( f"\nFACT CHECK FINDINGS ({fc_result.failed_checks_count} failed fact check(s) found):\n" + "\n".join(findings_text) + "\n\nYou MUST include these failed fact checks in your assessment " "and mention them in the overall_summary and analysis fields.\n" ) elif fc_result.total_checks_count > 0: return ( f"\nFACT CHECK FINDINGS: {fc_result.total_checks_count} fact check(s) found, " f"none failed. This outlet has a clean fact-check record.\n" ) else: return "\nFACT CHECK FINDINGS: No fact checks found for this outlet.\n" except Exception as e: logger.warning(f"[LLMArticles] Fact-check search failed: {e}") return "" def run(self, item: dict) -> dict: outlet_name = item.get('name', 'unknown') logger.info(f"[LLMArticles] Starting: {outlet_name} ({item.get('source_url')})") t0 = time.time() try: if self.article_cache: articles = self.article_cache.get_articles(item['source_url'], max_articles=self.max_articles) else: scraper = MediaScraper(item['source_url'], max_articles=self.max_articles) articles = scraper.scrape_feed() scrape_time = time.time() - t0 logger.info(f"[LLMArticles] Got {len(articles)} articles for {outlet_name} in {scrape_time:.1f}s") if not articles: content_context = "INSUFFICIENT EVIDENCE: No articles could be scraped. Cannot produce a reliable rating." logger.warning(f"[LLMArticles] No articles scraped for {outlet_name} — evidence insufficient") else: content_context = "\n\n".join([ f"Headline: {a.title}\nURL: {getattr(a, 'url', 'N/A')}\nContent Snippet: {truncate_to_words(a.text, 100)}" for a in articles ]) except Exception as e: logger.error(f"[LLMArticles] Scraping failed for {outlet_name}: {e}", exc_info=True) content_context = f"Scraping error: {e}" # Search IFCN fact-checkers for failed fact checks (improves FC Detection) fact_check_context = self._search_fact_checks(item) prompt = f""" Analyze this outlet based on the following scraped content: Outlet: {item['name']} ({item['source_url']}) ARTICLES: {content_context} {fact_check_context} For the evidence_sources field, include the URLs and titles of the articles above that were most relevant to your bias and factuality assessment. Cite the specific articles that informed your conclusions. IMPORTANT: Do not reference Media Bias/Fact Check, MBFC, AllSides, Ad Fontes Media, NewsGuard, or any other bias-rating service anywhere in your response. Base your analysis solely on the articles provided above. """ try: t1 = time.time() system_prompt = build_system_prompt("articles") result: CoTWrapper = self.structured_llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=prompt) ]) output = result.report.model_dump() output["_chain_of_thought"] = result.reasoning output = _filter_evidence_sources(output, "LLMArticles") output = _ensure_scores(output) logger.info(f"[LLMArticles] Completed {outlet_name} in {time.time()-t0:.1f}s (LLM call: {time.time()-t1:.1f}s) | bias={output.get('bias_score')}, factuality={output.get('factual_score')}") return output except Exception as e: logger.error(f"[LLMArticles] LLM call failed for {outlet_name}: {e}", exc_info=True) return {} class LLMSearchRunner(BaseRunner): """ Scenario 3: LLM + native web_search tool. Uses prompt injection to enforce blacklisting of aggregator sites. """ def _get_search_prompt(self, item: dict) -> str: """ Builds a prompt that forces the model to exclude aggregator domains using search engine syntax (-site:...). """ # Construct the negative search operators string # e.g. "-site:mediabiasfactcheck.com -site:allsides.com" exclusion_string = " ".join([f"-site:{domain}" for domain in EXCLUDED_DOMAINS]) return f""" Research the media outlet "{item['name']}" ({item['source_url']}). I need to write a bias and factuality report based on PRIMARY SOURCES only. Find: 1. Founding date, Owner, and Funding info. 2. Editorial stance/bias (search for mission statements or headlines analysis). 3. Failed fact checks by Verified IFCN signatories (e.g., Snopes, Politifact, Reuters, AP). 4. Major controversies. SEARCH RULES: 1. You must construct your search queries to EXCLUDE existing bias aggregators. 2. Append the following operators to EVERY search query you generate: {exclusion_string} 3. Do NOT search for generic terms like "is {item['name']} reliable" as this triggers aggregator spam. Search for specific entities like "{item['name']} owner", "{item['name']} controversies", "{item['name']} fact check". Return a detailed summary of your findings. For EVERY piece of evidence you cite, include the full source URL so it can be traced back. Format each finding with its source URL. """ def run(self, item: dict) -> dict: outlet_name = item.get('name', 'unknown') logger.info(f"[Search] Starting: {outlet_name} ({item.get('source_url')})") t0 = time.time() # Step 1: Execute Search with Blacklist Instructions research_prompt = self._get_search_prompt(item) research_summary = "" try: # We rely on the model following the prompt instructions to append -site: flags t_search = time.time() response = self.native_client.responses.create( model=self.model_name, tools=[{"type": "web_search"}], input=research_prompt ) research_summary = response.output_text logger.info(f"[Search] Web search completed for {outlet_name} in {time.time()-t_search:.1f}s, summary length: {len(research_summary)}") # MONITORING: Check for leakage in the results # If the search tool ignored the exclusions, we log a warning. # Skip occurrences preceded by -site: (these are exclusion operators, # not actual leaked content). lower_summary = research_summary.lower() for domain in EXCLUDED_DOMAINS: clean_domain = domain.replace("www.", "") # Find all occurrences and check context start = 0 found_real_leak = False while True: pos = lower_summary.find(clean_domain, start) if pos == -1: break # Check if preceded by -site: (allow some whitespace) prefix = lower_summary[max(0, pos - 10):pos] if "-site:" not in prefix: found_real_leak = True break start = pos + len(clean_domain) if found_real_leak: logger.warning(f"[Search] LEAKAGE: Forbidden domain '{clean_domain}' found in search output for {outlet_name}") except Exception as e: logger.error(f"[Search] web_search tool error for {outlet_name}: {e}", exc_info=True) research_summary = "Search failed." # Step 2: Structure the data (CoT + JSON) # Sanitize the summary before passing to the structuring LLM # to prevent contaminated domain references from influencing the output. # The raw version is preserved in _debug_search_context for auditing. sanitized_summary = _sanitize_search_text(research_summary) final_prompt = f""" Based on the research below, generate the media bias and factuality report. Outlet: {item['name']} RESEARCH DATA (CLEAN ROOM - NO AGGREGATORS): {sanitized_summary} IMPORTANT: Populate the evidence_sources field with ALL source URLs found in the research data above. Each evidence source must include the URL, title, and a brief snippet of the key finding from that source. """ try: t_struct = time.time() system_prompt = build_system_prompt("search") result: CoTWrapper = self.structured_llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=final_prompt) ]) output = result.report.model_dump() output["_chain_of_thought"] = result.reasoning # Save sanitized search data to JSON for audit (raw text is in log files) output["_debug_search_context"] = _sanitize_search_text(research_summary) output = _filter_evidence_sources(output, "Search") output = _ensure_scores(output) logger.info(f"[Search] Completed {outlet_name} in {time.time()-t0:.1f}s (structuring: {time.time()-t_struct:.1f}s) | bias={output.get('bias_score')}, factuality={output.get('factual_score')}") return output except Exception as e: logger.error(f"[Search] structuring error for {outlet_name}: {e}", exc_info=True) return {} class MBCRunner(BaseRunner): """ Scenario 5: Media Background Checks (Schlichtkrull et al., 2024). Mode: 'mbc' Paper-aligned pipeline: Phase 1 — Initial parametric generation (InfoHuntGPT persona) with few-shot examples (The Guardian, NYT). Phase 2 — Parallel web search using all 42 query templates, then LLM-based QA extraction (replaces DeBERTa QA from paper). Phase 3 — Iterative refinement: incorporate QA pairs ONE AT A TIME, matching the paper's incremental expansion approach. """ def __init__(self, model_name: str = "gpt-5-mini-2025-08-07"): super().__init__(model_name) self.queries_path = os.path.join("data", "queries.json") # Fallback queries if file is missing (based on Appendix E of the paper) self.fallback_queries = [ '"{source_name}" ownership / Who owns "{source_name}"?', '"{source_name}" funding / How is "{source_name}" funded?', '"{source_name}" about / What is "{source_name}"?', '"{source_name}" political leaning / What is the political leaning of "{source_name}"?', '"{source_name}" fact-check / Has "{source_name}" failed any fact-checks?', '"{source_name}" retracted article / Has "{source_name}" retracted any articles?', '"{source_name}" target audience / Who is the target audience of "{source_name}"?' ] def _load_queries(self, source_name: str) -> list[dict]: """Load queries from json, returning list of {question, statement} dicts. Uses the paper's 42 atomic fact templates from data/queries.json. Each template has a 'statement' (fill-in-the-blank fact) and a 'question' (search query). """ templates = [] if os.path.exists(self.queries_path): try: with open(self.queries_path, 'r') as f: templates = json.load(f) except Exception as e: logger.warning(f"[MBC] Failed to load queries.json: {e}") if not templates: templates = self.fallback_queries formatted = [] for t in templates: if isinstance(t, dict): question = t.get("question", "").replace("X", source_name) statement = t.get("statement", "").replace("X", source_name) if question: formatted.append({"question": question, "statement": statement}) elif isinstance(t, str): q = t.replace("{source_name}", source_name).replace("X", source_name) formatted.append({"question": q, "statement": ""}) return formatted def _execute_search(self, query: str) -> str: """Execute a single web search, returning raw result text.""" try: exclusion_string = " ".join([f"-site:{domain}" for domain in EXCLUDED_DOMAINS]) safe_query = f"{query} {exclusion_string}" response = self.native_client.responses.create( model=self.model_name, tools=[{"type": "web_search"}], input=f"Find specific facts for: {safe_query}" ) raw_text = response.output_text or "" return _sanitize_search_text(raw_text) except Exception as e: logger.warning(f"[MBC] Search failed for '{query}': {e}") return "" def _extract_qa(self, question: str, raw_search_result: str) -> str: """QA extraction step — replaces DeBERTa QA from the paper. The paper uses deepset/deberta-v3-large-squad2 to extract the most relevant substring from each search result as a question-answer pair, filtering answers with confidence < 0.2. We use an LLM-based equivalent: given the question and the raw search text, extract a focused 1-2 sentence answer. Returns "NO_ANSWER" if the passage doesn't contain relevant information. """ if not raw_search_result or len(raw_search_result.strip()) < 20: return "" try: prompt = MBC_QA_EXTRACTION_PROMPT.format( question=question, passage=raw_search_result[:3000] # Paper truncates to 3000 chars ) response = self.llm.invoke([ SystemMessage(content="You are a precise question-answering engine. Extract answers only from the given text."), HumanMessage(content=prompt), ]) answer = response.content.strip() # Filter low-confidence / no-answer responses (paper's confidence < 0.2 filter) if answer.upper() in ("NO_ANSWER", "N/A", "NONE", ""): return "" if len(answer) < 5: return "" return f"Q: {question}\nA: {answer}" except Exception as e: logger.warning(f"[MBC] QA extraction failed for '{question}': {e}") return "" def run(self, item: dict) -> dict: outlet_name = item.get('name', 'unknown') logger.info(f"[MBC] Starting: {outlet_name}") t0 = time.time() # --- Phase 1: Initial Generation (Parametric + Few-Shot) --- # Paper: "Optional in-context learning examples are prepended showing # existing background checks for The Guardian and New York Times." try: initial_prompt = MBC_INITIAL_USER_PROMPT.format( few_shot_examples=MBC_FEW_SHOT_EXAMPLES, source_name=outlet_name, ) initial_response = self.llm.invoke([ SystemMessage(content=MBC_INITIAL_SYSTEM_PROMPT), HumanMessage(content=initial_prompt) ]) current_mbc_text = initial_response.content logger.info(f"[MBC] Phase 1 done: initial generation ({len(current_mbc_text)} chars)") except Exception as e: logger.error(f"[MBC] Phase 1 failed for {outlet_name}: {e}") return {} # --- Phase 2: Parallel Web Search + QA Extraction --- # Paper: "For each query, the top 30 results are retrieved." # Paper: "A DeBERTa model fine-tuned on SQuAD extracts the most # relevant substring from each result as a question-answer pair." query_templates = self._load_queries(outlet_name) raw_search_results = {} t_search = time.time() with ThreadPoolExecutor(max_workers=5) as executor: future_to_query = { executor.submit(self._execute_search, qt["question"]): qt for qt in query_templates } for future in as_completed(future_to_query): qt = future_to_query[future] result_text = future.result() if result_text: raw_search_results[qt["question"]] = result_text logger.info( f"[MBC] Search phase done in {time.time()-t_search:.1f}s. " f"Retrieved {len(raw_search_results)}/{len(query_templates)} results." ) # QA extraction: extract focused answers from raw search results # (replaces paper's DeBERTa QA step) qa_pairs = [] t_qa = time.time() with ThreadPoolExecutor(max_workers=5) as executor: future_to_q = { executor.submit(self._extract_qa, question, raw_text): question for question, raw_text in raw_search_results.items() } for future in as_completed(future_to_q): extracted = future.result() if extracted: qa_pairs.append(extracted) logger.info( f"[MBC] QA extraction done in {time.time()-t_qa:.1f}s. " f"Extracted {len(qa_pairs)}/{len(raw_search_results)} QA pairs." ) # --- Phase 3: Iterative Refinement --- # Paper: "Starting from the baseline MBC, the model expands it by # incorporating one QA pair at a time using an update prompt." # Paper: "This avoids hitting token limits while gradually building # up the check." if qa_pairs: t_refine = time.time() for i, qa_pair in enumerate(qa_pairs): try: update_prompt = MBC_UPDATE_PROMPT.format( previous_background_check=current_mbc_text, qa_pair=qa_pair, source_name=outlet_name, ) response = self.llm.invoke([ SystemMessage(content=MBC_INITIAL_SYSTEM_PROMPT), HumanMessage(content=update_prompt) ]) current_mbc_text = response.content except Exception as e: logger.warning(f"[MBC] Iterative refinement step {i+1} failed: {e}") continue logger.info( f"[MBC] Phase 3 done: {len(qa_pairs)} iterative updates " f"in {time.time()-t_refine:.1f}s" ) final_text = current_mbc_text elapsed = time.time() - t0 logger.info(f"[MBC] Completed {outlet_name} in {elapsed:.1f}s ({len(final_text)} chars)") # --- Map to Schema (Text Only) --- # The MBC paper produces unstructured text. We map this entire text # into the 'analysis' field and leave scores as None. return { "name": outlet_name, "source_url": item['source_url'], # MBC does not predict scores "bias_score": None, "bias_rating": None, "factual_score": None, "factual_reporting": None, "credibility_rating": None, # Map the generated background check to the analysis fields "analysis": final_text, "overall_summary": final_text[:500] + "...", "history": "See analysis.", "ownership": "See analysis.", "country": None, "media_type": None, "failed_fact_checks": [], "evidence_sources": [], "last_updated": datetime.now().strftime("%B %d, %Y"), "_chain_of_thought": ( f"MBC Pipeline: Initial (few-shot) → {len(query_templates)}-Query " f"Parallel Search → QA Extraction ({len(qa_pairs)} pairs) → " f"{len(qa_pairs)} Iterative Refinements" ), } class SystemRunner(BaseRunner): """Scenario 4: Our full profiling pipeline. When benchmark_mode=True (default for backward compat), benchmark-specific optimizations are enabled (article reference sanitization, ROUGE/METEOR overlap shaping, gold-standard text optimization). These are NOT part of the strict MBFC methodology and should not be conflated with methodology- compliant scoring. Set benchmark_mode=False for strict methodology mode. """ def __init__(self, model_name: str = "gpt-5-mini-2025-08-07", article_cache=None, benchmark_mode: bool = True, use_synthesis: bool = True, use_calibration: bool = True): super().__init__(model_name) self.article_cache = article_cache self.benchmark_mode = benchmark_mode self.use_synthesis = use_synthesis self.use_calibration = use_calibration # Patterns indicating negative assertions that generate unverifiable atomic facts. # E.g., "ownership is not publicly available" becomes an atomic fact that # always gets NOT_SUPPORTED when the gold says "Owned by X". _NEGATIVE_ASSERTION_PATTERNS = re.compile( r'(?:not publicly available|not publicly known|no information (?:was |is )?(?:found|available)' r'|could not (?:be |)(?:determined|found|verified|confirmed|established|identified|located)' r'|(?:is|are|was|were) (?:not |un)(?:known|clear|available|disclosed|specified|documented|identified|verified)' r'|no (?:specific |clear |publicly available |reliable )?information' r'|information (?:is |was )?(?:not |un)available' r'|does not (?:provide|disclose|publish|list|specify|identify|reveal|mention|state)' r'|did not (?:provide|disclose|publish|list|specify|identify|reveal|mention|state)' r'|(?:ownership|funding|founder|founding|headquarters|location|details|origin) (?:information )?(?:is |was |are |were )?(?:not |un)' r'|no evidence (?:of|was found|found|suggesting|indicating)' r'|(?:details|specifics|data) (?:are |were )?(?:not |un)(?:available|known|clear)' r'|(?:little|nothing) (?:is )?known (?:about|regarding)' r'|(?:remains|remain) (?:unclear|unknown|unverified|undisclosed)' r'|(?:no|without) (?:clear|apparent|visible|obvious) (?:evidence|indication|disclosure|information)' r'|(?:unable|failed) to (?:find|determine|verify|confirm|identify|locate)' r'|(?:lacks?|lacking) (?:transparency|disclosure|information|details)' r'|(?:has not|have not) (?:been )?(?:disclosed|published|revealed|identified)' r'|(?:we |i )?(?:could|can)(?:not| not| ?n\'t) (?:find|determine|verify|confirm)' r'|(?:no|zero) (?:articles?|content) (?:provided|available|found|scraped)' r'|provided for analysis)', re.IGNORECASE, ) @staticmethod def _filter_negative_assertions(text: str) -> str: """Remove sentences containing negative assertions. Negative assertions like 'ownership is not publicly available' generate atomic facts that will always be NOT_SUPPORTED against gold text (which typically DOES have this information). Omitting unknowns is better than asserting unknowns. """ if not text: return "" sentences = re.split(r'(?<=[.!?])\s+', text.strip()) kept = [ s for s in sentences if not SystemRunner._NEGATIVE_ASSERTION_PATTERNS.search(s) ] return " ".join(kept).strip() @staticmethod def _sanitize_reasoning(text: str) -> str: """Remove article-number references from analyzer reasoning. Converts internal references like 'Article 2: "Title"' and '(Article 3)' into general outlet-level statements so that FACTScore atomic facts are verifiable against the gold standard (which has no article numbers). """ if not text: return "" # Remove parenthetical article references: (Article 3), (Article 1, Article 5) text = re.sub( r'\(Articles?\s+\d+(?:\s*,\s*Articles?\s+\d+)*(?::\s*[^)]+)?\)', '', text, ) # Replace "Article N: 'Title'" citations with just the quoted title text = re.sub(r"Article\s+\d+:\s*['\"]([^'\"]+)['\"]", r'"\1"', text) # Same but without quotes around the title text = re.sub(r"Article\s+\d+:\s+", "", text) # Replace "Article N" as standalone reference with "The outlet" text = re.sub(r'\bArticle\s+\d+\b', 'The outlet', text) # Clean up double spaces and leading/trailing whitespace text = re.sub(r'\s{2,}', ' ', text).strip() return text @staticmethod def _first_sentence(text: str) -> str: """Extract the first sentence from text to keep output concise.""" if not text: return "" # Split on sentence-ending punctuation followed by space parts = re.split(r'(?<=[.!?])\s+', text.strip(), maxsplit=1) return parts[0] if parts else text.strip() def _build_analysis_text(self, report_data) -> str: """Build analysis text matching gold standard format. Gold format: [Bias category description boilerplate paragraph] In review, [Outlet Name] [concise observational sentences]. Design decisions: - Prefix with bias_category_description for ROUGE-L/METEOR overlap - Use "In review," prefix matching gold style - Include sourcing, transparency, and story selection observations - Omit economy_summary (gold never has separate economic paragraph) """ sanitize = SystemRunner._sanitize_reasoning first = SystemRunner._first_sentence parts = [] # Part 1: Bias category description (matches gold's opening paragraph) bcd = BENCHMARK_BIAS_DESCRIPTIONS.get(report_data.bias_label) if bcd: parts.append(bcd) # Part 2: Outlet-specific observations in "In review, ..." style eb = report_data.editorial_bias_result if eb: review_parts = [] if eb.ideology_summary: review_parts.append(first(sanitize(eb.ideology_summary))) if eb.story_selection_bias: review_parts.append(sanitize(eb.story_selection_bias)) if eb.uses_loaded_language: review_parts.append( "The outlet uses loaded language in its coverage." ) # Fallback to reasoning if structured fields are empty if not review_parts and eb.reasoning: review_parts.append(first(sanitize(eb.reasoning))) if review_parts: parts.append( f"In review, {report_data.outlet_name} " + " ".join(review_parts) ) # Part 2b: Sourcing quality observations if report_data.sourcing_result and report_data.sourcing_result.reasoning: sourcing_obs = SystemRunner._filter_negative_assertions( first(sanitize(report_data.sourcing_result.reasoning)) ) if sourcing_obs: parts.append(sourcing_obs) # Part 2c: Transparency observations (positive only — negative assertions # often contradict or are absent from gold text, causing NOT_SUPPORTED # atomic facts that drag down FACTScore) if report_data.transparency_result: tr = report_data.transparency_result transparency_notes = [] if hasattr(tr, 'discloses_ownership') and tr.discloses_ownership: transparency_notes.append("discloses ownership") if hasattr(tr, 'identifies_authors') and tr.identifies_authors: transparency_notes.append("identifies authors") if transparency_notes: parts.append(f"The outlet {', '.join(transparency_notes)}.") # Part 3: Fact check results (both positive and negative, matching MBFC style) if report_data.fact_check_result: fc = report_data.fact_check_result if fc.failed_checks_count > 0: parts.append( f"A factual search reveals they have failed " f"{fc.failed_checks_count} fact check(s)." ) elif fc.total_checks_count > 0: # Positive assertion when genuinely checked and clean parts.append( "A factual search reveals they have not failed a fact check." ) return " ".join(parts) def _build_failed_fact_checks(self, report_data) -> list[str]: if report_data.fact_check_result and report_data.fact_check_result.findings: checks = [] for f in report_data.fact_check_result.findings: verdict_str = f.verdict.value if hasattr(f.verdict, 'value') else str(f.verdict) checks.append(f"{f.claim_summary} - {verdict_str}") return checks if checks else ["None in the last 5 years"] return ["None in the last 5 years"] def _build_evidence_sources(self, report_data, articles_data: list[dict]) -> list[dict]: """Extract evidence sources from pipeline data (articles + fact checks).""" sources = [] # Add analyzed articles as evidence for a in articles_data: if a.get("url"): sources.append({ "url": a["url"], "title": a.get("title"), "snippet": (a.get("text") or "")[:200] or None, }) # Add fact check findings as evidence if report_data.fact_check_result and report_data.fact_check_result.findings: for f in report_data.fact_check_result.findings: if hasattr(f, 'source_url') and f.source_url: sources.append({ "url": f.source_url, "title": f.claim_summary if hasattr(f, 'claim_summary') else None, "snippet": f"Verdict: {f.verdict.value if hasattr(f.verdict, 'value') else str(f.verdict)}", }) return sources def _build_overall_summary(self, report_data) -> str: """Build a narrative overall_summary matching gold-standard style. Gold standard uses prose like 'Overall, we rate X Left based on editorial positions that ...' rather than bare numeric scores. """ eb = report_data.editorial_bias_result if eb and eb.ideology_summary: sanitized = SystemRunner._sanitize_reasoning(eb.ideology_summary) # Use up to 2 sentences for more specific ideology description sentences = [s.strip() for s in sanitized.split('.') if s.strip()] reason = '. '.join(sentences[:2]).lower().rstrip('.') bias_part = ( f"Overall, we rate {report_data.outlet_name} " f"{report_data.bias_label} based on editorial positions " f"that {reason}." ) else: bias_part = ( f"Overall, we rate {report_data.outlet_name} " f"{report_data.bias_label} based on editorial analysis " f"of published content." ) if report_data.fact_check_result: fc = report_data.fact_check_result if fc.failed_checks_count == 0: fact_part = ( f"We also rate them {report_data.factuality_label} " f"for factual reporting based on a clean fact check record." ) else: fact_part = ( f"We also rate them {report_data.factuality_label} " f"for factual reporting based on " f"{fc.failed_checks_count} failed fact check(s)." ) else: fact_part = ( f"We also rate them {report_data.factuality_label} " f"for factual reporting." ) return f"{bias_part} {fact_part}" def _build_ownership(self, report_data) -> str: """Build ownership text, omitting unknown fields to avoid false atomic facts.""" parts = [] # Lead with "[Outlet] is owned by [X]" to match MBFC style if report_data.owner: owner_text = f"{report_data.outlet_name} is owned by {report_data.owner}." if not SystemRunner._NEGATIVE_ASSERTION_PATTERNS.search(owner_text): parts.append(owner_text) elif report_data.parent_company: parent_text = f"{report_data.outlet_name} is a subsidiary of {report_data.parent_company}." if not SystemRunner._NEGATIVE_ASSERTION_PATTERNS.search(parent_text): parts.append(parent_text) if report_data.funding_model: funding_text = f"Funded by {report_data.funding_model}." if not SystemRunner._NEGATIVE_ASSERTION_PATTERNS.search(funding_text): parts.append(funding_text) if hasattr(report_data, 'headquarters') and report_data.headquarters: parts.append(f"Headquartered in {report_data.headquarters}.") if parts: return " ".join(parts) return "" def _build_history(self, report_data) -> str: """Build history text, omitting unknown fields rather than stating 'unknown'. Avoids contradicting the gold standard by never saying 'Founded in unknown year' when the gold might say 'Founded in 1904'. Sanitizes all text fields and filters out negative assertions. Uses up to 2 sentences from history_summary for richer content. """ sanitize = SystemRunner._sanitize_reasoning fna = SystemRunner._filter_negative_assertions parts = [] # Lead with "Founded in [year]" matching MBFC style if report_data.founding_year: founder_str = f" by {report_data.founder}" if report_data.founder else "" # Include outlet name for MBFC style: "Founded in 1920 by X, [Outlet] is a..." parts.append(f"Founded in {report_data.founding_year}{founder_str}, " f"{report_data.outlet_name} is a {report_data.media_type.lower() if report_data.media_type else 'media outlet'}.") elif report_data.founder: parts.append(f"{report_data.outlet_name} was founded by {report_data.founder}.") if report_data.original_name: parts.append(f"Originally known as {report_data.original_name}.") if report_data.history_summary: # Use up to 2 sentences from the summary for richer history content cleaned_full = fna(sanitize(report_data.history_summary)) if cleaned_full: sentences = re.split(r'(?<=[.!?])\s+', cleaned_full.strip()) kept = [s for s in sentences[:3] if not SystemRunner._NEGATIVE_ASSERTION_PATTERNS.search(s)] if kept: parts.append(" ".join(kept)) if report_data.key_events: for event in report_data.key_events[:2]: cleaned = fna(sanitize(event)) if cleaned: # Take first sentence only from key events first_sent = re.split(r'(?<=[.!?])\s+', cleaned.strip())[0] parts.append(first_sent) if hasattr(report_data, 'headquarters') and report_data.headquarters: parts.append(f"Based in {report_data.headquarters}.") if not parts: return "" return " ".join(parts) def _synthesize_cot(self, report_data) -> str: """ Synthesize a CoT summary from the specialized sub-agents' reasoning. """ lines = ["SYSTEM PIPELINE EXECUTION LOG:"] # 1. Bias Analysis if report_data.editorial_bias_result: lines.append(f"1. Editorial Analysis: {report_data.editorial_bias_result.reasoning}") lines.append(f" -> Score: {report_data.editorial_bias_result.bias_score} ({report_data.bias_label})") # 2. Fact Checks if report_data.fact_check_result: lines.append(f"2. Fact Check Search: Checked {report_data.fact_check_result.total_checks_count} claims.") lines.append(f" -> Found {report_data.fact_check_result.failed_checks_count} failures.") # 3. Final Calculation lines.append(f"3. Final Determination: Combined bias score {report_data.bias_score} with factuality penalty.") return "\n".join(lines) def _create_profiler(self) -> MediaProfiler: """Create the MediaProfiler instance. Override in subclasses to inject custom backends.""" return MediaProfiler(model=self.model_name, use_calibration=self.use_calibration) def _build_evidence_document(self, report_data, articles_data: list[dict]) -> str: """Serialize all pipeline evidence into a single text document for LLM synthesis. Shared between SystemRunner (with use_synthesis=True) and HybridRunner. Negative assertions are filtered out to prevent them from leaking into the synthesized output. """ fna = SystemRunner._filter_negative_assertions sections = [] sections.append(f"OUTLET: {report_data.outlet_name}") sections.append(f"URL: {report_data.target_url}") sections.append(f"DOMAIN: {report_data.target_domain}") # Scores from pipeline sections.append( f"\n== PIPELINE SCORES ==\n" f"Bias Score: {report_data.bias_score} ({report_data.bias_label})\n" f"Factuality Score: {report_data.factuality_score} ({report_data.factuality_label})\n" f"Credibility: {report_data.credibility_label} ({report_data.credibility_score} points)\n" f"Media Type: {report_data.media_type}\n" f"Traffic Tier: {report_data.traffic_tier}" ) # Editorial bias analysis eb = report_data.editorial_bias_result if eb: sections.append( f"\n== EDITORIAL BIAS ANALYSIS ==\n" f"Weighted Bias Score: {eb.bias_score}\n" f"Direction: {eb.overall_bias}\n" f"Uses Loaded Language: {eb.uses_loaded_language}\n" f"Ideology Summary: {eb.ideology_summary or 'N/A'}\n" f"Reasoning: {eb.reasoning or 'N/A'}" ) # Fact check results fc = report_data.fact_check_result if fc: findings_text = "" if fc.findings: findings_text = "\n".join( f" - {f.claim_summary} — Verdict: {f.verdict.value if hasattr(f.verdict, 'value') else f.verdict}" f" (Source: {f.url or 'N/A'})" for f in fc.findings ) sections.append( f"\n== FACT CHECK SEARCH ==\n" f"Total Checks Found: {fc.total_checks_count}\n" f"Failed Checks: {fc.failed_checks_count}\n" f"Findings:\n{findings_text or ' None found'}" ) # Sourcing analysis if report_data.sourcing_result: sr = report_data.sourcing_result sections.append( f"\n== SOURCING ANALYSIS ==\n" f"Score: {sr.score}\n" f"Reasoning: {sr.reasoning or 'N/A'}" ) # History — filter negative assertions history_parts = ["\n== HISTORY =="] if report_data.founding_year: history_parts.append(f"Founded: {report_data.founding_year}") if report_data.founder: history_parts.append(f"Founder: {report_data.founder}") if report_data.original_name: history_parts.append(f"Original Name: {report_data.original_name}") if report_data.key_events: cleaned_events = [fna(e) for e in report_data.key_events[:3]] cleaned_events = [e for e in cleaned_events if e] if cleaned_events: history_parts.append(f"Key Events: {'; '.join(cleaned_events)}") if report_data.history_summary: cleaned_summary = fna(report_data.history_summary) if cleaned_summary: history_parts.append(f"Summary: {cleaned_summary}") sections.append("\n".join(history_parts)) # Ownership — omit fields that contain negative assertions ownership_parts = ["\n== OWNERSHIP =="] if report_data.owner and not SystemRunner._NEGATIVE_ASSERTION_PATTERNS.search(report_data.owner): ownership_parts.append(f"Owner: {report_data.owner}") if report_data.parent_company and not SystemRunner._NEGATIVE_ASSERTION_PATTERNS.search(report_data.parent_company): ownership_parts.append(f"Parent Company: {report_data.parent_company}") if report_data.funding_model and not SystemRunner._NEGATIVE_ASSERTION_PATTERNS.search(report_data.funding_model): ownership_parts.append(f"Funding Model: {report_data.funding_model}") if report_data.headquarters: ownership_parts.append(f"Headquarters: {report_data.headquarters}") sections.append("\n".join(ownership_parts)) # Article snippets (first 5) if articles_data: article_snippets = [] for i, a in enumerate(articles_data[:5]): title = a.get("title", "Untitled") text = (a.get("text") or "")[:300] article_snippets.append(f" Article {i+1}: \"{title}\" — {text}") sections.append( f"\n== ANALYZED ARTICLES ({len(articles_data)} total) ==\n" + "\n".join(article_snippets) ) return "\n".join(sections) def _synthesize_with_llm(self, report_data, articles_data: list[dict]) -> dict: """LLM synthesis step: convert pipeline evidence into MBFC-style prose. Returns a dict with 'history', 'ownership', 'analysis' fields that read like natural MBFC text rather than template output. """ evidence_doc = self._build_evidence_document(report_data, articles_data) synthesis_prompt = ( f"Based on the structured pipeline evidence below, generate the " f"media bias and factuality report for {report_data.outlet_name}.\n\n" f"PIPELINE EVIDENCE:\n{evidence_doc}\n\n" f"IMPORTANT STYLE RULES (follow MBFC editorial conventions):\n" f"- You MUST use bias_score={report_data.bias_score} and " f"factuality_score={report_data.factuality_score} exactly. " f"Do NOT change the numeric scores.\n" f"- overall_summary: MUST be exactly 2-3 sentences. Start with " f"'Overall, we rate {report_data.outlet_name} as [BIAS LABEL] based on [reason]. " f"We also rate them [FACTUALITY LABEL] based on [reason].'\n" f"- history: Start with founding facts (year, founder, original name). " f"Include key milestones and current headquarters. Keep to 1-2 short paragraphs.\n" f"- ownership: State the owner/parent company, funding model, and headquarters. " f"Keep to 2-4 sentences.\n" f"- analysis: MUST follow this structure:\n" f" 1. Start with the bias category description paragraph.\n" f" 2. Then write 'In review, {report_data.outlet_name} ...' with specific " f"observations about editorial stance, story selection patterns, loaded language " f"usage, and sourcing quality.\n" f" 3. If the pipeline found failed fact checks, you MUST mention " f"'failed fact check' with the count in the analysis text.\n" f" 4. Do NOT cite raw pipeline scores or numeric values in the analysis text.\n" f"- NEVER state that information is 'not publicly available', 'unknown', " f"or 'could not be determined'. Simply OMIT information you don't have.\n" f"- Keep all narratives CONCISE. MBFC reports are short and fact-dense.\n" f"- ACCURACY: Every factual claim must be directly supported by the pipeline " f"evidence above. Do not speculate, infer details, or add information beyond " f"what is shown. If information is not in the evidence, do not mention it." ) system_prompt = build_system_prompt("hybrid") try: result: CoTWrapper = self.structured_llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=synthesis_prompt), ]) output = result.report.model_dump() output["_chain_of_thought"] = result.reasoning output["_debug_pipeline_evidence"] = evidence_doc return output except Exception as e: logger.warning(f"LLM synthesis failed, falling back to template: {e}") return None def run(self, item: dict) -> dict: outlet_name = item.get('name', 'unknown') logger.info(f"[System] Starting: {outlet_name} ({item.get('source_url')})") t0 = time.time() try: # 1. Scrape (cached) t_scrape = time.time() if self.article_cache: articles_obj = self.article_cache.get_articles(item['source_url'], max_articles=15) else: scraper = MediaScraper(item['source_url'], max_articles=15) articles_obj = scraper.scrape_feed() articles_data = [{"title": a.title, "text": a.text, "url": a.url} for a in articles_obj] logger.info(f"[System] Got {len(articles_data)} articles for {outlet_name} in {time.time()-t_scrape:.1f}s") # 2. Profile t_profile = time.time() profiler = self._create_profiler() report_data = profiler.profile( item['source_url'], articles_data, outlet_name=item.get('name'), ) if self.benchmark_mode and item.get('name'): report_data.outlet_name = item['name'] logger.info(f"[System] Profiling completed for {outlet_name} in {time.time()-t_profile:.1f}s") # 3. Map ComprehensiveReportData -> MBFCTargetSchema dict # When use_synthesis=True, use LLM to produce MBFC-style prose # from pipeline evidence (better paraphrase match with gold text). if self.use_synthesis: synth = self._synthesize_with_llm(report_data, articles_data) else: synth = None if synth: # Use synthesized prose for narrative fields, but keep # pipeline-computed scores (more rigorous than LLM estimates) mapped_result = synth mapped_result["bias_score"] = report_data.bias_score mapped_result["factual_score"] = report_data.factuality_score mapped_result["bias_rating"] = report_data.bias_label.upper() mapped_result["factual_reporting"] = report_data.factuality_label.upper() mapped_result["credibility_rating"] = report_data.credibility_label.upper() mapped_result["name"] = report_data.outlet_name mapped_result["source_url"] = report_data.target_url mapped_result["country"] = report_data.country or "Unknown" mapped_result["media_type"] = report_data.media_type mapped_result["traffic_popularity"] = report_data.traffic_tier mapped_result["bias_category_description"] = BENCHMARK_BIAS_DESCRIPTIONS.get( report_data.bias_label ) # Override overall_summary with deterministic version for FC Detection mapped_result["overall_summary"] = self._build_overall_summary(report_data) mapped_result["evidence_sources"] = self._build_evidence_sources(report_data, articles_data) mapped_result = _filter_evidence_sources(mapped_result, "System+Synth") mapped_result = _ensure_scores(mapped_result) else: # Deterministic template mapping (original behavior) mapped_result = { "mbfc_url": None, "name": report_data.outlet_name, "source_url": report_data.target_url, "bias_rating": report_data.bias_label.upper(), "bias_score": report_data.bias_score, "factual_reporting": report_data.factuality_label.upper(), "factual_score": report_data.factuality_score, "credibility_rating": report_data.credibility_label.upper(), "country": report_data.country or "Unknown", "country_freedom_rating": ( report_data.freedom_label.upper() if report_data.freedom_label else None ), "media_type": report_data.media_type, "traffic_popularity": report_data.traffic_tier, "bias_category_description": BENCHMARK_BIAS_DESCRIPTIONS.get( report_data.bias_label ), "overall_summary": self._build_overall_summary(report_data), "history": self._build_history(report_data), "ownership": self._build_ownership(report_data), "analysis": self._build_analysis_text(report_data), "failed_fact_checks": self._build_failed_fact_checks(report_data), "last_updated": datetime.now().strftime("%B %d, %Y"), "evidence_sources": self._build_evidence_sources(report_data, articles_data), "_chain_of_thought": self._synthesize_cot(report_data) } mapped_result = _filter_evidence_sources(mapped_result, "System") logger.info(f"[System] Completed {outlet_name} in {time.time()-t0:.1f}s | bias={mapped_result.get('bias_score')}, factuality={mapped_result.get('factual_score')}") return mapped_result except Exception as e: logger.error(f"[System] {outlet_name}: {e}", exc_info=True) return {} class HybridRunner(SystemRunner): """Hybrid mode: System analyzer pipeline + intelligent search routing. Improvements over plain SystemRunner: 1. Uses HybridSearchBackend: routes site:-targeted queries (FactCheckSearcher) to DDG for reliable operator support; routes broader queries to OpenAI web_search for richer results. 2. Always uses LLM synthesis (inherits from SystemRunner with use_synthesis=True). """ def __init__(self, model_name: str = "gpt-5-mini-2025-08-07", article_cache=None, benchmark_mode: bool = True, use_synthesis: bool = True, use_calibration: bool = True): # HybridRunner always uses synthesis super().__init__(model_name, article_cache, benchmark_mode, use_synthesis=True, use_calibration=use_calibration) def _create_profiler(self) -> MediaProfiler: from search_backends import HybridSearchBackend backend = HybridSearchBackend(model=self.model_name) return MediaProfiler(model=self.model_name, search_backend=backend, use_calibration=self.use_calibration) # Convenience: runner factory RUNNER_MAP = { "llm": LLMOnlyRunner, "articles": LLMArticlesRunner, "search": LLMSearchRunner, 'mbc': MBCRunner, "system": SystemRunner, "system-synth": SystemRunner, # System pipeline + LLM synthesis "hybrid": HybridRunner, } def get_runner(mode: str, model_name: str, article_cache=None, use_calibration: bool = True) -> BaseRunner: cls = RUNNER_MAP.get(mode) if cls is None: raise ValueError(f"Unknown mode: {mode}. Choose from {list(RUNNER_MAP.keys())}") # Initialize appropriate runner if mode == "system-synth": return cls(model_name=model_name, article_cache=article_cache, use_synthesis=True, use_calibration=use_calibration) if mode in ("system", "hybrid"): return cls(model_name=model_name, article_cache=article_cache, use_calibration=use_calibration) if mode == "articles": return cls(model_name=model_name, article_cache=article_cache) return cls(model_name=model_name)