Skip to main content

Agents

The AI Ingredient Scanner uses four specialized agents, each handling a specific aspect of the analysis pipeline.

Agent Overview​

Supervisor Agent​

File: 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​

RouteConditionAction
researchNo ingredient dataFetch ingredient info
analysisNo report generatedGenerate safety report
criticReport needs validationValidate quality
endComplete or errorFinish workflow

Research Agent​

File: agents/research.py

The Research Agent fetches ingredient safety data using a dual-source strategy.

Architecture​

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:

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​

File: agents/analysis.py

The Analysis Agent generates personalized safety reports using Gemini 2.0 Flash.

Personalization Factors​

FactorImpact
AllergiesProminent warnings for matches
Skin TypeTargeted recommendations
ExpertiseBeginner (simple) / Expert (technical) tone

Report Generation​

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​

File: agents/critic.py

The Critic Agent validates report quality using a comprehensive 5-gate system.

5-Gate Validation​

Gate Details​

GateCheckPass Criteria
CompletenessAll ingredients addressed8/9 ingredients covered
FormatMarkdown table structureValid table with columns
Allergen MatchUser allergies flaggedMatching ingredients warned
ConsistencyRatings match concernsNo contradictions
ToneAppropriate for expertiseMatches beginner/expert

Validation Response​

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​