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 (quarry → curator → lens → forge → proof). 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.
• 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 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_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("192.168.1.20", 9000, api_key="…")
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: argus-cull's singguard service only).
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).

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

The stateful services are configured via CORTEX_* env vars — see .env.example. 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 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. Run copier update to pull template changes (CI, release, tooling).