Real-World Examples¶
Practical, production-ready examples for common PromptBeacon use cases.
Table of Contents¶
- Keyless Demo
- Share of Voice Comparison
- Stability Scan
- CI Visibility Check
- HTML Dashboard
- Brand Monitoring Dashboard
- Competitive Intelligence System
- PR Campaign Impact Tracking
- Multi-Brand Portfolio Management
- Automated Alerting System
- Weekly Report Generation
- BeaconGuard: Brand Safety
Keyless Demo¶
Try PromptBeacon with zero API keys. Returns realistic canned data.
CLI¶
pip install promptbeacon
promptbeacon demo "Nike"
Python¶
from promptbeacon import Beacon
# Demo mode — no keys, no network calls
report = Beacon("Nike").demo().scan()
print(f"Visibility Score: {report.visibility_score}/100")
print(f"Share of Voice: {report.share_of_voice.target_share:.0%}")
print(f"Presence Rate: {report.share_of_voice.target_presence_rate:.0%}")
print(f"Rank: #{report.share_of_voice.target_rank}")
print(f"Mentions: {report.mention_count}")
print(f"Positive Sent.: {report.sentiment_breakdown.positive:.0%}")
# Competitor SoV breakdown
for brand, entry in report.share_of_voice.aggregate.items():
print(f" {brand}: {entry.share_of_voice:.0%}")
# CI assertion in demo mode
report.assert_visibility(min_score=30, min_share_of_voice=0.1)
print("Visibility thresholds met.")
Demo mode also works with the --demo flag on all relevant CLI commands:
promptbeacon scan "Nike" --demo
promptbeacon quick "Nike" --demo
promptbeacon dashboard "Nike" --demo -o demo_report.html
Share of Voice Comparison¶
Measure your brand's fraction of AI mindshare vs. competitors.
from promptbeacon import Beacon, Provider
report = (
Beacon("Nike")
.with_competitors("Adidas", "Puma", "New Balance", "Under Armour")
.with_providers(Provider.OPENAI, Provider.ANTHROPIC, Provider.GOOGLE)
.with_industry("ecommerce")
.with_prompt_count(15)
.scan()
)
sov = report.share_of_voice
print(f"\n=== Share of Voice Report ===")
print(f"Target: {sov.target_share:.0%} | Presence: {sov.target_presence_rate:.0%} | Rank: #{sov.target_rank}")
print("\nAll brands:")
for brand, entry in sorted(
sov.aggregate.items(),
key=lambda x: x[1].share_of_voice,
reverse=True,
):
bar = "#" * int(entry.share_of_voice * 40)
print(f" {brand:<20} {entry.share_of_voice:.0%} {bar}")
# Per-provider breakdown
print("\nSoV by provider:")
for provider_name, provider_sov in sov.by_provider.items():
print(f" {provider_name}: {provider_sov.target_share:.0%}")
CLI comparison¶
promptbeacon compare "Nike" \
--against "Adidas" \
--against "Puma" \
--provider openai \
--provider anthropic \
--format json > comparison.json
Stability Scan¶
Measure how consistently AI mentions your brand across repeated scan runs.
from promptbeacon import Beacon, Provider
report = (
Beacon("Nike")
.with_competitors("Adidas", "Puma")
.with_providers(Provider.OPENAI, Provider.ANTHROPIC)
.with_stability(5) # 5 independent runs
.with_temperature(0.7) # Non-zero temperature for meaningful variance
.scan_stability()
)
s = report.stability
print(f"Stability Score: {s.stability_score:.1f}/100")
print(f"Rating: {s.volatility.stability_rating}") # stable/moderate/volatile
ci = s.score_confidence_interval
print(f"95% CI: [{ci[0]:.1f}, {ci[1]:.1f}]")
print(f"Per-run scores: {[f'{x:.1f}' for x in s.score_per_run]}")
print(f"Presence consist.: {s.overall_presence_consistency:.0%}")
print(f"Flip-flop count: {s.flip_flop_count}")
print("\nPer-prompt stability (most volatile first):")
for ps in sorted(report.stability.prompt_stability, key=lambda x: x.score_variance, reverse=True)[:5]:
print(f" [{ps.presence_rate:.0%} present] {ps.prompt[:60]}...")
CLI:
# 5 stability runs
promptbeacon scan "Nike" --stability 5
# Short form
promptbeacon scan "Nike" -r 5
Cost warning: --stability 5 multiplies API calls by 5. Start with --demo to explore the output structure:
promptbeacon scan "Nike" --stability 3 --demo
CI Visibility Check¶
Gate deployments or scheduled jobs on brand health metrics.
Python Assertion API¶
#!/usr/bin/env python3
"""CI visibility gate — fails with exit code 1 if thresholds not met."""
import sys
from promptbeacon import Beacon, Provider
from promptbeacon.core.exceptions import VisibilityAssertionError
report = (
Beacon("Nike")
.with_competitors("Adidas", "Puma")
.with_providers(Provider.OPENAI, Provider.ANTHROPIC)
.scan()
)
try:
report.assert_visibility(
min_score=40,
min_share_of_voice=0.15,
min_presence_rate=0.5,
max_rank=3,
)
print(f"Visibility check PASSED (score={report.visibility_score:.1f})")
except VisibilityAssertionError as e:
print("Visibility check FAILED:")
for failure in e.failures:
print(f" - {failure}")
sys.exit(1)
pytest Plugin¶
The plugin auto-registers via the promptbeacon entry point — no import needed:
# tests/test_brand_visibility.py
import pytest
@pytest.mark.visibility(
brand="Nike",
competitors=["Adidas", "Puma"],
min_score=40,
min_share_of_voice=0.15,
demo=True, # No API keys required
)
def test_nike_visibility_thresholds():
pass
@pytest.mark.visibility(
brand="Nike",
min_score=50,
demo=True,
)
def test_nike_minimum_score():
pass
Run with:
# Demo mode (no API keys)
PROMPTBEACON_DEMO=1 pytest tests/test_brand_visibility.py -v
# With real keys
pytest tests/test_brand_visibility.py -v
Tests skip cleanly when no API keys are present and demo=True is not set.
GitHub Action¶
# .github/workflows/brand-visibility.yml
name: AI Visibility Gate
on:
push:
branches: [main]
schedule:
- cron: '0 9 * * 1' # Weekly Mondays at 9 AM
jobs:
visibility-check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Check AI visibility
uses: yotambraun/promptbeacon@v1
with:
brand: "Nike"
competitors: "Adidas Puma New Balance"
providers: "openai anthropic"
min-score: "40"
min-share-of-voice: "0.15"
stability: "3"
min-stability: "60"
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
The action exits with code 1 if any threshold fails, blocking the workflow. For demo mode in CI without keys:
with:
brand: "Nike"
demo: "true"
min-score: "40"
HTML Dashboard¶
Generate a self-contained HTML report with charts (SoV bar, sentiment donut, score breakdown, stability band, history sparkline).
from promptbeacon import Beacon, Provider, to_dashboard_html
beacon = (
Beacon("Nike")
.with_competitors("Adidas", "Puma")
.with_providers(Provider.OPENAI, Provider.ANTHROPIC)
.with_storage("~/.promptbeacon/nike.db")
)
report = beacon.scan()
history = beacon.get_history(days=30)
# Generate dashboard with history sparkline
html = to_dashboard_html(report, history=history)
with open("nike_dashboard.html", "w") as f:
f.write(html)
print("Dashboard saved to nike_dashboard.html")
CLI (auto-opens in browser):
# Demo (no keys)
promptbeacon dashboard "Nike" --demo -o report.html
# Real scan
promptbeacon dashboard "Nike" \
--competitor "Adidas" \
--provider openai \
-o nike_dashboard.html
# Skip auto-open
promptbeacon dashboard "Nike" --demo -o report.html --no-open
Brand Monitoring Dashboard¶
A complete brand monitoring solution with daily scans, historical tracking, and alerts.
#!/usr/bin/env python3
"""Brand Monitoring Dashboard
Daily brand visibility monitoring with historical tracking and alerts.
"""
import asyncio
from datetime import datetime
from pathlib import Path
from promptbeacon import Beacon, Provider, to_json, to_markdown, to_dashboard_html
from rich.console import Console
from rich.table import Table
console = Console()
class BrandMonitor:
"""Automated brand monitoring system."""
def __init__(self, brand: str, competitors: list[str], storage_path: str):
self.brand = brand
self.competitors = competitors
self.storage_path = storage_path
self.beacon = (
Beacon(brand)
.with_aliases(f"{brand} Inc", f"{brand} Corporation")
.with_competitors(*competitors)
.with_providers(Provider.OPENAI, Provider.ANTHROPIC)
.with_industry("ecommerce")
.with_cache()
.with_storage(storage_path)
)
async def run_daily_scan(self):
"""Execute daily brand scan."""
console.print(f"\n[bold]Starting daily scan for {self.brand}[/bold]")
console.print(f"Time: {datetime.now()}")
report = await self.beacon.scan_async()
self._save_reports(report)
self._check_alerts(report)
self._display_summary(report)
return report
def _save_reports(self, report):
"""Save reports in multiple formats."""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_dir = Path("./reports") / self.brand
output_dir.mkdir(parents=True, exist_ok=True)
with open(output_dir / f"report_{timestamp}.json", "w") as f:
f.write(to_json(report))
with open(output_dir / f"report_{timestamp}.md", "w") as f:
f.write(to_markdown(report))
history = self.beacon.get_history(days=30)
with open(output_dir / f"dashboard_{timestamp}.html", "w") as f:
f.write(to_dashboard_html(report, history=history))
console.print(f"Reports saved to {output_dir}")
def _check_alerts(self, report):
"""Check for alert conditions."""
comparison = self.beacon.compare_with_previous()
if comparison:
if comparison.score_change < -5:
console.print(f"[red]ALERT: Visibility dropped {comparison.score_change:.1f} points[/red]")
elif comparison.score_change > 5:
console.print(f"[green]Visibility increased {comparison.score_change:+.1f} points[/green]")
if report.sentiment_breakdown.negative > 0.2:
console.print(f"[yellow]ALERT: Negative sentiment at {report.sentiment_breakdown.negative:.0%}[/yellow]")
def _display_summary(self, report):
"""Display scan summary."""
table = Table(title=f"{self.brand} Scan Summary")
table.add_column("Metric", style="cyan")
table.add_column("Value", style="green")
sov = report.share_of_voice
table.add_row("Visibility Score", f"{report.visibility_score:.1f}/100")
table.add_row("Share of Voice", f"{sov.target_share:.0%} (rank #{sov.target_rank})")
table.add_row("Presence Rate", f"{sov.target_presence_rate:.0%}")
table.add_row("Mentions", str(report.mention_count))
table.add_row("Positive Sentiment", f"{report.sentiment_breakdown.positive:.0%}")
table.add_row("Providers", ", ".join(report.providers_used))
table.add_row("Duration", f"{report.scan_duration_seconds:.1f}s")
if report.total_cost_usd:
table.add_row("Cost", f"${report.total_cost_usd:.4f}")
console.print("\n", table)
if report.competitor_comparison:
comp_table = Table(title="Competitor Comparison")
comp_table.add_column("Brand", style="cyan")
comp_table.add_column("Score")
comp_table.add_column("SoV")
comp_table.add_column("Mentions")
comp_table.add_row(
f"[bold]{self.brand}[/bold]",
f"[bold]{report.visibility_score:.1f}[/bold]",
f"[bold]{sov.target_share:.0%}[/bold]",
f"[bold]{report.mention_count}[/bold]",
)
for name, score in report.competitor_comparison.items():
comp_entry = sov.aggregate.get(name)
sov_str = f"{comp_entry.share_of_voice:.0%}" if comp_entry else "-"
comp_table.add_row(name, f"{score.visibility_score:.1f}", sov_str, str(score.mention_count))
console.print("\n", comp_table)
def get_weekly_summary(self):
"""Generate weekly summary."""
history = self.beacon.get_history(days=7)
if not history.data_points:
console.print("No data for weekly summary")
return
console.print(f"\n[bold]Weekly Summary: {self.brand}[/bold]")
console.print(f"Average Score: {history.average_score:.1f}")
console.print(f"Trend: {history.trend_direction}")
console.print(f"Volatility: {history.volatility:.2f}")
async def main():
monitor = BrandMonitor(
brand="Nike",
competitors=["Adidas", "Puma", "New Balance"],
storage_path="~/.promptbeacon/nike.db",
)
await monitor.run_daily_scan()
monitor.get_weekly_summary()
if __name__ == "__main__":
asyncio.run(main())
Cron Setup¶
# Run daily at 8 AM
0 8 * * * /usr/bin/python3 /path/to/brand_monitor.py
Competitive Intelligence System¶
Track multiple competitors with detailed analysis and reporting.
#!/usr/bin/env python3
"""Competitive Intelligence System"""
import asyncio
from datetime import datetime
from promptbeacon import Beacon, Provider
import pandas as pd
from pathlib import Path
class CompetitiveIntelligence:
def __init__(self, brands: list[str], categories: list[str]):
self.brands = brands
self.categories = categories
self.storage_path = "~/.promptbeacon/competitive.db"
async def run_competitive_scan(self):
results = []
for brand in self.brands:
print(f"\nScanning {brand}...")
beacon = (
Beacon(brand)
.with_providers(Provider.OPENAI, Provider.ANTHROPIC)
.with_categories(*self.categories)
.with_storage(self.storage_path)
.with_prompt_count(15)
.with_competitors(*[b for b in self.brands if b != brand])
)
report = await beacon.scan_async()
sov = report.share_of_voice
results.append({
"brand": brand,
"score": report.visibility_score,
"sov": sov.target_share,
"presence_rate": sov.target_presence_rate,
"rank": sov.target_rank,
"mentions": report.mention_count,
"positive": report.sentiment_breakdown.positive,
"negative": report.sentiment_breakdown.negative,
"timestamp": report.timestamp,
})
return pd.DataFrame(results)
def generate_competitive_report(self, df: pd.DataFrame):
output_dir = Path("./competitive_reports")
output_dir.mkdir(exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
df.to_csv(output_dir / f"competitive_data_{timestamp}.csv", index=False)
report_md = f"""# Competitive Intelligence Report
Generated: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
## Executive Summary
- **Brands Analyzed**: {len(df)}
- **Categories**: {", ".join(self.categories)}
- **Market Leader (Score)**: {df.loc[df['score'].idxmax(), 'brand']} ({df['score'].max():.1f})
- **Market Leader (SoV)**: {df.loc[df['sov'].idxmax(), 'brand']} ({df['sov'].max():.0%})
## Brand Rankings
| Rank | Brand | Score | Share of Voice | Presence Rate | Positive % |
|------|-------|-------|----------------|---------------|------------|
"""
for idx, row in df.sort_values("score", ascending=False).iterrows():
rank = df.sort_values("score", ascending=False).index.get_loc(idx) + 1
report_md += f"| {rank} | {row['brand']} | {row['score']:.1f} | {row['sov']:.0%} | {row['presence_rate']:.0%} | {row['positive']:.0%} |\n"
filepath = output_dir / f"report_{timestamp}.md"
with open(filepath, "w") as f:
f.write(report_md)
print(f"\nReport generated in {output_dir}")
async def main():
ci = CompetitiveIntelligence(
brands=["Nike", "Adidas", "Puma", "New Balance", "Under Armour"],
categories=["running shoes", "athletic wear", "sports brand"],
)
df = await ci.run_competitive_scan()
ci.generate_competitive_report(df)
print("\nCompetitive Intelligence Scan Complete!")
if __name__ == "__main__":
asyncio.run(main())
PR Campaign Impact Tracking¶
Track brand visibility before, during, and after PR campaigns.
#!/usr/bin/env python3
"""PR Campaign Impact Tracker"""
from promptbeacon import Beacon
from datetime import datetime
class CampaignTracker:
def __init__(self, brand: str, campaign_start: datetime):
self.brand = brand
self.campaign_start = campaign_start
self.beacon = Beacon(brand).with_storage("~/.promptbeacon/campaigns.db")
def track_campaign(self, days_before: int = 7, days_after: int = 14):
history = self.beacon.get_history(days=days_before + days_after)
if not history.data_points:
print("Insufficient historical data")
return
pre_campaign = [dp for dp in history.data_points if dp.timestamp < self.campaign_start]
post_campaign = [dp for dp in history.data_points if dp.timestamp >= self.campaign_start]
pre_avg = sum(dp.visibility_score for dp in pre_campaign) / len(pre_campaign) if pre_campaign else 0
post_avg = sum(dp.visibility_score for dp in post_campaign) / len(post_campaign) if post_campaign else 0
impact = post_avg - pre_avg
print(f"\n{self.brand} Campaign Impact Analysis")
print("=" * 50)
print(f"Campaign Start: {self.campaign_start}")
print(f"\nPre-Campaign Average: {pre_avg:.1f}")
print(f"Post-Campaign Average: {post_avg:.1f}")
print(f"Impact: {impact:+.1f} points")
if impact > 2:
print("Status: Positive impact detected")
elif impact < -2:
print("Status: Negative impact detected")
else:
print("Status: No significant impact")
return {"pre_avg": pre_avg, "post_avg": post_avg, "impact": impact}
if __name__ == "__main__":
tracker = CampaignTracker(
brand="Nike",
campaign_start=datetime(2026, 1, 10),
)
results = tracker.track_campaign(days_before=7, days_after=14)
Multi-Brand Portfolio Management¶
#!/usr/bin/env python3
"""Multi-Brand Portfolio Manager"""
import asyncio
from promptbeacon import Beacon, Provider
from rich.console import Console
from rich.table import Table
console = Console()
class PortfolioManager:
def __init__(self, brands: dict[str, list[str]]):
self.brands = brands
self.storage_path = "~/.promptbeacon/portfolio.db"
async def scan_portfolio(self):
results = {}
for brand, competitors in self.brands.items():
console.print(f"\n[cyan]Scanning {brand}...[/cyan]")
beacon = (
Beacon(brand)
.with_competitors(*competitors)
.with_providers(Provider.OPENAI)
.with_storage(self.storage_path)
.with_prompt_count(10)
)
report = await beacon.scan_async()
results[brand] = report
return results
def display_portfolio_summary(self, results: dict):
table = Table(title="Portfolio Summary")
table.add_column("Brand", style="cyan")
table.add_column("Score")
table.add_column("SoV")
table.add_column("Rank")
table.add_column("Sentiment")
table.add_column("Status")
for brand, report in results.items():
sov = report.share_of_voice
if report.visibility_score >= 70:
status = "[green]Excellent[/green]"
elif report.visibility_score >= 50:
status = "[yellow]Good[/yellow]"
else:
status = "[red]Needs Attention[/red]"
table.add_row(
brand,
f"{report.visibility_score:.1f}",
f"{sov.target_share:.0%}",
f"#{sov.target_rank}",
f"{report.sentiment_breakdown.positive:.0%}",
status,
)
console.print("\n", table)
async def main():
portfolio = {
"Brand A": ["Competitor A1", "Competitor A2"],
"Brand B": ["Competitor B1", "Competitor B2"],
}
manager = PortfolioManager(portfolio)
console.print("[bold]Starting Portfolio Scan...[/bold]")
results = await manager.scan_portfolio()
manager.display_portfolio_summary(results)
if __name__ == "__main__":
asyncio.run(main())
Automated Alerting System¶
#!/usr/bin/env python3
"""Automated Alerting System"""
from promptbeacon import Beacon
import requests
class AlertSystem:
def __init__(self, brand: str, slack_webhook: str = None):
self.brand = brand
self.slack_webhook = slack_webhook
self.beacon = Beacon(brand).with_storage("~/.promptbeacon/alerts.db")
def monitor_and_alert(self, thresholds: dict = None):
if thresholds is None:
thresholds = {
"score_drop": -5.0,
"score_increase": 5.0,
"negative_sentiment": 0.2,
"low_score": 40.0,
"low_sov": 0.1,
}
report = self.beacon.scan()
alerts = self._check_thresholds(report, thresholds)
for alert in alerts:
self._send_alert(alert)
return alerts
def _check_thresholds(self, report, thresholds):
alerts = []
sov = report.share_of_voice
comparison = self.beacon.compare_with_previous()
if comparison:
if comparison.score_change <= thresholds["score_drop"]:
alerts.append({
"severity": "high",
"type": "score_drop",
"message": f"Visibility dropped by {abs(comparison.score_change):.1f} points",
})
elif comparison.score_change >= thresholds["score_increase"]:
alerts.append({
"severity": "info",
"type": "score_increase",
"message": f"Visibility increased by {comparison.score_change:.1f} points",
})
if report.sentiment_breakdown.negative >= thresholds["negative_sentiment"]:
alerts.append({
"severity": "medium",
"type": "negative_sentiment",
"message": f"Negative sentiment at {report.sentiment_breakdown.negative:.0%}",
})
if report.visibility_score <= thresholds["low_score"]:
alerts.append({
"severity": "high",
"type": "low_score",
"message": f"Visibility score below threshold: {report.visibility_score:.1f}",
})
if sov.target_share <= thresholds.get("low_sov", 0):
alerts.append({
"severity": "medium",
"type": "low_sov",
"message": f"Share of Voice below threshold: {sov.target_share:.0%}",
})
return alerts
def _send_alert(self, alert: dict):
print(f"\nALERT [{alert['severity'].upper()}]: {alert['message']}")
if self.slack_webhook:
payload = {"text": f"Brand Alert: {self.brand} — {alert['message']}"}
try:
requests.post(self.slack_webhook, json=payload)
except Exception as e:
print(f"Slack alert failed: {e}")
if __name__ == "__main__":
alert_system = AlertSystem(
brand="Nike",
slack_webhook="https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
)
alerts = alert_system.monitor_and_alert()
if alerts:
print(f"\n{len(alerts)} alert(s) triggered")
else:
print("\nNo alerts — all metrics within thresholds")
Weekly Report Generation¶
#!/usr/bin/env python3
"""Weekly Report Generator"""
from promptbeacon import Beacon, to_markdown, to_dashboard_html
from datetime import datetime
from pathlib import Path
class WeeklyReporter:
def __init__(self, brand: str, competitors: list[str]):
self.brand = brand
self.competitors = competitors
self.beacon = (
Beacon(brand)
.with_competitors(*competitors)
.with_storage("~/.promptbeacon/weekly.db")
)
def generate_weekly_report(self):
current_report = self.beacon.scan()
history = self.beacon.get_history(days=7)
sov = current_report.share_of_voice
report = f"""# Weekly Visibility Report: {self.brand}
**Report Period:** {datetime.now().strftime("%Y-%m-%d")}
**Generated:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
---
## Executive Summary
- **Current Visibility Score:** {current_report.visibility_score:.1f}/100
- **Share of Voice:** {sov.target_share:.0%} (rank #{sov.target_rank})
- **Presence Rate:** {sov.target_presence_rate:.0%}
- **Total Mentions:** {current_report.mention_count}
- **Positive Sentiment:** {current_report.sentiment_breakdown.positive:.0%}
"""
if history.data_points:
report += f"- **7-Day Trend:** {history.trend_direction.upper()}\n"
report += f"- **7-Day Average:** {history.average_score:.1f}\n"
report += "\n## Competitive Analysis\n\n"
report += "| Brand | Score | SoV | Mentions | Positive % |\n"
report += "|-------|-------|-----|----------|------------|\n"
report += f"| **{self.brand}** | **{current_report.visibility_score:.1f}** | **{sov.target_share:.0%}** | **{current_report.mention_count}** | **{current_report.sentiment_breakdown.positive:.0%}** |\n"
for name, score in sorted(
current_report.competitor_comparison.items(),
key=lambda x: x[1].visibility_score,
reverse=True
):
comp_entry = sov.aggregate.get(name)
sov_str = f"{comp_entry.share_of_voice:.0%}" if comp_entry else "-"
report += f"| {name} | {score.visibility_score:.1f} | {sov_str} | {score.mention_count} | {score.sentiment.positive:.0%} |\n"
if current_report.recommendations:
report += "\n## Recommendations\n\n"
for rec in current_report.recommendations[:5]:
report += f"### [{rec.priority.upper()}] {rec.action}\n\n"
report += f"**Rationale:** {rec.rationale}\n\n"
if rec.expected_impact:
report += f"**Expected Impact:** {rec.expected_impact}\n\n"
# Save markdown
output_dir = Path("./weekly_reports")
output_dir.mkdir(exist_ok=True)
filename = f"weekly_report_{self.brand}_{datetime.now().strftime('%Y%m%d')}"
with open(output_dir / f"{filename}.md", "w") as f:
f.write(report)
# Save HTML dashboard
with open(output_dir / f"{filename}.html", "w") as f:
f.write(to_dashboard_html(current_report, history=history))
print(f"Reports saved to {output_dir}")
return report
if __name__ == "__main__":
reporter = WeeklyReporter(
brand="Nike",
competitors=["Adidas", "Puma", "New Balance"],
)
report = reporter.generate_weekly_report()
print(report)
BeaconGuard: Brand Safety¶
Real-time brand safety analysis for LLM outputs — no API calls needed.
Basic Guard¶
from promptbeacon import BeaconGuard
guard = BeaconGuard("Nike", competitors=["Adidas", "Puma"])
responses = [
"I recommend Nike for running shoes. Great quality and innovation.",
"Try Adidas instead — Nike has had quality issues lately.",
"Popular running brands include Brooks, Asics, and Saucony.",
]
for response in responses:
result = guard.analyze(response)
print(f"Risk: {result.risk_level} | Flags: {result.flags}")
Guard as Middleware¶
from promptbeacon import BeaconGuard
from promptbeacon.integrations.middleware import BeaconGuardMiddleware
guard = BeaconGuard("Acme", competitors=["CompetitorX"])
middleware = BeaconGuardMiddleware(
guard,
on_high_risk=lambda r: print(f"ALERT: {r.flags}")
)
result = middleware("I'd suggest CompetitorX over Acme for this use case.")
print(f"Risk: {result.risk_level}")
See the examples directory for runnable scripts: guard_example.py and langchain_guard.py.
See Also¶
- Quickstart Guide - Getting started
- API Reference - Complete API documentation
- Advanced Usage - Advanced patterns
- CLI Reference - Command-line tools
- Storage Guide - Data management