From 6fd396d36b2a376bfc6edec6a8d29091e56505b3 Mon Sep 17 00:00:00 2001 From: yuvalkh Date: Sun, 12 Jul 2026 14:32:15 +0300 Subject: [PATCH 1/4] added ambiguity node after schema explorer --- agent/scripts/upload_all_prompts.py | 144 ++++++++-- agent/src/agent/config.py | 5 +- agent/src/agent/graph.py | 71 ++++- agent/src/agent/mcp_server.py | 32 ++- agent/src/agent/nodes/detect_ambiguity.py | 275 ++++++++++++++++++++ agent/src/agent/nodes/refiner.py | 4 +- agent/src/agent/nodes/satisfaction_check.py | 13 +- agent/src/agent/nodes/schema_explorer.py | 89 +------ agent/src/agent/state.py | 13 +- agent/src/agent/utils/flag_bridge.py | 1 - agent/src/agent/utils/schema_enrichment.py | 63 +---- backend/app/routers/agent.py | 3 +- frontend/src/api/agent.ts | 1 + frontend/src/pages/AgentTestingPage.tsx | 35 ++- 14 files changed, 559 insertions(+), 190 deletions(-) create mode 100644 agent/src/agent/nodes/detect_ambiguity.py diff --git a/agent/scripts/upload_all_prompts.py b/agent/scripts/upload_all_prompts.py index deb2cdb..684ddca 100644 --- a/agent/scripts/upload_all_prompts.py +++ b/agent/scripts/upload_all_prompts.py @@ -1,9 +1,10 @@ import os from langfuse import Langfuse - +from dotenv import load_dotenv # ============================================================================== # CONFIGURATION - Set your private Langfuse server credentials here # ============================================================================== +load_dotenv() LANGFUSE_PUBLIC_KEY = os.getenv("LANGFUSE_PUBLIC_KEY") LANGFUSE_SECRET_KEY = os.getenv("LANGFUSE_SECRET_KEY") LANGFUSE_BASE_URL = os.getenv("LANGFUSE_BASE_URL") or os.getenv("LANGFUSE_HOST") @@ -65,21 +66,10 @@ def main(): "- Join strategy: If multiple tables are needed to answer the query, decide which tables to join and on which keys — do NOT ask the user.\n" "- Column selection: Choose the most appropriate columns yourself.\n" "- Filter strategy: Infer filters from the user's question.\n" - "- Table selection when one is clearly more appropriate: Pick the best match and proceed.\n\n" - "## When to Flag Ambiguity (ONLY these cases)\n\n" - "Set ambiguity_detected=true ONLY IF:\n" - "1. Two or more completely independent tables could each independently and fully answer the user's question, " - "and you have no way to determine which one the user wants (e.g., 'orders' vs 'orders_archive' with no " - "indication of time range, or two fact tables from different business domains that both seem equally relevant).\n" - "2. There is no table in the catalog that can answer the query at all.\n\n" - "Do NOT flag ambiguity for: join decisions, column choices, filter logic, or any decision you can make yourself.\n\n" - "## Output Format\n\n" - "Output MUST be a valid JSON object with the following keys:\n" - "- schema_plan: detailed query plan describing tables, columns, joins, and Esca reference IDs (empty string if ambiguity_detected is true)\n" - "- ambiguity_detected: boolean, true ONLY in the hard-blocker cases described above\n" - "- ambiguity_message: a concise question to ask the user to resolve the blocker (empty string if ambiguity_detected is false)\n" - "- candidate_options: list of strings (table names or options) for the user to choose from (empty list if ambiguity_detected is false)\n" - "Return only the raw JSON, no markdown formatting (no ```json code blocks)." + "- Table selection: When one table is clearly more appropriate, pick the best match and proceed.\n" + "- When uncertain between two tables, pick the most semantically appropriate one and document your reasoning in schema_plan.\n\n" + "## Output Instructions\n\n" + "Provide your output matching the requested schema. Ensure that `schema_plan` is a detailed string explanation, and `tables_used` is a list of table names." ) }, { @@ -172,6 +162,128 @@ def main(): } ], "type": "chat" + }, + { + "name": "text2sql/detect_ambiguity", + "prompt": [ + { + "role": "system", + "content": ( + "# **ROLE AND OBJECTIVE**\n" + "You are an advanced Text-to-SQL Diagnostic and Routing Agent that detects ambiguity in natural language requests intended for SQL generation. You sit at the intersection of Natural Language Intent, Database Schema Reality, and the Downstream SQL Generator's current interpretation.\n" + "Your primary objective is to determine if a User Request can be **deterministically** translated into SQL given the provided **Schema** and a **Current Agent SQL Attempt** after deeply analyzing their structural and semantic feasibility.\n" + "Your goal is to **pass queries to SQL generation whenever a reasonable, standard interpretation exists**, while intercepting only **truly ambiguous** or **impossible** request.\n\n" + "**Core Philosophy:**\n" + "1. **Lean towards \"CLEAR\":** If a human analyst would confidently answer the query using standard logic, mark it CLEAR.\n" + "2. **Allow Logical Inference and Heuristics:** Assume the downstream SQL generator can handle implicit table/column choices based on schema structure (e.g., choosing `active_users` over `archive_users` for \"current users\") unless there is a **direct conflict**. However, **do NOT assume fuzzy matching**. If the user's term does not explicitly match the schema values or column names, and there is no single obvious exact match, this may be an ambiguity.\n" + "3. **Intervene Only When Necessary:** Flag ambiguity ONLY when multiple interpretations lead to **drastically different data** or when the data is missing. Do not ask for optional details like specific date ranges, window sizes, or table names unless the request is genuinely unintelligible.\n" + "4. **Quality Assurance Auditor:** you will review the `Current Agent SQL Attempt` as the Agent's **proposed interpretation**. You do not reject the Agent's choice out of skepticism. Instead, you verify if the Agent's choice aligns with the **strongest available heuristic** in the schema.\n\n" + "You do not generate the final SQL. Instead, you act as the system's execution planner and ambiguity detector. You must protect the downstream SQL Composer from hallucinations, far fetched assumptions, and impossible requests by flaging queries that lack the necessary context to generate a logically correct SQL query. If a user query is ambiguous you will halt execution and formulate a user friendly clarification request.\n\n" + "# **EXECUTION WORKFLOW**\n" + "Before making a final determination, you must rigorously process the query through the following chronological steps. You will output this internal reasoning step-by-step.\n\n" + "1. Intent Deconstruction: Break down the natural language query into core components (desired output columns, temporal filters, aggregations, mathematical operations).\n" + "2. Logical SQL Planning: Draft a mental, step-by-step execution plan required to solve the query (e.g., \"I will need to JOIN the users table to the sales table on user_id, apply a WHERE filter for the date, and GROUP BY region\").\n" + "3. Schema Alignment: Systematically test your mental plan against the provided schema. Can every conceptual requirement be mapped to a concrete table and column?\n" + "4. State Determination: Based on the alignment test, classify the query's feasibility.\n\n" + "Process the query through these steps. If you can resolve any uncertainty using common sense or schema context, continue to CLEAR. Only stop if you hit a hard block.\n\n" + "1. **Intent Deconstruction:** Identify the core request: Who, What, When (Metric, Filters, Grouping).\n\n" + "2. **Active Schema Search & Heuristic Mapping:**\n" + " - **Search:** Actively look for columns/tables/values that semantically or syntactically match the user's terms.\n" + " - Map intent to schema.\n" + " - **Evaluate Confidence:**\n" + " - **Single Dominant Match:** Is there one column/table/value that is clearly the best fit by name or primary source for this concept?\n" + " *Example: User asks for \"User Name\". Schema has `full_name`, `first_name`, `last_name`, `username_handle`. `full_name` is the obvious default for \"Name\". → **CLEAR**.*\n" + " *Counter-Example: User asks for \"Date\". Schema has `created_at`, `updated_at`, `shipped_date`. No single default exists. → **FLAG**.*\n" + " - **Partial/Prefix Matches:** If the user says \"Premium\" and the schema has `plan_type='Premium'`, `customer_tier='Premium'`, and `discount_label LIKE 'Premium%'`, does business logic dictate one over the others? (e.g., if \"Premium\" usually implies Plan Type, use Plan Type). If yes → **CLEAR**.\n" + " - **Equal Weight Matches:** If multiple columns/tables contain the value and none is clearly superior (e.g., `region`, `district`, and `zone` all have \"North\" and are used interchangeably in business), → **FLAG**.\n" + " - **Value Check:** Does the user's filter value (e.g., \"New York\") explicitly exist in the column?\n" + " - If YES → Proceed.\n" + " - If NO, is there a single obvious alias or standard abbreviation (e.g., \"NY\") that is the *only* logical match? → Proceed.\n" + " - If NO, are there multiple potential matches or no matches? → **FLAG** as Ambiguous/Unanswerable.\n" + " - **Table Check:** If multiple tables contain similar data, does business logic favor one? (e.g., \"current status\" → `active_users`). If yes, proceed.\n" + " - **Semantic Column Priority:** Prioritize columns whose names semantically match the concept over columns whose values happen to contain the string.\n" + " Example: User asks for \"Price\". Schema has unit_price and description (values: 'Price is $5'). unit_price is the CLEAR winner.\n" + " - **Table Collision Check:** If the user asks for an entity (e.g. \"Sales\") and multiple tables exist with distinct scopes (e.g., main_sales, focus_sales, archived_sales), FLAG as Ambiguous.\n\n" + "3. **Ambiguity Check (The \"Stop\" Gate):**\n" + " Ask: *\"If I guess here, will I likely be wrong or return significantly different data?\"*\n" + " - **YES** → Flag as Ambiguous.\n" + " - **NO** (Standard interpretation exists) → Mark as Clear.\n" + " - **Specific Checks for Flagging:**\n" + " - **Direct Collision:** Two columns/tables have similar names and no context to distinguish them (e.g., `region_code` vs `area_code` for \"location\").\n" + " - **Missing Critical Logic:** User asks for \"Profit Margin\" but schema has no such column and no price/cost columns to derive it.\n" + " - **Contradiction:** User requests data that logically cannot exist together.\n\n" + "4. **Agent Proposal Audit**\n" + " Compare the `Current Agent SQL Attempt` against the **Dominant Standard** found in Step 2.\n" + " **Clear Standard Exists**\n" + " * Did the Agent use the Dominant Standard?\n" + " * YES → **CLEAR**.\n" + " * NO → **FLAG**. (Reason: Agent chose a sub-optimal column/value. e.g., Agent used `first_name` when `full_name` was available).\n\n" + " **No Clear Standard (Tie)**\n" + " * Are there two or more columns with equal heuristic weight?\n" + " * *Example:* User: \"Location\". Schema: `region`, `district`, `zone`. None is clearly \"the\" location.\n" + " * *Analysis:* There is no dominant standard. The Agent *must* guess.\n" + " * *Verdict:* **FLAG**. Even if the Agent picked `region`, it's an arbitrary choice among equals. A human would need to ask \"Do you mean region, district, or zone?\".\n\n" + " **No Standard & Agent Made a Reasonable Guess**\n" + " * *Example:* User: \"Date\". Schema: `created_at`, `updated_at`.\n" + " * *Analysis:* \"Date\" is ambiguous. `created_at` is a common default, but `updated_at` is also valid.\n" + " * *Verdict:* **FLAG**. The ambiguity lies in the User Request + Schema structure, not just the Agent's error. The Agent cannot be deterministic here.\n\n" + "5. **Final Decision:** Output JSON.\n\n" + "# **TAXONOMY OF FAILURES**\n" + "If the query cannot be processed (is not CLEAR), it must fall into one of the following exact failure modes.\n\n" + "## A. Unanswerable (Fatal Failure)\n" + "* **Missing Domain Data:** The requested metric/entity does not exist in the schema, and no reasonable calculation or combination of existing columns can derive it. No clarification will help.\n" + " *Example: Asking for \"support ticket resolution time\" when the database only contains \"marketing email campaigns\".*\n" + "* **No Value Match:** The user specifies a filter value that does not exist in the relevant column, and no standard alias exists.\n" + " *Example: User asks for \"Sales in 'North America'\" but the `region` column only contains 'NA', 'EU', 'APAC'. If 'NA' is the only logical match, CLEAR. If 'North America' could map to multiple ambiguous codes or none, FLAG.*\n\n" + "## B. Database-Related Ambiguity (Schema & Mapping Failures)\n" + "* **Missing Explicit Filter Value:** The user asks to filter by a specific entity (e.g. \"my specific ID\", \"that user\", \"a certain order\") but does NOT provide the actual value. Do NOT assume it is a parameter to be filled later. You MUST flag this as AMBIGUOUS and ask the user for the exact value.\n" + " *Example: User asks for \"Give the order with my specific ID\" but provides no ID. → FLAG.*\n" + "* **Direct Schema Collision:** The user asks for a concept that maps to **two or more equally valid columns/tables** without sufficient context to prefer one. Guessing would lead to significantly different data.\n" + " *Example: User asks for \"Location\". Schema has `shipping_address`, `billing_address`, and `current_gps`. No context provided. → FLAG.*\n" + " *Example: User asks for \"Revenue\". Schema has `gross_revenue` and `net_revenue`. No context provided. → FLAG.*\n" + " *Example: User asks for \"Date\". Schema has `created_at`, `updated_at`, `shipped_date`. No context provided. → FLAG.*\n" + "* **Unclear Schema Reference (Temporal/Attribute Conflict):** The query lacks sufficient contextual detail to determine the correct table or column, specifically where similarly named columns exist with different meanings.\n" + " *Example: \"Oldest user\" mapping to `date_of_birth` vs. `registration_date`. → FLAG.*\n" + "* **Unclear Value Reference (Colloquialism vs. Literal):** The user query's terminology refers to a specific data point using colloquialisms that do not directly match the literal values stored within the database records, and no single obvious mapping exists.\n" + " *Example: User refers to \"New York City,\" while the database strictly utilizes the string \"NYC\" or \"NY\". → FLAG.*\n\n" + "## C. Model-Related Ambiguity (Logic Failures)\n" + "* **Undefined Mathematical Dependencies:** The user uses a qualitative term (like \"best,\" \"top,\" or \"performance\") or Ambiguous Aggregation Intent (like summary or aggregation) that implies a specific mathematical aggregation or sorting order, but the schema supports **multiple distinct, valid calculations** with no dominant standard.\n" + " *Example: User asks for \"Best Customers\". Schema has `total_revenue`, `order_count`, and `avg_order_value`. \"Best\" could mean highest spender, most frequent, or highest ticket size. No single column is named \"best_score\" or \"customer_rating\". → FLAG.*\n" + " *Example: User asks for \"Total Product Performance\". Schema has `units_sold`, `revenue_generated`, and `profit_margin`. It is unclear which metric defines \"performance\". → FLAG.*\n" + " *Example: User asks \"Summarize server events\". Schema has `events` table with columns `event_type`, `server_id`, `severity`, `timestamp`. It is unclear if the user wants a count by `event_type`, by `server_id`, by `severity`, or a time-series count. → FLAG.*\n" + " *Example: User asks \"Give me a breakdown of orders\". Schema has `orders` table with `status`, `region`, `product_category`. It is unclear which dimension drives the \"breakdown\". → FLAG.*\n" + " *Counter-Example: User asks for \"Total Sales\". Schema has a single column `sales_amount`. → CLEAR (Implicit SUM).*\n" + " *Counter-Example: User asks for \"Most Expensive Products\". Schema has `price`. → CLEAR (Implicit ORDER BY price DESC).*\n" + "* **Ambiguous Operational Intent:** The operational intent is syntactically incomplete in a way that changes the result structure drastically, and no standard default exists.\n" + " *Example: \"Show users by date\" - is this an ORDER BY sort, or a GROUP BY aggregate? → FLAG.*\n" + "* **Conflicting Knowledge:** The query contains filters or entity requests that contradict factual logic or the known schema structure.\n\n" + "## D. General Ambiguity\n" + "* **Unclear Intent:** The query is structurally incoherent, fundamentally contradictory, or the primary objective cannot be reasonably deduced.\n\n" + "# **FLEXIBILITY INSTRUCTIONS**\n" + "* **Implicit Timeframes:** For example if the user says \"recent sales\" but doesn't say \"last 7 days,\" mark **CLEAR**. The SQL generator should infer a standard window or use the most recent data.\n" + "* **Implicit Tables:** For example if the user says \"Show user emails\" and both `users` and `customers` have emails, but `users` is the master table, mark **CLEAR**.\n" + "* **Contextual & Temporal Defaulting:** Assume standard real-world context and \"current state\" heuristics. For example if a user refers to \"*This* season,\" \"*Current* campaign,\" \"*Active* users,\" or \"Now,\" mark **CLEAR**. The system should assume the most recent, active, or highest-priority record relevant to that context. Do NOT flag these as ambiguous; they are standard semantic shortcuts.\n" + "* **Vague Aggregations:** For example if the user says \"Show me sales,\" assume `COUNT` or `SUM` is appropriate. Mark **CLEAR**.\n" + "* **Single Obvious Alias:** For example if the user says \"NY\" and the DB has \"New York\", and \"NY\" is a common standard abbreviation for it, and no other \"NY\" exists (e.g., not \"New York\" vs \"New Jersey\" both abbreviated NY), mark **CLEAR**.\n" + "* **Do NOT flag for minor syntactic differences** if the semantic match is unique and obvious.\n\n" + "# **CONSTRAINTS AND GUIDELINES**\n" + "* Fight for \"CLEAR\": Do not be overly pedantic. If a query strongly implies standard business logic that a human analyst would confidently execute, mark it CLEAR. Only flag queries where multiple interpretations are equally valid and result in drastically different data. For example, try to heuristically infer requested information via the LLM's external logic (e.g. if a user requests \"customer complaints,\" but the database lacks a dedicated complaints column, the system can search for relevant related column and for instance apply WHERE description LIKE '%complaint%' rather than flagging it as ambiguous.)\n" + "* Never Hallucinate Data: If the schema does not have the data, you must mark it UNANSWERABLE. DO NOT invent tables or columns.\n" + "* Clarification UX: Clarification questions must be strictly non-technical. Never use SQL jargon (JOIN, GROUP BY, schema names). Frame questions as clear business choices (e.g., \"Would you like to measure this by the date the account was created, or the date of their last login?\").\n\n" + "# **Additional Information:**\n\n" + "* The current time is: {{current_time}}" + ) + }, + { + "role": "user", + "content": ( + "User Request: {{user_query}}\n\n" + "Schema:\n{{schema}}\n\n" + "Current Agent SQL Attempt:\n{{current_sql_attempt}}" + ) + } + ], + "type": "chat" } ] diff --git a/agent/src/agent/config.py b/agent/src/agent/config.py index 41c3b09..dae2d93 100644 --- a/agent/src/agent/config.py +++ b/agent/src/agent/config.py @@ -43,6 +43,7 @@ class AgentSettings(BaseSettings): "text2sql/finalizer_sql_explanation" ) LANGFUSE_PROMPT_REJECTION_ROUTER: str = "text2sql/rejection_router" + LANGFUSE_PROMPT_DETECT_AMBIGUITY: str = "text2sql/detect_ambiguity" MAX_REFINER_ITERATIONS: int = Field(default=3, gt=0) REFINER_SCHEMA_CONTEXT_TABLES: int = Field(default=4, gt=0) @@ -54,7 +55,6 @@ class AgentSettings(BaseSettings): ENABLE_SEMANTIC_TYPING: bool = False # single batched LLM call — adds id/timestamp/category labels ENABLE_JOIN_GRAPH: bool = False ENABLE_SCHEMA_SUMMARIZATION: bool = False # generated once at profile-time, not at runtime - ENABLE_AMBIGUITY_DETECT: bool = False ENABLE_SKILL_INJECTION: bool = False # ── G2-04: Satisfaction Check ───────────────────────────────────────────── @@ -68,6 +68,9 @@ class AgentSettings(BaseSettings): SATISFACTION_SEMANTIC_THRESHOLD: float = 0.75 SATISFACTION_MAX_FAILURES: int = 2 # escalate to HITL after this many check failures + # ── Ambiguity Resolution ────────────────────────────────────────────────── + MAX_AMBIGUITY_RETRIES: int = 2 # max times user can clarify before hard stop + # ── G2-05: Redis Schema Cache ───────────────────────────────────────────── SCHEMA_CACHE_TTL: int = Field(default=600, gt=0) # seconds — DDL content PROFILE_CACHE_TTL: int = Field(default=1800, gt=0) # seconds — table profile statistics diff --git a/agent/src/agent/graph.py b/agent/src/agent/graph.py index aae6912..f58ed04 100644 --- a/agent/src/agent/graph.py +++ b/agent/src/agent/graph.py @@ -2,13 +2,21 @@ LangGraph agent graph — Group 2 hardened topology. Node order: - START → validate_config → extractor → schema_explorer → ... - (G2-01 fail-fast) - -HITL escalation compiles with interrupt_before=["hitl_escalation"] so -LangGraph pauses before executing that node. After a human injects -corrected state the graph resumes from hitl_escalation which immediately -routes to extractor (full state reset path). + START → init_flags → validate_config + → init_skills → extractor → schema_explorer + → sql_static_validations + → detect_ambiguity + ├─ [clear] → query_builder → ... + ├─ [ambiguous] → ambiguity_resolution ← interrupt + │ └─ schema_explorer (retry) + └─ [unanswerable] → END + +HITL escalation compiles with interrupt_before=["hitl_escalation", "ambiguity_resolution"] so +LangGraph pauses before executing either node. After a human injects corrected +state (feedback / clarification) the graph resumes: + • hitl_escalation → extractor (full reset path for hard errors) + • ambiguity_resolution → schema_explorer (targeted retry with user’s clarification + appended as feedback) Satisfaction check sits between refiner success path and finalizer (G2-04). """ @@ -23,12 +31,13 @@ from langchain_core.runnables.config import RunnableConfig from langchain_core.prompts import ChatPromptTemplate from agent.state import AgentState -from agent.utils.redis_publisher import publish_node_event_sync +from agent.utils.redis_publisher import publish_node_event_sync, publish_node_event from agent.nodes.extractor import extractor_node from agent.nodes.init_flags import init_flags_node from agent.nodes.init_skills import init_skills_node from agent.nodes.schema_explorer import schema_explorer_node, MAX_SCHEMA_RETRIES, sql_static_validations_node from agent.nodes.query_builder import query_builder_node +from agent.nodes.detect_ambiguity import detect_ambiguity_node, ambiguity_resolution_node from agent.nodes.refiner_graph import refiner_subgraph from agent.nodes.finalizer import finalizer_node from agent.config import settings @@ -184,6 +193,27 @@ def route_refiner_subagent(state: AgentState) -> str: return "finalizer" +def route_detect_ambiguity(state: AgentState) -> str: + """ + Route out of detect_ambiguity based on the resolved ambiguity_type. + + - "clear" → query_builder (proceed normally) + - "ambiguous" → ambiguity_resolution (HITL: user clarifies, then retry) + → END if MAX_AMBIGUITY_RETRIES exhausted + - "unanswerable" → END (data doesn’t exist; clarification won’t help) + """ + t = state.get("ambiguity_type") or "clear" + + if t == "clear": + return "query_builder" + + if t == "unanswerable": + return END + + # ambiguous — offer HITL (max retries logic is now handled inside detect_ambiguity_node) + return "ambiguity_resolution" + + def route_query_builder(state: AgentState) -> str: if state.get("feedback"): return "rejection_router" @@ -207,6 +237,8 @@ def route_rejection(state: AgentState) -> str: workflow.add_node("extractor", extractor_node) workflow.add_node("schema_explorer", schema_explorer_node) workflow.add_node("sql_static_validations", sql_static_validations_node) +workflow.add_node("detect_ambiguity", detect_ambiguity_node) +workflow.add_node("ambiguity_resolution", ambiguity_resolution_node) workflow.add_node("query_builder", query_builder_node) workflow.add_node("rejection_router", rejection_router_node) workflow.add_node("refiner_subagent", refiner_subgraph) @@ -226,11 +258,27 @@ def route_rejection(state: AgentState) -> str: route_schema_explorer, { "schema_explorer": "schema_explorer", - "query_builder": "query_builder", + # route_schema_explorer returns "query_builder" on success; + # we intercept it here and send to detect_ambiguity first. + "query_builder": "detect_ambiguity", "hitl_escalation": "hitl_escalation", # G2-02 }, ) +workflow.add_conditional_edges( + "detect_ambiguity", + route_detect_ambiguity, + { + "query_builder": "query_builder", + "ambiguity_resolution": "ambiguity_resolution", + END: END, + }, +) + +# ambiguity_resolution → schema_explorer: targeted retry (not a full extractor reset). +# The user’s clarification is in state["feedback"] which schema_explorer already reads. +workflow.add_edge("ambiguity_resolution", "schema_explorer") + workflow.add_conditional_edges( "query_builder", route_query_builder, @@ -263,5 +311,8 @@ def route_rejection(state: AgentState) -> str: memory = MemorySaver() agent_graph = workflow.compile( checkpointer=memory, - interrupt_before=["hitl_escalation"], # G2-02: pause before HITL node + interrupt_before=[ + "hitl_escalation", # G2-02: hard error escalation (full reset) + "ambiguity_resolution", # Ambiguity HITL: user clarifies, then schema_explorer retry + ], ) diff --git a/agent/src/agent/mcp_server.py b/agent/src/agent/mcp_server.py index e36fbc0..9f36b4c 100644 --- a/agent/src/agent/mcp_server.py +++ b/agent/src/agent/mcp_server.py @@ -64,14 +64,24 @@ async def chat_with_agent( } ) + update_dict = { + "last_error": None, + "trino_error": None, + "escalated": None, + "escalation_reason": None, + "refinement_count": 0, + } + + # If paused at a node breakpoint (interrupt_before), manually map the resume + # value into the state, since Command(resume=) only targets dynamic interrupt() + if state_snapshot.next: + if isinstance(resume_value, str): + update_dict["feedback"] = resume_value + elif isinstance(resume_value, dict) and "feedback" in resume_value: + update_dict["feedback"] = resume_value["feedback"] + result = await agent_graph.ainvoke(Command( - update={ - "last_error": None, - "trino_error": None, - "escalated": None, - "escalation_reason": None, - "refinement_count": 0, - }, + update=update_dict, resume=resume_value ), config=config) else: @@ -136,6 +146,9 @@ async def chat_with_agent( # Get trace_id from the Langfuse handler if available trace_id = getattr(langfuse_handler, "last_trace_id", None) if langfuse_handler else None + if langfuse_handler and hasattr(langfuse_handler, "flush"): + langfuse_handler.flush() + from agent.langfuse_client import langfuse_client langfuse_client.flush() @@ -177,6 +190,10 @@ async def chat_with_agent( } ) + is_unans = False + if final_state.values.get("ambiguity_type") == "unanswerable" or final_state.values.get("failure_reason"): + is_unans = True + return json.dumps( { "thread_id": thread_id, @@ -188,6 +205,7 @@ async def chat_with_agent( "schema_plan": result.get("schema_plan"), "trace_id": trace_id, "execution_path": result.get("execution_path", []), + "is_unanswerable": is_unans, } ) diff --git a/agent/src/agent/nodes/detect_ambiguity.py b/agent/src/agent/nodes/detect_ambiguity.py new file mode 100644 index 0000000..c7b6b46 --- /dev/null +++ b/agent/src/agent/nodes/detect_ambiguity.py @@ -0,0 +1,275 @@ +""" +detect_ambiguity — Pre-SQL Ambiguity Detection Node +==================================================== + +WHY THIS NODE SITS PRE-SQL (AFTER SCHEMA EXPLORER, BEFORE QUERY BUILDER): +-------------------------------------------------------------------------- +The ambiguity-detection prompt requires three inputs: + + 1. The original user request — always available + 2. The database schema — available from state["table_profiles"] after + schema_explorer runs + 3. The agent's current plan — the schema_plan produced by schema_explorer + describes WHICH tables and columns the agent + intends to use, making it a rich "proposed + interpretation" that the prompt's Agent Proposal + Audit (Step 4) can meaningfully evaluate. + +Using schema_plan (rather than waiting for actual SQL) means: + - Ambiguity is caught BEFORE query_builder runs (saves 1 LLM call) + - Ambiguity is caught BEFORE Trino executes (saves the DB round-trip) + - The schema_plan is actually MORE explicit about intent than the final SQL, + because it spells out the reasoning in plain language. + - False positives from SQL syntax errors are impossible (no SQL exists yet). + +If a query is genuinely ambiguous, we short-circuit the entire expensive +pipeline — no query_builder, no refiner, no Trino — and return a clarifying +question to the user immediately. + +Graph position (main agent graph): + sql_static_validations → detect_ambiguity + ├─ [clear] → query_builder → refiner_subagent → finalizer + ├─ [ambiguous] → END (clarifying_questions surfaced to caller) + └─ [unanswerable] → END (failure_reason surfaced to caller) +""" + +import json +import logging +from datetime import datetime, timezone + +from langchain_core.messages import HumanMessage, SystemMessage +from langchain_core.runnables.config import RunnableConfig +from pydantic import BaseModel, Field + +from agent.config import settings +from agent.langfuse_client import langfuse_client +from agent.llm import get_llm +from agent.nodes.refiner import build_refiner_schema_context +from agent.state import AgentState +from agent.utils.redis_publisher import publish_node_event + +logger = logging.getLogger(__name__) + + +def _resolve_ambiguity_type(parsed: dict) -> str: + clarifying: str | None = parsed.get("clarifying_questions") + reason: str = parsed.get("reason") or "" + is_ambiguous = parsed.get("is_ambiguous", False) + + # If the LLM generated a clarifying question (and it's not null/empty), it MUST be ambiguous + # regardless of whether it correctly flipped the boolean flag. + if clarifying is not None and clarifying.strip(): + return "ambiguous" + + # If flagged as ambiguous but no question provided + if is_ambiguous: + if "unanswer" in reason.lower(): + return "unanswerable" + return "ambiguous" + + return "clear" + + +class AmbiguityResult(BaseModel): + is_ambiguous: bool = Field(description="True if the request is ambiguous or unanswerable, False if clear") + reason: str = Field(default="", description="Explanation of the ambiguity detection decision") + clarifying_questions: str | None = Field(default=None, description="Questions to ask the user if ambiguous. Null if clear.") + + +async def detect_ambiguity_node( + state: AgentState, + config: RunnableConfig | None = None, +) -> dict: + """ + Pre-SQL ambiguity gate — runs after schema_explorer, before query_builder. + + Reads the schema_plan (the agent's natural-language query plan describing + which tables and columns it intends to use), the user query, and the enriched + schema profiles, then asks an LLM whether the interpretation is deterministic + or ambiguous/unanswerable. + + Returns a partial state update with: + - ambiguity_result : raw parsed JSON from the LLM + - ambiguity_type : "clear" | "ambiguous" | "unanswerable" + - clarifying_questions / failure_reason (populated on non-clear paths) + """ + thread_id = ( + config.get("configurable", {}).get("thread_id", "") if config else "" + ) + await publish_node_event(thread_id, "detect_ambiguity") + + runtime_flags = state.get("runtime_flags") or {} + + # ── Gather inputs ───────────────────────────────────────────────────────── + user_query: str = state.get("user_query") or "" + # schema_plan is the agent's explicit interpretation of the query before any + # SQL is written — richer than SQL for auditing intent. + agent_plan: str = state.get("schema_plan") or "(no query plan available)" + schema_context: str = build_refiner_schema_context(state) + current_time: str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") + + # ── Fetch system prompt from Langfuse ───────────────────────────────────── + langfuse_prompt = langfuse_client.get_prompt(settings.LANGFUSE_PROMPT_DETECT_AMBIGUITY) + if langfuse_prompt is None: + raise RuntimeError( + f"Langfuse prompt '{settings.LANGFUSE_PROMPT_DETECT_AMBIGUITY}' could not be retrieved." + ) + + # The prompt template uses {{current_time}} — inject it now. + raw_system = langfuse_prompt.prompt + if isinstance(raw_system, list): + # Chat-style prompt stored as list of role dicts; grab the system message. + system_text = next( + (m.get("content", "") for m in raw_system if m.get("role") == "system"), + "", + ) + else: + system_text = str(raw_system) + + system_text = system_text.replace("{{current_time}}", current_time) + + # ── Build user message ──────────────────────────────────────────────────── + # NOTE: The field is labelled "Current Agent SQL Attempt" in the prompt to + # stay consistent with the Langfuse prompt template. At this stage it + # contains the schema_plan (the agent's intent in natural language), which + # is equally valid — the prompt's Agent Proposal Audit works on intent, not + # syntax. + user_message = ( + f"User Request: {state.get('user_query', '')}\n\n" + f"Schema Context:\n{schema_context}\n\n" + f"Current Agent SQL Attempt:\n{agent_plan}\n" + ) + + # ── LLM call ───────────────────────────────────────────────────────────── + _llm = get_llm("detect_ambiguity", runtime_flags=runtime_flags) + structured_llm = _llm.with_structured_output(AmbiguityResult, method="json_schema") + messages = [SystemMessage(content=system_text), HumanMessage(content=user_message)] + + parsed: dict = {} + ambiguity_type: str = "unanswerable" + + try: + response = await structured_llm.ainvoke(messages) + parsed = response.model_dump() + ambiguity_type = _resolve_ambiguity_type(parsed) + except Exception as exc: + logger.error("detect_ambiguity: unexpected error parsing structured output: %s", exc) + parsed = { + "is_ambiguous": True, + "intent_deconstruction": "", + "logical_sql_plan": "", + "schema_alignment_check": "", + "reason": f"Ambiguity detection failed to parse LLM structured output: {exc}", + "clarifying_questions": "", + } + ambiguity_type = "ambiguous" + + # ── Langfuse tracing ────────────────────────────────────────────────────── + try: + if langfuse_client.get_current_trace_id(): + langfuse_client.update_current_span( + metadata={ + "ambiguity_type": ambiguity_type, + "ambiguity_result": parsed, + }, + ) + except Exception as exc: + logger.warning("detect_ambiguity: Langfuse trace update failed: %s", exc) + + # ── Build state update ──────────────────────────────────────────────────── + update: dict = { + "ambiguity_result": parsed, + "ambiguity_type": ambiguity_type, + "execution_path": ["detect_ambiguity"], + # Always clear terminal fields; they are set below only on non-clear paths. + "clarifying_questions": None, + "failure_reason": None, + } + + if ambiguity_type == "ambiguous": + retry_count = state.get("ambiguity_retry_count") or 0 + if retry_count >= settings.MAX_AMBIGUITY_RETRIES: + ambiguity_type = "unanswerable" + update["ambiguity_type"] = ambiguity_type + fr = "Max retries reached for ambiguous question. The agent was not able to resolve the ambiguity with the provided clarifications." + update["failure_reason"] = fr + update["escalation_reason"] = f"Unanswerable Query: {fr}" + update["summary"] = f"**Unanswerable Request:**\n\n{fr}" + return update + + cq = parsed.get("clarifying_questions") or parsed.get("reason") or "" + update["clarifying_questions"] = cq + update["escalation_reason"] = f"Ambiguity Detected: {cq}" + elif ambiguity_type == "unanswerable": + fr = parsed.get("reason") or parsed.get("clarifying_questions") or "" + update["failure_reason"] = fr + update["escalation_reason"] = f"Unanswerable Query: {fr}" + update["summary"] = f"**Unanswerable Request:**\n\n{fr}" + + return update + +# ── Ambiguity Resolution (HITL) ────────────────────────────────────────────── + + +async def ambiguity_resolution_node( + state: AgentState, + config: RunnableConfig | None = None, +) -> dict: + """ + HITL pause point for ambiguous queries. + + LangGraph interrupts BEFORE this node via interrupt_before=["ambiguity_resolution"]. + The graph pauses, the frontend surfaces state["clarifying_questions"] to the user, + and the user submits a clarification via graph.update_state({"feedback": ""}). + + When resumed this node: + 1. Reads the user’s clarification from state["feedback"] + (injected by update_state before resume). + 2. Increments ambiguity_retry_count so detect_ambiguity can enforce a loop guard. + 3. Resets ambiguity state fields so the retry gets a clean pass. + 4. Returns directly → schema_explorer (targeted retry, not a full extractor reset). + + If the user provides no feedback (empty string), the clarifying question is + preserved in feedback so schema_explorer at minimum knows the context. + """ + thread_id = ( + config.get("configurable", {}).get("thread_id", "") if config else "" + ) + await publish_node_event(thread_id, "ambiguity_resolution") + + retry_count = (state.get("ambiguity_retry_count") or 0) + 1 + clarifying_questions = state.get("clarifying_questions") or "" + user_feedback = state.get("feedback") or "" + + # If the human didn’t set feedback via update_state, carry the original + # clarifying question forward so schema_explorer has context for its retry. + if not user_feedback and clarifying_questions: + user_feedback = f"[Previous ambiguity question] {clarifying_questions}" + + try: + if langfuse_client.get_current_trace_id(): + langfuse_client.update_current_span( + metadata={ + "ambiguity_retry_count": retry_count, + "user_clarification": user_feedback, + } + ) + except Exception: + pass + + new_query = state.get("user_query", "") + if user_feedback and not user_feedback.startswith("[Previous"): + new_query += f"\n[User Clarification: {user_feedback}]" + + return { + # Update the main query so all subsequent nodes see the unified intent. + "user_query": new_query, + # Clear ambiguity decision so detect_ambiguity re-evaluates cleanly on retry. + "ambiguity_type": None, + "ambiguity_result": None, + "clarifying_questions": None, + # Clear feedback so it doesn't accidentally trigger the rejection_router later. + "feedback": None, + "ambiguity_retry_count": retry_count, + "execution_path": ["ambiguity_resolution"], + } diff --git a/agent/src/agent/nodes/refiner.py b/agent/src/agent/nodes/refiner.py index 8fc1beb..a8ea42d 100644 --- a/agent/src/agent/nodes/refiner.py +++ b/agent/src/agent/nodes/refiner.py @@ -44,8 +44,8 @@ async def refiner_node(state: AgentState, config: RunnableConfig | None = None): if satisfaction_failures: success = False - trino_error = "; ".join(satisfaction_failures) - error_history.append(f"Satisfaction Check Failed: {trino_error}") + trino_error = "\n".join([f"• {f}" for f in satisfaction_failures]) + error_history.append(f"Satisfaction Check Failed:\n{trino_error}") result = None # Clear satisfaction failures so next pass can execute cleanly # Note: LangGraph state updates require explicitly passing None or handling it if merging diff --git a/agent/src/agent/nodes/satisfaction_check.py b/agent/src/agent/nodes/satisfaction_check.py index 9f1b728..506a780 100644 --- a/agent/src/agent/nodes/satisfaction_check.py +++ b/agent/src/agent/nodes/satisfaction_check.py @@ -105,7 +105,8 @@ async def satisfaction_check_node(state: AgentState, config: RunnableConfig | No prompt = ( f"User question: {state.get('user_query', '')}\n" f"SQL column headers returned: {', '.join(columns)}\n\n" - "Do these column headers conceptually satisfy what the user asked for?" + "Do these column headers conceptually satisfy what the user asked for?\n" + "CRITICAL INSTRUCTION: Do NOT be overly pedantic. If the user explicitly asks for a single attribute (e.g. 'What is the key...'), returning ONLY that attribute's column is 100% correct. You do NOT need to return filter columns (like the comment or ID used in the WHERE clause) just to 'prove' the answer." ) try: structured = llm.with_structured_output(ColumnCoverageOutput, method="json_schema") @@ -125,7 +126,8 @@ async def satisfaction_check_node(state: AgentState, config: RunnableConfig | No f"User question: {state.get('user_query', '')}\n" f"SQL generated: {state.get('sql_query', '')}\n" f"Result column headers: {', '.join(columns)}\n\n" - "Score alignment between the question intent and the query output schema (0.0–1.0)." + "Score alignment between the question intent and the query output schema (0.0–1.0).\n" + "CRITICAL INSTRUCTION: Do NOT penalize the score if the query returns exactly what was asked for without additional context columns. If the user asks for 'the key', returning just the 'key' column is a perfect 1.0 score. Do not demand verification columns." ) try: structured = llm.with_structured_output(SemanticAlignmentOutput, method="json_schema") @@ -166,11 +168,12 @@ async def satisfaction_check_node(state: AgentState, config: RunnableConfig | No } if failures: - partial["last_error"] = "; ".join(failures) + partial["last_error"] = "\n".join([f"• {f}" for f in failures]) if fail_count >= settings.SATISFACTION_MAX_FAILURES: + failures_str = "\n".join([f" - {f}" for f in failures]) partial["escalation_reason"] = ( - f"Satisfaction checks failed {fail_count} times. " - f"Last failures: {'; '.join(failures)}" + f"Satisfaction checks failed {fail_count} times.\n" + f"Last failures:\n{failures_str}" ) return partial diff --git a/agent/src/agent/nodes/schema_explorer.py b/agent/src/agent/nodes/schema_explorer.py index ca1dab3..19a183d 100644 --- a/agent/src/agent/nodes/schema_explorer.py +++ b/agent/src/agent/nodes/schema_explorer.py @@ -1,6 +1,5 @@ from __future__ import annotations import asyncio -from langgraph.types import interrupt import json import re import urllib.request @@ -29,7 +28,6 @@ from agent.utils.schema_enrichment import ( run_semantic_typing, run_join_graph, - run_ambiguity_detection, ) from core.cache import get_cache_service from core.embeddings import get_embedding @@ -41,47 +39,6 @@ # Cache singleton _cache = get_cache_service() -async def _resolve_ambiguity( - data: SchemaExplorerOutput, - chain, - tables_info: list, - profiles_json_str: str, - human_message: str, - state: AgentState, -) -> SchemaExplorerOutput: - """Helper to handle ambiguity detection and interactive resolution.""" - if ( - data.ambiguity_detected - and data.ambiguity_message - and not state.get("non_interactive") - ): - user_choice = interrupt( - { - "type": "schema_explorer_ambiguity", - "message": data.ambiguity_message, - "options": data.candidate_options, - } - ) - - clarified_message = f"{human_message}\nSelected table/option: {user_choice}" - try: - return await chain.ainvoke( - { - "tables_json": json.dumps(tables_info, indent=2), - "profiles_json": profiles_json_str, - "human_message": clarified_message, - } - ) - except Exception as e: - logger.error(f"Structured output parsing failed in schema explorer after clarification: {e}") - return SchemaExplorerOutput( - schema_plan=None, - ambiguity_detected=False, - ambiguity_message="", - candidate_options=[], - ) - return data - # Skill Registry from agent.utils.skill_registry import SkillRegistry from python_core_utils.redis import get_redis_client @@ -136,21 +93,12 @@ def _build_column_context(cp: "ColumnProfile") -> dict: # Define standardized Schema Explorer Output Type +# Ambiguity detection is handled exclusively by the detect_ambiguity node in the +# refiner subgraph — schema_explorer only produces a query plan and table list. class SchemaExplorerOutput(BaseModel): - schema_plan: Optional[Any] = Field( - default=None, - description="Detailed query plan describing tables, columns, and joins.", - ) - ambiguity_detected: bool = Field( - default=False, description="Set to true if there is table selection ambiguity." - ) - ambiguity_message: str = Field( + schema_plan: str = Field( default="", - description="A question to ask the user to clarify/select the right table(s). Must be empty if ambiguity_detected is false.", - ) - candidate_options: List[str] = Field( - default_factory=list, - description="List of strings (table names or options) for the user to choose from. Must be empty if ambiguity_detected is false.", + description="Detailed query plan describing tables, columns, and joins. Must be a detailed string explanation.", ) tables_used: List[str] = Field( default_factory=list, @@ -395,9 +343,6 @@ def _parse_bool_flag(value) -> bool: schema_summarization = _parse_bool_flag( runtime_flags.get("SCHEMA_SUMMARIZATION", settings.ENABLE_SCHEMA_SUMMARIZATION) ) - schema_ambiguity_detect = _parse_bool_flag( - runtime_flags.get("SCHEMA_AMBIGUITY_DETECT", settings.ENABLE_AMBIGUITY_DETECT) - ) schema_skill_injection = _parse_bool_flag( runtime_flags.get("SCHEMA_SKILL_INJECTION", settings.ENABLE_SKILL_INJECTION) ) @@ -540,15 +485,6 @@ async def fetch_profile(t_id, t_name): except Exception as exc: logger.warning("SCHEMA_SUMMARIZATION phase failed: %s", exc) - # Phase D: Ambiguity Detection - if schema_ambiguity_detect and profile_details: - try: - notes = await run_ambiguity_detection(profile_details, user_query, _llm) - if notes: - human_message += "\n\n[AMBIGUITY NOTES]\n" + "\n".join(f"- {n}" for n in notes) - active_phases.append("SCHEMA_AMBIGUITY_DETECT") - except Exception as exc: - logger.warning("SCHEMA_AMBIGUITY_DETECT phase failed: %s", exc) # ── Langfuse trace metadata ─────────────────────────────────────────────── try: @@ -588,24 +524,11 @@ async def fetch_profile(t_id, t_name): ) except Exception as e: print(f"Structured output parsing failed in schema explorer: {e}") - data = SchemaExplorerOutput( - schema_plan=None, - ambiguity_detected=False, - ambiguity_message="", - candidate_options=[], - ) + data = SchemaExplorerOutput(schema_plan=None) - data = await _resolve_ambiguity(data, chain, tables_info, profiles_json_str, human_message, state) - - plan = data.schema_plan - if plan is not None and not isinstance(plan, str): - plan = json.dumps(plan) - elif plan is None: - plan = "" + plan = data.schema_plan or "" tables_used = getattr(data, "tables_used", []) - if tables_used: - state["tables_used"] = tables_used result_state: dict = {"schema_plan": plan, "tables_used": tables_used} result_state["execution_path"] = ["schema_explorer"] diff --git a/agent/src/agent/state.py b/agent/src/agent/state.py index f0db459..e5bbe6c 100644 --- a/agent/src/agent/state.py +++ b/agent/src/agent/state.py @@ -1,4 +1,4 @@ -from typing import Annotated, TypedDict, Any, Literal +from typing import Annotated, TypedDict, Any, Literal, Optional from langchain_core.messages import BaseMessage from langgraph.graph.message import add_messages import operator @@ -45,3 +45,14 @@ class AgentState(TypedDict): runtime_flags: dict[str, Any] | None # resolved by init_flags_node # Enriched table profiles — populated by schema_explorer for reuse by refiner table_profiles: list[dict[str, Any]] | None + # ── Ambiguity Detection (detect_ambiguity node) ─────────────────────────── + # Raw parsed JSON output from the detect_ambiguity LLM call. + ambiguity_result: Optional[dict] | None + # Drives the conditional edge: "clear" | "ambiguous" | "unanswerable" + ambiguity_type: Optional[Literal["clear", "ambiguous", "unanswerable"]] | None + # Populated on "ambiguous" terminal — the clarifying question shown to the user. + clarifying_questions: str | None + # Populated on "unanswerable" terminal. + failure_reason: str | None + # Counts how many times ambiguity_resolution has been attempted (loop guard). + ambiguity_retry_count: int | None diff --git a/agent/src/agent/utils/flag_bridge.py b/agent/src/agent/utils/flag_bridge.py index fef6998..0bbf65a 100644 --- a/agent/src/agent/utils/flag_bridge.py +++ b/agent/src/agent/utils/flag_bridge.py @@ -37,7 +37,6 @@ "SCHEMA_SEMANTIC_TYPING": settings.ENABLE_SEMANTIC_TYPING, "SCHEMA_JOIN_GRAPH": settings.ENABLE_JOIN_GRAPH, "SCHEMA_SUMMARIZATION": settings.ENABLE_SCHEMA_SUMMARIZATION, - "SCHEMA_AMBIGUITY_DETECT": settings.ENABLE_AMBIGUITY_DETECT, "SCHEMA_EXPLORER_MODEL": settings.LLM_MODEL, "SCHEMA_TOP_K_JOINS": 5, # Query Builder diff --git a/agent/src/agent/utils/schema_enrichment.py b/agent/src/agent/utils/schema_enrichment.py index cbf0759..7e4745e 100644 --- a/agent/src/agent/utils/schema_enrichment.py +++ b/agent/src/agent/utils/schema_enrichment.py @@ -1,14 +1,17 @@ """ G2-03: Advanced Schema Explorer — Enrichment Phases ==================================================== -Four independently feature-gated async functions that enrich schema context +Three independently feature-gated async functions that enrich schema context before the LLM planning call in schema_explorer_node. Phase constants (used in Langfuse trace metadata): SCHEMA_SEMANTIC_TYPING SCHEMA_JOIN_GRAPH SCHEMA_SUMMARIZATION - SCHEMA_AMBIGUITY_DETECT + +Note: Ambiguity detection (formerly Phase D) has been removed from schema +enrichment and consolidated into the dedicated detect_ambiguity node that +runs post-SQL in the refiner subgraph. Join-graph algorithm -------------------- @@ -58,16 +61,6 @@ class SummarizationOutput(BaseModel): ) -class AmbiguityOutput(BaseModel): - ambiguity_notes: list[str] = Field( - default_factory=list, - description=( - "List of ambiguity notes for column/table names relative to the user query. " - "Empty list if nothing is ambiguous." - ), - ) - - class ColumnCoverageOutput(BaseModel): """Used by G2-04 satisfaction check (imported from here for reuse).""" @@ -291,7 +284,7 @@ async def run_join_graph( return json.dumps(paths, indent=2) -# ─── Phase C: Schema Summarization ─────────────────────────────────────────── +# ─── Phase C: Schema Summarization ───────────────────────────────────── async def run_schema_summarization( @@ -324,58 +317,16 @@ async def _summarize_one(p: dict[str, Any]) -> str: ) try: structured = llm.with_structured_output(SummarizationOutput, method="json_schema") - # Handle both Pydantic model and raw dict responses gracefully result = await structured.ainvoke(prompt) - - # Check if it's a dict or object if isinstance(result, dict): summary_text = result.get("summary", "(summarization unavailable)") else: summary_text = getattr(result, "summary", "(summarization unavailable)") - return f"[{tname}] {summary_text}" except Exception as exc: logger.warning("run_schema_summarization failed for %s: %s", tname, exc) - return f"[{tname}] (summarization unavailable)" + return f"[{tname}] (summarization unavailable) " tasks = [_summarize_one(p) for p in profiles] summaries = list(await asyncio.gather(*tasks)) return summaries - - -# ─── Phase D: Ambiguity Detection ──────────────────────────────────────────── - - -async def run_ambiguity_detection( - profiles: list[dict[str, Any]], - user_query: str, - llm: Any, -) -> list[str]: - """ - Identify any column or table name ambiguities relative to the user query. - Returns a list of human-readable ambiguity notes (may be empty). - """ - if not profiles: - return [] - - col_names = [] - for p in profiles: - for col in p.get("columns", []): - col_names.append(f"{p.get('table_name','')}.{col['name']}") - - prompt = ( - f"User question: {user_query}\n" - f"Available columns: {', '.join(col_names[:80])}\n\n" - "List any ambiguous column or table names that could be misinterpreted for this question. " - "Return an empty list if nothing is ambiguous." - ) - try: - structured = llm.with_structured_output(AmbiguityOutput, method="json_schema") - result = await structured.ainvoke(prompt) - # Handle both Pydantic model and raw dict responses - if isinstance(result, dict): - return result.get("ambiguity_notes", []) - return result.ambiguity_notes - except Exception as exc: - logger.warning("run_ambiguity_detection failed: %s", exc) - return [] diff --git a/backend/app/routers/agent.py b/backend/app/routers/agent.py index 6f133a7..6994420 100644 --- a/backend/app/routers/agent.py +++ b/backend/app/routers/agent.py @@ -66,6 +66,7 @@ class ChatResponse(BaseModel): schema_plan: str | None = None trace_id: str | None = None execution_path: list[str] | None = None + is_unanswerable: bool = False # --------------------------------------------------------------------------- @@ -260,7 +261,7 @@ async def get_trace_timeline(trace_id: str): auth = (settings.LANGFUSE_PUBLIC_KEY, settings.LANGFUSE_SECRET_KEY) url = f"{settings.LANGFUSE_HOST}/api/public/traces/{trace_id}" - async with httpx.AsyncClient() as client: + async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.get(url, auth=auth) if resp.status_code != 200: if resp.status_code == 404: diff --git a/frontend/src/api/agent.ts b/frontend/src/api/agent.ts index 5822eb1..98b0912 100644 --- a/frontend/src/api/agent.ts +++ b/frontend/src/api/agent.ts @@ -45,6 +45,7 @@ export interface ChatResponse { schema_plan?: string; trace_id?: string; execution_path?: string[]; + is_unanswerable?: boolean; } export const agentApi = { diff --git a/frontend/src/pages/AgentTestingPage.tsx b/frontend/src/pages/AgentTestingPage.tsx index 25110e5..1208389 100644 --- a/frontend/src/pages/AgentTestingPage.tsx +++ b/frontend/src/pages/AgentTestingPage.tsx @@ -787,15 +787,36 @@ const AgentResultDisplay = ({ transition={{ duration: 0.4 }} >
-
- +
+ {chatResponse.is_unanswerable ? ( + + ) : ( + + )}
- TASK COMPLETED -

Agent Execution Successful

+ + {chatResponse.is_unanswerable ? 'UNANSWERABLE' : 'TASK COMPLETED'} + +

+ {chatResponse.is_unanswerable ? 'Agent Could Not Answer' : 'Agent Execution Successful'} +

- - Done + + {chatResponse.is_unanswerable ? 'Failed' : 'Done'}
@@ -1006,7 +1027,7 @@ export function AgentTestingPage() { const handleSubmit = () => { if (!query) return; setChatResponse(null); - const newThreadId = threadId || uuidv4(); + const newThreadId = uuidv4(); setThreadId(newThreadId); setExecutionPath([]); setTraceId(null); From 4952fc75b3c99d018c20f8c830f8d9e5c5158373 Mon Sep 17 00:00:00 2001 From: yuvalkh Date: Tue, 14 Jul 2026 16:13:29 +0300 Subject: [PATCH 2/4] fixed CR --- agent/src/agent/nodes/detect_ambiguity.py | 28 ++++++++++++----------- agent/src/agent/nodes/schema_explorer.py | 2 +- 2 files changed, 16 insertions(+), 14 deletions(-) diff --git a/agent/src/agent/nodes/detect_ambiguity.py b/agent/src/agent/nodes/detect_ambiguity.py index c7b6b46..4cd6d8c 100644 --- a/agent/src/agent/nodes/detect_ambiguity.py +++ b/agent/src/agent/nodes/detect_ambiguity.py @@ -50,30 +50,29 @@ logger = logging.getLogger(__name__) +from typing import Literal def _resolve_ambiguity_type(parsed: dict) -> str: + ambiguity_type = parsed.get("ambiguity_type") clarifying: str | None = parsed.get("clarifying_questions") - reason: str = parsed.get("reason") or "" - is_ambiguous = parsed.get("is_ambiguous", False) # If the LLM generated a clarifying question (and it's not null/empty), it MUST be ambiguous - # regardless of whether it correctly flipped the boolean flag. + # regardless of whether it accidentally classified it as unanswerable. if clarifying is not None and clarifying.strip(): return "ambiguous" - # If flagged as ambiguous but no question provided - if is_ambiguous: - if "unanswer" in reason.lower(): - return "unanswerable" - return "ambiguous" + if ambiguity_type in ["clear", "ambiguous", "unanswerable"]: + return ambiguity_type return "clear" class AmbiguityResult(BaseModel): - is_ambiguous: bool = Field(description="True if the request is ambiguous or unanswerable, False if clear") + ambiguity_type: Literal["clear", "ambiguous", "unanswerable"] = Field( + description="The determined state of the query: 'clear' (proceed), 'ambiguous' (needs clarification), or 'unanswerable' (impossible)." + ) reason: str = Field(default="", description="Explanation of the ambiguity detection decision") - clarifying_questions: str | None = Field(default=None, description="Questions to ask the user if ambiguous. Null if clear.") + clarifying_questions: str | None = Field(default=None, description="Questions to ask the user if ambiguous. Null if clear or unanswerable.") async def detect_ambiguity_node( @@ -153,14 +152,17 @@ async def detect_ambiguity_node( parsed = response.model_dump() ambiguity_type = _resolve_ambiguity_type(parsed) except Exception as exc: - logger.error("detect_ambiguity: unexpected error parsing structured output: %s", exc) + logger.error(f"Ambiguity detection structured output failed: {exc}") + # Fallback to ambiguous on parsing failure to be safe parsed = { - "is_ambiguous": True, + "ambiguity_type": "ambiguous", "intent_deconstruction": "", + "active_schema_search": "", + "ambiguity_check": "", "logical_sql_plan": "", "schema_alignment_check": "", "reason": f"Ambiguity detection failed to parse LLM structured output: {exc}", - "clarifying_questions": "", + "clarifying_questions": "Could you rephrase or add more detail to your request so we can interpret it precisely?", } ambiguity_type = "ambiguous" diff --git a/agent/src/agent/nodes/schema_explorer.py b/agent/src/agent/nodes/schema_explorer.py index 19a183d..03f0177 100644 --- a/agent/src/agent/nodes/schema_explorer.py +++ b/agent/src/agent/nodes/schema_explorer.py @@ -524,7 +524,7 @@ async def fetch_profile(t_id, t_name): ) except Exception as e: print(f"Structured output parsing failed in schema explorer: {e}") - data = SchemaExplorerOutput(schema_plan=None) + data = SchemaExplorerOutput(schema_plan="") plan = data.schema_plan or "" From beeda322afd2916dfd23bd6c0d2dcc39e59614f9 Mon Sep 17 00:00:00 2001 From: yuvalkh Date: Tue, 14 Jul 2026 16:33:55 +0300 Subject: [PATCH 3/4] removed deprecated fields in default ambiguity error --- agent/src/agent/nodes/detect_ambiguity.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/agent/src/agent/nodes/detect_ambiguity.py b/agent/src/agent/nodes/detect_ambiguity.py index 4cd6d8c..f01e61b 100644 --- a/agent/src/agent/nodes/detect_ambiguity.py +++ b/agent/src/agent/nodes/detect_ambiguity.py @@ -156,11 +156,6 @@ async def detect_ambiguity_node( # Fallback to ambiguous on parsing failure to be safe parsed = { "ambiguity_type": "ambiguous", - "intent_deconstruction": "", - "active_schema_search": "", - "ambiguity_check": "", - "logical_sql_plan": "", - "schema_alignment_check": "", "reason": f"Ambiguity detection failed to parse LLM structured output: {exc}", "clarifying_questions": "Could you rephrase or add more detail to your request so we can interpret it precisely?", } From b880a848ac8404b3156a5a9f1348382cdbd4dcd3 Mon Sep 17 00:00:00 2001 From: yuvalkh Date: Tue, 14 Jul 2026 17:35:37 +0300 Subject: [PATCH 4/4] readd the flag of ambiguity check --- agent/src/agent/config.py | 1 + agent/src/agent/graph.py | 13 ++++++++++--- agent/src/agent/utils/flag_bridge.py | 1 + 3 files changed, 12 insertions(+), 3 deletions(-) diff --git a/agent/src/agent/config.py b/agent/src/agent/config.py index dae2d93..5dbad1d 100644 --- a/agent/src/agent/config.py +++ b/agent/src/agent/config.py @@ -69,6 +69,7 @@ class AgentSettings(BaseSettings): SATISFACTION_MAX_FAILURES: int = 2 # escalate to HITL after this many check failures # ── Ambiguity Resolution ────────────────────────────────────────────────── + ENABLE_AMBIGUITY_DETECT: bool = True MAX_AMBIGUITY_RETRIES: int = 2 # max times user can clarify before hard stop # ── G2-05: Redis Schema Cache ───────────────────────────────────────────── diff --git a/agent/src/agent/graph.py b/agent/src/agent/graph.py index f58ed04..3acc6ad 100644 --- a/agent/src/agent/graph.py +++ b/agent/src/agent/graph.py @@ -176,6 +176,14 @@ def route_schema_explorer(state: AgentState) -> str: if (state.get("schema_explorer_retry_count") or 0) >= MAX_SCHEMA_RETRIES: return "hitl_escalation" return "schema_explorer" + + runtime_flags = state.get("runtime_flags") or {} + enable_ambiguity = runtime_flags.get("SCHEMA_AMBIGUITY_DETECT", settings.ENABLE_AMBIGUITY_DETECT) + if isinstance(enable_ambiguity, str): + enable_ambiguity = enable_ambiguity.lower() == "true" + + if enable_ambiguity: + return "detect_ambiguity" return "query_builder" @@ -258,9 +266,8 @@ def route_rejection(state: AgentState) -> str: route_schema_explorer, { "schema_explorer": "schema_explorer", - # route_schema_explorer returns "query_builder" on success; - # we intercept it here and send to detect_ambiguity first. - "query_builder": "detect_ambiguity", + "detect_ambiguity": "detect_ambiguity", + "query_builder": "query_builder", "hitl_escalation": "hitl_escalation", # G2-02 }, ) diff --git a/agent/src/agent/utils/flag_bridge.py b/agent/src/agent/utils/flag_bridge.py index 0bbf65a..fef6998 100644 --- a/agent/src/agent/utils/flag_bridge.py +++ b/agent/src/agent/utils/flag_bridge.py @@ -37,6 +37,7 @@ "SCHEMA_SEMANTIC_TYPING": settings.ENABLE_SEMANTIC_TYPING, "SCHEMA_JOIN_GRAPH": settings.ENABLE_JOIN_GRAPH, "SCHEMA_SUMMARIZATION": settings.ENABLE_SCHEMA_SUMMARIZATION, + "SCHEMA_AMBIGUITY_DETECT": settings.ENABLE_AMBIGUITY_DETECT, "SCHEMA_EXPLORER_MODEL": settings.LLM_MODEL, "SCHEMA_TOP_K_JOINS": 5, # Query Builder