Spaces:
Running
Running
| from typing import Callable | |
| from langgraph.config import get_stream_writer | |
| from app.core.topic import extract_topic | |
| from app.models.pipeline import PipelineState | |
| from app.security.guard_classifier import GuardClassifier | |
| from app.security.sanitizer import sanitize_input, redact_pii | |
| def make_guard_node(classifier: GuardClassifier) -> Callable[[PipelineState], dict]: | |
| def guard_node(state: PipelineState) -> dict: | |
| writer = get_stream_writer() | |
| original_query = state["query"] | |
| # 1. Sanitize and PII-redact before any LLM or classifier call. | |
| sanitized = sanitize_input(original_query) | |
| clean_query = redact_pii(sanitized) | |
| # Emit the first status event now that we have a clean query to describe. | |
| # Topic extraction is O(N) set lookup — adds zero measurable latency. | |
| if clean_query: | |
| topic = extract_topic(clean_query) | |
| label = f"Checking your question about {topic}" if topic else "Checking your question" | |
| else: | |
| topic = "" | |
| label = "Checking your question" | |
| writer({"type": "status", "label": label}) | |
| if len(clean_query) == 0: | |
| return { | |
| "query": clean_query, | |
| "guard_passed": False, | |
| "answer": "I can only answer questions about Darshan's work, projects, and background.", | |
| "path": "blocked", | |
| "query_topic": topic, | |
| } | |
| # 2. Classify (scope evaluation). | |
| is_safe, score = classifier.is_in_scope(clean_query) | |
| if not is_safe: | |
| return { | |
| "query": clean_query, | |
| "guard_passed": False, | |
| "answer": "I can only answer questions about Darshan's work, projects, and background.", | |
| "path": "blocked", | |
| "query_topic": topic, | |
| } | |
| return { | |
| "query": clean_query, | |
| "guard_passed": True, | |
| "query_topic": topic, | |
| } | |
| return guard_node | |