Spaces:
Sleeping
Sleeping
File size: 5,604 Bytes
ca20ec1 | 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 | 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() |