Spaces:
Sleeping
Sleeping
| """ | |
| Query analyzer agent: classifies user intent (data retrieval, parameter info, or out-of-scope) | |
| to route the chatbot to the appropriate handler. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| from typing import Callable | |
| from tools.state import GraphState | |
| def _format_history(history: list[dict], max_turns: int = 6) -> str: | |
| if not history: | |
| return "" | |
| clipped = history[-max_turns:] | |
| lines = [] | |
| for msg in clipped: | |
| role = str(msg.get("role", "user")).strip().lower() | |
| content = str(msg.get("content", "")).strip() | |
| if content: | |
| lines.append(f"{role}: {content}") | |
| return "\n".join(lines) | |
| def build_query_analyzer_node( | |
| llm_json_call: Callable[[str], dict] | None = None, | |
| ): | |
| """ | |
| Factory that returns a query analyzer node for the graph. | |
| The node classifies user intent via LLM into: data_retrieval, parameter_info, | |
| both, or out_of_scope. | |
| """ | |
| def query_analyzer_node(state: GraphState) -> dict: | |
| user_query = state.get("user_query", "") | |
| intent, is_valid, error = "out_of_scope", False, "" | |
| prior_filters = state.get("filters", {}) or {} | |
| history_text = _format_history(state.get("conversation_history", []) or []) | |
| if llm_json_call is not None: | |
| prompt = f""" | |
| You are a query classifier for a pharma market metrics chatbot. | |
| Classify user queries into one intent: | |
| - data_retrieval: user wants metrics/data (volume, share, growth, etc.) | |
| - parameter_info: user asks about parameters (cluster mapping, regions, etc.) | |
| - both: user asks for data AND parameter information at the same time | |
| - out_of_scope: unrelated topics (weather, general knowledge, etc.) | |
| Return strict JSON only with keys: | |
| intent, is_valid, error_message, confidence | |
| CRITICAL: Set is_valid=true for data_retrieval when the user asks a data question, | |
| including when they want to CHANGE filters (e.g. "same but for Spain", "what about Germany", | |
| "do the same in LATAM"). Do NOT set is_valid=false just because the user requests | |
| different country/region/period than previously applied - the filter extraction step | |
| will handle that. Only set is_valid=false for genuinely out-of-scope questions. | |
| User query: {user_query} | |
| Previously applied filters: {json.dumps(prior_filters, ensure_ascii=True)} | |
| Recent conversation: | |
| {history_text} | |
| """ | |
| try: | |
| result = llm_json_call(prompt) | |
| if isinstance(result, str): | |
| result = json.loads(result) | |
| intent = result.get("intent", intent) | |
| is_valid = bool(result.get("is_valid", is_valid)) | |
| error = result.get("error_message", error) or error | |
| except Exception: | |
| pass | |
| if intent not in {"data_retrieval", "parameter_info", "both", "out_of_scope"}: | |
| intent = "out_of_scope" | |
| is_valid = False | |
| if intent == "out_of_scope" and not error: | |
| error = ( | |
| "This question is outside my knowledge domain. I can help with data " | |
| "queries and parameter information about the demo pharma market." | |
| ) | |
| return { | |
| "query_intent": intent, | |
| "is_valid": is_valid, | |
| "error_message": error, | |
| } | |
| return query_analyzer_node | |