diff --git a/use-cases/governance_observability/.gitignore b/use-cases/governance_observability/.gitignore new file mode 100644 index 0000000..dd3d02a --- /dev/null +++ b/use-cases/governance_observability/.gitignore @@ -0,0 +1 @@ +governance_extract_*.json diff --git a/use-cases/governance_observability/agent_tracing.py b/use-cases/governance_observability/agent_tracing.py index 6bfc9b5..d4b3299 100644 --- a/use-cases/governance_observability/agent_tracing.py +++ b/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") diff --git a/use-cases/governance_observability/data_governance_team.py b/use-cases/governance_observability/data_governance_team.py index 63e5faa..70df65a 100644 --- a/use-cases/governance_observability/data_governance_team.py +++ b/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() diff --git a/use-cases/governance_observability/run.sh b/use-cases/governance_observability/run.sh new file mode 100644 index 0000000..3c55c1e --- /dev/null +++ b/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 diff --git a/use-cases/governance_observability/use_case_team.py b/use-cases/governance_observability/use_case_team.py index 9b4b9a5..5205239 100644 --- a/use-cases/governance_observability/use_case_team.py +++ b/use-cases/governance_observability/use_case_team.py @@ -37,10 +37,13 @@ # Config # --------------------------------------------------------------------------- -WORKSPACE = os.environ["OPIK_WORKSPACE"] -PROJECT_NAME = os.environ.get("OPIK_PROJECT_NAME", "governance-data-demo") -OPIK_BASE_URL = os.environ.get("OPIK_URL_OVERRIDE", "https://www.comet.com/opik/api") -GOVERNANCE_TAG = "governance" +WORKSPACE = os.environ.get("OPIK_WORKSPACE") +PROJECT_NAME = os.environ.get("OPIK_PROJECT_NAME", "governance-data-demo") +OPIK_BASE_URL = os.environ.get("OPIK_URL_OVERRIDE", "https://www.comet.com/opik/api") +GOVERNANCE_TAG = "governance" + +# No Opik credentials -> describe the batch and exit without calling the Opik API. +DRY_RUN = not (os.environ.get("OPIK_API_KEY") and WORKSPACE) def build_client() -> OpikApi: @@ -100,7 +103,7 @@ def derive_oversight_risk_flag(composite_risk: float | None) -> float | None: def derive_cost_per_quality_unit(scores: dict[str, float]) -> float | None: """Cost in USD divided by response quality. Returns None when either score is absent.""" - cost = scores.get("cost_usd") + cost = scores.get("cost_usd") quality = scores.get("response_quality") if cost is None or quality is None or quality == 0: return None @@ -111,11 +114,12 @@ def derive_cost_per_quality_unit(scores: dict[str, float]) -> float | None: # Batch job # --------------------------------------------------------------------------- + def run_retrospective_batch(client: OpikApi, project_id: str) -> None: - now = datetime.now(UTC) + now = datetime.now(UTC) yesterday = now - timedelta(days=1) - from_time = yesterday.replace(hour=0, minute=0, second=0, microsecond=0) - to_time = yesterday.replace(hour=23, minute=59, second=59, microsecond=999999) + from_time = yesterday.replace(hour=0, minute=0, second=0, microsecond=0) + to_time = yesterday.replace(hour=23, minute=59, second=59, microsecond=999999) print(f"Date range : {from_time.date()} 00:00 → 23:59 UTC") @@ -123,7 +127,7 @@ def run_retrospective_batch(client: OpikApi, project_id: str) -> None: page, page_size = 1, 100 total_processed = 0 - total_scored = 0 + total_scored = 0 while True: response = client.traces.get_traces_by_project( @@ -142,30 +146,32 @@ def run_retrospective_batch(client: OpikApi, project_id: str) -> None: total_processed += 1 existing: dict[str, float] = { - fs.name: fs.value - for fs in (trace.feedback_scores or []) - if fs.name and fs.value is not None + fs.name: fs.value for fs in (trace.feedback_scores or []) if fs.name and fs.value is not None } - composite_risk = derive_composite_risk_score(existing) - risk_flag = derive_oversight_risk_flag(composite_risk) + composite_risk = derive_composite_risk_score(existing) + risk_flag = derive_oversight_risk_flag(composite_risk) cost_per_quality = derive_cost_per_quality_unit(existing) derived = [ - s for s in [ - composite_risk is not None and { - "name": "composite_risk_score", - "value": composite_risk, + s + for s in [ + composite_risk is not None + and { + "name": "composite_risk_score", + "value": composite_risk, "reason": "0.5×compliance + 0.3×(1−hallucination) + 0.2×quality", }, - risk_flag is not None and { - "name": "oversight_risk_flag", - "value": risk_flag, + risk_flag is not None + and { + "name": "oversight_risk_flag", + "value": risk_flag, "reason": f"1.0 = composite_risk_score < {RISK_THRESHOLD}", }, - cost_per_quality is not None and { - "name": "cost_per_quality_unit", - "value": cost_per_quality, + cost_per_quality is not None + and { + "name": "cost_per_quality_unit", + "value": cost_per_quality, "reason": "cost_usd / response_quality", }, ] @@ -200,10 +206,17 @@ def run_retrospective_batch(client: OpikApi, project_id: str) -> None: # --------------------------------------------------------------------------- if __name__ == "__main__": + if DRY_RUN: + print( + "[DRY RUN] Opik creds not set — would derive composite governance scores " + f"for yesterday's '{GOVERNANCE_TAG}'-tagged traces in project '{PROJECT_NAME}'." + ) + raise SystemExit(0) + print(f"\nProject : {PROJECT_NAME}") print(f"Workspace: {WORKSPACE}\n") - client = build_client() + client = build_client() project_id = get_project_id(client) run_retrospective_batch(client, project_id)