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.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).

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).