Hermes Hitting Walls: Search Tool Solutions

import requests, os, json

# Tool definition for Hermes function calling
SEARCH_TOOL = {
    "type": "function",
    "function": {
        "name": "web_search",
        "description": "Search the web for current information. Use when you need: pricing, documentation, recent news, version numbers, or any fact that could have changed recently.",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query. Be specific and include year if relevant."
                }
            },
            "required": ["query"]
        }
    }
}

def execute_search(query: str) -> str:
    """Execute search and return formatted results for the agent."""
    try:
        resp = requests.post("https://api.scavio.dev/api/v1/search",
            headers={"x-api-key": os.environ["SCAVIO_API_KEY"]},
            json={"platform": "google", "query": query},
            timeout=10)
        resp.raise_for_status()
        results = resp.json().get("organic", [])[:5]
        if not results:
            return "No results found. Try a different query."
        formatted = []
        for r in results:
            formatted.append(f"Title: {r.get('title', '')}")
            formatted.append(f"Snippet: {r.get('snippet', '')}")
            formatted.append(f"URL: {r.get('link', '')}")
            formatted.append("---")
        return "\\n".join(formatted)
    except requests.exceptions.Timeout:
        return "Search timed out. Try again with a simpler query."
    except requests.exceptions.HTTPError as e:
        return f"Search failed: HTTP {e.response.status_code}"
    except Exception as e:
        return f"Search error: {str(e)}"

from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

def run_hermes_with_search(user_message: str) -> str:
    """Run Hermes with search tool available."""
    messages = [
        {"role": "system", "content": "You have access to web_search. Use it when you need current information."},
        {"role": "user", "content": user_message}
    ]

    # Generate response (may include tool call)
    response = generate(messages, tools=[SEARCH_TOOL])

    # Check if model wants to use a tool
    if response.get("tool_calls"):
        for call in response["tool_calls"]:
            if call["function"]["name"] == "web_search":
                args = json.loads(call["function"]["arguments"])
                search_result = execute_search(args["query"])

                # Add tool result to conversation
                messages.append({"role": "assistant", "content": None, "tool_calls": [call]})
                messages.append({
                    "role": "tool",
                    "tool_call_id": call["id"],
                    "content": search_result
                })

        # Generate final response with search results in context
        final = generate(messages, tools=[SEARCH_TOOL])
        return final["content"]

    return response["content"]

import time
from dataclasses import dataclass, field

@dataclass
class ToolHealthMonitor:
    tool_name: str
    consecutive_failures: int = 0
    last_success: float = 0
    last_failure: float = 0
    total_calls: int = 0
    total_failures: int = 0

    def record_success(self):
        self.consecutive_failures = 0
        self.last_success = time.time()
        self.total_calls += 1

    def record_failure(self):
        self.consecutive_failures += 1
        self.last_failure = time.time()
        self.total_calls += 1
        self.total_failures += 1

    @property
    def is_healthy(self) -> bool:
        return self.consecutive_failures < 3

    @property
    def failure_rate(self) -> float:
        if self.total_calls == 0:
            return 0
        return self.total_failures / self.total_calls

search_health = ToolHealthMonitor(tool_name="web_search")

def execute_search_with_health(query: str) -> str:
    """Execute search with health monitoring."""
    if not search_health.is_healthy:
        return "[Search tool temporarily unavailable. Answering from context only.]"

    result = execute_search(query)
    if "failed" in result.lower() or "error" in result.lower():
        search_health.record_failure()
    else:
        search_health.record_success()
    return result

def startup_health_check() -> dict:
    """Run on agent startup to verify all tools work."""
    checks = {}

    # Verify search tool
    try:
        result = execute_search("test query")
        checks["web_search"] = "healthy" if "Title:" in result else "degraded"
    except Exception as e:
        checks["web_search"] = f"failed: {str(e)}"

    # Verify API key is valid
    try:
        resp = requests.post("https://api.scavio.dev/api/v1/search",
            headers={"x-api-key": os.environ.get("SCAVIO_API_KEY", "")},
            json={"platform": "google", "query": "ping"},
            timeout=5)
        checks["api_auth"] = "valid" if resp.status_code == 200 else f"error: {resp.status_code}"
    except Exception:
        checks["api_auth"] = "unreachable"

    return checks