File size: 5,741 Bytes
cff1e0e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """Conversation Orchestrator for managing evaluation workflow."""
import time
from typing import Any, Dict, List, Optional
import streamlit as st
from evaluators import create_evaluator
class ConversationOrchestrator:
"""Orchestrates conversation evaluation using multiple evaluators."""
def __init__(self, api_keys: Optional[Dict[str, str]] = None):
"""Initialize orchestrator with API keys.
Args:
api_keys: Dict of API keys, e.g., {"openai": "...", "hf": "..."}
"""
self.api_keys = api_keys or {}
def _extract_scores_flat(self, item: Any) -> Dict[str, Any]:
"""Normalize a per-utterance item into a flat score map.
Supports both:
- {"metrics": {...}} (older shape)
- {...} (already flat map)
"""
if not isinstance(item, dict):
return {}
if "metrics" in item and isinstance(item["metrics"], dict):
return item["metrics"]
return item
def evaluate_conversation(
self, utterances: List[Dict[str, Any]], selected_metrics: List[str]
) -> List[Dict[str, Any]]:
"""Evaluate conversation using selected metrics.
Returns a list of per-utterance rows. For each selected metric, the row
gets a key f"{metric_name}_scores" containing a flat map of scores.
"""
progress_bar = st.progress(0)
status_text = st.empty()
all_evaluator_results: Dict[str, Dict[str, Any]] = {}
total_evaluators = max(1, len(selected_metrics))
for i, metric_name in enumerate(selected_metrics):
status_text.text(
f"Running {metric_name} evaluator ({i+1}/{total_evaluators})..."
)
progress_bar.progress((i + 1) / total_evaluators)
# Create evaluator - pass api_key (singular) from the dict
# Most evaluators use HuggingFace models, so try 'hf' first, then 'openai'
api_key = self.api_keys.get("hf") or self.api_keys.get("openai") or None
evaluator = create_evaluator(metric_name, api_key=api_key)
if evaluator is None:
st.warning(f"Evaluator for metric '{metric_name}' not found")
continue
try:
# Many evaluators ignore **kwargs; it's fine.
result = evaluator.execute(utterances, granularity="utterance")
if result is None:
st.warning(f"Evaluator '{metric_name}' returned no result")
else:
all_evaluator_results[metric_name] = result
except Exception as e:
st.warning(f"Evaluator '{metric_name}' failed: {str(e)}")
time.sleep(0.05)
# Merge results per utterance
results: List[Dict[str, Any]] = []
for idx, utt in enumerate(utterances):
row: Dict[str, Any] = {
"speaker": utt.get("speaker", ""),
"text": utt.get("text", ""),
"index": idx,
}
for eval_name, eval_result in all_evaluator_results.items():
# Determine granularity (default to utterance)
granularity = eval_result.get("granularity")
if not granularity:
# infer by available keys
if "per_utterance" in eval_result:
granularity = "utterance"
elif "per_conversation" in eval_result or "overall" in eval_result:
granularity = "conversation"
elif "per_segment" in eval_result:
granularity = "segment"
else:
granularity = "utterance"
scores: Dict[str, Any] = {}
if granularity == "utterance":
per_u = eval_result.get("per_utterance") or []
if idx < len(per_u):
scores = self._extract_scores_flat(per_u[idx])
else:
scores = {}
elif granularity == "conversation":
# try overall, then per_conversation, normalize to flat metrics
overall = eval_result.get("overall")
if isinstance(overall, dict):
scores = self._extract_scores_flat(overall)
else:
per_conv = eval_result.get("per_conversation", {})
scores = self._extract_scores_flat(per_conv)
elif granularity == "segment":
# Attach the first matching segment that covers this utterance
seg_scores = {}
segments = eval_result.get("per_segment") or []
for seg in segments:
try:
indices = seg.get("utterance_indices") or []
if idx in indices:
seg_scores = self._extract_scores_flat(
seg.get("metrics", {})
)
break
except Exception:
continue
scores = seg_scores
else:
# Unknown granularity; try to be helpful
per_u = eval_result.get("per_utterance") or []
if idx < len(per_u):
scores = self._extract_scores_flat(per_u[idx])
row[f"{eval_name}_scores"] = scores
results.append(row)
status_text.text("Evaluation complete!")
return results
|