Gestione delle domande ambigue sui database

Il problema dell'ambiguità in NL-to-SQL

Tipi di ambiguità

# Examples of each type
ambiguous_examples = {
    'temporal': 'Show me recent orders',         # last 7 days? 30 days? 1 year?
    'scope':    'Who are the top customers?',     # by revenue, order count, or recency?
    'table':    'Show me user activity',          # from users, sessions, or audit_log?
    'filter':   'List all active products'        # active = in_stock? not discontinued?
}

if __name__ == '__main__':
    print('Types of ambiguous questions:')
    for kind, example in ambiguous_examples.items():
        print(f'  {kind}: "{example}"')

Rilevamento dell'ambiguità con l'LLM

import json

AMBIGUITY_CHECK_PROMPT = '''You are a SQL assistant. Given a database schema and a user question,
determine if the question is clear enough to write a single correct SQL query.

Schema:
{schema}

Question: {question}

Respond with JSON:
- If clear: {{"ambiguous": false, "sql": "SELECT ..."}}
- If ambiguous: {{"ambiguous": true, "clarification": "What time range counts as recent?",
  "options": ["Last 7 days", "Last 30 days", "Last 90 days"]}}

JSON:'''

def check_and_generate(question, schema):
    response = llm_call(AMBIGUITY_CHECK_PROMPT.format(
        schema=schema, question=question
    ))
    return json.loads(response)

Il ciclo di chiarimento

def nl_to_sql_with_clarification(user_question, schema, conn, ask_user_fn):
    for attempt in range(2):  # max 2 clarification rounds
        result = check_and_generate(user_question, schema)

        if not result.get('ambiguous'):
            # Clear question — execute
            rows = execute_query(conn, result['sql'])
            return format_results(rows, user_question)

        # Ambiguous — ask user
        clarification = result['clarification']
        options = result.get('options', [])
        user_reply = ask_user_fn(clarification, options)

        # Enrich the question with the answer
        user_question = f'{user_question} ({clarification}: {user_reply})'
        print(f'Enriched question: {user_question}')

    # After 2 rounds, generate with best guess
    return check_and_generate(user_question, schema)

Disambiguazione temporale di «recenti»

from datetime import datetime, timedelta

TEMPORAL_DEFAULTS = {
    'recent':   7,    # days
    'latest':   1,    # days
    'new':      30,
    'old':      365,
    'this week': 7,
    'this month': 30,
    'this year': 365
}

def resolve_temporal(question):
    lower = question.lower()
    for term, days in TEMPORAL_DEFAULTS.items():
        if term in lower:
            since = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
            return question + f" ('{term}' means since {since})"
    return question

print(resolve_temporal('Show me recent orders'))
# Show me recent orders ('recent' means since 2024-05-22)

Disambiguazione dello schema: più tabelle

TABLE_SEMANTIC_MAP = {
    'user activity':    ['sessions', 'user_events', 'audit_log'],
    'purchases':        ['orders', 'transactions', 'invoices'],
    'product catalog':  ['products', 'items', 'listings'],
    'sign-ups':         ['users', 'registrations', 'accounts']
}

def disambiguate_tables(question, schema_dict, ask_user_fn):
    lower = question.lower()
    for concept, tables in TABLE_SEMANTIC_MAP.items():
        if concept in lower:
            available = [t for t in tables if t in schema_dict]
            if len(available) > 1:
                chosen = ask_user_fn(
                    f"Which table should I query for '{concept}'?",
                    available
                )
                return question + f" (use the {chosen} table)"
    return question

if __name__ == '__main__':
    def ask_user_fn(prompt, options):
        print(f'{prompt} -> choosing "{options[0]}" (demo default)')
        return options[0]

    demo_schema = {'sessions': [], 'transactions': [], 'orders': []}
    resolved = disambiguate_tables('Show me user activity', demo_schema, ask_user_fn)
    print('Resolved question:', resolved)

Gestione dell'ambiguità dell'ambito «Top N»

RANKING_AMBIGUITY_PROMPT = '''The question asks for a ranking but the metric is unclear.

Question: {question}
Table columns available: {columns}

List 2-3 reasonable ranking metrics as a JSON array of objects:
[{{"label": "By total revenue", "sql_expr": "SUM(total) DESC"}},
 {{"label": "By order count", "sql_expr": "COUNT(*) DESC"}}]

JSON:'''

def resolve_ranking(question, columns, ask_user_fn):
    import json
    response = llm_call(RANKING_AMBIGUITY_PROMPT.format(
        question=question, columns=columns
    ))
    options = json.loads(response)
    labels = [o['label'] for o in options]
    chosen_label = ask_user_fn('How should I rank the results?', labels)
    chosen = next(o for o in options if o['label'] == chosen_label)
    return question + f" (rank by: {chosen['sql_expr']})", chosen['sql_expr']

Proposta di valori predefiniti intelligenti

def nl_to_sql_with_assumptions(question, schema, conn):
    # Resolve common ambiguities with defaults
    enriched = resolve_temporal(question)
    result = check_and_generate(enriched, schema)

    if result.get('ambiguous'):
        # Still ambiguous — pick default option
        options = result.get('options', ['the most common interpretation'])
        default = options[0]
        enriched = enriched + f' ({result["clarification"]}: {default})'
        result = check_and_generate(enriched, schema)

    rows = execute_query(conn, result['sql'])
    answer = format_results(rows, question)

    # Append assumption note
    if enriched != question:
        assumption = enriched[len(question):].strip().strip('()')
        answer += f'\n\n[Note: I assumed {assumption}]'

    return answer

Ambiguità a livello di colonna

COLUMN_DESCRIPTIONS = {
    ('orders', 'created_at'):   'When the order was placed',
    ('orders', 'updated_at'):   'When the order was last modified',
    ('orders', 'shipped_at'):   'When the order was shipped to customer',
    ('orders', 'delivery_date'): 'Expected or actual delivery date'
}

def format_columns_with_descriptions(table, columns):
    parts = []
    for col in columns:
        desc = COLUMN_DESCRIPTIONS.get((table, col['name']), '')
        label = f"{col['name']} ({col['type']})"
        if desc:
            label += f' [{desc}]'
        parts.append(label)
    return ', '.join(parts)

if __name__ == '__main__':
    demo_columns = [
        {'name': 'created_at', 'type': 'timestamp'},
        {'name': 'shipped_at', 'type': 'timestamp'},
        {'name': 'total', 'type': 'float'},
    ]
    print(format_columns_with_descriptions('orders', demo_columns))

Registrazione delle risoluzioni dell'ambiguità

import json
from datetime import datetime

ambiguity_log = []

def log_ambiguity(original_question, clarification, resolution, method):
    ambiguity_log.append({
        'timestamp': datetime.now().isoformat(),
        'question': original_question,
        'clarification': clarification,
        'resolution': resolution,
        'method': method  # 'asked_user' | 'default' | 'llm_inferred'
    })

# Periodically analyze to improve defaults
def analyze_ambiguity_log():
    from collections import Counter
    types = Counter(entry['clarification'] for entry in ambiguity_log)
    print('Most common ambiguities:')
    for q, count in types.most_common(5):
        print(f'  {count}x: {q}')

if __name__ == '__main__':
    log_ambiguity('Show recent orders', 'temporal', 'last 30 days', 'default')
    log_ambiguity('Show recent orders', 'temporal', 'last 7 days', 'asked_user')
    log_ambiguity('Top customers', 'scope', 'by revenue', 'llm_inferred')
    analyze_ambiguity_log()

Contesto della conversazione su più turni

def nl_to_sql_with_context(question, schema, conn, conversation_history):
    context_prompt = ''
    if conversation_history:
        context_prompt = 'Previous conversation context:\n'
        for turn in conversation_history[-3:]:  # last 3 turns
            context_prompt += f"User: {turn['user']}\n"
            if 'assumption' in turn:
                context_prompt += f"Assumption made: {turn['assumption']}\n"

    full_prompt = context_prompt + f'Current question: {question}'
    result = check_and_generate(full_prompt, schema)

    conversation_history.append({
        'user': question,
        'sql': result.get('sql', ''),
        'assumption': result.get('assumption', '')
    })
    return result

Verifica delle conoscenze

Riepilogo: gestione delle domande ambigue