BLACKCYTE

Autonomous competitive intelligence platform for continuous monitoring, analysis, and scheduled executive brief delivery.

Tagline: Set targets, walk away, get the brief Monday morning.

Why This Exists

Most competitive intel pipelines fail because they depend on manual curation. BLACKCYTE runs the full loop automatically:

• Collect from web/news/social in parallel.
• Score and synthesize findings with LLM analysis.
• Persist evidence and embeddings for traceability.
• Deliver executive-ready email briefs on schedule.

System Overview

COLLECTION LAYER               ANALYSIS LAYER                  DELIVERY
+-------------------+          +--------------------------+     +-----------------+
| ScraperAgent      | -------> | Analyst                  | --> | Executive Brief |
| NewsAgent         | -------> | - dedup + rank           |     | HTML Email      |
| SocialAgent       | -------> | - trend detection        |     +-----------------+
+-------------------+          | - LLM synthesis          |
                               +-----------+--------------+
                                           |
                                           v
                                +--------------------------+
                                | PostgreSQL + pgvector    |
                                | findings, runs, embeddings|
                                +--------------------------+

Quick Start

1) Local run (recommended)

git clone https://github.com/unified-methods-division/blackcyte.git
cd blackcyte
cp .env.example .env
# edit .env or .env.local with DATABASE_URL and OPENAI_API_KEY at minimum
uv sync
# migrate + seed sample competitors
uv run scaffold-db
uv run uvicorn src.app:app --host 0.0.0.0 --port 8000

Health check:

curl http://localhost:8000/health

Manual run trigger:

curl -X POST http://localhost:8000/run

Database scaffold (migrate + seed sample competitors):

uv run scaffold-db

Script convention (important):

# Use script entrypoints from pyproject when available
uv run scaffold-db
# For direct module execution:
uv run python -m scripts.seed_competitors

2) Docker run

docker build -t blackcyte .
docker run --rm -p 8000:8000 --env-file .env blackcyte

API Surface

Method	Endpoint	Description
GET	/health	Service status, scheduler state, and most recent pipeline run
POST	/run	Executes one ad-hoc orchestration run immediately

Configuration

Configuration is loaded via Pydantic settings from .env and .env.local (.env.local can override .env). Full template: .env.example.

Required

Variable	Purpose
DATABASE_URL	PostgreSQL connection string
OPENAI_API_KEY	LLM analysis and embeddings

Optional (graceful degradation)

Variable	Behavior when missing
NEWS_API_KEY	News collection is skipped
YOUTUBE_API_KEY	YouTube collection is skipped
RESEND_API_KEY	Brief generation continues, email delivery is skipped

Operational defaults

Variable	Default
SCHEDULE_CRON	0 8 * * 1 (Monday 08:00 UTC)
MAX_CONCURRENT_REQUESTS	5
REQUEST_TIMEOUT_SECONDS	15.0
RELEVANCE_THRESHOLD	0.7
DEDUP_THRESHOLD	0.15

Agent Roster

Agent	Module	Mission
SCRAPER	src/agents/scraper.py	Monitors target pages and detects changes
NEWS	src/agents/news.py	Pulls keyword-targeted news coverage
SOCIAL	src/agents/social.py	Collects Reddit and YouTube signals
CLASSIFIER	src/agents/classifier.py	Performs relevance scoring
ANALYST	src/analyst.py	Deduplicates, trends, synthesizes, prepares brief content

All collectors implement the GatheringAgent protocol. The orchestrator executes them in parallel and tolerates partial source failure.

Add a Competitor Target

INSERT INTO competitors (name, website_url, keywords, monitored_pages)
VALUES (
  'Acme Corp',
  'https://acme.com',
  '["Acme Corp", "acme.com", "acme pricing"]',
  '["/pricing", "/blog", "/careers"]'
);

Seed 3 real targets (xAI, Anduril, Varick Agents):

uv run python -m scripts.seed_competitors

Notes:

• Seed script is idempotent by competitor name (safe to run multiple times).
• Seed script only requires DATABASE_URL.
• Run migrations first on fresh databases: uv run alembic upgrade head.

Extend With a New Agent

1. Add src/agents/your_agent.py.
2. Implement agent_name: str and async gather(target) -> list[FindingData].
3. Use retry policy helpers for outbound HTTP calls.
4. Register the agent in src/app.py inside run_pipeline().
5. Add tests under tests/test_agents/.

Tech Stack

• Python 3.12, asyncio
• FastAPI + Uvicorn
• PostgreSQL + pgvector
• SQLAlchemy (async), asyncpg
• APScheduler
• OpenAI API
• structlog + tenacity
• Docker

Repository Layout

src/
  agents/        # source-specific collectors + classifier
  db/            # models, sessions, migration config
  prompts/       # LLM prompt templates
  report/        # brief rendering + delivery
  analyst.py     # synthesis pipeline
  orchestrator.py
  app.py         # FastAPI entrypoint + scheduler wiring

Built By

Built by Unified Methods Division.

License

Proprietary. All rights reserved.