The Problem

AI agents in multi-agent environments reinforce each other’s initial interpretations without systematic evaluation of alternatives. This leads to:

• False consensus on incorrect conclusions
• Coordinated error propagation
• Poor uncertainty calibration
• Degraded performance under adversarial information

The Solution

Contrastive Inquiry disrupts confirmation bias by requiring agents to generate and systematically evaluate competing explanations before committing to conclusions. Expected outcomes:

• 30-50% reduction in false consensus rates
• Improved confidence calibration
• Enhanced adversity response
• Measurable epistemic updating

pip install openai fastapi uvicorn
export OPENAI_API_KEY="your-api-key-here"

from contrastive_inquiry import contrastive_inquiry
# Example: Evaluating a bug diagnosis
result = contrastive_inquiry(
    conclusion="The bug is caused by improper input validation",
    context="Error occurs when users submit forms with special characters",
    evidence=[
        "Error logs show SQL syntax errors",
        "Users report success with alphanumeric-only inputs",
        "Database uses UTF-8 encoding"
    ]
)
print(result['recommended_conclusion'])
print(f"Confidence: {result['confidence']}")

# README: Contrastive Inquiry Implementation for AI Agents
#
# Overview:
# This Python module implements the Contrastive Inquiry Method from Reasoned Leadership,
# designed to reduce epistemic rigidity and confirmation bias in AI agents. It generates
# alternative hypotheses that contradict an initial conclusion, evaluates them against
# provided evidence, and logs the process for IBOT-style tracking.
#
# Key Features:
# - Modular functions for alternative generation, evidence evaluation, and IBOT logging
# - Error handling for insufficient inputs or API failures
# - Output includes scores, recommendations, confidence levels, and logs
# - Theoretical basis: Disrupts bias by forcing contrastive evaluation
#
# Dependencies:
# - openai (pip install openai)
# - Python 3.8+
# - Set OPENAI_API_KEY environment variable for LLM access
#
# Usage:
# 1. Import the module: from contrastive_inquiry import contrastive_inquiry
# 2. Call the core function with conclusion, context, and evidence list
# 3. For API: Run the FastAPI example at the bottom
#
# Limitations:
# - Relies on LLM quality; test with your model (e.g., gpt-4o)
# - Evidence should be concise strings for accurate scoring
import os
import datetime
import json
from typing import List, Dict, Any
import openai
# Set up OpenAI client
openai.api_key = os.getenv("OPENAI_API_KEY")
if not openai.api_key:
    raise ValueError("OPENAI_API_KEY environment variable not set.")
def generate_alternative_hypothesis(conclusion: str, context: str) -> str:
    """
    Generate a plausible alternative hypothesis that genuinely contradicts
    the initial conclusion.
   
    Args:
        conclusion (str): The initial conclusion or hypothesis.
        context (str): Background context for the scenario.
   
    Returns:
        str: An alternative hypothesis that contradicts substantive claims.
   
    Raises:
        ValueError: If inputs are empty.
        RuntimeError: If LLM API call fails.
    """
    if not conclusion or not context:
        raise ValueError("Conclusion and context must be provided.")
   
    prompt = f"""
    Based on the context: "{context}"
    The initial conclusion is: "{conclusion}"
   
    Generate ONE alternative hypothesis that directly contradicts the substantive
    claims of the initial conclusion. Make it plausible and grounded in the context,
    but ensure it challenges key assumptions or interpretations. Do not contradict
    trivial details; focus on core causal or explanatory elements.
    Output only the alternative hypothesis as a single sentence.
    """
   
    try:
        response = openai.ChatCompletion.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "You are a reasoning assistant for contrastive inquiry."},
                {"role": "user", "content": prompt}
            ]
        )
        alternative = response.choices[0].message.content.strip()
        return alternative
    except Exception as e:
        raise RuntimeError(f"LLM API error: {str(e)}")
def evaluate_evidence_against_hypotheses(hypotheses: List[str], evidence: List[str]) -> Dict[str, float]:
    """
    Evaluate how well each hypothesis fits the provided evidence,
    returning scores on a 0-1 scale.
   
    Args:
        hypotheses (list[str]): List of hypotheses to evaluate (initial and alternative).
        evidence (list[str]): List of evidence strings.
   
    Returns:
        dict[str, float]: Hypothesis as key, average fit score (0-1) as value.
   
    Raises:
        ValueError: If no evidence or hypotheses provided.
        RuntimeError: If LLM API call fails.
    """
    if not hypotheses or not evidence:
        raise ValueError("Hypotheses and evidence must be provided.")
   
    scores = {}
    for hypo in hypotheses:
        prompt = f"""
        For the hypothesis: "{hypo}"
        Evaluate its fit against each piece of evidence below on a scale of 0
        (no fit/contradicts) to 1 (perfect fit/supports).
        Evidence:
        {json.dumps(evidence, indent=2)}
       
        Output as JSON: {{"evidence_scores": [score1, score2, ...]}}
        """
        try:
            response = openai.ChatCompletion.create(
                model="gpt-4o",
                messages=[
                    {"role": "system", "content": "You are an evidence evaluator."},
                    {"role": "user", "content": prompt}
                ]
            )
            result = json.loads(response.choices[0].message.content.strip())
            avg_score = sum(result["evidence_scores"]) / len(result["evidence_scores"]) if result["evidence_scores"] else 0.0
            scores[hypo] = avg_score
        except Exception as e:
            raise RuntimeError(f"LLM API error during evaluation: {str(e)}")
   
    return scores
def determine_confidence_and_recommendation(scores: Dict[str, float], evidence_count: int) -> tuple[str, str]:
    """
    Determine confidence level and recommended conclusion based on scores.
   
    Args:
        scores (dict[str, float]): Hypothesis scores.
        evidence_count (int): Number of evidence items.
   
    Returns:
        tuple[str, str]: (recommended_conclusion, confidence_level)
    """
    # Current design assumes exactly two hypotheses: initial + one alternative
    if evidence_count < 2:
        return "Insufficient evidence to recommend", "uncertain"
   
    # Access scores explicitly by hypothesis to avoid reliance on dict ordering
    hypotheses_list = list(scores.keys())
    initial_score = scores[hypotheses_list[0]]
    alt_score = scores[hypotheses_list[1]]
    diff = abs(initial_score - alt_score)
   
    if diff > 0.3:
        recommended = max(scores, key=scores.get)
        confidence = "high" if max(initial_score, alt_score) > 0.7 else "moderate"
    else:
        recommended = "Both plausible; further evidence needed"
        confidence = "low" if evidence_count < 5 else "moderate"
   
    if max(initial_score, alt_score) < 0.4:
        confidence = "uncertain"
   
    return recommended, confidence
def ibot_log(agent_id: str, initial_hypo: str, alt_hypo: str, evidence: List[str],
             conclusion: str, confidence: str, reasoning: str) -> Dict[str, Any]:
    """
    Generate IBOT-compatible log for longitudinal tracking.
   
    Args:
        agent_id (str): Identifier for the AI agent.
        initial_hypo (str): Initial hypothesis.
        alt_hypo (str): Alternative hypothesis.
        evidence (list[str]): Evidence sources.
        conclusion (str): Recommended conclusion.
        confidence (str): Confidence level.
        reasoning (str): Chain of reasoning.
   
    Returns:
        dict: Log data in JSON-serializable format.
    """
    log = {
        "agent_id": agent_id,
        "timestamp": datetime.datetime.utcnow().isoformat(),
        "initial_hypothesis": initial_hypo,
        "alternative_hypothesis": alt_hypo,
        "evidence_sources": evidence,
        "final_conclusion": conclusion,
        "confidence_level": confidence,
        "reasoning_chain": reasoning
    }
    return log
def contrastive_inquiry(conclusion: str, context: str, evidence: List[str],
                       agent_id: str = "default_agent") -> Dict[str, Any]:
    """
    Core Contrastive Inquiry function to reduce bias in AI decisions.
   
    Args:
        conclusion (str): Initial conclusion.
        context (str): Background context.
        evidence (list[str]): List of evidence strings.
        agent_id (str, optional): Agent identifier for logging.
   
    Returns:
        dict: Structured output with evaluation and log.
   
    Raises:
        ValueError: For invalid inputs.
    """
    if not evidence:
        raise ValueError("At least one piece of evidence required.")
   
    # Generate alternative
    alt_hypo = generate_alternative_hypothesis(conclusion, context)
   
    # Evaluate evidence
    hypotheses = [conclusion, alt_hypo]
    scores = evaluate_evidence_against_hypotheses(hypotheses, evidence)
   
    # Determine recommendation and confidence
    recommended, confidence = determine_confidence_and_recommendation(scores, len(evidence))
   
    # Reasoning chain for log (simple string summary)
    reasoning = f"Initial score: {scores[conclusion]:.2f}, Alt score: {scores[alt_hypo]:.2f}. Diff: {abs(scores[conclusion] - scores[alt_hypo]):.2f}."
   
    # IBOT log
    log = ibot_log(agent_id, conclusion, alt_hypo, evidence, recommended, confidence, reasoning)
   
    return {
        "initial_hypothesis": conclusion,
        "alternative_hypothesis": alt_hypo,
        "evidence_evaluation": scores,
        "recommended_conclusion": recommended,
        "confidence": confidence,
        "epistemic_log": log
    }
# Example Usage Scenarios
# Example 1: Basic call
# input_conclusion = "The project failed due to poor leadership."
# input_context = "Team faced tight deadlines and resource constraints."
# input_evidence = ["Team members reported high stress levels.",
# "Budget was cut midway.",
# "Leader changed strategy multiple times."]
# result = contrastive_inquiry(input_conclusion, input_context, input_evidence)
# print(json.dumps(result, indent=2))
# Example 2: Insufficient evidence
# try:
# contrastive_inquiry("Hypothesis A", "Context", [])
# except ValueError as e:
# print(e) # "At least one piece of evidence required."
# Example 3: Custom agent ID
# result = contrastive_inquiry(
# "AI will replace all jobs.",
# "Economic trends in automation.",
# ["Job growth in tech sectors.", "Historical data on industrial revolutions."],
# agent_id="agent_123"
# )
# print(result["epistemic_log"]["agent_id"]) # "agent_123"
# Integration Example: Wrap into FastAPI endpoint
# Install FastAPI and uvicorn: pip install fastapi uvicorn
# Run: uvicorn contrastive_inquiry:app --reload
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI(title="Contrastive Inquiry API")
class InquiryRequest(BaseModel):
    conclusion: str
    context: str
    evidence: List[str]
    agent_id: str = "default_agent"
@app.post("/contrastive_inquiry")
def api_contrastive_inquiry(request: InquiryRequest):
    try:
        result = contrastive_inquiry(
            request.conclusion,
            request.context,
            request.evidence,
            request.agent_id
        )
        return result
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except RuntimeError as e:
        raise HTTPException(status_code=500, detail=str(e))
# To integrate into an AI agent's pipeline:
# Example: In an agent's decision loop
# def agent_decide(query):
# # ... gather conclusion, context, evidence ...
# ci_result = contrastive_inquiry(conclusion=concl, context=ctx, evidence=evid)
# if ci_result["confidence"] == "uncertain":
# # Fetch more evidence via web search
# pass
# return ci_result["recommended_conclusion"]

from contrastive_inquiry import contrastive_inquiry
import json
# Test Case: Agent observes error in system
test_cases = [
    {
        "conclusion": "The system is slow because of database queries",
        "context": "Users report 3-second delays during peak hours",
        "evidence": [
            "Database query logs show 200ms average response time",
            "Network latency spikes to 2.5 seconds during peak hours",
            "CPU usage remains at 30% during slowdowns"
        ],
        "correct_answer": "Network latency is the primary cause"
    },
    {
        "conclusion": "Login failures are due to incorrect passwords",
        "context": "Users unable to log in after password reset",
        "evidence": [
            "Password reset emails contain correct temporary passwords",
            "Session tokens expire after 10 minutes",
            "Users report delays of 15+ minutes between reset and login attempt"
        ],
        "correct_answer": "Session token expiration is the issue"
    }
]
# Run validation
results = []
for i, test in enumerate(test_cases):
    result = contrastive_inquiry(
        conclusion=test["conclusion"],
        context=test["context"],
        evidence=test["evidence"]
    )
   
    # Check if recommended conclusion aligns with correct answer
    correct = test["correct_answer"].lower() in result["recommended_conclusion"].lower()
   
    results.append({
        "test_case": i + 1,
        "initial_hypothesis": result["initial_hypothesis"],
        "alternative_hypothesis": result["alternative_hypothesis"],
        "recommended_conclusion": result["recommended_conclusion"],
        "confidence": result["confidence"],
        "correct_identification": correct
    })
   
    print(f"\n=== Test Case {i + 1} ===")
    print(f"Initial: {result['initial_hypothesis']}")
    print(f"Alternative: {result['alternative_hypothesis']}")
    print(f"Recommended: {result['recommended_conclusion']}")
    print(f"Confidence: {result['confidence']}")
    print(f"Correct: {correct}")
# Calculate accuracy
accuracy = sum(r["correct_identification"] for r in results) / len(results) * 100
print(f"\n=== Overall Accuracy: {accuracy}% ===")

import random
def run_adversarial_test(num_trials=10):
    """Test agent performance with mixed accurate/misleading evidence"""
   
    # Define test with known ground truth
    ground_truth = "Market decline caused by interest rate hikes"
   
    accurate_evidence = [
        "Federal Reserve raised rates by 0.75%",
        "Bond yields increased sharply",
        "Historical correlation between rate hikes and market corrections"
    ]
   
    misleading_evidence = [
        "Social media sentiment turned negative",
        "One tech CEO made pessimistic comments",
        "Retail investors sold heavily on Tuesday"
    ]
   
    results = []
   
    for trial in range(num_trials):
        # Mix evidence randomly (60% accurate, 40% misleading)
        evidence_pool = accurate_evidence + misleading_evidence
        selected_evidence = random.sample(evidence_pool, k=4)
       
        # Alternate initial conclusions
        if trial % 2 == 0:
            conclusion = "Market decline caused by negative sentiment"
        else:
            conclusion = ground_truth
       
        result = contrastive_inquiry(
            conclusion=conclusion,
            context="Stock market declined 3% in one day",
            evidence=selected_evidence,
            agent_id=f"trial_{trial}"
        )
       
        # Check if conclusion aligns with ground truth
        correct = "interest rate" in result["recommended_conclusion"].lower() or \
                  "rate hike" in result["recommended_conclusion"].lower()
       
        results.append({
            "trial": trial,
            "evidence_quality": sum(e in accurate_evidence for e in selected_evidence) / 4,
            "correct_conclusion": correct,
            "confidence": result["confidence"]
        })
   
    # Calculate metrics
    accuracy = sum(r["correct_conclusion"] for r in results) / num_trials * 100
    avg_confidence_when_correct = sum(
        1 if r["confidence"] in ["high", "moderate"] else 0
        for r in results if r["correct_conclusion"]
    ) / sum(r["correct_conclusion"] for r in results) * 100
   
    print(f"Accuracy: {accuracy}%")
    print(f"Appropriate confidence when correct: {avg_confidence_when_correct}%")
   
    return results
# Run test
adversarial_results = run_adversarial_test(num_trials=20)

# Install: pip install slowapi
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi import FastAPI, Request
limiter = Limiter(key_func=get_remote_address)
app = FastAPI(title="Contrastive Inquiry API")
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.post("/contrastive_inquiry")
@limiter.limit("10/minute") # Max 10 requests per minute per IP
def api_contrastive_inquiry(request: Request, inquiry: InquiryRequest):
    try:
        result = contrastive_inquiry(
            inquiry.conclusion,
            inquiry.context,
            inquiry.evidence,
            inquiry.agent_id
        )
        return result
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except RuntimeError as e:
        raise HTTPException(status_code=500, detail=str(e))

from fastapi import Header, HTTPException
import secrets
# Generate secure API keys: secrets.token_urlsafe(32)
VALID_API_KEYS = {
    "your-secure-api-key-here",
    "another-api-key-here"
}
async def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key not in VALID_API_KEYS:
        raise HTTPException(
            status_code=403,
            detail="Invalid API key"
        )
    return x_api_key
@app.post("/contrastive_inquiry")
async def api_contrastive_inquiry(
    inquiry: InquiryRequest,
    api_key: str = Depends(verify_api_key)
):
    # Your existing code here
    pass
# Clients must include header: X-API-Key: your-secure-api-key-here

from pydantic import BaseModel, validator
class InquiryRequest(BaseModel):
    conclusion: str
    context: str
    evidence: List[str]
    agent_id: str = "default_agent"
   
    @validator('conclusion', 'context')
    def check_length(cls, v):
        if len(v) > 500:
            raise ValueError('Text must be 500 characters or less')
        return v
   
    @validator('evidence')
    def check_evidence_count(cls, v):
        if len(v) > 10:
            raise ValueError('Maximum 10 evidence items allowed')
        if any(len(e) > 300 for e in v):
            raise ValueError('Each evidence item must be 300 characters or less')
        return v

# Run the API server locally
uvicorn contrastive_inquiry:app --reload
# Test with curl
curl -X POST "http://localhost:8000/contrastive_inquiry" \
  -H "Content-Type: application/json" \
  -d '{
    "conclusion": "The system is slow due to database queries",
    "context": "Users report delays during peak hours",
    "evidence": ["Query logs show 200ms response time", "Network latency is 2.5s"]
  }'