Pharma Supply Intelligence Platform

AI-native platform that monitors pharma supply chains using free public APIs, detects anomalies via multi-modal signals, and validates them through a LangChain Deep Agent with hosted MCP tools.

Status: Phase 3 complete — ingestion → anomaly detection → Deep Agent investigation running end-to-end.

Current Architecture

flowchart TB
    subgraph APIs["Free Public APIs (no auth)"]
        FDA1[openFDA Recalls]
        FDA2[openFDA Shortages]
        FDA3[openFDA NDC Directory]
        FDA4[openFDA Labels]
        RXN[RxNorm]
        DM[DailyMed]
        CT[ClinicalTrials.gov]
    end

    subgraph Ingestion["Ingestion Layer"]
        AIR[Airflow DAG / Scripts]
        TF[Transformers<br/>recall / shortage / ndc / label]
        ER[Entity Resolution<br/>NDC → RxCUI]
    end

    subgraph Storage["Storage Layer"]
        N4J[(Neo4j AuraDB<br/>Knowledge Graph<br/>Drug / Mfr / Recall / Shortage / Anomaly)]
        QDR[(Qdrant Cloud<br/>Vector Store<br/>Cohere embed-english-v3.0<br/>1024-d)]
        GOV[(SQLite<br/>Governance SoR<br/>anomaly_event / evidence / agent_run)]
    end

    subgraph Detection["Anomaly Detection"]
        RULES[Business Rules<br/>sole_source / class_i / coupling]
        STAT[Statistical<br/>IsolationForest / z-score]
        GRAPH[Graph<br/>Cypher degree / paths]
        SEM[Semantic<br/>Cohere + Qdrant kNN]
        FUSION[Late Fusion<br/>0.40 / 0.25 / 0.35 weights<br/>Support Gate ≥2]
    end

    subgraph Agent["Investigation Agent"]
        LLM[Groq LLM<br/>llama-3.3-70b-versatile]
        MCP[MCP Tools<br/>neo4j / openfda / qdrant / rxnorm / governance]
        SKILLS[Skills<br/>supply_risk / manufacturing_recall / evidence_fusion]
        LANGRAPH[LangGraph<br/>React Agent]
    end

    APIs --> Ingestion
    Ingestion --> Storage
    Storage --> Detection
    Detection -->|CandidateAnomalyEvent| Agent
    Agent -->|validated_system / rejected_system / needs_human| GOV
    Agent -->|update status| N4J

End-to-End Flow

sequenceDiagram
    participant APIs as Free APIs
    participant Ingest as Ingestion
    participant Neo4j as Neo4j Aura
    participant Qdrant as Qdrant Cloud
    participant Detect as Anomaly Detectors
    participant Gov as Governance (SQLite)
    participant Agent as Deep Agent (Groq)

    APIs->>Ingest: Poll recalls, shortages, NDC, labels
    Ingest->>Neo4j: Upsert Drug / Mfr / Recall / Shortage nodes + edges
    Ingest->>Qdrant: Embed + upsert supply_events (Cohere v3)
    Ingest->>Gov: Log source timestamps

    Detect->>Neo4j: Build DrugFeatures (mfr_count, recall_count, shortage_count)
    Detect->>Neo4j: Cypher graph scores
    Detect->>Qdrant: Semantic novelty (kNN distance)
    Detect->>Detect: Rules (sole_source, class_ii, coupling)
    Detect->>Detect: IForest tabular anomaly
    Detect->>Detect: Fuse 0.40/0.25/0.35 + support gate → CandidateAnomalyEvent
    Detect->>Gov: Persist hypothesis (anomaly_event + evidence)
    Detect->>Neo4j: Create (:Anomaly) nodes + ANOMALY_ABOUT edges

    Agent->>Gov: Load top hypothesis
    Agent->>MCP: neo4j_drug_neighborhood()
    Agent->>MCP: openfda_search_shortages() / openfda_search_recalls()
    Agent->>MCP: qdrant_semantic_search()
    Agent->>MCP: neo4j_manufacturer_profile()
    Agent->>MCP: rxnorm_lookup()
    Agent->>Agent: Apply skills + multi-source evidence → conclusion
    Agent->>Gov: Persist agent_run (input, output, tool_trace)
    Agent->>Neo4j: Update Anomaly status (validated_system/rejected/needs_human)

Examples from Latest Run

Drug	Type	Fused Score	Status	Key Signal
Spiriva Respimat	M1+S1+T1	0.855	hypothesis	sole_source + open shortage + IForest
EDARBI	M1+R1+T1	0.855	hypothesis	sole_source + Class II recall
Stool Softener Laxative	M1+S1+T1	—	validated_system (0.9)	confirmed shortage, no linked recall
ENOXAPARIN SODIUM	R2	—	needs_human (0.3)	tool call schema mismatch

Future Architecture (Phase 4)

Event-Driven Ingestion with AWS SNS/SQS + DLQ

flowchart TB
    subgraph Sources["Data Sources"]
        FDA_R[openFDA Recalls<br/>webhook / poll]
        FDA_S[openFDA Shortages]
        FDA_N[openFDA NDC]
        RXN[RxNorm]
        DM[DailyMed]
        CT[ClinicalTrials.gov]
    end

    subgraph Ingest_Service["Ingestion Service<br/>(AWS Lambda / ECS Task)"]
        ING_1[Poll API<br/>Scheduled EventBridge]
        ING_2[Transform + Normalize]
        ING_3[Publish to SNS]
    end

    subgraph SNS["AWS SNS Topic<br/>pharma-ingestion-events"]
        SNST[(SNS Topic)]
    end

    subgraph SQS["AWS SQS Queues"]
        SQSR[Queue: Recalls<br/>DLQ: Recalls-DLQ]
        SQSS[Queue: Shortages<br/>DLQ: Shortages-DLQ]
        SQSN[Queue: NDC<br/>DLQ: NDC-DLQ]
        SQSL[Queue: Labels<br/>DLQ: Labels-DLQ]
    end

    subgraph Workers["Worker Services<br/>(AWS ECS Fargate / Lambda)"]
        W_R[Recall Worker]
        W_S[Shortage Worker]
        W_N[NDC Worker]
        W_L[Label Worker]
    end

    subgraph DLQ["Dead Letter Queues<br/>(after 3 retries)"]
        D_R[Recalls-DLQ]
        D_S[Shortages-DLQ]
        D_N[NDC-DLQ]
        D_L[Labels-DLQ]
    end

    subgraph Storage["Persistent Storage"]
        N4J[(Neo4j AuraDB)]
        QDR[(Qdrant Cloud)]
        GOV[(PostgreSQL / Aurora<br/>Governance SoR)]
    end

    subgraph Monitoring["Monitoring & Alerting"]
        CW[CloudWatch<br/>Metrics + Alarms]
        SL[DLQ Alarm → SNS<br/>(email / PagerDuty)]
        TR[AWS X-Ray<br/>Trace ingestion failures]
    end

    Sources --> Ingest_Service
    Ingest_Service --> SNST
    SNST --> SQSR & SQSS & SQSN & SQSL
    SQSR --> W_R
    SQSS --> W_S
    SQSN --> W_N
    SQSL --> W_L
    W_R -->|fail 3x| D_R
    W_S -->|fail 3x| D_S
    W_N -->|fail 3x| D_N
    W_L -->|fail 3x| D_L
    W_R & W_S & W_N & W_L --> N4J
    W_R & W_S & W_N & W_L --> QDR
    W_R & W_S & W_N & W_L --> GOV
    D_R & D_S & D_N & D_L -.->|alarm| CW
    CW -.-> SL

Kubernetes Deployment (EKS)

flowchart TB
    subgraph EKS["Amazon EKS Cluster"]
        subgraph Ingress["Ingress"]
            ALB[AWS ALB<br/>Ingress Controller]
            KG[Keycloak Gatekeeper<br/>OIDC Proxy]
        end

        subgraph Services["Core Services"]
            API[Pharma API<br/>3 replicas]
            MCP_N[MCP: Neo4j Tools<br/>2 replicas]
            MCP_O[MCP: openFDA Tools<br/>2 replicas]
            MCP_Q[MCP: Qdrant Tools<br/>2 replicas]
            MCP_G[MCP: Governance<br/>2 replicas]
            MCP_R[MCP: RxNorm<br/>2 replicas]
        end

        subgraph Detectors["Detector Pods"]
            D_RULE[Rules Detector<br/>CronJob]
            D_STAT[Statistical Detector<br/>CronJob]
            D_GRAPH[Graph Detector<br/>CronJob]
            D_SEM[Semantic Detector<br/>CronJob]
        end

        subgraph Agent_Pods["Agent Pods"]
            A_SUP[Supervisor Agent<br/>Deployment]
            A_S[Subagent: Supply<br/>Sidecar]
            A_M[Subagent: Mfg<br/>Sidecar]
            A_F[Subagent: Fusion<br/>Sidecar]
        end

        subgraph Infra["Infrastructure"]
            HPA[HPA<br/>CPU/Memory]
            PD[Pod Disruption Budget]
            PDB[PostgreSQL / Aurora<br/>Governance]
        end
    end

    subgraph Auth["Auth Layer"]
        KC[Keycloak Cluster<br/>OAuth2 / OIDC]
        USR[User Realm<br/>pharma-admins]
        SVC[Service Account Realm<br/>mcp-services / agent-services]
    end

    Internet --> ALB
    ALB --> KG
    KG -->|JWT validation| API
    API --> KC
    MCP_N & MCP_O & MCP_Q & MCP_G & MCP_R --> KC
    A_SUP --> KC
    KG -->|403| Internet

Keycloak Auth Flow for MCPs

sequenceDiagram
    participant Agent as Deep Agent
    participant KC as Keycloak
    participant MCP as MCP Service
    participant API as External API

    Agent->>KC: Request token (client_credentials)<br/>scope: mcp:neo4j mcp:openfda
    KC-->>Agent: JWT access_token<br/>(with audience + roles)

    Agent->>MCP: tool_call()<br/>Authorization: Bearer JWT
    MCP->>KC: Verify token (JWKS)<br/>check audience + roles
    KC-->>MCP: Token valid

    MCP->>API: Execute (with sub-claims for audit)
    API-->>MCP: Response
    MCP-->>Agent: Tool result

    Note over KC: Token expiry → Agent refreshes via refresh_token

Service Mesh & Observability

flowchart LR
    subgraph Mesh["AWS App Mesh / Istio"]
        GW[Gateway]
        VS[Virtual Services<br/>per MCP domain]
        VR[Virtual Routers<br/>traffic split / canary]
    end

    subgraph Observability
        OTEL[OpenTelemetry<br/>Collector Sidecar]
        JAEG[Jaeger / Tempo<br/>Distributed Tracing]
        PROM[Prometheus<br/>Metrics]
        GRAF[Grafana<br/>Dashboards]
        LOKI[Loki<br/>Log Aggregation]
    end

    subgraph Alerting
        ALRT[Alertmanager]
        PD[PagerDuty]
        SLK[Slack]
    end

    Mesh --> OTEL
    OTEL --> JAEG & PROM & LOKI
    PROM --> ALRT
    ALRT --> PD & SLK
    PROM --> GRAF
    LOKI --> GRAF

Project Structure

pharma-supply-intel/
├── src/
│   ├── common/               Config + shared models
│   ├── ingestion/
│   │   ├── apis/             7 free API clients
│   │   └── transforms/       Entity transformers
│   ├── storage/
│   │   ├── neo4j/            Graph DB client + schema
│   │   └── qdrant/           Vector DB client + Cohere embeddings
│   ├── anomaly_detection/
│   │   ├── detectors/        rules / statistical / graph / semantic
│   │   ├── features.py       Feature matrix builder
│   │   ├── fusion.py         Weighted late fusion + support gate
│   │   ├── models.py         Pydantic schemas
│   │   └── pipeline.py       Orchestrator
│   ├── governance/           SQLite SoR + Neo4j anomaly persistence
│   ├── agents/               LangGraph investigation agent + skills
│   └── mcp_server/           FastMCP server (10 tools)
├── dags/                     Airflow DAG definition
├── scripts/                  Standalone pipeline scripts
├── docs/                     HLD, LLD, anomaly detection design
├── docker-compose.yml        Local infra (Neo4j, Qdrant, Postgres)
├── requirements.txt          Python dependencies
└── pyproject.toml            Package config

Quick Start

# Install
pip install -e .

# Configure
cp .env.example .env        # Fill in your API keys

# Ingest FDA data → Neo4j Aura
python scripts/ingest_and_query_neo4j.py

# Ingest + embed → Qdrant Cloud
python scripts/ingest_and_query_qdrant.py

# Run anomaly detection → hypotheses
python scripts/run_anomaly_detection.py

# Investigate top hypothesis with Deep Agent
python scripts/run_investigation.py --mode react --limit 2

# Verify all APIs are reachable
python scripts/verify_apis.py

# Run full demo
python scripts/demo_pipeline.py

Environment

Variable	Required	Default	Description
`NEO4J_URI`	Yes	`bolt://localhost:7687`	Neo4j Aura connection string
`NEO4J_USER` / `NEO4J_USERNAME`	Yes	`neo4j`	Neo4j username
`NEO4J_PASSWORD`	Yes	`pharma_dev`	Neo4j password
`QDRANT_URL`	Yes	`http://localhost:6333`	Qdrant Cloud URL
`QDRANT_API_KEY`	Yes	—	Qdrant Cloud API key
`COHERE_API_KEY`	Yes	—	Cohere embed-english-v3.0
`GROQ_API_KEY`	Yes	—	Groq llama-3.3-70b-versatile
`EMBEDDING_MODEL`	No	`embed-english-v3.0`	Cohere embedding model
`GROQ_MODEL`	No	`llama-3.3-70b-versatile`	Groq LLM model

Tech Stack

Component	Current	Future
Orchestration	Airflow / scripts	EventBridge + SNS/SQS
Graph DB	Neo4j AuraDB (cloud)	Neo4j AuraDB
Vector DB	Qdrant Cloud	Qdrant Cloud
Embeddings	Cohere embed-english-v3.0	Cohere
LLM	Groq llama-3.3-70b-versatile	Groq
Agent Framework	LangGraph react / deepagents	LangGraph
MCP Transport	stdio subprocess	FastMCP over HTTP (hosted)
Auth	None (dev)	Keycloak (OAuth2 / OIDC)
Compute	Local / script	EKS Fargate + Lambda
Monitoring	stdout	OpenTelemetry + Grafana
Governance	SQLite	PostgreSQL / Aurora
Containerization	docker-compose	Kubernetes (EKS)

Anomaly Detection

Multi-signal late fusion with support gate:

Detector	Weight	Signal
Tabular (rules + IForest)	0.40	manufacturer concentration, recall severity, shortage coupling
Semantic (Cohere + Qdrant)	0.25	Semantic novelty vs historical events
Graph (Neo4j Cypher)	0.35	Degree anomalies, path risk, sole-source
Fused	—	Hypothesis emitted if support ≥ 2 or rule hit + high score

Status levels: hypothesis → validated_system → needs_human → confirmed

License

MIT

Navigation Menu

Repository navigation

About

Repository files navigation

Pharma Supply Intelligence Platform

Current Architecture

End-to-End Flow

Examples from Latest Run

Future Architecture (Phase 4)

Event-Driven Ingestion with AWS SNS/SQS + DLQ

Kubernetes Deployment (EKS)

Keycloak Auth Flow for MCPs

Service Mesh & Observability

Project Structure

Quick Start

Environment

Tech Stack

Anomaly Detection

License

About

Releases

Packages

Contributors

Languages

Name		Name	Last commit message	Last commit date
dags		dags
docs		docs
scripts		scripts
src		src
.env.example		.env.example
.gitignore		.gitignore
README.md		README.md
docker-compose.yml		docker-compose.yml
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt

Latest commit

History

Folders and files

Repository files navigation

Pharma Supply Intelligence Platform

Current Architecture

End-to-End Flow

Examples from Latest Run

Future Architecture (Phase 4)

Event-Driven Ingestion with AWS SNS/SQS + DLQ

Kubernetes Deployment (EKS)

Keycloak Auth Flow for MCPs

Service Mesh & Observability

Project Structure

Quick Start

Environment

Tech Stack

Anomaly Detection

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages