| 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): |
| |
| 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 |
|
|
