Agents

The AI Ingredient Scanner uses four specialized agents, each handling a specific aspect of the analysis pipeline. This document provides implementation details and code examples for each agent.

Agent Overview

┌─────────────────────────────────────────────────────────────────────┐
│                         AGENT PIPELINE                               │
│                                                                      │
│   ┌─────────────┐    ┌─────────────┐    ┌─────────────┐            │
│   │  SUPERVISOR │ → │  RESEARCH   │ → │  ANALYSIS   │            │
│   │   (Router)  │    │  (Lookup)   │    │  (Report)   │            │
│   └─────────────┘    └──────┬──────┘    └──────┬──────┘            │
│                             │                   │                    │
│                      ┌──────┴──────┐           │                    │
│                      ▼             ▼           ▼                    │
│                 ┌─────────┐  ┌──────────┐  ┌─────────────┐         │
│                 │ Qdrant  │  │  Google  │  │   Gemini    │         │
│                 │ Vector  │  │  Search  │  │   2.0 LLM   │         │
│                 └─────────┘  └──────────┘  └─────────────┘         │
│                                                   │                  │
│                                                   ▼                  │
│                                           ┌─────────────┐           │
│                                           │   CRITIC    │           │
│                                           │ (Validate)  │           │
│                                           └──────┬──────┘           │
│                                                  │                   │
│                              ┌───────────────────┼───────────────┐  │
│                              ▼                   ▼               ▼  │
│                         [APPROVED]          [REJECTED]    [ESCALATED]│
│                              │                   │               │  │
│                              ▼                   ▼               ▼  │
│                          Output              Retry           Output │
│                          Report             Analysis        + Warn  │
└─────────────────────────────────────────────────────────────────────┘

Supervisor Agent

agents/supervisor.py

The Supervisor orchestrates the workflow by determining the next step based on current state.

Routing Logic

def route_next(state: WorkflowState) -> RouteType:
    """Determine the next node in the workflow."""

    # Check for errors
    if state.get("error"):
        return NODE_END

    # Step 1: Need research data?
    if not has_research_data(state):
        return NODE_RESEARCH

    # Step 2: Need analysis report?
    if not has_analysis_report(state):
        return NODE_ANALYSIS

    # Step 3: Check validation status
    critic_feedback = state.get("critic_feedback")

    if critic_feedback is None:
        return NODE_CRITIC

    # Step 4: Handle validation results
    if is_approved(state):
        return NODE_END

    if is_escalated(state):
        return NODE_END

    if is_rejected(state):
        return NODE_ANALYSIS  # Retry

    return NODE_END

Route Types

Route	Condition	Action
research	No ingredient data	Fetch ingredient info
analysis	No report generated	Generate safety report
critic	Report needs validation	Validate quality
end	Complete or error	Finish workflow

Research Agent

agents/research.py

The Research Agent fetches ingredient safety data using a dual-source strategy with parallel processing for large ingredient lists.

Dual-Source Strategy

def _research_single_ingredient(ingredient_name: str) -> IngredientData | None:
    # Try vector database first
    result = lookup_ingredient(ingredient_name)

    if result and result["confidence"] >= CONFIDENCE_THRESHOLD:
        return result  # High confidence match

    # Fall back to Google Search
    grounded_result = grounded_ingredient_search(ingredient_name)

    if grounded_result:
        _save_to_qdrant(grounded_result)  # Learn for next time
        return grounded_result

    return None  # Will use unknown record

Parallel Processing

For lists with more than 3 ingredients, research is parallelized across multiple workers:

BATCH_SIZE = 3  # Ingredients per worker

def _research_parallel(ingredients: list[str]) -> list[IngredientData]:
    batches = _create_batches(ingredients, BATCH_SIZE)

    with ThreadPoolExecutor(max_workers=len(batches)) as executor:
        futures = {
            executor.submit(_research_batch, idx, batch): idx
            for idx, batch in enumerate(batches)
        }

        for future in as_completed(futures):
            batch_results = future.result()
            results_by_batch[futures[future]] = batch_results

    return _reassemble_results(results_by_batch)

Output Schema

class IngredientData(TypedDict):
    name: str
    purpose: str
    safety_rating: int        # 1-10 scale
    concerns: str
    recommendation: str       # SAFE/CAUTION/AVOID
    allergy_risk_flag: str    # HIGH/LOW
    origin: str               # Natural/Synthetic
    category: str             # Food/Cosmetics/Both
    regulatory_status: str
    source: str               # qdrant/google_search
    confidence: float

Analysis Agent

agents/analysis.py

The Analysis Agent generates personalized safety reports using Gemini 2.0 Flash, adapting content based on user profile.

Personalization Factors

Factor	Impact on Report
Allergies	Prominent AVOID warnings for matched ingredients
Skin Type	Targeted recommendations for sensitivities
Expertise Level	Beginner (simple) vs Expert (technical) tone

Tone Adaptation

TONE_INSTRUCTIONS = {
    "beginner": """
        Use simple, everyday language.
        Avoid technical jargon.
        Explain concepts clearly for someone new to ingredient analysis.
    """,
    "expert": """
        Use technical terminology when appropriate.
        Include chemical details and concentrations.
        Reference regulatory standards and scientific studies.
    """
}

Risk Calculation

def _parse_llm_overall_risk(llm_analysis: str) -> tuple[RiskLevel, int]:
    """Determine overall risk from analysis."""

    # Rule 1: Any AVOID → HIGH risk
    if has_avoid_recommendation(llm_analysis):
        return RiskLevel.HIGH, avg_rating

    # Rule 2: Any banned ingredient → HIGH risk
    if has_regulatory_ban(llm_analysis):
        return RiskLevel.HIGH, avg_rating

    # Rule 3: Calculate from average
    if avg_rating <= 3:
        return RiskLevel.HIGH, avg_rating
    elif avg_rating <= 6:
        return RiskLevel.MEDIUM, avg_rating
    else:
        return RiskLevel.LOW, avg_rating

Critic Agent

agents/critic.py

The Critic Agent validates report quality using a comprehensive 5-gate validation system with automatic retry logic.

5-Gate Validation System

Completeness

All ingredients addressed (8/9 minimum)

Format

Valid markdown table structure with required columns

Allergen Match

User allergies properly flagged and warned

Consistency

Safety ratings align with stated concerns

Tone

Appropriate for user expertise level

Validation Response Schema

class CriticFeedback(TypedDict):
    result: ValidationResult      # APPROVED/REJECTED/ESCALATED
    completeness_ok: bool
    format_ok: bool
    allergens_ok: bool
    consistency_ok: bool
    tone_ok: bool
    feedback: str
    failed_gates: list[str]

Retry Logic

def validate_report(state: WorkflowState) -> dict:
    # Run validation
    validation_result = _run_multi_gate_validation(report, ...)

    all_gates_passed = all([
        validation_result["completeness_ok"],
        validation_result["format_ok"],
        validation_result["allergens_ok"],
        validation_result["consistency_ok"],
        validation_result["tone_ok"],
    ])

    if all_gates_passed:
        result = ValidationResult.APPROVED
    elif retry_count >= max_retries:
        result = ValidationResult.ESCALATED
    else:
        result = ValidationResult.REJECTED
        new_retry_count = retry_count + 1

Agent Interaction Sequence

Supervisor          Research           Analysis           Critic
    │                   │                  │                  │
    │   route_next()    │                  │                  │
    ├──────────────────>│                  │                  │
    │                   │                  │                  │
    │        Start workflow                │                  │
    │<──────────────────│                  │                  │
    │                   │                  │                  │
    │  ingredient_data[]│                  │                  │
    │<──────────────────│                  │                  │
    │   (parallel)      │                  │                  │
    │                   │                  │                  │
    │   route_next()    │                  │                  │
    ├───────────────────────────────────-->│                  │
    │                                      │                  │
    │              analysis_report         │                  │
    │<─────────────────────────────────────│                  │
    │                                      │                  │
    │   route_next()                       │                  │
    ├─────────────────────────────────────────────────────────>│
    │                                                         │
    │                            [APPROVED / REJECTED]        │
    │<────────────────────────────────────────────────────────│
    │                                                         │
    │   if REJECTED && retries < 2:                          │
    ├───────────────────────────────────-->│  (retry)        │
    │                                      │                  │