pacing

PACING: Platform for Agentic Clinical Intelligence with Networked Graphs

A clinical decision support system for Substance Use Disorder (SUD) treatment with two main operational modes:

1. Live Session Mode: Real-time audio transcription with parallel agentic sidecars

2. Longitudinal Analysis Mode: Risk assessment and What-If scenario simulation

Key Features: - Extensible plugin architecture with dependency injection - Privacy-aware: DEV_MODE (persists data) vs PROD_MODE (ephemeral) - Auditable: Human-in-the-loop verification with UncertaintyAuditor - Explainable: Risk factors with evidence and contribution scores

class pacing.AdaptiveConfidenceTranscriber(script: List[str] | None = None, latency_ms: float = 50.0, base_confidence: float = 0.85, confidence_variance: float = 0.1)

Mock transcriber that adjusts confidence based on text characteristics.

This variant simulates more realistic confidence scores by: - Lowering confidence for longer utterances (harder to transcribe) - Lowering confidence for utterances with medical/technical terms - Lowering confidence for utterances with negations (tricky)

async transcribe_chunk(audio_chunk: ndarray, sample_rate: int, is_final: bool = False) → TranscriptionResult

Transcribe with adaptive confidence.

class pacing.Event(*, event_id: str, event_type: ~pacing.models.data_models.EventType, description: str, date: ~datetime.datetime, impact_score: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=-1.0), ~annotated_types.Le(le=1.0)] = None, metadata: ~typing.Dict[str, ~typing.Any] = <factory>)

A life event in the patient’s history.

Events represent significant occurrences that may impact relapse risk.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pacing.EventType(value)

Types of life events tracked in the patient graph.

class pacing.Intervention(*, intervention_id: str, intervention_type: InterventionType, description: str, start_date: datetime, end_date: datetime | None = None, effectiveness_score: Annotated[float | None, Ge(ge=0.0), Le(le=1.0)] = None)

A clinical intervention in the patient’s treatment plan.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pacing.InterventionType(value)

Types of clinical interventions.

class pacing.MockAudioProvider(sample_rate: int = 16000, chunk_duration_ms: int = 100, total_duration_sec: float = 60.0)

Mock audio provider that generates synthetic audio signals.

This implementation is useful for: - Testing the platform without audio hardware - Demonstrating the system in controlled scenarios - Development and CI/CD pipelines

The mock provider can: - Generate silence - Generate synthetic tones - Simulate realistic audio chunk timing

get_audio_chunks() → Iterator[ndarray]

Generate mock audio chunks.

Yields:

np.ndarray – Synthetic audio data (float32, normalized to [-1, 1])

async get_audio_chunks_async() → AsyncIterator[ndarray]

Asynchronously generate mock audio chunks.

Yields:

np.ndarray – Synthetic audio data

get_sample_rate() → int

Get the sample rate.

property is_streaming: bool

Check if currently streaming.

start_stream() → None

Start the mock audio stream.

stop_stream() → None

Stop the mock audio stream.

class pacing.MockBayesianModel(base_risk: float = 0.5)

Mock risk model using simple rule-based logic.

This implementation demonstrates the IRiskModel interface without requiring actual Bayesian network inference. It uses heuristics like: - Recent substance use increases risk - Negative life events increase risk - Active interventions decrease risk - Longer sobriety decreases risk

Risk Score Calculation: 1. Start with base risk (0.50) 2. Adjust based on recent substance use 3. Adjust based on recent negative events 4. Adjust based on active interventions 5. Clamp to [0.0, 1.0]

calculate_risk(patient_data: PatientGraph, options: Dict[str, Any] | None = None) → RiskReport

Calculate risk using simple heuristics.

Parameters:

• patient_data – Patient clinical history

• options – Optional configuration

Returns:

Risk assessment with contributing factors

Return type:

RiskReport

get_model_metadata() → dict

Get model metadata.

get_model_version() → str

Get model version.

class pacing.MockSimulationModel(base_risk: float = 0.5)

Mock model that supports What-If simulation.

Extends MockBayesianModel to enable scenario comparison.

calculate_risk_delta(baseline: PatientGraph, modified: PatientGraph, options: Dict[str, Any] | None = None) → Dict[str, Any]

Calculate risk change between baseline and modified scenarios.

Parameters:

• baseline – Current patient state

• modified – Hypothetical patient state

• options – Optional configuration

Returns:

Risk comparison with delta and explanation

Return type:

dict

class pacing.MockTranscriber(script: List[str] | None = None, latency_ms: float = 50.0, base_confidence: float = 0.85, confidence_variance: float = 0.1)

Mock transcriber that yields pre-scripted text.

This implementation simulates a real transcription service by: - Returning text from a pre-defined script - Adding realistic latency (processing delay) - Generating plausible confidence scores - Supporting partial (streaming) results

Useful for: - Testing the platform without API costs - Creating reproducible demonstrations - Development and CI/CD

get_model_info() → dict

Get mock model information.

reset_script() → None

Reset the script index to start from the beginning.

supports_speaker_diarization() → bool

Mock transcriber does not support speaker diarization.

async transcribe_chunk(audio_chunk: ndarray, sample_rate: int, is_final: bool = False) → TranscriptionResult

Simulate transcription of an audio chunk.

Parameters:

• audio_chunk – Audio samples (ignored in mock)

• sample_rate – Sample rate (ignored in mock)

• is_final – Whether this is the final chunk

Returns:

Mock transcription with confidence score

Return type:

TranscriptionResult

class pacing.Mutation(mutation_type: MutationType, parameters: Dict[str, Any], description: str | None = None)

A modification to apply to a patient graph for simulation.

Mutations represent “What If” questions: - What if housing became stable? - What if the patient started medication-assisted treatment? - What if the patient experienced a traumatic event?

apply(graph: PatientGraph) → PatientGraph

Apply this mutation to a patient graph.

Parameters:

graph – The graph to modify

Returns:

A new graph with the mutation applied

Return type:

PatientGraph

class pacing.MutationType(value)

Types of mutations that can be applied to a patient graph.

class pacing.OperatingMode(value)

Operating mode for the platform.

DEV_MODE: Persists raw data (audio, transcripts) for debugging and auditing PROD_MODE: Ephemeral raw data, only structured extractions are retained

class pacing.PacingPlatform(operating_mode: OperatingMode = OperatingMode.PROD_MODE, transcriber: ITranscriber | None = None, risk_model: IRiskModel | None = None)

Main PACING platform class.

This is the primary interface for initializing and configuring the PACING system. It provides dependency injection for all major components: - Audio providers - Transcribers - Sidecar agents - Risk models - Operating mode (DEV/PROD)

Example Usage:

# Initialize platform in DEV mode platform = PacingPlatform(operating_mode=OperatingMode.DEV_MODE)

# Inject custom components platform.set_transcriber(DeepgramTranscriber(api_key=…)) platform.set_risk_model(MyBayesianNetwork())

# Register agents platform.register_agent(UncertaintyAuditor(confidence_threshold=0.70)) platform.register_agent(GuideAgent()) platform.register_agent(ScribeAgent())

# Start a session session = SessionMetadata(

session_id=”session_001”, patient_id=”patient_123”, clinician_id=”clinician_456”, start_time=datetime.now()

)

audio = MyAudioProvider() await platform.start_live_session(session, audio)

calculate_risk(patient_graph: PatientGraph, options: Dict[str, Any] | None = None)

Calculate relapse risk for a patient (Longitudinal Mode).

Parameters:

• patient_graph – Patient clinical history

• options – Optional model configuration

Returns:

Risk assessment

Return type:

RiskReport

Raises:

ValueError – If risk model not configured

get_platform_status() → dict

Get current platform status.

Returns:

Platform configuration and status

Return type:

dict

get_simulation_context(patient_graph: PatientGraph)

Create a simulation context for What-If analysis.

Parameters:

patient_graph – Patient clinical history

Returns:

Context for running simulations

Return type:

SimulationContext

Raises:

ValueError – If risk model doesn’t support simulation

register_agent(agent: ISidecarAgent) → None

Register a sidecar agent.

Parameters:

agent – Sidecar agent implementation

set_risk_model(model: IRiskModel) → None

Inject a custom risk model.

Parameters:

model – Risk model implementation

set_transcriber(transcriber: ITranscriber) → None

Inject a custom transcriber.

Parameters:

transcriber – Transcriber implementation

async start_live_session(session_metadata: SessionMetadata, audio_provider: IAudioProvider) → None

Start a live clinical session.

Parameters:

• session_metadata – Session information

• audio_provider – Audio source

Raises:

ValueError – If transcriber not configured

async stop_live_session() → None

Stop the current live session.

unregister_agent(agent: ISidecarAgent) → None

Unregister a sidecar agent.

Parameters:

agent – Sidecar agent to remove

class pacing.PatientGraph(*, patient_id: str, events: ~typing.List[~pacing.models.data_models.Event] = <factory>, substance_use_records: ~typing.List[~pacing.models.data_models.SubstanceUse] = <factory>, interventions: ~typing.List[~pacing.models.data_models.Intervention] = <factory>, edges: ~typing.List[~pacing.models.data_models.GraphEdge] = <factory>, metadata: ~typing.Dict[str, ~typing.Any] = <factory>)

A structured representation of a patient’s clinical history.

This graph contains nodes (events, substance use, interventions) and edges (temporal/causal relationships) that are used for longitudinal reasoning.

get_all_node_ids() → List[str]

Get all node IDs in the graph.

>>> from datetime import datetime
>>> graph = PatientGraph(patient_id="p123")
>>> graph.get_all_node_ids()
[]

>>> graph.events.append(Event(
...     event_id="e1", event_type=EventType.JOB_CHANGE,
...     description="test", date=datetime.now()
... ))
>>> "e1" in graph.get_all_node_ids()
True

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pacing.RiskReport(*, patient_id: str, risk_score: ~typing.Annotated[float, ~annotated_types.Ge(ge=0.0), ~annotated_types.Le(le=1.0)], risk_factors: ~typing.List[~pacing.models.data_models.RiskFactor] = <factory>, timestamp: ~datetime.datetime = <factory>, model_version: str = 'default', confidence_interval: tuple[float, float] | None = None)

Output from a risk assessment model.

patient_id

Patient identifier

Type:

str

risk_score

Overall relapse risk probability (0.0-1.0)

Type:

float

risk_factors

Contributing factors ranked by importance

Type:

List[pacing.models.data_models.RiskFactor]

timestamp

When this assessment was generated

Type:

datetime.datetime

model_version

Identifier for the model that generated this report

Type:

str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pacing.SessionMetadata(*, session_id: str, patient_id: str, clinician_id: str, start_time: datetime, end_time: datetime | None = None, session_type: str = 'counseling')

Metadata for a clinical session.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pacing.SimulationContext(baseline_graph: PatientGraph, model: ISimulationModel)

Context for running What-If simulations on patient data.

The SimulationContext: 1. Takes a baseline patient graph (current state) 2. Applies hypothetical mutations (e.g., “What if housing stable?”) 3. Runs risk models on both baseline and modified graphs 4. Compares the results to show the impact of the hypothetical change

Example Usage:

# Load patient data patient_graph = load_patient_data(patient_id)

# Create simulation context model = MockSimulationModel() sim = SimulationContext(patient_graph, model)

# Run “What If” scenario result = sim.simulate_mutation(

Mutation(

MutationType.MODIFY_HOUSING, {“status”: “stable”, “impact_score”: 0.8}, description=”Housing becomes stable”

)

)

print(f”Current risk: {result[‘baseline_risk’]:.2%}”) print(f”Predicted risk with stable housing: {result[‘modified_risk’]:.2%}”) print(f”Risk reduction: {abs(result[‘delta’]):.2%}”)

Design Philosophy: - Simulations are non-destructive (original data never modified) - Multiple scenarios can be compared side-by-side - Results include explanations for interpretability - Suitable for clinical decision support and patient education

compare_scenarios(scenarios: Dict[str, List[Mutation]], options: Dict[str, Any] | None = None) → Dict[str, Dict[str, Any]]

Compare multiple alternative scenarios side-by-side.

Example

scenarios = {

“Scenario A: MAT Only”: [mat_mutation], “Scenario B: MAT + Housing”: [mat_mutation, housing_mutation], “Scenario C: MAT + Housing + Employment”: [mat_mutation, housing_mutation, job_mutation]

}

comparison = sim.compare_scenarios(scenarios)

Parameters:

• scenarios – Dict mapping scenario names to mutation lists

• options – Optional model configuration

Returns:

Results for each scenario

Return type:

dict

get_simulation_history() → List[Dict[str, Any]]

Get the history of all simulations run in this context.

Returns:

List of simulation results

Return type:

List[Dict[str, Any]]

reset_baseline(new_baseline: PatientGraph) → None

Update the baseline graph (e.g., when patient data changes).

Parameters:

new_baseline – The new baseline patient state

simulate_multiple_mutations(mutations: List[Mutation], options: Dict[str, Any] | None = None) → Dict[str, Any]

Simulate the combined impact of multiple mutations.

This allows exploring complex scenarios like: “What if housing stabilizes AND patient starts MAT AND gets employed?”

Parameters:

• mutations – List of mutations to apply together

• options – Optional model configuration

Returns:

Simulation results with risk comparison

Return type:

dict

simulate_mutation(mutation: Mutation, options: Dict[str, Any] | None = None) → Dict[str, Any]

Simulate the impact of a single mutation.

Parameters:

• mutation – The hypothetical change to apply

• options – Optional model configuration

Returns:

Simulation results with risk comparison

Return type:

dict

class pacing.SubstanceType(value)

Types of substances tracked.

class pacing.SubstanceUse(*, use_id: str, substance_type: SubstanceType, status: SubstanceUseStatus, date: datetime, severity: Annotated[int | None, Ge(ge=1), Le(le=10)] = None, notes: str | None = None)

A substance use event or status change.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pacing.SubstanceUseStatus(value)

Status of substance use.

class pacing.TranscriptionResult(*, text: str, timestamp: ~datetime.datetime = <factory>, confidence_score: ~typing.Annotated[float, ~annotated_types.Ge(ge=0.0), ~annotated_types.Le(le=1.0)], speaker_id: str | None = None, is_partial: bool = False)

Result from a transcription operation.

text

The transcribed text

Type:

str

timestamp

When this transcription was generated

Type:

datetime.datetime

confidence_score

Acoustic confidence (0.0-1.0)

Type:

float

speaker_id

Optional speaker identification

Type:

str | None

is_partial

Whether this is a partial (streaming) result

Type:

bool

property confidence_level: ConfidenceLevel

Categorize confidence score into levels.

>>> from datetime import datetime
>>> t1 = TranscriptionResult(text="hello", confidence_score=0.95)
>>> t1.confidence_level
<ConfidenceLevel.HIGH: 'high'>

>>> t2 = TranscriptionResult(text="world", confidence_score=0.75)
>>> t2.confidence_level
<ConfidenceLevel.MEDIUM: 'medium'>

>>> t3 = TranscriptionResult(text="test", confidence_score=0.5)
>>> t3.confidence_level
<ConfidenceLevel.LOW: 'low'>

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pacing.UncertaintyAuditor(confidence_threshold: float = 0.7, max_queue_size: int = 100, auto_flag_medical_terms: bool = True)

Sidecar agent that audits transcription confidence and flags uncertain segments.

The UncertaintyAuditor implements a critical safety mechanism in clinical decision support: it identifies transcription segments where the acoustic model has low confidence and queues them for human verification.

Design Rationale: - Clinical decisions should never be based on uncertain transcriptions - Medical/legal terminology often has low acoustic confidence - Human verification creates an auditable trail - This pattern enables “glass box” AI (explainable and verifiable)

Configuration: - confidence_threshold: Below this, segment is flagged (default: 0.70) - priority_rules: Customize prioritization logic - max_queue_size: Maximum review queue size before warning

Privacy Note: - The auditor stores transcriptions (text) for review, not raw audio - In production, implement retention policies for the review queue

clear_reviewed_items() → int

Remove reviewed items from the queue.

Returns:

Number of items removed

Return type:

int

get_agent_name() → str

Get the agent name.

get_agent_status() → dict

Get the current status of the auditor.

get_review_queue() → List[ReviewQueueItem]

Get all items in the review queue.

Returns:

Items sorted by priority (highest first)

Return type:

List[ReviewQueueItem]

get_statistics() → dict

Get comprehensive statistics about auditor performance.

Returns:

Statistics including flagging rates, priority distribution, etc.

Return type:

dict

get_unreviewed_items() → List[ReviewQueueItem]

Get only unreviewed items from the queue.

Returns:

Unreviewed items

Return type:

List[ReviewQueueItem]

mark_reviewed(item_id: str, reviewer_notes: str | None = None) → bool

Mark an item as reviewed.

Parameters:

• item_id – The item ID to mark as reviewed

• reviewer_notes – Optional notes from the reviewer

Returns:

True if item was found and marked, False otherwise

Return type:

bool

on_session_end(session_id: str) → None

Finalize processing for a session.

Parameters:

session_id – Session identifier

on_session_start(session_id: str, metadata: dict) → None

Initialize for a new session.

Parameters:

• session_id – Session identifier

• metadata – Session metadata

async on_transcription_update(transcription: TranscriptionResult, context: dict | None = None) → None

Process a transcription and flag if confidence is low.

Parameters:

• transcription – The transcription to audit

• context – Optional context (session_id, patient_id, etc.)

pacing.create_employment_mutation(employed: bool = True) → Mutation

Create a mutation representing employment status change.

>>> mutation = create_employment_mutation(employed=True)
>>> mutation.parameters['status']
'employed'
>>> mutation.parameters['impact_score']
0.6

>>> mutation = create_employment_mutation(employed=False)
>>> mutation.parameters['status']
'unemployed'
>>> mutation.parameters['impact_score']
-0.6

pacing.create_mat_intervention_mutation(medication: str = 'buprenorphine') → Mutation

Create a mutation for starting medication-assisted treatment (MAT).

pacing.create_stable_housing_mutation() → Mutation

Create a mutation representing stable housing.

>>> mutation = create_stable_housing_mutation()
>>> mutation.mutation_type
<MutationType.MODIFY_HOUSING: 'modify_housing'>
>>> mutation.description
'Stable Housing'
>>> mutation.parameters['status']
'stable'