Propaganda Pipeline — Usage Tutorial¶

This notebook shows how to use propaganda_pipeline.py to detect propaganda techniques in text using LLMs (via DSPy + OpenAI).

Pipeline overview¶

Text
 │
 ▼
Analisis.run()          ← runs all 23 technique detectors in sequence
 │  (or run_consistency() for robustness via multiple trials)
 ▼
candidates list         ← one dict per technique: answer, confidence, span, rationale
 │
 ▼
RunJuez()               ← LLM judge scores candidates, synthesizes final answer
 │
 ▼
result dict             ← judged, topk, synthesis, selected_models_posthoc
 │
 ├── show_rubric_table_from_result()   ← per-technique rubric scores table
 ├── synthesis_report()               ← narrative summary card
 └── visualize_spans()                ← highlighted text with color-coded spans

Detected techniques (23 total)¶

#	Name	Description
1	REPETITION	Persuasion through repetition of words/phrases
2	EXAGERATION	Hyperbole or minimization without evidence
3	OBFUSCATION	Intentional vagueness or confusion
4	LOADED_LANGUAGE	Emotionally charged terms without reasoning
5	WHATABOUTISM	Deflecting criticism by pointing to others
6	KAIROS	Exploiting timing/urgency to manipulate
7	KILLER	Conversation killers that suppress discussion
8	SLIPPERY	Slippery slope fallacy
9	SLOGAN	Brief striking phrases used as sole argument
10	VALUES	Appeal to abstract values instead of evidence
11	RED_HERRING	Introducing irrelevant information
12	STRAWMAN	Misrepresenting someone’s position
13	FEAR	Appeal to fear or prejudice
14	AUTHORITY	Fallacious appeal to authority
15	BANDWAGON	Appeal to popularity
16	CASTING_DOUBT	Attacking credibility instead of the argument
17	FLAG_WAVING	Appeal to group pride/identity
18	SMEAR_POISONING	Poisoning the well / reputation attacks
19	TU_QUOQUE	“You do it too” / appeal to hypocrisy
20	GUILT_BY_ASSOCIATION	Discrediting by association
21	NAME_CALLING	Derogatory labeling
22	CAUSAL_OVERSIMPLIFICATION	Attributing complex outcomes to a single cause
23	FALSE_DILEMMA	Presenting only two options when more exist

1. Setup¶

Import the pipeline components and configure the LLM. Replace YOUR_OPENAI_API_KEY with your key or load it from an environment variable.

[ ]:

import os
from propaganda_pipeline import (
    Configuracion,
    Analisis,
    run_consistency,
    RunJuez,
    visualize_spans,
    synthesis_report,
    show_rubric_table_from_result,
)

# Configure the LLM — gpt-4o-2024-08-06 is recommended for best accuracy
cfg = Configuracion(
    model_name="gpt-4o-2024-08-06",
    api_key=os.environ.get("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY"),
)
cfg.setup()
print("LLM configured successfully.")

2. Define the input text¶

The pipeline works on a single string. It can be in English or Spanish.

[ ]:

# Example: False Dilemma + Appeal to Fear
TEXT = (
    "Referring to your claim that providing medicare for all citizens would be costly "
    "and a danger to the free market, I infer that you don't care if people die from "
    "not having healthcare, so we are not going to support your endeavour."
)

print(f"Text ({len(TEXT)} chars):")
print(TEXT)

3. Run the detectors¶

Option A — Single run (`Analisis.run`)¶

Fast, but may produce inconsistent results across runs due to LLM stochasticity. Use this during development or for quick tests.

[ ]:

analisis = Analisis()

# Single run — returns a list of 23 candidate dicts
candidates_single = analisis.run(TEXT)

# Quick peek at the structure of one candidate
import json
example = candidates_single[0]
print(json.dumps({k: v for k, v in example.items() if k != "raw"}, indent=2, ensure_ascii=False))

Option B — Consistency run (`run_consistency`)¶

Runs the full analysis trials times and consolidates results by detection ratio. A technique is reported as detected only if it appears in at least threshold fraction of trials.

trials=5, threshold=1.0 → technique must be detected in all 5 runs (strictest)
trials=5, threshold=0.6 → technique must be detected in at least 3 of 5 runs

Each consolidated candidate has an extra ratio field.

[ ]:

# Recommended for production — runs 5 trials and requires unanimous agreement
candidates = run_consistency(analisis, TEXT, trials=5, threshold=1.0)

# Show techniques detected (answer = "Sí") with their ratio
detected = [(c["model"], c["ratio"], c["confidence"]) for c in candidates if c["answer"] == "Sí"]
print(f"Detected {len(detected)} technique(s):")
for model, ratio, conf in detected:
    print(f"  {model:35s}  ratio={ratio:.2f}  confidence={conf:.2f}")

4. Inspect individual candidates¶

Each candidate dict has these standard keys:

Key	Type	Description
`model`	str	Technique name (e.g. `"FALSE_DILEMMA"`)
`answer`	str	`"Sí"` (detected) or `"No"`
`confidence`	float	Model certainty in [0, 1]
`ratio`	float	Detection rate across trials (consistency mode only)
`rationale_summary`	str	Brief explanation of the decision
`span`	list[str]	Literal text excerpts supporting the detection
`labels`	dict	Technique-specific metadata
`raw`	dict	Full Pydantic model dump

[ ]:

# Print all detected candidates with their rationale and spans
for c in candidates:
    if c["answer"] != "Sí":
        continue
    print(f"{'='*60}")
    print(f"Technique : {c['model']}")
    print(f"Answer    : {c['answer']}  (ratio={c.get('ratio', '-'):.2f}, confidence={c['confidence']:.2f})")
    print(f"Rationale : {c['rationale_summary']}")
    print(f"Spans     : {c['span']}")

5. Run the judge (`RunJuez`)¶

RunJuez takes the candidates list and runs a three-step LLM evaluation:

Judge — scores each candidate against the DEFAULT_RUBRIC (consistency × confidence × yes-preference) and selects which techniques to include.
Synthesizer — generates a natural-language final answer listing all confirmed techniques.
Post-hoc selector — maps the synthesis back to specific model names with per-technique evidence.

The result dict keys:

Key	Description
`judged`	Rubric scores for each candidate and list of selected indices
`topk`	Selected candidate dicts (include=True in rubric)
`synthesis`	Final consolidated answer + brief reasoning
`selected_models_posthoc`	Detected technique names + per-model evidence
`elapsed_sec`	Wall-clock time for the judge pipeline

[ ]:

result = RunJuez(TEXT, candidates, confidence_threshold=0.9)

print(f"Judge completed in {result['elapsed_sec']:.1f}s")
print(f"\nFinal answer:\n{result['synthesis']['final_answer']}")
print(f"\nReasoning:\n{result['synthesis']['brief_reasoning']}")
print(f"\nSelected models: {result['selected_models_posthoc']['selected_models']}")

6. Visualizations¶

6.1 Rubric scores table¶

Shows per-technique scores (consistency, confidence, yes-preference) and whether the technique was included.

[ ]:

show_rubric_table_from_result(
    result,
    candidates=candidates,
    title="Rubric scores by technique"
)

6.2 Synthesis report card¶

A formatted HTML card summarizing all detected techniques, the final answer, and per-technique evidence.

[ ]:

html_card = synthesis_report(
    TEXT,
    judged=result["judged"],
    topk=result["topk"],
    synthesis=result["synthesis"],
    selection=result["selected_models_posthoc"],
    display_inline=True
)

6.3 Highlighted span visualization¶

Color-codes the original text by technique. Each color corresponds to a detected propaganda technique. Overlapping regions show a gradient of multiple colors.

[ ]:

selection = result["selected_models_posthoc"]

html_spans = visualize_spans(
    TEXT,
    candidates,
    selection,
    display_inline=True,
    show_legend="bottom",
    show_badges=False,
    boxed=True,
    title="Detected spans",
    icon="⚖️",
    accent_color="#111827"
)

7. Running on multiple texts¶

To analyze a batch of texts, simply loop over them. Each call to run_consistency + RunJuez is independent.

[ ]:

TEXTS = {
    "false_dilemma": (
        "Either you support our policy completely or you want the country to fail. "
        "There is no middle ground."
    ),
    "appeal_to_authority": (
        "Dr. Johnson, one of the world's leading economists, confirmed that "
        "this tax cut will definitely grow the economy. Therefore it's true."
    ),
    "name_calling": (
        "These radical socialist extremists want to destroy everything we've built. "
        "Don't listen to their dangerous, un-American agenda."
    ),
}

results_batch = {}
for label, text in TEXTS.items():
    print(f"Analyzing: {label}...")
    cands = run_consistency(analisis, text, trials=3, threshold=0.67)  # faster: 3 trials
    res = RunJuez(text, cands)
    selected = res["selected_models_posthoc"]["selected_models"]
    results_batch[label] = selected
    print(f"  → {selected}")

print("\nBatch complete.")

8. Using a subset of techniques¶

You can instantiate Analisis with only the detectors you need, which speeds up the pipeline significantly.

[ ]:

from propaganda_pipeline import FalseDilemmaRunner, StrawmanRunner, FearPrejudiceRunner

# Only run three specific detectors
analisis_subset = Analisis(techniques=[FalseDilemmaRunner, StrawmanRunner, FearPrejudiceRunner])

cands_subset = analisis_subset.run(TEXT)
for c in cands_subset:
    print(f"{c['model']:25s}  answer={c['answer']}  confidence={c['confidence']:.2f}")

9. Accessing raw LLM output¶

Each candidate’s raw field contains the full Pydantic model dump from the LLM, including all technique-specific fields.

[ ]:

# Find the False Dilemma candidate and inspect its raw output
fd_candidate = next((c for c in candidates if c["model"] == "FALSE_DILEMMA"), None)

if fd_candidate:
    raw = fd_candidate["raw"]
    print(f"is_fd                : {raw.get('is_fd')}")
    print(f"options_extracted    : {raw.get('options_extracted')}")
    print(f"third_options_possible: {raw.get('third_options_possible')}")
    print(f"erases_continuum     : {raw.get('erases_continuum')}")
    print(f"justification        : {raw.get('justification')}")
    print(f"confidence           : {raw.get('confidence')}")
else:
    print("FALSE_DILEMMA not in candidates list.")

10. Quick reference — API summary¶

# 1. Configure
cfg = Configuracion(model_name="gpt-4o-2024-08-06", api_key="...")
cfg.setup()

# 2. Detect
analisis = Analisis()                                # all 23 detectors
analisis = Analisis(techniques=[SloganRunner, ...])  # subset

candidates = analisis.run(TEXT)                      # single run
candidates = run_consistency(                        # multi-trial
    analisis, TEXT, trials=5, threshold=1.0
)

# 3. Judge
result = RunJuez(TEXT, candidates, confidence_threshold=0.9)

# result keys:
#   result["judged"]                  → rubric scores
#   result["topk"]                    → included candidates
#   result["synthesis"]               → final_answer, brief_reasoning
#   result["selected_models_posthoc"] → selected_models, evidence, confidence
#   result["elapsed_sec"]             → timing

# 4. Visualize
selection = result["selected_models_posthoc"]

show_rubric_table_from_result(result, candidates=candidates, title="...")

synthesis_report(
    TEXT,
    judged=result["judged"],
    topk=result["topk"],
    synthesis=result["synthesis"],
    selection=selection,
    display_inline=True
)

visualize_spans(
    TEXT, candidates, selection,
    display_inline=True, show_legend="bottom",
    boxed=True, title="Detected spans"
)