File size: 3,734 Bytes
b6ed6f7 0aa8e87 b6ed6f7 18ddfa2 b6ed6f7 0aa8e87 b6ed6f7 18ddfa2 0aa8e87 b6ed6f7 0aa8e87 18ddfa2 0aa8e87 18ddfa2 0aa8e87 18ddfa2 0aa8e87 b6ed6f7 0aa8e87 | 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 | from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from src.agents.CallState import CallState
from src.agents.schemas import QualityScores
import inspect
class QualityScoringAgent:
RUBRIC_VERSION = "v1"
RUBRIC_TEXT = (
"Scoring rubric (0–10 each):\n"
"- Tone: 0 hostile/arguing; 3 curt/tense; 5 neutral; 7 friendly/empathic; 10 consistently calm, respectful, and de-escalating.\n"
"- Professionalism: 0 rude/unprofessional; 3 unclear or dismissive; 5 acceptable; 7 clear, courteous, policy-aligned; 10 excellent clarity, appropriate boundaries, and ownership.\n"
"- Structured resolution: 0 no attempt; 3 vague/no next steps; 5 partial (some questions/steps); 7 clear diagnosis + next steps + confirmation; 10 fully structured (issue, actions, timelines, confirmation, and closure).\n"
"Notes must cite 1–3 specific behaviors from the transcript (no long quotes)."
)
def __init__(self):
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
def __call__(self, state: CallState) -> CallState:
"""Evaluates tone, professionalism, and structured resolution with rubric."""
clean_text = state.get("clean_content", state.get("content", ""))
if not clean_text:
return state
prompt = PromptTemplate.from_template(
"Evaluate the following call transcript for tone, professionalism, and structured resolution.\n\n"
"{rubric}\n\n"
"Return scores and brief notes.\n\n"
"Transcript:\n{text}"
)
structured_llm = self._structured_llm()
chain = prompt | structured_llm
result = chain.invoke({"text": clean_text, "rubric": self.RUBRIC_TEXT})
if isinstance(result, QualityScores):
if hasattr(result, "model_dump"):
result_dict = result.model_dump()
else:
result_dict = result.dict()
else:
result_dict = dict(result or {})
result_dict["rubric_version"] = self.RUBRIC_VERSION
result_dict["rubric"] = self.RUBRIC_TEXT
profanity_count = clean_text.count("***")
if profanity_count > 0:
result_dict["profanity"] = profanity_count
for key in ["tone", "professionalism", "structured_resolution"]:
if key in result_dict and isinstance(result_dict[key], (int, float)):
result_dict[key] = max(0, result_dict[key] - 3)
if state.get("metadata") is None or not isinstance(state.get("metadata"), dict):
state["metadata"] = {}
state["metadata"]["qa_rubric_version"] = self.RUBRIC_VERSION
state["quality_scores"] = result_dict
return state
def _structured_llm(self):
# Prefer OpenAI function calling enforcement when supported by the installed LangChain version.
sig = None
try:
sig = inspect.signature(self.llm.with_structured_output)
except Exception:
sig = None
if sig and "method" in sig.parameters:
try:
return self.llm.with_structured_output(
QualityScores, method="function_calling"
)
except Exception:
pass
return self.llm.with_structured_output(QualityScores)
@staticmethod
def fallback(state: CallState) -> CallState:
state["quality_scores"] = {
"tone": None,
"professionalism": None,
"structured_resolution": None,
"notes": "Scoring failed (rate limit or parse error). Showing placeholders.",
}
return state
|