Implementation:Vibrantlabsai Ragas Workflow Eval

Knowledge Sources	Vibrantlabsai_Ragas
Domains	LLM Evaluation, Workflow Evaluation, Support Triage, Information Extraction
Last Updated	2026-02-12 00:00 GMT

Overview

This module implements a configurable support triage agent with pluggable extraction strategies (deterministic regex-based and LLM-based), full trace logging, and a three-step email processing workflow for workflow evaluation examples.

Description

The workflow.py module provides a complete multi-step support email triage system designed as a workflow evaluation example for Ragas. The architecture centers on the ConfigurableSupportTriageAgent class, which orchestrates a three-step pipeline:

Step 1: Email Classification -- Uses an OpenAI LLM (gpt-3.5-turbo) to classify incoming emails into one of three categories: "Bug Report", "Billing", or "Feature Request". On LLM failure, it falls back to "Bug Report" as a default.

Step 2: Information Extraction -- Uses a pluggable BaseExtractor to extract category-specific structured information. Two concrete extractors are provided:

• DeterministicExtractor -- Uses regex patterns and keyword matching. For Bug Reports, it extracts product versions (e.g., "2.1.4") and error codes (e.g., "XYZ-123"). For Billing emails, it extracts invoice numbers and dollar amounts. For Feature Requests, it detects urgency level via keyword hierarchy (urgent > high > medium > low), identifies product areas from a predefined list, and attempts to extract the requested feature description.

• LLMExtractor -- Uses OpenAI gpt-3.5-turbo with structured prompts to extract the same fields via natural language understanding. Each category has a dedicated prompt template that instructs the LLM to respond with valid JSON containing the relevant fields.

Step 3: Response Generation -- Uses an OpenAI LLM to generate a professional customer support response template based on the classified category and extracted information. On failure, it returns a generic acknowledgment message.

Trace Logging -- Every step emits TraceEvent dataclass instances that record the event type (e.g., "llm_call", "llm_response", "extraction", "error", "init"), the component responsible, and associated data. After processing, export_traces_to_log serializes all traces along with the email content and result to a timestamped JSON log file in the configured log directory.

The ExtractionMode enum tracks whether the agent uses deterministic or LLM extraction, and the set_extractor method allows runtime switching between extraction strategies.

The default_workflow_client factory function creates a properly configured agent instance, defaulting to deterministic extraction. When using the LLM extractor, it requires the OPENAI_API_KEY environment variable.

Usage

Import this module when you need a configurable support email triage agent for workflow evaluation experiments. Use ConfigurableSupportTriageAgent directly for full control, or default_workflow_client for quick setup. The module is designed to be evaluated with Ragas metrics by comparing agent outputs against expected classifications, extractions, and response quality.

Code Reference

Source Location

• Repository: Vibrantlabsai_Ragas
• File: examples/ragas_examples/workflow_eval/workflow.py

Signature

@dataclass
class TraceEvent:
    event_type: str
    component: str
    data: Dict[str, Any]

class ExtractionMode(Enum):
    DETERMINISTIC = "deterministic"
    LLM = "llm"

class BaseExtractor(ABC):
    @abstractmethod
    def extract(self, email_content: str, category: str) -> Dict[str, Optional[str]]

class DeterministicExtractor(BaseExtractor):
    def extract(self, email_content: str, category: str) -> Dict[str, Optional[str]]

class LLMExtractor(BaseExtractor):
    def __init__(self, client: OpenAI) -> None
    def extract(self, email_content: str, category: str) -> Dict[str, Optional[str]]

class ConfigurableSupportTriageAgent:
    def __init__(self, api_key: str, extractor: Optional[BaseExtractor] = None,
                 logdir: str = "logs") -> None
    def set_extractor(self, extractor: BaseExtractor) -> None
    def classify_email(self, email_content: str) -> str
    def extract_info(self, email_content: str, category: str) -> Dict[str, Optional[str]]
    def generate_response(self, category: str, extracted_info: Dict[str, Any]) -> str
    def export_traces_to_log(self, run_id: str, email_content: str,
                             result: Optional[Dict[str, Any]] = None) -> str
    def process_email(self, email_content: str, run_id: Optional[str] = None) -> Dict[str, Any]

def default_workflow_client(
    extractor_type: Literal["deterministic", "llm"] = "deterministic"
) -> ConfigurableSupportTriageAgent

def main() -> None

Import

from ragas_examples.workflow_eval.workflow import (
    ConfigurableSupportTriageAgent,
    DeterministicExtractor,
    LLMExtractor,
    TraceEvent,
    ExtractionMode,
    default_workflow_client,
)

I/O Contract

Inputs

ConfigurableSupportTriageAgent.__init__

Name	Type	Required	Description
api_key	str	Yes	OpenAI API key for LLM calls
extractor	Optional[BaseExtractor]	No	Pluggable extractor instance; defaults to DeterministicExtractor
logdir	str	No	Directory for trace log files (default: "logs")

ConfigurableSupportTriageAgent.process_email

Name	Type	Required	Description
email_content	str	Yes	The full text of the customer email to process
run_id	Optional[str]	No	Optional run identifier; auto-generated from timestamp and content hash if not provided

DeterministicExtractor.extract / LLMExtractor.extract

Name	Type	Required	Description
email_content	str	Yes	The full text of the email to extract information from
category	str	Yes	The classified category: "Bug Report", "Billing", or "Feature Request"

default_workflow_client

Name	Type	Required	Description
extractor_type	Literal["deterministic", "llm"]	No	Type of extractor to use (default: "deterministic")

Outputs

ConfigurableSupportTriageAgent.process_email

Name	Type	Description
return	Dict[str, Any]	Dictionary with keys: "category" (str), "extracted_info" (dict), "response_template" (str), "extraction_mode" (str)

DeterministicExtractor.extract (Bug Report)

Name	Type	Description
return	Dict[str, Optional[str]]	Dictionary with keys: "product_version", "error_code"

DeterministicExtractor.extract (Billing)

Name	Type	Description
return	Dict[str, Optional[str]]	Dictionary with keys: "invoice_number", "amount"

DeterministicExtractor.extract (Feature Request)

Name	Type	Description
return	Dict[str, Optional[str]]	Dictionary with keys: "requested_feature", "product_area", "urgency_level"

export_traces_to_log

Name	Type	Description
return	str	File path of the saved trace log JSON file

Usage Examples

Using the Default Workflow Client

from workflow import default_workflow_client

# Create agent with deterministic extraction (no API key needed for extraction)
agent = default_workflow_client(extractor_type="deterministic")

# Process a customer email
result = agent.process_email(
    "Hi, I'm getting error code XYZ-123 when using version 2.1.4. Please help!"
)

print(f"Category: {result['category']}")
print(f"Extracted info: {result['extracted_info']}")
print(f"Response: {result['response_template']}")

Switching Extractors at Runtime

from openai import OpenAI
from workflow import ConfigurableSupportTriageAgent, DeterministicExtractor, LLMExtractor

api_key = "your-api-key"
agent = ConfigurableSupportTriageAgent(api_key=api_key)

# Start with deterministic extraction
result_det = agent.process_email("Invoice #INV-2024-001 for $299.99 seems incorrect.")

# Switch to LLM-based extraction
llm_extractor = LLMExtractor(OpenAI(api_key=api_key))
agent.set_extractor(llm_extractor)

# Re-process with LLM extraction
result_llm = agent.process_email("Invoice #INV-2024-001 for $299.99 seems incorrect.")

# Compare extraction results
print(f"Deterministic: {result_det['extracted_info']}")
print(f"LLM: {result_llm['extracted_info']}")

Accessing Trace Logs

import json
from workflow import default_workflow_client

agent = default_workflow_client()
result = agent.process_email("Add dark mode support for the dashboard, it's urgent!")

# Traces are automatically saved; also available in memory
for trace in agent.traces:
    print(f"[{trace.event_type}] {trace.component}: {trace.data}")

Related Pages

• Environment:Vibrantlabsai_Ragas_Python_3_9_Core_Environment