Building a Compensation Inquiry Agent: Pay Stub, Tax, and Benefits Questions

Why Compensation Questions Need an Agent

Payroll and benefits questions are among the most time-sensitive and anxiety-inducing inquiries employees have. "Why is my paycheck lower this month?", "How much goes to my HSA?", "What is the difference between my W-4 allowances and my actual tax rate?" These questions have precise answers buried in payroll systems that most employees cannot navigate. A compensation agent provides instant, clear answers while maintaining strict data security — because compensation data is among the most sensitive information in any organization.

Data Models for Compensation

from dataclasses import dataclass, field
from datetime import date
from typing import Optional
from agents import Agent, Runner, function_tool
import json

@dataclass
class PayStub:
    pay_period_end: date
    gross_pay: float
    federal_tax: float
    state_tax: float
    social_security: float
    medicare: float
    health_premium: float
    dental_premium: float
    vision_premium: float
    hsa_contribution: float
    retirement_401k: float
    other_deductions: dict[str, float] = field(default_factory=dict)

    @property
    def total_deductions(self) -> float:
        fixed = (self.federal_tax + self.state_tax + self.social_security
                 + self.medicare + self.health_premium + self.dental_premium
                 + self.vision_premium + self.hsa_contribution + self.retirement_401k)
        return fixed + sum(self.other_deductions.values())

    @property
    def net_pay(self) -> float:
        return self.gross_pay - self.total_deductions

@dataclass
class EmployeeCompensation:
    employee_id: str
    annual_salary: float
    pay_frequency: str  # "biweekly", "semi_monthly", "monthly"
    filing_status: str  # "single", "married_joint", "married_separate"
    federal_allowances: int
    state: str
    pay_stubs: list[PayStub] = field(default_factory=list)

@dataclass
class BenefitsAccount:
    account_type: str  # "hsa", "fsa", "401k"
    balance: float
    ytd_contributions: float
    employer_match: float
    annual_limit: float
    remaining_limit: float

COMPENSATION_DB: dict[str, EmployeeCompensation] = {}
BENEFITS_ACCOUNTS: dict[str, list[BenefitsAccount]] = {}

Pay Stub Explanation Tool

The most common compensation question is "Why does my paycheck look different?" The agent breaks down each line item and highlights changes from the previous period.

flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus<br/>classify"]
    PLAN["Plan and tool<br/>selection"]
    AGENT["Agent loop<br/>LLM plus tools"]
    GUARD{"Guardrails<br/>and policy"}
    EXEC["Execute and<br/>verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus<br/>next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff

@function_tool
def get_pay_stub(employee_id: str, period: str = "latest") -> str:
    """Retrieve and explain a pay stub for the specified period."""
    comp = COMPENSATION_DB.get(employee_id)
    if not comp:
        return json.dumps({"error": "Compensation record not found"})

    if not comp.pay_stubs:
        return json.dumps({"error": "No pay stubs available"})

    stub = comp.pay_stubs[-1]  # latest by default

    breakdown = {
        "pay_period_ending": str(stub.pay_period_end),
        "gross_pay": f"${stub.gross_pay:,.2f}",
        "deductions": {
            "Federal Income Tax": f"${stub.federal_tax:,.2f}",
            "State Income Tax": f"${stub.state_tax:,.2f}",
            "Social Security (6.2%)": f"${stub.social_security:,.2f}",
            "Medicare (1.45%)": f"${stub.medicare:,.2f}",
            "Health Insurance": f"${stub.health_premium:,.2f}",
            "Dental Insurance": f"${stub.dental_premium:,.2f}",
            "Vision Insurance": f"${stub.vision_premium:,.2f}",
            "HSA Contribution": f"${stub.hsa_contribution:,.2f}",
            "401(k) Contribution": f"${stub.retirement_401k:,.2f}",
        },
        "total_deductions": f"${stub.total_deductions:,.2f}",
        "net_pay": f"${stub.net_pay:,.2f}",
    }

    # Compare with previous period if available
    if len(comp.pay_stubs) >= 2:
        prev = comp.pay_stubs[-2]
        diff = stub.net_pay - prev.net_pay
        if abs(diff) > 1.0:
            changes = []
            if stub.federal_tax != prev.federal_tax:
                changes.append(f"Federal tax changed by ${stub.federal_tax - prev.federal_tax:+,.2f}")
            if stub.health_premium != prev.health_premium:
                changes.append(f"Health premium changed by ${stub.health_premium - prev.health_premium:+,.2f}")
            if stub.retirement_401k != prev.retirement_401k:
                changes.append(f"401(k) contribution changed by ${stub.retirement_401k - prev.retirement_401k:+,.2f}")
            breakdown["period_over_period"] = {
                "net_pay_change": f"${diff:+,.2f}",
                "contributing_factors": changes if changes else ["Minor rounding adjustments"],
            }

    return json.dumps(breakdown)

Tax Withholding Explanation Tool

@function_tool
def explain_tax_withholding(employee_id: str) -> str:
    """Explain how federal and state tax withholding is calculated."""
    comp = COMPENSATION_DB.get(employee_id)
    if not comp:
        return json.dumps({"error": "Compensation record not found"})

    pay_periods = {"biweekly": 26, "semi_monthly": 24, "monthly": 12}
    periods = pay_periods.get(comp.pay_frequency, 26)
    per_period_gross = comp.annual_salary / periods

    # Simplified 2026 federal bracket illustration
    brackets = [
        (11600, 0.10), (47150, 0.12), (100525, 0.22),
        (191950, 0.24), (243725, 0.32), (609350, 0.35),
    ]

    explanation = {
        "annual_salary": f"${comp.annual_salary:,.2f}",
        "pay_frequency": comp.pay_frequency,
        "gross_per_period": f"${per_period_gross:,.2f}",
        "filing_status": comp.filing_status,
        "federal_allowances": comp.federal_allowances,
        "state": comp.state,
        "note": "Federal withholding is based on IRS tax tables "
                "using your W-4 filing status and allowances. "
                "Actual withholding may differ slightly from the "
                "marginal bracket calculation due to per-period adjustments.",
        "how_to_adjust": "Submit an updated W-4 form to HR to change your "
                         "federal withholding. Use the IRS Tax Withholding "
                         "Estimator at irs.gov for guidance.",
    }

    return json.dumps(explanation)

Benefits Account Tool

@function_tool
def get_benefits_accounts(employee_id: str) -> str:
    """Get HSA, FSA, and 401(k) account details."""
    accounts = BENEFITS_ACCOUNTS.get(employee_id, [])
    if not accounts:
        return json.dumps({"message": "No benefits accounts found"})

    result = []
    for acct in accounts:
        entry = {
            "account_type": acct.account_type.upper(),
            "current_balance": f"${acct.balance:,.2f}",
            "ytd_contributions": f"${acct.ytd_contributions:,.2f}",
            "employer_match": f"${acct.employer_match:,.2f}",
            "annual_limit": f"${acct.annual_limit:,.2f}",
            "remaining_contribution_room": f"${acct.remaining_limit:,.2f}",
        }

        # Add account-specific guidance
        if acct.account_type == "hsa":
            entry["note"] = ("HSA funds roll over year to year and are yours to keep. "
                             "Triple tax advantage: pre-tax contributions, "
                             "tax-free growth, tax-free qualified withdrawals.")
        elif acct.account_type == "fsa":
            entry["note"] = ("FSA funds are use-it-or-lose-it. "
                             f"You have ${acct.remaining_limit:,.2f} remaining "
                             "to spend before the plan year ends.")
        elif acct.account_type == "401k":
            match_pct = (acct.employer_match / max(acct.ytd_contributions, 1)) * 100
            entry["note"] = (f"Your employer matches approximately {match_pct:.0f}% "
                             "of your contributions up to the matching limit.")

        result.append(entry)

    return json.dumps(result)

@function_tool
def update_contribution(
    employee_id: str,
    account_type: str,
    new_amount: float,
    effective_date: str,
) -> str:
    """Request a change to HSA, FSA, or 401(k) contribution amounts."""
    accounts = BENEFITS_ACCOUNTS.get(employee_id, [])
    target = next((a for a in accounts if a.account_type == account_type.lower()), None)

    if not target:
        return json.dumps({"error": f"No {account_type} account found"})

    # Validate against limits
    remaining_periods = 12  # simplified
    projected = target.ytd_contributions + (new_amount * remaining_periods)
    if projected > target.annual_limit:
        return json.dumps({
            "status": "warning",
            "message": f"Projected annual contribution of ${projected:,.2f} "
                       f"exceeds the ${target.annual_limit:,.2f} limit. "
                       f"Maximum per-period contribution: "
                       f"${(target.annual_limit - target.ytd_contributions) / remaining_periods:,.2f}",
        })

    return json.dumps({
        "status": "submitted",
        "account": account_type.upper(),
        "new_per_period_amount": f"${new_amount:,.2f}",
        "effective_date": effective_date,
        "note": "Changes take effect on the next full pay period after the effective date.",
    })

compensation_agent = Agent(
    name="CompBot",
    instructions="""You are CompBot, a compensation and benefits assistant.
Help employees understand their pay stubs, tax withholdings, and benefits accounts.
Always verify the employee's identity before sharing compensation data.
Explain deductions in plain language, avoiding jargon.
For tax advice beyond withholding mechanics, direct employees to a tax professional.
Never share one employee's compensation data with another.""",
    tools=[get_pay_stub, explain_tax_withholding, get_benefits_accounts, update_contribution],
)

FAQ

How do you secure compensation data in the agent?

Implement strict authentication before every tool call. The agent verifies the requesting user's identity against the employee ID in the query. All compensation data is encrypted at rest and in transit. Audit logs record every data access with timestamps and the authenticated user ID.

What if an employee's pay stub has an actual error?

The agent can identify potential errors (such as a deduction that was not authorized or a gross pay discrepancy) and flag them for payroll review. However, the agent never modifies payroll records directly. It generates a payroll inquiry ticket that routes to the payroll team with the specific discrepancy details.

How do you handle employees in multiple states?

Employees who work in multiple states may owe taxes in each state. The agent explains which state withholdings apply based on the employee's work location records and home state. For complex multi-state situations, the agent recommends consulting with a tax professional while providing the factual withholding details from payroll.

#Compensation #Payroll #TaxWithholding #BenefitsEnrollment #AgenticAI #LearnAI #AIEngineering