Spaces:
Sleeping
Sleeping
| import json | |
| from typing import Any | |
| from pydantic import BaseModel | |
| from app.llm.prompt_manager import PromptManager | |
| from app.query.execution_engine import QueryResult | |
| from app.query.path_extractor import GraphEdge, GraphNode | |
| from app.query.query_router import RoutedQuery | |
| from app.utils.logger import get_logger | |
| logger = get_logger("query.response_formatter") | |
| class QueryResponse(BaseModel): | |
| answer: str | |
| explanation: str | |
| query_used: str | |
| nodes: list[GraphNode] | |
| edges: list[GraphEdge] | |
| data: list[dict] | |
| metadata: dict | |
| class ResponseFormatter: | |
| """Builds the final response from RoutedQuery + QueryResult.""" | |
| def __init__( | |
| self, | |
| llm_client: Any = None, | |
| model: str = "", | |
| prompt_manager: PromptManager | None = None, | |
| ) -> None: | |
| self.llm_client = llm_client | |
| self.model = model | |
| self.prompt_manager = prompt_manager | |
| def format(self, routed_query: RoutedQuery, result: QueryResult) -> QueryResponse: | |
| answer = self._llm_answer(routed_query, result) | |
| explanation = self._build_explanation(routed_query) | |
| return QueryResponse( | |
| answer=answer, | |
| explanation=explanation, | |
| query_used=routed_query.cypher, | |
| nodes=result.graph.nodes, | |
| edges=result.graph.edges, | |
| data=self._sanitize_data(result.data), | |
| metadata={ | |
| "template_name": routed_query.template_name, | |
| "intent_type": routed_query.intent.intent_type, | |
| "confidence": routed_query.intent.confidence, | |
| "execution_time_ms": result.execution_time_ms, | |
| "record_count": len(result.data), | |
| "node_count": len(result.graph.nodes), | |
| "edge_count": len(result.graph.edges), | |
| }, | |
| ) | |
| def _llm_answer(self, routed_query: RoutedQuery, result: QueryResult) -> str: | |
| """Try to generate a natural language answer via LLM. Falls back to hardcoded.""" | |
| if not self.llm_client or not self.prompt_manager: | |
| return self._build_answer(routed_query, result) | |
| try: | |
| sanitized = self._sanitize_data(result.data) | |
| preview = sanitized[:10] | |
| results_json = json.dumps(preview, default=str, ensure_ascii=False) | |
| prompt = self.prompt_manager.render("response_synthesis", { | |
| "user_query": routed_query.intent.raw_query, | |
| "cypher": routed_query.cypher, | |
| "results_json": results_json, | |
| "total_count": str(len(sanitized)), | |
| }) | |
| response = self.llm_client.chat.completions.create( | |
| model=self.model, | |
| messages=[{"role": "user", "content": prompt}], | |
| temperature=0.0, | |
| timeout=15, | |
| ) | |
| answer = response.choices[0].message.content.strip() | |
| if answer: | |
| return answer | |
| except Exception as e: | |
| logger.warning("LLM response synthesis failed, using fallback: %s", e) | |
| return self._build_answer(routed_query, result) | |
| def _build_answer(self, routed_query: RoutedQuery, result: QueryResult) -> str: | |
| """Generate a human-readable answer from the query results.""" | |
| data = result.data | |
| intent = routed_query.intent | |
| if not data: | |
| return "No results found for your query." | |
| if intent.intent_type == "count": | |
| # Expect a single row with a 'total' field | |
| total = data[0].get("total", len(data)) | |
| label = routed_query.parameters.get("label", "nodes") | |
| return f"There are {total} {label} nodes." | |
| if intent.intent_type == "aggregate": | |
| row = data[0] | |
| agg_result = row.get("result", "N/A") | |
| total_nodes = row.get("total_nodes", "N/A") | |
| func = routed_query.parameters.get("aggregate_function", "SUM") | |
| prop = routed_query.parameters.get("property_name", "") | |
| label = routed_query.parameters.get("label", "") | |
| return f"The {func} of {prop} across {total_nodes} {label} nodes is {agg_result}." | |
| if intent.intent_type in ("find_node", "search"): | |
| return f"Found {len(data)} matching node(s)." | |
| if intent.intent_type == "find_path": | |
| return f"Found {len(data)} path(s) between the specified nodes." | |
| if intent.intent_type == "list_neighbors": | |
| return f"Found {len(data)} connected node(s)." | |
| if intent.intent_type == "flow_gaps": | |
| # Summarize gap types | |
| gap_types: dict[str, int] = {} | |
| for row in data: | |
| gap = row.get("gap_type", "Unknown") | |
| gap_types[gap] = gap_types.get(gap, 0) + 1 | |
| summary = ", ".join(f"{count} {gap}" for gap, count in gap_types.items()) | |
| return f"Found {len(data)} sales orders with incomplete flows: {summary}." | |
| if intent.intent_type == "find_unlinked": | |
| label = routed_query.parameters.get("label", "nodes") | |
| target = routed_query.parameters.get("target_label", "") | |
| return f"Found {len(data)} {label} node(s) not linked to {target}." | |
| if intent.intent_type == "rank": | |
| label = routed_query.parameters.get("label", "nodes") | |
| target = routed_query.parameters.get("target_label", "") | |
| return f"Top {len(data)} {label} node(s) ranked by connections to {target}." | |
| if intent.intent_type == "find_latest": | |
| label = routed_query.parameters.get("label", "nodes") | |
| direction = routed_query.parameters.get("order_direction", "DESC") | |
| qualifier = "most recent" if direction == "DESC" else "oldest" | |
| return f"Found {len(data)} {qualifier} {label} node(s)." | |
| if intent.intent_type == "distinct": | |
| prop = routed_query.parameters.get("property_name", "") | |
| label = routed_query.parameters.get("label", "nodes") | |
| count = len(data) | |
| values = [str(row.get("value", "")) for row in data[:20]] | |
| values_str = ", ".join(values) | |
| if count > 20: | |
| values_str += f", ... ({count - 20} more)" | |
| return f"Found {count} distinct {prop} values for {label}: {values_str}." | |
| return f"Query returned {len(data)} result(s)." | |
| def _build_explanation(self, routed_query: RoutedQuery) -> str: | |
| """Build a human-readable explanation of what the query does.""" | |
| intent = routed_query.intent | |
| params = routed_query.parameters | |
| if intent.intent_type == "count": | |
| return f"Counting all {params.get('label', '')} nodes in the graph." | |
| if intent.intent_type == "find_node": | |
| label = params.get("label", "") | |
| prop = params.get("property_name", "") | |
| val = params.get("property_value", "") | |
| return f"Finding {label} nodes where {prop} = '{val}'." | |
| if intent.intent_type == "search": | |
| label = params.get("label", "") | |
| val = params.get("search_value", "") | |
| return f"Searching all properties of {label} nodes for value '{val}'." | |
| if intent.intent_type == "find_path": | |
| return ( | |
| f"Finding the shortest path from " | |
| f"{params.get('start_label', '')} ({params.get('start_value', '')}) to " | |
| f"{params.get('end_label', '')} ({params.get('end_value', '')})." | |
| ) | |
| if intent.intent_type == "list_neighbors": | |
| label = params.get("label", "") | |
| prop = params.get("property_name", "") | |
| val = params.get("property_value", "") | |
| return f"Listing all nodes connected to {label} where {prop} = '{val}'." | |
| if intent.intent_type == "aggregate": | |
| func = params.get("aggregate_function", "SUM") | |
| prop = params.get("property_name", "") | |
| label = params.get("label", "") | |
| return f"Computing {func} of {prop} for all {label} nodes." | |
| if intent.intent_type == "flow_gaps": | |
| return "Analyzing Order-to-Cash flow for sales orders with missing delivery, billing, journal entry, or payment steps." | |
| if intent.intent_type == "find_unlinked": | |
| label = params.get("label", "") | |
| rel = params.get("relationship_type", "") | |
| target = params.get("target_label", "") | |
| return f"Finding {label} nodes that have no {rel} relationship to {target}." | |
| if intent.intent_type == "rank": | |
| label = params.get("label", "") | |
| rel = params.get("relationship_type", "") | |
| target = params.get("target_label", "") | |
| return f"Ranking {label} nodes by number of {rel} relationships to {target}." | |
| if intent.intent_type == "find_latest": | |
| label = params.get("label", "") | |
| prop = params.get("order_property", "creationDate") | |
| direction = params.get("order_direction", "DESC") | |
| qualifier = "most recent" if direction == "DESC" else "oldest" | |
| return f"Finding {qualifier} {label} nodes ordered by {prop}." | |
| if intent.intent_type == "distinct": | |
| prop = params.get("property_name", "") | |
| label = params.get("label", "") | |
| return f"Listing distinct values of {prop} across all {label} nodes." | |
| if intent.intent_type == "custom": | |
| return f"Running a custom query generated from your question." | |
| return f"Executing a {intent.intent_type} query." | |
| def _sanitize_data(data: list[dict]) -> list[dict]: | |
| """Convert raw Neo4j records to JSON-serializable dicts. | |
| Neo4j Node/Relationship objects are flattened into the row so that | |
| their properties appear as regular columns. | |
| """ | |
| sanitized = [] | |
| for row in data: | |
| clean: dict = {} | |
| for key, value in row.items(): | |
| # Neo4j Node or Relationship — flatten properties into row | |
| if hasattr(value, "_properties"): | |
| for pk, pv in value._properties.items(): | |
| clean[pk] = pv | |
| elif hasattr(value, "items") and not isinstance(value, dict): | |
| for pk, pv in value.items(): | |
| clean[pk] = pv | |
| elif isinstance(value, dict): | |
| # Already a dict (e.g. from driver deserialization) | |
| for pk, pv in value.items(): | |
| clean[pk] = pv | |
| else: | |
| clean[key] = value | |
| sanitized.append(clean) | |
| return sanitized | |