DSPy Reasoning Strategies

DSPy integration for optimized reasoning patterns including Chain-of-Thought (CoT), ReAct, self-consistency, and meta-prompting with automatic optimization.

Features

• Multiple Reasoning Strategies: CoT, ReAct, self-consistency, meta-prompting
• Automatic Optimization: DSPy-based prompt optimization
• Confidence Scoring: ML-based confidence estimation
• Performance Tracking: Monitor reasoning effectiveness
• Custom Strategies: Add domain-specific reasoning patterns

Quick Start

from recoagent.reasoning import DSPyReasoningEngine, DSPyStrategy, DSPyConfig

# Configure DSPy
config = DSPyConfig(
    lm=dspy_lm,  # DSPy language model
    max_bootstrapped_demos=4,
    max_labeled_demos=16
)

# Initialize reasoning engine
engine = DSPyReasoningEngine(
    default_strategy=DSPyStrategy.COT,
    llm_config={
        "provider": "openai",
        "model_name": "gpt-4",
        "api_key": "your-api-key"
    }
)

# Use reasoning
result = engine.reason(
    query="Solve this problem: 2x + 5 = 13",
    context={"subject": "algebra", "difficulty": "medium"},
    strategy=DSPyStrategy.COT
)

print(f"Answer: {result['conclusion']}")
print(f"Confidence: {result['confidence']:.2f}")
print(f"Reasoning Steps: {result['reasoning_trace']}")

Reasoning Strategies

Chain-of-Thought (CoT)

# Use Chain-of-Thought reasoning
result = engine.reason(
    query="What is the capital of France?",
    strategy=DSPyStrategy.COT
)

# CoT provides step-by-step reasoning
print("Reasoning Steps:")
for step in result['reasoning_trace']:
    print(f"- {step}")
print(f"Final Answer: {result['conclusion']}")

ReAct (Reasoning + Acting)

# Use ReAct reasoning
result = engine.reason(
    query="Find the weather in New York",
    strategy=DSPyStrategy.REACT,
    context={"available_tools": ["web_search", "weather_api"]}
)

# ReAct combines reasoning with tool usage
print("Reasoning and Actions:")
for step in result['reasoning_trace']:
    print(f"- {step}")
print(f"Final Answer: {result['conclusion']}")

Self-Consistency

# Use self-consistency for better accuracy
result = engine.reason(
    query="What is 15% of 200?",
    strategy=DSPyStrategy.COT,
    use_self_consistency=True,
    num_samples=5  # Generate 5 reasoning paths
)

print(f"Answer: {result['conclusion']}")
print(f"Confidence: {result['confidence']:.2f}")
print(f"Consistency Score: {result['consistency_score']:.2f}")

Advanced Configuration

Custom DSPy Programs

# Define custom reasoning program
class CustomReasoning(dspy.Module):
    def __init__(self):
        super().__init__()
        self.reasoning = dspy.ChainOfThought("question -> reasoning -> answer")
    
    def forward(self, question):
        return self.reasoning(question=question)

# Register custom program
engine.register_custom_program("custom_reasoning", CustomReasoning)

# Use custom program
result = engine.reason(
    query="Custom reasoning problem",
    strategy="custom_reasoning"
)

Multi-Strategy Reasoning

# Combine multiple strategies
strategies = [DSPyStrategy.COT, DSPyStrategy.REACT]

results = []
for strategy in strategies:
    result = engine.reason(
        query="Complex reasoning problem",
        strategy=strategy
    )
    results.append(result)

# Compare results
for i, result in enumerate(results):
    print(f"Strategy {i+1}: {result['conclusion']} (confidence: {result['confidence']:.2f})")

Optimization

Strategy Optimization

# Optimize a specific strategy
training_data = [
    {"query": "What is 2+2?", "answer": "4"},
    {"query": "What is 3*3?", "answer": "9"},
    {"query": "What is 10/2?", "answer": "5"}
]

success = engine.optimize_strategy(
    strategy=DSPyStrategy.COT,
    training_data=training_data
)

if success:
    print("Strategy optimized successfully")
else:
    print("Strategy optimization failed")

Performance Monitoring

# Monitor reasoning performance
stats = engine.get_performance_stats()
print(f"Total Reasoning Calls: {stats['total_calls']}")
print(f"Average Confidence: {stats['avg_confidence']:.2f}")
print(f"Success Rate: {stats['success_rate']:.2f}")

# By strategy
for strategy, stats in engine.get_strategy_stats().items():
    print(f"{strategy}: {stats['calls']} calls, {stats['avg_confidence']:.2f} confidence")

Integration Examples

With Workflows

from packages.observability import trace_reasoning

@trace_reasoning(strategy="chain_of_thought")
async def reasoning_workflow(problem):
    engine = DSPyReasoningEngine()
    
    result = engine.reason(
        query=problem,
        strategy=DSPyStrategy.COT
    )
    
    return result

With Caching

from recoagent.reasoning import ReasoningCache, CacheConfig

# Configure caching
cache_config = CacheConfig(
    redis_url="redis://localhost:6379/0",
    ttl_seconds=3600,
    enable_compression=True
)

cache = ReasoningCache(cache_config)

# Use with reasoning engine
engine = DSPyReasoningEngine(
    enable_caching=True,
    cache=cache
)

# Cached reasoning
result = engine.reason(
    query="Cached reasoning problem",
    use_cache=True
)

With Evaluation

from packages.observability import get_evaluation_metrics

# Evaluate reasoning quality
metrics = get_evaluation_metrics()

result = metrics.evaluate_reasoning(
    query="What is machine learning?",
    expected_answer="Machine learning is a subset of AI...",
    actual_answer=engine.reason("What is machine learning?")['conclusion'],
    reasoning_trace=engine.reason("What is machine learning?")['reasoning_trace']
)

print(f"Reasoning Quality Score: {result.overall_score:.2f}")

Custom Strategies

Define Custom Strategy

# Define custom reasoning strategy
class CustomCoT(dspy.Module):
    def __init__(self):
        super().__init__()
        self.reasoning = dspy.ChainOfThought("question -> step1 -> step2 -> answer")
    
    def forward(self, question):
        return self.reasoning(question=question)

# Register custom strategy
engine.register_strategy("custom_cot", CustomCoT)

# Use custom strategy
result = engine.reason(
    query="Custom reasoning problem",
    strategy="custom_cot"
)

Multi-Step Reasoning

# Define multi-step reasoning
class MultiStepReasoning(dspy.Module):
    def __init__(self):
        super().__init__()
        self.step1 = dspy.ChainOfThought("question -> analysis")
        self.step2 = dspy.ChainOfThought("analysis -> reasoning")
        self.step3 = dspy.ChainOfThought("reasoning -> answer")
    
    def forward(self, question):
        analysis = self.step1(question=question)
        reasoning = self.step2(analysis=analysis.analysis)
        answer = self.step3(reasoning=reasoning.reasoning)
        return answer

# Register and use
engine.register_strategy("multi_step", MultiStepReasoning)

Performance Optimization

Batch Processing

# Process multiple queries in batch
queries = [
    "What is 2+2?",
    "What is 3*3?",
    "What is 10/2?"
]

results = engine.batch_reason(
    queries=queries,
    strategy=DSPyStrategy.COT
)

for query, result in zip(queries, results):
    print(f"Q: {query}")
    print(f"A: {result['conclusion']}")
    print(f"Confidence: {result['confidence']:.2f}")

Parallel Processing

import asyncio

# Parallel reasoning
async def parallel_reasoning(queries):
    tasks = []
    for query in queries:
        task = engine.areason(
            query=query,
            strategy=DSPyStrategy.COT
        )
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    return results

# Use parallel reasoning
queries = ["Problem 1", "Problem 2", "Problem 3"]
results = await parallel_reasoning(queries)

Best Practices

1. Choose Right Strategy: Select strategy based on problem type
2. Use Self-Consistency: For critical reasoning tasks
3. Optimize Strategies: Use training data to optimize
4. Monitor Performance: Track reasoning effectiveness
5. Cache Results: Cache expensive reasoning operations
6. Handle Errors: Implement proper error handling
7. Custom Strategies: Create domain-specific strategies

Troubleshooting

Common Issues

1. Low Confidence: Check reasoning quality and data
2. Slow Performance: Optimize strategies and use caching
3. Inconsistent Results: Use self-consistency
4. Memory Issues: Monitor memory usage and optimize

Debug Mode

# Enable debug logging
import logging
logging.getLogger('recoagent.reasoning.dspy_strategies').setLevel(logging.DEBUG)

API Reference

DSPyReasoningEngine

Parameter	Type	Description
default_strategy	DSPyStrategy	Default reasoning strategy
llm_config	Dict	LLM configuration
strategy_configs	Dict	Strategy-specific configurations

DSPyStrategy

Value	Description
COT	Chain-of-Thought reasoning
REACT	Reasoning + Acting
SELF_CONSISTENCY	Self-consistency reasoning
META_PROMPTING	Meta-prompting

Methods

Method	Description
reason(query, strategy, context)	Execute reasoning
optimize_strategy(strategy, training_data)	Optimize strategy
get_performance_stats()	Get performance statistics
register_strategy(name, program)	Register custom strategy