Skip to content

nhra_gt.agents.base

Base classes and orchestrators for NHRA strategic agents.

This module defines the interfaces for strategic actors in the simulation, including heuristic models and LLM-driven agents. It also provides utilities for generating narratives and validating strategic traces.

Classes

Agent

Bases: ABC

Abstract base class for all NHRA strategic agents.

Source code in src/nhra_gt/agents/base.py 
34
35
36
37
38
39
class Agent(ABC):
    """Abstract base class for all NHRA strategic agents."""

    @abstractmethod
    def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
        """Choose strategies based on current system state."""

Functions

decide(state, params, rng) abstractmethod

Choose strategies based on current system state.

Source code in src/nhra_gt/agents/base.py 
37
38
39
@abstractmethod
def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
    """Choose strategies based on current system state."""

BriefGenerator

Converts simulation state into a textual narrative for LLM consumption.

Source code in src/nhra_gt/agents/base.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class BriefGenerator:
    """Converts simulation state into a textual narrative for LLM consumption."""

    @staticmethod
    def generate(state: State, params: Params, role: str) -> str:
        """Generate a policy brief for a specific role (Commonwealth/State/Provider)."""
        mode_desc = {
            "normal": "The system is currently stable.",
            "stress": "The system is under significant pressure.",
            "crisis": "CRITICAL: The system is in a state of failure (Code Red).",
            "recovery": "The system is recovering from a crisis.",
        }

        # Get values from first jurisdiction if available, fallback to defaults
        eff_gap = state.reported_efficiency_gap
        eff_share = (
            state.jurisdictions[0].effective_cth_share
            if hasattr(state, "jurisdictions") and state.jurisdictions
            else 0.38
        )
        pol_cap = (
            state.jurisdictions[0].political_capital
            if hasattr(state, "jurisdictions") and state.jurisdictions
            else 1.0
        )
        eq_idx = (
            state.jurisdictions[0].equity_index
            if hasattr(state, "jurisdictions") and state.jurisdictions
            else 1.0
        )

        brief = f"""
### NHRA POLICY BRIEF - {role.upper()}
**Current Year:** {state.year}, Month {state.month}
**System Mode:** {state.system_mode.value.upper()} - {mode_desc.get(state.system_mode.value, "")}

**Key Metrics:**
- System Pressure: {state.pressure:.2f} (Base: 1.0)
- ED Wait Time: {state.within4 * 100:.1f}% within 4h
- Efficiency Gap: {eff_gap * 100:.1f}% (Cost vs NEP)
- Effective Cth Share: {eff_share * 100:.1f}%
- Political Capital: {pol_cap:.2f}
- Equity Index: {eq_idx:.2f}

**Strategic Context:**
- Your goal is to maximize your utility while maintaining system stability.
- Agreement (A) restores trust but may increase bailout expectations.
- Upcoding (U) increases revenue but risks Auditor detection and penalties.
- Fragmentation (F) in Aged Care/NDIS reduces your immediate load but worsens downstream pressure.

**Decision Required:**
You must choose strategies for the current period.
"""
        return brief

Functions

generate(state, params, role) staticmethod

Generate a policy brief for a specific role (Commonwealth/State/Provider).

Source code in src/nhra_gt/agents/base.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
    @staticmethod
    def generate(state: State, params: Params, role: str) -> str:
        """Generate a policy brief for a specific role (Commonwealth/State/Provider)."""
        mode_desc = {
            "normal": "The system is currently stable.",
            "stress": "The system is under significant pressure.",
            "crisis": "CRITICAL: The system is in a state of failure (Code Red).",
            "recovery": "The system is recovering from a crisis.",
        }

        # Get values from first jurisdiction if available, fallback to defaults
        eff_gap = state.reported_efficiency_gap
        eff_share = (
            state.jurisdictions[0].effective_cth_share
            if hasattr(state, "jurisdictions") and state.jurisdictions
            else 0.38
        )
        pol_cap = (
            state.jurisdictions[0].political_capital
            if hasattr(state, "jurisdictions") and state.jurisdictions
            else 1.0
        )
        eq_idx = (
            state.jurisdictions[0].equity_index
            if hasattr(state, "jurisdictions") and state.jurisdictions
            else 1.0
        )

        brief = f"""
### NHRA POLICY BRIEF - {role.upper()}
**Current Year:** {state.year}, Month {state.month}
**System Mode:** {state.system_mode.value.upper()} - {mode_desc.get(state.system_mode.value, "")}

**Key Metrics:**
- System Pressure: {state.pressure:.2f} (Base: 1.0)
- ED Wait Time: {state.within4 * 100:.1f}% within 4h
- Efficiency Gap: {eff_gap * 100:.1f}% (Cost vs NEP)
- Effective Cth Share: {eff_share * 100:.1f}%
- Political Capital: {pol_cap:.2f}
- Equity Index: {eq_idx:.2f}

**Strategic Context:**
- Your goal is to maximize your utility while maintaining system stability.
- Agreement (A) restores trust but may increase bailout expectations.
- Upcoding (U) increases revenue but risks Auditor detection and penalties.
- Fragmentation (F) in Aged Care/NDIS reduces your immediate load but worsens downstream pressure.

**Decision Required:**
You must choose strategies for the current period.
"""
        return brief

AuditorValidator

Evaluates agent strategic traces from an 'Auditor' perspective. Scores realism and detects unlikely gaming behavior.

Source code in src/nhra_gt/agents/base.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class AuditorValidator:
    """
    Evaluates agent strategic traces from an 'Auditor' perspective.
    Scores realism and detects unlikely gaming behavior.
    """

    def validate(self, trace: list[dict[str, Any]]) -> dict[str, Any]:
        """Evaluates agent strategic traces from an 'Auditor' perspective.

        Scores realism and detects unlikely gaming behavior.

        Args:
            trace: A list of step dictionaries containing state, strategy, and rationale.

        Returns:
            A dictionary with keys `realism_score` (float) and `findings` (list of str).
        """
        # Logic would involve an LLM prompt: "Does this look like a real hospital administrator?"
        # For now, return a placeholder score.
        return {
            "realism_score": 0.85,
            "findings": ["Gaming behavior consistent with high pressure periods."],
        }

Functions

validate(trace)

Evaluates agent strategic traces from an 'Auditor' perspective.

Scores realism and detects unlikely gaming behavior.

Parameters:

Name Type Description Default
trace list[dict[str, Any]]

A list of step dictionaries containing state, strategy, and rationale.

 required

Returns:

Type Description
dict[str, Any]

A dictionary with keys realism_score (float) and findings (list of str).
Source code in src/nhra_gt/agents/base.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def validate(self, trace: list[dict[str, Any]]) -> dict[str, Any]:
    """Evaluates agent strategic traces from an 'Auditor' perspective.

    Scores realism and detects unlikely gaming behavior.

    Args:
        trace: A list of step dictionaries containing state, strategy, and rationale.

    Returns:
        A dictionary with keys `realism_score` (float) and `findings` (list of str).
    """
    # Logic would involve an LLM prompt: "Does this look like a real hospital administrator?"
    # For now, return a placeholder score.
    return {
        "realism_score": 0.85,
        "findings": ["Gaming behavior consistent with high pressure periods."],
    }

LLMAgent

Bases: Agent

An agent that uses a Large Language Model to make strategic decisions.

Source code in src/nhra_gt/agents/base.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class LLMAgent(Agent):
    """
    An agent that uses a Large Language Model to make strategic decisions.
    """

    def __init__(self, role: str, model_name: str = "gemini-pro"):
        self.role = role
        self.model_name = model_name
        self.brief_gen = BriefGenerator()

    def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
        """
        Generates a brief, calls the LLM, and parses the strategic response.
        Note: Actual LLM call is stubbed/delegated to the environment.
        """
        self.brief_gen.generate(state, params, self.role)

        # In a real implementation, we would call the LLM here.
        # For the simulation engine, we can use a callback or a mock
        # that uses the HeuristicAgent as a 'smart fallback'.

        # MOCK/STUB logic for now:
        heuristic_response = CommonwealthAgent().decide(state, params, rng)
        heuristic_response.update(JurisdictionAgent().decide(state, params, rng))
        heuristic_response.update(LHNAgent().decide(state, params, rng))

        # Add 'Rationale' for Cognitive Trace
        heuristic_response["RATIONALE"] = (
            f"Decided based on multi-agent refactor for {self.role}. Pressure is {state.pressure:.2f}."
        )

        return heuristic_response

Functions

decide(state, params, rng)

Generates a brief, calls the LLM, and parses the strategic response. Note: Actual LLM call is stubbed/delegated to the environment.

Source code in src/nhra_gt/agents/base.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
    """
    Generates a brief, calls the LLM, and parses the strategic response.
    Note: Actual LLM call is stubbed/delegated to the environment.
    """
    self.brief_gen.generate(state, params, self.role)

    # In a real implementation, we would call the LLM here.
    # For the simulation engine, we can use a callback or a mock
    # that uses the HeuristicAgent as a 'smart fallback'.

    # MOCK/STUB logic for now:
    heuristic_response = CommonwealthAgent().decide(state, params, rng)
    heuristic_response.update(JurisdictionAgent().decide(state, params, rng))
    heuristic_response.update(LHNAgent().decide(state, params, rng))

    # Add 'Rationale' for Cognitive Trace
    heuristic_response["RATIONALE"] = (
        f"Decided based on multi-agent refactor for {self.role}. Pressure is {state.pressure:.2f}."
    )

    return heuristic_response

CommonwealthAgent

Bases: Agent

The Principal (Federal Govt) setting price and compliance rules.

Focuses on fiscal sustainability, compliance targets, and framing the NEP.

Source code in src/nhra_gt/agents/base.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class CommonwealthAgent(Agent):
    """
    The Principal (Federal Govt) setting price and compliance rules.

    Focuses on fiscal sustainability, compliance targets, and framing the NEP.
    """

    def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
        """Chooses framing (DEF) and compliance (COMP) strategies."""
        obs_pressure = getattr(state, "reported_pressure", state.pressure)
        obs_efficiency_gap = state.reported_efficiency_gap

        # Commonwealth moves: DEF, COMP, BARG(offer)
        results = {}

        # 1. Compliance Strategy (COMP)
        comp_prob = logistic(0.9 * params.audit_pressure - 0.7 * obs_efficiency_gap)
        results["COMP"] = "T" if rng.random() < comp_prob else "L"

        # 2. Framing Strategy (DEF)
        def_prob = logistic(1.3 * (obs_efficiency_gap - 0.25) + 0.9 * (obs_pressure - 1.0))
        results["DEF"] = "R" if rng.random() < def_prob else "E"

        return results

Functions

decide(state, params, rng)

Chooses framing (DEF) and compliance (COMP) strategies.

Source code in src/nhra_gt/agents/base.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
    """Chooses framing (DEF) and compliance (COMP) strategies."""
    obs_pressure = getattr(state, "reported_pressure", state.pressure)
    obs_efficiency_gap = state.reported_efficiency_gap

    # Commonwealth moves: DEF, COMP, BARG(offer)
    results = {}

    # 1. Compliance Strategy (COMP)
    comp_prob = logistic(0.9 * params.audit_pressure - 0.7 * obs_efficiency_gap)
    results["COMP"] = "T" if rng.random() < comp_prob else "L"

    # 2. Framing Strategy (DEF)
    def_prob = logistic(1.3 * (obs_efficiency_gap - 0.25) + 0.9 * (obs_pressure - 1.0))
    results["DEF"] = "R" if rng.random() < def_prob else "E"

    return results

JurisdictionAgent

Bases: Agent

The Intermediary (State/Territory Govt) managing the Agreement.

Focuses on budget reconciliation, political capital, and cost-shifting.

Source code in src/nhra_gt/agents/base.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
class JurisdictionAgent(Agent):
    """
    The Intermediary (State/Territory Govt) managing the Agreement.

    Focuses on budget reconciliation, political capital, and cost-shifting.
    """

    def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
        """Chooses bargaining (BARG), cost-shifting (SHIFT), and governance (GOV) strategies."""
        obs_pressure = getattr(state, "reported_pressure", state.pressure)
        obs_efficiency_gap = state.reported_efficiency_gap

        # Get bailout expectation from first jurisdiction
        bailout_exp = (
            state.jurisdictions[0].bailout_expectation
            if hasattr(state, "jurisdictions") and state.jurisdictions
            else 0.0
        )

        # State moves: BARG, SHIFT, GOV
        results = {}

        # 1. Bargaining Strategy (BARG)
        barg_prob = logistic(0.6 * (1.2 - obs_pressure) + bailout_exp)
        results["BARG"] = "A" if rng.random() < barg_prob else "D"

        # 2. Cost Shifting (SHIFT)
        shift_prob = logistic(-1.1 * (obs_pressure - 1.0) - 1.0 * obs_efficiency_gap)
        results["SHIFT"] = "I" if rng.random() < shift_prob else "S"

        # 3. Governance (GOV)
        gov_prob = logistic(-0.8 * (obs_pressure - 1.0) - 0.7 * params.political_salience)
        results["GOV"] = "I" if rng.random() < gov_prob else "S"

        return results

Functions

decide(state, params, rng)

Chooses bargaining (BARG), cost-shifting (SHIFT), and governance (GOV) strategies.

Source code in src/nhra_gt/agents/base.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
    """Chooses bargaining (BARG), cost-shifting (SHIFT), and governance (GOV) strategies."""
    obs_pressure = getattr(state, "reported_pressure", state.pressure)
    obs_efficiency_gap = state.reported_efficiency_gap

    # Get bailout expectation from first jurisdiction
    bailout_exp = (
        state.jurisdictions[0].bailout_expectation
        if hasattr(state, "jurisdictions") and state.jurisdictions
        else 0.0
    )

    # State moves: BARG, SHIFT, GOV
    results = {}

    # 1. Bargaining Strategy (BARG)
    barg_prob = logistic(0.6 * (1.2 - obs_pressure) + bailout_exp)
    results["BARG"] = "A" if rng.random() < barg_prob else "D"

    # 2. Cost Shifting (SHIFT)
    shift_prob = logistic(-1.1 * (obs_pressure - 1.0) - 1.0 * obs_efficiency_gap)
    results["SHIFT"] = "I" if rng.random() < shift_prob else "S"

    # 3. Governance (GOV)
    gov_prob = logistic(-0.8 * (obs_pressure - 1.0) - 0.7 * params.political_salience)
    results["GOV"] = "I" if rng.random() < gov_prob else "S"

    return results

LHNAgent

Bases: Agent

The Operator (Hospital/LHN) managing clinical demand and revenue.

Focuses on upcoding, venue shifting, and coordination with aged care/NDIS.

Source code in src/nhra_gt/agents/base.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class LHNAgent(Agent):
    """
    The Operator (Hospital/LHN) managing clinical demand and revenue.

    Focuses on upcoding, venue shifting, and coordination with aged care/NDIS.
    """

    def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
        """Chooses coding, venue, and clinical coordination strategies."""
        obs_pressure = getattr(state, "reported_pressure", state.pressure)
        obs_efficiency_gap = state.reported_efficiency_gap

        # LHN moves: CODING, VENUE_SHIFT, DISC, AGED, NDIS, COMPETITION
        results = {}

        # 1. Revenue Strategy (CODING)
        coding_prob = logistic(1.5 * (obs_pressure - 1.1) + 1.2 * obs_efficiency_gap)
        results["CODING"] = "U" if rng.random() < coding_prob else "H"

        # 2. Venue Selection (VENUE_SHIFT)
        venue_prob = logistic(1.2 * (obs_pressure - 1.1) + 0.8 * obs_efficiency_gap)
        results["VENUE_SHIFT"] = "B" if rng.random() < venue_prob else "A"

        # 3. Operations Coordination (DISC, AGED, NDIS)
        results["DISC"] = "C" if rng.random() < 0.7 else "F"
        results["AGED"] = "C" if rng.random() < 0.6 else "F"
        results["NDIS"] = "C" if rng.random() < 0.6 else "F"

        # 4. Competition Strategy
        comp_prob = logistic(1.1 * (obs_pressure - 1.0) + 0.5 * params.cannibalization_beta)
        results["COMPETITION"] = "A" if rng.random() < comp_prob else "M"

        return results

Functions

decide(state, params, rng)

Chooses coding, venue, and clinical coordination strategies.

Source code in src/nhra_gt/agents/base.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
    """Chooses coding, venue, and clinical coordination strategies."""
    obs_pressure = getattr(state, "reported_pressure", state.pressure)
    obs_efficiency_gap = state.reported_efficiency_gap

    # LHN moves: CODING, VENUE_SHIFT, DISC, AGED, NDIS, COMPETITION
    results = {}

    # 1. Revenue Strategy (CODING)
    coding_prob = logistic(1.5 * (obs_pressure - 1.1) + 1.2 * obs_efficiency_gap)
    results["CODING"] = "U" if rng.random() < coding_prob else "H"

    # 2. Venue Selection (VENUE_SHIFT)
    venue_prob = logistic(1.2 * (obs_pressure - 1.1) + 0.8 * obs_efficiency_gap)
    results["VENUE_SHIFT"] = "B" if rng.random() < venue_prob else "A"

    # 3. Operations Coordination (DISC, AGED, NDIS)
    results["DISC"] = "C" if rng.random() < 0.7 else "F"
    results["AGED"] = "C" if rng.random() < 0.6 else "F"
    results["NDIS"] = "C" if rng.random() < 0.6 else "F"

    # 4. Competition Strategy
    comp_prob = logistic(1.1 * (obs_pressure - 1.0) + 0.5 * params.cannibalization_beta)
    results["COMPETITION"] = "A" if rng.random() < comp_prob else "M"

    return results

HeuristicAgent

Bases: Agent

Orchestrator that delegates to distinct Commonwealth, State, and LHN agents.

Source code in src/nhra_gt/agents/base.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
class HeuristicAgent(Agent):
    """
    Orchestrator that delegates to distinct Commonwealth, State, and LHN agents.
    """

    def decide(self, state: State, params: Params, rng: np.random.Generator) -> dict[str, Any]:
        cw = CommonwealthAgent()
        jr = JurisdictionAgent()
        lhn = LHNAgent()

        results = {
            "SIGNAL": "L",
            "SIGNAL_QUALITY": 0.9,
        }

        # Aggregate decisions from all levels
        results.update(cw.decide(state, params, rng))
        results.update(jr.decide(state, params, rng))
        results.update(lhn.decide(state, params, rng))

        results["RATIONALE"] = f"Multi-agent consensus reached in {params.orchestration_mode} mode."
        return results

Functions

logistic(x)

Standard sigmoid function.

Source code in src/nhra_gt/agents/base.py 
21
22
23
def logistic(x: float) -> float:
    """Standard sigmoid function."""
    return 1.0 / (1.0 + math.exp(-x))

softmax(u, tau=0.25)

Numpy implementation of tau-tempered softmax.

Source code in src/nhra_gt/agents/base.py 
26
27
28
29
30
31
def softmax(u: np.ndarray, tau: float = 0.25) -> np.ndarray:
    """Numpy implementation of tau-tempered softmax."""
    u = np.asarray(u, dtype=float)
    u = u - u.max()
    z = np.exp(u / max(1e-9, tau))
    return np.asarray(z / z.sum(), dtype=float)