POC Portfolio: Data & AI Solutions for Salesforce Employee Success

Configuration-as-code pattern for declarative, version-controlled data ingestion:

# master_config/workday_hcm.yaml
source:
  name: workday_hcm
  type: api
  connection:
    base_url: "https://wd5-impl.workday.com/ccx/api/v1"
    auth: oauth2
    credentials_ref: "vault://es-secrets/workday-oauth"

  ingestion:
    mode: incremental  # full | incremental | cdc
    schedule: "0 2 * * *"  # Daily at 2 AM
    watermark_column: "last_modified_date"
    batch_size: 10000

  schema:
    version: "3.2"
    drift_detection: true
    evolution_strategy: merge  # merge | strict | log_only

  pii_classification:
    auto_detect: true
    known_pii_fields:
      - employee_ssn: ssn
      - date_of_birth: dob
      - home_address: address
      - salary: compensation
    encryption: fernet
    key_ref: "vault://es-secrets/pii-encryption-key"

  governance:
    owner: "es-people-analytics"
    access_tier: restricted
    tags:
      - data_domain: employee_master
      - sensitivity: high
      - retention_days: 2555  # 7 years
      - gdpr_relevant: true

  quality_rules:
    - rule: not_null
      columns: [employee_id, hire_date, department_code]
    - rule: unique
      columns: [employee_id]
    - rule: referential_integrity
      column: manager_id
      reference: employee_id
    - rule: range_check
      column: salary
      min: 0
      max: 10000000

Probabilistic identity resolution for HR data across multiple source systems with confidence scoring:

# identity_resolution/probabilistic_matcher.py
"""
Probabilistic identity resolution for HR data across multiple source systems.
Handles: name variations, cultural name orders, nickname matching,
and fuzzy matching with confidence scoring.
"""

from dataclasses import dataclass
from typing import List, Tuple

@dataclass
class EmployeeCandidate:
    source_system: str
    source_id: str
    first_name: str
    last_name: str
    email: str
    department: str
    hire_date: str

class IdentityResolver:
    """
    Multi-strategy identity resolution pipeline.

    Strategy 1: Deterministic — exact match on employee_id or email
    Strategy 2: Probabilistic — weighted scoring on name, dept, hire_date
    Strategy 3: NLP-assisted — handles cultural name variations, nicknames
    """

    CONFIDENCE_THRESHOLD = 0.85
    MANUAL_REVIEW_THRESHOLD = 0.65

    WEIGHTS = {
        'email_exact': 0.40,
        'name_similarity': 0.25,
        'department_match': 0.15,
        'hire_date_match': 0.15,
        'source_id_pattern': 0.05
    }

    def resolve(self, candidates: List[EmployeeCandidate]):
        """
        Three-pass resolution:
        Pass 1: Deterministic matching (email, employee ID)
        Pass 2: Probabilistic matching (fuzzy name + dept + hire date)
        Pass 3: Flag low-confidence matches for manual review
        """
        results = []
        unmatched = list(candidates)

        # Pass 1: Deterministic
        deterministic_groups = self._deterministic_match(unmatched)
        for group in deterministic_groups:
            golden_id = self._generate_golden_id(group)
            results.append(MatchResult(
                golden_id=golden_id,
                candidates=group,
                confidence=1.0,
                match_method='deterministic'
            ))

        # Pass 2: Probabilistic
        probabilistic_groups = self._probabilistic_match(unmatched)
        for group, score in probabilistic_groups:
            if score >= self.CONFIDENCE_THRESHOLD:
                # High confidence → approve
                results.append(...)
            elif score >= self.MANUAL_REVIEW_THRESHOLD:
                # Medium confidence → flag for review
                results.append(...)

        return results

NLP-based and regex pattern matching for automatic sensitive field detection:

# governance/pii_classifier.py
PII_PATTERNS = {
    'ssn': r'\b\d{3}-?\d{2}-?\d{4}\b',
    'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
    'phone': r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
    'date_of_birth': r'\b(0[1-9]|1[0-2])/(0[1-9]|[12]\d|3[01])/\d{4}\b',
    'salary': r'\$[\d,]+\.?\d{0,2}',
    'bank_account': r'\b\d{8,17}\b'
}

SENSITIVE_COLUMN_NAMES = {
    'high': ['ssn', 'social_security', 'salary', 'compensation'],
    'medium': ['date_of_birth', 'home_address', 'phone'],
    'low': ['department', 'job_title', 'hire_date']
}

def classify_and_tag(dataframe, source_config):
    """
    Scan DataFrame columns for PII, apply governance tags,
    and encrypt high-sensitivity fields.
    """
    audit_log = []

    for column in dataframe.columns:
        sensitivity = _detect_sensitivity(column, dataframe[column])

        # Tag the column in metadata catalog
        tag_in_catalog(
            column=column,
            sensitivity=sensitivity,
            source=source_config['source']['name']
        )

        # Encrypt high-sensitivity fields
        if sensitivity == 'high':
            dataframe[column] = encrypt_field(dataframe[column])
            audit_log.append({
                'action': 'encrypt',
                'column': column,
                'sensitivity': sensitivity
            })

    return dataframe, audit_log

Example: Voluntary Attrition Rate metric with full business definition, formula, dimensions, and quality checks:

# metrics/attrition/voluntary_attrition_rate.yaml
metric:
  name: voluntary_attrition_rate
  display_name: "Voluntary Attrition Rate"
  version: "2.1"
  owner: es-people-analytics
  description: "Percentage of employees who voluntarily separated"

  formula:
    numerator: "COUNT(separations WHERE separation_type = 'voluntary')"
    denominator: "AVG(active_headcount)"
    expression: "(numerator / denominator) * 100"

  dimensions:
    - department
    - job_level
    - tenure_band
    - location
    - manager_id
    - gender
    - ethnicity

  data_sources:
    primary: gold.employee_360
    supplementary:
      - gold.separation_events
      - gold.headcount_snapshots

  quality_checks:
    - type: range
      expected: "0-50%"
    - type: month_over_month_change
      threshold: "25%"
    - type: null_check
      fields: [separation_type, separation_date, employee_id]

  access_control:
    base: all_es_team
    restricted_dimensions:
      ethnicity: dei_reporting_group

  changelog:
    - version: "2.1"
      date: "2026-03-15"
      change: "Added tenure_band dimension"

Centralized computation engine for People Analytics metrics with validation and lineage tracking:

# metrics/engine/compute.py
class MetricEngine:
    """
    Core computation engine for People Analytics metrics.

    Responsibilities:
    - Parse metric YAML definitions
    - Generate and execute SQL against Gold tables
    - Run quality validation checks
    - Track lineage from source tables → metrics → consumers
    """

    def compute_metric(self, metric_name: str,
                       dimensions: list = None,
                       date_range: tuple = None):
        """
        Compute a metric by name with optional dimensional slicing.

        Example:
            engine.compute_metric(
                'voluntary_attrition_rate',
                dimensions=['department', 'tenure_band'],
                date_range=('2025-04-01', '2026-03-31')
            )
        """
        definition = self._load_definition(metric_name)

        # Generate SQL from YAML formula
        sql = self._generate_sql(definition, dimensions, date_range)

        # Execute against Gold layer
        raw_result = self._execute(sql)

        # Run quality checks
        quality_report = self._validate(raw_result,
                                         definition['quality_checks'])

        if quality_report.has_failures:
            self._alert_data_team(metric_name, quality_report)

        # Update lineage graph
        self._update_lineage(metric_name, definition['data_sources'])

        return MetricResult(
            metric_name=metric_name,
            value=raw_result,
            quality_status=quality_report.status,
            definition_version=definition['metric']['version']
        )

Designed to map directly to Agentforce architecture with Topics, Actions, and Guardrails:

# agent/es_pulse_agent.py
"""
ES Pulse Agent — Agentic Employee Sentiment & Attrition Intelligence
Designed to map directly to Agentforce architecture
"""

AGENT_TOPICS = {
    "attrition_analysis": {
        "description": "Analyze employee attrition patterns and root causes",
        "triggers": ["attrition", "turnover", "leaving", "quit"],
        "instructions": """
            1. Query the semantic metrics layer for quantitative data
            2. Search exit interview embeddings for qualitative themes
            3. Correlate with engagement survey sentiment
            4. Identify top contributing factors with confidence scores
            5. Compare to historical baselines and industry benchmarks
            6. Suggest data-backed interventions
        """
    },
    "engagement_insights": {
        "description": "Surface employee engagement trends and sentiment patterns",
        "triggers": ["engagement", "satisfaction", "morale", "sentiment"]
    }
}

AGENT_ACTIONS = {
    "query_metrics": {
        "description": "Query the ES semantic metrics layer",
        "function": "metric_engine.compute_metric",
        "parameters": ["metric_name", "dimensions", "date_range"]
    },
    "search_feedback": {
        "description": "Semantic search across employee feedback",
        "function": "vector_db.similarity_search",
        "parameters": ["query", "filters", "top_k"]
    },
    "detect_anomalies": {
        "description": "Identify statistical anomalies in HR metrics",
        "function": "anomaly_detector.detect",
        "parameters": ["metric_name", "lookback_period"]
    }
}

AGENT_GUARDRAILS = {
    "pii_protection": {
        "rule": "Never expose individual employee names or IDs",
        "action": "Aggregate to minimum group size of 5"
    },
    "access_control": {
        "rule": "Verify user's access level before returning sensitive dimensions",
        "action": "Check user role against metric access_control definitions"
    },
    "bias_detection": {
        "rule": "Flag when recommendations may have demographic bias",
        "action": "Run bias audit on any recommendation involving protected categories"
    }
}

Implements self-correcting retrieval with relevance grading, query transformation, and multi-hop reasoning:

# agent/adaptive_rag.py
"""
Adaptive RAG pipeline for ES Pulse Agent.
Implements self-correcting retrieval with relevance grading,
query transformation, and multi-hop reasoning.
"""

from langgraph.graph import StateGraph, END

def build_es_pulse_graph():
    """
    Build the adaptive reasoning graph for ES Pulse Agent.

    Flow:
    1. Classify question → route to appropriate topic
    2. Retrieve relevant context (metrics + unstructured feedback)
    3. Grade retrieval relevance — if low, transform query and retry
    4. Synthesize answer with confidence scoring
    5. Self-check: does answer fully address the question?
    6. If not, identify gaps and retrieve additional context
    """

    graph = StateGraph(AgentState)

    # Add nodes for each step
    graph.add_node("classify_question", classify_question)
    graph.add_node("retrieve_metrics", retrieve_metrics)
    graph.add_node("retrieve_feedback", retrieve_feedback)
    graph.add_node("grade_relevance", grade_relevance)
    graph.add_node("transform_query", transform_query)
    graph.add_node("synthesize_answer", synthesize_answer)
    graph.add_node("self_check", self_check)
    graph.add_node("apply_guardrails", apply_guardrails)

    # Set entry point
    graph.set_entry_point("classify_question")
    graph.add_edge("classify_question", "retrieve_metrics")

    # Conditional edge: if retrieval quality is low, retry with transformed query
    graph.add_conditional_edges(
        "grade_relevance",
        lambda state: "transform"
            if state["confidence"] < 0.7
            else "synthesize",
        {"transform": "transform_query",
         "synthesize": "synthesize_answer"}
    )

    # Self-check: is the answer complete?
    graph.add_conditional_edges(
        "self_check",
        lambda state: "retrieve_more"
            if not state.get("answer_complete", False)
            else "guardrails",
        {"retrieve_more": "retrieve_metrics",
         "guardrails": "apply_guardrails"}
    )

    graph.add_edge("apply_guardrails", END)

    return graph.compile()

# Example agent response:
#
# User: "Why did attrition spike 40% in Engineering last quarter?"
#
# Agent identifies root causes:
# - New VP Engineering hired (org restructuring uncertainty) — 35% weight
# - Compensation below market for ML Engineers — 28% weight
# - 3 team leads departed in Dec (cascade effect) — 22% weight
# - Remote work policy change announced Jan — 15% weight
#
# Returns data-backed analysis with recommendations