feat: CI support for governance_observability

use-cases/governance_observability/.gitignore

@@ -0,0 +1 @@
+governance_extract_*.json

use-cases/governance_observability/agent_tracing.py

@@ -34,18 +34,23 @@
 import opik
 from opik import opik_context
 
-PROJECT_NAME    = os.environ.get("OPIK_PROJECT_NAME", "governance-data-demo")
+WORKSPACE = os.environ.get("OPIK_WORKSPACE")
+PROJECT_NAME = os.environ.get("OPIK_PROJECT_NAME", "governance-data-demo")
+
+# No Opik credentials -> describe what would be traced and exit without calling Opik.
+DRY_RUN = not (os.environ.get("OPIK_API_KEY") and WORKSPACE)
 
 # Tag applied to every trace. The oversight/reporting team filters on this tag
 # to identify which traces belong to the governance programme.
 # Replace with whatever tag your organisation uses.
-GOVERNANCE_TAG  = "governance"
+GOVERNANCE_TAG = "governance"
 
 
 # ---------------------------------------------------------------------------
 # Agent implementation
 # ---------------------------------------------------------------------------
 
+
 @opik.track(name="retrieve_context", type="tool")
 def retrieve_context(query: str) -> list[str]:
     opik_context.update_current_span(metadata={"retriever": "vector-index-v3", "top_k": 5})
@@ -93,26 +98,26 @@ def run_agent(
         metadata={
             # Governance fields — the oversight team filters and slices on all of these.
             # Adapt field names and values to match your organisation's schema.
-            "env":                 "prod",
-            "region":              "us-east",
-            "use_case_id":         "loan-approval",
-            "use_case_version":    "2.1.0",
-            "team":                "risk-analytics",
-            "business_unit":       business_unit,
-            "model_name":          model,
-            "model_version":       "2024-11-20",
-            "risk_tier":           risk_tier,
+            "env": "prod",
+            "region": "us-east",
+            "use_case_id": "loan-approval",
+            "use_case_version": "2.1.0",
+            "team": "risk-analytics",
+            "business_unit": business_unit,
+            "model_name": model,
+            "model_version": "2024-11-20",
+            "risk_tier": risk_tier,
             "data_classification": "confidential",
-            "regulatory_scope":    "internal",
+            "regulatory_scope": "internal",
             # Call-level runtime fields
-            "request_id":          request_id,
-            "channel":             "api",
+            "request_id": request_id,
+            "channel": "api",
         }
     )
 
     context_docs = retrieve_context(query)
-    answer       = call_llm(query, context_docs, model)
-        
+    answer = call_llm(query, context_docs, model)
+
     return {"answer": answer, "sources": context_docs}
 
 
@@ -122,35 +127,42 @@ def run_agent(
 
 SAMPLE_RUNS = [
     {
-        "query":              "Assess the risk for a $20,000 personal loan application.",
-        "request_id":         "req-001",
-        "business_unit":      "retail",
-        "risk_tier":          "high",
+        "query": "Assess the risk for a $20,000 personal loan application.",
+        "request_id": "req-001",
+        "business_unit": "retail",
+        "risk_tier": "high",
         "hallucination_rate": 0.03,
-        "response_quality":   0.91,
-        "cost_usd":           0.0042,
+        "response_quality": 0.91,
+        "cost_usd": 0.0042,
     },
     {
-        "query":              "Evaluate eligibility for a $500,000 business loan.",
-        "request_id":         "req-002",
-        "business_unit":      "commercial",
-        "risk_tier":          "medium",
+        "query": "Evaluate eligibility for a $500,000 business loan.",
+        "request_id": "req-002",
+        "business_unit": "commercial",
+        "risk_tier": "medium",
         "hallucination_rate": 0.07,
-        "response_quality":   0.84,
-        "cost_usd":           0.0061,
+        "response_quality": 0.84,
+        "cost_usd": 0.0061,
     },
     {
-        "query":              "Review a credit limit increase request from $10,000 to $25,000.",
-        "request_id":         "req-003",
-        "business_unit":      "wealth",
-        "risk_tier":          "low",
+        "query": "Review a credit limit increase request from $10,000 to $25,000.",
+        "request_id": "req-003",
+        "business_unit": "wealth",
+        "risk_tier": "low",
         "hallucination_rate": 0.01,
-        "response_quality":   0.97,
-        "cost_usd":           0.0038,
+        "response_quality": 0.97,
+        "cost_usd": 0.0038,
     },
 ]
 
 if __name__ == "__main__":
+    if DRY_RUN:
+        print(
+            "[DRY RUN] Opik creds not set — would trace 3 governance-tagged "
+            f"loan-approval agent runs to project '{PROJECT_NAME}'."
+        )
+        raise SystemExit(0)
+
     print(f"Project  : {PROJECT_NAME}")
     print(f"Tag      : {GOVERNANCE_TAG}\n")
 
@@ -161,4 +173,4 @@ def run_agent(
 
     opik.flush_tracker()
     print("Done. Traces are visible in the Opik UI under the project:")
-    print(f"  https://www.comet.com/opik/{os.environ['OPIK_WORKSPACE']}/{PROJECT_NAME}/traces")
+    print(f"  https://www.comet.com/opik/{WORKSPACE}/{PROJECT_NAME}/traces")

use-cases/governance_observability/data_governance_team.py

@@ -35,16 +35,19 @@
 # Config
 # ---------------------------------------------------------------------------
 
-WORKSPACE       = os.environ["OPIK_WORKSPACE"]
-OPIK_BASE_URL   = os.environ.get("OPIK_URL_OVERRIDE", "https://www.comet.com/opik/api")
-GOVERNANCE_TAG  = "governance"  # must match the tag used in agent_tracing.py
+WORKSPACE = os.environ.get("OPIK_WORKSPACE")
+OPIK_BASE_URL = os.environ.get("OPIK_URL_OVERRIDE", "https://www.comet.com/opik/api")
+GOVERNANCE_TAG = "governance"  # must match the tag used in agent_tracing.py
+
+# No Opik credentials -> describe the extraction and exit without calling the Opik API.
+DRY_RUN = not (os.environ.get("OPIK_API_KEY") and WORKSPACE)
 
 _now = datetime.now(UTC)
 
 # Metric types to extract. Each maps to one get_project_metrics() call.
 # See the full list of available values in the SDK docs linked above.
 METRIC_TYPES = [
-    "FEEDBACK_SCORES",   # average per named feedback score
+    "FEEDBACK_SCORES",  # average per named feedback score
 ]
 
 # Interval for aggregation. Choose one: "HOURLY" | "DAILY" | "WEEKLY" | "TOTAL"
@@ -66,6 +69,7 @@ def build_client() -> OpikApi:
 # Project enumeration
 # ---------------------------------------------------------------------------
 
+
 def list_all_projects(client: OpikApi) -> list[dict]:
     """Page through find_projects() and return [{"id": ..., "name": ...}, ...]."""
     projects = []
@@ -96,6 +100,7 @@ def list_all_projects(client: OpikApi) -> list[dict]:
 #   TraceFilterPublic(field="metadata", key="business_unit", operator="=", value="retail")
 # ---------------------------------------------------------------------------
 
+
 def _governance_filters(metadata_slice: dict[str, str] | None = None) -> list[TraceFilterPublic]:
     """
     Build the filter list for a governance extraction.
@@ -105,9 +110,7 @@ def _governance_filters(metadata_slice: dict[str, str] | None = None) -> list[Tr
         TraceFilterPublic(field="tags", operator="contains", value=GOVERNANCE_TAG),
     ]
     for key, value in (metadata_slice or {}).items():
-        filters.append(
-            TraceFilterPublic(field="metadata", key=key, operator="=", value=value)
-        )
+        filters.append(TraceFilterPublic(field="metadata", key=key, operator="=", value=value))
     return filters
 
 
@@ -138,6 +141,7 @@ def _governance_filters(metadata_slice: dict[str, str] | None = None) -> list[Tr
 # Metrics extraction
 # ---------------------------------------------------------------------------
 
+
 def fetch_metrics_for_project(
     client: OpikApi,
     project_id: str,
@@ -153,7 +157,7 @@ def fetch_metrics_for_project(
         result.name     — score name (e.g. "composite_risk_score")
         result.data     — list of DataPointNumberPublic (time, value) data points
     """
-    trace_filters  = _governance_filters(metadata_slice)
+    trace_filters = _governance_filters(metadata_slice)
     interval_start = _now - timedelta(days=LOOKBACK_DAYS)
     req_opts: RequestOptions = {"timeout_in_seconds": 60}
     metrics: dict = {}
@@ -192,15 +196,16 @@ def fetch_metrics_for_project(
 # Main pipeline
 # ---------------------------------------------------------------------------
 
+
 def run_extraction() -> list[dict]:
-    print(f"\n{'='*60}")
+    print(f"\n{'=' * 60}")
     print(f"Governance Metrics Extraction  {_now.strftime('%Y-%m-%d %H:%M UTC')}")
     print(f"Workspace : {WORKSPACE}")
     print(f"Tag       : {GOVERNANCE_TAG}")
     print(f"Interval  : {INTERVAL}  |  Look-back: {LOOKBACK_DAYS} days")
-    print(f"{'='*60}\n")
+    print(f"{'=' * 60}\n")
 
-    client   = build_client()
+    client = build_client()
     projects = list_all_projects(client)
     payloads = []
 
@@ -220,11 +225,7 @@ def run_extraction() -> list[dict]:
             trace_filters=_governance_filters(),
             request_options={"timeout_in_seconds": 60},
         )
-        probe_has_data = any(
-            point.value
-            for result in (probe.results or [])
-            for point in (result.data or [])
-        )
+        probe_has_data = any(point.value for result in (probe.results or []) for point in (result.data or []))
         if not probe_has_data:
             print("  No governance-tagged traces — skipping.\n")
             continue
@@ -239,17 +240,17 @@ def run_extraction() -> list[dict]:
         for label, metadata_slice in slices_to_run:
             sliced_metrics[label] = {
                 "slice_filter": metadata_slice,
-                "metrics":      fetch_metrics_for_project(client, project["id"], metadata_slice),
+                "metrics": fetch_metrics_for_project(client, project["id"], metadata_slice),
             }
 
         payload = {
-            "schema_version":  "2.0",
-            "extracted_at":    _now.isoformat(),
-            "workspace":       WORKSPACE,
-            "project_id":      project["id"],
-            "project_name":    project["name"],
-            "governance_tag":  GOVERNANCE_TAG,
-            "slices":          sliced_metrics,
+            "schema_version": "2.0",
+            "extracted_at": _now.isoformat(),
+            "workspace": WORKSPACE,
+            "project_id": project["id"],
+            "project_name": project["name"],
+            "governance_tag": GOVERNANCE_TAG,
+            "slices": sliced_metrics,
         }
         payloads.append(payload)
 
@@ -282,4 +283,11 @@ def _push_to_reporting_endpoint(payloads: list[dict]) -> None:
 
 
 if __name__ == "__main__":
+    if DRY_RUN:
+        print(
+            "[DRY RUN] Opik creds not set — would extract governance metrics across "
+            f"all projects for '{GOVERNANCE_TAG}'-tagged traces and build the reporting payload."
+        )
+        raise SystemExit(0)
+
     run_extraction()

use-cases/governance_observability/run.sh

@@ -0,0 +1,15 @@
+#!/usr/bin/env bash
+set -e
+
+export OPIK_PROJECT_NAME="governance-observability"
+
+uv sync
+
+echo "--- Step 1: agent tracing ---"
+uv run python agent_tracing.py
+
+echo "--- Step 2: use case team ---"
+uv run python use_case_team.py
+
+echo "--- Step 3: data governance team ---"
+uv run python data_governance_team.py