🚀 -> Project on GitHub <-

Adaptive Hops – Quick Start Guide

What is Adaptive Hops?

An intelligent routing system that automatically selects the optimal agent for every query:

Installation

All files are already in the project:

core/
  ├── adaptive_hops.py           # Core logic
  └── adaptive_integration.py    # Integration layer
examples/
  ├── adaptive_demo.py           # Demo script
  └── app_integration_example.py # Integration example
ADAPTIVE_HOPS.md                 # Full documentation

Quick Start (3 Steps)

1. Add Import

In app.py after line 23:

from core.adaptive_integration import initialize_adaptive_system

2. Initialize System

In app.py after line 295 (after multihop_agent init):

# Initialize adaptive system
adaptive_manager = None
adaptive_processor = None
try:
    adaptive_manager, adaptive_processor = initialize_adaptive_system(
        llm=agent.llm,
        agent=agent,
        multihop_agent=multihop_agent,
        system_monitor=system_monitor,
        performance_tracker=performance_tracker
    )
    logger.info("Adaptive system initialized")
except Exception as e:
    logger.error(f"Failed to initialize adaptive system: {e}")

3. Add Endpoint

In app.py after line 768 (after /query endpoint):

from typing import Dict, Any, List, Optional
from pydantic import BaseModel, Field, validator

class AdaptiveQueryRequest(BaseModel):
    query: str = Field(..., min_length=1, max_length=MAX_QUERY_LENGTH)
    force_complexity: Optional[str] = None
    enable_escalation: bool = True

    @validator('query')
    def sanitize_query_input(cls, v):
        return sanitize_query(v)

class AdaptiveQueryResponse(BaseModel):
    answer: str
    confidence: Optional[float]
    strategy: Dict[str, Any]
    metadata: Dict[str, Any]
    steps: Optional[int] = None
    search_queries: Optional[List[str]] = None
    reasoning_path: Optional[List[str]] = None

@app.post("/query-adaptive", response_model=AdaptiveQueryResponse, dependencies=[Depends(check_rate_limit)])
async def query_adaptive_endpoint(request: AdaptiveQueryRequest):
    if not adaptive_processor:
        raise HTTPException(status_code=503, detail="Adaptive system not available")
    result = adaptive_processor.process_query(
        query=request.query,
        force_complexity=request.force_complexity,
        enable_escalation=request.enable_escalation
    )
    return AdaptiveQueryResponse(**result)

Test It

Option A: Run Demo Script

python examples/adaptive_demo.py

Shows all features: complexity analysis, strategy decisions, escalation, etc.

Option B: Test API Endpoint

  1. Start server: 
    python app.py
  2. Send request: 
    curl -X POST "http://localhost:8000/query-adaptive" \
  -H "X-API-Key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
 "query": "Compare AI in healthcare vs manufacturing",
 "enable_escalation": true
  }'
  3. Check response: 
    {
  "answer": "...",
  "confidence": 0.85,
  "strategy": {
 "complexity": "high",
 "agent_type": "MultiHopReasoningAgent",
 "max_hops": 5,
 "reasoning": ["High complexity: MultiHop with 5 hops"]
  },
  "metadata": {
 "attempts": 1,
 "elapsed_time": 12.5,
 "escalation_history": []
  }
}

Usage Examples

Simple Query (LOW)

curl ... -d '{"query": "What is Python?"}'
# → SearchAgent, no tools, ~0.8s

Web Search Query (MID)

curl ... -d '{"query": "Latest AI news 2025"}'
# → SearchAgent + web tools, ~2.5s

Complex Query (HIGH)

curl ... -d '{"query": "Compare X and Y, analyze trends, recommend best option"}'
# → MultiHopReasoningAgent, 5 hops, ~15s

Force Complexity

curl ... -d '{"query": "Simple question", "force_complexity": "high"}'
# → Forces MultiHopAgent even for simple queries

How It Works

Query → Complexity Analysis → Strategy Decision → Agent Execution
          │                        │
          ├─ LLM Classification     ├─ Resource Check
          ├─ Length Heuristic       ├─ Agent Selection
          └─ Keyword Detection      └─ Configuration
                   ↓ (if confidence < 0.5)
              Escalation Loop
              LOW → MID → HIGH

Features

Customize Configuration

from core.adaptive_hops import AdaptiveConfig

custom_config = AdaptiveConfig(
    max_hops_high=3,               # Reduce high-complexity hops
    cpu_threshold_high=70.0,       # Stricter CPU limit
    confidence_low=0.6,            # Higher escalation threshold
    enable_resource_monitoring=False  # Disable monitoring
)
# Pass custom_config during initialization

Next Steps

  1. Read Full Docs: ADAPTIVE_HOPS.md
    • Architecture
    • API reference
    • Best practices
    • Troubleshooting
  2. Run Demo: python examples/adaptive_demo.py
  3. Integrate into API: See examples/app_integration_example.py
  4. Monitor in Production: Use logs + /stats endpoint

Benefits

Feature Before With Adaptive Hops
Agent Selection Manual (use_multihop=true/false) Automatic based on complexity
Performance Fixed strategy Optimized per query
Resources No adaptation Dynamic degradation
Confidence Not monitored Auto-escalation
Transparency Black box Full strategy + metadata

Support

Happy Adaptive Hopping!