Multi-Agent Pharmacovigilance Detector

Build a multi-agent AI pipeline to detect drug safety signals from FDA data.

Introduction

30 Second Summary

Every prescription drug on the market is continuously monitored for unexpected side effects. When patterns of adverse reactions emerge in patient reports, pharmaceutical safety teams need to detect and investigate them quickly.

In this project, you will build a multi-agent AI system that queries real FDA adverse event data and detects potential safety signals using statistical methods. The system generates structured assessment reports grounded in medical literature.

What You'll Build

A multi-agent pharmacovigilance signal detection system powered by LangGraph, Groq, ChromaDB, and FastAPI that mirrors what pharmaceutical companies are building for drug safety monitoring.

By the end of this project, you'll have:

  • A LangGraph multi-agent pipeline with 4 specialized agents (Intake, Signal Detection, Evidence, Assessment) coordinated by a supervisor with conditional routing.
  • A RAG pipeline over medical literature using ChromaDB for evidence retrieval and citation grounding.
  • A Dockerized FastAPI service exposing the full signal detection pipeline as a production-ready API.
  • Secret Mission: A human-in-the-loop review step using LangGraph interrupts for safety-critical workflow control.

Are there any prerequisites?

This project assumes familiarity with Python and a basic understanding of LLMs and APIs. Prior experience with LangChain or LangGraph is helpful but not required.

Set Up Your Development Environment

Your pharmacovigilance system relies on several tools working together: LangGraph for agent orchestration, ChromaDB for vector search, Groq for fast LLM inference, and FastAPI for serving the pipeline.

In this step, you will install all the Python packages, create a free Groq account for LLM access, and verify that both the Groq API and the OpenFDA FAERS API respond correctly.

In this step, get ready to:
  • Install Python dependencies (LangGraph, LangChain, ChromaDB, FastAPI, Pydantic, and the Groq integration).
  • Create a Groq account and configure your free API key.
  • Verify the full stack works with test queries against Groq and OpenFDA.
Install Python dependencies

The project uses ten pinned packages that handle everything from agent orchestration to serving an API. You will create a project folder, set up a virtual environment, and install them all at once from a requirements file.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir pharma-signal-detector
cd pharma-signal-detector
  • Create the src directory for your application code by running:
mkdir src
  • Create a Python virtual environment and activate it by running:
python3 -m venv venv
source venv/bin/activate

What does this do?

A virtual environment isolates your project's packages from your system Python. The source venv/bin/activate command activates the environment so that any pip install goes into this project only.

On Windows, use venv\Scripts\activate instead.

  • Create the requirements.txt file inside your pharma-signal-detector folder. Open your code editor and paste the following contents:
langchain==0.3.18
langchain-groq==0.2.4
langchain-core==0.3.34
langgraph==0.2.74
chromadb==1.0.0
fastapi==0.115.0
uvicorn==0.32.0
pydantic==2.10.4
httpx==0.27.0
python-dotenv==1.0.1

What does this file do?

Each line pins a package to an exact version. This ensures everyone building this project gets the same behavior. The key packages are:

  • langgraph and langchain for multi-agent orchestration.
  • langchain-groq for connecting to Groq's fast inference API.
  • chromadb for local vector search (the RAG component).
  • fastapi and uvicorn for serving the pipeline as an API.
  • httpx for making HTTP requests to OpenFDA.
  • Save requirements.txt.
  • Install all packages by running:
pip install -r requirements.txt
  • This may take a minute or two. When it finishes, you should see a line ending with Successfully installed followed by a list of packages.

Install failing or hanging?

  • Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt.
  • If you see a permissions error, confirm you are NOT using sudo. Virtual environments do not need elevated permissions.
  • If a specific package fails to build, make sure you are running Python 3.10 or higher by checking python3 --version.

Help me debug my pip install failure for this pharmacovigilance project.

Create a Groq account and set up your free API key

Groq provides free-tier access to Llama 3.3 70B with 1,000 requests per day. No credit card is required.

  • Open your browser and navigate to console.groq.com.
  • Click Sign Up and create an account using your email or Google OAuth.
  • Once logged in, navigate to the API Keys page at console.groq.com/keys.
  • Click Create API Key.
  • Give it a name (e.g. pharma-signal-detector) and confirm.
  • Copy the key immediately. You will not be able to see it again after closing the dialog.

Why Groq instead of running a local model?

Pharmacovigilance analysis requires a large model that can reliably produce structured JSON output from complex medical text. Groq serves Llama 3.3 70B at extremely fast speeds on their free tier. Local models (3B or 7B parameter) would produce unreliable structured output for this use case.

The code uses a config file that makes swapping to any LangChain-compatible provider (OpenAI, Anthropic) a one-line change later.

Now create the environment files and config module to store your key securely.

  • In your code editor, create a file called .env.example in the root of your pharma-signal-detector folder and paste:
GROQ_API_KEY=your-groq-api-key-here
  • Create a second file called .env (no .example suffix) in the same location. Paste the following, replacing your-groq-api-key-here with the real API key you just copied:
GROQ_API_KEY=your-groq-api-key-here

Why two env files?

The .env.example file is safe to commit to version control. It shows teammates which environment variables are needed without exposing real credentials. Your .env file holds the real key and should never be committed.

  • Create the config module by opening src/config.py in your editor and pasting:
import os
from dotenv import load_dotenv

load_dotenv()

GROQ_API_KEY = os.getenv("GROQ_API_KEY")
GROQ_MODEL = "llama-3.3-70b-versatile"
OPENFDA_BASE_URL = "https://api.fda.gov/drug/event.json"
CHROMA_COLLECTION_NAME = "pubmed_drug_safety"

What does this code do?

  • load_dotenv() reads the .env file and makes its values available as environment variables.
  • GROQ_API_KEY pulls your key from the environment so it never appears in application code.
  • GROQ_MODEL sets the LLM to llama-3.3-70b-versatile, Groq's free-tier model.
  • OPENFDA_BASE_URL points to the public FDA adverse event reporting API.
  • CHROMA_COLLECTION_NAME sets the vector store collection name for later use.
  • Save all three files (.env.example, .env, src/config.py).

Seeing a permissions error on the API Keys page?

If you see a message saying only team owners or developers can create keys, log out and log back in. New accounts sometimes need a session refresh before the role is fully applied.

Help me troubleshoot my Groq API key creation issue.

Verify the full stack with test queries

Before building agents, confirm that both external services respond. You will send a test message to Groq and a test query to the OpenFDA FAERS API.

  • Create a file called test_setup.py in the root of your pharma-signal-detector folder and paste:
# Test Groq connection
from langchain_groq import ChatGroq
from src.config import GROQ_API_KEY, GROQ_MODEL

print("Testing Groq API...")
llm = ChatGroq(api_key=GROQ_API_KEY, model_name=GROQ_MODEL, temperature=0)
response = llm.invoke("Say hello in one sentence.")
print(f"Groq response: {response.content}")
print("Groq OK!\n")

# Test OpenFDA FAERS API
import httpx
from src.config import OPENFDA_BASE_URL

print("Testing OpenFDA FAERS API...")
params = {
    "search": 'patient.drug.medicinalproduct:"metformin"',
    "limit": 1,
}
fda_response = httpx.get(OPENFDA_BASE_URL, params=params)
print(f"OpenFDA status: {fda_response.status_code}")
results = fda_response.json().get("results", [])
print(f"OpenFDA returned {len(results)} event(s)")
print("OpenFDA OK!")
print("\nAll systems ready.")

What does this test do?

  • The Groq test creates a ChatGroq instance using your API key and sends a simple prompt. If it prints a response, your key and network connection work.
  • The OpenFDA test queries real adverse event reports for metformin (a common diabetes drug). The OpenFDA API is free and requires no key.
  • Save test_setup.py and run it with:
python test_setup.py
  • You should see output similar to:
Testing Groq API...
Groq response: Hello, how can I assist you today?
Groq OK!

Testing OpenFDA FAERS API...
OpenFDA status: 200
OpenFDA returned 1 event(s)
OpenFDA OK!

All systems ready.

Test failing?

  • If you see an authentication error from Groq, double-check that your .env file is in the project root (same level as test_setup.py) and contains no extra spaces or quotes around the key.
  • If the OpenFDA request times out, check your internet connection. The OpenFDA API is public and does not require authentication.
  • If you see ModuleNotFoundError, make sure your virtual environment is active ((venv) prefix in your prompt) and that pip install -r requirements.txt completed successfully.

Help me debug my test_setup.py failures.

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files.

ⓧ I'd like to double check the full code

Here is what each file should contain at the end of this step.

requirements.txt

langchain==0.3.18
langchain-groq==0.2.4
langchain-core==0.3.34
langgraph==0.2.74
chromadb==1.0.0
fastapi==0.115.0
uvicorn==0.32.0
pydantic==2.10.4
httpx==0.27.0
python-dotenv==1.0.1

.env.example

GROQ_API_KEY=your-groq-api-key-here

.env (with your real key):

GROQ_API_KEY=your-groq-api-key-here

src/config.py

import os
from dotenv import load_dotenv

load_dotenv()

GROQ_API_KEY = os.getenv("GROQ_API_KEY")
GROQ_MODEL = "llama-3.3-70b-versatile"
OPENFDA_BASE_URL = "https://api.fda.gov/drug/event.json"
CHROMA_COLLECTION_NAME = "pubmed_drug_safety"

Your environment is ready and both external APIs are responding. Next up, you will build the Data Ingestion Agent that queries real FDA adverse event reports and normalizes them into structured data.

Build the Data Ingestion Agent

Your pipeline needs real-world data to analyze. The FDA maintains a public database called FAERS (FDA Adverse Event Reporting System) that contains millions of adverse event reports submitted by healthcare professionals and patients.

In this step, you will define the data models for the entire pipeline, then build an agent that queries the OpenFDA API and transforms raw reports into clean, structured data your other agents can work with.

In this step, get ready to:
  • Define Pydantic data models for the pharmacovigilance pipeline.
  • Create the OpenFDA FAERS API client for adverse event data.
  • Test the intake agent with a real drug query.
Create the OpenFDA FAERS API client that queries adverse event reports by drug name

Before writing any agent logic, you need to define the data shapes that flow through the pipeline. Pydantic models enforce structure and validation on every piece of data your agents produce.

  • Create a new file called src/models.py inside your src/ folder.
  • Add the following code to define the core input types your intake agent will produce:
from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum


class SignalStrength(str, Enum):
    STRONG = "strong"
    MODERATE = "moderate"
    WEAK = "weak"
    NONE = "none"


class AdverseEvent(BaseModel):
    drug_name: str
    reaction: str
    case_count: int
    serious: Optional[int] = None

What does this code do?

SignalStrength is an enum that classifies how concerning a detected signal is. The values range from STRONG (high confidence safety concern) down to NONE (no signal detected).

AdverseEvent represents one drug-reaction pair aggregated from the raw FDA data. It tracks the drug name, the reaction term, how many cases reported it, and whether those cases were serious.

  • Below the AdverseEvent class, add the output models that later agents will use:
class DetectedSignal(BaseModel):
    drug_name: str = Field(description="Name of the drug")
    reaction: str = Field(description="Adverse reaction identified")
    prr: float = Field(description="Proportional Reporting Ratio")
    ror: float = Field(description="Reporting Odds Ratio")
    case_count: int = Field(description="Number of reported cases")
    signal_strength: SignalStrength = Field(description="Assessed signal strength")


class EvidenceItem(BaseModel):
    source: str = Field(description="Source document title or ID")
    relevance_score: float = Field(description="Similarity score from vector search")
    summary: str = Field(description="Key finding from the source")
    supports_signal: bool = Field(description="Whether evidence supports the signal")


class SafetyReport(BaseModel):
    drug_name: str
    query_reaction: str
    signals_detected: list[DetectedSignal]
    evidence: list[EvidenceItem]
    overall_assessment: str
    confidence_score: float = Field(ge=0.0, le=1.0)
    recommendation: str
    sources_cited: list[str]

What do these models represent?

DetectedSignal holds the statistical results from disproportionality analysis. The prr and ror fields store the Proportional Reporting Ratio and Reporting Odds Ratio that measure whether a drug-reaction pair appears more often than expected.

EvidenceItem captures a single piece of supporting literature found by the RAG agent. SafetyReport is the final output that combines signals, evidence, and the LLM's assessment into one structured document.

  • Save src/models.py.

Now you have the complete data schema. Next, you will build the API client that produces AdverseEvent objects from real FDA data.

  • Create the src/agents/ directory by running this command from your project root:
mkdir -p src/agents
  • Confirm the directory exists by running:
ls src/

You should see agents listed alongside config.py and models.py.

  • Create a new file called src/agents/intake.py and add the API client function:
import httpx
from src.config import OPENFDA_BASE_URL
from src.models import AdverseEvent


def fetch_adverse_events(drug_name: str, limit: int = 100) -> list[dict]:
    """Fetch adverse event reports from OpenFDA FAERS API."""
    params = {
        "search": f'patient.drug.medicinalproduct:"{drug_name}"',
        "limit": limit,
    }
    response = httpx.get(OPENFDA_BASE_URL, params=params)
    if response.status_code != 200:
        return []
    data = response.json()
    return data.get("results", [])

What does this code do?

fetch_adverse_events sends a GET request to the OpenFDA FAERS API. The search parameter uses the query syntax patient.drug.medicinalproduct:"drug_name" to find adverse event reports mentioning a specific drug.

If the API returns a non-200 status, the function returns an empty list so the pipeline can continue gracefully. Otherwise it extracts the results array from the JSON response.

Why httpx instead of requests?

The httpx library provides the same developer experience as requests but also supports async calls. Since the FastAPI service you build later is async, using httpx now means you can upgrade to httpx.AsyncClient without rewriting your API logic.

Build the Intake Agent node that normalizes and structures raw FAERS data

Raw FAERS data is messy. Each API response contains nested JSON with patient information, drug lists, and reaction arrays. Your normalization function will aggregate this into clean AdverseEvent objects sorted by frequency.

  • In src/agents/intake.py, add the normalization function below fetch_adverse_events:
def normalize_events(raw_events: list[dict], drug_name: str) -> list[AdverseEvent]:
    """Normalize raw FAERS data into structured AdverseEvent objects."""
    reaction_counts: dict[str, dict] = {}

    for event in raw_events:
        reactions = event.get("patient", {}).get("reaction", [])
        is_serious = 1 if event.get("serious") == 1 else 0

        for reaction in reactions:
            term = reaction.get("reactionmeddrapt", "").lower()
            if term:
                if term not in reaction_counts:
                    reaction_counts[term] = {"count": 0, "serious": 0}
                reaction_counts[term]["count"] += 1
                reaction_counts[term]["serious"] += is_serious

    events = []
    for reaction, data in reaction_counts.items():
        events.append(
            AdverseEvent(
                drug_name=drug_name,
                reaction=reaction,
                case_count=data["count"],
                serious=data["serious"],
            )
        )

    return sorted(events, key=lambda x: x.case_count, reverse=True)

What does this code do?

normalize_events loops through every raw event and extracts each reaction term from the nested patient.reaction array. It uses MedDRA preferred terms (stored in the reactionmeddrapt field) as the canonical reaction name.

The function aggregates counts per reaction and tracks how many cases were flagged as serious. The final list is sorted by case count so the most frequently reported reactions appear first.

  • Below normalize_events, add the LangGraph node function that ties both functions together:
def intake_node(state: dict) -> dict:
    """LangGraph node: Fetch and normalize adverse events."""
    drug_name = state["drug_name"]
    raw_events = fetch_adverse_events(drug_name, limit=100)
    normalized = normalize_events(raw_events, drug_name)
    return {"adverse_events": normalized, "raw_event_count": len(raw_events)}

What is a LangGraph node?

In LangGraph, each agent is a Python function (a "node") that accepts the current pipeline state and returns updates to that state. The intake_node function reads drug_name from state and returns the normalized adverse events plus the raw event count.

Later, the supervisor graph will wire this node to the Signal Detection Agent so data flows automatically through the pipeline.

  • Save src/agents/intake.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved both src/models.py and src/agents/intake.py.

ⓧ I'd like to double check the full code

Here is the complete src/models.py:

from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum


class SignalStrength(str, Enum):
    STRONG = "strong"
    MODERATE = "moderate"
    WEAK = "weak"
    NONE = "none"


class AdverseEvent(BaseModel):
    drug_name: str
    reaction: str
    case_count: int
    serious: Optional[int] = None


class DetectedSignal(BaseModel):
    drug_name: str = Field(description="Name of the drug")
    reaction: str = Field(description="Adverse reaction identified")
    prr: float = Field(description="Proportional Reporting Ratio")
    ror: float = Field(description="Reporting Odds Ratio")
    case_count: int = Field(description="Number of reported cases")
    signal_strength: SignalStrength = Field(description="Assessed signal strength")


class EvidenceItem(BaseModel):
    source: str = Field(description="Source document title or ID")
    relevance_score: float = Field(description="Similarity score from vector search")
    summary: str = Field(description="Key finding from the source")
    supports_signal: bool = Field(description="Whether evidence supports the signal")


class SafetyReport(BaseModel):
    drug_name: str
    query_reaction: str
    signals_detected: list[DetectedSignal]
    evidence: list[EvidenceItem]
    overall_assessment: str
    confidence_score: float = Field(ge=0.0, le=1.0)
    recommendation: str
    sources_cited: list[str]

And here is the complete src/agents/intake.py:

import httpx
from src.config import OPENFDA_BASE_URL
from src.models import AdverseEvent


def fetch_adverse_events(drug_name: str, limit: int = 100) -> list[dict]:
    """Fetch adverse event reports from OpenFDA FAERS API."""
    params = {
        "search": f'patient.drug.medicinalproduct:"{drug_name}"',
        "limit": limit,
    }
    response = httpx.get(OPENFDA_BASE_URL, params=params)
    if response.status_code != 200:
        return []
    data = response.json()
    return data.get("results", [])


def normalize_events(raw_events: list[dict], drug_name: str) -> list[AdverseEvent]:
    """Normalize raw FAERS data into structured AdverseEvent objects."""
    reaction_counts: dict[str, dict] = {}

    for event in raw_events:
        reactions = event.get("patient", {}).get("reaction", [])
        is_serious = 1 if event.get("serious") == 1 else 0

        for reaction in reactions:
            term = reaction.get("reactionmeddrapt", "").lower()
            if term:
                if term not in reaction_counts:
                    reaction_counts[term] = {"count": 0, "serious": 0}
                reaction_counts[term]["count"] += 1
                reaction_counts[term]["serious"] += is_serious

    events = []
    for reaction, data in reaction_counts.items():
        events.append(
            AdverseEvent(
                drug_name=drug_name,
                reaction=reaction,
                case_count=data["count"],
                serious=data["serious"],
            )
        )

    return sorted(events, key=lambda x: x.case_count, reverse=True)


def intake_node(state: dict) -> dict:
    """LangGraph node: Fetch and normalize adverse events."""
    drug_name = state["drug_name"]
    raw_events = fetch_adverse_events(drug_name, limit=100)
    normalized = normalize_events(raw_events, drug_name)
    return {"adverse_events": normalized, "raw_event_count": len(raw_events)}
Test the agent with a real drug query

Time to verify that your intake agent can reach the FDA's servers and return structured data. You will query adverse events for metformin, one of the most widely prescribed diabetes medications.

  • Run the following test command from your project root:

macOS

PYTHONPATH=. python -c "
from src.agents.intake import intake_node
result = intake_node({'drug_name': 'metformin'})
print(f'Raw events fetched: {result[\"raw_event_count\"]}')
print('Top 5 reactions:')
for e in result['adverse_events'][:5]:
    print(f'  {e.reaction}: {e.case_count} cases')
"

Windows

$env:PYTHONPATH = "."
python -c "
from src.agents.intake import intake_node
result = intake_node({'drug_name': 'metformin'})
print(f'Raw events fetched: {result[\"raw_event_count\"]}')
print('Top 5 reactions:')
for e in result['adverse_events'][:5]:
    print(f'  {e.reaction}: {e.case_count} cases')
"

You should see output showing 100 raw events fetched and a list of the top 5 most frequently reported reactions for metformin.

Why metformin for testing?

Metformin is one of the most commonly reported drugs in the FAERS database. Using it for testing guarantees you will get a full set of results back from the API, making it easy to confirm your code works.

Not seeing any output or getting an error?

If you see Raw events fetched: 0, check that your machine has internet access. The OpenFDA API is public and requires no authentication.

If you see a ModuleNotFoundError, make sure you are running the command from the project root folder (the directory containing the src/ folder). The PYTHONPATH=. prefix tells Python to look in the current directory for imports.

help me debug my intake agent test

Your intake agent is pulling real adverse event data from the FDA and structuring it for analysis. Next up, you will build the Signal Detection Agent that applies statistical methods to identify which drug-reaction pairs are appearing more often than expected.

Build the Signal Detection Agent

Your Intake Agent can now pull real adverse event reports from the FDA. The next question is: which of those reactions are statistically unusual enough to warrant investigation?

In this step, you will build the Signal Detection Agent that applies disproportionality analysis to identify drug-reaction pairs that appear more frequently than expected. This is the same statistical approach used by real pharma safety teams and the FDA.

In this step, get ready to:
  • Implement Proportional Reporting Ratio (PRR) and Reporting Odds Ratio (ROR) calculations.
  • Build the Signal Detection Agent that flags reactions occurring more than expected.
  • Test the agent and verify it produces structured signal output using your Pydantic models.
Implement PRR and ROR calculations

Pharmacovigilance signal detection relies on comparing how often a drug-reaction pair is reported versus how often you would expect it if there were no real association. Two standard ratios quantify this: PRR (Proportional Reporting Ratio) and ROR (Reporting Odds Ratio).

What are PRR and ROR?

PRR compares the proportion of a specific reaction for one drug against the proportion of that same reaction for all other drugs. A PRR of 2.0 means the reaction is reported twice as often for this drug compared to the background.

ROR is similar but uses an odds ratio instead of a proportion. Together, PRR and ROR form the foundation of disproportionality analysis. A PRR >= 2 with at least 3 cases is generally considered a signal worth investigating.

  • Create a new file called src/agents/signal_detection.py in your src/agents folder.
  • Add the imports and the calculate_prr function by pasting this code:
from src.models import AdverseEvent, DetectedSignal, SignalStrength


def calculate_prr(drug_reaction_count, drug_total, other_reaction_count, other_total):
    # Guard against division by zero
    if drug_total == 0 or other_total == 0 or other_reaction_count == 0:
        return 0.0
    # Proportion of this reaction for the target drug
    a = drug_reaction_count / drug_total
    # Proportion of this reaction across all other drugs
    b = other_reaction_count / other_total
    if b == 0:
        return 0.0
    # PRR = target proportion / background proportion
    return a / b

What does this code do?

The import line pulls in the Pydantic models you defined earlier. AdverseEvent is the input data structure, DetectedSignal is the output, and SignalStrength is the enum for rating severity.

The calculate_prr function divides the reaction's proportion for the target drug by its proportion across all other drugs. A result above 1.0 means the reaction is over-represented for this drug.

  • Save src/agents/signal_detection.py.
  • You won't see a change yet. The next chunk adds the ROR calculation that complements this function.

Now add the calculate_ror function below calculate_prr. ROR uses an odds ratio approach, which is more robust when baseline rates are very low.

  • Add the calculate_ror function below your calculate_prr function:
def calculate_ror(drug_reaction_count, drug_no_reaction, other_reaction_count, other_no_reaction):
    # Guard against division by zero
    if drug_no_reaction == 0 or other_reaction_count == 0 or other_no_reaction == 0:
        return 0.0
    # Cross-multiply to get the odds ratio
    numerator = drug_reaction_count * other_no_reaction
    denominator = drug_no_reaction * other_reaction_count
    if denominator == 0:
        return 0.0
    return numerator / denominator

What does this code do?

ROR works like an odds ratio in a 2x2 contingency table. It compares the odds of the reaction occurring with the target drug versus the odds of it occurring with all other drugs. The cross-multiplication pattern (a * d / b * c) is the standard formula.

  • Save src/agents/signal_detection.py.
  • You won't see a change yet. The next chunk uses both calculations together to classify signal strength.
Build the signal detection node

With PRR and ROR calculated, you need a function that interprets the numbers. A PRR of 1.2 is noise. A PRR of 5.0 with many cases is a serious safety signal.

  • Add the assess_signal_strength function below calculate_ror:
def assess_signal_strength(prr, ror, case_count):
    # Strong: high PRR with enough cases to be statistically meaningful
    if prr >= 2.0 and ror >= 2.0 and case_count >= 3:
        if prr >= 5.0 and case_count >= 10:
            return SignalStrength.STRONG
        return SignalStrength.MODERATE
    # Weak: slightly elevated but not conclusive
    elif prr >= 1.5 or ror >= 1.5:
        return SignalStrength.WEAK
    return SignalStrength.NONE

What does this code do?

This function maps raw PRR/ROR numbers into the SignalStrength enum you defined earlier. The thresholds follow standard pharmacovigilance criteria.

  • STRONG: PRR >= 5.0 with 10+ cases. This is a clear safety signal.
  • MODERATE: PRR >= 2.0, ROR >= 2.0, and 3+ cases. Worth investigating.
  • WEAK: PRR or ROR >= 1.5 but below the moderate threshold.
  • NONE: No signal detected. This reaction is within expected rates.
  • Save src/agents/signal_detection.py.
  • You won't see a change yet. The next chunk builds the LangGraph node that ties everything together.

Now you need the signal_detection_node function. This is the LangGraph node that receives adverse events from the Intake Agent, runs PRR/ROR on each one, and outputs only the reactions that qualify as signals.

  • Add the signal_detection_node function below assess_signal_strength:
def signal_detection_node(state: dict) -> dict:
    """LangGraph node: Detect disproportionate signals from adverse events."""
    adverse_events: list[AdverseEvent] = state["adverse_events"]
    drug_total = sum(e.case_count for e in adverse_events)
    signals = []
    # Process top 20 reactions by case count
    for event in adverse_events[:20]:
        # Simplified background rates for comparison
        other_reaction = max(1, drug_total // 10)
        other_total = drug_total * 5
        prr = calculate_prr(event.case_count, drug_total, other_reaction, other_total)
        ror = calculate_ror(event.case_count, drug_total - event.case_count, other_reaction, other_total - other_reaction)
        strength = assess_signal_strength(prr, ror, event.case_count)
        # Only keep reactions that register as a signal
        if strength != SignalStrength.NONE:
            signals.append(DetectedSignal(drug_name=event.drug_name, reaction=event.reaction, prr=round(prr, 2), ror=round(ror, 2), case_count=event.case_count, signal_strength=strength))
    return {"detected_signals": signals}

What does this code do?

  • It reads the adverse_events list from the graph state (output of the Intake Agent).
  • It processes only the top 20 reactions (sorted by frequency from intake). This keeps computation fast without missing the most important signals.
  • It uses simplified background rates (other_reaction and other_total) derived from the target drug's total case count. In a production system, these would come from a full database of all drugs.
  • It filters out any reaction rated NONE and returns only genuine signals as DetectedSignal objects.
  • Save src/agents/signal_detection.py.

Seeing a syntax error?

  • Check that each function is separated by two blank lines (Python convention for top-level functions).
  • Verify the import at the top of the file matches exactly: from src.models import AdverseEvent, DetectedSignal, SignalStrength.
  • Make sure the signal_detection_node function's long signals.append() line is all on one line, not accidentally broken across multiple lines.

Help me debug my signal_detection.py file

Test signal detection with structured output

Your Signal Detection Agent is complete. Time to verify it produces properly structured DetectedSignal objects from real FDA data.

  • Test the signal detection pipeline end-to-end by running this command from your project root:
python -c "
from src.agents.intake import intake_node
from src.agents.signal_detection import signal_detection_node

state = intake_node({'drug_name': 'metformin'})
print(f'Adverse events found: {len(state[\"adverse_events\"])}')

result = signal_detection_node(state)
print(f'Signals detected: {len(result[\"detected_signals\"])}')
for s in result['detected_signals'][:5]:
    print(f'  {s.reaction}: PRR={s.prr}, ROR={s.ror}, strength={s.signal_strength.value}')
"

What should I see?

You should see output listing the number of adverse events found and a handful of detected signals. Each signal shows the reaction name, PRR value, ROR value, and signal strength (strong, moderate, or weak).

Metformin typically produces signals for reactions like diarrhoea, nausea, and lactic acidosis. The exact numbers depend on the current FDA database, but you should see at least a few moderate or strong signals.

Seeing zero signals?

  • If adverse_events is empty, the OpenFDA API might be temporarily unavailable. Wait a moment and try again.
  • If you see adverse events but no signals, check that assess_signal_strength uses >= (not >) for the threshold comparisons.
  • Verify you are running the command from the project root folder (where src/ lives).

My signal detection test is returning zero signals

How does DetectedSignal structure the output?

Each signal the agent produces is a validated DetectedSignal Pydantic model containing the drug name, reaction, PRR, ROR, case count, and signal strength. Pydantic ensures every field has the correct type before the data moves to the next agent in the pipeline.

This structured output is what makes multi-agent systems reliable. Downstream agents can trust the shape of the data they receive without writing defensive parsing code.

✔️ Awesome, I've got everything!

Double check you have saved src/agents/signal_detection.py and that your test printed detected signals.

ⓧ I'd like to double check the full code

Here is the complete src/agents/signal_detection.py file:

from src.models import AdverseEvent, DetectedSignal, SignalStrength


def calculate_prr(drug_reaction_count, drug_total, other_reaction_count, other_total):
    if drug_total == 0 or other_total == 0 or other_reaction_count == 0:
        return 0.0
    a = drug_reaction_count / drug_total
    b = other_reaction_count / other_total
    if b == 0:
        return 0.0
    return a / b


def calculate_ror(drug_reaction_count, drug_no_reaction, other_reaction_count, other_no_reaction):
    if drug_no_reaction == 0 or other_reaction_count == 0 or other_no_reaction == 0:
        return 0.0
    numerator = drug_reaction_count * other_no_reaction
    denominator = drug_no_reaction * other_reaction_count
    if denominator == 0:
        return 0.0
    return numerator / denominator


def assess_signal_strength(prr, ror, case_count):
    if prr >= 2.0 and ror >= 2.0 and case_count >= 3:
        if prr >= 5.0 and case_count >= 10:
            return SignalStrength.STRONG
        return SignalStrength.MODERATE
    elif prr >= 1.5 or ror >= 1.5:
        return SignalStrength.WEAK
    return SignalStrength.NONE


def signal_detection_node(state: dict) -> dict:
    adverse_events: list[AdverseEvent] = state["adverse_events"]
    drug_total = sum(e.case_count for e in adverse_events)
    signals = []
    for event in adverse_events[:20]:
        other_reaction = max(1, drug_total // 10)
        other_total = drug_total * 5
        prr = calculate_prr(event.case_count, drug_total, other_reaction, other_total)
        ror = calculate_ror(event.case_count, drug_total - event.case_count, other_reaction, other_total - other_reaction)
        strength = assess_signal_strength(prr, ror, event.case_count)
        if strength != SignalStrength.NONE:
            signals.append(DetectedSignal(drug_name=event.drug_name, reaction=event.reaction, prr=round(prr, 2), ror=round(ror, 2), case_count=event.case_count, signal_strength=strength))
    return {"detected_signals": signals}

Your Signal Detection Agent can now flag statistically unusual drug-reaction pairs from real FDA data. Next up, you will build the Evidence Agent that cross-references these signals against medical literature using a vector database.

Build the Evidence RAG Agent

Your Signal Detection Agent can now identify drug-reaction pairs that appear more frequently than expected. But a statistical signal alone is not enough to make a safety decision.

Real pharmacovigilance teams cross-reference every signal against published medical literature. In this step, you will build a RAG (Retrieval-Augmented Generation) pipeline that searches a ChromaDB vector store of PubMed abstracts to find evidence supporting or contradicting each detected signal.

In this step, get ready to:
  • Set up ChromaDB with sample medical literature and cosine similarity search.
  • Create a data loading pipeline that ingests PubMed abstracts into the vector store.
  • Build the Evidence Agent node that queries the vector store for each detected signal.
Set up ChromaDB and define sample abstracts

ChromaDB is a vector database that stores text as numerical embeddings, making it possible to search by meaning rather than exact keywords. You will populate it with sample PubMed abstracts covering known drug safety topics.

  • Create a new file called evidence.py inside your src/agents/ folder.
  • Add the imports and sample abstract data by pasting this code:
import chromadb
from src.config import CHROMA_COLLECTION_NAME
from src.models import DetectedSignal, EvidenceItem

# Sample PubMed abstracts covering common drug safety topics
SAMPLE_ABSTRACTS = [
    {"id": "pmid_001", "text": "A systematic review of metformin-associated lactic acidosis found that the incidence is approximately 3-10 cases per 100,000 patient-years. Risk factors include renal impairment, hepatic dysfunction, and concurrent use of nephrotoxic agents.", "metadata": {"drug": "metformin", "topic": "lactic acidosis"}},
    {"id": "pmid_002", "text": "Gastrointestinal adverse effects of metformin including nausea, diarrhea, and abdominal pain are reported in up to 25% of patients. Extended-release formulations may reduce GI intolerance.", "metadata": {"drug": "metformin", "topic": "gastrointestinal effects"}},
    {"id": "pmid_003", "text": "Long-term metformin use has been associated with vitamin B12 deficiency in 5-10% of patients. Regular monitoring of B12 levels is recommended for patients on prolonged therapy.", "metadata": {"drug": "metformin", "topic": "vitamin b12 deficiency"}},
    {"id": "pmid_004", "text": "Post-marketing surveillance data suggest that metformin has a favorable cardiovascular safety profile compared to sulfonylureas, with reduced risk of myocardial infarction and all-cause mortality.", "metadata": {"drug": "metformin", "topic": "cardiovascular safety"}},
    {"id": "pmid_005", "text": "Aspirin-associated upper gastrointestinal bleeding remains a significant concern. Meta-analyses estimate the relative risk at 1.5-2.0 for low-dose aspirin users compared to non-users.", "metadata": {"drug": "aspirin", "topic": "gi bleeding"}},
]

What does this code do?

The imports pull in chromadb for vector storage, plus the config constant and Pydantic models you defined earlier.

SAMPLE_ABSTRACTS simulates a corpus of PubMed research papers. Each entry has an ID, the abstract text, and metadata tagging the drug and topic. In a production system, you would replace this with a live PubMed E-utilities integration. For this project, hardcoded abstracts keep the focus on the multi-agent architecture.

Why sample data instead of a live PubMed API?

The core skill pharma companies want to see is the multi-agent RAG architecture, not NCBI API authentication. Sample abstracts let you demonstrate the full pipeline without adding external API complexity. The pattern is identical to what you would use with real data.

  • Save src/agents/evidence.py.

You will not see a visible change yet. The next chunk adds the function that loads these abstracts into ChromaDB.

Create the data loading pipeline

ChromaDB needs a collection to store and query your embeddings. The get_or_create_collection() function initializes an in-memory ChromaDB client, creates the collection with cosine similarity, and loads the sample abstracts if the collection is empty.

  • In src/agents/evidence.py, add this function below the SAMPLE_ABSTRACTS list:
def get_or_create_collection():
    # Initialize an in-memory ChromaDB client
    client = chromadb.Client()
    # Create or retrieve the collection with cosine similarity
    collection = client.get_or_create_collection(name=CHROMA_COLLECTION_NAME, metadata={"hnsw:space": "cosine"})
    # Load sample abstracts only if collection is empty
    if collection.count() == 0:
        collection.add(ids=[a["id"] for a in SAMPLE_ABSTRACTS], documents=[a["text"] for a in SAMPLE_ABSTRACTS], metadatas=[a["metadata"] for a in SAMPLE_ABSTRACTS])
    return collection

What does this code do?

chromadb.Client() creates an in-memory vector store. No server or database file is needed.

get_or_create_collection() retrieves the collection if it already exists, or creates a new one. The metadata={"hnsw:space": "cosine"} configures cosine similarity for nearest-neighbor search.

The collection.count() == 0 check prevents duplicate loading. ChromaDB automatically embeds the text using its default model (all-MiniLM-L6-v2) when you call collection.add(). The model downloads automatically on first use.

  • Save src/agents/evidence.py.

You will not see a visible change yet. The next function uses this collection to search for evidence.

Seeing an import error for chromadb?

Make sure you installed all packages from requirements.txt in Step 1. If chromadb is missing, run pip install chromadb==1.0.0 from your project folder.

help me fix a chromadb import error in my evidence agent

Build the evidence retrieval node

The Evidence Agent needs two pieces: a search function that queries ChromaDB, and a LangGraph node that iterates over detected signals and collects supporting or contradicting literature for each one.

  • In src/agents/evidence.py, add these two functions below get_or_create_collection():
def search_evidence(query: str, n_results: int = 3) -> list[dict]:
    # Get the ChromaDB collection (creates it if needed)
    collection = get_or_create_collection()
    # Query by semantic similarity, returning docs, metadata, and distances
    return collection.query(query_texts=[query], n_results=n_results, include=["documents", "metadatas", "distances"])


def evidence_node(state: dict) -> dict:
    """LangGraph node: Retrieve evidence for detected signals."""
    signals: list[DetectedSignal] = state["detected_signals"]
    all_evidence: list[EvidenceItem] = []
    # Search for evidence on the top 5 signals
    for signal in signals[:5]:
        query = f"{signal.drug_name} {signal.reaction} adverse event safety"
        results = search_evidence(query, n_results=2)
        if results["documents"] and results["documents"][0]:
            for i, doc in enumerate(results["documents"][0]):
                # Convert distance to relevance (lower distance = higher relevance)
                distance = results["distances"][0][i] if results["distances"] else 1.0
                relevance = max(0.0, 1.0 - distance)
                metadata = results["metadatas"][0][i] if results["metadatas"] else {}
                all_evidence.append(EvidenceItem(source=f"PubMed Abstract ({metadata.get('drug', 'unknown')})", relevance_score=round(relevance, 3), summary=doc[:200], supports_signal=relevance > 0.3))
    return {"evidence": all_evidence}

What does this code do?

search_evidence() wraps ChromaDB's query() method. You pass a natural language query and it returns the most semantically similar documents along with their distances.

evidence_node() is the LangGraph node. It loops through the top 5 detected signals, builds a search query from the drug name and reaction, then converts ChromaDB's distance scores into relevance scores (1.0 minus the distance). A relevance score above 0.3 means the evidence supports the signal.

The node returns a dictionary with an evidence key containing a list of EvidenceItem objects. This matches the state schema you will define when wiring the full graph together.

  • Save src/agents/evidence.py.
  • Test the evidence agent by running this command from your project root:
python -c "from src.agents.evidence import evidence_node; from src.models import DetectedSignal, SignalStrength; result = evidence_node({'detected_signals': [DetectedSignal(drug_name='metformin', reaction='lactic acidosis', prr=3.5, ror=4.1, case_count=12, signal_strength=SignalStrength.STRONG)]}); print(f'Found {len(result[\"evidence\"])} evidence items'); [print(f'  - {e.source}: relevance={e.relevance_score}') for e in result['evidence']]"

You should see output showing 2 evidence items retrieved from the vector store, each with a source label and a relevance score.

What should I see?

The output shows evidence items retrieved for the "metformin lactic acidosis" signal. Each item has a source like PubMed Abstract (metformin) and a relevance score between 0 and 1. The first time you run this, ChromaDB downloads the all-MiniLM-L6-v2 embedding model automatically. This may take a few seconds.

Seeing a ModuleNotFoundError?

Make sure you are running the command from your project root folder (where requirements.txt lives). The src.agents.evidence import path requires your terminal to be at the project root.

If you see an error about CHROMA_COLLECTION_NAME, check that your src/config.py file has the CHROMA_COLLECTION_NAME = "pubmed_drug_safety" line from Step 1.

help me debug a ModuleNotFoundError in my pharmacovigilance evidence agent

✔️ Awesome, I've got everything!

Great. Double check you have saved src/agents/evidence.py and move on to the next step.

ⓧ I'd like to double check the full code

import chromadb
from src.config import CHROMA_COLLECTION_NAME
from src.models import DetectedSignal, EvidenceItem

SAMPLE_ABSTRACTS = [
    {"id": "pmid_001", "text": "A systematic review of metformin-associated lactic acidosis found that the incidence is approximately 3-10 cases per 100,000 patient-years. Risk factors include renal impairment, hepatic dysfunction, and concurrent use of nephrotoxic agents.", "metadata": {"drug": "metformin", "topic": "lactic acidosis"}},
    {"id": "pmid_002", "text": "Gastrointestinal adverse effects of metformin including nausea, diarrhea, and abdominal pain are reported in up to 25% of patients. Extended-release formulations may reduce GI intolerance.", "metadata": {"drug": "metformin", "topic": "gastrointestinal effects"}},
    {"id": "pmid_003", "text": "Long-term metformin use has been associated with vitamin B12 deficiency in 5-10% of patients. Regular monitoring of B12 levels is recommended for patients on prolonged therapy.", "metadata": {"drug": "metformin", "topic": "vitamin b12 deficiency"}},
    {"id": "pmid_004", "text": "Post-marketing surveillance data suggest that metformin has a favorable cardiovascular safety profile compared to sulfonylureas, with reduced risk of myocardial infarction and all-cause mortality.", "metadata": {"drug": "metformin", "topic": "cardiovascular safety"}},
    {"id": "pmid_005", "text": "Aspirin-associated upper gastrointestinal bleeding remains a significant concern. Meta-analyses estimate the relative risk at 1.5-2.0 for low-dose aspirin users compared to non-users.", "metadata": {"drug": "aspirin", "topic": "gi bleeding"}},
]


def get_or_create_collection():
    client = chromadb.Client()
    collection = client.get_or_create_collection(name=CHROMA_COLLECTION_NAME, metadata={"hnsw:space": "cosine"})
    if collection.count() == 0:
        collection.add(ids=[a["id"] for a in SAMPLE_ABSTRACTS], documents=[a["text"] for a in SAMPLE_ABSTRACTS], metadatas=[a["metadata"] for a in SAMPLE_ABSTRACTS])
    return collection


def search_evidence(query: str, n_results: int = 3) -> list[dict]:
    collection = get_or_create_collection()
    return collection.query(query_texts=[query], n_results=n_results, include=["documents", "metadatas", "distances"])


def evidence_node(state: dict) -> dict:
    signals: list[DetectedSignal] = state["detected_signals"]
    all_evidence: list[EvidenceItem] = []
    for signal in signals[:5]:
        query = f"{signal.drug_name} {signal.reaction} adverse event safety"
        results = search_evidence(query, n_results=2)
        if results["documents"] and results["documents"][0]:
            for i, doc in enumerate(results["documents"][0]):
                distance = results["distances"][0][i] if results["distances"] else 1.0
                relevance = max(0.0, 1.0 - distance)
                metadata = results["metadatas"][0][i] if results["metadatas"] else {}
                all_evidence.append(EvidenceItem(source=f"PubMed Abstract ({metadata.get('drug', 'unknown')})", relevance_score=round(relevance, 3), summary=doc[:200], supports_signal=relevance > 0.3))
    return {"evidence": all_evidence}

Your Evidence Agent is ready. Next up, you will build the Assessment Agent that synthesizes signals and evidence into a structured safety report, then wire all four agents together into a LangGraph pipeline.

Build the Assessment Agent and Supervisor

You have three specialized agents that each handle one part of the pharmacovigilance workflow. The Intake Agent fetches data, the Signal Detection Agent applies statistics, and the Evidence Agent retrieves literature. Now you need a fourth agent to synthesize everything into a final report.

The Assessment Agent takes the detected signals and supporting evidence, then uses Groq's LLM to produce a structured safety assessment. After building it, you will wire all four agents into a LangGraph StateGraph with conditional routing that acts as the supervisor for the entire pipeline.

In this step, get ready to:
  • Build the Assessment Agent that uses an LLM to generate structured safety reports.
  • Create the LangGraph StateGraph that orchestrates all four agents with conditional routing.
  • Test the full pipeline end-to-end with a real drug query.
Build the Assessment Agent with LLM synthesis

The Assessment Agent is the final analytical node in the pipeline. It receives detected signals and evidence, sends them to an LLM with a carefully crafted prompt, and parses the response into a structured SafetyReport object.

  • Create a new file src/agents/assessment.py in your code editor.
  • Add the imports and prompt template by pasting this code:
from langchain_groq import ChatGroq
from langchain_core.prompts import ChatPromptTemplate
from src.config import GROQ_API_KEY, GROQ_MODEL
from src.models import DetectedSignal, EvidenceItem, SafetyReport

# Define the prompt that instructs the LLM to act as a safety assessor
ASSESSMENT_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a pharmacovigilance safety assessor. Analyze the detected signals and supporting evidence to produce a safety assessment. Be precise, cite sources, and assign a confidence score between 0 and 1.\n\nRespond ONLY with a JSON object matching this schema:\n- overall_assessment: A 2-3 sentence summary of findings\n- confidence_score: Float between 0.0 and 1.0\n- recommendation: One of \"monitor\", \"investigate_further\", \"urgent_review\", \"no_action\""""),
    ("human", """Drug: {drug_name}\nQuery Reaction: {query_reaction}\n\nDetected Signals:\n{signals_text}\n\nSupporting Evidence:\n{evidence_text}\n\nProvide your safety assessment as JSON."""),
])

What does this prompt do?

The system message instructs the LLM to behave as a pharmacovigilance assessor and return structured JSON. The human message template has placeholders for the drug name, reaction, signal data, and evidence that get filled at runtime.

Constraining the output to a specific JSON schema makes parsing reliable. The recommendation field is limited to four specific values so downstream systems can act on it programmatically.

  • Below the prompt template, add the two helper functions that format signal and evidence data for the prompt:
def format_signals(signals):
    """Format detected signals into a readable string for the LLM."""
    lines = [f"- {s.reaction}: PRR={s.prr}, ROR={s.ror}, cases={s.case_count}, strength={s.signal_strength.value}" for s in signals]
    return "\n".join(lines) if lines else "No significant signals detected."


def format_evidence(evidence):
    """Format evidence items into a readable string for the LLM."""
    lines = [f"- [{e.source}] (relevance: {e.relevance_score}): {e.summary[:150]}" for e in evidence]
    return "\n".join(lines) if lines else "No supporting evidence found."

Why format data as text for the LLM?

LLMs work with text, not Python objects. These helpers convert your Pydantic models into clean, readable strings that give the LLM all the context it needs. Truncating evidence summaries to 150 characters prevents the prompt from exceeding token limits when many signals are detected.

  • Now add the main assessment_node function that ties it all together:
def assessment_node(state: dict) -> dict:
    """LangGraph node: Generate safety assessment report using LLM."""
    import json
    drug_name = state["drug_name"]
    query_reaction = state.get("query_reaction", "general")
    signals = state.get("detected_signals", [])
    evidence = state.get("evidence", [])
    # Instantiate the Groq LLM with zero temperature for deterministic output
    llm = ChatGroq(api_key=GROQ_API_KEY, model_name=GROQ_MODEL, temperature=0)
    # Create a chain: prompt template piped into the LLM
    chain = ASSESSMENT_PROMPT | llm
    response = chain.invoke({"drug_name": drug_name, "query_reaction": query_reaction, "signals_text": format_signals(signals), "evidence_text": format_evidence(evidence)})
    # Parse the LLM response as JSON, with a fallback for malformed responses
    try:
        assessment_data = json.loads(response.content)
    except json.JSONDecodeError:
        assessment_data = {"overall_assessment": response.content, "confidence_score": 0.5, "recommendation": "investigate_further"}
    # Build the structured SafetyReport from the LLM output
    report = SafetyReport(drug_name=drug_name, query_reaction=query_reaction, signals_detected=signals, evidence=evidence, overall_assessment=assessment_data.get("overall_assessment", ""), confidence_score=assessment_data.get("confidence_score", 0.5), recommendation=assessment_data.get("recommendation", "investigate_further"), sources_cited=[e.source for e in evidence])
    return {"safety_report": report}

What does this node do?

The assessment_node function follows the same pattern as your other agents. It reads from the shared state, does its work, and returns a dictionary that updates the state.

The ASSESSMENT_PROMPT | llm syntax creates a LangChain "chain" that pipes the formatted prompt directly into the LLM. The try/except block handles cases where the LLM returns malformed JSON by falling back to safe defaults.

  • Save src/agents/assessment.py.

Seeing an import error?

Make sure you saved the file inside the src/agents/ directory. If Python cannot find langchain_groq, confirm your virtual environment is activated and you installed all packages from requirements.txt in Step 1.

help me debug an import error in my assessment agent

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

ⓧ I'd like to double check the full code

from langchain_groq import ChatGroq
from langchain_core.prompts import ChatPromptTemplate
from src.config import GROQ_API_KEY, GROQ_MODEL
from src.models import DetectedSignal, EvidenceItem, SafetyReport

ASSESSMENT_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a pharmacovigilance safety assessor. Analyze the detected signals and supporting evidence to produce a safety assessment. Be precise, cite sources, and assign a confidence score between 0 and 1.\n\nRespond ONLY with a JSON object matching this schema:\n- overall_assessment: A 2-3 sentence summary of findings\n- confidence_score: Float between 0.0 and 1.0\n- recommendation: One of \"monitor\", \"investigate_further\", \"urgent_review\", \"no_action\""""),
    ("human", """Drug: {drug_name}\nQuery Reaction: {query_reaction}\n\nDetected Signals:\n{signals_text}\n\nSupporting Evidence:\n{evidence_text}\n\nProvide your safety assessment as JSON."""),
])


def format_signals(signals):
    lines = [f"- {s.reaction}: PRR={s.prr}, ROR={s.ror}, cases={s.case_count}, strength={s.signal_strength.value}" for s in signals]
    return "\n".join(lines) if lines else "No significant signals detected."


def format_evidence(evidence):
    lines = [f"- [{e.source}] (relevance: {e.relevance_score}): {e.summary[:150]}" for e in evidence]
    return "\n".join(lines) if lines else "No supporting evidence found."


def assessment_node(state: dict) -> dict:
    import json
    drug_name = state["drug_name"]
    query_reaction = state.get("query_reaction", "general")
    signals = state.get("detected_signals", [])
    evidence = state.get("evidence", [])
    llm = ChatGroq(api_key=GROQ_API_KEY, model_name=GROQ_MODEL, temperature=0)
    chain = ASSESSMENT_PROMPT | llm
    response = chain.invoke({"drug_name": drug_name, "query_reaction": query_reaction, "signals_text": format_signals(signals), "evidence_text": format_evidence(evidence)})
    try:
        assessment_data = json.loads(response.content)
    except json.JSONDecodeError:
        assessment_data = {"overall_assessment": response.content, "confidence_score": 0.5, "recommendation": "investigate_further"}
    report = SafetyReport(drug_name=drug_name, query_reaction=query_reaction, signals_detected=signals, evidence=evidence, overall_assessment=assessment_data.get("overall_assessment", ""), confidence_score=assessment_data.get("confidence_score", 0.5), recommendation=assessment_data.get("recommendation", "investigate_further"), sources_cited=[e.source for e in evidence])
    return {"safety_report": report}
Create the LangGraph StateGraph with conditional routing

A StateGraph is how LangGraph models multi-agent workflows. You define a shared state, register each agent as a node, then connect them with edges. Conditional edges let the graph make routing decisions at runtime based on the current state.

  • Create a new file src/graph.py in your code editor.
  • Add the imports and state definition by pasting this code:
import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from src.models import AdverseEvent, DetectedSignal, EvidenceItem, SafetyReport
from src.agents.intake import intake_node
from src.agents.signal_detection import signal_detection_node
from src.agents.evidence import evidence_node
from src.agents.assessment import assessment_node


# Define the shared state that flows between all agents
class PharmacovigilanceState(TypedDict):
    drug_name: str
    query_reaction: str
    adverse_events: list[AdverseEvent]
    raw_event_count: int
    detected_signals: list[DetectedSignal]
    evidence: list[EvidenceItem]
    safety_report: SafetyReport | None

What is this state schema?

The PharmacovigilanceState TypedDict defines every piece of data that flows through the pipeline. Each agent reads what it needs from the state and writes its output back. This is the "shared memory" that all agents communicate through.

LangGraph automatically merges each node's return dictionary into this shared state before passing it to the next node.

  • Below the state definition, add the routing function and graph builder:
def should_continue_to_evidence(state: PharmacovigilanceState) -> str:
    """Route based on whether signals were detected."""
    if state.get("detected_signals"):
        return "evidence"
    return "assessment"


def build_graph() -> StateGraph:
    """Build the pharmacovigilance multi-agent graph."""
    builder = StateGraph(PharmacovigilanceState)
    # Register each agent as a named node
    builder.add_node("intake", intake_node)
    builder.add_node("signal_detection", signal_detection_node)
    builder.add_node("evidence", evidence_node)
    builder.add_node("assessment", assessment_node)
    # Define the execution flow
    builder.add_edge(START, "intake")
    builder.add_edge("intake", "signal_detection")
    # Conditional: skip evidence retrieval if no signals found
    builder.add_conditional_edges("signal_detection", should_continue_to_evidence, ["evidence", "assessment"])
    builder.add_edge("evidence", "assessment")
    builder.add_edge("assessment", END)
    return builder.compile()


# Compile the graph once at module level for reuse
graph = build_graph()

How does the supervisor pattern work here?

The supervisor logic lives in the graph structure itself. The add_conditional_edges call checks whether any signals were detected after the Signal Detection Agent runs. If signals exist, the pipeline routes to the Evidence Agent for literature lookup. If no signals were found, it skips straight to Assessment.

This is the recommended LangGraph pattern for multi-agent coordination. Instead of a separate "supervisor" LLM deciding what to do next, you encode the routing logic in conditional edges that inspect the state.

  • Save src/graph.py.

Seeing a compilation error?

LangGraph validates that all nodes referenced in edges actually exist. If you see an error about missing nodes, check that the import paths at the top of src/graph.py match your file structure. Each agent file must be inside the src/agents/ directory.

help me fix a LangGraph compilation error in my graph.py file

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

ⓧ I'd like to double check the full code

import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from src.models import AdverseEvent, DetectedSignal, EvidenceItem, SafetyReport
from src.agents.intake import intake_node
from src.agents.signal_detection import signal_detection_node
from src.agents.evidence import evidence_node
from src.agents.assessment import assessment_node


class PharmacovigilanceState(TypedDict):
    drug_name: str
    query_reaction: str
    adverse_events: list[AdverseEvent]
    raw_event_count: int
    detected_signals: list[DetectedSignal]
    evidence: list[EvidenceItem]
    safety_report: SafetyReport | None


def should_continue_to_evidence(state: PharmacovigilanceState) -> str:
    if state.get("detected_signals"):
        return "evidence"
    return "assessment"


def build_graph() -> StateGraph:
    builder = StateGraph(PharmacovigilanceState)
    builder.add_node("intake", intake_node)
    builder.add_node("signal_detection", signal_detection_node)
    builder.add_node("evidence", evidence_node)
    builder.add_node("assessment", assessment_node)
    builder.add_edge(START, "intake")
    builder.add_edge("intake", "signal_detection")
    builder.add_conditional_edges("signal_detection", should_continue_to_evidence, ["evidence", "assessment"])
    builder.add_edge("evidence", "assessment")
    builder.add_edge("assessment", END)
    return builder.compile()


graph = build_graph()
Test the full pipeline end-to-end

All four agents are built and wired together. Time to run the complete pipeline with a real drug query to confirm everything works from data ingestion through to the final safety report.

  • Run the end-to-end test from your project root by running this command:
python -c "
from src.graph import graph
result = graph.invoke({
    'drug_name': 'metformin',
    'query_reaction': 'general',
    'adverse_events': [],
    'raw_event_count': 0,
    'detected_signals': [],
    'evidence': [],
    'safety_report': None,
})
print('Signals found:', len(result['detected_signals']))
print('Recommendation:', result['safety_report'].recommendation)
"

What is this test doing?

You invoke the compiled graph with an initial state containing only the drug name. The pipeline then runs each agent in sequence: Intake fetches real adverse events from the OpenFDA API, Signal Detection analyzes them, Evidence retrieves supporting literature from ChromaDB, and Assessment generates the final report.

The empty lists in the initial state get populated as each agent writes its output. You should see a non-zero signal count and a recommendation like monitor or investigate_further.

  • You should see output showing detected signals and a recommendation. The exact numbers depend on current OpenFDA data, but you should see at least a few signals for metformin.

Getting a Groq API error?

Check that your .env file contains a valid GROQ_API_KEY value. The Groq free tier allows 1,000 requests per day. If you are seeing rate limit errors, wait a moment and try again.

help me debug a Groq API error when running my pharmacovigilance pipeline

Your multi-agent pipeline is fully operational. All four agents communicate through the shared state, and the supervisor's conditional routing ensures the Evidence Agent only runs when there are signals to investigate. Next up, you will expose this pipeline as a production-ready API using FastAPI and Docker.

Expose as a FastAPI Service with Docker

Your multi-agent pipeline works end-to-end in Python. However, a production pharmacovigilance system needs to be accessible as a service that other applications can call.

In this step, you will wrap your LangGraph pipeline in a FastAPI web service and containerize it with Docker. This gives you a single command to spin up the entire signal detection system as a production-ready API.

In this step, get ready to:
  • Create a FastAPI application with endpoints for signal detection queries.
  • Write a Dockerfile and docker-compose.yml for containerized deployment.
  • Test the full pipeline end-to-end through the API.
Create the FastAPI application

FastAPI is a modern Python web framework built on Pydantic. It automatically validates request and response data against your models, which is ideal for a system where structured safety reports are the output.

  • Create a new file called src/main.py in your src folder.
  • Add the app definition and request/response models by pasting this code:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from src.graph import graph
from src.models import SafetyReport

# Create the FastAPI application instance
app = FastAPI(
    title="Pharmacovigilance Signal Detector",
    description="Multi-agent system for detecting drug safety signals from FDA FAERS data",
    version="1.0.0",
)


# Define the shape of incoming requests
class SignalRequest(BaseModel):
    drug_name: str
    reaction: str = "general"


# Define the shape of API responses
class SignalResponse(BaseModel):
    report: SafetyReport
    raw_event_count: int
    signals_found: int

What does this code do?

The FastAPI() call creates your web server with metadata that appears in the auto-generated docs.

SignalRequest defines what the caller sends (a drug name and an optional reaction filter). SignalResponse defines what comes back (the full safety report plus summary counts). FastAPI uses these Pydantic models to validate data at the boundary automatically.

  • Below the SignalResponse class, add the two API endpoints:
@app.post("/detect", response_model=SignalResponse)
async def detect_signals(request: SignalRequest):
    """Run the pharmacovigilance signal detection pipeline."""
    try:
        # Invoke the full LangGraph pipeline with initial state
        result = graph.invoke(
            {
                "drug_name": request.drug_name,
                "query_reaction": request.reaction,
                "adverse_events": [],
                "raw_event_count": 0,
                "detected_signals": [],
                "evidence": [],
                "safety_report": None,
            }
        )

        if not result.get("safety_report"):
            raise HTTPException(
                status_code=500, detail="Pipeline failed to generate report"
            )

        return SignalResponse(
            report=result["safety_report"],
            raw_event_count=result.get("raw_event_count", 0),
            signals_found=len(result.get("detected_signals", [])),
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "pharmacovigilance-signal-detector"}

What does this code do?

  • POST /detect is the main endpoint. It takes a SignalRequest, invokes the full LangGraph pipeline with an initialized state dictionary, and returns the structured safety report.
  • GET /health is a lightweight health check that confirms the service is running. Production container orchestrators (like Kubernetes) use this pattern to monitor service availability.
  • The try/except block catches any pipeline failure and returns a clean HTTP 500 error with details, rather than crashing the server.
  • Save src/main.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/main.py with both the app definition and the two endpoints.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from src.graph import graph
from src.models import SafetyReport

app = FastAPI(
    title="Pharmacovigilance Signal Detector",
    description="Multi-agent system for detecting drug safety signals from FDA FAERS data",
    version="1.0.0",
)


class SignalRequest(BaseModel):
    drug_name: str
    reaction: str = "general"


class SignalResponse(BaseModel):
    report: SafetyReport
    raw_event_count: int
    signals_found: int


@app.post("/detect", response_model=SignalResponse)
async def detect_signals(request: SignalRequest):
    """Run the pharmacovigilance signal detection pipeline."""
    try:
        result = graph.invoke(
            {
                "drug_name": request.drug_name,
                "query_reaction": request.reaction,
                "adverse_events": [],
                "raw_event_count": 0,
                "detected_signals": [],
                "evidence": [],
                "safety_report": None,
            }
        )

        if not result.get("safety_report"):
            raise HTTPException(
                status_code=500, detail="Pipeline failed to generate report"
            )

        return SignalResponse(
            report=result["safety_report"],
            raw_event_count=result.get("raw_event_count", 0),
            signals_found=len(result.get("detected_signals", [])),
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "pharmacovigilance-signal-detector"}
Write the Dockerfile and docker-compose.yml

Docker packages your application and all its dependencies into a single image that runs identically on any machine. This is how pharma companies deploy AI services to production.

  • Create a file called Dockerfile in the root of your project folder (the same level as requirements.txt).
  • Add the following container definition:
FROM python:3.11-slim

WORKDIR /app

# Install Python dependencies first (layer caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy the rest of the application
COPY . .

EXPOSE 8000

# Start the FastAPI server
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

What does this Dockerfile do?

  • FROM python:3.11-slim starts from a lightweight Python image.
  • COPY requirements.txt and RUN pip install are separated from the main COPY . . so Docker can cache the dependency layer. This means rebuilds are fast when only your code changes.
  • CMD launches uvicorn (the ASGI server) pointing at your FastAPI app object.
  • Save the Dockerfile.
  • Create another file called docker-compose.yml in the same root folder.
  • Paste the following service definition:
version: "3.8"

services:
  pharma-signal-detector:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    environment:
      - PYTHONPATH=/app
    volumes:
      - ./data:/app/data

What does this compose file do?

  • build: . tells Docker to build the image from the Dockerfile in the current directory.
  • ports: "8000:8000" maps port 8000 inside the container to port 8000 on your machine.
  • env_file: .env loads your Groq API key into the container without baking it into the image.
  • PYTHONPATH=/app ensures Python can find the src package from the container's working directory.
  • Save docker-compose.yml.
  • Confirm both files exist by running:
ls Dockerfile docker-compose.yml

You should see both filenames listed in the output.

Don't see both files?

Make sure you created the files in the project root (the same folder that contains requirements.txt and the src/ folder). If you accidentally placed them inside src/, move them up one level.

help me find where my Dockerfile and docker-compose.yml should live in this project

Test the full pipeline through the API

Everything is wired together. Time to build the Docker image, start the container, and send a real request through the API.

  • Build and start the container by running:
docker compose up --build

What does this command do?

docker compose up --build builds your Docker image from the Dockerfile and then starts the container. The --build flag forces a fresh image build so any code changes are included.

The first build takes a few minutes because it downloads the base Python image and installs all packages. Subsequent builds use cached layers and are much faster.

  • Wait for the container to finish starting. You should see output ending with a line like Uvicorn running on http://0.0.0.0:8000.
  • Open a new terminal window (keep the Docker container running in the first one).
  • Test the health endpoint by running:
curl http://localhost:8000/health

You should see a JSON response confirming the service is alive.

  • Now test the full signal detection pipeline by running:
curl -X POST http://localhost:8000/detect \
  -H "Content-Type: application/json" \
  -d '{"drug_name": "metformin"}'

What should I see?

After 10-30 seconds (the pipeline calls the OpenFDA API and Groq), you should receive a large JSON response containing a SafetyReport with detected signals, evidence items, an overall assessment, a confidence score, and a recommendation.

The signals_found field in the response tells you how many drug-reaction pairs exceeded the PRR/ROR thresholds. The raw_event_count shows how many FAERS reports were ingested.

Seeing a connection error?

  • Make sure the Docker container is still running in your first terminal. If it crashed, check the logs for import errors or missing environment variables.
  • Verify your .env file contains a valid GROQ_API_KEY value. Docker loads this file at container start.
  • If you see a ModuleNotFoundError, confirm that PYTHONPATH=/app is set in your docker-compose.yml under environment.

help me debug my Docker container for the pharmacovigilance API

Your pharmacovigilance system is now running as a production-ready API inside Docker. Any application can send a drug name to your service and receive a structured safety signal report backed by real FDA data and LLM-powered analysis.

Secret mission

Add Human-in-the-Loop Signal Review

In real pharma companies, no safety signal goes directly into a report without human oversight. LangGraph supports interrupt-based workflows that pause execution and wait for a reviewer's decision. Your mission: add a human review gate between signal detection and the assessment report.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally with Docker and uses free-tier APIs. There are no ongoing costs to worry about.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker container and image for the pharmacovigilance API service.
  • Project folder with all source code, configuration files, and the .env file containing your API key.

Keep everything running

No action needed. Choose this if you're still actively building or plan to extend the project with the Secret Mission.

  • Your Docker container continues serving the API on port 8000.
  • Your Groq free tier resets daily (1,000 requests per day), so there is no accumulating cost.
  • ChromaDB runs in-memory inside the container, so no persistent storage builds up on disk.
  • Your Groq account remains active at no charge. No credit card is attached.

Pause - I'll come back to this later

Stop the running Docker container to free up memory and CPU. Your project files stay on disk so you can restart later.

  • Stop the Docker container by running this command from your project folder:
docker compose down
  • This stops and removes the container but keeps the image cached locally for a fast restart.
  • To start again later, run:
docker compose up --build

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below.

  • Stop the container and remove the Docker image by running:
docker compose down --rmi all -v
  • Delete the project folder. On macOS or Linux, run:
cd ~/Desktop
rm -rf pharma-signal-detector
  • On Windows, run:
cd ~\Desktop
rmdir /s /q pharma-signal-detector

Nice Work!

Nice Work!

Nice work. You just built a production-grade multi-agent pharmacovigilance signal detection system using real FDA data.

You've learned how to:

  • Build a 4-agent LangGraph pipeline with specialized agents for intake, signal detection, evidence retrieval, and assessment.
  • Apply real pharmacovigilance statistical methods (PRR and ROR) to FDA adverse event data from the OpenFDA FAERS API.
  • Deploy a RAG-powered assessment system as a Dockerized FastAPI service with structured Pydantic outputs and source citations.
  • Secret Mission: Add human-in-the-loop review using LangGraph interrupts for safety-critical workflow control.

Ready to quiz yourself?