diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS new file mode 100644 index 0000000..ab89f97 --- /dev/null +++ b/.github/CODEOWNERS @@ -0,0 +1 @@ +* @scope-maintainers diff --git a/.github/workflows/backend-main.yaml b/.github/workflows/backend-main.yaml index 925ac3d..d1ca827 100644 --- a/.github/workflows/backend-main.yaml +++ b/.github/workflows/backend-main.yaml @@ -41,6 +41,10 @@ jobs: - intelligence-worker - stateful-eval-worker - cognitive-failure-worker + - cognition-protocols-worker + - ingestion-worker + - semantic-analysis-worker + - eval-worker permissions: contents: read # required for actions/checkout packages: write # required to push the image to GHCR diff --git a/.github/workflows/backend-verify.yaml b/.github/workflows/backend-verify.yaml index 161a372..a1b9b96 100644 --- a/.github/workflows/backend-verify.yaml +++ b/.github/workflows/backend-verify.yaml @@ -30,6 +30,10 @@ jobs: - intelligence-worker - stateful-eval-worker - cognitive-failure-worker + - cognition-protocols-worker + - ingestion-worker + - semantic-analysis-worker + - eval-worker permissions: contents: read # required for actions/checkout packages: read diff --git a/README.md b/README.md index d72e402..c4659bb 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ SCOPE observes, analyzes and explains the executions of Multi-Agent Systems (MAS Run the whole stack on your machine with Docker Compose, send it the trace of a small two-agent example, and look at it in the dashboard. It takes about 15 minutes, most of it the first image build. -You need Docker (Compose v2.17 or later, with at least **12 GB** of memory given to it) and [`uv`](https://docs.astral.sh/uv/) to run the example. An OpenAI-compatible LLM endpoint is only needed for the analyses that call a model: without it, traces are still collected and stored in the knowledge graph. +You need Docker (Compose v2.17 or later, with at least **6 GB** of memory given to it) and [`uv`](https://docs.astral.sh/uv/) to run the example. An OpenAI-compatible LLM endpoint is only needed for the analyses that call a model: without it, traces are still collected and stored in the knowledge graph. ### 1. Start SCOPE @@ -19,7 +19,7 @@ cp .env.template .env # set LLM_MODEL_NAME, EMBEDDING_MODEL_NAME, LLM_GAT docker compose up -d --build ``` -### 2. Send a trace** +### 2. Send a trace ```bash cd ../../examples/currency-demo diff --git a/backend/CHANGELOG.md b/backend/CHANGELOG.md index 74e9a3b..d000c6e 100644 --- a/backend/CHANGELOG.md +++ b/backend/CHANGELOG.md @@ -63,6 +63,60 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **Workers can run in one process, and Compose runs three services instead of eight.** `worker_base.runner` starts the + workers of a group in threads of one process, each with its own event loop, queues and options, so the interpreter and + the libraries they share are loaded once. The new images are `ingestion-worker` (norm, embedding), + `semantic-analysis-worker` (grouping, hierarchical-grouping, analysis, intelligence) and `eval-worker` (mce, + stateful-eval); the analysis worker, which forks a pool of CPU-bound processes, runs as a child process of its group. + `--workers` (or `INGESTION_WORKERS`, `SEMANTIC_ANALYSIS_WORKERS`, `EVAL_WORKERS`) picks some of them, and the settings + that all the workers read from one variable can be set per worker with its prefix (`NORM_MAX_INFLIGHT_MESSAGES`, + `HIERARCHICAL_GROUPING_PERIODIC_TRIGGER_INTERVAL_SECONDS`; `scoped_getenv` in `worker_base.utils`), and + `_ARGS` adds command line options. The single-worker images and CLIs are unchanged (and still needed for Argo + run-once mode); the Helm chart still deploys the single workers. On a 33-session replay the three services took 1.6 GB + against 2.0 GB for the eight containers, and the pipeline finished in 85 s against 101 s. +- **The MCE can ignore the metric plugins of some packages (`MCE_IGNORED_PLUGIN_PACKAGES`).** It loads every package that + registers a metric in the `metrics_computation_engine.plugins` entry-point group and skips the ones that are not MCE + metrics, but loading is importing: in an image that also has the stateful evaluation (`eval-worker`), the MCE imported + `mdt-metrics` and `litellm` with it, which took its process from 102 MB to 388 MB after discovery. The `eval-worker` image + sets it to `mdt-metrics`; the metric catalog is the same (45 metrics). +- **The stateful evaluation calls the judge model with the OpenAI SDK, not LiteLLM.** `metrics-computation-engine`'s + `LLMClient` is a thin wrapper around `litellm.completion`, and importing it (which `BaseMetric` and `Jury` do) loaded + LiteLLM, about 300 MB per process. `stateful_evals_be.llm_client.LLMClient` has the same constructor and `query` + signature and calls the `openai` SDK (Azure OpenAI when `api_version` is set); it replaces the package's module when + `stateful_evals_be` is imported. A parameter the endpoint refuses (`reasoning_effort`, `temperature`, `response_format`) + is dropped and the call retried, as LiteLLM's `drop_params` did. LiteLLM is still installed, since + `metrics-computation-engine` depends on it, but no longer imported. +- **Insights from the cognitive failure detections.** Two intelligence-catalog templates read the + `CognitiveFailureReport` nodes: one insight per failure and session (priority from the detector's confidence), and one + per failure and semantic group when it is recurring (at least 2 sessions and 30% of the group, medium or high + confidence). +- **Fixed: the intelligence worker failed with `Insight failed ontology validation: concerns: Less than 1 values`.** + `ingest_insights` validated each insight without its `concerns` edge to the node it is attached to, which the + ontology requires, so no insight could be written. It now passes that edge. +- **`cognition-protocols-worker` depends on `ioc-scope-mce-core`, not the `ioc-scope-mce` meta package.** It only runs the + ACCORD and CONCORD plugin metrics, but the meta package installs every provider (native, sdk, deepeval, opik, ragas, + legacy), with `litellm`, `pyarrow`, `scipy`, `scikit-learn` and `pandas` behind them. The image's virtual environment goes + from 788 MB to 233 MB; the same 49 protocol metrics are discovered. +- **`clickhouse-sqlalchemy` is an optional extra of `ioc-scope-api`.** Nothing imports it (span reads use + `clickhouse-connect`), but it pulled `asynch` (66 MB) into every worker image. The API image still installs it + (`clickhouse-sqlalchemy` extra), for a `SQLAlchemyConnector` pointed at a ClickHouse URL. +- **`umap-learn` and `numba` are only installed in the API image.** They serve the UMAP projection of + `GET /sessions/{id}/latent-space`, but `ioc-scope-api` is also the knowledge-graph client of every worker, so all the worker + images carried them, with llvmlite, scikit-learn and scipy behind them. They are now the `latent-space` extra of + `ioc-scope-api`, which the API Dockerfile installs (`uv sync --extra latent-space`). The norm worker image goes from + 970 MB to 494 MB, and no worker image has numba, llvmlite or umap any more (three Dockerfiles no longer need their + `--no-install-package numba` workaround). Without the extra, the latent-space endpoint answers 501 and names it. +- **The embedding worker image no longer installs torch.** `dem.embedding` imported torch, transformers, + sentence-transformers, gensim and google-genai as soon as it was imported, so the worker paid about 360 MB of memory for + libraries that `OpenAIEmbedder` (the embedder of the Compose file and the Helm chart) never uses. Each embedder now imports + its library when it is built. `dem[embedding]` is the API embedders only (openai, google-genai); the local models + (`SentenceTransformerEmbedder`, `Qwen3Embedder`, `GensimEmbedder`) moved to the new `dem[embedding-local]` extra, and the + embedding worker has a matching `local` extra, which `Dockerfile.gpu` installs. The default image goes from 1.54 GB to + 0.72 GB and the process from about 470 MB to about 105 MB. A local embedder without its libraries fails with a message + that names the extra. To use one in the default image, build a variant with `--extra local`. +- **The cognition-protocols image builds without an SSH agent.** The ACCORD and CONCORD metric packages are read from the + public `ioc-project/cognition-protocols` repository over HTTPS (`ssh: [default]` is gone from the Compose file and the bake + target), and the image is part of the CI matrix like the others. - **Breaking: metrics are stored as the ontology's `Metric` + `MetricResult`.** A `MetricResult` (`result`, `reasoning`, `timestamp`, `evaluation*` cost fields) is linked to its `Metric` definition (one per metric name and provider, with `metricProvider`, `unit`, `valueRange`) by `ofMetricType`, and diff --git a/backend/api/Dockerfile b/backend/api/Dockerfile index ed9670a..06a16b6 100644 --- a/backend/api/Dockerfile +++ b/backend/api/Dockerfile @@ -25,14 +25,14 @@ COPY --from=ontology . /ontology # Install dependencies without installing source code yet. RUN --mount=type=cache,target=/root/.cache/uv \ - cd /src && uv sync --frozen --no-dev --no-install-project --no-install-workspace --no-sources --project /src/api + cd /src && uv sync --frozen --no-dev --no-install-project --no-install-workspace --no-sources --extra latent-space --extra clickhouse-sqlalchemy --project /src/api # Copy the API source code. COPY api /src/api # Install the ioc-scope-api package RUN --mount=type=cache,target=/root/.cache/uv \ - cd /src && uv sync --frozen --no-dev --no-editable --project /src/api + cd /src && uv sync --frozen --no-dev --no-editable --extra latent-space --extra clickhouse-sqlalchemy --project /src/api FROM python:3.13-slim-trixie AS runner diff --git a/backend/api/pyproject.toml b/backend/api/pyproject.toml index 6673a39..5679c6d 100644 --- a/backend/api/pyproject.toml +++ b/backend/api/pyproject.toml @@ -10,23 +10,30 @@ readme = "README.md" requires-python = ">=3.13" dependencies = [ "clickhouse-connect>=0.11.0", - "clickhouse-sqlalchemy>=0.3.2", "fastapi>=0.129.0", "neo4j>=6.1.0", "packaging>=26.0", - "numba>=0.60.0", "numpy>=2.3.5", "pydantic-settings>=2.13.1", "python-dotenv>=1.2.1", "requests>=2.32.5", "redis>=5.0.0", "sqlalchemy>=2.0.46", - "umap-learn>=0.5.7", "uvicorn>=0.41.0", "ruff<=0.15.22", "ioc-scope-ontology[models,verify]>=1.1.0", ] +[project.optional-dependencies] +# The SQLAlchemy dialect for ClickHouse (and its asynch driver), for a SQLAlchemyConnector pointed at a ClickHouse URL. +# Nothing imports it by default: the span reads use clickhouse-connect. Only the API image installs it. +clickhouse-sqlalchemy = ["clickhouse-sqlalchemy>=0.3.2"] +# The UMAP projection behind GET /sessions/{id}/latent-space. It pulls in numba and llvmlite, so only the API image +# installs it: the workers depend on this package for the knowledge-graph client and do not need a projection. +latent-space = [ + "umap-learn>=0.5.7", + "numba>=0.60.0", +] [build-system] @@ -45,7 +52,9 @@ dev = [ "httpx>=0.28.1", "ioc-scope-mce-provider-native", "ioc-scope-mce-provider-sdk", + "numba>=0.60.0", "pytest>=9.0.2", + "umap-learn>=0.5.7", ] [tool.uv] diff --git a/backend/api/scope/client/dal.py b/backend/api/scope/client/dal.py index 7ce485c..4443455 100644 --- a/backend/api/scope/client/dal.py +++ b/backend/api/scope/client/dal.py @@ -13,6 +13,7 @@ from scope_ontology import OntologyValidationError, verify_kg_object from scope_ontology.models.edges import ( aboutMetric, + concerns, containsSession, hasContext, hasMedioidSession, @@ -1835,7 +1836,14 @@ def ingest_insights( current_ids_by_target_template: Dict[Tuple[str, str, str], set] = {} for insight in insights: - verification = verify_kg_object(insight) + # ``concerns`` is mandatory on an Insight. It is not stored (the target's ``hasInsight`` edge below is what + # the graph keeps), so it is asserted here for the shape to see the target the insight is attached to. + concerns_edges = ( + [concerns(source_id=insight.id, target_id=insight.targetNodeId)] + if insight.targetNodeId + else [] + ) + verification = verify_kg_object(insight, edges=concerns_edges) if not verification.conforms: raise OntologyValidationError(verification) diff --git a/backend/api/scope/client/sessions.py b/backend/api/scope/client/sessions.py index 4cf4817..1cb8c24 100644 --- a/backend/api/scope/client/sessions.py +++ b/backend/api/scope/client/sessions.py @@ -1476,7 +1476,14 @@ def _get_session_latent_space( """Fetch state embeddings and project them to 2D using UMAP.""" import numpy as np from fastapi import HTTPException - from umap import UMAP + + try: + from umap import UMAP + except ImportError as exc: # the `latent-space` extra is only installed in the API image + raise HTTPException( + status_code=501, + detail="The latent space needs umap-learn: install ioc-scope-api[latent-space].", + ) from exc query, params = ui_queries.latent_space_query(session_id=session_id) try: diff --git a/backend/api/tests/test_ingest_insights.py b/backend/api/tests/test_ingest_insights.py new file mode 100644 index 0000000..25ef260 --- /dev/null +++ b/backend/api/tests/test_ingest_insights.py @@ -0,0 +1,58 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""``ingest_insights`` validates an Insight against its shape, which requires the ``concerns`` edge to its target.""" + +import datetime +from unittest.mock import MagicMock + +import pytest +from scope_ontology import OntologyValidationError +from scope_ontology.models.nodes.insight import Insight + +from scope.client import dal + + +def _insight(**overrides) -> Insight: + fields = { + "id": "insight-1", + "dataType": "text", + "templateId": "high_cost_agent", + "description": "Agent X spent 3x the median cost.", + "scope": "Agent", + "labels": "[]", + "targetNodeId": "agent-x", + "createdAt": datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), + } + fields.update(overrides) + return Insight(**fields) + + +@pytest.fixture() +def graph(monkeypatch): + nodes, rels = [], [] + monkeypatch.setattr( + dal, "_ingest_node", lambda db, label, props: nodes.append(props) + ) + monkeypatch.setattr(dal, "_create_rel", lambda *args: rels.append(args)) + return nodes, rels + + +def test_an_insight_with_a_target_is_written_and_attached(graph): + nodes, rels = graph + db = MagicMock() + + dal.ingest_insights(db, [_insight()]) + + assert [n["id"] for n in nodes] == ["insight-1"] + assert rels[0][1:4] == ("Agent", ["agentName"], ["agent-x"]) + assert rels[0][-1] == "hasInsight" + db.execute.assert_called_once() # clean-up of the outdated insights of this template + + +def test_an_insight_without_a_target_is_rejected(graph): + nodes, _ = graph + + with pytest.raises(OntologyValidationError): + dal.ingest_insights(MagicMock(), [_insight(targetNodeId="")]) + + assert nodes == [] diff --git a/backend/api/tests/test_latent_space_extra.py b/backend/api/tests/test_latent_space_extra.py new file mode 100644 index 0000000..7d692e1 --- /dev/null +++ b/backend/api/tests/test_latent_space_extra.py @@ -0,0 +1,20 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""The UMAP projection is an optional extra of the API: without it the endpoint says so instead of failing.""" + +from __future__ import annotations + +import sys + +import pytest +from fastapi import HTTPException + +from scope.client import sessions + + +def test_latent_space_without_umap_answers_501(monkeypatch): + monkeypatch.setitem(sys.modules, "umap", None) # importing it raises ImportError, as when it is not installed + with pytest.raises(HTTPException) as raised: + sessions._get_session_latent_space(None, None, session_id="s1") + assert raised.value.status_code == 501 + assert "ioc-scope-api[latent-space]" in raised.value.detail diff --git a/backend/dem/pyproject.toml b/backend/dem/pyproject.toml index 404c355..190737d 100644 --- a/backend/dem/pyproject.toml +++ b/backend/dem/pyproject.toml @@ -39,13 +39,17 @@ grouping = [ llm = [ "openai>=2.8.1", ] +# Embedders that call an API (OpenAI-compatible gateway, Google). This is what the embedding worker image installs. embedding = [ + "openai>=2.8.1", + "google-genai>=1.51.0", +] +# Embedders that run a model in the process (sentence-transformers, Qwen3, gensim): they pull in torch. +embedding-local = [ "sentence-transformers>=5.1.2", "transformers>=4.57.1", "torch>=2.0.0", "gensim>=4.4.0", - "openai>=2.8.1", - "google-genai>=1.51.0", ] analysis = [ "scikit-learn>=1.7.2,<1.9.0", @@ -55,6 +59,7 @@ all = [ "dem[grouping]", "dem[llm]", "dem[embedding]", + "dem[embedding-local]", "dem[analysis]", ] dev = [ diff --git a/backend/dem/src/dem/embedding/embedding.py b/backend/dem/src/dem/embedding/embedding.py index 1afd76f..09cd8d2 100644 --- a/backend/dem/src/dem/embedding/embedding.py +++ b/backend/dem/src/dem/embedding/embedding.py @@ -6,16 +6,66 @@ from collections import OrderedDict from typing import Optional -import gensim.downloader import numpy as np -import torch -from google import genai -from openai import OpenAI -from sentence_transformers import SentenceTransformer -from transformers import AutoModel, AutoTokenizer from .cache import CacheManager +# The embedders import their client library when they are built, not when this module is: the local models +# need torch and transformers (the ``embedding-local`` extra), which an API-only deployment does not install and +# should not pay about 400 MB of memory to import. ``_lazy`` and the module ``__getattr__`` resolve the names +# below on first use, so ``dem.embedding.embedding.OpenAI`` and the like still work (and can be patched). +_EXTRAS = { + "gensim": "embedding-local", + "torch": "embedding-local", + "SentenceTransformer": "embedding-local", + "AutoModel": "embedding-local", + "AutoTokenizer": "embedding-local", + "genai": "embedding", + "OpenAI": "embedding", +} + + +def __getattr__(name: str): + try: + if name == "gensim": + import gensim.downloader # noqa: F401 (makes gensim.downloader available) + + return gensim + if name == "torch": + import torch + + return torch + if name == "SentenceTransformer": + from sentence_transformers import SentenceTransformer + + return SentenceTransformer + if name in ("AutoModel", "AutoTokenizer"): + import transformers + + return getattr(transformers, name) + if name == "genai": + from google import genai + + return genai + if name == "OpenAI": + from openai import OpenAI + + return OpenAI + except ImportError as exc: + raise ImportError( + f"{name} is needed by this embedder and is not installed: " + f"install dem[{_EXTRAS[name]}]. ({exc})" + ) from exc + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + + +def _lazy(name: str): + """Resolve one of the optional libraries above, honouring a value set on the module (a test's patch).""" + try: + return globals()[name] + except KeyError: + return __getattr__(name) + class EmbeddingModel: """Base class for all embedding models.""" @@ -85,7 +135,8 @@ def get_word_embedding_in_context( class SentenceTransformerEmbedder(EmbeddingModel): def __init__(self, model_id, cache_manager=None): super().__init__(model_id, cache_manager) - self.sbert_model = SentenceTransformer(self.model_id) + torch = _lazy("torch") + self.sbert_model = _lazy("SentenceTransformer")(self.model_id) self.tokenizer = self.sbert_model._modules["0"].tokenizer self.transformer_model = self.sbert_model._modules["0"].auto_model @@ -140,6 +191,7 @@ def _normalize_offset_mapping(self, raw_offset): def get_word_embedding_in_context( self, text: str, char_start: int, char_end: int ) -> np.ndarray: + torch = _lazy("torch") # Tokenize the input text with offset mapping tokens = self.tokenizer.encode_plus( text, @@ -185,7 +237,7 @@ def get_word_embedding_in_context( class GensimEmbedder(EmbeddingModel): def __init__(self, model_id, cache_manager=None): super().__init__(model_id, cache_manager) - self.model = gensim.downloader.load(self.model_id) + self.model = _lazy("gensim").downloader.load(self.model_id) self.embedding_dimension = self.model.vector_size def _internal_encode(self, sentences: list[str]) -> np.ndarray: @@ -238,7 +290,7 @@ def __init__( ) # Create client with optional base_url (for gateway support) - self.client = OpenAI( + self.client = _lazy("OpenAI")( api_key=api_key_to_use, base_url=base_url_to_use, max_retries=max_retries, @@ -283,7 +335,7 @@ def __init__(self, model_id, cache_manager=None): api_key = os.getenv("GEMINI_API_KEY") if not api_key: raise ValueError("Google API key not found.") - self.client = genai.Client(api_key=api_key) + self.client = _lazy("genai").Client(api_key=api_key) # Avoid eager network/API-key validation at startup. # Dimension is discovered lazily on first real encode call. self.embedding_dimension = None @@ -328,6 +380,8 @@ def __init__( super().__init__(model_id, cache_manager) self.batch_size = batch_size try: + torch = _lazy("torch") + AutoModel, AutoTokenizer = _lazy("AutoModel"), _lazy("AutoTokenizer") # Keep device as a simple string for test determinism when torch.device is patched device_obj = torch.device("cuda" if torch.cuda.is_available() else "cpu") try: @@ -360,6 +414,7 @@ def __init__( raise Exception(f"Failed to load Qwen3 model {model_id}: {e}") def _internal_encode(self, sentences: list[str]) -> np.ndarray: + torch = _lazy("torch") all_sentence_embeddings = [] # Process sentences in smaller batches to manage memory for i in range(0, len(sentences), self.batch_size): @@ -424,6 +479,7 @@ def _internal_encode(self, sentences: list[str]) -> np.ndarray: def get_word_embedding_in_context( self, text: str, char_start: int, char_end: int ) -> np.ndarray: + torch = _lazy("torch") tokens = self.tokenizer.encode_plus( text, return_offsets_mapping=True, diff --git a/backend/dem/src/dem/tests/test_embedding/test_lazy_imports.py b/backend/dem/src/dem/tests/test_embedding/test_lazy_imports.py new file mode 100644 index 0000000..57756b0 --- /dev/null +++ b/backend/dem/src/dem/tests/test_embedding/test_lazy_imports.py @@ -0,0 +1,57 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""The embedders import their client library when they are built, so an API-only install can use dem.embedding.""" + +import subprocess +import sys +import textwrap + + +def _run(code: str) -> str: + return subprocess.run( + [sys.executable, "-c", textwrap.dedent(code)], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + + +def test_importing_the_package_loads_no_model_library(): + loaded = _run( + """ + import sys + import dem.embedding + heavy = ("torch", "transformers", "sentence_transformers", "gensim", "openai", "google.genai") + print(",".join(name for name in heavy if name in sys.modules)) + """ + ) + assert loaded == "" + + +def test_api_embedder_works_without_the_local_model_libraries(): + out = _run( + """ + import sys + for name in ("torch", "transformers", "sentence_transformers", "gensim"): + sys.modules[name] = None # importing them raises ImportError, as when they are not installed + from dem.embedding import OpenAIEmbedder + OpenAIEmbedder("text-embedding-3-small", api_key="key", base_url="http://localhost") + print("torch" in sys.modules and sys.modules["torch"] is not None) + """ + ) + assert out == "False" + + +def test_local_embedder_without_its_libraries_says_which_extra_to_install(): + out = _run( + """ + import sys + sys.modules["torch"] = None + from dem.embedding import SentenceTransformerEmbedder + try: + SentenceTransformerEmbedder("all-MiniLM-L6-v2") + except ImportError as exc: + print(exc) + """ + ) + assert "dem[embedding-local]" in out diff --git a/backend/intelligence-catalog/insight-templates/group/recurring-cognitive-failure-group.json b/backend/intelligence-catalog/insight-templates/group/recurring-cognitive-failure-group.json new file mode 100644 index 0000000..85abcf9 --- /dev/null +++ b/backend/intelligence-catalog/insight-templates/group/recurring-cognitive-failure-group.json @@ -0,0 +1,23 @@ +{ + "nameTemplate": "Recurring $failureName ($code) in group \"$groupName\"", + "descriptionTemplate": "The cognitive failure $failureName ($code) was detected, with medium or high confidence, in $affected of the $groupSize sessions of group \"$groupName\" ($sharePercent%).", + "kgQuery": [ + "MATCH (sg:SemanticGroup)-[:containsSession]->(s:Session)", + "WITH sg, count(DISTINCT s) AS groupSize", + "MATCH (sg)-[:containsSession]->(failed:Session)<-[:reportsOn]-(r:CognitiveFailureReport)", + "WHERE r.confidence IN ['medium', 'high']", + "WITH sg, groupSize, r.code AS code, head(collect(r.name)) AS reportName, count(DISTINCT failed) AS affected", + "WHERE affected >= 2 AND toFloat(affected) / groupSize >= 0.3", + "RETURN sg.id AS groupId,", + " sg.groupName AS groupName,", + " code,", + " coalesce(reportName, code) AS failureName,", + " affected,", + " groupSize,", + " toInteger(round(100.0 * affected / groupSize)) AS sharePercent" + ], + "labels": ["Cognitive Failure", "Reliability", "$code"], + "scope": "SemanticGroup", + "priority": "P2", + "targetNodeId": "$groupId" +} diff --git a/backend/intelligence-catalog/insight-templates/session/cognitive-failure-session.json b/backend/intelligence-catalog/insight-templates/session/cognitive-failure-session.json new file mode 100644 index 0000000..0dfb5fe --- /dev/null +++ b/backend/intelligence-catalog/insight-templates/session/cognitive-failure-session.json @@ -0,0 +1,21 @@ +{ + "nameTemplate": "$failureName ($code) detected in session $sessionId", + "descriptionTemplate": "The cognitive failure detector found $failureName ($code) in session $sessionId, with $confidence confidence, at $occurrences point(s) of the trace.\nReasoning: $reasoning", + "kgQuery": [ + "MATCH (r:CognitiveFailureReport)-[:reportsOn]->(s:Session)", + "WITH s, r, CASE r.confidence WHEN 'high' THEN 3 WHEN 'medium' THEN 2 ELSE 1 END AS rank", + "ORDER BY rank DESC", + "WITH s, r.code AS code, collect(r)[0] AS best, count(r) AS occurrences", + "RETURN s.sessionId AS sessionId,", + " code,", + " coalesce(best.name, code) AS failureName,", + " best.confidence AS confidence,", + " coalesce(best.reasoning, '') AS reasoning,", + " occurrences,", + " CASE best.confidence WHEN 'high' THEN 'P1' WHEN 'medium' THEN 'P2' ELSE 'P3' END AS priority" + ], + "labels": ["Cognitive Failure", "$code"], + "scope": "Session", + "priority": "$priority", + "targetNodeId": "$sessionId" +} diff --git a/backend/mce/mce-core/src/mce/core/registry/discovery.py b/backend/mce/mce-core/src/mce/core/registry/discovery.py index e3a713c..7a8d8fe 100644 --- a/backend/mce/mce-core/src/mce/core/registry/discovery.py +++ b/backend/mce/mce-core/src/mce/core/registry/discovery.py @@ -13,6 +13,7 @@ from collections import defaultdict import logging +import os import pkgutil import importlib import inspect @@ -33,6 +34,15 @@ # Entry-point group name (same as telemetry-hub for cross-compatibility) _PLUGIN_GROUP = "metrics_computation_engine.plugins" +# Packages whose plugins are not loaded, comma separated. A venv that runs the MCE next to another package that +# registers plugins in the same group, for metrics of a different engine (the stateful evaluation's), would otherwise +# import them, and everything they bring, only for the MCE to skip them. +_IGNORED_PLUGIN_PACKAGES_ENV = "MCE_IGNORED_PLUGIN_PACKAGES" + + +def _normalize_package_name(name: str) -> str: + return name.strip().lower().replace("_", "-").replace(".", "-") + # Module-level cache: None means "not yet computed", list means already computed. _DISCOVERY_CACHE: list[Metric] | None = None _DISCOVERY_LOCK = threading.Lock() @@ -652,8 +662,17 @@ def _discover_plugin_metrics() -> list[Metric]: ensuring cross-compatibility between both projects. """ plugins: list[Metric] = [] + ignored = { + _normalize_package_name(name) + for name in os.environ.get(_IGNORED_PLUGIN_PACKAGES_ENV, "").split(",") + if name.strip() + } eps = entry_points(group=_PLUGIN_GROUP) for ep in eps: + package = getattr(getattr(ep, "dist", None), "name", None) + if ignored and isinstance(package, str) and _normalize_package_name(package) in ignored: + logger.debug("Plugin %s of %s is ignored (%s)", ep.name, package, _IGNORED_PLUGIN_PACKAGES_ENV) + continue try: cls = ep.load() if not inspect.isclass(cls): diff --git a/backend/mce/tests/core/test_plugin_discovery.py b/backend/mce/tests/core/test_plugin_discovery.py index fde3d8e..0f7b2e7 100644 --- a/backend/mce/tests/core/test_plugin_discovery.py +++ b/backend/mce/tests/core/test_plugin_discovery.py @@ -145,3 +145,40 @@ def test_abstract_metric_is_skipped(): result = _discover_plugin_metrics() assert result == [] + + +# --------------------------------------------------------------------------- +# _discover_plugin_metrics — packages whose plugins are ignored +# --------------------------------------------------------------------------- + + +def _entry_point(name: str, package: str, cls): + ep = MagicMock(spec=EntryPoint) + ep.name = name + ep.value = f"{package}.module:{name}" + ep.dist = MagicMock() + ep.dist.name = package + ep.load.return_value = cls + return ep + + +def test_plugins_of_an_ignored_package_are_not_even_loaded(monkeypatch): + """Loading a plugin imports its package: the ones that are ignored must not be loaded at all.""" + monkeypatch.setenv("MCE_IGNORED_PLUGIN_PACKAGES", "mdt_metrics, other") + ignored = _entry_point("Groundedness", "mdt-metrics", _StubMetric) + kept = _entry_point("StubMetric", "my-plugins", _StubMetric) + + with patch("mce.core.registry.discovery.entry_points", return_value=[ignored, kept]): + result = _discover_plugin_metrics() + + assert len(result) == 1 + ignored.load.assert_not_called() + kept.load.assert_called_once() + + +def test_every_plugin_is_loaded_when_none_is_ignored(monkeypatch): + monkeypatch.delenv("MCE_IGNORED_PLUGIN_PACKAGES", raising=False) + eps = [_entry_point("A", "mdt-metrics", _StubMetric), _entry_point("B", "my-plugins", _StubMetric)] + + with patch("mce.core.registry.discovery.entry_points", return_value=eps): + assert len(_discover_plugin_metrics()) == 2 diff --git a/backend/pyproject.toml b/backend/pyproject.toml index db2690c..3018096 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -43,6 +43,9 @@ members = [ "workers/cognitive-failure-worker", "workers/cognition-protocols-worker", "workers/intelligence-worker", + "workers/ingestion-worker", + "workers/semantic-analysis-worker", + "workers/eval-worker", "worker-base", "stateful-evals", ] @@ -78,6 +81,9 @@ stateful-eval-worker = { workspace = true } cognition-protocols-worker = { workspace = true } stateful-evals = { workspace = true } intelligence-worker = { workspace = true } +ingestion-worker = { workspace = true } +semantic-analysis-worker = { workspace = true } +eval-worker = { workspace = true } worker-base = { workspace = true } [dependency-groups] diff --git a/backend/stateful-evals/pyproject.toml b/backend/stateful-evals/pyproject.toml index 5b04659..97f6a81 100644 --- a/backend/stateful-evals/pyproject.toml +++ b/backend/stateful-evals/pyproject.toml @@ -14,9 +14,9 @@ readme = "README.md" requires-python = ">=3.11" dependencies = [ "metrics-computation-engine>=1.2.7", - # Judge calls go through LiteLLM; releases before 1.83 emit Pydantic serializer warnings on every response. - "litellm>=1.83.0", "mdt-metrics", + # Judge calls go through the OpenAI SDK (stateful_evals_be.llm_client), not LiteLLM, which loads ~300 MB. + "openai>=1.0.0", "pydantic>=2.0.0", ] diff --git a/backend/stateful-evals/stateful_evals_be/__init__.py b/backend/stateful-evals/stateful_evals_be/__init__.py index f6ef600..149e348 100644 --- a/backend/stateful-evals/stateful_evals_be/__init__.py +++ b/backend/stateful-evals/stateful_evals_be/__init__.py @@ -4,6 +4,11 @@ from typing import TYPE_CHECKING, Any +from stateful_evals_be import llm_client as _llm_client + +# Before anything imports metrics_computation_engine: its LLMClient would load LiteLLM. +_llm_client.install() + if TYPE_CHECKING: from stateful_evals_be.api import ( EvaluationEngine, diff --git a/backend/stateful-evals/stateful_evals_be/evaluation/inter_step_review.py b/backend/stateful-evals/stateful_evals_be/evaluation/inter_step_review.py index 64cb726..3f03033 100644 --- a/backend/stateful-evals/stateful_evals_be/evaluation/inter_step_review.py +++ b/backend/stateful-evals/stateful_evals_be/evaluation/inter_step_review.py @@ -17,18 +17,14 @@ import re from typing import Any, Dict, List -import litellm -from metrics_computation_engine.llm_judge.llm import LLMClient - from stateful_evals_be.evaluation.span_judge import SpanJudge +from stateful_evals_be.llm_client import LLMClient from stateful_evals_be.models.requests import FailureDetail, MetricFailure from stateful_evals_be.prompts.inter_step_review import ( INTER_STEP_REVIEW_SYSTEM_PROMPT, INTER_STEP_REVIEW_USER_PROMPT, ) -litellm.drop_params = True - _MAX_FAILURES_PER_BATCH = 15 diff --git a/backend/stateful-evals/stateful_evals_be/evaluation/processor.py b/backend/stateful-evals/stateful_evals_be/evaluation/processor.py index c0b23a4..407bacf 100644 --- a/backend/stateful-evals/stateful_evals_be/evaluation/processor.py +++ b/backend/stateful-evals/stateful_evals_be/evaluation/processor.py @@ -25,8 +25,8 @@ from functools import partial from typing import TYPE_CHECKING, Any, Dict, List, Optional -from metrics_computation_engine.llm_judge.llm import LLMClient from metrics_computation_engine.models.requests import LLMJudgeConfig +from stateful_evals_be.llm_client import LLMClient from stateful_evals_be.evaluation.cost_estimator import ( estimate_token_cost, estimate_usage_cost_usd, diff --git a/backend/stateful-evals/stateful_evals_be/evaluation/trajectory_context_metrics/judge.py b/backend/stateful-evals/stateful_evals_be/evaluation/trajectory_context_metrics/judge.py index 8688dda..7144c7c 100644 --- a/backend/stateful-evals/stateful_evals_be/evaluation/trajectory_context_metrics/judge.py +++ b/backend/stateful-evals/stateful_evals_be/evaluation/trajectory_context_metrics/judge.py @@ -12,8 +12,8 @@ from datetime import datetime, timezone from typing import Any, Protocol -from metrics_computation_engine.llm_judge.llm import LLMClient +from stateful_evals_be.llm_client import LLMClient from stateful_evals_be.evaluation.trajectory_context import TrajectoryContext from .base import EvidenceRef, HighLevelMetricFailure, MetricInputConfig diff --git a/backend/stateful-evals/stateful_evals_be/llm_client.py b/backend/stateful-evals/stateful_evals_be/llm_client.py new file mode 100644 index 0000000..f6ad38d --- /dev/null +++ b/backend/stateful-evals/stateful_evals_be/llm_client.py @@ -0,0 +1,86 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""OpenAI-compatible judge client, in place of the LiteLLM one of ``metrics-computation-engine``. + +``metrics_computation_engine.llm_judge.llm.LLMClient`` is a thin wrapper around ``litellm.completion`` and importing it +loads LiteLLM, about 300 MB of memory per process. The judge only talks to one OpenAI-compatible gateway, so this client +calls the ``openai`` SDK directly, with the same constructor and ``query`` signature. + +``install()`` registers it as ``metrics_computation_engine.llm_judge.llm``, so the ``Jury`` and ``BaseMetric`` classes +of that package (which import ``LLMClient`` when they are imported) use it too. +""" + +import sys +import threading +import types +from typing import Any, Dict, List, Optional + +_LLM_MODULE = "metrics_computation_engine.llm_judge.llm" + +# Parameters the judge adds when it can, and that some models refuse. LiteLLM's ``drop_params`` removed them silently. +_OPTIONAL_PARAMS = ("reasoning_effort", "temperature", "response_format") + + +class LLMClient: + """One model behind an OpenAI-compatible endpoint (Azure OpenAI when ``api_version`` is set).""" + + def __init__(self, model_config: Dict[str, Any]) -> None: + self.model = model_config.get("LLM_MODEL_NAME", "") + self.api_base = model_config["LLM_BASE_MODEL_URL"] + self.api_key = model_config["LLM_API_KEY"] + self.api_version = model_config.get("api_version", "") + self.num_retries = int(model_config.get("NUM_LLM_RETRIES", 3)) + self._client: Optional[Any] = None + self._lock = threading.Lock() + + def _get_client(self) -> Any: + # Built on the first query: the key may be empty when the client is only constructed (tests, no gateway). + with self._lock: + if self._client is None: + import openai + + if self.api_version: + self._client = openai.AzureOpenAI( + azure_endpoint=self.api_base, + api_key=self.api_key, + api_version=self.api_version, + max_retries=self.num_retries, + ) + else: + self._client = openai.OpenAI( + base_url=self.api_base or None, + api_key=self.api_key, + max_retries=self.num_retries, + ) + return self._client + + def query(self, messages: List[Dict[str, str]], **kwargs: Any) -> Any: + """Run one chat completion and return the SDK response (``choices``, ``usage``).""" + import openai + + client = self._get_client() + try: + return client.chat.completions.create( + model=self.model, messages=messages, **kwargs + ) + except openai.BadRequestError as exc: + # The endpoint rejected an optional parameter: retry without it, as LiteLLM's drop_params did. + refused = [ + name for name in _OPTIONAL_PARAMS if name in kwargs and name in str(exc) + ] + if not refused: + raise + kept = {k: v for k, v in kwargs.items() if k not in refused} + return client.chat.completions.create( + model=self.model, messages=messages, **kept + ) + + +def install() -> None: + """Make ``metrics_computation_engine.llm_judge.llm`` resolve to this client, unless LiteLLM's is already loaded.""" + if _LLM_MODULE in sys.modules: + return + module = types.ModuleType(_LLM_MODULE) + module.__doc__ = __doc__ + module.LLMClient = LLMClient # type: ignore[attr-defined] + sys.modules[_LLM_MODULE] = module diff --git a/backend/stateful-evals/stateful_evals_be/scripts/compare_span_context_vs_deepeval.py b/backend/stateful-evals/stateful_evals_be/scripts/compare_span_context_vs_deepeval.py index f8d13c9..fe2aecd 100644 --- a/backend/stateful-evals/stateful_evals_be/scripts/compare_span_context_vs_deepeval.py +++ b/backend/stateful-evals/stateful_evals_be/scripts/compare_span_context_vs_deepeval.py @@ -21,7 +21,7 @@ os.environ.setdefault("DEEPEVAL_TELEMETRY_OPT_OUT", "YES") -from metrics_computation_engine.llm_judge.llm import LLMClient +from stateful_evals_be.llm_client import LLMClient from stateful_evals_be.evaluation.trajectory_context_metrics import ( PAPER_METRIC_SUITE, diff --git a/backend/stateful-evals/stateful_evals_be/tests/test_llm_client.py b/backend/stateful-evals/stateful_evals_be/tests/test_llm_client.py new file mode 100644 index 0000000..1c00619 --- /dev/null +++ b/backend/stateful-evals/stateful_evals_be/tests/test_llm_client.py @@ -0,0 +1,89 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +import subprocess +import sys +from unittest.mock import patch + +import httpx +import openai +import pytest + +from stateful_evals_be.llm_client import LLMClient + +CONFIG = { + "LLM_MODEL_NAME": "gpt-4o", + "LLM_BASE_MODEL_URL": "https://llm.example/v1", + "LLM_API_KEY": "key", +} + + +def _bad_request(message: str) -> openai.BadRequestError: + request = httpx.Request("POST", "https://llm.example/v1/chat/completions") + response = httpx.Response(400, request=request) + return openai.BadRequestError(message, response=response, body=None) + + +def test_query_calls_the_openai_sdk(): + with patch("openai.OpenAI") as factory: + create = factory.return_value.chat.completions.create + create.return_value = "response" + client = LLMClient(CONFIG) + assert ( + client.query([{"role": "user", "content": "hi"}], max_tokens=5) + == "response" + ) + factory.assert_called_once_with( + base_url="https://llm.example/v1", api_key="key", max_retries=3 + ) + create.assert_called_once_with( + model="gpt-4o", messages=[{"role": "user", "content": "hi"}], max_tokens=5 + ) + + +def test_a_refused_optional_parameter_is_dropped_and_retried(): + with patch("openai.OpenAI") as factory: + create = factory.return_value.chat.completions.create + create.side_effect = [ + _bad_request("Unsupported parameter: reasoning_effort"), + "ok", + ] + assert LLMClient(CONFIG).query([], reasoning_effort="low", max_tokens=5) == "ok" + assert create.call_args_list[1].kwargs == { + "model": "gpt-4o", + "messages": [], + "max_tokens": 5, + } + + +def test_another_bad_request_is_raised(): + with patch("openai.OpenAI") as factory: + factory.return_value.chat.completions.create.side_effect = _bad_request( + "context too long" + ) + with pytest.raises(openai.BadRequestError): + LLMClient(CONFIG).query([], reasoning_effort="low") + + +def test_api_version_selects_azure(): + with patch("openai.AzureOpenAI") as factory: + LLMClient({**CONFIG, "api_version": "2024-06-01"}).query([]) + assert factory.call_args.kwargs["api_version"] == "2024-06-01" + + +def test_a_client_without_a_key_is_built_without_a_gateway(): + LLMClient({**CONFIG, "LLM_API_KEY": ""}) + + +def test_importing_the_library_and_the_metrics_base_does_not_load_litellm(): + code = ( + "import sys, stateful_evals_be\n" + "from metrics_computation_engine.metrics.base import BaseMetric\n" + "from metrics_computation_engine.llm_judge.jury import Jury\n" + "from stateful_evals_be.llm_client import LLMClient\n" + "assert Jury({'LLM_BASE_MODEL_URL': '', 'LLM_API_KEY': ''}).llms[0].__class__ is LLMClient\n" + "assert 'litellm' not in sys.modules\n" + ) + result = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True + ) + assert result.returncode == 0, result.stderr diff --git a/backend/uv.lock b/backend/uv.lock index 1a01c99..400966d 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -16,8 +16,10 @@ members = [ "consistency-worker", "dem", "embedding-worker", + "eval-worker", "grouping-worker", "hierarchical-grouping-worker", + "ingestion-worker", "intelligence-worker", "ioc-scope-api", "ioc-scope-lib", @@ -36,6 +38,7 @@ members = [ "norm", "norm-worker", "normal-behaviour-worker", + "semantic-analysis-worker", "stateful-eval-worker", "stateful-evals", "worker-base", @@ -640,8 +643,8 @@ dependencies = [ { name = "ioc-accord-mce-metrics" }, { name = "ioc-concord-mce-metrics" }, { name = "ioc-scope-api" }, - { name = "ioc-scope-mce" }, { name = "ioc-scope-mce-client" }, + { name = "ioc-scope-mce-core" }, { name = "python-dotenv" }, { name = "pyyaml" }, { name = "worker-base" }, @@ -656,11 +659,11 @@ dev = [ [package.metadata] requires-dist = [ { name = "click", specifier = ">=8.0" }, - { name = "ioc-accord-mce-metrics", git = "ssh://git@github.com/cisco-eti/ioc-core-l9-extensions.git?subdirectory=evals%2Faccord_custom_evals%2Faccord-mce-metrics&branch=feat%2Fadding-mce-metrics-plugin" }, - { name = "ioc-concord-mce-metrics", git = "ssh://git@github.com/ioc-project/cognition-protocols.git?subdirectory=protocols%2Fconcord%2Fmce-metrics&branch=main" }, + { name = "ioc-accord-mce-metrics", git = "https://github.com/ioc-project/cognition-protocols.git?subdirectory=protocols%2Faccord%2Fmce-metrics&branch=main" }, + { name = "ioc-concord-mce-metrics", git = "https://github.com/ioc-project/cognition-protocols.git?subdirectory=protocols%2Fconcord%2Fmce-metrics&branch=main" }, { name = "ioc-scope-api", editable = "api" }, - { name = "ioc-scope-mce", editable = "mce/mce-meta" }, { name = "ioc-scope-mce-client", editable = "mce/mce-client" }, + { name = "ioc-scope-mce-core", editable = "mce/mce-core" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0" }, { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.21.0" }, { name = "python-dotenv", specifier = ">=1.0" }, @@ -1015,9 +1018,11 @@ dev = [ { name = "pytest-mock" }, ] embedding = [ - { name = "gensim" }, { name = "google-genai" }, { name = "openai" }, +] +embedding-local = [ + { name = "gensim" }, { name = "sentence-transformers" }, { name = "torch" }, { name = "transformers" }, @@ -1041,9 +1046,10 @@ dev = [ requires-dist = [ { name = "dem", extras = ["analysis"], marker = "extra == 'all'", editable = "dem" }, { name = "dem", extras = ["embedding"], marker = "extra == 'all'", editable = "dem" }, + { name = "dem", extras = ["embedding-local"], marker = "extra == 'all'", editable = "dem" }, { name = "dem", extras = ["grouping"], marker = "extra == 'all'", editable = "dem" }, { name = "dem", extras = ["llm"], marker = "extra == 'all'", editable = "dem" }, - { name = "gensim", marker = "extra == 'embedding'", specifier = ">=4.4.0" }, + { name = "gensim", marker = "extra == 'embedding-local'", specifier = ">=4.4.0" }, { name = "google-genai", marker = "extra == 'embedding'", specifier = ">=1.51.0" }, { name = "hdbscan", marker = "extra == 'grouping'", specifier = ">=0.8.41" }, { name = "neo4j", specifier = ">=5.28.2" }, @@ -1058,11 +1064,11 @@ requires-dist = [ { name = "scikit-learn", marker = "extra == 'analysis'", specifier = ">=1.7.2,<1.9.0" }, { name = "scikit-learn", marker = "extra == 'grouping'", specifier = ">=1.7.2,<1.9.0" }, { name = "scipy", marker = "extra == 'analysis'", specifier = ">=1.0.0" }, - { name = "sentence-transformers", marker = "extra == 'embedding'", specifier = ">=5.1.2" }, - { name = "torch", marker = "extra == 'embedding'", specifier = ">=2.0.0" }, - { name = "transformers", marker = "extra == 'embedding'", specifier = ">=4.57.1" }, + { name = "sentence-transformers", marker = "extra == 'embedding-local'", specifier = ">=5.1.2" }, + { name = "torch", marker = "extra == 'embedding-local'", specifier = ">=2.0.0" }, + { name = "transformers", marker = "extra == 'embedding-local'", specifier = ">=4.57.1" }, ] -provides-extras = ["grouping", "llm", "embedding", "analysis", "all", "dev"] +provides-extras = ["grouping", "llm", "embedding", "embedding-local", "analysis", "all", "dev"] [package.metadata.requires-dev] dev = [ @@ -1138,12 +1144,16 @@ dev = [ { name = "pytest-mock" }, { name = "ruff" }, ] +local = [ + { name = "dem", extra = ["embedding-local"] }, +] [package.metadata] requires-dist = [ { name = "black", marker = "extra == 'dev'", specifier = ">=23.0.0" }, { name = "click", specifier = ">=8.0.0" }, { name = "dem", extras = ["embedding"], editable = "dem" }, + { name = "dem", extras = ["embedding-local"], marker = "extra == 'local'", editable = "dem" }, { name = "ioc-scope-api", editable = "api" }, { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.0.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" }, @@ -1154,6 +1164,30 @@ requires-dist = [ { name = "ruff", marker = "extra == 'dev'", specifier = "<=0.15.22" }, { name = "worker-base", editable = "worker-base" }, ] +provides-extras = ["local", "dev"] + +[[package]] +name = "eval-worker" +version = "1.0.0" +source = { editable = "workers/eval-worker" } +dependencies = [ + { name = "mce-worker" }, + { name = "stateful-eval-worker" }, + { name = "worker-base" }, +] + +[package.optional-dependencies] +dev = [ + { name = "pytest" }, +] + +[package.metadata] +requires-dist = [ + { name = "mce-worker", editable = "workers/mce-worker" }, + { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" }, + { name = "stateful-eval-worker", editable = "workers/stateful-eval-worker" }, + { name = "worker-base", editable = "worker-base" }, +] provides-extras = ["dev"] [[package]] @@ -1655,6 +1689,30 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b3/55/ecca97ae19075f1fac62def77731e7f535e6c1fb8f92ff08160c5e6dade8/importlib_metadata-9.0.1-py3-none-any.whl", hash = "sha256:bba5600596a7e21f3eef53281cf28d6a5195634d2f2b78ff9501a3272c6eaab0", size = 27920, upload-time = "2026-08-28T15:30:33.433Z" }, ] +[[package]] +name = "ingestion-worker" +version = "1.0.0" +source = { editable = "workers/ingestion-worker" } +dependencies = [ + { name = "embedding-worker" }, + { name = "norm-worker" }, + { name = "worker-base" }, +] + +[package.optional-dependencies] +dev = [ + { name = "pytest" }, +] + +[package.metadata] +requires-dist = [ + { name = "embedding-worker", editable = "workers/embedding-worker" }, + { name = "norm-worker", editable = "workers/norm-worker" }, + { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" }, + { name = "worker-base", editable = "worker-base" }, +] +provides-extras = ["dev"] + [[package]] name = "iniconfig" version = "2.3.1" @@ -1735,7 +1793,7 @@ provides-extras = ["dev"] [[package]] name = "ioc-accord-mce-metrics" version = "0.1.0" -source = { git = "ssh://git@github.com/cisco-eti/ioc-core-l9-extensions.git?subdirectory=evals%2Faccord_custom_evals%2Faccord-mce-metrics&branch=feat%2Fadding-mce-metrics-plugin#2b796d8869d9220f6883e4906da746083b022e97" } +source = { git = "https://github.com/ioc-project/cognition-protocols.git?subdirectory=protocols%2Faccord%2Fmce-metrics&branch=main#01a5efd471ce77e5897bddfcaeac056b86ed0263" } dependencies = [ { name = "oxp-mce-core" }, ] @@ -1743,7 +1801,7 @@ dependencies = [ [[package]] name = "ioc-concord-mce-metrics" version = "0.1.0" -source = { git = "ssh://git@github.com/ioc-project/cognition-protocols.git?subdirectory=protocols%2Fconcord%2Fmce-metrics&branch=main#d68e1c4835150aa18b0d818b72b3eed61f285682" } +source = { git = "https://github.com/ioc-project/cognition-protocols.git?subdirectory=protocols%2Fconcord%2Fmce-metrics&branch=main#01a5efd471ce77e5897bddfcaeac056b86ed0263" } dependencies = [ { name = "openai" }, { name = "oxp-mce-core" }, @@ -1756,11 +1814,9 @@ version = "1.0.0" source = { editable = "api" } dependencies = [ { name = "clickhouse-connect" }, - { name = "clickhouse-sqlalchemy" }, { name = "fastapi" }, { name = "ioc-scope-ontology", extra = ["models", "verify"] }, { name = "neo4j" }, - { name = "numba" }, { name = "numpy" }, { name = "packaging" }, { name = "pydantic-settings" }, @@ -1769,27 +1825,37 @@ dependencies = [ { name = "requests" }, { name = "ruff" }, { name = "sqlalchemy" }, - { name = "umap-learn" }, { name = "uvicorn" }, ] +[package.optional-dependencies] +clickhouse-sqlalchemy = [ + { name = "clickhouse-sqlalchemy" }, +] +latent-space = [ + { name = "numba" }, + { name = "umap-learn" }, +] + [package.dev-dependencies] dev = [ { name = "fakeredis" }, { name = "httpx" }, { name = "ioc-scope-mce-provider-native" }, { name = "ioc-scope-mce-provider-sdk" }, + { name = "numba" }, { name = "pytest" }, + { name = "umap-learn" }, ] [package.metadata] requires-dist = [ { name = "clickhouse-connect", specifier = ">=0.11.0" }, - { name = "clickhouse-sqlalchemy", specifier = ">=0.3.2" }, + { name = "clickhouse-sqlalchemy", marker = "extra == 'clickhouse-sqlalchemy'", specifier = ">=0.3.2" }, { name = "fastapi", specifier = ">=0.129.0" }, { name = "ioc-scope-ontology", extras = ["models", "verify"], editable = "../ontology" }, { name = "neo4j", specifier = ">=6.1.0" }, - { name = "numba", specifier = ">=0.60.0" }, + { name = "numba", marker = "extra == 'latent-space'", specifier = ">=0.60.0" }, { name = "numpy", specifier = ">=2.3.5" }, { name = "packaging", specifier = ">=26.0" }, { name = "pydantic-settings", specifier = ">=2.13.1" }, @@ -1798,9 +1864,10 @@ requires-dist = [ { name = "requests", specifier = ">=2.32.5" }, { name = "ruff", specifier = "<=0.15.22" }, { name = "sqlalchemy", specifier = ">=2.0.46" }, - { name = "umap-learn", specifier = ">=0.5.7" }, + { name = "umap-learn", marker = "extra == 'latent-space'", specifier = ">=0.5.7" }, { name = "uvicorn", specifier = ">=0.41.0" }, ] +provides-extras = ["clickhouse-sqlalchemy", "latent-space"] [package.metadata.requires-dev] dev = [ @@ -1808,7 +1875,9 @@ dev = [ { name = "httpx", specifier = ">=0.28.1" }, { name = "ioc-scope-mce-provider-native", editable = "mce/mce-provider-native" }, { name = "ioc-scope-mce-provider-sdk", editable = "mce/mce-provider-sdk" }, + { name = "numba", specifier = ">=0.60.0" }, { name = "pytest", specifier = ">=9.0.2" }, + { name = "umap-learn", specifier = ">=0.5.7" }, ] [[package]] @@ -3283,7 +3352,7 @@ wheels = [ [[package]] name = "oxp-mce-core" version = "2.0.0rc1" -source = { git = "ssh://git@github.com/outshift-open/observe-and-explain-platform.git?subdirectory=backend%2Fmce%2Fmce-core&branch=main#87c8c8a39ea0420e647d94650deec66607f191c6" } +source = { git = "https://github.com/outshift-open/observe-and-explain-platform.git?subdirectory=backend%2Fmce%2Fmce-core&branch=main#e154d84d2e3f5e5d78c347083044dd3d4a354852" } dependencies = [ { name = "click" }, { name = "networkx" }, @@ -4119,6 +4188,34 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/50/a8/6a77f5f267c555108f0a864b6db714363dab567a8266422a79a385f9232b/scipy-1.18.1-cp313-cp313-win_arm64.whl", hash = "sha256:cd479fc04dd9401e3b4f49e76518768ef99c4f517a98c284eb091fd725719adf", size = 24399315, upload-time = "2026-08-21T23:25:23.458Z" }, ] +[[package]] +name = "semantic-analysis-worker" +version = "1.0.0" +source = { editable = "workers/semantic-analysis-worker" } +dependencies = [ + { name = "analysis-worker" }, + { name = "grouping-worker" }, + { name = "hierarchical-grouping-worker" }, + { name = "intelligence-worker" }, + { name = "worker-base" }, +] + +[package.optional-dependencies] +dev = [ + { name = "pytest" }, +] + +[package.metadata] +requires-dist = [ + { name = "analysis-worker", editable = "workers/analysis-worker" }, + { name = "grouping-worker", editable = "workers/grouping-worker" }, + { name = "hierarchical-grouping-worker", editable = "workers/hierarchical-grouping-worker" }, + { name = "intelligence-worker", editable = "workers/intelligence-worker" }, + { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" }, + { name = "worker-base", editable = "worker-base" }, +] +provides-extras = ["dev"] + [[package]] name = "sentence-transformers" version = "6.2.0" @@ -4288,9 +4385,9 @@ name = "stateful-evals" version = "1.0.0" source = { editable = "stateful-evals" } dependencies = [ - { name = "litellm" }, { name = "mdt-metrics" }, { name = "metrics-computation-engine" }, + { name = "openai" }, { name = "pydantic" }, ] @@ -4308,9 +4405,9 @@ dev = [ [package.metadata] requires-dist = [ { name = "clickhouse-connect", marker = "extra == 'import'", specifier = ">=0.8.0" }, - { name = "litellm", specifier = ">=1.83.0" }, { name = "mdt-metrics", directory = "stateful-evals/stateful_evals_be/plugins/mdt_metrics" }, { name = "metrics-computation-engine", specifier = ">=1.2.7" }, + { name = "openai", specifier = ">=1.0.0" }, { name = "pydantic", specifier = ">=2.0.0" }, ] provides-extras = ["import"] diff --git a/backend/worker-base/src/worker_base/runner.py b/backend/worker-base/src/worker_base/runner.py new file mode 100644 index 0000000..115bde7 --- /dev/null +++ b/backend/worker-base/src/worker_base/runner.py @@ -0,0 +1,169 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""Run several workers in one process, to share the interpreter and the libraries they have in common. + +Each worker keeps its own command line, queues and configuration: a member is the ``main`` click command of the +worker's CLI, started in a thread of its own with its own event loop (``asyncio.run`` is what the CLI does anyway). A +few settings that all the workers read from the same variable can be set for one of them with its prefix: +``NORM_MAX_INFLIGHT_MESSAGES`` rather than ``MAX_INFLIGHT_MESSAGES`` (see ``worker_base.utils.env_scope``), and +``_ARGS`` adds command line options for one member (``ANALYSIS_ARGS="--batch_size 20"``). + +A member that is CPU-bound and forks processes of its own (the analysis worker) is run as a child process instead +(``isolated``): forking from a process that has other threads is not safe. +""" + +from __future__ import annotations + +import argparse +import importlib +import logging +import os +import shlex +import signal +import subprocess +import sys +import threading +from dataclasses import dataclass +from typing import Callable, Sequence + +from worker_base.utils import env_scope + +logger = logging.getLogger(__name__) + +# The settings that every worker reads from one variable, and that can be set per worker with the member's prefix. +SCOPED_SETTINGS = ("MAX_INFLIGHT_MESSAGES", "PERIODIC_TRIGGER_INTERVAL_SECONDS") + + +@dataclass(frozen=True) +class Member: + """A worker of the group: ``entry`` is ``"package.cli:main"``, the click command of its CLI.""" + + name: str + prefix: str + entry: str + isolated: bool = False + + def args(self) -> list[str]: + return shlex.split(os.getenv(f"{self.prefix}_ARGS", "")) + + def load(self) -> Callable: + module, _, attribute = self.entry.partition(":") + return getattr(importlib.import_module(module), attribute) + + +def _run_in_thread(member: Member, command: Callable, errors: dict[str, BaseException]) -> None: + with env_scope(member.prefix): + try: + command.main(args=member.args(), prog_name=member.name, standalone_mode=False) + except BaseException as exc: # noqa: BLE001 - reported to the supervisor, which stops the process + logger.exception("Worker %s stopped with an error", member.name) + errors[member.name] = exc + + +def _isolated_environment(member: Member) -> dict[str, str]: + env = dict(os.environ) + for setting in SCOPED_SETTINGS: + value = os.getenv(f"{member.prefix}_{setting}") + if value not in (None, ""): + env[setting] = value + return env + + +def _start_isolated(member: Member) -> subprocess.Popen: + module, _, attribute = member.entry.partition(":") + code = f"import sys; from {module} import {attribute} as main; main(sys.argv[1:], prog_name={member.name!r})" + return subprocess.Popen( + [sys.executable, "-c", code, *member.args()], + env=_isolated_environment(member), + ) + + +def select(members: Sequence[Member], names: str | Sequence[str] | None) -> list[Member]: + """The members called ``names`` (a comma separated string or a list), in the order of the group; all by default.""" + if isinstance(names, str): + names = [name for name in names.split(",")] + wanted = [name.strip() for name in names or [] if name.strip()] + if not wanted: + return list(members) + known = {member.name for member in members} + unknown = [name for name in wanted if name not in known] + if unknown: + raise ValueError(f"Unknown worker(s) {', '.join(unknown)}: this image runs {', '.join(sorted(known))}") + return [member for member in members if member.name in wanted] + + +def run_members(members: Sequence[Member]) -> int: + """Run the members until one of them fails or all of them have stopped. Returns the exit code.""" + commands = {member.name: member.load() for member in members if not member.isolated} # import errors show up now + + errors: dict[str, BaseException] = {} + threads = [ + threading.Thread( + target=_run_in_thread, args=(member, commands[member.name], errors), name=member.name, daemon=True + ) + for member in members + if not member.isolated + ] + children = {member.name: _start_isolated(member) for member in members if member.isolated} + + def _terminate(signum, frame): # the orchestrator stopped the container + raise SystemExit(128 + signum) + + signal.signal(signal.SIGTERM, _terminate) + + for thread in threads: + thread.start() + logger.info("Running workers: %s", ", ".join(member.name for member in members)) + + code = 0 + try: + stop = threading.Event() + while not stop.wait(0.5): + if errors: + code = 1 + break + exited = {name: child.poll() for name, child in children.items() if child.poll() is not None} + failed = {name: rc for name, rc in exited.items() if rc != 0} + if failed: + logger.error("Worker(s) %s exited with %s", ", ".join(failed), failed) + code = 1 + break + if not any(thread.is_alive() for thread in threads) and len(exited) == len(children): + break + except (KeyboardInterrupt, SystemExit) as exc: + logger.info("Stopping workers") + code = exc.code if isinstance(exc, SystemExit) and isinstance(exc.code, int) else 0 + finally: + for child in children.values(): + if child.poll() is None: + child.terminate() + for child in children.values(): + try: + child.wait(timeout=10) + except subprocess.TimeoutExpired: + child.kill() + return code + + +def main(group: str, members: Sequence[Member], argv: Sequence[str] | None = None) -> int: + """Command line of a group of workers: ``--workers norm,embedding`` picks some of them (``_WORKERS``).""" + env_name = f"{group.upper().replace('-', '_')}_WORKERS" + parser = argparse.ArgumentParser( + prog=f"{group}-worker-cli", + description=f"Run the {group} workers in one process: {', '.join(member.name for member in members)}.", + ) + parser.add_argument( + "--workers", + default=os.getenv(env_name, ""), + help=f"comma separated workers to run (default: all of them; env: {env_name})", + ) + parser.add_argument("--test", action="store_true", help="print 'I'm Alive' and exit") + args = parser.parse_args(argv) + if args.test: + print("I'm Alive") + return 0 + try: + chosen = select(members, args.workers) + except ValueError as exc: + parser.error(str(exc)) + return run_members(chosen) diff --git a/backend/worker-base/src/worker_base/utils.py b/backend/worker-base/src/worker_base/utils.py index 558f4ea..87edd3d 100644 --- a/backend/worker-base/src/worker_base/utils.py +++ b/backend/worker-base/src/worker_base/utils.py @@ -1,6 +1,8 @@ # Copyright (c) 2026 Cisco Systems, Inc. and its affiliates # SPDX-License-Identifier: Apache-2.0 +import contextlib +import contextvars import logging import os from urllib.parse import ( @@ -14,6 +16,33 @@ DEFAULT_RABBITMQ_PORT = "5672" +_env_scope: contextvars.ContextVar[str | None] = contextvars.ContextVar("worker_env_scope", default=None) + + +@contextlib.contextmanager +def env_scope(prefix: str): + """Make ``scoped_getenv`` prefer ``_`` over ```` for the code that runs inside. + + Several workers can run in one process (see ``worker_base.runner``), and a few settings that every worker reads from + the same variable (``MAX_INFLIGHT_MESSAGES``, ``PERIODIC_TRIGGER_INTERVAL_SECONDS``) must still be set per worker. + """ + token = _env_scope.set(prefix) + try: + yield + finally: + _env_scope.reset(token) + + +def scoped_getenv(name: str, default: str | None = None) -> str | None: + """``os.getenv`` that first looks for the variable with the prefix of the current ``env_scope``.""" + prefix = _env_scope.get() + if prefix: + value = os.getenv(f"{prefix}_{name}") + if value not in (None, ""): + return value + return os.getenv(name, default) + + class ColoredFormatter(logging.Formatter): """ANSI-colored formatter for terminal logs.""" @@ -137,7 +166,7 @@ def resolve_max_inflight_messages( `-1` means auto (= available CPU count), any non-positive value is rejected. """ - env_value = os.getenv("MAX_INFLIGHT_MESSAGES") + env_value = scoped_getenv("MAX_INFLIGHT_MESSAGES") raw = env_value if env_value not in (None, "") else str(cli_value) parsed = int(raw) if parsed == -1: diff --git a/backend/worker-base/tests/test_runner.py b/backend/worker-base/tests/test_runner.py new file mode 100644 index 0000000..0d56216 --- /dev/null +++ b/backend/worker-base/tests/test_runner.py @@ -0,0 +1,72 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""Workers that share a process each keep their own settings.""" + +import threading +import time + +import click +import pytest + +from worker_base import runner +from worker_base.utils import env_scope, resolve_max_inflight_messages, scoped_getenv + +seen: dict[str, dict] = {} + + +def _command(name: str, *, fail: bool = False): + @click.command() + @click.option("--flag", default="") + def main(flag): + with_scope = scoped_getenv("MAX_INFLIGHT_MESSAGES") + seen[name] = {"inflight": with_scope, "flag": flag, "thread": threading.current_thread().name} + if fail: + raise RuntimeError("boom") + time.sleep(0.05) + + return main + + +good, bad = _command("good"), _command("bad", fail=True) + + +def test_scoped_setting_wins_over_the_shared_one(monkeypatch): + monkeypatch.setenv("MAX_INFLIGHT_MESSAGES", "8") + monkeypatch.setenv("NORM_MAX_INFLIGHT_MESSAGES", "128") + assert scoped_getenv("MAX_INFLIGHT_MESSAGES") == "8" + with env_scope("NORM"): + assert scoped_getenv("MAX_INFLIGHT_MESSAGES") == "128" + assert resolve_max_inflight_messages(1, error_subject="x") == 128 + with env_scope("EMBEDDING"): # nothing set for it: the shared setting applies + assert resolve_max_inflight_messages(1, error_subject="x") == 8 + + +def test_select_keeps_the_order_of_the_group_and_rejects_unknown_workers(): + members = [runner.Member(n, n.upper(), "m:main") for n in ("a", "b", "c")] + assert [m.name for m in runner.select(members, "c,a")] == ["a", "c"] + assert [m.name for m in runner.select(members, "")] == ["a", "b", "c"] + with pytest.raises(ValueError, match="Unknown worker"): + runner.select(members, "a,z") + + +def test_members_run_in_their_own_thread_with_their_own_settings(monkeypatch): + seen.clear() + monkeypatch.setenv("ONE_MAX_INFLIGHT_MESSAGES", "3") + monkeypatch.setenv("TWO_MAX_INFLIGHT_MESSAGES", "5") + monkeypatch.setenv("TWO_ARGS", "--flag extra") + one = runner.Member("one", "ONE", f"{__name__}:good") + two = runner.Member("two", "TWO", f"{__name__}:good") + assert runner.run_members([one, two]) == 0 + assert seen["good"]["inflight"] in ("3", "5") # both members use the same command object in this test + + +def test_a_failing_member_stops_the_process_with_an_error(monkeypatch): + seen.clear() + ok = runner.Member("ok", "OK", f"{__name__}:good") + ko = runner.Member("ko", "KO", f"{__name__}:bad") + assert runner.run_members([ok, ko]) == 1 + + +def test_member_args_come_from_the_environment(monkeypatch): + monkeypatch.setenv("ANALYSIS_ARGS", "--batch_size 20 --debug") + assert runner.Member("analysis", "ANALYSIS", "m:main").args() == ["--batch_size", "20", "--debug"] diff --git a/backend/workers/README.md b/backend/workers/README.md index 973028b..171bfb4 100644 --- a/backend/workers/README.md +++ b/backend/workers/README.md @@ -39,6 +39,26 @@ Sessions flow through the pipeline in stages: └────────────────────────────┘ ``` +## Grouped images + +The workers run on their own with their CLI, and `docker compose` and the images below run several of them in one +process (`worker_base.runner`), so the interpreter and the libraries they have in common are loaded once: + +| Image | Workers | CLI | +|---|---|---| +| `ingestion-worker` | norm, embedding | `ingestion-worker-cli` | +| `semantic-analysis-worker` | grouping, hierarchical-grouping, analysis, intelligence | `semantic-analysis-worker-cli` | +| `eval-worker` | mce, stateful-eval | `eval-worker-cli` | + +Each worker keeps its queues and options, set the way they are for the worker alone (`NORM_INPUT_QUEUE`, ...), and +runs in a thread with an event loop of its own. The two settings that all the workers read from the same variable are +set per worker with its prefix, `NORM_MAX_INFLIGHT_MESSAGES` rather than `MAX_INFLIGHT_MESSAGES` (and +`HIERARCHICAL_GROUPING_PERIODIC_TRIGGER_INTERVAL_SECONDS`), and `_ARGS` adds command line options to one +worker (`ANALYSIS_ARGS="--batch_size 20"`). `--workers norm` (or `INGESTION_WORKERS`, `SEMANTIC_ANALYSIS_WORKERS`, +`EVAL_WORKERS`) runs only some of them. The analysis worker forks a pool of CPU-bound processes, so it runs as a child +process of the group, not as a thread. If a worker fails, the process stops and the container restarts. Argo run-once +mode needs the image of the single worker. + ## Workers ### [norm-worker](norm-worker/README.md) diff --git a/backend/workers/analysis-worker/deploy/docker/Dockerfile b/backend/workers/analysis-worker/deploy/docker/Dockerfile index 8466b0e..d531987 100644 --- a/backend/workers/analysis-worker/deploy/docker/Dockerfile +++ b/backend/workers/analysis-worker/deploy/docker/Dockerfile @@ -31,9 +31,7 @@ COPY mce /app/mce COPY api /app/api COPY --from=ontology . /ontology -RUN uv sync --package analysis-worker --frozen --no-editable \ - --no-install-package numba \ - --no-install-package llvmlite +RUN uv sync --package analysis-worker --frozen --no-editable # ── Run stage ───────────────────────────────────────────────────────────────── FROM python:3.13-slim AS runner diff --git a/backend/workers/cognition-protocols-worker/deploy/docker/Dockerfile b/backend/workers/cognition-protocols-worker/deploy/docker/Dockerfile index 881060f..8496b13 100644 --- a/backend/workers/cognition-protocols-worker/deploy/docker/Dockerfile +++ b/backend/workers/cognition-protocols-worker/deploy/docker/Dockerfile @@ -2,8 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 # Build with the backend/ directory as the context (see docker-bake.hcl). The two protocol -# MCE-metrics packages (ACCORD, CONCORD) are private git dependencies, so the build needs an -# SSH agent with access to them (docker buildx bake --ssh default). +# MCE-metrics packages (ACCORD, CONCORD) are git dependencies on the public cognition-protocols repository. # ── Build stage ────────────────────────────────────────────────────────────── FROM python:3.13-slim AS builder @@ -17,14 +16,11 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ curl \ ca-certificates \ git \ - openssh-client \ gcc \ g++ \ python3-dev \ && rm -rf /var/lib/apt/lists/* \ - && curl -LsSf https://astral.sh/uv/install.sh | sh \ - && mkdir -p -m 0700 /root/.ssh \ - && ssh-keyscan github.com >> /root/.ssh/known_hosts + && curl -LsSf https://astral.sh/uv/install.sh | sh ENV PATH="/root/.local/bin:${PATH}" @@ -35,8 +31,7 @@ COPY mce /opt/mce COPY api /opt/api COPY --from=ontology . /ontology -RUN --mount=type=ssh \ - uv sync --package cognition-protocols-worker --frozen --no-editable --no-dev +RUN uv sync --package cognition-protocols-worker --frozen --no-editable --no-dev # ── Run stage ───────────────────────────────────────────────────────────────── FROM python:3.13-slim AS runner diff --git a/backend/workers/cognition-protocols-worker/pyproject.toml b/backend/workers/cognition-protocols-worker/pyproject.toml index 9344dce..8e6b463 100644 --- a/backend/workers/cognition-protocols-worker/pyproject.toml +++ b/backend/workers/cognition-protocols-worker/pyproject.toml @@ -12,7 +12,7 @@ license = { text = "Apache-2.0" } dependencies = [ "ioc-accord-mce-metrics", "ioc-concord-mce-metrics", - "ioc-scope-mce", + "ioc-scope-mce-core", "ioc-scope-mce-client", "ioc-scope-api", "worker-base", @@ -29,11 +29,13 @@ cognition-protocols-worker-cli = "cognition_protocols_worker.cli:main" # Graph access goes through the platform's API package (SCOPEKGProvider to read a session, # SCOPECognitionProtocolReportProvider to write the Cognition protocol reports) and metrics run on the -# platform's MCE: both are workspace members. The protocol MCE metrics live with their protocols. +# platform's MCE: both are workspace members. It depends on mce-core and mce-client, not on the ioc-scope-mce meta package, +# because it only runs the protocols' plugin metrics and has no use for the native, deepeval, opik or ragas providers. +# The protocol MCE metrics live with their protocols. [tool.uv.sources] -ioc-accord-mce-metrics = { git = "ssh://git@github.com/cisco-eti/ioc-core-l9-extensions.git", subdirectory = "evals/accord_custom_evals/accord-mce-metrics", branch = "feat/adding-mce-metrics-plugin" } -ioc-concord-mce-metrics = { git = "ssh://git@github.com/ioc-project/cognition-protocols.git", subdirectory = "protocols/concord/mce-metrics", branch = "main" } -ioc-scope-mce = { workspace = true } +ioc-accord-mce-metrics = { git = "https://github.com/ioc-project/cognition-protocols.git", subdirectory = "protocols/accord/mce-metrics", branch = "main" } +ioc-concord-mce-metrics = { git = "https://github.com/ioc-project/cognition-protocols.git", subdirectory = "protocols/concord/mce-metrics", branch = "main" } +ioc-scope-mce-core = { workspace = true } ioc-scope-mce-client = { workspace = true } ioc-scope-api = { workspace = true } worker-base = { workspace = true } diff --git a/backend/workers/embedding-worker/deploy/docker/Dockerfile b/backend/workers/embedding-worker/deploy/docker/Dockerfile index 919d8b8..c17be71 100644 --- a/backend/workers/embedding-worker/deploy/docker/Dockerfile +++ b/backend/workers/embedding-worker/deploy/docker/Dockerfile @@ -1,8 +1,9 @@ # Copyright (c) 2026 Cisco Systems, Inc. and its affiliates # SPDX-License-Identifier: Apache-2.0 -# CPU-optimized image variant. -# Uses CPU-only torch wheels to avoid large CUDA/NVIDIA runtime dependencies. +# API-only image: the embeddings come from an OpenAI-compatible (or Google) endpoint, so torch and the other +# local-model libraries (the dem[embedding-local] extra) are not installed. To run a SentenceTransformer, Qwen3 +# or gensim embedder, build a variant with that extra (and CPU-only torch wheels, to avoid the CUDA runtime). # ── Builder ────────────────────────────────────────────────────────────────── FROM python:3.13-slim AS builder @@ -36,28 +37,6 @@ COPY api /app/api COPY --from=ontology . /ontology RUN uv sync --package embedding-worker --frozen --no-editable --no-dev \ - --no-install-package torch \ - --no-install-package numba \ - --no-install-package llvmlite \ - --no-install-package triton \ - --no-install-package cuda-bindings \ - --no-install-package cuda-toolkit \ - --no-install-package nvidia-cublas \ - --no-install-package nvidia-cuda-cupti \ - --no-install-package nvidia-cuda-nvrtc \ - --no-install-package nvidia-cuda-runtime \ - --no-install-package nvidia-cudnn-cu13 \ - --no-install-package nvidia-cufft \ - --no-install-package nvidia-cufile \ - --no-install-package nvidia-curand \ - --no-install-package nvidia-cusolver \ - --no-install-package nvidia-cusparse \ - --no-install-package nvidia-cusparselt-cu13 \ - --no-install-package nvidia-nccl-cu13 \ - --no-install-package nvidia-nvjitlink \ - --no-install-package nvidia-nvshmem-cu13 \ - --no-install-package nvidia-nvtx \ - && uv pip install --python /opt/venv/bin/python --no-cache-dir --index-url https://download.pytorch.org/whl/cpu --no-deps torch==2.11.0 \ && find /opt/venv -type d \( -name __pycache__ -o -name tests -o -name test \) -prune -exec rm -rf {} + # ── Runtime ─────────────────────────────────────────────────────────────────── diff --git a/backend/workers/embedding-worker/deploy/docker/Dockerfile.gpu b/backend/workers/embedding-worker/deploy/docker/Dockerfile.gpu index af8585a..3df54a3 100644 --- a/backend/workers/embedding-worker/deploy/docker/Dockerfile.gpu +++ b/backend/workers/embedding-worker/deploy/docker/Dockerfile.gpu @@ -31,7 +31,7 @@ COPY dem /app/dem COPY api /app/api COPY --from=ontology . /ontology -RUN uv sync --package embedding-worker --frozen --no-editable --no-dev \ +RUN uv sync --package embedding-worker --extra local --frozen --no-editable --no-dev \ --no-install-package torch \ --no-install-package triton \ --no-install-package cuda-bindings \ diff --git a/backend/workers/embedding-worker/pyproject.toml b/backend/workers/embedding-worker/pyproject.toml index 652eb5a..d737218 100644 --- a/backend/workers/embedding-worker/pyproject.toml +++ b/backend/workers/embedding-worker/pyproject.toml @@ -38,6 +38,11 @@ ioc-scope-api = { workspace = true } ioc-scope-mce-core = { workspace = true } [project.optional-dependencies] +# Embedders that run a model in the process (SentenceTransformerEmbedder, Qwen3Embedder, GensimEmbedder): they +# pull in torch. The default image does not install them (see deploy/docker/Dockerfile.gpu for one that does). +local = [ + "dem[embedding-local]", +] dev = [ "pytest>=7.0.0", "pytest-asyncio>=0.20.0", diff --git a/backend/workers/eval-worker/README.md b/backend/workers/eval-worker/README.md new file mode 100644 index 0000000..aab1ba9 --- /dev/null +++ b/backend/workers/eval-worker/README.md @@ -0,0 +1,3 @@ +# eval-worker + +See the [workers README](../README.md). diff --git a/backend/workers/eval-worker/deploy/docker/Dockerfile b/backend/workers/eval-worker/deploy/docker/Dockerfile new file mode 100644 index 0000000..3e2cbd8 --- /dev/null +++ b/backend/workers/eval-worker/deploy/docker/Dockerfile @@ -0,0 +1,66 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 + +# Build with the backend/ directory as the context (see docker-bake.hcl). +# mce and stateful-eval workers, in one process. + +# ── Build stage ────────────────────────────────────────────────────────────── +FROM python:3.13-slim AS builder + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + UV_LINK_MODE=copy \ + UV_PROJECT_ENVIRONMENT=/opt/venv + +WORKDIR /app + +RUN apt-get update && apt-get install -y --no-install-recommends \ + curl \ + ca-certificates \ + git \ + gcc \ + g++ \ + python3-dev \ + && rm -rf /var/lib/apt/lists/* \ + && curl -LsSf https://astral.sh/uv/install.sh | sh + +ENV PATH="/root/.local/bin:${PATH}" + +COPY pyproject.toml uv.lock /app/ +COPY workers/eval-worker /app/workers/eval-worker +COPY workers/mce-worker /app/workers/mce-worker +COPY workers/stateful-eval-worker /app/workers/stateful-eval-worker +COPY worker-base /app/worker-base +COPY mce /app/mce +COPY stateful-evals /app/stateful-evals +COPY api /app/api +COPY --from=ontology . /ontology + +RUN uv sync --package eval-worker --frozen --no-editable --no-dev + +ARG MODEL_PRICES_REFRESHED_AT=unset +COPY scripts/update_model_prices.py /tmp/update_model_prices.py +RUN /opt/venv/bin/python /tmp/update_model_prices.py \ + --trimmed-file "$(/opt/venv/bin/python -c 'import pathlib, mce.engine as p; print(pathlib.Path(p.__file__).parent / "data" / "model_prices.json")')" \ + || echo "WARNING: could not refresh the LLM price table of the MCE from LiteLLM; keeping the one committed in the repo" +RUN /opt/venv/bin/python /tmp/update_model_prices.py \ + --full-file "$(/opt/venv/bin/python -c 'import pathlib, stateful_evals_be as p; print(pathlib.Path(p.__file__).parent / "data" / "model_prices_and_context_window.json")')" \ + || echo "WARNING: could not refresh the LLM price table of the stateful evals from LiteLLM; keeping the one committed in the repo" + +# ── Run stage ───────────────────────────────────────────────────────────────── +FROM python:3.13-slim AS runner + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + # The stateful evaluation's metrics (mdt-metrics) are plugins of another engine: the MCE would import them, and + # litellm with them (about 300 MB), only to skip them. + MCE_IGNORED_PLUGIN_PACKAGES=mdt-metrics + +WORKDIR /app + +COPY --from=builder /opt/venv /opt/venv +COPY workers/eval-worker/deploy/docker/entrypoint.sh /entrypoint.sh + +RUN chmod +x /entrypoint.sh + +ENTRYPOINT ["/entrypoint.sh"] diff --git a/backend/workers/eval-worker/deploy/docker/entrypoint.sh b/backend/workers/eval-worker/deploy/docker/entrypoint.sh new file mode 100755 index 0000000..16b6df5 --- /dev/null +++ b/backend/workers/eval-worker/deploy/docker/entrypoint.sh @@ -0,0 +1,6 @@ +#!/bin/sh +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +set -eu + +exec /opt/venv/bin/eval-worker-cli "$@" diff --git a/backend/workers/eval-worker/pyproject.toml b/backend/workers/eval-worker/pyproject.toml new file mode 100644 index 0000000..bfeaa3b --- /dev/null +++ b/backend/workers/eval-worker/pyproject.toml @@ -0,0 +1,58 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "eval-worker" +version = "1.0.0" +authors = [ + { name = "Outshift Open", email = "oss-github@cisco.com" } +] +license = "Apache-2.0" +description = "Evaluation workers (mce, stateful-eval) in one process" +readme = "README.md" +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", +] + +requires-python = ">=3.11" +dependencies = [ + "mce-worker", + "stateful-eval-worker", + "worker-base", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.0.0", +] + +[project.scripts] +eval-worker-cli = "eval_worker.cli:main" + +[tool.uv.sources] +mce-worker = { workspace = true } +stateful-eval-worker = { workspace = true } +worker-base = { workspace = true } + +[tool.hatch.build.targets.wheel] +packages = ["src/eval_worker"] + +[tool.ruff] +line-length = 120 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "W", "I"] +ignore = ["E501"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = "test_*.py" diff --git a/backend/workers/eval-worker/src/eval_worker/__init__.py b/backend/workers/eval-worker/src/eval_worker/__init__.py new file mode 100644 index 0000000..d494ce9 --- /dev/null +++ b/backend/workers/eval-worker/src/eval_worker/__init__.py @@ -0,0 +1,2 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 diff --git a/backend/workers/eval-worker/src/eval_worker/cli.py b/backend/workers/eval-worker/src/eval_worker/cli.py new file mode 100644 index 0000000..7d80fd3 --- /dev/null +++ b/backend/workers/eval-worker/src/eval_worker/cli.py @@ -0,0 +1,22 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""Evaluation workers in one process: `--workers` (or `EVAL_WORKERS`) picks some of them.""" + +import sys + +from worker_base.runner import Member, main as run_group + +GROUP = "eval" + +MEMBERS = ( + Member("mce", "MCE", "mce_worker.cli:main"), + Member("stateful-eval", "STATEFUL_EVAL", "stateful_eval_worker.cli:main"), +) + + +def main() -> None: + sys.exit(run_group(GROUP, MEMBERS)) + + +if __name__ == "__main__": + main() diff --git a/backend/workers/eval-worker/tests/test_cli.py b/backend/workers/eval-worker/tests/test_cli.py new file mode 100644 index 0000000..260d00b --- /dev/null +++ b/backend/workers/eval-worker/tests/test_cli.py @@ -0,0 +1,12 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""The group starts the workers of its members, and every member's command can be loaded.""" + +import pytest + +from eval_worker.cli import MEMBERS + + +@pytest.mark.parametrize("member", MEMBERS, ids=lambda member: member.name) +def test_member_command_loads(member): + assert member.load().name # a click command diff --git a/backend/workers/hierarchical-grouping-worker/src/hierarchical_grouping_worker/cli.py b/backend/workers/hierarchical-grouping-worker/src/hierarchical_grouping_worker/cli.py index fcb2a1c..7d903f8 100644 --- a/backend/workers/hierarchical-grouping-worker/src/hierarchical_grouping_worker/cli.py +++ b/backend/workers/hierarchical-grouping-worker/src/hierarchical_grouping_worker/cli.py @@ -14,7 +14,7 @@ from typing import Optional # noqa: E402 import click # noqa: E402 -from worker_base.utils import configure_worker_logging # noqa: E402 +from worker_base.utils import configure_worker_logging, scoped_getenv # noqa: E402 from hierarchical_grouping_worker import queues # noqa: E402 from hierarchical_grouping_worker.worker import HierarchicalGroupingWorker # noqa: E402 @@ -145,7 +145,7 @@ def main( "HIERARCHICAL_GROUPING_EMBEDDING_MODEL", "azure/text-embedding-3-small" ) resolved_max_inflight_messages = int( - os.getenv( + scoped_getenv( "MAX_INFLIGHT_MESSAGES", str(max_inflight_messages), ) @@ -154,7 +154,7 @@ def main( resolved_min_samples = int(os.getenv("HIERARCHICAL_GROUPING_MIN_SAMPLES", str(min_samples))) resolved_max_distance = float(os.getenv("HIERARCHICAL_GROUPING_MAX_DISTANCE", str(max_distance))) resolved_periodic_trigger_interval_seconds = float( - os.getenv( + scoped_getenv( "PERIODIC_TRIGGER_INTERVAL_SECONDS", str(periodic_trigger_interval_seconds), ) diff --git a/backend/workers/ingestion-worker/README.md b/backend/workers/ingestion-worker/README.md new file mode 100644 index 0000000..1083cb5 --- /dev/null +++ b/backend/workers/ingestion-worker/README.md @@ -0,0 +1,3 @@ +# ingestion-worker + +See the [workers README](../README.md). diff --git a/backend/workers/ingestion-worker/deploy/docker/Dockerfile b/backend/workers/ingestion-worker/deploy/docker/Dockerfile new file mode 100644 index 0000000..83e8984 --- /dev/null +++ b/backend/workers/ingestion-worker/deploy/docker/Dockerfile @@ -0,0 +1,55 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 + +# Build with the backend/ directory as the context (see docker-bake.hcl). +# norm and embedding workers, in one process. + +# ── Build stage ────────────────────────────────────────────────────────────── +FROM python:3.13-slim AS builder + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + UV_LINK_MODE=copy \ + UV_PROJECT_ENVIRONMENT=/opt/venv + +WORKDIR /app + +RUN apt-get update && apt-get install -y --no-install-recommends \ + curl \ + ca-certificates \ + git \ + gcc \ + g++ \ + python3-dev \ + && rm -rf /var/lib/apt/lists/* \ + && curl -LsSf https://astral.sh/uv/install.sh | sh + +ENV PATH="/root/.local/bin:${PATH}" + +COPY pyproject.toml uv.lock /app/ +COPY workers/ingestion-worker /app/workers/ingestion-worker +COPY workers/norm-worker /app/workers/norm-worker +COPY workers/embedding-worker /app/workers/embedding-worker +COPY worker-base /app/worker-base +COPY norm /app/norm +COPY dem /app/dem +COPY mce /app/mce +COPY api /app/api +COPY --from=ontology . /ontology + +RUN uv sync --package ingestion-worker --frozen --no-editable --no-dev + +# ── Run stage ───────────────────────────────────────────────────────────────── +FROM python:3.13-slim AS runner + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /app + +COPY --from=builder /opt/venv /opt/venv +COPY workers/ingestion-worker/deploy/docker/entrypoint.sh /entrypoint.sh + +RUN chmod +x /entrypoint.sh + +ENTRYPOINT ["/entrypoint.sh"] diff --git a/backend/workers/ingestion-worker/deploy/docker/entrypoint.sh b/backend/workers/ingestion-worker/deploy/docker/entrypoint.sh new file mode 100755 index 0000000..5823205 --- /dev/null +++ b/backend/workers/ingestion-worker/deploy/docker/entrypoint.sh @@ -0,0 +1,6 @@ +#!/bin/sh +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +set -eu + +exec /opt/venv/bin/ingestion-worker-cli "$@" diff --git a/backend/workers/ingestion-worker/pyproject.toml b/backend/workers/ingestion-worker/pyproject.toml new file mode 100644 index 0000000..76866c2 --- /dev/null +++ b/backend/workers/ingestion-worker/pyproject.toml @@ -0,0 +1,58 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "ingestion-worker" +version = "1.0.0" +authors = [ + { name = "Outshift Open", email = "oss-github@cisco.com" } +] +license = "Apache-2.0" +description = "Ingestion workers (norm, embedding) in one process" +readme = "README.md" +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", +] + +requires-python = ">=3.11" +dependencies = [ + "norm-worker", + "embedding-worker", + "worker-base", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.0.0", +] + +[project.scripts] +ingestion-worker-cli = "ingestion_worker.cli:main" + +[tool.uv.sources] +norm-worker = { workspace = true } +embedding-worker = { workspace = true } +worker-base = { workspace = true } + +[tool.hatch.build.targets.wheel] +packages = ["src/ingestion_worker"] + +[tool.ruff] +line-length = 120 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "W", "I"] +ignore = ["E501"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = "test_*.py" diff --git a/backend/workers/ingestion-worker/src/ingestion_worker/__init__.py b/backend/workers/ingestion-worker/src/ingestion_worker/__init__.py new file mode 100644 index 0000000..d494ce9 --- /dev/null +++ b/backend/workers/ingestion-worker/src/ingestion_worker/__init__.py @@ -0,0 +1,2 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 diff --git a/backend/workers/ingestion-worker/src/ingestion_worker/cli.py b/backend/workers/ingestion-worker/src/ingestion_worker/cli.py new file mode 100644 index 0000000..0378d8e --- /dev/null +++ b/backend/workers/ingestion-worker/src/ingestion_worker/cli.py @@ -0,0 +1,22 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""Ingestion workers in one process: `--workers` (or `INGESTION_WORKERS`) picks some of them.""" + +import sys + +from worker_base.runner import Member, main as run_group + +GROUP = "ingestion" + +MEMBERS = ( + Member("norm", "NORM", "norm_worker.cli:main"), + Member("embedding", "EMBEDDING", "embedding_worker.cli:main"), +) + + +def main() -> None: + sys.exit(run_group(GROUP, MEMBERS)) + + +if __name__ == "__main__": + main() diff --git a/backend/workers/ingestion-worker/tests/test_cli.py b/backend/workers/ingestion-worker/tests/test_cli.py new file mode 100644 index 0000000..18be525 --- /dev/null +++ b/backend/workers/ingestion-worker/tests/test_cli.py @@ -0,0 +1,12 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""The group starts the workers of its members, and every member's command can be loaded.""" + +import pytest + +from ingestion_worker.cli import MEMBERS + + +@pytest.mark.parametrize("member", MEMBERS, ids=lambda member: member.name) +def test_member_command_loads(member): + assert member.load().name # a click command diff --git a/backend/workers/intelligence-worker/src/intelligence_worker/cli.py b/backend/workers/intelligence-worker/src/intelligence_worker/cli.py index e38e13d..17d6160 100644 --- a/backend/workers/intelligence-worker/src/intelligence_worker/cli.py +++ b/backend/workers/intelligence-worker/src/intelligence_worker/cli.py @@ -18,6 +18,7 @@ configure_worker_logging, resolve_max_inflight_messages, resolve_rabbitmq_url, + scoped_getenv, ) from intelligence_worker import queues # noqa: E402 @@ -193,7 +194,7 @@ def main( error_subject="Intelligence max inflight messages", ) resolved_periodic_trigger_interval_seconds = float( - os.getenv( + scoped_getenv( "PERIODIC_TRIGGER_INTERVAL_SECONDS", str(periodic_trigger_interval_seconds), ) diff --git a/backend/workers/intelligence-worker/tests/test_cognitive_failure_templates.py b/backend/workers/intelligence-worker/tests/test_cognitive_failure_templates.py new file mode 100644 index 0000000..6f98c16 --- /dev/null +++ b/backend/workers/intelligence-worker/tests/test_cognitive_failure_templates.py @@ -0,0 +1,62 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +from pathlib import Path + +import pytest +from dem.intelligence import load_templates_from_disk + +from intelligence_worker.wrapper.intelligence_wrapper import _build_insight + +CATALOG = Path(__file__).resolve().parents[3] / "intelligence-catalog" + + +@pytest.fixture(scope="module") +def templates(): + return {t.scope + ":" + t.nameTemplate: t for t in load_templates_from_disk(CATALOG)} + + +def _find(templates, scope, text): + return next(t for key, t in templates.items() if key.startswith(scope) and text in t.nameTemplate) + + +def test_the_cognitive_failure_templates_are_valid_and_loaded(templates): + assert _find(templates, "Session", "detected in session") + assert _find(templates, "SemanticGroup", "Recurring") + + +def test_a_session_insight_names_the_failure_and_takes_its_priority_from_the_confidence(templates): + template = _find(templates, "Session", "detected in session") + row = { + "sessionId": "s1", + "code": "IR-1", + "failureName": "Context Drift", + "confidence": "high", + "reasoning": "Instruction lost", + "occurrences": 2, + "priority": "P1", + } + + insight = _build_insight(template, row) + + assert insight.name == "Context Drift (IR-1) detected in session s1" + assert insight.priority == "P1" + assert insight.targetNodeId == "s1" + assert "Instruction lost" in insight.description + assert "IR-1" in insight.labels + + +def test_two_failures_of_one_session_are_two_insights(templates): + template = _find(templates, "Session", "detected in session") + base = { + "sessionId": "s1", + "failureName": "x", + "confidence": "low", + "reasoning": "r", + "occurrences": 1, + "priority": "P3", + } + + first = _build_insight(template, {**base, "code": "IR-1"}) + second = _build_insight(template, {**base, "code": "IC-2"}) + + assert first.id != second.id diff --git a/backend/workers/mce-worker/deploy/docker/Dockerfile b/backend/workers/mce-worker/deploy/docker/Dockerfile index fa5760c..e7d3fe8 100644 --- a/backend/workers/mce-worker/deploy/docker/Dockerfile +++ b/backend/workers/mce-worker/deploy/docker/Dockerfile @@ -30,9 +30,7 @@ COPY mce /app/mce COPY api /app/api COPY --from=ontology . /ontology -RUN uv sync --package mce-worker --frozen --no-editable \ - --no-install-package numba \ - --no-install-package llvmlite +RUN uv sync --package mce-worker --frozen --no-editable # ── LLM price table ────────────────────────────────────────────────────────── # The price table the MCE bills with is refreshed from LiteLLM on every diff --git a/backend/workers/semantic-analysis-worker/README.md b/backend/workers/semantic-analysis-worker/README.md new file mode 100644 index 0000000..eeeb642 --- /dev/null +++ b/backend/workers/semantic-analysis-worker/README.md @@ -0,0 +1,3 @@ +# semantic-analysis-worker + +See the [workers README](../README.md). diff --git a/backend/workers/semantic-analysis-worker/deploy/docker/Dockerfile b/backend/workers/semantic-analysis-worker/deploy/docker/Dockerfile new file mode 100644 index 0000000..33e2692 --- /dev/null +++ b/backend/workers/semantic-analysis-worker/deploy/docker/Dockerfile @@ -0,0 +1,59 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 + +# Build with the backend/ directory as the context (see docker-bake.hcl). +# grouping, hierarchical grouping, analysis and intelligence workers: the first three in one process, and analysis as a +# child process (it forks a pool of CPU-bound processes of its own). + +# ── Build stage ────────────────────────────────────────────────────────────── +FROM python:3.13-slim AS builder + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + UV_LINK_MODE=copy \ + UV_PROJECT_ENVIRONMENT=/opt/venv + +WORKDIR /app + +RUN apt-get update && apt-get install -y --no-install-recommends \ + curl \ + ca-certificates \ + git \ + gcc \ + g++ \ + python3-dev \ + && rm -rf /var/lib/apt/lists/* \ + && curl -LsSf https://astral.sh/uv/install.sh | sh + +ENV PATH="/root/.local/bin:${PATH}" + +COPY pyproject.toml uv.lock /app/ +COPY workers/semantic-analysis-worker /app/workers/semantic-analysis-worker +COPY workers/grouping-worker /app/workers/grouping-worker +COPY workers/hierarchical-grouping-worker /app/workers/hierarchical-grouping-worker +COPY workers/analysis-worker /app/workers/analysis-worker +COPY workers/intelligence-worker /app/workers/intelligence-worker +COPY worker-base /app/worker-base +COPY dem /app/dem +COPY mce /app/mce +COPY api /app/api +COPY intelligence-catalog /app/intelligence-catalog +COPY --from=ontology . /ontology + +RUN uv sync --package semantic-analysis-worker --frozen --no-editable --no-dev + +# ── Run stage ───────────────────────────────────────────────────────────────── +FROM python:3.13-slim AS runner + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /app + +COPY --from=builder /opt/venv /opt/venv +COPY --from=builder /app/intelligence-catalog /app/intelligence-catalog +COPY workers/semantic-analysis-worker/deploy/docker/entrypoint.sh /entrypoint.sh + +RUN chmod +x /entrypoint.sh + +ENTRYPOINT ["/entrypoint.sh"] diff --git a/backend/workers/semantic-analysis-worker/deploy/docker/entrypoint.sh b/backend/workers/semantic-analysis-worker/deploy/docker/entrypoint.sh new file mode 100755 index 0000000..2136190 --- /dev/null +++ b/backend/workers/semantic-analysis-worker/deploy/docker/entrypoint.sh @@ -0,0 +1,6 @@ +#!/bin/sh +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +set -eu + +exec /opt/venv/bin/semantic-analysis-worker-cli "$@" diff --git a/backend/workers/semantic-analysis-worker/pyproject.toml b/backend/workers/semantic-analysis-worker/pyproject.toml new file mode 100644 index 0000000..23ddebd --- /dev/null +++ b/backend/workers/semantic-analysis-worker/pyproject.toml @@ -0,0 +1,62 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "semantic-analysis-worker" +version = "1.0.0" +authors = [ + { name = "Outshift Open", email = "oss-github@cisco.com" } +] +license = "Apache-2.0" +description = "Semantic analysis workers (grouping, hierarchical grouping, analysis, intelligence) in one process" +readme = "README.md" +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", +] + +requires-python = ">=3.11" +dependencies = [ + "grouping-worker", + "hierarchical-grouping-worker", + "analysis-worker", + "intelligence-worker", + "worker-base", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.0.0", +] + +[project.scripts] +semantic-analysis-worker-cli = "semantic_analysis_worker.cli:main" + +[tool.uv.sources] +grouping-worker = { workspace = true } +hierarchical-grouping-worker = { workspace = true } +analysis-worker = { workspace = true } +intelligence-worker = { workspace = true } +worker-base = { workspace = true } + +[tool.hatch.build.targets.wheel] +packages = ["src/semantic_analysis_worker"] + +[tool.ruff] +line-length = 120 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "W", "I"] +ignore = ["E501"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = "test_*.py" diff --git a/backend/workers/semantic-analysis-worker/src/semantic_analysis_worker/__init__.py b/backend/workers/semantic-analysis-worker/src/semantic_analysis_worker/__init__.py new file mode 100644 index 0000000..d494ce9 --- /dev/null +++ b/backend/workers/semantic-analysis-worker/src/semantic_analysis_worker/__init__.py @@ -0,0 +1,2 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 diff --git a/backend/workers/semantic-analysis-worker/src/semantic_analysis_worker/cli.py b/backend/workers/semantic-analysis-worker/src/semantic_analysis_worker/cli.py new file mode 100644 index 0000000..1fd2cb6 --- /dev/null +++ b/backend/workers/semantic-analysis-worker/src/semantic_analysis_worker/cli.py @@ -0,0 +1,24 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""Semantic analysis workers in one process: `--workers` (or `SEMANTIC_ANALYSIS_WORKERS`) picks some of them.""" + +import sys + +from worker_base.runner import Member, main as run_group + +GROUP = "semantic-analysis" + +MEMBERS = ( + Member("grouping", "GROUPING", "grouping_worker.cli:main"), + Member("hierarchical-grouping", "HIERARCHICAL_GROUPING", "hierarchical_grouping_worker.cli:main"), + Member("analysis", "ANALYSIS", "analysis_worker.cli:main", isolated=True), + Member("intelligence", "INTELLIGENCE", "intelligence_worker.cli:main"), +) + + +def main() -> None: + sys.exit(run_group(GROUP, MEMBERS)) + + +if __name__ == "__main__": + main() diff --git a/backend/workers/semantic-analysis-worker/tests/test_cli.py b/backend/workers/semantic-analysis-worker/tests/test_cli.py new file mode 100644 index 0000000..9649875 --- /dev/null +++ b/backend/workers/semantic-analysis-worker/tests/test_cli.py @@ -0,0 +1,12 @@ +# Copyright (c) 2026 Cisco Systems, Inc. and its affiliates +# SPDX-License-Identifier: Apache-2.0 +"""The group starts the workers of its members, and every member's command can be loaded.""" + +import pytest + +from semantic_analysis_worker.cli import MEMBERS + + +@pytest.mark.parametrize("member", MEMBERS, ids=lambda member: member.name) +def test_member_command_loads(member): + assert member.load().name # a click command diff --git a/deploy/docker-compose/.env.template b/deploy/docker-compose/.env.template index dddc2c9..3f83a25 100644 --- a/deploy/docker-compose/.env.template +++ b/deploy/docker-compose/.env.template @@ -24,10 +24,16 @@ NEO4J_PASSWORD=change-me # --- Workers ------------------------------------------------------------------- # The embedding worker feeds these queues. The cognitive-failure worker makes a judge-model call for each session, -# and the cognition-protocols worker is built with an SSH agent that can read the private protocol packages. +# and the cognition-protocols worker evaluates the Cognition protocols (ACCORD, CONCORD) that a session used. # To turn one off, remove its queue here and stop its service (docker compose stop ). # EMBEDDING_OUTPUT_QUEUE=new_session_to_grouping,new_session_to_stateful_eval,new_session_to_cognitive_failure,new_session_to_cognition_protocols +# The workers run in three services: ingestion (norm, embedding), semantic-analysis (grouping, hierarchical-grouping, +# analysis, intelligence) and eval (mce, stateful-eval). To run only some of the workers of a service, list them: +# INGESTION_WORKERS=norm,embedding +# SEMANTIC_ANALYSIS_WORKERS=grouping,hierarchical-grouping,analysis,intelligence +# EVAL_WORKERS=mce + # --- Published ports --------------------------------------------------------- SCOPE_PORT=8080 # gateway: UI + API OTLP_GRPC_PORT=4317 # otel collector, point your instrumented apps here diff --git a/deploy/docker-compose/README.md b/deploy/docker-compose/README.md index 593d3b1..4b630d2 100644 --- a/deploy/docker-compose/README.md +++ b/deploy/docker-compose/README.md @@ -11,7 +11,7 @@ Runs the whole SCOPE stack on one machine, with the same components as the | `clickhouse`, `clickhouse-setup` | `clickhouse.yaml`, `clickhouse-init-job.yaml` | | `session-end-trigger` | `session-end-trigger.yaml` | | `rabbitmq`, `neo4j` | `rabbitmq.yaml`, `neo4j` subchart | -| `norm`, `embedding`, `mce`, `grouping`, `hierarchical-grouping`, `intelligence`, `analysis`, `stateful-eval`, `cognitive-failure`, `cognition-protocols` (see below) | `worker-*.yaml` | +| `ingestion` (norm, embedding), `semantic-analysis` (grouping, hierarchical-grouping, analysis, intelligence), `eval` (mce, stateful-eval), `cognitive-failure`, `cognition-protocols` (see below) | `worker-*.yaml` | Kubernetes-only concerns (ingress, HPA, cert-manager, ExternalSecrets, resource limits) have no equivalent here. @@ -21,7 +21,8 @@ equivalent here. - Docker with Compose v2.17+ and BuildKit. - A checkout of the whole repository: the images are built from `../../backend`, `../../ontology` and `../../ui`. -- Memory: give Docker at least 12 GB. The chart's worker limits go up to 12 GB each. +- Memory: give Docker at least 6 GB. The whole stack uses about 4.5 GB once a few dozen sessions are processed, and + Neo4j and ClickHouse grow with the data. - An OpenAI-compatible LLM endpoint and a text-embedding model. ## Run @@ -53,7 +54,7 @@ starting the other or change the ports in `.env`. Point your instrumented application at the collector, for example `OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318`. When a session ends, the `session-end-trigger` -publishes it to the `norm` worker and the pipeline takes over. +publishes it to the `norm` worker (in the `ingestion` service) and the pipeline takes over. ## Configuration @@ -66,21 +67,21 @@ publishes it to the `norm` worker and the pipeline takes over. `docker compose stop cognitive-failure` and set `EMBEDDING_OUTPUT_QUEUE` in `.env` without its queue (see `.env.template`). - Queue names and worker tuning (inflight messages, grouping distances, trigger intervals) are set - in `docker-compose.yml` with the chart's default values. + in `docker-compose.yml` with the chart's default values. The `ingestion`, `semantic-analysis` and `eval` + services each run several workers in one process, so a worker's settings carry its prefix + (`NORM_MAX_INFLIGHT_MESSAGES`, `HIERARCHICAL_GROUPING_PERIODIC_TRIGGER_INTERVAL_SECONDS`, ...) and + `ANALYSIS_ARGS` adds command line options to one of them. `INGESTION_WORKERS`, `SEMANTIC_ANALYSIS_WORKERS` and + `EVAL_WORKERS` run only some of a service's workers: `EVAL_WORKERS=mce` leaves out the stateful evaluation. - `scripts/` holds the two shell scripts that the chart inlines in its Job and Deployment; keep them in sync with the chart when it changes. ### Cognition protocols worker The `cognition-protocols` worker detects which Cognition protocols (ACCORD, CONCORD) a session used, and -evaluates the ones it used. It is part of the stack by default, but its image has to be built with an SSH agent -that can read the private protocol packages: `docker compose up -d --build` forwards your agent -(`ssh: [default]` in the compose file), so `ssh-add -l` must list a key with access to those repositories, and -Docker may ask you to allow the forwarding (`--allow=ssh`). This setup has not been tested end to end. +evaluates the ones it used. It is part of the stack by default. To run the stack without it, stop the service (`docker compose stop cognition-protocols`), and set -`EMBEDDING_OUTPUT_QUEUE` in `.env` without `new_session_to_cognition_protocols` (see `.env.template`). Without a -key, the build of this one service fails. +`EMBEDDING_OUTPUT_QUEUE` in `.env` without `new_session_to_cognition_protocols` (see `.env.template`). Data lives in the named volumes `clickhouse-data`, `rabbitmq-data`, `neo4j-data` and `neo4j-logs`. `docker compose down -v` deletes it. diff --git a/deploy/docker-compose/docker-compose.yml b/deploy/docker-compose/docker-compose.yml index 6c9aab0..a3554a8 100644 --- a/deploy/docker-compose/docker-compose.yml +++ b/deploy/docker-compose/docker-compose.yml @@ -170,7 +170,7 @@ services: condition: service_healthy rabbitmq: condition: service_healthy - norm: + ingestion: condition: service_started # ------------------------------------------------------------------ api and ui @@ -261,79 +261,51 @@ services: condition: service_healthy # --------------------------------------------------------------------- workers + # + # The workers run in three images, each one in a process of its own: they share the interpreter and the libraries + # they have in common. Their settings keep the name they have in the worker's CLI, with the worker's prefix + # (NORM_INPUT_QUEUE, ...); MAX_INFLIGHT_MESSAGES and PERIODIC_TRIGGER_INTERVAL_SECONDS are set per worker the same + # way (NORM_MAX_INFLIGHT_MESSAGES), and _ARGS adds command line options. `--workers` (or INGESTION_WORKERS, + # SEMANTIC_ANALYSIS_WORKERS, EVAL_WORKERS) runs only some of the workers of an image. - norm: - image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-norm-worker:${SCOPE_VERSION:-1.0.0} + # Reads the spans of a session into the knowledge graph (norm), and embeds its states (embedding). + ingestion: + image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-ingestion-worker:${SCOPE_VERSION:-1.0.0} build: context: ../../backend additional_contexts: ontology: ../../ontology - dockerfile: workers/norm-worker/deploy/docker/Dockerfile + dockerfile: workers/ingestion-worker/deploy/docker/Dockerfile <<: *worker environment: - <<: [*rabbitmq-env, *neo4j-env, *clickhouse-env] + <<: [*rabbitmq-env, *neo4j-env, *clickhouse-env, *llm-env] NORM_INPUT_QUEUE: new_session_to_norm NORM_OUTPUT_QUEUE: new_session_to_mce,new_session_to_emb - MAX_INFLIGHT_MESSAGES: "128" - depends_on: - rabbitmq: - condition: service_healthy - neo4j: - condition: service_healthy - clickhouse: - condition: service_healthy - - embedding: - image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-embedding-worker:${SCOPE_VERSION:-1.0.0} - build: - context: ../../backend - additional_contexts: - ontology: ../../ontology - dockerfile: workers/embedding-worker/deploy/docker/Dockerfile - <<: *worker - environment: - <<: [*rabbitmq-env, *neo4j-env, *llm-env] + NORM_MAX_INFLIGHT_MESSAGES: "128" EMBEDDING_INPUT_QUEUE: new_session_to_emb # The queues of the cognitive-failure and cognition-protocols workers are last: remove one here, and stop the # worker, to turn it off. EMBEDDING_OUTPUT_QUEUE: ${EMBEDDING_OUTPUT_QUEUE:-new_session_to_grouping,new_session_to_stateful_eval,new_session_to_cognitive_failure,new_session_to_cognition_protocols} EMBEDDING_EMBEDDER_TYPE: OpenAIEmbedder - MAX_INFLIGHT_MESSAGES: "32" + EMBEDDING_MAX_INFLIGHT_MESSAGES: "32" + INGESTION_WORKERS: ${INGESTION_WORKERS:-} depends_on: rabbitmq: condition: service_healthy neo4j: condition: service_healthy - - mce: - image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-mce-worker:${SCOPE_VERSION:-1.0.0} - build: - context: ../../backend - additional_contexts: - ontology: ../../ontology - dockerfile: workers/mce-worker/deploy/docker/Dockerfile - <<: *worker - environment: - <<: [*rabbitmq-env, *neo4j-env, *llm-env] - MCE_INPUT_QUEUE: new_session_to_mce - MCE_OUTPUT_QUEUE: "" - MCE_CONFIG_PATH: /scope/mce_config.yaml - MAX_INFLIGHT_MESSAGES: "64" - volumes: - - ./config/mce_config.yaml:/scope/mce_config.yaml:ro - depends_on: - rabbitmq: - condition: service_healthy - neo4j: + clickhouse: condition: service_healthy - grouping: - image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-grouping-worker:${SCOPE_VERSION:-1.0.0} + # Groups the sessions by semantic similarity (grouping, hierarchical-grouping), and analyses the groups: anomalies, + # consistency and normal behaviour (analysis), and insights (intelligence). + semantic-analysis: + image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-semantic-analysis-worker:${SCOPE_VERSION:-1.0.0} build: context: ../../backend additional_contexts: ontology: ../../ontology - dockerfile: workers/grouping-worker/deploy/docker/Dockerfile + dockerfile: workers/semantic-analysis-worker/deploy/docker/Dockerfile <<: *worker environment: <<: [*rabbitmq-env, *neo4j-env, *llm-env] @@ -342,94 +314,54 @@ services: GROUPING_EMBEDDING_MODEL: ${EMBEDDING_MODEL_NAME:?set EMBEDDING_MODEL_NAME in .env} GROUPING_MAX_DISTANCE: "0.3" GROUPING_MAX_NEIGHBORS: "10" - MAX_INFLIGHT_MESSAGES: "16" - depends_on: - rabbitmq: - condition: service_healthy - neo4j: - condition: service_healthy - - hierarchical-grouping: - image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-hierarchical-grouping-worker:${SCOPE_VERSION:-1.0.0} - build: - context: ../../backend - additional_contexts: - ontology: ../../ontology - dockerfile: workers/hierarchical-grouping-worker/deploy/docker/Dockerfile - <<: *worker - environment: - <<: [*rabbitmq-env, *neo4j-env, *llm-env] + GROUPING_MAX_INFLIGHT_MESSAGES: "16" HIERARCHICAL_GROUPING_INPUT_QUEUE: new_session_to_periodic_grouping HIERARCHICAL_GROUPING_OUTPUT_QUEUE: new_session_to_analysis HIERARCHICAL_GROUPING_EMBEDDING_MODEL: ${EMBEDDING_MODEL_NAME:?set EMBEDDING_MODEL_NAME in .env} HIERARCHICAL_GROUPING_MAX_NEIGHBORS: "10" HIERARCHICAL_GROUPING_MIN_SAMPLES: "20" HIERARCHICAL_GROUPING_MAX_DISTANCE: "0.15" - PERIODIC_TRIGGER_INTERVAL_SECONDS: "900" - MAX_INFLIGHT_MESSAGES: "1" - depends_on: - rabbitmq: - condition: service_healthy - neo4j: - condition: service_healthy - - intelligence: - image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-intelligence-worker:${SCOPE_VERSION:-1.0.0} - build: - context: ../../backend - additional_contexts: - ontology: ../../ontology - dockerfile: workers/intelligence-worker/deploy/docker/Dockerfile - <<: *worker - environment: - <<: [*rabbitmq-env, *neo4j-env] + HIERARCHICAL_GROUPING_PERIODIC_TRIGGER_INTERVAL_SECONDS: "900" + HIERARCHICAL_GROUPING_MAX_INFLIGHT_MESSAGES: "1" INTELLIGENCE_INPUT_QUEUE: new_session_to_intelligence INTELLIGENCE_CATALOG_ROOT: /app/intelligence-catalog - PERIODIC_TRIGGER_INTERVAL_SECONDS: "300" - MAX_INFLIGHT_MESSAGES: "1" - depends_on: - rabbitmq: - condition: service_healthy - neo4j: - condition: service_healthy - - analysis: - image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-analysis-worker:${SCOPE_VERSION:-1.0.0} - build: - context: ../../backend - additional_contexts: - ontology: ../../ontology - dockerfile: workers/analysis-worker/deploy/docker/Dockerfile - <<: *worker - environment: - <<: [*rabbitmq-env, *neo4j-env, *llm-env] + INTELLIGENCE_PERIODIC_TRIGGER_INTERVAL_SECONDS: "300" + INTELLIGENCE_MAX_INFLIGHT_MESSAGES: "1" ANALYSIS_INPUT_QUEUE: new_session_to_analysis ANOMALY_CONFIG_FILE: "" CONSISTENCY_CONFIG_FILE: "" NORMAL_BEHAVIOUR_CONFIG_FILE: "" - MAX_INFLIGHT_MESSAGES: "6" - command: ["--batch_size", "20"] + ANALYSIS_MAX_INFLIGHT_MESSAGES: "6" + ANALYSIS_ARGS: "--batch_size 20" + SEMANTIC_ANALYSIS_WORKERS: ${SEMANTIC_ANALYSIS_WORKERS:-} depends_on: rabbitmq: condition: service_healthy neo4j: condition: service_healthy - stateful-eval: - image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-stateful-eval-worker:${SCOPE_VERSION:-1.0.0} + # Computes the metrics of a session (mce) and evaluates its states (stateful-eval). + eval: + image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-eval-worker:${SCOPE_VERSION:-1.0.0} build: context: ../../backend additional_contexts: ontology: ../../ontology - dockerfile: workers/stateful-eval-worker/deploy/docker/Dockerfile + dockerfile: workers/eval-worker/deploy/docker/Dockerfile <<: *worker environment: <<: [*rabbitmq-env, *neo4j-env, *llm-env] + MCE_INPUT_QUEUE: new_session_to_mce + MCE_OUTPUT_QUEUE: "" + MCE_CONFIG_PATH: /scope/mce_config.yaml + MCE_MAX_INFLIGHT_MESSAGES: "64" STATEFUL_EVAL_INPUT_QUEUE: new_session_to_stateful_eval STATEFUL_EVAL_OUTPUT_QUEUE: "" STATEFUL_EVAL_CONFIG_PATH: /scope/stateful_eval_config.yaml - MAX_INFLIGHT_MESSAGES: "32" + STATEFUL_EVAL_MAX_INFLIGHT_MESSAGES: "32" + EVAL_WORKERS: ${EVAL_WORKERS:-} volumes: + - ./config/mce_config.yaml:/scope/mce_config.yaml:ro - ./config/stateful_eval_config.yaml:/scope/stateful_eval_config.yaml:ro depends_on: rabbitmq: @@ -461,9 +393,7 @@ services: condition: service_healthy # Detects which Cognition protocols (ACCORD, CONCORD) a session used, and evaluates the ones it used. - # The protocol metric packages are private git dependencies, so the image is built with an SSH agent that - # can read them (docker compose up --build needs `ssh-add -l` to list such a key). To run the stack without - # it, stop the service and remove its queue from EMBEDDING_OUTPUT_QUEUE. + # To run the stack without it, stop the service and remove its queue from EMBEDDING_OUTPUT_QUEUE. cognition-protocols: image: ${SCOPE_IMAGE_REGISTRY:-ghcr.io/ioc-project/container-images}/scope-cognition-protocols-worker:${SCOPE_VERSION:-1.0.0} build: @@ -471,7 +401,6 @@ services: additional_contexts: ontology: ../../ontology dockerfile: workers/cognition-protocols-worker/deploy/docker/Dockerfile - ssh: [default] <<: *worker environment: <<: [*rabbitmq-env, *neo4j-env, *llm-env] diff --git a/deploy/k8s/charts/scope/README.md b/deploy/k8s/charts/scope/README.md index 67deb84..81c03ab 100644 --- a/deploy/k8s/charts/scope/README.md +++ b/deploy/k8s/charts/scope/README.md @@ -121,10 +121,7 @@ run, and each one makes the judge prompt longer. ### 7. Cognition Protocols Worker The `cognition-protocols` worker detects which Cognition protocols (ACCORD, CONCORD) a session used, and evaluates -the ones it used. It is **enabled by default**. Its image is not published by CI yet, because the packages with the -protocol metrics are private git dependencies, so the image has to be built with an SSH agent that can read them -(see its Dockerfile) and made available to the cluster. Until then, its pod cannot pull the image: turn the worker -off. +the ones it used. It is **enabled by default**. To turn it off: ```bash helm upgrade -i -n scope scope charts/scope \ diff --git a/deploy/k8s/charts/scope/values.yaml b/deploy/k8s/charts/scope/values.yaml index f221f89..c76007f 100644 --- a/deploy/k8s/charts/scope/values.yaml +++ b/deploy/k8s/charts/scope/values.yaml @@ -769,9 +769,6 @@ workers: cpu: "1" cognition-protocols: - # Enabled by default, but its image is not published by CI yet: it needs the ACCORD and CONCORD metric - # packages, which are private git dependencies, so the image is built with an SSH agent (see the worker's - # Dockerfile). Until the image is available to the cluster, set enabled to false to avoid an ImagePullBackOff. # While enabled, its input queue is added to the embedding worker's output queues. enabled: true replicaCount: 1 diff --git a/docker-bake.hcl b/docker-bake.hcl index e21717d..c51f400 100644 --- a/docker-bake.hcl +++ b/docker-bake.hcl @@ -25,6 +25,9 @@ group "backend" { "stateful-eval-worker", "cognitive-failure-worker", "cognition-protocols-worker", + "ingestion-worker", + "semantic-analysis-worker", + "eval-worker", ] } @@ -155,5 +158,28 @@ target "cognition-protocols-worker" { contexts = { ontology = "ontology" } dockerfile = "workers/cognition-protocols-worker/deploy/docker/Dockerfile" platforms = ["linux/amd64"] - ssh = ["default"] +} + +# The workers of a group in one image and one process (norm + embedding, grouping + hierarchical grouping + analysis + +# intelligence, mce + stateful-eval). The images of the single workers above are still built, for Argo run-once mode. +target "ingestion-worker" { + inherits = ["docker-metadata-action"] + context = "backend" + contexts = { ontology = "ontology" } + dockerfile = "workers/ingestion-worker/deploy/docker/Dockerfile" + platforms = ["linux/amd64"] +} +target "semantic-analysis-worker" { + inherits = ["docker-metadata-action"] + context = "backend" + contexts = { ontology = "ontology" } + dockerfile = "workers/semantic-analysis-worker/deploy/docker/Dockerfile" + platforms = ["linux/amd64"] +} +target "eval-worker" { + inherits = ["docker-metadata-action"] + context = "backend" + contexts = { ontology = "ontology" } + dockerfile = "workers/eval-worker/deploy/docker/Dockerfile" + platforms = ["linux/amd64"] } diff --git a/docs/analysis/components/analysis.md b/docs/analysis/components/analysis.md index 5799d51..b2a78a6 100644 --- a/docs/analysis/components/analysis.md +++ b/docs/analysis/components/analysis.md @@ -43,9 +43,10 @@ persists the results back: | `utils` | `src/dem/utils/` | Shared logging, graph (networkx) conversions, and data-store helpers | Optional dependency extras (`dem/pyproject.toml`) gate the heavier submodules: -`dem[grouping]` (hdbscan, scikit-learn), `dem[llm]` (openai), `dem[embedding]` (sentence-transformers, -transformers, torch, gensim, openai, google-genai), `dem[analysis]` (scikit-learn, scipy), or -`dem[all]` for everything. +`dem[grouping]` (hdbscan, scikit-learn), `dem[llm]` (openai), `dem[embedding]` (the API embedders: openai, +google-genai), `dem[embedding-local]` (the embedders that run a model in the process: sentence-transformers, +transformers, torch, gensim), `dem[analysis]` (scikit-learn, scipy), or `dem[all]` for everything. The embedders +import their library when they are built, so `import dem.embedding` is cheap whatever is installed. ## Public API diff --git a/docs/analysis/components/intelligence-catalog.md b/docs/analysis/components/intelligence-catalog.md index 2c36821..6aed63b 100644 --- a/docs/analysis/components/intelligence-catalog.md +++ b/docs/analysis/components/intelligence-catalog.md @@ -16,8 +16,8 @@ Unlike the other backend components, this isn't a Python library — it's a data ``` intelligence-catalog/ ├── insight-templates/ -│ ├── group/ # SemanticGroup-scoped templates (3 today) -│ └── session/ # Session-scoped templates (6 today) +│ ├── group/ # SemanticGroup-scoped templates (4 today) +│ └── session/ # Session-scoped templates (7 today) ├── recommendations/ # empty (.gitkeep) — reserved for future remediation content └── root-causes/ # empty (.gitkeep) — reserved for future root-cause patterns ``` @@ -67,4 +67,17 @@ Session-scoped templates follow the same shape, reading from `Metric`, `AnomalyR 2. Make sure every `RETURN`-clause alias needed by the templates/labels/priority/targetNodeId is present — the loader rejects templates where a `$variable` isn't returned by the query. 3. No build step is required: the catalog is read from disk at worker startup (`catalog_root` passed to `IntelligenceWrapper`), so a new template only needs redeploying/restarting the [intelligence-worker](../workers/intelligence-worker.md). +### Cognitive failure templates + +Two templates turn the detections of the [cognitive-failure-worker](../workers/cognitive-failure-worker.md) +(`CognitiveFailureReport` nodes) into insights: + +- `session/cognitive-failure-session.json`: one insight per failure and session, attached to the session. A failure + detected at several calls of a session is one insight, described by its most confident detection. The priority + follows that confidence: `P1` for high, `P2` for medium, `P3` for low (the template's `priority` is the `$priority` + variable of its query). +- `group/recurring-cognitive-failure-group.json`: one insight per failure and semantic group, attached to the group, + when the failure was detected with medium or high confidence in at least 2 sessions and 30% of the sessions of the + group. Priority `P2`. + See also [analysis](analysis.md) for the `ConsistencyReport`/`AnomalyReport` nodes these templates query against. diff --git a/docs/analysis/components/stateful-evals.md b/docs/analysis/components/stateful-evals.md index a71e701..4b3de40 100644 --- a/docs/analysis/components/stateful-evals.md +++ b/docs/analysis/components/stateful-evals.md @@ -76,7 +76,7 @@ uv pip install --python .venv-evals/bin/python \ ./stateful-evals ``` -A judge model is configured via `LLM_MODEL_NAME`, `LLM_BASE_MODEL_URL` (an OpenAI-compatible endpoint, called through LiteLLM), and `LLM_API_KEY`. +A judge model is configured via `LLM_MODEL_NAME`, `LLM_BASE_MODEL_URL` (an OpenAI-compatible endpoint, called with the OpenAI SDK), and `LLM_API_KEY`. See the [quickstart](https://github.com/ioc-project/scope/blob/main/backend/stateful-evals/quickstart/README.md) for a full walkthrough — inspecting a `TrajectoryContext` with no model calls, then running `evaluate_file` on a bundled NOA trip planner trajectory — and [Trajectory Context](https://github.com/ioc-project/scope/blob/main/backend/stateful-evals/stateful_evals_be/evaluation/README.md) / [Defining Evaluation Metrics](https://github.com/ioc-project/scope/blob/main/backend/stateful-evals/stateful_evals_be/evaluation/trajectory_context_metrics/README.md) for the state model and how to register a new metric. diff --git a/docs/analysis/components/worker-base.md b/docs/analysis/components/worker-base.md index 822f86a..3402d7a 100644 --- a/docs/analysis/components/worker-base.md +++ b/docs/analysis/components/worker-base.md @@ -63,6 +63,11 @@ Key methods/attributes a subclass uses: set) around each `handle_message()` call. The event loop's thread pool, which `asyncio.to_thread` uses for the blocking work of a handler, is sized to `max_inflight_messages` too: by default it would hold only `min(32, cpus + 4)` threads and cap the worker below its limit. +- `worker_base.runner` — runs several workers in one process (`Member`, `run_members`, `main`): each worker's CLI command + is started in a thread with an event loop of its own, and `env_scope` / `scoped_getenv` (in `worker_base.utils`) let + `NORM_MAX_INFLIGHT_MESSAGES` set the one worker whose CLI reads `MAX_INFLIGHT_MESSAGES`. A member marked `isolated` + runs as a child process instead (the analysis worker, which forks a pool of its own). The `ingestion-worker`, + `semantic-analysis-worker` and `eval-worker` packages are built on it. - `async run_argo_message(raw_input, output_path)` — single-message (or batch-of-messages, if `raw_input` is a JSON array) mode with no RabbitMQ involved: validates the input against `input_message_class`, calls `handle_message()`, and writes the serialized output message(s) to diff --git a/docs/analysis/pipelines.md b/docs/analysis/pipelines.md index 12b7c2c..90629e8 100644 --- a/docs/analysis/pipelines.md +++ b/docs/analysis/pipelines.md @@ -29,7 +29,7 @@ jobs. Nothing else in the pipeline needs to know it exists. | [hierarchical-grouping](workers/hierarchical-grouping-worker.md) | Periodic | :material-check: | :material-check: | | | [intelligence](workers/intelligence-worker.md) | Periodic | :material-check: | :material-check: | | | [cognitive-failure](workers/cognitive-failure-worker.md) | Ingestion | :material-check: | :material-check: | Fed by embedding. It makes a judge-model call for each session, and can be turned off in both deployments | -| [cognition-protocols](workers/cognition-protocols-worker.md) | Ingestion | :material-check: | :material-check: | Fed by embedding. Its image needs an SSH agent that can read private packages to build, and is not published by CI yet | +| [cognition-protocols](workers/cognition-protocols-worker.md) | Ingestion | :material-check: | :material-check: | Fed by embedding. | | [anomaly-detection](workers/anomaly-detection-worker.md), [consistency](workers/consistency-worker.md), [normal-behaviour](workers/normal-behaviour-worker.md) | Ingestion | :material-minus: | :material-minus: | Images built, but their analyses run inside the analysis worker | :material-check: deployed · :material-close: not deployed · :material-minus: not deployed on its own diff --git a/docs/analysis/workers/cognition-protocols-worker.md b/docs/analysis/workers/cognition-protocols-worker.md index 9c6aac6..a394594 100644 --- a/docs/analysis/workers/cognition-protocols-worker.md +++ b/docs/analysis/workers/cognition-protocols-worker.md @@ -95,12 +95,8 @@ queues. It is deployed by default: - **Docker Compose:** the `cognition-protocols` service is part of the stack, and the queue is in the default `EMBEDDING_OUTPUT_QUEUE`. -!!! warning "The image is not published by CI yet" - The ACCORD and CONCORD metric packages are private git dependencies, so the image can only be built with an - SSH agent that can read them (see the worker's Dockerfile), with `docker buildx bake cognition-protocols-worker`. - It is not part of the CI image matrix yet. Until the image is available to your cluster, turn the worker off: - `workers.cognition-protocols.enabled=false` in the chart, or stop the service in Compose and remove its queue - from `EMBEDDING_OUTPUT_QUEUE`. +To turn the worker off, set `workers.cognition-protocols.enabled=false` in the chart, or stop the service in +Compose and remove its queue from `EMBEDDING_OUTPUT_QUEUE`. ```bash cognition-protocols-worker-cli # consume new_session_to_cognition_protocols diff --git a/docs/analysis/workers/embedding-worker.md b/docs/analysis/workers/embedding-worker.md index e676545..0f7ca70 100644 --- a/docs/analysis/workers/embedding-worker.md +++ b/docs/analysis/workers/embedding-worker.md @@ -60,7 +60,7 @@ via CLI flags / environment variables: | `EMBEDDING_INPUT_QUEUE` | `--input-queue` | Input queue name (default `new_session_to_embedding`) | | `EMBEDDING_OUTPUT_QUEUE` | `--output-queue` | Output queue(s), comma-separated (default `new_session_to_grouping`) | | `EMBEDDING_FEEDBACK_QUEUE` | `--feedback-queue` | Optional feedback queue | -| `EMBEDDING_EMBEDDER_TYPE` | `--embedder-type` | `SentenceTransformerEmbedder` (default) or `OpenAIEmbedder` | +| `EMBEDDING_EMBEDDER_TYPE` | `--embedder-type` | `SentenceTransformerEmbedder` (default of the CLI) or `OpenAIEmbedder` (set by the Compose file and the chart). `SentenceTransformerEmbedder` runs the model in the process and needs the `local` extra, which the default image does not install | | `EMBEDDING_MODEL` | `--embedding-model` | Model identifier (default `all-MiniLM-L6-v2`; e.g. `azure/text-embedding-3-small` for `OpenAIEmbedder`) | | `NEO4J_URI` / `NEO4J_USERNAME` / `NEO4J_PASSWORD` / `NEO4J_DATABASE` | — | Neo4j connection, used by the KG DAL | | `AI_GATEWAY_API_KEY` / `AI_GATEWAY_BASE_URL` | — | Required when using `OpenAIEmbedder` | diff --git a/docs/tutorials/deploy-scope.md b/docs/tutorials/deploy-scope.md index 54a49ca..97c0d0d 100644 --- a/docs/tutorials/deploy-scope.md +++ b/docs/tutorials/deploy-scope.md @@ -16,7 +16,8 @@ SCOPE ships as one image per component, published under |---|---| | API | `scope-api` | | Dashboard | `scope-ui` | -| Workers | `scope-norm-worker`, `scope-embedding-worker`, `scope-mce-worker`, `scope-grouping-worker`, `scope-hierarchical-grouping-worker`, `scope-analysis-worker`, `scope-intelligence-worker`, `scope-stateful-eval-worker`, `scope-cognitive-failure-worker`, `scope-cognition-protocols-worker` (not published by CI yet) | +| Workers (Docker Compose) | `scope-ingestion-worker` (norm, embedding), `scope-semantic-analysis-worker` (grouping, hierarchical grouping, analysis, intelligence), `scope-eval-worker` (MCE, stateful evaluation), `scope-cognitive-failure-worker`, `scope-cognition-protocols-worker` | +| Workers (Kubernetes) | `scope-norm-worker`, `scope-embedding-worker`, `scope-mce-worker`, `scope-grouping-worker`, `scope-hierarchical-grouping-worker`, `scope-analysis-worker`, `scope-intelligence-worker`, `scope-stateful-eval-worker`, `scope-cognitive-failure-worker`, `scope-cognition-protocols-worker` | The deployment also needs four pieces of infrastructure: an OTel collector, a span datastore (ClickHouse), a message broker (RabbitMQ) and the knowledge graph store @@ -27,7 +28,7 @@ datastore (ClickHouse), a message broker (RabbitMQ) and the knowledge graph stor - An LLM gateway with an API key. The embedding, grouping, metric and stateful-evaluation workers call it. Without a key, spans are still collected and normalised into the knowledge graph, but those analyses do not run. -- Memory: about 6 GiB of free memory for the default Kubernetes install, and 12 GB +- Memory: about 6 GiB of free memory for the default Kubernetes install, and 6 GB given to Docker for Compose. Pick one of the two options. @@ -35,7 +36,7 @@ Pick one of the two options. ## Option A: Docker Compose Best for a laptop or a single VM. Docker with the Compose plugin (v2.17 or later) is -the only prerequisite, and Docker should have at least **12 GB of memory**. +the only prerequisite, and Docker should have at least **6 GB of memory**. ### 1. Clone SCOPE repo @@ -109,7 +110,7 @@ The collector answers `{"partialSuccess":{}}`. After a few seconds, the norm wor should report the session: ```bash -docker compose logs norm | grep "sess-1" +docker compose logs ingestion | grep "sess-1" ``` You should see the worker processing `sess-1`, then sending the session on to the diff --git a/docs/tutorials/first-trace.md b/docs/tutorials/first-trace.md index 114e7b9..d9b8232 100644 --- a/docs/tutorials/first-trace.md +++ b/docs/tutorials/first-trace.md @@ -11,7 +11,7 @@ image build. You need: -- Docker with the Compose plugin (v2.17 or later) and at least **12 GB of memory** given to Docker. +- Docker with the Compose plugin (v2.17 or later) and at least **6 GB of memory** given to Docker. - A checkout of this repository. - [`uv`](https://docs.astral.sh/uv/), to run the example application. - An OpenAI-compatible LLM endpoint and key. Without one, spans are still collected and @@ -91,7 +91,7 @@ The spans go through four steps. Each one is visible in the logs of the stack, f where you find the session id: ```bash - docker compose logs norm | grep session_id + docker compose logs ingestion | grep session_id ``` ```text diff --git a/docs/tutorials/semantic-analysis.md b/docs/tutorials/semantic-analysis.md index 0013381..11752fd 100644 --- a/docs/tutorials/semantic-analysis.md +++ b/docs/tutorials/semantic-analysis.md @@ -68,11 +68,13 @@ You need: !!! note "Keeping the cost down" Besides the judge metrics of the metrics computation engine (MCE), which the analysis needs, every session goes through two more workers that call your model for each session: the stateful evaluation and the cognitive-failure detection. To skip them - for this walkthrough, stop them before loading the sessions: + for this walkthrough, stop them before loading the sessions. The stateful evaluation runs next to the MCE in the + `eval` service: restart it with only the MCE. ```bash cd deploy/docker-compose - docker compose stop stateful-eval cognitive-failure cognition-protocols + docker compose stop cognitive-failure cognition-protocols + EVAL_WORKERS=mce docker compose up -d eval ``` ## 1. Run one trace of the trip planner @@ -130,16 +132,16 @@ count every session the `norm` worker has processed so far: ```bash cd deploy/docker-compose -docker compose logs norm | grep -c "Completed processing" +docker compose logs ingestion | grep -c "Completed processing" ``` Give the metrics a few more minutes: the judged ones are the slowest. ## 3. Trigger the grouping -The grouping is done by the `hierarchical-grouping` worker. It wakes up every 15 minutes -(`PERIODIC_TRIGGER_INTERVAL_SECONDS` in `docker-compose.yml`), and only groups an application once it has 20 sessions -with an embedding (`HIERARCHICAL_GROUPING_MIN_SAMPLES`). Rather than wait, send it the message that its timer would +The grouping is done by the `hierarchical-grouping` worker, which runs in the `semantic-analysis` service. It wakes up +every 15 minutes (`HIERARCHICAL_GROUPING_PERIODIC_TRIGGER_INTERVAL_SECONDS` in `docker-compose.yml`), and only groups an +application once it has 20 sessions with an embedding (`HIERARCHICAL_GROUPING_MIN_SAMPLES`). Rather than wait, send it the message that its timer would send, with the name of the application. Use the `RABBITMQ_USERNAME` and `RABBITMQ_PASSWORD` of your `.env`: ```bash @@ -161,7 +163,7 @@ anomaly analyses on it. That takes a few minutes: ```bash cd deploy/docker-compose -docker compose logs hierarchical-grouping | grep "semantic groups" +docker compose logs semantic-analysis | grep "semantic groups" ``` ```text @@ -317,8 +319,9 @@ For the routes session it reads, for example, *"The score is 0.00 because the ou ```bash cd deploy/docker-compose -docker compose start stateful-eval cognitive-failure cognition-protocols # if you stopped them -docker compose down -v # to delete the data as well +docker compose start cognitive-failure cognition-protocols # if you stopped them +docker compose up -d eval # with the stateful evaluation again +docker compose down -v # to delete the data as well ``` ## Where to go next