Spaces:
Sleeping
Sleeping
| from app.Nlu import NLUStage | |
| from app.Planner import decide | |
| from app.Responder import ResponderStage | |
| from app.database import db_manager | |
| class SHLAgent: | |
| """Orchestrates the 3-stage pipeline for one turn: | |
| 1. NLU (LLM) -> structured facts about the conversation | |
| 2. Planner (Python) -> deterministic decision of what to do next | |
| 3. Responder (LLM) -> phrasing, ONLY when the action needs one | |
| (off_topic / ask_question / close never call the LLM | |
| at all, so they can never hallucinate) | |
| Recommendations returned by the Responder are validated against the actual | |
| retrieved catalog candidates before being returned to the API layer — any | |
| name/url the model invents that isn't in the candidate set is dropped. | |
| """ | |
| def __init__(self): | |
| self.nlu = NLUStage() | |
| self.responder = ResponderStage() | |
| def handle_conversation(self, conversation: list) -> dict: | |
| latest = conversation[-1].content | |
| history_str = "\n".join(f"{t.role.capitalize()}: {t.content}" for t in conversation[:-1]) | |
| state = self.nlu.run(history=history_str, latest_message=latest) | |
| # If the NLU call failed due to the LLM provider itself being unavailable | |
| # (not a real extraction gap), say so honestly instead of asking a | |
| # clarifying question the user already answered. | |
| if state.get("_upstream_failure"): | |
| return { | |
| "reply": "I'm having trouble reaching the assessment engine right now — could you try again in a moment?", | |
| "recommendations": [], | |
| "end_of_conversation": False, | |
| } | |
| action = decide(state, conversation) | |
| # --- No-LLM paths: cannot hallucinate by construction --- | |
| if action.kind == "off_topic": | |
| return { | |
| "reply": "I'm focused on helping with SHL assessment recommendations for hiring — happy to help once you've got a role, skill, or candidate pool in mind!", | |
| "recommendations": [], | |
| "end_of_conversation": False, | |
| } | |
| if action.kind == "ask_question": | |
| return { | |
| "reply": action.question, | |
| "recommendations": [], | |
| "end_of_conversation": False, | |
| } | |
| if action.kind == "close": | |
| return { | |
| "reply": "Great — glad that fits. Locking in this shortlist.", | |
| "recommendations": getattr(action, "reuse_recommendations", []), | |
| "end_of_conversation": True, | |
| } | |
| # --- LLM path: compare / recommend / redirect, all need retrieval first --- | |
| keywords = getattr(action, "topic_keywords", state.get("topic_keywords", [])) | |
| # FIX: If we are updating an existing list, focus the vector search STRICTLY | |
| # on the newest message so the new test isn't drowned out by old keywords. | |
| if getattr(action, "updating", False): | |
| query = latest | |
| else: | |
| query = " ".join(keywords) if keywords else latest | |
| candidates = db_manager.query_catalog_structured(query, n_results=8) | |
| candidates_str = "\n".join( | |
| f"- {c['name']} | {c['test_type']} | {c['url']}" for c in candidates | |
| ) or "No close matches found in catalog." | |
| result = self.responder.run( | |
| action=action.kind, # "compare" | "recommend" | "redirect" | |
| role_summary=state.get("role_summary") or "not specified", | |
| purpose=state.get("purpose") or "not specified", | |
| updating=getattr(action, "updating", False), | |
| prior_recommendations=getattr(action, "prior_recommendations", []), | |
| candidates=candidates_str, | |
| history=history_str, | |
| input=latest, | |
| ) | |
| # --- Validation --- | |
| # Combine the fresh candidates with the prior recommendations to create the allowed whitelist | |
| prior_recs = getattr(action, "prior_recommendations", []) | |
| allowed_tests = candidates.copy() | |
| allowed_urls = {c["url"] for c in allowed_tests} | |
| for pr in prior_recs: | |
| if isinstance(pr, dict) and pr.get("url") not in allowed_urls: | |
| allowed_tests.append(pr) | |
| allowed_urls.add(pr["url"]) | |
| # Build validation dictionaries from the combined whitelist | |
| valid_by_url = {c["url"]: c for c in allowed_tests} | |
| valid_by_name = {c["name"].strip().lower(): c for c in allowed_tests} | |
| def _resolve(rec): | |
| if isinstance(rec, dict): | |
| url = rec.get("url") | |
| if url in valid_by_url: | |
| return valid_by_url[url] | |
| name = (rec.get("name") or "").strip().lower() | |
| return valid_by_name.get(name) | |
| if isinstance(rec, str): | |
| return valid_by_name.get(rec.strip().lower()) | |
| return None | |
| resolved = [_resolve(r) for r in result.get("recommendations", [])] | |
| # Clean the final list and deduplicate (in case the LLM added a test twice) | |
| seen_urls = set() | |
| final_recs = [] | |
| for r in resolved: | |
| if r is not None and r["url"] not in seen_urls: | |
| final_recs.append(r) | |
| seen_urls.add(r["url"]) | |
| result["recommendations"] = final_recs | |
| result["end_of_conversation"] = False # closing only ever happens via the planner's "close" path | |
| return result | |
| # Global singleton instance | |
| shl_agent = SHLAgent() |