How to Build LangChain Search Tool with Rate Limiting

import os, requests, time, json, threading
from datetime import datetime

API_KEY = os.environ['SCAVIO_API_KEY']
SH = {'x-api-key': API_KEY, 'Content-Type': 'application/json'}

class RateLimitedSearch:
    def __init__(self, max_per_minute=30, daily_budget=500):
        self.max_per_minute = max_per_minute
        self.daily_budget = daily_budget
        self.calls = []
        self.daily_calls = 0
        self.daily_reset = datetime.now().strftime('%Y-%m-%d')
        self.lock = threading.Lock()
    
    def _check_limits(self):
        now = time.time()
        # Reset daily counter
        today = datetime.now().strftime('%Y-%m-%d')
        if today != self.daily_reset:
            self.daily_calls = 0
            self.daily_reset = today
        # Check daily budget
        if self.daily_calls >= self.daily_budget:
            return False, f'Daily budget ({self.daily_budget}) reached. Cost: ${self.daily_calls * 0.005:.2f}'
        # Check per-minute rate
        self.calls = [t for t in self.calls if now - t < 60]
        if len(self.calls) >= self.max_per_minute:
            wait = 60 - (now - self.calls[0])
            return False, f'Rate limit: {self.max_per_minute}/min. Wait {wait:.0f}s.'
        return True, 'OK'
    
    def search(self, query, num_results=5):
        with self.lock:
            allowed, reason = self._check_limits()
            if not allowed:
                return {'error': reason, 'results': []}
            self.calls.append(time.time())
            self.daily_calls += 1
        resp = requests.post('https://api.scavio.dev/api/v1/search',
            headers=SH, json={'query': query, 'country_code': 'us'}, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        results = [{'title': r.get('title', ''), 'url': r.get('link', ''),
                    'snippet': r.get('snippet', '')} for r in data.get('organic_results', [])[:num_results]]
        return {'results': results, 'count': len(results),
                'credits_used': self.daily_calls, 'budget_remaining': self.daily_budget - self.daily_calls}

rl_search = RateLimitedSearch(max_per_minute=30, daily_budget=500)
result = rl_search.search('langchain search tool tutorial')
print(f'Results: {result.get("count", 0)}')
print(f'Credits used today: {result.get("credits_used", 0)}')
print(f'Budget remaining: {result.get("budget_remaining", 0)}')

# LangChain tool definition
# from langchain.tools import Tool

def langchain_search(query: str) -> str:
    """Search the web with rate limiting and budget controls.
    Use for finding documentation, current information, or verifying facts."""
    result = rl_search.search(query)
    if 'error' in result:
        return f'Search blocked: {result["error"]}'
    output = []
    for r in result['results']:
        output.append(f'Title: {r["title"]}')
        output.append(f'URL: {r["url"]}')
        output.append(f'Snippet: {r["snippet"]}')
        output.append('')
    output.append(f'[Credits: {result["credits_used"]}/{rl_search.daily_budget} today]')
    return '\n'.join(output)

# Register with LangChain:
# search_tool = Tool(
#     name='web_search',
#     func=langchain_search,
#     description='Search the web. Rate limited to prevent runaway costs.'
# )

# Test
print(langchain_search('how to build langchain agent with tools'))
print(f'\nRate: {rl_search.max_per_minute}/min, Budget: {rl_search.daily_budget}/day')
print(f'Cost cap: ${rl_search.daily_budget * 0.005:.2f}/day')

def usage_report():
    print(f'\n=== Search Usage Report ===')
    print(f'  Date: {datetime.now().strftime("%Y-%m-%d %H:%M")}')
    print(f'  Calls today: {rl_search.daily_calls}')
    print(f'  Budget: {rl_search.daily_calls}/{rl_search.daily_budget}')
    print(f'  Cost today: ${rl_search.daily_calls * 0.005:.3f}')
    print(f'  Budget remaining: ${(rl_search.daily_budget - rl_search.daily_calls) * 0.005:.3f}')
    # Usage rate
    pct = rl_search.daily_calls / rl_search.daily_budget * 100
    print(f'  Usage: {pct:.0f}%')
    if pct > 80:
        print(f'  WARNING: Over 80% of daily budget used!')
    elif pct > 50:
        print(f'  NOTE: Over 50% of daily budget used.')
    # Projected monthly cost
    monthly = rl_search.daily_calls * 0.005 * 30
    print(f'\n  Projected monthly cost: ${monthly:.2f}')
    print(f'  Max monthly cost: ${rl_search.daily_budget * 0.005 * 30:.2f}')

# Simulate multiple searches
for i in range(5):
    rl_search.search(f'test query {i}')

usage_report()
print(f'\n  Rate limit: {rl_search.max_per_minute} calls/minute')
print(f'  Daily budget: {rl_search.daily_budget} calls (${ rl_search.daily_budget * 0.005:.2f})')
print(f'  Per call: $0.005')