Sentinel-State

Agentic AI pipeline for Medicaid fee schedule automation with LangGraph.

Overview

Sentinel-State is a fully autonomous AI-driven data ingestion pipeline that discovers, qualifies, extracts, and transforms Medicaid fee schedule data from state sources. It uses:

LangGraph: Agent orchestration framework
Multi-Provider LLM Fallback: Configurable primary + auto fallback across configured providers, including optional AWS Bedrock Agent Runtime
AI-Driven Intelligence: Semantic source naming, value interpretation, column mapping, drift detection
True Upsert Semantics: Row-level idempotence for raw data; per-raw-column updates for mappings
Drift Governance: Business analyst validates column mapping changes with configurable review policies

Architecture

Agent Graph (LangGraph)

navigator (URL discovery) 
  ↓
extractor (crawl & qualify sources)
  ↓
business_analyst (drift policy + mapping guidance)
  ↓
analyst (column mapping + confidence routing)
  ↓
human_review / auto_reject (if policy triggers)
  ↓
archivist (DB writes: curated + Bronze/Silver/Gold)

Data Model

Core tables manage state/source/column lifecycle and agent coordination:

Table	Purpose	Key Fields
`state_registry`	Active states to ingest	state_name, state_home_link, is_active
`source_metadata`	Discovered/qualified sources	state_name, source_url, source_table_name, extraction_status
`mapping_column`	Column name transformations (per source URL)	state_name, source_url, raw_column, canonical_column, confidence, approved
`canonical_column_mapping`	Cross-state column naming standards	dataset_type, reference_state, state_name, source_column_name, canonical_column_name
`agent_memory`	Persistent per-agent memory	state_name, agent_id, memory_key, memory_value, confidence
`agent_handoff`	Agent-to-agent messages	state_name, from_agent, to_agent, message_type, acknowledged

Per-Source Data Tables

Each extracted source creates its own physical table per state with clean, approachable naming:

Name pattern: {state_name}_{dataset_type}
Examples: alaska_dmepos, arizona_dmepos, colorado_pharmacy, nevada_physician
Schema: canonical fields only (procedure_code, fee_amount, effective_date, etc.) + state_id, created_at, updated_at
Rows: sourced from raw data after semantic value normalization and LLM-driven column mapping
Isolation: Each state's data isolated via state_id column within the table (composite uniqueness key with other fields)
Column Names: Identical across all instances of same dataset type (e.g., all dmepos tables have same column names)

Cross-State Column Normalization

When multiple states provide the same dataset type (e.g., dmepos), they often use different column names:

Alaska: hcpcs_code vs Arizona: Hcpcs vs Colorado: HCPCS_CODE

Solution (canonical_column_mapping table + business_analyst agent):

LLM-Driven Naming: The business_analyst agent receives cross-state context (how other states named columns for this dataset_type)
Reference Standard: First state loaded sets the naming pattern; subsequent states' columns are matched semantically by the LLM
Semantic Matching: Agent understands Medicaid conventions (e.g., HCPCS_CODE and hcpcs_code map to same semantic field)
Column Name Consistency: All states' tables for same category use identical column names despite separate tables
Confidence Tracking: LLM confidence score per mapping; only approved mappings (above threshold) are used for future states

How It Works:

First state (Alaska) loads dmepos → Analyst maps columns (hcpcs_code, fee_amount, ...) → approves at 95% confidence
Naming stored as reference standard in canonical_column_mapping
Second state (Arizona) processes dmepos → Business analyst receives Alaska's column mappings as context
Arizona's raw columns: Hcpcs, RATE → LLM sees Alaska's hcpcs_code, fee_amount in context
LLM reasons: "Hcpcs is semantic match to hcpcs_code; RATE matches fee_amount based on Medicaid conventions"
Arizona approved mappings use same canonical names → arizona_dmepos table has identical columns to alaska_dmepos
Third state (Colorado) sees both Alaska + Arizona's mappings as context → even more informed semantic decision

Database:

canonical_column_mapping table: (dataset_type, reference_state, state_name, source_column_name, canonical_column_name, confidence)
Enables agent reasoning across all states' data patterns

Result: Separate per-state tables (alaska_dmepos, arizona_dmepos, colorado_dmepos) with identical column names for full compatibility. When multi-state merging is needed later, data is already semantically aligned.

Key Features

1. Multi-Provider LLM Fallback

Seamless provider rotation with auto-retry on failure:

# Configured in .env
LLM_PROVIDER=groq
# Fallback chain is auto-derived from configured providers in this order:
# bedrock_agent -> gemini -> openai -> groq -> ollama -> bedrock

Behavior:

Tries primary provider first
On error, rotates through fallback list
Logs provider + error reason for each attempt
Returns first successful response or None

Parallel Execution:

Business analyst node (drift detection) + Analyst node (column mapping) run simultaneously via ThreadPoolExecutor
Reduces pipeline latency by ~50%

2. Source Discovery & Pre-Ingestion Qualification

URL Filtering (_relevant_links()):

Rejects URLs with negative tokens: contact, faq, form, login, policy, handbook, directory, privacy, etc.
Ranks by file extension: .xlsx (100) > .xls (95) > .csv (90) > .html (50)
Prefers spreadsheets over web pages

Pre-Ingestion Study (_study_source_before_ingestion()):

Heuristic scoring: URL tokens + column names + sample data patterns
LLM semantic qualification: "Is this a Medicaid fee schedule?" (60% confidence threshold)
Rejection at write boundary: Non-fee-schedules never written to source_metadata

Example: ContactUs page filtered out despite being linked from state site.

3. AI-Driven Semantic Source Naming

Problem: Sources stored with years/versions like:

fee_schedule_physician_fy2026 + fee_schedule_physician_fy2027 → duplicated in metadata
fees_therapies_fy2026 → unclear category
fee_hcpcs_ndc_crosswalk_2024 → mixed naming styles

Solution (_categorize_medicaid_source() + _derive_source_name()):

Semantic Categorization: Classifies source into standard Medicaid service categories:
- physician, dental, optometry, pharmacy, dmepos
- therapy (PT/OT/Speech), transportation, laboratory, radiology
- hospital, mental_health, home_health, hospice, rehabilitation
- crosswalk (code mapping tables)
Two-Stage Classification:
- Stage 1 (Fast): Heuristic keyword matching on URL
- Stage 2 (AI): LLM semantic understanding if heuristics unclear

Clean Table Names: Uses category name directly, removes all years/versions:

Raw URL	Category	Table Name
`fee_schedule_physician_fy2026.xlsx`	physician	`physician`
`fees_dmepos_interim_202601.xls`	dmepos	`dmepos`
`fee_hcpcs_ndc_crosswalk.xlsx`	crosswalk	`crosswalk`
`fee-schedule-dental-2026.csv`	dental	`dental`
`therapies_fy2026.xlsx`	therapy	`therapy`

Metadata Reconciliation: Same category on re-run updates existing source (no duplicate)

Result: Global standard naming, not file-dependent, easy to understand

4. Semantic Value Normalization

Problem: X/Y/Yes/No/True/False columns have inconsistent representations

Solution (_infer_semantic_rules_with_llm()):

Per-column rule inference: LLM learns column semantics from data samples (up to 80 columns)
Boolean token mappings:
- Example: x → false, X. → false, Required → true, Opt → false
- Derived from LLM understanding of data context (not hardcoded)
All-column coverage: Processes every column, not just obvious boolean ones
Strong auth-column override: For auth-like columns, any x marker forced to false for data safety

Example Transformation:

Raw:  procedure_code=99213, auth_required=X, fee_type=Y, notes=Not applicable
      ↓ (semantic rules applied)
      procedure_code=99213, auth_required=false, fee_type=true, notes=Not applicable

Default Rules (fallback if LLM unavailable):

False tokens: x, n, no, false, 0
True tokens: y, yes, true, 1

5. Column Mapping with Drift Detection

Business Analyst Node (business_analyst_node()):

Loads approved mappings from previous runs
Compares current LLM-proposed mappings against approved baseline
Computes drift score (0-100) + critical field remapping flags
Enforces drift policy (DRIFT_POLICY env var)

Drift Policies:

Policy	Force Review Trigger
`strict`	Medium/high drift OR any critical-field remap
`balanced`	High drift OR any critical-field remap (default)
`lenient`	Only high drift + critical-field remap together

Critical Fields (protected by policy):

procedure_code, fee_amount, effective_date

6. Column Mapping Upsert (per raw_column)

Problem: Re-running pipeline on same source created duplicate mapping_column rows

Solution (_save_mapping_details() with ON CONFLICT):

INSERT INTO mapping_column (state_name, source_url, raw_column, canonical_column, ...)
VALUES (...)
ON CONFLICT (state_name, source_url, raw_column)
DO UPDATE SET
    canonical_column = EXCLUDED.canonical_column,
    confidence = EXCLUDED.confidence,
    approved = EXCLUDED.approved,
    updated_at = EXCLUDED.updated_at

Semantics:

Unique key: (state_name, source_url, raw_column)
Insert: New mapping for RAW column → inserts row
Update: Same state/source/raw_column → updates canonical mapping (no duplicate)
Migration-safe: Includes constraint migration for older DBs (3-column key)
Idempotent: Running twice produces same DB state

7. Raw Table Upsert (row-level)

Problem: Re-ingesting source data created duplicate rows

Solution (_replace_raw_source_rows() with row hashing):

-- Insert or update based on row content hash (SHA-256)
INSERT INTO {table} (...) VALUES (...)
ON CONFLICT (row_hash) DO UPDATE SET ...;

-- Clean stale rows (hash not in current batch)
DELETE FROM {table} WHERE row_hash NOT IN (...)

Semantics:

Each row hashed via SHA-256 of canonical payload
Same content → same hash → upserts (no duplicate)
Different content → different hash → new row
Stale-row cleanup prevents orphaned data from partial failures
Idempotent snapshots: re-running produces same data set

Quick Start (WSL)

Clone and move into project directory:

cd medicaid

Create env file from template:

cp .env.example .env

Configure LLM providers in .env:

# Primary provider (required)
LLM_PROVIDER=groq
GROQ_API_KEY=xxx

# Fallback providers (optional but recommended)
GEMINI_API_KEY=xxx
OPENAI_API_KEY=xxx
# ... add others as needed

Build and start infrastructure:

docker compose up --build

This starts:

db: Postgres 16
adminer: DB UI (http://localhost:8080)
worker: Idle Python 3.11 worker

Optional HITL dashboard:

docker compose --profile hitl up --build

Dashboard URL: http://localhost:8501

Run ingestion for Alaska (state_id=1):

docker compose run --rm worker python main.py --state-id 1

Open Adminer DB UI:

URL: http://localhost:8080
System: PostgreSQL
Server: db
Username: postgres
Password: postgres
Database: sentinel_state

Verification & Debugging

Run Commands

Run one state by id:

docker compose run --rm worker python main.py --state-id 1

Run all active states:

docker compose run --rm worker python main.py --all

Run with EventBridge-style payload:

docker compose run --rm worker python main.py --event-json '{"detail":{"state_name":"alaska","state_home_link":"https://...","dataset_category":"physician","run_id":"evt-001"}}'

Verify Ingestion Completed

-- Check source was discovered
SELECT state_name, source_name, source_url, extraction_status
FROM source_metadata
WHERE state_name = 'alaska'
ORDER BY discovered_at DESC;

-- Check column mappings were created
SELECT state_name, raw_column, canonical_column, confidence, approved
FROM mapping_column
WHERE state_name = 'alaska'
ORDER BY updated_at DESC;

-- Check extracted data (curated table)
SELECT COUNT(*) as row_count, created_at
FROM fee_schedule_physician_alaska_1
GROUP BY created_at;

-- Check Gold SCD2 output
SELECT state_name, dataset_type, procedure_code, modifier, fee_amount, effective_date, end_date, is_active
FROM gold_medicaid_rates
WHERE state_name = 'alaska'
ORDER BY ingestion_timestamp DESC
LIMIT 50;

Pre-Push Checklist

Reset volume and rebuild:

docker compose down -v
docker compose up --build -d

Run single-state ingestion:

docker compose run --rm worker python main.py --state-id 1

Verify DB state:

-- Sources discovered and qualified
SELECT COUNT(*) FROM source_metadata WHERE state_name = 'alaska' AND extraction_status NOT IN ('failed', 'rejected');

-- Mappings created
SELECT COUNT(*) FROM mapping_column WHERE state_name = 'alaska' AND approved = true;

-- Data loaded
SELECT COUNT(*) FROM fee_schedule_physician_alaska_1;

Stop and cleanup:

docker compose down

Troubleshooting Logs

View worker logs during run:

docker compose logs -f worker

Common messages:

Source qualified: fee_schedule_physician (confidence=92%) → Pre-ingestion passed
Drift detected: high (80 → physician) force_review=true → Policy triggered review
Saved mapping details: state=alaska, source=fee_schedule_physician, mappings=15 → Upsert succeeded
Saved curated table: fee_schedule_physician_alaska_1 (1523 rows) → Data ingested

Environment Configuration

Required Variables

DATABASE_URL=postgresql://postgres:postgres@db:5432/sentinel_state
LLM_PROVIDER=groq
GROQ_API_KEY=xxx

Optional LLM Providers

GEMINI_API_KEY=xxx
OPENAI_API_KEY=xxx
USE_BEDROCK=false  # Set true + AWS creds to enable Bedrock
USE_BEDROCK_AGENT=false
BEDROCK_AGENT_ID=...
BEDROCK_AGENT_ALIAS_ID=...

Policy Configuration

DRIFT_POLICY=balanced        # strict, balanced, lenient
BLOCK_ON_REVIEW=false        # true=reject, false=route to human_review
ANALYST_CONFIDENCE_THRESHOLD=85  # Auto-approve if >= 85%

Runtime Mode Configuration

RUNTIME_MODE=local           # local | aws | hybrid
PIPELINE_VERSION=sentinel-state-v2

# AWS integrations (used when mode=aws|hybrid)
BRONZE_BUCKET=
SILVER_BUCKET=
CHECKPOINT_TABLE_NAME=
HITL_SNS_TOPIC_ARN=

# Local fallbacks
DATA_LAKE_ROOT=data_lake
LOCAL_CHECKPOINT_DIR=checkpoints

Architecture Decisions

Why Multiple LLM Providers?

Reliability: No single point of failure (auto-fallback on timeout/error)
Cost: Groq is fast + cheap (primary); Gemini/OpenAI for backup
Flexibility: Easy to add/swap providers (Ollama local, Bedrock hosted, Bedrock Agent Runtime)
Parallelization: Multiple async LLM calls without blocking

Why AI-Driven Configuration?

User requested "implement AI not the logic please":

Source naming learned from URL patterns (AI understands domain)
Semantic rules learned from data samples (AI understands context)
Drift policy enforcement by LLM understanding of column semantics (AI decides if remap is critical)
Maintainable: behavior lives in prompts, not hardcoded lists

Why Upsert Instead of Truncate-Recreate?

Idempotent: Re-running produces same state (no data loss on failures)
Audit trail: Updates preserve created_at, track changes via updated_at
Safety: Partial failures don't orphan rows (stale cleanup in upsert transaction)

Why Per-Source Tables Instead of Single Wide Table?

Scalability: Each source grows independently (no huge wide table)
Maintenance: Source schema changes don't impact others
Query performance: Smaller tables + indexed searches
Autonomy: New sources added without schema migration

Development

Run All States

docker compose run --rm worker python main.py --all

Run Specific State with Debug Logging

docker compose run --rm -e DEBUG=true worker python main.py --state-id 1

Reset Database (Fresh Start)

docker compose down -v
docker compose up --build

Connect to Postgres Directly

# From host machine (WSL)
psql -h localhost -U postgres -d sentinel_state  # password: postgres

Inspect Graph Execution

Graph state transitions logged to console. Key state fields:

status: current node state (navigator → extractor → analyst → archivist)
log: timestamped event log per node
column_mappings: raw → canonical mapping dict
standardized_records: approved column-renamed data ready for curated table

AWS Lambda: URL Discovery Handler

Path: lambdas/url_discovery_handler/lambda_function.py

This Lambda is intentionally self-contained for deployment and does not import navigator/extractor helper functions from the main pipeline modules.

What It Does

Reads state_id from event body.
Loads the active state record from state_registry.
Navigates the state home URL and discovers links.
Filters relevant Medicaid fee-schedule URLs using one provider only.
Upserts discovered URLs into source_metadata with extraction_status='discovered'.

Provider Selection Rule (No Fallback)

Provider is selected from LLM_PROVIDER:

LLM_PROVIDER=ollama → uses OLLAMA_MODEL + OLLAMA_BASE_URL
LLM_PROVIDER=bedrock → uses BEDROCK_MODEL + AWS_REGION

No multi-provider fallback is used in this Lambda.

Required Environment Variables

Env Var	Required	Example	Purpose
`DATABASE_URL`	Yes	`postgresql+psycopg2://user:pass@host:5432/db`	PostgreSQL connection
`LLM_PROVIDER`	Yes	`ollama` or `bedrock`	Selects single provider
`OLLAMA_MODEL`	Ollama only	`qwen3.5:0.8b`	Ollama model name
`OLLAMA_BASE_URL`	Ollama only	`http://192.168.1.79:11434`	Ollama endpoint
`BEDROCK_MODEL`	Bedrock only	`anthropic.claude-3-5-sonnet-20241022-v2:0`	Bedrock model id
`AWS_REGION`	Bedrock only	`us-east-1`	Bedrock runtime region

Optional Environment Variables (Timeout Safety)

Env Var	Default	Purpose
`URL_DISCOVERY_MAX_CANDIDATE_LINKS`	`200`	Caps raw discovered links
`URL_DISCOVERY_MAX_LINKS`	`50`	Caps URLs written to DB
`NAVIGATION_HTTP_TIMEOUT_SEC`	`20`	Homepage fetch timeout
`LLM_HTTP_TIMEOUT_SEC`	`45`	LLM HTTP call timeout

Event Format

Supports direct or API Gateway body payload.

{
  "state_id": 1
}

or

{
  "body": "{\"state_id\": 1}"
}

Lambda Handler

Handler: lambda_function.lambda_handler
Runtime: Python 3.11+
Recommended timeout: 600 seconds

Dockerized Local Development & Testing (Separated)

Local Lambda container files:

Run local Lambda + DB:

docker compose -f docker-compose.yml -f docker-compose.lambda.local.yml up --build db lambda-url-discovery-local

Invoke Lambda locally:

curl -s -X POST "http://localhost:9000/2015-03-31/functions/function/invocations" \
  -d '{"body":"{\"state_id\":1}"}'

Dockerized Deployment Image (Separated)

Deployment image file:

lambdas/url_discovery_handler/Dockerfile.deploy

Build deployment image:

docker build -f lambdas/url_discovery_handler/Dockerfile.deploy -t medicaid-url-discovery-lambda:latest .

Tag + push to ECR (example):

docker tag medicaid-url-discovery-lambda:latest <account-id>.dkr.ecr.<region>.amazonaws.com/medicaid-url-discovery-lambda:latest
docker push <account-id>.dkr.ecr.<region>.amazonaws.com/medicaid-url-discovery-lambda:latest

Then update the Lambda function to use the pushed container image URI.

Performance Notes

Cold start: First run builds LLM connections (~5-10s)
Typical run: Single state ingestion ~2-5 minutes (depends on source size)
Parallel LLM calls: Business analyst + Analyst run simultaneously (~50% faster)
Large sources (>100K rows): May trigger chunking logic; set MAX_CHUNK_SIZE in .env

Known Limitations & Future Work

Semantic rule coverage: Very wide sources (>150 columns) may exceed LLM token limits; chunking recommended
Similarity threshold (0.86): May over-match source names; monitor first few runs and adjust if needed
Schema migrations: Manual ALTER TABLE for new canonical fields; consider auto-migration framework
Revert capability: No built-in rollback; use DB backups for point-in-time recovery
Rate limiting: LLM providers have rate limits; add exponential backoff for high-volume ingestion

Support & Debugging

For issues:

Check worker logs: docker compose logs worker | tail -50
Verify DB connectivity: docker compose exec db psql -U postgres -d sentinel_state -c "SELECT 1"
Test LLM provider: Check .env keys and network connectivity
Review mapping_column for upsert issues: Query distinct (state_name, source_url, raw_column) tuples

Name		Name	Last commit message	Last commit date
Documents		Documents
app		app
lambdas/url_discovery_handler		lambdas/url_discovery_handler
new		new
.env.example		.env.example
.gitignore		.gitignore
.python-version		.python-version
Dockerfile		Dockerfile
README.md		README.md
agents.py		agents.py
database.py		database.py
docker-compose.lambda.local.yml		docker-compose.lambda.local.yml
docker-compose.yml		docker-compose.yml
extractor_agent_integrated.py		extractor_agent_integrated.py
graph.py		graph.py
hitl_dashboard.py		hitl_dashboard.py
init.sql		init.sql
main.py		main.py
requirements.txt		requirements.txt
test_lambda.http		test_lambda.http

Latest commit

History

Folders and files

Repository files navigation