Module 5: ReAct, Self-Consistency, and Advanced Patterns

8. Project: Multi-Strategy Problem Solver

Overview

In this project you'll build a complete system that takes any kind of problem, classifies it automatically, selects the most suitable prompt engineering technique, runs that technique, and optionally compares the results of several techniques to validate the answer.

This project pulls together everything you learned in the module: ReAct, Self-Consistency, Tree-of-Thought, Meta-prompting and Self-Refine.


System Architecture

                    MULTI-STRATEGY PROBLEM SOLVER
                    
    Input: Problem/Question
           │
           ▼
    ┌─────────────────┐
    │   CLASSIFIER    │  → type: math/factual/reasoning/creative/code/planning
    └────────┬────────┘
             │
             ▼
    ┌─────────────────┐
    │     ROUTER      │  → picks the optimal technique from type + constraints
    └────────┬────────┘
             │
    ┌────────┴──────────────────────────────────┐
    │                                           │
    ▼                                           ▼
 Technique A                                Technique B (if comparing)
 (main)                                     (alternative)
    │                                           │
    └────────────────────┬──────────────────────┘
                         ▼
                  ┌──────────────┐
                  │  EVALUATOR   │  → quality score, confidence
                  └──────┬───────┘
                         │
                         ▼
                  Structured output
                  {answer, technique_used, confidence, metrics}

Full Implementation

from openai import OpenAI
from collections import Counter
from dataclasses import dataclass, field
from typing import Optional, Any
import json
import re
import time
import math

client = OpenAI()

# ============================================================
# DATA MODELS
# ============================================================

@dataclass
class ClassifiedProblem:
    text: str
    problem_type: str             # math, factual, reasoning, creative, code, planning, qa_realtime
    complexity: str               # simple, medium, high
    needs_external: bool
    accuracy_critical: bool
    has_unique_answer: bool       # False for creative

@dataclass
class TechniqueResult:
    technique: str
    answer: str
    time_seconds: float
    n_calls: int
    confidence: float
    metadata: dict = field(default_factory=dict)

@dataclass
class FinalResult:
    problem: str
    classification: ClassifiedProblem
    main_result: TechniqueResult
    alternative_result: Optional[TechniqueResult]
    consensus_answer: str
    metrics: dict
    recommended_technique: str


# ============================================================
# MODULE 1: CLASSIFIER
# ============================================================

def classify_problem(problem: str) -> ClassifiedProblem:
    """
    Analyzes a problem and classifies it for routing.
    
    Args:
        problem: The text of the problem to classify
    
    Returns:
        ClassifiedProblem with every attribute the routing needs
    """
    classification_prompt = f"""Analyze the following problem and classify it precisely.

Problem: {problem}

Reply in JSON with this exact schema:
{{
    "type": "math|factual|reasoning|creative|code|planning|qa_realtime",
    "complexity": "simple|medium|high",
    "needs_external": true|false,
    "accuracy_critical": true|false,
    "has_unique_answer": true|false,
    "type_reason": "explanation in 10 words"
}}

Type guide:
- math: numeric operations, algebra, calculus, statistics
- factual: facts the model knows from training
- reasoning: logical deductions, inferences, multi-step without calculation
- creative: writing, generation, no single correct answer
- code: writing, reviewing, or debugging code
- planning: designing strategies, architectures, roadmaps
- qa_realtime: questions about current data (prices, news, dates)

needs_external guide: true if the answer changes over time or requires APIs/a database
accuracy_critical guide: true if errors have a high impact (finance, medicine, legal)"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": classification_prompt}],
        temperature=0,
        response_format={"type": "json_object"}
    )
    
    try:
        data = json.loads(response.choices[0].message.content)
        return ClassifiedProblem(
            text=problem,
            problem_type=data.get("type", "reasoning"),
            complexity=data.get("complexity", "medium"),
            needs_external=data.get("needs_external", False),
            accuracy_critical=data.get("accuracy_critical", False),
            has_unique_answer=data.get("has_unique_answer", True)
        )
    except Exception as e:
        # Conservative fallback
        return ClassifiedProblem(
            text=problem,
            problem_type="reasoning",
            complexity="medium",
            needs_external=False,
            accuracy_critical=False,
            has_unique_answer=True
        )


# ============================================================
# MODULE 2: TECHNIQUE IMPLEMENTATIONS
# ============================================================

def _extract_number(text: str) -> str:
    """Extracts the final number from a reasoning text."""
    patterns = [
        r'(?:answer|result)[:\s=]+(-?\d+\.?\d*)',
        r'\*\*(-?\d+\.?\d*)\*\*',
        r'=\s*(-?\d+\.?\d*)\s*$'
    ]
    for p in patterns:
        m = re.search(p, text, re.IGNORECASE | re.MULTILINE)
        if m:
            return m.group(1)
    nums = re.findall(r'-?\d+\.?\d*', text)
    return nums[-1] if nums else text.strip()[-20:]

def run_zero_shot(problem: str) -> TechniqueResult:
    """Direct solving with no special technique."""
    t0 = time.time()
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": problem}],
        temperature=0.3,
        max_tokens=400
    ).choices[0].message.content
    return TechniqueResult(
        technique="zero_shot",
        answer=resp,
        time_seconds=time.time() - t0,
        n_calls=1,
        confidence=0.7
    )

def run_cot(problem: str) -> TechniqueResult:
    """Chain-of-Thought: step-by-step reasoning."""
    t0 = time.time()
    prompt = f"""{problem}

Think step by step:
1. Identify what you know and what you need to find
2. Apply the right process step by step
3. Check whether your answer makes sense
4. Write: "Final answer: [answer]"
"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        max_tokens=500
    ).choices[0].message.content
    return TechniqueResult(
        technique="chain_of_thought",
        answer=resp,
        time_seconds=time.time() - t0,
        n_calls=1,
        confidence=0.82,
        metadata={"has_reasoning": True}
    )

def run_self_consistency(problem: str, n: int = 5) -> TechniqueResult:
    """Self-Consistency: N answers with a majority vote."""
    t0 = time.time()
    answers = []
    
    for _ in range(n):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"{problem}\nThink step by step. Answer: [value]"}],
            temperature=0.7,
            max_tokens=400
        ).choices[0].message.content
        answers.append(_extract_number(resp))
    
    counts = Counter(answers)
    winner, votes = counts.most_common(1)[0]
    confidence = votes / n
    
    return TechniqueResult(
        technique="self_consistency",
        answer=winner,
        time_seconds=time.time() - t0,
        n_calls=n,
        confidence=confidence,
        metadata={"n_samples": n, "distribution": dict(counts), "winner_votes": votes}
    )

def run_react(problem: str) -> TechniqueResult:
    """
    Simplified ReAct: reasoning with calculation and search tools.
    """
    t0 = time.time()
    
    TOOLS = [
        {
            "type": "function",
            "function": {
                "name": "calculate",
                "description": "Evaluates mathematical expressions. Use it for numeric calculations.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "expression": {"type": "string", "description": "Expression to compute"}
                    },
                    "required": ["expression"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "get_fact",
                "description": "Simulates fetching a current factual data point. In production: call a real API.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string"}
                    },
                    "required": ["query"]
                }
            }
        }
    ]
    
    messages = [
        {
            "role": "system",
            "content": "Use tools to get data and to compute. Don't make up numbers. When you have the answer, finish with 'Answer: [result]'"
        },
        {"role": "user", "content": problem}
    ]
    
    n_calls = 1
    tools_used = []
    
    for _ in range(6):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
            temperature=0
        )
        msg = response.choices[0].message
        n_calls += 1
        
        if not msg.tool_calls:
            final_answer = msg.content or "No answer"
            break
        
        messages.append({
            "role": "assistant",
            "content": msg.content or "",
            "tool_calls": [tc.model_dump() for tc in msg.tool_calls]
        })
        
        for tc in msg.tool_calls:
            tool_name = tc.function.name
            args = json.loads(tc.function.arguments)
            tools_used.append(tool_name)
            
            if tool_name == "calculate":
                expr = args.get("expression", "0")
                try:
                    # Safe eval, math operations only
                    allowed = set("0123456789+-*/.() ")
                    if all(c in allowed for c in expr):
                        result = str(round(eval(expr, {"__builtins__": {}}, {"math": math}), 4))
                    else:
                        result = "Expression not allowed"
                except Exception as e:
                    result = f"Error: {e}"
            else:
                result = f"[Simulated data for: {args.get('query', '')}]"
            
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": result
            })
    else:
        final_answer = "Step limit reached"
    
    return TechniqueResult(
        technique="react",
        answer=final_answer,
        time_seconds=time.time() - t0,
        n_calls=n_calls,
        confidence=0.85 if tools_used else 0.7,
        metadata={"tools_used": tools_used}
    )

def run_tot_simplified(problem: str, breadth: int = 3) -> TechniqueResult:
    """Simplified Tree-of-Thought: generate approaches, evaluate, continue with the best."""
    t0 = time.time()
    n_calls = 0
    
    # Generate approaches
    approaches_resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""Problem: {problem}

Generate {breadth} DISTINCT approaches to solve this problem.
Each one has to be a different strategy.
Format: "1. [approach]" per line."""}],
        temperature=0.8,
        max_tokens=300
    )
    n_calls += 1
    
    lines = [l.lstrip('0123456789.-) ').strip() 
             for l in approaches_resp.choices[0].message.content.split('\n')
             if l.strip() and (l.strip()[0].isdigit() or l.strip()[0] == '-')]
    approaches = lines[:breadth]
    
    if not approaches:
        approaches = ["Direct step-by-step approach"]
    
    # Evaluate the approaches
    scored = []
    for approach in approaches:
        eval_resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"Problem: {problem}\nApproach: {approach}\nPromising? Score 0.0-1.0. The number only."}],
            temperature=0,
            max_tokens=10
        )
        n_calls += 1
        try:
            score = float(eval_resp.choices[0].message.content.strip()[:4])
            score = max(0.0, min(1.0, score))
        except ValueError:
            score = 0.5
        scored.append((score, approach))
    
    best_approach = max(scored, key=lambda x: x[0])[1]
    
    # Solve with the best approach
    solution = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""Problem: {problem}

Starting from: {best_approach}

Keep going step by step until the complete solution.
At the end: "Answer: [answer]"
"""}],
        temperature=0,
        max_tokens=500
    )
    n_calls += 1
    
    return TechniqueResult(
        technique="tree_of_thought",
        answer=solution.choices[0].message.content,
        time_seconds=time.time() - t0,
        n_calls=n_calls,
        confidence=0.78,
        metadata={"best_approach": best_approach, "n_approaches_evaluated": len(approaches)}
    )

def run_self_refine(problem: str, max_iter: int = 2) -> TechniqueResult:
    """Self-Refine: generate, critique, improve."""
    t0 = time.time()
    n_calls = 0
    
    # Initial generation
    initial_resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": problem}],
        temperature=0.3,
        max_tokens=500
    ).choices[0].message.content
    n_calls += 1
    
    answer = initial_resp
    current_score = 0.5
    
    for _ in range(max_iter):
        # Critique
        critique_resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"""Problem: {problem}
Proposed answer: {answer}

Evaluate: Are there errors? Is anything missing? Can it be improved?
Reply in JSON: {{"score": 0.0-1.0, "improvements": ["improvement1", "improvement2"], "can_improve": true/false}}"""}],
            temperature=0,
            response_format={"type": "json_object"}
        ).choices[0].message.content
        n_calls += 1
        
        try:
            critique = json.loads(critique_resp)
            score = float(critique.get("score", 0.5))
            improvements = critique.get("improvements", [])
            can_improve = critique.get("can_improve", True)
        except Exception:
            break
        
        if not can_improve or score >= 0.88:
            current_score = score
            break
        
        # Refinement
        improvements_str = "\n".join([f"- {m}" for m in improvements[:3]])
        refined = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"""Problem: {problem}
Previous answer: {answer}
Improvements required:
{improvements_str}

Generate an improved answer incorporating all the improvements."""}],
            temperature=0.2,
            max_tokens=500
        ).choices[0].message.content
        n_calls += 1
        
        answer = refined
        current_score = score
    
    return TechniqueResult(
        technique="self_refine",
        answer=answer,
        time_seconds=time.time() - t0,
        n_calls=n_calls,
        confidence=min(current_score, 0.95),
        metadata={"iterations": max_iter}
    )


# ============================================================
# MODULE 3: ROUTER
# ============================================================

def select_and_run(
    classification: ClassifiedProblem,
    compare_mode: bool = False,
    n_sc: int = 5
) -> tuple[TechniqueResult, Optional[TechniqueResult]]:
    """
    Selects and runs the technique based on the problem's classification.
    
    Args:
        classification: The classifier's result
        compare_mode: If True, also runs an alternative technique to compare
        n_sc: Number of samples for Self-Consistency
    
    Returns:
        Tuple (main_result, alternative_result)
    """
    problem_type = classification.problem_type
    accuracy = classification.accuracy_critical
    external = classification.needs_external
    complexity = classification.complexity
    has_unique = classification.has_unique_answer
    
    # Main technique selection
    if external:
        main = run_react(classification.text)
    elif problem_type == "math" and accuracy and complexity in ["medium", "high"]:
        main = run_self_consistency(classification.text, n=n_sc)
    elif problem_type == "math":
        main = run_cot(classification.text)
    elif problem_type in ["reasoning", "logic"] and complexity == "high":
        main = run_self_consistency(classification.text, n=3)
    elif problem_type in ["reasoning", "logic"]:
        main = run_cot(classification.text)
    elif problem_type == "planning" and complexity == "high":
        main = run_tot_simplified(classification.text)
    elif problem_type == "code":
        main = run_self_refine(classification.text, max_iter=2)
    elif problem_type == "creative" or not has_unique:
        main = run_zero_shot(classification.text)
    else:
        # factual, qa_realtime, default
        main = run_cot(classification.text) if external else run_zero_shot(classification.text)
    
    # Alternative technique for comparison
    alternative = None
    if compare_mode:
        if main.technique == "chain_of_thought":
            alternative = run_self_consistency(classification.text, n=3)
        elif main.technique == "self_consistency":
            alternative = run_cot(classification.text)
        elif main.technique == "tree_of_thought":
            alternative = run_cot(classification.text)
        elif main.technique == "self_refine":
            alternative = run_cot(classification.text)
    
    return main, alternative


# ============================================================
# MODULE 4: EVALUATOR AND CONSENSUS
# ============================================================

def evaluate_answer_quality(problem: str, answer: str) -> dict:
    """
    Evaluates the quality of an answer without knowing the correct one.
    """
    prompt = f"""Evaluate this answer:

Problem: {problem}
Answer: {answer}

Reply in JSON:
{{
    "score_completeness": 0.0-1.0,
    "score_accuracy": 0.0-1.0,
    "score_clarity": 0.0-1.0,
    "has_answer": true/false,
    "possible_errors": ["error if there is one"]
}}"""
    
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        response_format={"type": "json_object"}
    ).choices[0].message.content
    
    try:
        data = json.loads(resp)
        average_score = (
            data.get("score_completeness", 0.5) * 0.4 +
            data.get("score_accuracy", 0.5) * 0.4 +
            data.get("score_clarity", 0.5) * 0.2
        )
        return {**data, "average_score": average_score}
    except Exception:
        return {"average_score": 0.5, "has_answer": True, "possible_errors": []}

def determine_consensus(
    main_result: TechniqueResult,
    alternative_result: Optional[TechniqueResult],
    classification: ClassifiedProblem
) -> str:
    """
    Determines the consensus answer when there are two techniques.
    """
    if alternative_result is None:
        return main_result.answer
    
    # For math: compare the numeric answers
    if classification.problem_type == "math":
        num_main = _extract_number(main_result.answer)
        num_alt = _extract_number(alternative_result.answer)
        if num_main == num_alt:
            return main_result.answer  # Consensus
        # They differ: pick the one with higher confidence
        if main_result.confidence >= alternative_result.confidence:
            return main_result.answer
        return alternative_result.answer
    
    # For the rest: pick the one with the higher quality score
    eval_main = evaluate_answer_quality(classification.text, main_result.answer)
    eval_alt = evaluate_answer_quality(classification.text, alternative_result.answer)
    
    if eval_main["average_score"] >= eval_alt["average_score"]:
        return main_result.answer
    return alternative_result.answer


# ============================================================
# MAIN SYSTEM
# ============================================================

def solve_problem(
    problem: str,
    compare_mode: bool = False,
    n_sc: int = 5,
    verbose: bool = True
) -> FinalResult:
    """
    The Multi-Strategy Problem Solver's main system.
    
    Args:
        problem: The problem or question to solve
        compare_mode: If True, runs two techniques and compares them
        n_sc: N for Self-Consistency if it applies
        verbose: If True, prints the process
    
    Returns:
        FinalResult with the answer, metrics and complete metadata
    """
    t_start = time.time()
    
    if verbose:
        print(f"\n{'='*60}")
        print(f"MULTI-STRATEGY PROBLEM SOLVER")
        print(f"{'='*60}")
        print(f"Problem: {problem[:80]}...")
    
    # Step 1: Classify
    if verbose:
        print("\n[1/4] Classifying the problem...")
    classification = classify_problem(problem)
    
    if verbose:
        print(f"  Type: {classification.problem_type}")
        print(f"  Complexity: {classification.complexity}")
        print(f"  External data: {classification.needs_external}")
        print(f"  Accuracy critical: {classification.accuracy_critical}")
    
    # Step 2: Select and run the technique
    if verbose:
        print("\n[2/4] Selecting and running the technique...")
    
    main_result, alternative_result = select_and_run(
        classification, compare_mode, n_sc
    )
    
    if verbose:
        print(f"  Main technique: {main_result.technique}")
        print(f"  API calls: {main_result.n_calls}")
        print(f"  Time: {main_result.time_seconds:.2f}s")
        if alternative_result:
            print(f"  Alternative technique: {alternative_result.technique}")
    
    # Step 3: Evaluate the quality
    if verbose:
        print("\n[3/4] Evaluating the quality...")
    evaluation = evaluate_answer_quality(problem, main_result.answer)
    
    if verbose:
        print(f"  Quality score: {evaluation.get('average_score', 0):.2f}")
    
    # Step 4: Determine the consensus answer
    if verbose:
        print("\n[4/4] Determining the final answer...")
    consensus_answer = determine_consensus(
        main_result, alternative_result, classification
    )
    
    # Compute the metrics
    total_calls = main_result.n_calls
    if alternative_result:
        total_calls += alternative_result.n_calls
    # Include the classification and evaluation calls
    total_calls += 2
    
    metrics = {
        "total_time_seconds": time.time() - t_start,
        "total_calls": total_calls,
        "estimated_cost_usd": total_calls * 0.0002,  # gpt-4o-mini estimate
        "quality_score": evaluation.get("average_score", 0),
        "technique_confidence": main_result.confidence,
        "has_comparison": alternative_result is not None
    }
    
    if verbose:
        print(f"\n{'='*60}")
        print(f"FINAL RESULT:")
        print(f"Technique used: {main_result.technique}")
        print(f"Answer: {consensus_answer[:200]}...")
        print(f"Total time: {metrics['total_time_seconds']:.2f}s")
        print(f"API calls: {total_calls}")
        print(f"Estimated cost: ${metrics['estimated_cost_usd']:.4f}")
        print(f"{'='*60}")
    
    return FinalResult(
        problem=problem,
        classification=classification,
        main_result=main_result,
        alternative_result=alternative_result,
        consensus_answer=consensus_answer,
        metrics=metrics,
        recommended_technique=main_result.technique
    )

Test Suite

# ============================================================
# SYSTEM TESTS
# ============================================================

TEST_CASES = [
    {
        "category": "Simple math",
        "problem": "What is 15% of 2,400?",
        "expected_answer": "360",
        "expected_technique": "chain_of_thought"
    },
    {
        "category": "Critical math",
        "problem": "If I invest €10,000 with compound interest at 6% a year for 5 years, how much will I have?",
        "expected_answer": "13382",
        "expected_technique": "self_consistency"
    },
    {
        "category": "Logical reasoning",
        "problem": "All managers work on Mondays. Alice works on Mondays. Is Alice a manager?",
        "expected_answer": "not necessarily",
        "expected_technique": "chain_of_thought"
    },
    {
        "category": "Planning",
        "problem": "Design a 3-month plan to learn Machine Learning starting from basic Python. Include specific resources.",
        "expected_answer": None,  # Qualitative evaluation
        "expected_technique": "tree_of_thought"
    },
    {
        "category": "Code",
        "problem": "Write a Python function that finds all the prime numbers up to N using the Sieve of Eratosthenes.",
        "expected_answer": None,
        "expected_technique": "self_refine"
    },
    {
        "category": "Creative",
        "problem": "Write a haiku about programming.",
        "expected_answer": None,
        "expected_technique": "zero_shot"
    }
]

def run_test_suite(verbose: bool = True) -> dict:
    """
    Runs every test case and produces a report.
    """
    print("\n" + "="*70)
    print("TEST SUITE: MULTI-STRATEGY PROBLEM SOLVER")
    print("="*70)
    
    suite_results = []
    
    for i, case in enumerate(TEST_CASES, 1):
        print(f"\n[Test {i}/{len(TEST_CASES)}] {case['category']}")
        print(f"Problem: {case['problem'][:60]}...")
        
        # Run it
        result = solve_problem(
            case["problem"],
            compare_mode=False,
            verbose=False  # Quiet inside the suite
        )
        
        # Evaluate
        technique_correct = result.recommended_technique == case.get("expected_technique")
        
        answer_correct = None
        if case.get("expected_answer"):
            resp_num = _extract_number(result.consensus_answer)
            try:
                expected_num = case.get("expected_answer", "")
                answer_correct = abs(float(resp_num) - float(expected_num)) < 1.0
            except ValueError:
                answer_correct = case["expected_answer"].lower() in result.consensus_answer.lower()
        
        test_result = {
            "category": case["category"],
            "technique_used": result.recommended_technique,
            "expected_technique": case.get("expected_technique"),
            "technique_correct": technique_correct,
            "answer_correct": answer_correct,
            "confidence": result.metrics["technique_confidence"],
            "calls": result.metrics["total_calls"],
            "time": result.metrics["total_time_seconds"]
        }
        suite_results.append(test_result)
        
        if verbose:
            print(f"  Technique: {result.recommended_technique} ({'✓' if technique_correct else '✗'})")
            if answer_correct is not None:
                print(f"  Answer: {'✓ Correct' if answer_correct else '✗ Incorrect'}")
            print(f"  Calls: {result.metrics['total_calls']}, Time: {result.metrics['total_time_seconds']:.2f}s")
    
    # Summary
    correct_techniques = sum(1 for r in suite_results if r["technique_correct"])
    evaluable_answers = [r for r in suite_results if r["answer_correct"] is not None]
    correct_answers = sum(1 for r in evaluable_answers if r["answer_correct"])
    
    total_calls = sum(r["calls"] for r in suite_results)
    average_time = sum(r["time"] for r in suite_results) / len(suite_results)
    
    print(f"\n{'='*70}")
    print("TEST SUMMARY")
    print(f"{'='*70}")
    print(f"Tests run: {len(TEST_CASES)}")
    print(f"Correct technique: {correct_techniques}/{len(TEST_CASES)} ({correct_techniques/len(TEST_CASES):.0%})")
    if evaluable_answers:
        print(f"Correct answer: {correct_answers}/{len(evaluable_answers)} ({correct_answers/len(evaluable_answers):.0%})")
    print(f"Total API calls: {total_calls}")
    print(f"Average time: {average_time:.2f}s")
    print(f"Total estimated cost: ${total_calls * 0.0002:.4f}")
    
    return {
        "details": suite_results,
        "technique_accuracy": correct_techniques / len(TEST_CASES),
        "total_calls": total_calls,
        "average_time": average_time
    }

Optional Extensions

Extension 1: Command-line interface

import sys

def cli_interactive():
    """
    An interactive interface to use the solver from the terminal.
    Type 'quit' to exit.
    """
    print("\n=== MULTI-STRATEGY PROBLEM SOLVER ===")
    print("Type any problem or question.")
    print("Commands: 'compare' (toggles comparison mode), 'quit'\n")
    
    compare_mode = False
    
    while True:
        problem = input("Your question: ").strip()
        
        if problem.lower() == "quit":
            print("See you!")
            break
        
        if problem.lower() == "compare":
            compare_mode = not compare_mode
            print(f"Comparison mode: {'ON' if compare_mode else 'OFF'}")
            continue
        
        if not problem:
            continue
        
        result = solve_problem(problem, compare_mode=compare_mode, verbose=True)
        print(f"\nFINAL ANSWER:\n{result.consensus_answer}\n")

Extension 2: Logging and persistent metrics

import json
from datetime import datetime
from pathlib import Path

class SolverLogger:
    def __init__(self, log_file: str = "solver_logs.jsonl"):
        self.log_file = Path(log_file)
    
    def log(self, result: FinalResult):
        """Saves every solve into a JSONL file for later analysis."""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "problem": result.problem[:200],
            "type": result.classification.problem_type,
            "technique": result.recommended_technique,
            "n_calls": result.metrics["total_calls"],
            "time_sec": result.metrics["total_time_seconds"],
            "quality_score": result.metrics["quality_score"],
            "confidence": result.metrics["technique_confidence"]
        }
        with self.log_file.open("a") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    
    def stats(self) -> dict:
        """Computes usage statistics for the solver."""
        if not self.log_file.exists():
            return {}
        
        logs = [json.loads(l) for l in self.log_file.read_text().strip().split('\n') if l]
        
        from collections import Counter
        technique_counts = Counter(l["technique"] for l in logs)
        type_counts = Counter(l["type"] for l in logs)
        
        return {
            "total_solutions": len(logs),
            "most_used_techniques": technique_counts.most_common(5),
            "most_common_types": type_counts.most_common(5),
            "average_time": sum(l["time_sec"] for l in logs) / len(logs),
            "average_score": sum(l["quality_score"] for l in logs) / len(logs),
            "average_calls": sum(l["n_calls"] for l in logs) / len(logs)
        }

# Usage with logging:
logger = SolverLogger()
result = solve_problem("What is the square root of 2025?")
logger.log(result)
print(logger.stats())

Extension 3: Cache of similar answers

from hashlib import md5
import shelve

class CachedSolver:
    def __init__(self, cache_file: str = "solver_cache"):
        self.cache_file = cache_file
    
    def _hash_problem(self, problem: str) -> str:
        """Creates a unique hash for the problem."""
        return md5(problem.strip().lower().encode()).hexdigest()
    
    def solve_with_cache(self, problem: str, ttl_hours: int = 24) -> FinalResult:
        """
        Solve with a cache: if the same problem was solved recently,
        return the cached result.
        """
        key = self._hash_problem(problem)
        
        with shelve.open(self.cache_file) as cache:
            if key in cache:
                entry = cache[key]
                # Check the TTL
                age_hours = (time.time() - entry["timestamp"]) / 3600
                if age_hours < ttl_hours:
                    print(f"[Cache hit] Result found ({age_hours:.1f}h old)")
                    return entry["result"]
            
            # Not in the cache: solve it and store it
            result = solve_problem(problem, verbose=False)
            cache[key] = {
                "result": result,
                "timestamp": time.time()
            }
            return result

Project Success Criteria

Check that your implementation meets these criteria:

  • The classifier correctly identifies the problem type (math/factual/reasoning/creative/code/planning) in >80% of cases
  • The router selects the appropriate technique based on the type and the complexity
  • Each technique produces results of consistent quality
  • The evaluation system assigns reasonable quality scores
  • Comparison mode runs two techniques and determines the better answer
  • The system handles errors gracefully (no crashes on edge cases)
  • Logs and metrics are recorded correctly (extension)
  • The cache works and avoids repeated calls (extension)

Main Entry Point

if __name__ == "__main__":
    # Basic demo
    demo_problems = [
        "What is 30% of 1,750?",
        "Explain the difference between a list and a tuple in Python",
        "Design the architecture of a REST API for an inventory system",
        "Write a function that reverses a string without using reversed()",
    ]
    
    print("=== DEMO: MULTI-STRATEGY PROBLEM SOLVER ===\n")
    
    for problem in demo_problems:
        print(f"\n{'─'*50}")
        result = solve_problem(problem, verbose=True)
    
    # Test suite
    print("\n\n=== RUNNING THE TEST SUITE ===")
    report = run_test_suite(verbose=True)
    
    print(f"\nAccuracy on technique selection: {report['technique_accuracy']:.0%}")

Summary

In this project you built a complete Multi-Strategy Problem Solver that:

  1. Automatically classifies the problem type (math, reasoning, code, planning, etc.)
  2. Selects the optimal technique based on the type, the complexity, and whether it needs external data
  3. Runs the selected technique with appropriate parameters
  4. Evaluates the quality of the answer
  5. Determines consensus when there's a comparison between techniques
  6. Records metrics on usage, cost and quality

The techniques implemented are: Zero-shot, CoT, Self-Consistency, ReAct, Tree-of-Thought, and Self-Refine.

This system is extensible: you can add new problem types, new techniques, and hook up real external tools inside the ReAct module.


Additional resources

  1. ReAct: Synergizing Reasoning and Acting (Yao et al., 2022)
  2. Self-Consistency (Wang et al., 2022)
  3. Tree of Thoughts (Yao et al., 2023)
  4. Self-Refine (Madaan et al., 2023)
  5. OpenAI Function Calling
  6. LangChain Agent framework
  7. Evaluating LLM outputs - OpenAI evals