argus-cortex

Shared foundation for the Argus suite: taxonomy, versioned wire-schema tooling, and the local/remote AI backend contract.

Part of the Argus suite; SUITE.md (opens in new tab) lists every repo, what it owns, and how they fit together. This is the library the stages import so the code they share — the target taxonomy, the wire-schema versioning discipline, and the way they call local or hosted models — lives in one place instead of being copy-pasted per repo.

What's here

• argus_cortex.taxonomy — TargetProfile / TargetStyle / TargetCategory, the "moat" types every stage inherits verbatim. taxonomy_schema() renders them into the committed taxonomy-schema.json (make schema regenerates it; a test fails when it is stale), so non-Python clients generate their types instead of copying the literals.
• argus_cortex.provenance: Provenance / ProvenanceKind, where a face, voice or image came from (the shape argus-mask#19 defined). personal needs only a source and is never cleared; stock, synthetic and consented are cleared, each states its licence, and consented also names its subject and the release_sha256 of the signed release (never the document). curated is reserved and not cleared. Strict: unknown fields and a blank source are refused, licence is read as license, and a stored cleared is recomputed. A synthetic or personal source may name its generator (model, model_sha256, seed, prompt; optional, refused on other kinds). provenance_schema() renders the committed provenance-schema.json (make schema; a test fails when it is stale).
• argus_cortex.consent: a real person's likeness release, as argus-mask#26 defined it. ConsentRecord is keyed by the release_sha256 a consented Provenance names, and holds the subject, who collected it, signing and expiry dates (expiry must be stated, null for none), the age check (who and when, never the ID), the face/voice opt-ins, voice deliveries, uses (avatar, video, speech, training; training is off unless listed), sha256s of the reference images and recordings, revocation terms and any revocation. consent_valid(record, use=, likeness=, at=, delivery=) returns a ConsentVerdict (ok, reason, detail), checking in order revoked, not-yet-signed, expired, likeness-not-covered, use-not-covered, delivery-not-covered; at must be timezone-aware and dates compare as UTC days, so the check never reads the clock. consent_matches(record, provenance) checks the record is the release and subject a provenance names. A record names a person, so services return only verdicts, never records. consent_schema() renders the committed consent-schema.json.
• argus_cortex.wire — the versioned wire-schema toolkit: check_version() (major-compatibility gate), make_versioned_base() (a Pydantic base that stamps + checks a version field like proof_version / manifest_version), and wire_schema() / render_schema() for the committed-schema schema --check CLI pattern.
• argus_cortex.backends — the AI-backend contract generalised from argus-lens: Backend / LocalBackend / RemoteBackend (point at a hosted service by base_url, host/port, or a known RemoteProvider), plus with_retries() and resolve_device(). httpx lives behind the [remote] extra.
• argus_cortex.server — shared write-guard + env-flag scaffolding for the suite's FastAPI micro-servers: WriteGuard (a pure-ASGI method-gate that never buffers streaming responses), the constant_refuse() / cross_site_refuse() predicates (read-only/replay mode and cross-site-write protection are two configs of the one guard), the UNSAFE_METHODS / TRUTHY / FALSY constants, and env_flag() (warns on a mistyped protection flag rather than silently disabling it). Starlette lives behind the [server] extra.
• argus_cortex.xmp — the suite's one XMP sidecar implementation (hoisted from argus-lens): write_xmp() / read_xmp() / merge_xmp() / parse_xmp() over an XmpData model covering dc (keywords, description), xmp (Rating, Label), lr (hierarchical keywords), Iptc4xmpExt (PersonInImage), mwg-rs face/pet regions, and argus:ContentRating. Writes merge: each property you set replaces that property (latest write wins), everything else in the sidecar — Lightroom develop settings, GPS, another stage's fields — is preserved. None leaves a property alone, an empty value removes it. Writes are atomic; an unparseable existing sidecar raises XmpError instead of being clobbered. Stdlib + pydantic only.
• argus_cortex.moderation — the suite's content-moderation clients, over HTTP to servers the deployer runs (nothing here imports a detector): NudeNetClient (a NudeNet server's /infer → NsfwResult: detections, per-class maxima, and an exposed score) and SingGuardClient (a SingGuard server's /api/guard/check → GuardResult: verdict, the policy rules that fired, explanation; optional custom policy text). DEFAULT_EXPOSED_LABELS is the suite's one definition of nudity: the five exposed buttocks/breast/genitalia/anus classes — faces, covered parts, and bare feet, belly, armpits or male chest never count. Stdlib + pydantic; Pillow (for SingGuard's downscaling of large images, on by default) is behind [images]: without it SingGuardClient(...) raises ModerationConfigError at construction unless given downscale=False. A user:password@ in a URL is sent as HTTP Basic auth. URLs are passed in, never read from the environment here: each stage owns its config. NudeNet itself is AGPL-3.0, which is why it stays behind an HTTP boundary.
• argus_cortex.gpu: GPU tenancy, generalised from argus-lens. coordinator_from_env() returns a coordinator whose lease(caller=, min_vram_mb=) holds a GPU slot for a block of work. none (the default) grants every lease at once. lease takes an exclusive flock that every argus process on the host shares (one lock for the host, not per GPU), then waits until free_vram_mb() (NVML, behind [gpu]) reports min_vram_mb free across every process on the card, ComfyUI and Ollama included, which never take the lock. It raises GpuLeaseTimeout after ARGUS_GPU_LEASE_TIMEOUT_S (300 s). The VRAM check is a snapshot, not a reservation, and a probe that cannot run lets the lease through. Containers each have their own /tmp, so set ARGUS_GPU_LEASE_PATH to a file on a volume every GPU service mounts, or the lock coordinates nothing. ARGUS_GPU_DEVICE (default 0; an invalid value warns and falls back to 0) picks which card NVML probes for free VRAM; the lock stays host-wide.
• argus_cortex.store — the optional stateful services layer (the suite's persistent "memory"). External and opt-in — every store is configured by CORTEX_* env vars and no-ops when its URL is unset.
  • Phase 1 — Postgres lineage store. Persists the source_asset → caption → human_edit → dataset_membership → training_run DAG for LoRA reproducibility, edit capture, and the feedback loop. Backed by a psycopg connection pool (concurrent asyncio.to_thread calls are safe; a dropped connection is replaced), with a transaction() boundary for atomic multi-step writes. open_lineage_store() returns a no-op store when CORTEX_PG_URL is unset; psycopg lives behind the [postgres] extra.
  • Phase 2 — Qdrant vector store. Stores image + tag-set embeddings for retrieval-augmented few-shot and near-duplicate curation; put a caption_id/asset_id in a point's payload and a search hit joins straight back to the lineage. open_vector_store() no-ops when CORTEX_QDRANT_URL is unset; qdrant-client lives behind the [qdrant] extra. cortex stores vectors it's handed — computing embeddings is the caller's job.
  • Phase 3 — MinIO/S3 blob store. Owns image bytes when cortex can't rely on the filesystem/Immich (exported/selected training images). Blobs are content-addressed on sha256 (content_key() == source_asset.sha256), so storage, dedup, and lineage join share one identity. open_blob_store() no-ops when CORTEX_S3_ENDPOINT/CORTEX_S3_BUCKET are unset; the minio client lives behind the [s3] extra.

from argus_cortex.wire import make_versioned_base, render_schema
from argus_cortex.backends import RemoteBackend, RemoteProvider

# per-package versioned base (readable field name kept local, logic shared)
_Versioned = make_versioned_base("proof_version", "1.0", ("1",))

# reach a hosted scorer by IP/port, or a known provider
scorer = RemoteBackend.from_host("<LAN_HOST>", 9000, api_key="…")  # real host/IP: local .env
nim = RemoteBackend.from_provider(RemoteProvider.NVIDIA_NIM, api_key="…")

from argus_cortex.xmp import XmpData, XmpRegion, XmpRegions, write_xmp

# lens writes captions/keywords …
write_xmp("pics/beach.jpg", XmpData(keywords=["beach", "sunset"], description="a beach at sunset"))
# … curator later adds faces + a content rating to the same sidecar; lens's fields survive
write_xmp(
    "pics/beach.jpg",
    XmpData(
        persons=["Ada"],
        regions=XmpRegions(width=4000, height=3000, regions=[XmpRegion(x=0.52, y=0.4, w=0.1, h=0.18, name="Ada")]),
        content_rating="safe",
    ),
)

from argus_cortex.moderation import NudeNetClient, SingGuardClient, rule_matches

nudenet = NudeNetClient("http://nudenet:8080")                  # its origin; /infer is appended
result = nudenet.check(image_bytes, "beach.jpg")
result.exposed_score    # 0.0-1.0 over DEFAULT_EXPOSED_LABELS; result.labels = per-class maxima

guard = SingGuardClient("http://guard:9150", policy=None)  # None = SingGuard's default rules
# thinking="fast" (default) is a verdict and rule; "fast-slow" also writes why; "slow" writes why
# with no fast verdict ahead of the reasoning; "cascade" is fast, then slow only for what fast
# flagged (the last two: agent-composer's guard or argus-cull's singguard service).
verdict = guard.check(image_bytes, text="the prompt, if any")
preflight = guard.check_text("the prompt")    # no image: judge a prompt before it is run
verdict.safe, verdict.rules   # False, ["B. Real-World Crimes & Public Safety"]
any(rule_matches(r, "B") for r in verdict.rules)   # match a rule by letter or full name

from argus_cortex.moderation import SingGuardClient, policies

guard = SingGuardClient("http://guard:9150", policy=policies.policy_text("adult-granular"))
verdict = guard.check(image_bytes)
illegal = [r for r in verdict.rules if policies.is_illegal("adult-granular", r)]   # A, H or I
policies.mode("adult-granular"), policies.mode("default")   # "nsfw", "sfw"

Stages ask for a policy by name; policies holds each one's text, the rules that mean illegal content under it, and its SFW/NSFW mode (default and sfw-* are SFW). The known names are policies.POLICIES; any other name raises ModerationConfigError (an sfw-* name gets mode sfw but has no text):

adult-granular, adult-cautious and adult-cinematic are under evaluation against adult; one adult policy remains at release, beside default.

from argus_cortex.gpu import GpuLeaseTimeout, coordinator_from_env

gpu = coordinator_from_env()   # ARGUS_GPU_COORDINATOR=lease, ARGUS_GPU_LEASE_PATH=/shared/argus-gpu.lock
try:
    with gpu.lease(caller="faces", min_vram_mb=1500):   # one argus job at a time, and 1.5 GB free
        run_the_model()
except GpuLeaseTimeout:
    ...   # no slot in time: back off, or answer 503

Each check() raises CheckFailed (NudeNetCheckFailed / GuardCheckFailed) for a network error or an unusable answer, so a caller checking many images can record "unchecked" against the one image — and should treat unchecked as not safe. Bad configuration raises ModerationConfigError, whose message never echoes the URL.

from argus_cortex.store import open_lineage_store, SourceAsset, Caption, HumanEdit

# no-op if CORTEX_PG_URL is unset; PostgresLineageStore if it's set
store = open_lineage_store()          # reads CORTEX_* from the environment
store.ensure_schema()                 # idempotent bootstrap of the lineage tables

# one atomic unit — the asset + caption + edit commit together or not at all
with store.transaction():
    asset_id = store.record_asset(SourceAsset(uri="immich://…", sha256="…"))
    # lens emits a CaptionResult; cortex ingests it (cortex never imports lens)
    cap_id = store.record_caption(asset_id, Caption.from_caption_result(result, profile={...}))
    store.record_edit(cap_id, HumanEdit(edited_caption="…", editor="alice"))

# the (model_caption, edited_caption) pairs the summariser feedback loop trains on
pairs = store.caption_edit_pairs()

from argus_cortex.store import open_vector_store, IMAGE_COLLECTION

# no-op if CORTEX_QDRANT_URL is unset; QdrantVectorStore if it's set
vec = open_vector_store()
vec.ensure_collection(IMAGE_COLLECTION, dim=512)   # idempotent

# store an embedding you computed, tagged with its lineage id
vec.upsert(IMAGE_COLLECTION, point_id=cap_id, vector=embedding, payload={"caption_id": cap_id})

# retrieve similar past captions to steer the summariser (few-shot)
hits = vec.search(IMAGE_COLLECTION, vector=query_embedding, top_k=5)
similar_caption_ids = [h.payload["caption_id"] for h in hits]   # join back to Postgres

from argus_cortex.store import open_blob_store

# no-op if CORTEX_S3_ENDPOINT / CORTEX_S3_BUCKET are unset; S3BlobStore if set
blobs = open_blob_store()
blobs.ensure_bucket()

# content-addressed: the returned key IS the sha256 (== source_asset.sha256)
sha = blobs.put_content_addressed(image_bytes, content_type="image/jpeg")
data = blobs.get(sha)                    # bytes, or None if absent
link = blobs.url(sha, expires=3600)      # presigned GET URL

Install

uv pip install argus-cortex              # core (pydantic only)
uv pip install "argus-cortex[remote]"    # + httpx for RemoteBackend
uv pip install "argus-cortex[server]"    # + starlette for argus_cortex.server (WriteGuard, env_flag)
uv pip install "argus-cortex[postgres]"  # + psycopg for the lineage store
uv pip install "argus-cortex[qdrant]"    # + qdrant-client for the vector store
uv pip install "argus-cortex[s3]"        # + minio for the blob store
uv pip install "argus-cortex[images]"    # + Pillow for SingGuardClient downscaling (on by default)
uv pip install "argus-cortex[gpu]"       # + nvidia-ml-py for the GPU lease's free-VRAM probe

The stateful services are configured via CORTEX_* env vars — see .env.example (opens in new tab). Each is optional: leave a URL unset and that feature degrades to a no-op.

Develop

make install   # venv + editable install with the "dev" extras
make test
make lint

CI / Release

• CI runs via the shared argus-ci (opens in new tab) reusable workflow.
• Release publishes to PyPI (OIDC trusted publishing) on v* tags. No container image — this is a library.
• Versioning is derived from git tags via hatch-vcs — tag vX.Y.Z to cut a release.

This repo was scaffolded from argus-pkg-template (opens in new tab). Run copier update to pull template changes (CI, release, tooling).