Agent Observability#

Trace: prior-auth-P12345-95810
├── Span: workflow.run (root)
│   ├── Span: node.fetch_patient_data
│   │   └── Span: tool.get_patient_summary
│   ├── Span: node.retrieve_guidelines
│   │   └── Span: tool.search_clinical_guidelines
│   ├── Span: node.evaluate_criteria
│   │   └── Span: llm.call (claude-opus-4-8)
│   ├── Span: hitl.physician_review [INTERRUPT]
│   └── Span: node.finalize_determination
│       └── Span: tool.submit_determination

"""
Structured observability for an agentic workflow.
Educational Example — Illustrative observability patterns.
"""
from __future__ import annotations

import json
import logging
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Optional


class EventType(str, Enum):
    WORKFLOW_START = "workflow.start"
    WORKFLOW_COMPLETE = "workflow.complete"
    WORKFLOW_ERROR = "workflow.error"
    NODE_START = "node.start"
    NODE_COMPLETE = "node.complete"
    NODE_ERROR = "node.error"
    TOOL_CALL = "tool.call"
    TOOL_RESULT = "tool.result"
    TOOL_ERROR = "tool.error"
    LLM_CALL = "llm.call"
    LLM_RESPONSE = "llm.response"
    HITL_TRIGGERED = "hitl.triggered"
    HITL_RESOLVED = "hitl.resolved"


@dataclass
class AgentEvent:
    """Structured event for agent observability."""
    event_type: EventType
    trace_id: str
    span_id: str
    workflow_name: str
    timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    duration_ms: Optional[float] = None
    node_name: Optional[str] = None
    tool_name: Optional[str] = None
    model: Optional[str] = None
    input_tokens: Optional[int] = None
    output_tokens: Optional[int] = None
    success: Optional[bool] = None
    error_code: Optional[str] = None
    error_message: Optional[str] = None
    # Clinical compliance fields
    patient_id: Optional[str] = None       # For compliance logging
    reviewer_id: Optional[str] = None      # For HITL audit
    decision: Optional[str] = None         # For outcome logging
    metadata: dict = field(default_factory=dict)

    def to_log_dict(self) -> dict:
        return {k: v for k, v in self.__dict__.items() if v is not None}


class AgentObservability:
    """
    Structured observability wrapper for agentic workflows.
    Emits structured JSON log events compatible with CloudWatch, Datadog, etc.
    """

    def __init__(self, workflow_name: str, logger: Optional[logging.Logger] = None):
        self.workflow_name = workflow_name
        self.logger = logger or logging.getLogger(workflow_name)
        self._configure_json_logging()

    def _configure_json_logging(self):
        if not self.logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(logging.Formatter("%(message)s"))
            self.logger.addHandler(handler)
            self.logger.setLevel(logging.INFO)

    def _emit(self, event: AgentEvent):
        self.logger.info(json.dumps(event.to_log_dict()))

    def trace_workflow(self, patient_id: str):
        """Context manager for tracing a complete workflow."""
        import contextlib

        @contextlib.contextmanager
        def _trace():
            trace_id = str(uuid.uuid4())
            span_id = str(uuid.uuid4())
            start = time.monotonic()

            self._emit(AgentEvent(
                event_type=EventType.WORKFLOW_START,
                trace_id=trace_id,
                span_id=span_id,
                workflow_name=self.workflow_name,
                patient_id=patient_id,
            ))

            try:
                yield trace_id
                duration = (time.monotonic() - start) * 1000
                self._emit(AgentEvent(
                    event_type=EventType.WORKFLOW_COMPLETE,
                    trace_id=trace_id,
                    span_id=span_id,
                    workflow_name=self.workflow_name,
                    patient_id=patient_id,
                    duration_ms=duration,
                    success=True,
                ))
            except Exception as e:
                duration = (time.monotonic() - start) * 1000
                self._emit(AgentEvent(
                    event_type=EventType.WORKFLOW_ERROR,
                    trace_id=trace_id,
                    span_id=span_id,
                    workflow_name=self.workflow_name,
                    patient_id=patient_id,
                    duration_ms=duration,
                    success=False,
                    error_code=type(e).__name__,
                    error_message=str(e),
                ))
                raise

        return _trace()

    def trace_tool_call(
        self,
        trace_id: str,
        tool_name: str,
        tool_input: dict,
        patient_id: Optional[str] = None,
    ) -> dict:
        """Execute a tool call with tracing. Returns tool result."""
        span_id = str(uuid.uuid4())
        start = time.monotonic()

        self._emit(AgentEvent(
            event_type=EventType.TOOL_CALL,
            trace_id=trace_id,
            span_id=span_id,
            workflow_name=self.workflow_name,
            tool_name=tool_name,
            patient_id=patient_id,
            metadata={"input_keys": list(tool_input.keys())},
        ))

        try:
            # Actual tool dispatch would happen here
            result = {"status": "stub", "tool": tool_name}
            duration = (time.monotonic() - start) * 1000
            self._emit(AgentEvent(
                event_type=EventType.TOOL_RESULT,
                trace_id=trace_id,
                span_id=span_id,
                workflow_name=self.workflow_name,
                tool_name=tool_name,
                patient_id=patient_id,
                duration_ms=duration,
                success=True,
            ))
            return result

        except Exception as e:
            duration = (time.monotonic() - start) * 1000
            self._emit(AgentEvent(
                event_type=EventType.TOOL_ERROR,
                trace_id=trace_id,
                span_id=span_id,
                workflow_name=self.workflow_name,
                tool_name=tool_name,
                patient_id=patient_id,
                duration_ms=duration,
                success=False,
                error_code=type(e).__name__,
                error_message=str(e),
            ))
            raise

    def log_llm_call(
        self,
        trace_id: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
        duration_ms: float,
        success: bool,
    ):
        """Log an LLM call's token consumption and latency."""
        self._emit(AgentEvent(
            event_type=EventType.LLM_RESPONSE,
            trace_id=trace_id,
            span_id=str(uuid.uuid4()),
            workflow_name=self.workflow_name,
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            duration_ms=duration_ms,
            success=success,
        ))

    def log_hitl_triggered(
        self,
        trace_id: str,
        trigger_reason: str,
        patient_id: str,
        metadata: dict,
    ):
        """Log a HITL trigger event for compliance and monitoring."""
        self._emit(AgentEvent(
            event_type=EventType.HITL_TRIGGERED,
            trace_id=trace_id,
            span_id=str(uuid.uuid4()),
            workflow_name=self.workflow_name,
            patient_id=patient_id,
            metadata={"trigger_reason": trigger_reason, **metadata},
        ))

"""
LangSmith integration for LangGraph workflow tracing.
Educational Example — Illustrative integration pattern.
Verify current LangSmith API at smith.langchain.com/docs
"""
import os
from typing import Any


def configure_langsmith(project_name: str, environment: str = "development"):
    """
    Configure LangSmith tracing for a LangGraph application.
    LangSmith traces LangGraph workflows automatically when LANGCHAIN_TRACING_V2=true.
    """
    # Set these in environment configuration, not hardcoded
    os.environ["LANGCHAIN_TRACING_V2"] = "true"
    os.environ["LANGCHAIN_PROJECT"] = f"{project_name}-{environment}"
    # os.environ["LANGCHAIN_API_KEY"] = "..."  # Set via secrets manager in production
    # os.environ["LANGCHAIN_ENDPOINT"] = "..."  # Required for self-hosted LangSmith


def add_run_metadata(config: dict, metadata: dict) -> dict:
    """
    Add run-level metadata to a LangGraph invocation config.
    Metadata appears in LangSmith UI for filtering and searching.
    """
    if "metadata" not in config:
        config["metadata"] = {}

    config["metadata"].update(metadata)
    return config


def run_with_langsmith_metadata(
    graph: Any,
    initial_state: dict,
    thread_id: str,
    patient_id: str,
    procedure_code: str,
) -> dict:
    """
    Run a LangGraph workflow with LangSmith metadata for filtering in the UI.
    """
    config = {
        "configurable": {"thread_id": thread_id},
        "metadata": {
            "patient_id": patient_id,        # For filtering in LangSmith UI
            "procedure_code": procedure_code,
            "workflow_version": "2.0",
            "environment": os.environ.get("APP_ENV", "development"),
        },
        "tags": ["prior-auth", procedure_code, "hms-workflow"],
    }

    return graph.invoke(initial_state, config)

"""
Automated quality evaluation pipeline for agentic outputs.
Educational Example — Illustrative evaluation patterns.
"""
from dataclasses import dataclass
from typing import Optional


@dataclass
class EvaluationResult:
    workflow_id: str
    patient_id: str
    procedure_code: str
    # Quality dimensions
    criteria_accuracy: Optional[float] = None    # 0.0–1.0: were criteria applied correctly?
    guideline_citation_accuracy: float = 0.0     # 0.0–1.0: were citations correct and relevant?
    determination_consistency: float = 0.0       # 0.0–1.0: consistent with similar cases?
    output_format_compliance: float = 0.0        # 0.0–1.0: did output match expected schema?
    # Aggregate
    overall_quality_score: float = 0.0
    passed_quality_gate: bool = False
    evaluation_notes: str = ""


def evaluate_prior_auth_output(
    workflow_output: dict,
    ground_truth: Optional[dict] = None,
) -> EvaluationResult:
    """
    Evaluate a prior authorization workflow output.
    For production: integrate with RAGAS or LangSmith evaluation datasets.

    ground_truth: if provided, compare against known-correct evaluation.
    If not provided, apply heuristic quality checks only.
    """
    result = EvaluationResult(
        workflow_id=workflow_output.get("workflow_id", "unknown"),
        patient_id=workflow_output.get("patient_id", "unknown"),
        procedure_code=workflow_output.get("procedure_code", "unknown"),
    )

    determination = workflow_output.get("final_determination", {})

    # 1. Output format compliance — does the output have required fields?
    required_fields = ["criteria_met", "rationale", "supporting_guidelines", "determination"]
    present_fields = sum(1 for f in required_fields if f in determination)
    result.output_format_compliance = present_fields / len(required_fields)

    # 2. Guideline citation presence — are citations non-empty?
    citations = determination.get("supporting_guidelines", [])
    result.guideline_citation_accuracy = min(len(citations) / 3, 1.0)  # expect ≥ 3 citations

    # 3. If ground truth is available, compare determination
    if ground_truth:
        expected = ground_truth.get("criteria_met")
        actual = determination.get("criteria_met")
        result.criteria_accuracy = 1.0 if expected == actual else 0.0

    # 4. Overall score
    scores = [
        result.output_format_compliance,
        result.guideline_citation_accuracy,
    ]
    if result.criteria_accuracy is not None:
        scores.append(result.criteria_accuracy)

    result.overall_quality_score = sum(scores) / len(scores)

    # Quality gate: all required fields present AND citations present
    result.passed_quality_gate = (
        result.output_format_compliance >= 1.0
        and result.guideline_citation_accuracy >= 0.33  # At least 1 citation
    )

    return result


class EvaluationPipeline:
    """
    Continuous evaluation pipeline for production agent monitoring.
    In production: integrate with LangSmith Evaluation or RAGAS.
    """

    def __init__(self, quality_gate_threshold: float = 0.80):
        self.quality_gate_threshold = quality_gate_threshold
        self.evaluation_history: list[EvaluationResult] = []

    def evaluate_batch(
        self,
        workflow_outputs: list[dict],
        ground_truths: Optional[list[dict]] = None,
    ) -> dict:
        """Evaluate a batch of workflow outputs and return aggregate metrics."""
        results = []
        for i, output in enumerate(workflow_outputs):
            gt = ground_truths[i] if ground_truths else None
            result = evaluate_prior_auth_output(output, gt)
            results.append(result)
            self.evaluation_history.append(result)

        pass_rate = sum(1 for r in results if r.passed_quality_gate) / max(len(results), 1)
        avg_quality = sum(r.overall_quality_score for r in results) / max(len(results), 1)

        return {
            "batch_size": len(results),
            "pass_rate": pass_rate,
            "avg_quality_score": avg_quality,
            "below_gate_count": sum(1 for r in results if not r.passed_quality_gate),
            "gate_threshold": self.quality_gate_threshold,
            "alert_required": avg_quality < self.quality_gate_threshold,
        }