Spaces:
Running
Running
File size: 5,358 Bytes
1721aea |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 |
"""
LLM assistance functions for Regression Discontinuity Design (RDD).
"""
from typing import List, Dict, Any, Optional
import logging
# Imported for type hinting
from langchain.chat_models.base import BaseChatModel
# Import shared LLM helpers
from auto_causal.utils.llm_helpers import call_llm_with_json_output
logger = logging.getLogger(__name__)
def suggest_rdd_parameters(
df_cols: List[str],
query: str,
llm: Optional[BaseChatModel] = None
) -> Dict[str, Any]:
"""
(Placeholder) Use LLM to suggest RDD parameters (running variable, cutoff).
Args:
df_cols: List of available column names.
query: User's causal query text.
llm: Optional LLM model instance.
Returns:
Dictionary containing suggested 'running_variable' and 'cutoff', or empty.
"""
logger.info("LLM RDD parameter suggestion is not implemented yet.")
if llm:
# Placeholder: Analyze columns, distributions, query for potential
# running variables (e.g., 'score', 'age') and cutoffs (e.g., 50, 65).
pass
return {}
def interpret_rdd_results(
results: Dict[str, Any],
diagnostics: Optional[Dict[str, Any]],
llm: Optional[BaseChatModel] = None
) -> str:
"""
Use LLM to interpret Regression Discontinuity Design (RDD) results.
Args:
results: Dictionary of estimation results from the RDD estimator.
diagnostics: Dictionary of diagnostic test results.
llm: Optional LLM model instance.
Returns:
String containing natural language interpretation.
"""
default_interpretation = "LLM interpretation not available for RDD."
if llm is None:
logger.info("LLM not provided for RDD interpretation.")
return default_interpretation
try:
# --- Prepare summary for LLM ---
results_summary = {}
effect = results.get('effect_estimate')
p_val = results.get('p_value')
ci = results.get('confidence_interval')
results_summary['Method Used'] = results.get('method_used', 'RDD')
results_summary['Effect Estimate'] = f"{effect:.3f}" if isinstance(effect, (int, float)) else str(effect)
results_summary['P-value'] = f"{p_val:.3f}" if isinstance(p_val, (int, float)) else str(p_val)
if isinstance(ci, (list, tuple)) and len(ci) == 2:
results_summary['Confidence Interval'] = f"[{ci[0]:.3f}, {ci[1]:.3f}]"
else:
results_summary['Confidence Interval'] = str(ci) if ci is not None else "N/A"
diag_summary = {}
if diagnostics and diagnostics.get("status", "").startswith("Success"):
diag_details = diagnostics.get("details", {})
diag_summary['Covariate Balance Status'] = "Checked" if 'covariate_balance' in diag_details else "Not Checked"
if isinstance(diag_details.get('covariate_balance'), dict):
num_unbalanced = sum(1 for cov, res in diag_details['covariate_balance'].items() if isinstance(res, dict) and res.get('balanced', '').startswith("No"))
diag_summary['Number of Unbalanced Covariates (p<=0.05)'] = num_unbalanced
diag_summary['Density Continuity Test'] = diag_details.get('continuity_density_test', 'N/A')
diag_summary['Visual Inspection Recommended'] = "Yes" if 'visual_inspection' in diag_details else "No"
elif diagnostics:
diag_summary['Status'] = diagnostics.get("status", "Unknown")
if "error" in diagnostics:
diag_summary['Error'] = diagnostics["error"]
else:
diag_summary['Status'] = "Diagnostics not available or failed."
# --- Construct Prompt ---
prompt = f"""
You are assisting with interpreting Regression Discontinuity Design (RDD) results.
Estimation Results Summary:
{results_summary}
Diagnostics Summary:
{diag_summary}
Explain these RDD results in 2-4 concise sentences. Focus on:
1. The estimated causal effect at the cutoff (magnitude, direction, statistical significance based on p-value < 0.05, if available).
2. Key diagnostic findings (specifically mention covariate balance issues if present, and note that other checks like density continuity were not performed).
3. Mention that visual inspection of the running variable vs outcome is recommended.
Return ONLY a valid JSON object with the following structure (no explanations or surrounding text):
{{
"interpretation": "<your concise interpretation text>"
}}
"""
# --- Call LLM ---
response = call_llm_with_json_output(llm, prompt)
# --- Process Response ---
if response and isinstance(response, dict) and \
"interpretation" in response and isinstance(response["interpretation"], str):
return response["interpretation"]
else:
logger.warning(f"Failed to get valid interpretation from LLM for RDD. Response: {response}")
return default_interpretation
except Exception as e:
logger.error(f"Error during LLM interpretation for RDD: {e}")
return f"Error generating interpretation: {e}"
|