diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 000000000..17ad4e69e --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,28 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "hermes-agent" +version = "0.1.0" +description = "AI agent with advanced tool-calling and toolsets" +readme = "README.md" +requires-python = ">=3.10" +authors = [{ name = "Hermes Agent" }] +license = { text = "MIT" } +dependencies = [ + "firecrawl-py", + "openai", + "fal-client", + "python-dotenv", + "fire" +] + +[project.scripts] +hermes-agent = "run_agent:main" + +[tool.setuptools] +py-modules = ["run_agent", "model_tools", "toolsets"] + +[tool.setuptools.packages.find] +include = ["tools"] diff --git a/test_run.sh b/scripts/test_run.sh similarity index 69% rename from test_run.sh rename to scripts/test_run.sh index ff4ffc3c2..a559e120f 100644 --- a/test_run.sh +++ b/scripts/test_run.sh @@ -13,6 +13,16 @@ PROMPT="$1" # Set debug mode for web tools export WEB_TOOLS_DEBUG=true +# Resolve repository root relative to this script and run from there +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" +cd "$REPO_ROOT" + +# Prefer local venv if present +if [ -f "venv/bin/activate" ]; then + source venv/bin/activate +fi + # Run the agent with the provided prompt python run_agent.py \ --query "$PROMPT" \ diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/test_web_tools.py b/tests/test_web_tools.py similarity index 97% rename from test_web_tools.py rename to tests/test_web_tools.py index 7c86becb6..f6eea2c30 100644 --- a/test_web_tools.py +++ b/tests/test_web_tools.py @@ -23,8 +23,8 @@ import argparse from datetime import datetime from typing import List, Dict, Any -# Import the web tools to test -from web_tools import ( +# Import the web tools to test (updated path after moving tools/) +from tools.web_tools import ( web_search_tool, web_extract_tool, web_crawl_tool, diff --git a/tools/mixture_of_agents_tool.py b/tools/mixture_of_agents_tool.py index 206b14963..0e0ca741b 100644 --- a/tools/mixture_of_agents_tool.py +++ b/tools/mixture_of_agents_tool.py @@ -1,586 +1,586 @@ -#!/usr/bin/env python3 -""" -Mixture-of-Agents Tool Module - -This module implements the Mixture-of-Agents (MoA) methodology that leverages -the collective strengths of multiple LLMs through a layered architecture to -achieve state-of-the-art performance on complex reasoning tasks. - -Based on the research paper: "Mixture-of-Agents Enhances Large Language Model Capabilities" -by Junlin Wang et al. (arXiv:2406.04692v1) - -Key Features: -- Multi-layer LLM collaboration for enhanced reasoning -- Parallel processing of reference models for efficiency -- Intelligent aggregation and synthesis of diverse responses -- Specialized for extremely difficult problems requiring intense reasoning -- Optimized for coding, mathematics, and complex analytical tasks - -Available Tool: -- mixture_of_agents_tool: Process complex queries using multiple frontier models - -Architecture: -1. Reference models generate diverse initial responses in parallel -2. Aggregator model synthesizes responses into a high-quality output -3. Multiple layers can be used for iterative refinement (future enhancement) - -Models Used: -- Reference Models: claude-opus-4-20250514, gemini-2.5-pro, o4-mini, deepseek-r1 -- Aggregator Model: claude-opus-4-20250514 (highest capability for synthesis) - -Configuration: - To customize the MoA setup, modify the configuration constants at the top of this file: - - REFERENCE_MODELS: List of models for generating diverse initial responses - - AGGREGATOR_MODEL: Model used to synthesize the final response - - REFERENCE_TEMPERATURE/AGGREGATOR_TEMPERATURE: Sampling temperatures - - MIN_SUCCESSFUL_REFERENCES: Minimum successful models needed to proceed - -Usage: - from mixture_of_agents_tool import mixture_of_agents_tool - import asyncio - - # Process a complex query - result = await mixture_of_agents_tool( - user_prompt="Solve this complex mathematical proof..." - ) -""" - -import json -import os -import asyncio -import uuid -import datetime -from pathlib import Path -from typing import Dict, Any, List, Optional -from openai import AsyncOpenAI - -# Initialize Nous Research API client for MoA processing -nous_client = AsyncOpenAI( - api_key=os.getenv("NOUS_API_KEY"), - base_url="https://inference-api.nousresearch.com/v1" -) - -# Configuration for MoA processing -# Reference models - these generate diverse initial responses in parallel -REFERENCE_MODELS = [ - "claude-opus-4-20250514", - "gemini-2.5-pro", - "gpt-5", - "deepseek-r1" -] - -# Aggregator model - synthesizes reference responses into final output -AGGREGATOR_MODEL = "claude-opus-4-20250514" # Use highest capability model for aggregation - -# Temperature settings optimized for MoA performance -REFERENCE_TEMPERATURE = 0.6 # Balanced creativity for diverse perspectives -AGGREGATOR_TEMPERATURE = 0.4 # Focused synthesis for consistency - -# Failure handling configuration -MIN_SUCCESSFUL_REFERENCES = 1 # Minimum successful reference models needed to proceed - -# System prompt for the aggregator model (from the research paper) -AGGREGATOR_SYSTEM_PROMPT = """You have been provided with a set of responses from various open-source models to the latest user query. Your task is to synthesize these responses into a single, high-quality response. It is crucial to critically evaluate the information provided in these responses, recognizing that some of it may be biased or incorrect. Your response should not simply replicate the given answers but should offer a refined, accurate, and comprehensive reply to the instruction. Ensure your response is well-structured, coherent, and adheres to the highest standards of accuracy and reliability. - -Responses from models:""" - -# Debug mode configuration -DEBUG_MODE = os.getenv("MOA_TOOLS_DEBUG", "false").lower() == "true" -DEBUG_SESSION_ID = str(uuid.uuid4()) -DEBUG_LOG_PATH = Path("./logs") -DEBUG_DATA = { - "session_id": DEBUG_SESSION_ID, - "start_time": datetime.datetime.now().isoformat(), - "debug_enabled": DEBUG_MODE, - "tool_calls": [] -} if DEBUG_MODE else None - -# Create logs directory if debug mode is enabled -if DEBUG_MODE: - DEBUG_LOG_PATH.mkdir(exist_ok=True) - print(f"šŸ› MoA debug mode enabled - Session ID: {DEBUG_SESSION_ID}") - - -def _log_debug_call(tool_name: str, call_data: Dict[str, Any]) -> None: - """ - Log a debug call entry to the global debug data structure. - - Args: - tool_name (str): Name of the tool being called - call_data (Dict[str, Any]): Data about the call including parameters and results - """ - if not DEBUG_MODE or not DEBUG_DATA: - return - - call_entry = { - "timestamp": datetime.datetime.now().isoformat(), - "tool_name": tool_name, - **call_data - } - - DEBUG_DATA["tool_calls"].append(call_entry) - - -def _save_debug_log() -> None: - """ - Save the current debug data to a JSON file in the logs directory. - """ - if not DEBUG_MODE or not DEBUG_DATA: - return - - try: - debug_filename = f"moa_tools_debug_{DEBUG_SESSION_ID}.json" - debug_filepath = DEBUG_LOG_PATH / debug_filename - - # Update end time - DEBUG_DATA["end_time"] = datetime.datetime.now().isoformat() - DEBUG_DATA["total_calls"] = len(DEBUG_DATA["tool_calls"]) - - with open(debug_filepath, 'w', encoding='utf-8') as f: - json.dump(DEBUG_DATA, f, indent=2, ensure_ascii=False) - - print(f"šŸ› MoA debug log saved: {debug_filepath}") - - except Exception as e: - print(f"āŒ Error saving MoA debug log: {str(e)}") - - -def _construct_aggregator_prompt(system_prompt: str, responses: List[str]) -> str: - """ - Construct the final system prompt for the aggregator including all model responses. - - Args: - system_prompt (str): Base system prompt for aggregation - responses (List[str]): List of responses from reference models - - Returns: - str: Complete system prompt with enumerated responses - """ - response_text = "\n".join([f"{i+1}. {response}" for i, response in enumerate(responses)]) - return f"{system_prompt}\n\n{response_text}" - - -async def _run_reference_model_safe( - model: str, - user_prompt: str, - temperature: float = REFERENCE_TEMPERATURE, - max_tokens: int = 32000, - max_retries: int = 3 -) -> tuple[str, str, bool]: - """ - Run a single reference model with retry logic and graceful failure handling. - - Args: - model (str): Model identifier to use - user_prompt (str): The user's query - temperature (float): Sampling temperature for response generation - max_tokens (int): Maximum tokens in response - max_retries (int): Maximum number of retry attempts - - Returns: - tuple[str, str, bool]: (model_name, response_content_or_error, success_flag) - """ - for attempt in range(max_retries): - try: - print(f"šŸ¤– Querying {model} (attempt {attempt + 1}/{max_retries})") - - # Build parameters for the API call - api_params = { - "model": model, - "messages": [{"role": "user", "content": user_prompt}] - } - - # GPT models (especially gpt-4o-mini) don't support custom temperature values - # Only include temperature for non-GPT models - if not model.lower().startswith('gpt-'): - api_params["temperature"] = temperature - - response = await nous_client.chat.completions.create(**api_params) - - content = response.choices[0].message.content.strip() - print(f"āœ… {model} responded ({len(content)} characters)") - return model, content, True - - except Exception as e: - error_str = str(e) - # Log more detailed error information for debugging - if "invalid" in error_str.lower(): - print(f"āš ļø {model} invalid request error (attempt {attempt + 1}): {error_str}") - elif "rate" in error_str.lower() or "limit" in error_str.lower(): - print(f"āš ļø {model} rate limit error (attempt {attempt + 1}): {error_str}") - else: - print(f"āš ļø {model} unknown error (attempt {attempt + 1}): {error_str}") - - if attempt < max_retries - 1: - # Exponential backoff for rate limiting - sleep_time = 2 ** attempt - print(f" Retrying in {sleep_time}s...") - await asyncio.sleep(sleep_time) - else: - error_msg = f"{model} failed after {max_retries} attempts: {error_str}" - print(f"āŒ {error_msg}") - return model, error_msg, False - - -async def _run_aggregator_model( - system_prompt: str, - user_prompt: str, - temperature: float = AGGREGATOR_TEMPERATURE, - max_tokens: int = None -) -> str: - """ - Run the aggregator model to synthesize the final response. - - Args: - system_prompt (str): System prompt with all reference responses - user_prompt (str): Original user query - temperature (float): Focused temperature for consistent aggregation - max_tokens (int): Maximum tokens in final response - - Returns: - str: Synthesized final response - """ - print(f"šŸ§  Running aggregator model: {AGGREGATOR_MODEL}") - - # Build parameters for the API call - api_params = { - "model": AGGREGATOR_MODEL, - "messages": [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_prompt} - ] - } - - # GPT models (especially gpt-4o-mini) don't support custom temperature values - # Only include temperature for non-GPT models - if not AGGREGATOR_MODEL.lower().startswith('gpt-'): - api_params["temperature"] = temperature - - response = await nous_client.chat.completions.create(**api_params) - - content = response.choices[0].message.content.strip() - print(f"āœ… Aggregation complete ({len(content)} characters)") - return content - - -async def mixture_of_agents_tool( - user_prompt: str, - reference_models: Optional[List[str]] = None, - aggregator_model: Optional[str] = None -) -> str: - """ - Process a complex query using the Mixture-of-Agents methodology. - - This tool leverages multiple frontier language models to collaboratively solve - extremely difficult problems requiring intense reasoning. It's particularly - effective for: - - Complex mathematical proofs and calculations - - Advanced coding problems and algorithm design - - Multi-step analytical reasoning tasks - - Problems requiring diverse domain expertise - - Tasks where single models show limitations - - The MoA approach uses a fixed 2-layer architecture: - 1. Layer 1: Multiple reference models generate diverse responses in parallel (temp=0.6) - 2. Layer 2: Aggregator model synthesizes the best elements into final response (temp=0.4) - - Args: - user_prompt (str): The complex query or problem to solve - reference_models (Optional[List[str]]): Custom reference models to use - aggregator_model (Optional[str]): Custom aggregator model to use - - Returns: - str: JSON string containing the MoA results with the following structure: - { - "success": bool, - "response": str, - "models_used": { - "reference_models": List[str], - "aggregator_model": str - }, - "processing_time": float - } - - Raises: - Exception: If MoA processing fails or API key is not set - """ - start_time = datetime.datetime.now() - - debug_call_data = { - "parameters": { - "user_prompt": user_prompt[:200] + "..." if len(user_prompt) > 200 else user_prompt, - "reference_models": reference_models or REFERENCE_MODELS, - "aggregator_model": aggregator_model or AGGREGATOR_MODEL, - "reference_temperature": REFERENCE_TEMPERATURE, - "aggregator_temperature": AGGREGATOR_TEMPERATURE, - "min_successful_references": MIN_SUCCESSFUL_REFERENCES - }, - "error": None, - "success": False, - "reference_responses_count": 0, - "failed_models_count": 0, - "failed_models": [], - "final_response_length": 0, - "processing_time_seconds": 0, - "models_used": {} - } - - try: - print(f"šŸš€ Starting Mixture-of-Agents processing...") - print(f"šŸ“ Query: {user_prompt[:100]}{'...' if len(user_prompt) > 100 else ''}") - - # Validate API key availability - if not os.getenv("NOUS_API_KEY"): - raise ValueError("NOUS_API_KEY environment variable not set") - - # Use provided models or defaults - ref_models = reference_models or REFERENCE_MODELS - agg_model = aggregator_model or AGGREGATOR_MODEL - - print(f"šŸ”„ Using {len(ref_models)} reference models in 2-layer MoA architecture") - - # Layer 1: Generate diverse responses from reference models (with failure handling) - print("šŸ“” Layer 1: Generating reference responses...") - model_results = await asyncio.gather(*[ - _run_reference_model_safe(model, user_prompt, REFERENCE_TEMPERATURE) - for model in ref_models - ]) - - # Separate successful and failed responses - successful_responses = [] - failed_models = [] - - for model_name, content, success in model_results: - if success: - successful_responses.append(content) - else: - failed_models.append(model_name) - - successful_count = len(successful_responses) - failed_count = len(failed_models) - - print(f"šŸ“Š Reference model results: {successful_count} successful, {failed_count} failed") - - if failed_models: - print(f"āš ļø Failed models: {', '.join(failed_models)}") - - # Check if we have enough successful responses to proceed - if successful_count < MIN_SUCCESSFUL_REFERENCES: - raise ValueError(f"Insufficient successful reference models ({successful_count}/{len(ref_models)}). Need at least {MIN_SUCCESSFUL_REFERENCES} successful responses.") - - debug_call_data["reference_responses_count"] = successful_count - debug_call_data["failed_models_count"] = failed_count - debug_call_data["failed_models"] = failed_models - - # Layer 2: Aggregate responses using the aggregator model - print("šŸ§  Layer 2: Synthesizing final response...") - aggregator_system_prompt = _construct_aggregator_prompt( - AGGREGATOR_SYSTEM_PROMPT, - successful_responses - ) - - final_response = await _run_aggregator_model( - aggregator_system_prompt, - user_prompt, - AGGREGATOR_TEMPERATURE - ) - - # Calculate processing time - end_time = datetime.datetime.now() - processing_time = (end_time - start_time).total_seconds() - - print(f"āœ… MoA processing completed in {processing_time:.2f} seconds") - - # Prepare successful response (only final aggregated result, minimal fields) - result = { - "success": True, - "response": final_response, - "models_used": { - "reference_models": ref_models, - "aggregator_model": agg_model - } - } - - debug_call_data["success"] = True - debug_call_data["final_response_length"] = len(final_response) - debug_call_data["processing_time_seconds"] = processing_time - debug_call_data["models_used"] = result["models_used"] - - # Log debug information - _log_debug_call("mixture_of_agents_tool", debug_call_data) - _save_debug_log() - - return json.dumps(result, indent=2) - - except Exception as e: - error_msg = f"Error in MoA processing: {str(e)}" - print(f"āŒ {error_msg}") - - # Calculate processing time even for errors - end_time = datetime.datetime.now() - processing_time = (end_time - start_time).total_seconds() - - # Prepare error response (minimal fields) - result = { - "success": False, - "response": "MoA processing failed. Please try again or use a single model for this query.", - "models_used": { - "reference_models": reference_models or REFERENCE_MODELS, - "aggregator_model": aggregator_model or AGGREGATOR_MODEL - }, - "error": error_msg - } - - debug_call_data["error"] = error_msg - debug_call_data["processing_time_seconds"] = processing_time - _log_debug_call("mixture_of_agents_tool", debug_call_data) - _save_debug_log() - - return json.dumps(result, indent=2) - - -def check_nous_api_key() -> bool: - """ - Check if the Nous Research API key is available in environment variables. - - Returns: - bool: True if API key is set, False otherwise - """ - return bool(os.getenv("NOUS_API_KEY")) - - -def check_moa_requirements() -> bool: - """ - Check if all requirements for MoA tools are met. - - Returns: - bool: True if requirements are met, False otherwise - """ - return check_nous_api_key() - - -def get_debug_session_info() -> Dict[str, Any]: - """ - Get information about the current debug session. - - Returns: - Dict[str, Any]: Dictionary containing debug session information - """ - if not DEBUG_MODE or not DEBUG_DATA: - return { - "enabled": False, - "session_id": None, - "log_path": None, - "total_calls": 0 - } - - return { - "enabled": True, - "session_id": DEBUG_SESSION_ID, - "log_path": str(DEBUG_LOG_PATH / f"moa_tools_debug_{DEBUG_SESSION_ID}.json"), - "total_calls": len(DEBUG_DATA["tool_calls"]) - } - - -def get_available_models() -> Dict[str, List[str]]: - """ - Get information about available models for MoA processing. - - Returns: - Dict[str, List[str]]: Dictionary with reference and aggregator models - """ - return { - "reference_models": REFERENCE_MODELS, - "aggregator_models": [AGGREGATOR_MODEL], - "supported_models": REFERENCE_MODELS + [AGGREGATOR_MODEL] - } - - -def get_moa_configuration() -> Dict[str, Any]: - """ - Get the current MoA configuration settings. - - Returns: - Dict[str, Any]: Dictionary containing all configuration parameters - """ - return { - "reference_models": REFERENCE_MODELS, - "aggregator_model": AGGREGATOR_MODEL, - "reference_temperature": REFERENCE_TEMPERATURE, - "aggregator_temperature": AGGREGATOR_TEMPERATURE, - "min_successful_references": MIN_SUCCESSFUL_REFERENCES, - "total_reference_models": len(REFERENCE_MODELS), - "failure_tolerance": f"{len(REFERENCE_MODELS) - MIN_SUCCESSFUL_REFERENCES}/{len(REFERENCE_MODELS)} models can fail" - } - - -if __name__ == "__main__": - """ - Simple test/demo when run directly - """ - print("šŸ¤– Mixture-of-Agents Tool Module") - print("=" * 50) - - # Check if API key is available - api_available = check_nous_api_key() - - if not api_available: - print("āŒ NOUS_API_KEY environment variable not set") - print("Please set your API key: export NOUS_API_KEY='your-key-here'") - print("Get API key at: https://inference-api.nousresearch.com/") - exit(1) - else: - print("āœ… Nous Research API key found") - - print("šŸ› ļø MoA tools ready for use!") - - # Show current configuration - config = get_moa_configuration() - print(f"\nāš™ļø Current Configuration:") - print(f" šŸ¤– Reference models ({len(config['reference_models'])}): {', '.join(config['reference_models'])}") - print(f" šŸ§  Aggregator model: {config['aggregator_model']}") - print(f" šŸŒ”ļø Reference temperature: {config['reference_temperature']}") - print(f" šŸŒ”ļø Aggregator temperature: {config['aggregator_temperature']}") - print(f" šŸ›”ļø Failure tolerance: {config['failure_tolerance']}") - print(f" šŸ“Š Minimum successful models: {config['min_successful_references']}") - - # Show debug mode status - if DEBUG_MODE: - print(f"\nšŸ› Debug mode ENABLED - Session ID: {DEBUG_SESSION_ID}") - print(f" Debug logs will be saved to: ./logs/moa_tools_debug_{DEBUG_SESSION_ID}.json") - else: - print("\nšŸ› Debug mode disabled (set MOA_TOOLS_DEBUG=true to enable)") - - print("\nBasic usage:") - print(" from mixture_of_agents_tool import mixture_of_agents_tool") - print(" import asyncio") - print("") - print(" async def main():") - print(" result = await mixture_of_agents_tool(") - print(" user_prompt='Solve this complex mathematical proof...'") - print(" )") - print(" print(result)") - print(" asyncio.run(main())") - - print("\nBest use cases:") - print(" - Complex mathematical proofs and calculations") - print(" - Advanced coding problems and algorithm design") - print(" - Multi-step analytical reasoning tasks") - print(" - Problems requiring diverse domain expertise") - print(" - Tasks where single models show limitations") - - print("\nPerformance characteristics:") - print(" - Higher latency due to multiple model calls") - print(" - Significantly improved quality for complex tasks") - print(" - Parallel processing for efficiency") - print(f" - Optimized temperatures: {REFERENCE_TEMPERATURE} for reference models, {AGGREGATOR_TEMPERATURE} for aggregation") - print(" - Token-efficient: only returns final aggregated response") - print(" - Resilient: continues with partial model failures") - print(f" - Configurable: easy to modify models and settings at top of file") - print(" - State-of-the-art results on challenging benchmarks") - - print("\nDebug mode:") - print(" # Enable debug logging") - print(" export MOA_TOOLS_DEBUG=true") - print(" # Debug logs capture all MoA processing steps and metrics") - print(" # Logs saved to: ./logs/moa_tools_debug_UUID.json") +#!/usr/bin/env python3 +""" +Mixture-of-Agents Tool Module + +This module implements the Mixture-of-Agents (MoA) methodology that leverages +the collective strengths of multiple LLMs through a layered architecture to +achieve state-of-the-art performance on complex reasoning tasks. + +Based on the research paper: "Mixture-of-Agents Enhances Large Language Model Capabilities" +by Junlin Wang et al. (arXiv:2406.04692v1) + +Key Features: +- Multi-layer LLM collaboration for enhanced reasoning +- Parallel processing of reference models for efficiency +- Intelligent aggregation and synthesis of diverse responses +- Specialized for extremely difficult problems requiring intense reasoning +- Optimized for coding, mathematics, and complex analytical tasks + +Available Tool: +- mixture_of_agents_tool: Process complex queries using multiple frontier models + +Architecture: +1. Reference models generate diverse initial responses in parallel +2. Aggregator model synthesizes responses into a high-quality output +3. Multiple layers can be used for iterative refinement (future enhancement) + +Models Used: +- Reference Models: claude-opus-4-20250514, gemini-2.5-pro, o4-mini, deepseek-r1 +- Aggregator Model: claude-opus-4-20250514 (highest capability for synthesis) + +Configuration: + To customize the MoA setup, modify the configuration constants at the top of this file: + - REFERENCE_MODELS: List of models for generating diverse initial responses + - AGGREGATOR_MODEL: Model used to synthesize the final response + - REFERENCE_TEMPERATURE/AGGREGATOR_TEMPERATURE: Sampling temperatures + - MIN_SUCCESSFUL_REFERENCES: Minimum successful models needed to proceed + +Usage: + from mixture_of_agents_tool import mixture_of_agents_tool + import asyncio + + # Process a complex query + result = await mixture_of_agents_tool( + user_prompt="Solve this complex mathematical proof..." + ) +""" + +import json +import os +import asyncio +import uuid +import datetime +from pathlib import Path +from typing import Dict, Any, List, Optional +from openai import AsyncOpenAI + +# Initialize Nous Research API client for MoA processing +nous_client = AsyncOpenAI( + api_key=os.getenv("NOUS_API_KEY"), + base_url="https://inference-api.nousresearch.com/v1" +) + +# Configuration for MoA processing +# Reference models - these generate diverse initial responses in parallel +REFERENCE_MODELS = [ + "claude-opus-4-20250514", + "gemini-2.5-pro", + "gpt-5", + "deepseek-r1" +] + +# Aggregator model - synthesizes reference responses into final output +AGGREGATOR_MODEL = "claude-opus-4-20250514" # Use highest capability model for aggregation + +# Temperature settings optimized for MoA performance +REFERENCE_TEMPERATURE = 0.6 # Balanced creativity for diverse perspectives +AGGREGATOR_TEMPERATURE = 0.4 # Focused synthesis for consistency + +# Failure handling configuration +MIN_SUCCESSFUL_REFERENCES = 1 # Minimum successful reference models needed to proceed + +# System prompt for the aggregator model (from the research paper) +AGGREGATOR_SYSTEM_PROMPT = """You have been provided with a set of responses from various open-source models to the latest user query. Your task is to synthesize these responses into a single, high-quality response. It is crucial to critically evaluate the information provided in these responses, recognizing that some of it may be biased or incorrect. Your response should not simply replicate the given answers but should offer a refined, accurate, and comprehensive reply to the instruction. Ensure your response is well-structured, coherent, and adheres to the highest standards of accuracy and reliability. + +Responses from models:""" + +# Debug mode configuration +DEBUG_MODE = os.getenv("MOA_TOOLS_DEBUG", "false").lower() == "true" +DEBUG_SESSION_ID = str(uuid.uuid4()) +DEBUG_LOG_PATH = Path("./logs") +DEBUG_DATA = { + "session_id": DEBUG_SESSION_ID, + "start_time": datetime.datetime.now().isoformat(), + "debug_enabled": DEBUG_MODE, + "tool_calls": [] +} if DEBUG_MODE else None + +# Create logs directory if debug mode is enabled +if DEBUG_MODE: + DEBUG_LOG_PATH.mkdir(exist_ok=True) + print(f"šŸ› MoA debug mode enabled - Session ID: {DEBUG_SESSION_ID}") + + +def _log_debug_call(tool_name: str, call_data: Dict[str, Any]) -> None: + """ + Log a debug call entry to the global debug data structure. + + Args: + tool_name (str): Name of the tool being called + call_data (Dict[str, Any]): Data about the call including parameters and results + """ + if not DEBUG_MODE or not DEBUG_DATA: + return + + call_entry = { + "timestamp": datetime.datetime.now().isoformat(), + "tool_name": tool_name, + **call_data + } + + DEBUG_DATA["tool_calls"].append(call_entry) + + +def _save_debug_log() -> None: + """ + Save the current debug data to a JSON file in the logs directory. + """ + if not DEBUG_MODE or not DEBUG_DATA: + return + + try: + debug_filename = f"moa_tools_debug_{DEBUG_SESSION_ID}.json" + debug_filepath = DEBUG_LOG_PATH / debug_filename + + # Update end time + DEBUG_DATA["end_time"] = datetime.datetime.now().isoformat() + DEBUG_DATA["total_calls"] = len(DEBUG_DATA["tool_calls"]) + + with open(debug_filepath, 'w', encoding='utf-8') as f: + json.dump(DEBUG_DATA, f, indent=2, ensure_ascii=False) + + print(f"šŸ› MoA debug log saved: {debug_filepath}") + + except Exception as e: + print(f"āŒ Error saving MoA debug log: {str(e)}") + + +def _construct_aggregator_prompt(system_prompt: str, responses: List[str]) -> str: + """ + Construct the final system prompt for the aggregator including all model responses. + + Args: + system_prompt (str): Base system prompt for aggregation + responses (List[str]): List of responses from reference models + + Returns: + str: Complete system prompt with enumerated responses + """ + response_text = "\n".join([f"{i+1}. {response}" for i, response in enumerate(responses)]) + return f"{system_prompt}\n\n{response_text}" + + +async def _run_reference_model_safe( + model: str, + user_prompt: str, + temperature: float = REFERENCE_TEMPERATURE, + max_tokens: int = 32000, + max_retries: int = 3 +) -> tuple[str, str, bool]: + """ + Run a single reference model with retry logic and graceful failure handling. + + Args: + model (str): Model identifier to use + user_prompt (str): The user's query + temperature (float): Sampling temperature for response generation + max_tokens (int): Maximum tokens in response + max_retries (int): Maximum number of retry attempts + + Returns: + tuple[str, str, bool]: (model_name, response_content_or_error, success_flag) + """ + for attempt in range(max_retries): + try: + print(f"šŸ¤– Querying {model} (attempt {attempt + 1}/{max_retries})") + + # Build parameters for the API call + api_params = { + "model": model, + "messages": [{"role": "user", "content": user_prompt}] + } + + # GPT models (especially gpt-4o-mini) don't support custom temperature values + # Only include temperature for non-GPT models + if not model.lower().startswith('gpt-'): + api_params["temperature"] = temperature + + response = await nous_client.chat.completions.create(**api_params) + + content = response.choices[0].message.content.strip() + print(f"āœ… {model} responded ({len(content)} characters)") + return model, content, True + + except Exception as e: + error_str = str(e) + # Log more detailed error information for debugging + if "invalid" in error_str.lower(): + print(f"āš ļø {model} invalid request error (attempt {attempt + 1}): {error_str}") + elif "rate" in error_str.lower() or "limit" in error_str.lower(): + print(f"āš ļø {model} rate limit error (attempt {attempt + 1}): {error_str}") + else: + print(f"āš ļø {model} unknown error (attempt {attempt + 1}): {error_str}") + + if attempt < max_retries - 1: + # Exponential backoff for rate limiting + sleep_time = 2 ** attempt + print(f" Retrying in {sleep_time}s...") + await asyncio.sleep(sleep_time) + else: + error_msg = f"{model} failed after {max_retries} attempts: {error_str}" + print(f"āŒ {error_msg}") + return model, error_msg, False + + +async def _run_aggregator_model( + system_prompt: str, + user_prompt: str, + temperature: float = AGGREGATOR_TEMPERATURE, + max_tokens: int = None +) -> str: + """ + Run the aggregator model to synthesize the final response. + + Args: + system_prompt (str): System prompt with all reference responses + user_prompt (str): Original user query + temperature (float): Focused temperature for consistent aggregation + max_tokens (int): Maximum tokens in final response + + Returns: + str: Synthesized final response + """ + print(f"šŸ§  Running aggregator model: {AGGREGATOR_MODEL}") + + # Build parameters for the API call + api_params = { + "model": AGGREGATOR_MODEL, + "messages": [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt} + ] + } + + # GPT models (especially gpt-4o-mini) don't support custom temperature values + # Only include temperature for non-GPT models + if not AGGREGATOR_MODEL.lower().startswith('gpt-'): + api_params["temperature"] = temperature + + response = await nous_client.chat.completions.create(**api_params) + + content = response.choices[0].message.content.strip() + print(f"āœ… Aggregation complete ({len(content)} characters)") + return content + + +async def mixture_of_agents_tool( + user_prompt: str, + reference_models: Optional[List[str]] = None, + aggregator_model: Optional[str] = None +) -> str: + """ + Process a complex query using the Mixture-of-Agents methodology. + + This tool leverages multiple frontier language models to collaboratively solve + extremely difficult problems requiring intense reasoning. It's particularly + effective for: + - Complex mathematical proofs and calculations + - Advanced coding problems and algorithm design + - Multi-step analytical reasoning tasks + - Problems requiring diverse domain expertise + - Tasks where single models show limitations + + The MoA approach uses a fixed 2-layer architecture: + 1. Layer 1: Multiple reference models generate diverse responses in parallel (temp=0.6) + 2. Layer 2: Aggregator model synthesizes the best elements into final response (temp=0.4) + + Args: + user_prompt (str): The complex query or problem to solve + reference_models (Optional[List[str]]): Custom reference models to use + aggregator_model (Optional[str]): Custom aggregator model to use + + Returns: + str: JSON string containing the MoA results with the following structure: + { + "success": bool, + "response": str, + "models_used": { + "reference_models": List[str], + "aggregator_model": str + }, + "processing_time": float + } + + Raises: + Exception: If MoA processing fails or API key is not set + """ + start_time = datetime.datetime.now() + + debug_call_data = { + "parameters": { + "user_prompt": user_prompt[:200] + "..." if len(user_prompt) > 200 else user_prompt, + "reference_models": reference_models or REFERENCE_MODELS, + "aggregator_model": aggregator_model or AGGREGATOR_MODEL, + "reference_temperature": REFERENCE_TEMPERATURE, + "aggregator_temperature": AGGREGATOR_TEMPERATURE, + "min_successful_references": MIN_SUCCESSFUL_REFERENCES + }, + "error": None, + "success": False, + "reference_responses_count": 0, + "failed_models_count": 0, + "failed_models": [], + "final_response_length": 0, + "processing_time_seconds": 0, + "models_used": {} + } + + try: + print(f"šŸš€ Starting Mixture-of-Agents processing...") + print(f"šŸ“ Query: {user_prompt[:100]}{'...' if len(user_prompt) > 100 else ''}") + + # Validate API key availability + if not os.getenv("NOUS_API_KEY"): + raise ValueError("NOUS_API_KEY environment variable not set") + + # Use provided models or defaults + ref_models = reference_models or REFERENCE_MODELS + agg_model = aggregator_model or AGGREGATOR_MODEL + + print(f"šŸ”„ Using {len(ref_models)} reference models in 2-layer MoA architecture") + + # Layer 1: Generate diverse responses from reference models (with failure handling) + print("šŸ“” Layer 1: Generating reference responses...") + model_results = await asyncio.gather(*[ + _run_reference_model_safe(model, user_prompt, REFERENCE_TEMPERATURE) + for model in ref_models + ]) + + # Separate successful and failed responses + successful_responses = [] + failed_models = [] + + for model_name, content, success in model_results: + if success: + successful_responses.append(content) + else: + failed_models.append(model_name) + + successful_count = len(successful_responses) + failed_count = len(failed_models) + + print(f"šŸ“Š Reference model results: {successful_count} successful, {failed_count} failed") + + if failed_models: + print(f"āš ļø Failed models: {', '.join(failed_models)}") + + # Check if we have enough successful responses to proceed + if successful_count < MIN_SUCCESSFUL_REFERENCES: + raise ValueError(f"Insufficient successful reference models ({successful_count}/{len(ref_models)}). Need at least {MIN_SUCCESSFUL_REFERENCES} successful responses.") + + debug_call_data["reference_responses_count"] = successful_count + debug_call_data["failed_models_count"] = failed_count + debug_call_data["failed_models"] = failed_models + + # Layer 2: Aggregate responses using the aggregator model + print("šŸ§  Layer 2: Synthesizing final response...") + aggregator_system_prompt = _construct_aggregator_prompt( + AGGREGATOR_SYSTEM_PROMPT, + successful_responses + ) + + final_response = await _run_aggregator_model( + aggregator_system_prompt, + user_prompt, + AGGREGATOR_TEMPERATURE + ) + + # Calculate processing time + end_time = datetime.datetime.now() + processing_time = (end_time - start_time).total_seconds() + + print(f"āœ… MoA processing completed in {processing_time:.2f} seconds") + + # Prepare successful response (only final aggregated result, minimal fields) + result = { + "success": True, + "response": final_response, + "models_used": { + "reference_models": ref_models, + "aggregator_model": agg_model + } + } + + debug_call_data["success"] = True + debug_call_data["final_response_length"] = len(final_response) + debug_call_data["processing_time_seconds"] = processing_time + debug_call_data["models_used"] = result["models_used"] + + # Log debug information + _log_debug_call("mixture_of_agents_tool", debug_call_data) + _save_debug_log() + + return json.dumps(result, indent=2) + + except Exception as e: + error_msg = f"Error in MoA processing: {str(e)}" + print(f"āŒ {error_msg}") + + # Calculate processing time even for errors + end_time = datetime.datetime.now() + processing_time = (end_time - start_time).total_seconds() + + # Prepare error response (minimal fields) + result = { + "success": False, + "response": "MoA processing failed. Please try again or use a single model for this query.", + "models_used": { + "reference_models": reference_models or REFERENCE_MODELS, + "aggregator_model": aggregator_model or AGGREGATOR_MODEL + }, + "error": error_msg + } + + debug_call_data["error"] = error_msg + debug_call_data["processing_time_seconds"] = processing_time + _log_debug_call("mixture_of_agents_tool", debug_call_data) + _save_debug_log() + + return json.dumps(result, indent=2) + + +def check_nous_api_key() -> bool: + """ + Check if the Nous Research API key is available in environment variables. + + Returns: + bool: True if API key is set, False otherwise + """ + return bool(os.getenv("NOUS_API_KEY")) + + +def check_moa_requirements() -> bool: + """ + Check if all requirements for MoA tools are met. + + Returns: + bool: True if requirements are met, False otherwise + """ + return check_nous_api_key() + + +def get_debug_session_info() -> Dict[str, Any]: + """ + Get information about the current debug session. + + Returns: + Dict[str, Any]: Dictionary containing debug session information + """ + if not DEBUG_MODE or not DEBUG_DATA: + return { + "enabled": False, + "session_id": None, + "log_path": None, + "total_calls": 0 + } + + return { + "enabled": True, + "session_id": DEBUG_SESSION_ID, + "log_path": str(DEBUG_LOG_PATH / f"moa_tools_debug_{DEBUG_SESSION_ID}.json"), + "total_calls": len(DEBUG_DATA["tool_calls"]) + } + + +def get_available_models() -> Dict[str, List[str]]: + """ + Get information about available models for MoA processing. + + Returns: + Dict[str, List[str]]: Dictionary with reference and aggregator models + """ + return { + "reference_models": REFERENCE_MODELS, + "aggregator_models": [AGGREGATOR_MODEL], + "supported_models": REFERENCE_MODELS + [AGGREGATOR_MODEL] + } + + +def get_moa_configuration() -> Dict[str, Any]: + """ + Get the current MoA configuration settings. + + Returns: + Dict[str, Any]: Dictionary containing all configuration parameters + """ + return { + "reference_models": REFERENCE_MODELS, + "aggregator_model": AGGREGATOR_MODEL, + "reference_temperature": REFERENCE_TEMPERATURE, + "aggregator_temperature": AGGREGATOR_TEMPERATURE, + "min_successful_references": MIN_SUCCESSFUL_REFERENCES, + "total_reference_models": len(REFERENCE_MODELS), + "failure_tolerance": f"{len(REFERENCE_MODELS) - MIN_SUCCESSFUL_REFERENCES}/{len(REFERENCE_MODELS)} models can fail" + } + + +if __name__ == "__main__": + """ + Simple test/demo when run directly + """ + print("šŸ¤– Mixture-of-Agents Tool Module") + print("=" * 50) + + # Check if API key is available + api_available = check_nous_api_key() + + if not api_available: + print("āŒ NOUS_API_KEY environment variable not set") + print("Please set your API key: export NOUS_API_KEY='your-key-here'") + print("Get API key at: https://inference-api.nousresearch.com/") + exit(1) + else: + print("āœ… Nous Research API key found") + + print("šŸ› ļø MoA tools ready for use!") + + # Show current configuration + config = get_moa_configuration() + print(f"\nāš™ļø Current Configuration:") + print(f" šŸ¤– Reference models ({len(config['reference_models'])}): {', '.join(config['reference_models'])}") + print(f" šŸ§  Aggregator model: {config['aggregator_model']}") + print(f" šŸŒ”ļø Reference temperature: {config['reference_temperature']}") + print(f" šŸŒ”ļø Aggregator temperature: {config['aggregator_temperature']}") + print(f" šŸ›”ļø Failure tolerance: {config['failure_tolerance']}") + print(f" šŸ“Š Minimum successful models: {config['min_successful_references']}") + + # Show debug mode status + if DEBUG_MODE: + print(f"\nšŸ› Debug mode ENABLED - Session ID: {DEBUG_SESSION_ID}") + print(f" Debug logs will be saved to: ./logs/moa_tools_debug_{DEBUG_SESSION_ID}.json") + else: + print("\nšŸ› Debug mode disabled (set MOA_TOOLS_DEBUG=true to enable)") + + print("\nBasic usage:") + print(" from mixture_of_agents_tool import mixture_of_agents_tool") + print(" import asyncio") + print("") + print(" async def main():") + print(" result = await mixture_of_agents_tool(") + print(" user_prompt='Solve this complex mathematical proof...'") + print(" )") + print(" print(result)") + print(" asyncio.run(main())") + + print("\nBest use cases:") + print(" - Complex mathematical proofs and calculations") + print(" - Advanced coding problems and algorithm design") + print(" - Multi-step analytical reasoning tasks") + print(" - Problems requiring diverse domain expertise") + print(" - Tasks where single models show limitations") + + print("\nPerformance characteristics:") + print(" - Higher latency due to multiple model calls") + print(" - Significantly improved quality for complex tasks") + print(" - Parallel processing for efficiency") + print(f" - Optimized temperatures: {REFERENCE_TEMPERATURE} for reference models, {AGGREGATOR_TEMPERATURE} for aggregation") + print(" - Token-efficient: only returns final aggregated response") + print(" - Resilient: continues with partial model failures") + print(f" - Configurable: easy to modify models and settings at top of file") + print(" - State-of-the-art results on challenging benchmarks") + + print("\nDebug mode:") + print(" # Enable debug logging") + print(" export MOA_TOOLS_DEBUG=true") + print(" # Debug logs capture all MoA processing steps and metrics") + print(" # Logs saved to: ./logs/moa_tools_debug_UUID.json") diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index e01d7a617..a17eae4f1 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -22,8 +22,6 @@ Usage: import json import os from typing import Optional, Dict, Any -from hecate import run_tool_with_lifecycle_management -from morphcloud._llm import ToolCall # Detailed description for the terminal tool based on Hermes Terminal system prompt TERMINAL_TOOL_DESCRIPTION = """Execute commands on a secure, persistent Linux VM environment with full interactive application support. @@ -114,6 +112,22 @@ def terminal_tool( >>> result = terminal_tool(command="sleep 60", background=True) """ try: + # Import hecate and ToolCall lazily so this module can be imported + # even when hecate is not installed. If unavailable, gracefully + # indicate that the terminal tool is disabled. + try: + from hecate import run_tool_with_lifecycle_management + from morphcloud._llm import ToolCall + except ImportError: + return json.dumps({ + "output": "", + "screen": "", + "session_id": None, + "exit_code": -1, + "error": "Terminal tool is disabled: 'hecate' is not installed. Install with: pip install hecate", + "status": "disabled" + }) + # Build tool input based on provided parameters tool_input = {} diff --git a/tools/vision_tools.py b/tools/vision_tools.py index 3183713bd..737b8d495 100644 --- a/tools/vision_tools.py +++ b/tools/vision_tools.py @@ -1,346 +1,346 @@ -#!/usr/bin/env python3 -""" -Vision Tools Module - -This module provides vision analysis tools that work with image URLs. -Uses Gemini Flash via Nous Research API for intelligent image understanding. - -Available tools: -- vision_analyze_tool: Analyze images from URLs with custom prompts - -Features: -- Comprehensive image description -- Context-aware analysis based on user queries -- Proper error handling and validation -- Debug logging support - -Usage: - from vision_tools import vision_analyze_tool - import asyncio - - # Analyze an image - result = await vision_analyze_tool( - image_url="https://example.com/image.jpg", - user_prompt="What architectural style is this building?" - ) -""" - -import json -import os -import asyncio -import uuid -import datetime -from pathlib import Path -from typing import Dict, Any, Optional -from openai import AsyncOpenAI - -# Initialize Nous Research API client for vision processing -nous_client = AsyncOpenAI( - api_key=os.getenv("NOUS_API_KEY"), - base_url="https://inference-api.nousresearch.com/v1" -) - -# Configuration for vision processing -DEFAULT_VISION_MODEL = "gemini-2.5-flash" - -# Debug mode configuration -DEBUG_MODE = os.getenv("VISION_TOOLS_DEBUG", "false").lower() == "true" -DEBUG_SESSION_ID = str(uuid.uuid4()) -DEBUG_LOG_PATH = Path("./logs") -DEBUG_DATA = { - "session_id": DEBUG_SESSION_ID, - "start_time": datetime.datetime.now().isoformat(), - "debug_enabled": DEBUG_MODE, - "tool_calls": [] -} if DEBUG_MODE else None - -# Create logs directory if debug mode is enabled -if DEBUG_MODE: - DEBUG_LOG_PATH.mkdir(exist_ok=True) - print(f"šŸ› Vision debug mode enabled - Session ID: {DEBUG_SESSION_ID}") - - -def _log_debug_call(tool_name: str, call_data: Dict[str, Any]) -> None: - """ - Log a debug call entry to the global debug data structure. - - Args: - tool_name (str): Name of the tool being called - call_data (Dict[str, Any]): Data about the call including parameters and results - """ - if not DEBUG_MODE or not DEBUG_DATA: - return - - call_entry = { - "timestamp": datetime.datetime.now().isoformat(), - "tool_name": tool_name, - **call_data - } - - DEBUG_DATA["tool_calls"].append(call_entry) - - -def _save_debug_log() -> None: - """ - Save the current debug data to a JSON file in the logs directory. - """ - if not DEBUG_MODE or not DEBUG_DATA: - return - - try: - debug_filename = f"vision_tools_debug_{DEBUG_SESSION_ID}.json" - debug_filepath = DEBUG_LOG_PATH / debug_filename - - # Update end time - DEBUG_DATA["end_time"] = datetime.datetime.now().isoformat() - DEBUG_DATA["total_calls"] = len(DEBUG_DATA["tool_calls"]) - - with open(debug_filepath, 'w', encoding='utf-8') as f: - json.dump(DEBUG_DATA, f, indent=2, ensure_ascii=False) - - print(f"šŸ› Vision debug log saved: {debug_filepath}") - - except Exception as e: - print(f"āŒ Error saving vision debug log: {str(e)}") - - -def _validate_image_url(url: str) -> bool: - """ - Basic validation of image URL format. - - Args: - url (str): The URL to validate - - Returns: - bool: True if URL appears to be valid, False otherwise - """ - if not url or not isinstance(url, str): - return False - - # Check if it's a valid URL format - if not (url.startswith('http://') or url.startswith('https://')): - return False - - # Check for common image extensions (optional, as URLs may not have extensions) - image_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp', '.svg'] - - return True # Allow all HTTP/HTTPS URLs for flexibility - - -async def vision_analyze_tool( - image_url: str, - user_prompt: str, - model: str = DEFAULT_VISION_MODEL -) -> str: - """ - Analyze an image from a URL using vision AI. - - This tool processes images using Gemini Flash via Nous Research API. - The user_prompt parameter is expected to be pre-formatted by the calling - function (typically model_tools.py) to include both full description - requests and specific questions. - - Args: - image_url (str): The URL of the image to analyze - user_prompt (str): The pre-formatted prompt for the vision model - model (str): The vision model to use (default: gemini-2.5-flash) - - Returns: - str: JSON string containing the analysis results with the following structure: - { - "success": bool, - "analysis": str (defaults to error message if None) - } - - Raises: - Exception: If analysis fails or API key is not set - """ - debug_call_data = { - "parameters": { - "image_url": image_url, - "user_prompt": user_prompt, - "model": model - }, - "error": None, - "success": False, - "analysis_length": 0, - "model_used": model - } - - try: - print(f"šŸ” Analyzing image from URL: {image_url[:60]}{'...' if len(image_url) > 60 else ''}") - print(f"šŸ“ User prompt: {user_prompt[:100]}{'...' if len(user_prompt) > 100 else ''}") - - # Validate image URL - if not _validate_image_url(image_url): - raise ValueError("Invalid image URL format. Must start with http:// or https://") - - # Check API key availability - if not os.getenv("NOUS_API_KEY"): - raise ValueError("NOUS_API_KEY environment variable not set") - - # Use the prompt as provided (model_tools.py now handles full description formatting) - comprehensive_prompt = user_prompt - - # Prepare the message with image URL format - messages = [ - { - "role": "user", - "content": [ - { - "type": "text", - "text": comprehensive_prompt - }, - { - "type": "image_url", - "image_url": { - "url": image_url - } - } - ] - } - ] - - print(f"šŸ§  Processing image with {model}...") - - # Call the vision API - response = await nous_client.chat.completions.create( - model=model, - messages=messages, - temperature=0.1, # Low temperature for consistent analysis - max_tokens=2000 # Generous limit for detailed analysis - ) - - # Extract the analysis - analysis = response.choices[0].message.content.strip() - analysis_length = len(analysis) - - print(f"āœ… Image analysis completed ({analysis_length} characters)") - - # Prepare successful response - result = { - "success": True, - "analysis": analysis or "There was a problem with the request and the image could not be analyzed." - } - - debug_call_data["success"] = True - debug_call_data["analysis_length"] = analysis_length - - # Log debug information - _log_debug_call("vision_analyze_tool", debug_call_data) - _save_debug_log() - - return json.dumps(result, indent=2) - - except Exception as e: - error_msg = f"Error analyzing image: {str(e)}" - print(f"āŒ {error_msg}") - - # Prepare error response - result = { - "success": False, - "analysis": "There was a problem with the request and the image could not be analyzed." - } - - debug_call_data["error"] = error_msg - _log_debug_call("vision_analyze_tool", debug_call_data) - _save_debug_log() - - return json.dumps(result, indent=2) - - -def check_nous_api_key() -> bool: - """ - Check if the Nous Research API key is available in environment variables. - - Returns: - bool: True if API key is set, False otherwise - """ - return bool(os.getenv("NOUS_API_KEY")) - - -def check_vision_requirements() -> bool: - """ - Check if all requirements for vision tools are met. - - Returns: - bool: True if requirements are met, False otherwise - """ - return check_nous_api_key() - - -def get_debug_session_info() -> Dict[str, Any]: - """ - Get information about the current debug session. - - Returns: - Dict[str, Any]: Dictionary containing debug session information - """ - if not DEBUG_MODE or not DEBUG_DATA: - return { - "enabled": False, - "session_id": None, - "log_path": None, - "total_calls": 0 - } - - return { - "enabled": True, - "session_id": DEBUG_SESSION_ID, - "log_path": str(DEBUG_LOG_PATH / f"vision_tools_debug_{DEBUG_SESSION_ID}.json"), - "total_calls": len(DEBUG_DATA["tool_calls"]) - } - - -if __name__ == "__main__": - """ - Simple test/demo when run directly - """ - print("šŸ‘ļø Vision Tools Module") - print("=" * 40) - - # Check if API key is available - api_available = check_nous_api_key() - - if not api_available: - print("āŒ NOUS_API_KEY environment variable not set") - print("Please set your API key: export NOUS_API_KEY='your-key-here'") - print("Get API key at: https://inference-api.nousresearch.com/") - exit(1) - else: - print("āœ… Nous Research API key found") - - print("šŸ› ļø Vision tools ready for use!") - print(f"šŸ§  Using model: {DEFAULT_VISION_MODEL}") - - # Show debug mode status - if DEBUG_MODE: - print(f"šŸ› Debug mode ENABLED - Session ID: {DEBUG_SESSION_ID}") - print(f" Debug logs will be saved to: ./logs/vision_tools_debug_{DEBUG_SESSION_ID}.json") - else: - print("šŸ› Debug mode disabled (set VISION_TOOLS_DEBUG=true to enable)") - - print("\nBasic usage:") - print(" from vision_tools import vision_analyze_tool") - print(" import asyncio") - print("") - print(" async def main():") - print(" result = await vision_analyze_tool(") - print(" image_url='https://example.com/image.jpg',") - print(" user_prompt='What do you see in this image?'") - print(" )") - print(" print(result)") - print(" asyncio.run(main())") - - print("\nExample prompts:") - print(" - 'What architectural style is this building?'") - print(" - 'Describe the emotions and mood in this image'") - print(" - 'What text can you read in this image?'") - print(" - 'Identify any safety hazards visible'") - print(" - 'What products or brands are shown?'") - - print("\nDebug mode:") - print(" # Enable debug logging") - print(" export VISION_TOOLS_DEBUG=true") - print(" # Debug logs capture all vision analysis calls and results") - print(" # Logs saved to: ./logs/vision_tools_debug_UUID.json") +#!/usr/bin/env python3 +""" +Vision Tools Module + +This module provides vision analysis tools that work with image URLs. +Uses Gemini Flash via Nous Research API for intelligent image understanding. + +Available tools: +- vision_analyze_tool: Analyze images from URLs with custom prompts + +Features: +- Comprehensive image description +- Context-aware analysis based on user queries +- Proper error handling and validation +- Debug logging support + +Usage: + from vision_tools import vision_analyze_tool + import asyncio + + # Analyze an image + result = await vision_analyze_tool( + image_url="https://example.com/image.jpg", + user_prompt="What architectural style is this building?" + ) +""" + +import json +import os +import asyncio +import uuid +import datetime +from pathlib import Path +from typing import Dict, Any, Optional +from openai import AsyncOpenAI + +# Initialize Nous Research API client for vision processing +nous_client = AsyncOpenAI( + api_key=os.getenv("NOUS_API_KEY"), + base_url="https://inference-api.nousresearch.com/v1" +) + +# Configuration for vision processing +DEFAULT_VISION_MODEL = "gemini-2.5-flash" + +# Debug mode configuration +DEBUG_MODE = os.getenv("VISION_TOOLS_DEBUG", "false").lower() == "true" +DEBUG_SESSION_ID = str(uuid.uuid4()) +DEBUG_LOG_PATH = Path("./logs") +DEBUG_DATA = { + "session_id": DEBUG_SESSION_ID, + "start_time": datetime.datetime.now().isoformat(), + "debug_enabled": DEBUG_MODE, + "tool_calls": [] +} if DEBUG_MODE else None + +# Create logs directory if debug mode is enabled +if DEBUG_MODE: + DEBUG_LOG_PATH.mkdir(exist_ok=True) + print(f"šŸ› Vision debug mode enabled - Session ID: {DEBUG_SESSION_ID}") + + +def _log_debug_call(tool_name: str, call_data: Dict[str, Any]) -> None: + """ + Log a debug call entry to the global debug data structure. + + Args: + tool_name (str): Name of the tool being called + call_data (Dict[str, Any]): Data about the call including parameters and results + """ + if not DEBUG_MODE or not DEBUG_DATA: + return + + call_entry = { + "timestamp": datetime.datetime.now().isoformat(), + "tool_name": tool_name, + **call_data + } + + DEBUG_DATA["tool_calls"].append(call_entry) + + +def _save_debug_log() -> None: + """ + Save the current debug data to a JSON file in the logs directory. + """ + if not DEBUG_MODE or not DEBUG_DATA: + return + + try: + debug_filename = f"vision_tools_debug_{DEBUG_SESSION_ID}.json" + debug_filepath = DEBUG_LOG_PATH / debug_filename + + # Update end time + DEBUG_DATA["end_time"] = datetime.datetime.now().isoformat() + DEBUG_DATA["total_calls"] = len(DEBUG_DATA["tool_calls"]) + + with open(debug_filepath, 'w', encoding='utf-8') as f: + json.dump(DEBUG_DATA, f, indent=2, ensure_ascii=False) + + print(f"šŸ› Vision debug log saved: {debug_filepath}") + + except Exception as e: + print(f"āŒ Error saving vision debug log: {str(e)}") + + +def _validate_image_url(url: str) -> bool: + """ + Basic validation of image URL format. + + Args: + url (str): The URL to validate + + Returns: + bool: True if URL appears to be valid, False otherwise + """ + if not url or not isinstance(url, str): + return False + + # Check if it's a valid URL format + if not (url.startswith('http://') or url.startswith('https://')): + return False + + # Check for common image extensions (optional, as URLs may not have extensions) + image_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp', '.svg'] + + return True # Allow all HTTP/HTTPS URLs for flexibility + + +async def vision_analyze_tool( + image_url: str, + user_prompt: str, + model: str = DEFAULT_VISION_MODEL +) -> str: + """ + Analyze an image from a URL using vision AI. + + This tool processes images using Gemini Flash via Nous Research API. + The user_prompt parameter is expected to be pre-formatted by the calling + function (typically model_tools.py) to include both full description + requests and specific questions. + + Args: + image_url (str): The URL of the image to analyze + user_prompt (str): The pre-formatted prompt for the vision model + model (str): The vision model to use (default: gemini-2.5-flash) + + Returns: + str: JSON string containing the analysis results with the following structure: + { + "success": bool, + "analysis": str (defaults to error message if None) + } + + Raises: + Exception: If analysis fails or API key is not set + """ + debug_call_data = { + "parameters": { + "image_url": image_url, + "user_prompt": user_prompt, + "model": model + }, + "error": None, + "success": False, + "analysis_length": 0, + "model_used": model + } + + try: + print(f"šŸ” Analyzing image from URL: {image_url[:60]}{'...' if len(image_url) > 60 else ''}") + print(f"šŸ“ User prompt: {user_prompt[:100]}{'...' if len(user_prompt) > 100 else ''}") + + # Validate image URL + if not _validate_image_url(image_url): + raise ValueError("Invalid image URL format. Must start with http:// or https://") + + # Check API key availability + if not os.getenv("NOUS_API_KEY"): + raise ValueError("NOUS_API_KEY environment variable not set") + + # Use the prompt as provided (model_tools.py now handles full description formatting) + comprehensive_prompt = user_prompt + + # Prepare the message with image URL format + messages = [ + { + "role": "user", + "content": [ + { + "type": "text", + "text": comprehensive_prompt + }, + { + "type": "image_url", + "image_url": { + "url": image_url + } + } + ] + } + ] + + print(f"šŸ§  Processing image with {model}...") + + # Call the vision API + response = await nous_client.chat.completions.create( + model=model, + messages=messages, + temperature=0.1, # Low temperature for consistent analysis + max_tokens=2000 # Generous limit for detailed analysis + ) + + # Extract the analysis + analysis = response.choices[0].message.content.strip() + analysis_length = len(analysis) + + print(f"āœ… Image analysis completed ({analysis_length} characters)") + + # Prepare successful response + result = { + "success": True, + "analysis": analysis or "There was a problem with the request and the image could not be analyzed." + } + + debug_call_data["success"] = True + debug_call_data["analysis_length"] = analysis_length + + # Log debug information + _log_debug_call("vision_analyze_tool", debug_call_data) + _save_debug_log() + + return json.dumps(result, indent=2) + + except Exception as e: + error_msg = f"Error analyzing image: {str(e)}" + print(f"āŒ {error_msg}") + + # Prepare error response + result = { + "success": False, + "analysis": "There was a problem with the request and the image could not be analyzed." + } + + debug_call_data["error"] = error_msg + _log_debug_call("vision_analyze_tool", debug_call_data) + _save_debug_log() + + return json.dumps(result, indent=2) + + +def check_nous_api_key() -> bool: + """ + Check if the Nous Research API key is available in environment variables. + + Returns: + bool: True if API key is set, False otherwise + """ + return bool(os.getenv("NOUS_API_KEY")) + + +def check_vision_requirements() -> bool: + """ + Check if all requirements for vision tools are met. + + Returns: + bool: True if requirements are met, False otherwise + """ + return check_nous_api_key() + + +def get_debug_session_info() -> Dict[str, Any]: + """ + Get information about the current debug session. + + Returns: + Dict[str, Any]: Dictionary containing debug session information + """ + if not DEBUG_MODE or not DEBUG_DATA: + return { + "enabled": False, + "session_id": None, + "log_path": None, + "total_calls": 0 + } + + return { + "enabled": True, + "session_id": DEBUG_SESSION_ID, + "log_path": str(DEBUG_LOG_PATH / f"vision_tools_debug_{DEBUG_SESSION_ID}.json"), + "total_calls": len(DEBUG_DATA["tool_calls"]) + } + + +if __name__ == "__main__": + """ + Simple test/demo when run directly + """ + print("šŸ‘ļø Vision Tools Module") + print("=" * 40) + + # Check if API key is available + api_available = check_nous_api_key() + + if not api_available: + print("āŒ NOUS_API_KEY environment variable not set") + print("Please set your API key: export NOUS_API_KEY='your-key-here'") + print("Get API key at: https://inference-api.nousresearch.com/") + exit(1) + else: + print("āœ… Nous Research API key found") + + print("šŸ› ļø Vision tools ready for use!") + print(f"šŸ§  Using model: {DEFAULT_VISION_MODEL}") + + # Show debug mode status + if DEBUG_MODE: + print(f"šŸ› Debug mode ENABLED - Session ID: {DEBUG_SESSION_ID}") + print(f" Debug logs will be saved to: ./logs/vision_tools_debug_{DEBUG_SESSION_ID}.json") + else: + print("šŸ› Debug mode disabled (set VISION_TOOLS_DEBUG=true to enable)") + + print("\nBasic usage:") + print(" from vision_tools import vision_analyze_tool") + print(" import asyncio") + print("") + print(" async def main():") + print(" result = await vision_analyze_tool(") + print(" image_url='https://example.com/image.jpg',") + print(" user_prompt='What do you see in this image?'") + print(" )") + print(" print(result)") + print(" asyncio.run(main())") + + print("\nExample prompts:") + print(" - 'What architectural style is this building?'") + print(" - 'Describe the emotions and mood in this image'") + print(" - 'What text can you read in this image?'") + print(" - 'Identify any safety hazards visible'") + print(" - 'What products or brands are shown?'") + + print("\nDebug mode:") + print(" # Enable debug logging") + print(" export VISION_TOOLS_DEBUG=true") + print(" # Debug logs capture all vision analysis calls and results") + print(" # Logs saved to: ./logs/vision_tools_debug_UUID.json") diff --git a/tools/web_tools.py b/tools/web_tools.py index 706eb1ff1..f54db58d2 100644 --- a/tools/web_tools.py +++ b/tools/web_tools.py @@ -1,1009 +1,1017 @@ -#!/usr/bin/env python3 -""" -Standalone Web Tools Module - -This module provides generic web tools that work with multiple backend providers. -Currently uses Firecrawl as the backend, and the interface makes it easy to swap -providers without changing the function signatures. - -Available tools: -- web_search_tool: Search the web for information -- web_extract_tool: Extract content from specific web pages -- web_crawl_tool: Crawl websites with specific instructions - -Backend compatibility: -- Firecrawl: https://docs.firecrawl.dev/introduction - -LLM Processing: -- Uses Nous Research API with Gemini 2.5 Flash for intelligent content extraction -- Extracts key excerpts and creates markdown summaries to reduce token usage - -Debug Mode: -- Set WEB_TOOLS_DEBUG=true to enable detailed logging -- Creates web_tools_debug_UUID.json in ./logs directory -- Captures all tool calls, results, and compression metrics - -Usage: - from web_tools import web_search_tool, web_extract_tool, web_crawl_tool - - # Search the web - results = web_search_tool("Python machine learning libraries", limit=3) - - # Extract content from URLs - content = web_extract_tool(["https://example.com"], format="markdown") - - # Crawl a website - crawl_data = web_crawl_tool("example.com", "Find contact information") -""" - -#TODO: Search Capabilities over the scraped pages -#TODO: Store the pages in something -#TODO: Tool to see what pages are available/saved to search over - -import json -import os -import re -import asyncio -import uuid -import datetime -from pathlib import Path -from typing import List, Dict, Any, Optional -from firecrawl import Firecrawl -from openai import AsyncOpenAI - -# Initialize Firecrawl client once at module level -firecrawl_client = Firecrawl(api_key=os.getenv("FIRECRAWL_API_KEY")) - -# Initialize Nous Research API client for LLM processing (async) -nous_client = AsyncOpenAI( - api_key=os.getenv("NOUS_API_KEY"), - base_url="https://inference-api.nousresearch.com/v1" -) - -# Configuration for LLM processing -DEFAULT_SUMMARIZER_MODEL = "gemini-2.5-flash" -DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION = 5000 - -# Debug mode configuration -DEBUG_MODE = os.getenv("WEB_TOOLS_DEBUG", "false").lower() == "true" -DEBUG_SESSION_ID = str(uuid.uuid4()) -DEBUG_LOG_PATH = Path("./logs") -DEBUG_DATA = { - "session_id": DEBUG_SESSION_ID, - "start_time": datetime.datetime.now().isoformat(), - "debug_enabled": DEBUG_MODE, - "tool_calls": [] -} if DEBUG_MODE else None - -# Create logs directory if debug mode is enabled -if DEBUG_MODE: - DEBUG_LOG_PATH.mkdir(exist_ok=True) - print(f"šŸ› Debug mode enabled - Session ID: {DEBUG_SESSION_ID}") - - -def _log_debug_call(tool_name: str, call_data: Dict[str, Any]) -> None: - """ - Log a debug call entry to the global debug data structure. - - Args: - tool_name (str): Name of the tool being called - call_data (Dict[str, Any]): Data about the call including parameters and results - """ - if not DEBUG_MODE or not DEBUG_DATA: - return - - call_entry = { - "timestamp": datetime.datetime.now().isoformat(), - "tool_name": tool_name, - **call_data - } - - DEBUG_DATA["tool_calls"].append(call_entry) - - -def _save_debug_log() -> None: - """ - Save the current debug data to a JSON file in the logs directory. - """ - if not DEBUG_MODE or not DEBUG_DATA: - return - - try: - debug_filename = f"web_tools_debug_{DEBUG_SESSION_ID}.json" - debug_filepath = DEBUG_LOG_PATH / debug_filename - - # Update end time - DEBUG_DATA["end_time"] = datetime.datetime.now().isoformat() - DEBUG_DATA["total_calls"] = len(DEBUG_DATA["tool_calls"]) - - with open(debug_filepath, 'w', encoding='utf-8') as f: - json.dump(DEBUG_DATA, f, indent=2, ensure_ascii=False) - - print(f"šŸ› Debug log saved: {debug_filepath}") - - except Exception as e: - print(f"āŒ Error saving debug log: {str(e)}") - - -async def process_content_with_llm( - content: str, - url: str = "", - title: str = "", - model: str = DEFAULT_SUMMARIZER_MODEL, - min_length: int = DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION -) -> Optional[str]: - """ - Process web content using LLM to create intelligent summaries with key excerpts. - - This function uses Gemini 2.5 Flash (or specified model) via Nous Research API - to intelligently extract key information and create markdown summaries, - significantly reducing token usage while preserving all important information. - - Args: - content (str): The raw content to process - url (str): The source URL (for context, optional) - title (str): The page title (for context, optional) - model (str): The model to use for processing (default: gemini-2.5-flash) - min_length (int): Minimum content length to trigger processing (default: 5000) - - Returns: - Optional[str]: Processed markdown content, or None if content too short or processing fails - """ - try: - # Skip processing if content is too short - if len(content) < min_length: - print(f"šŸ“ Content too short ({len(content)} < {min_length} chars), skipping LLM processing") - return None - - print(f"šŸ§  Processing content with LLM ({len(content)} characters)") - - # Create context information - context_info = [] - if title: - context_info.append(f"Title: {title}") - if url: - context_info.append(f"Source: {url}") - - context_str = "\n".join(context_info) + "\n\n" if context_info else "" - - # Simplified prompt for better quality markdown output - system_prompt = """You are an expert content analyst. Your job is to process web content and create a comprehensive yet concise summary that preserves all important information while dramatically reducing bulk. - -Create a well-structured markdown summary that includes: -1. Key excerpts (quotes, code snippets, important facts) in their original format -2. Comprehensive summary of all other important information -3. Proper markdown formatting with headers, bullets, and emphasis - -Your goal is to preserve ALL important information while reducing length. Never lose key facts, figures, insights, or actionable information. Make it scannable and well-organized.""" - - user_prompt = f"""Please process this web content and create a comprehensive markdown summary: - -{context_str}CONTENT TO PROCESS: -{content} - -Create a markdown summary that captures all key information in a well-organized, scannable format. Include important quotes and code snippets in their original formatting. Focus on actionable information, specific details, and unique insights.""" - - # Call the LLM asynchronously - response = await nous_client.chat.completions.create( - model=model, - messages=[ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_prompt} - ], - temperature=0.1, # Low temperature for consistent extraction - max_tokens=4000 # Generous limit for comprehensive processing - ) - - # Get the markdown response directly - processed_content = response.choices[0].message.content.strip() - - # Calculate compression metrics for logging - original_length = len(content) - processed_length = len(processed_content) - compression_ratio = processed_length / original_length if original_length > 0 else 1.0 - - print(f"āœ… Content processed: {original_length} ā†’ {processed_length} chars ({compression_ratio:.1%})") - - return processed_content - - except Exception as e: - print(f"āŒ Error processing content with LLM: {str(e)}") - return None - - -def clean_base64_images(text: str) -> str: - """ - Remove base64 encoded images from text to reduce token count and clutter. - - This function finds and removes base64 encoded images in various formats: - - (data:image/png;base64,...) - - (data:image/jpeg;base64,...) - - (data:image/svg+xml;base64,...) - - data:image/[type];base64,... (without parentheses) - - Args: - text: The text content to clean - - Returns: - Cleaned text with base64 images replaced with placeholders - """ - # Pattern to match base64 encoded images wrapped in parentheses - # Matches: (data:image/[type];base64,[base64-string]) - base64_with_parens_pattern = r'\(data:image/[^;]+;base64,[A-Za-z0-9+/=]+\)' - - # Pattern to match base64 encoded images without parentheses - # Matches: data:image/[type];base64,[base64-string] - base64_pattern = r'data:image/[^;]+;base64,[A-Za-z0-9+/=]+' - - # Replace parentheses-wrapped images first - cleaned_text = re.sub(base64_with_parens_pattern, '[BASE64_IMAGE_REMOVED]', text) - - # Then replace any remaining non-parentheses images - cleaned_text = re.sub(base64_pattern, '[BASE64_IMAGE_REMOVED]', cleaned_text) - - return cleaned_text - - -def web_search_tool(query: str, limit: int = 5) -> str: - """ - Search the web for information using available search API backend. - - This function provides a generic interface for web search that can work - with multiple backends. Currently uses Firecrawl. - - Note: This function returns search result metadata only (URLs, titles, descriptions). - Use web_extract_tool to get full content from specific URLs. - - Args: - query (str): The search query to look up - limit (int): Maximum number of results to return (default: 5) - - Returns: - str: JSON string containing search results with the following structure: - { - "success": bool, - "data": { - "web": [ - { - "title": str, - "url": str, - "description": str, - "position": int - }, - ... - ] - } - } - - Raises: - Exception: If search fails or API key is not set - """ - debug_call_data = { - "parameters": { - "query": query, - "limit": limit - }, - "error": None, - "results_count": 0, - "original_response_size": 0, - "final_response_size": 0 - } - - try: - print(f"šŸ” Searching the web for: '{query}' (limit: {limit})") - - # Use Firecrawl's v2 search functionality WITHOUT scraping - # We only want search result metadata, not scraped content - # Docs: https://docs.firecrawl.dev/features/search - response = firecrawl_client.search( - query=query, - limit=limit - ) - - # The response is a SearchData object with web, news, and images attributes - # When not scraping, the results are directly in these attributes - web_results = [] - - # Check if response has web attribute (SearchData object) - if hasattr(response, 'web'): - # Response is a SearchData object with web attribute - if response.web: - # Convert each SearchResultWeb object to dict - for result in response.web: - if hasattr(result, 'model_dump'): - # Pydantic model - use model_dump - web_results.append(result.model_dump()) - elif hasattr(result, '__dict__'): - # Regular object - use __dict__ - web_results.append(result.__dict__) - elif isinstance(result, dict): - # Already a dict - web_results.append(result) - elif hasattr(response, 'model_dump'): - # Response has model_dump method - use it to get dict - response_dict = response.model_dump() - if 'web' in response_dict and response_dict['web']: - web_results = response_dict['web'] - elif isinstance(response, dict): - # Response is already a dictionary - if 'web' in response and response['web']: - web_results = response['web'] - - results_count = len(web_results) - print(f"āœ… Found {results_count} search results") - - # Build response with just search metadata (URLs, titles, descriptions) - response_data = { - "success": True, - "data": { - "web": web_results - } - } - - # Capture debug information - debug_call_data["results_count"] = results_count - - # Convert to JSON - result_json = json.dumps(response_data, indent=2) - - debug_call_data["final_response_size"] = len(result_json) - - # Log debug information - _log_debug_call("web_search_tool", debug_call_data) - _save_debug_log() - - return result_json - - except Exception as e: - error_msg = f"Error searching web: {str(e)}" - print(f"āŒ {error_msg}") - - debug_call_data["error"] = error_msg - _log_debug_call("web_search_tool", debug_call_data) - _save_debug_log() - - return json.dumps({"error": error_msg}) - - -async def web_extract_tool( - urls: List[str], - format: str = None, - use_llm_processing: bool = True, - model: str = DEFAULT_SUMMARIZER_MODEL, - min_length: int = DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION -) -> str: - """ - Extract content from specific web pages using available extraction API backend. - - This function provides a generic interface for web content extraction that - can work with multiple backends. Currently uses Firecrawl. - - Args: - urls (List[str]): List of URLs to extract content from - format (str): Desired output format ("markdown" or "html", optional) - use_llm_processing (bool): Whether to process content with LLM for summarization (default: True) - model (str): The model to use for LLM processing (default: gemini-2.5-flash) - min_length (int): Minimum content length to trigger LLM processing (default: 5000) - - Returns: - str: JSON string containing extracted content. If LLM processing is enabled and successful, - the 'content' field will contain the processed markdown summary instead of raw content. - - Raises: - Exception: If extraction fails or API key is not set - """ - debug_call_data = { - "parameters": { - "urls": urls, - "format": format, - "use_llm_processing": use_llm_processing, - "model": model, - "min_length": min_length - }, - "error": None, - "pages_extracted": 0, - "pages_processed_with_llm": 0, - "original_response_size": 0, - "final_response_size": 0, - "compression_metrics": [], - "processing_applied": [] - } - - try: - print(f"šŸ“„ Extracting content from {len(urls)} URL(s)") - - # Determine requested formats for Firecrawl v2 - formats: List[str] = [] - if format == "markdown": - formats = ["markdown"] - elif format == "html": - formats = ["html"] - else: - # Default: request markdown for LLM-readiness and include html as backup - formats = ["markdown", "html"] - - # Always use individual scraping for simplicity and reliability - # Batch scraping adds complexity without much benefit for small numbers of URLs - results: List[Dict[str, Any]] = [] - - for url in urls: - try: - print(f" šŸ“„ Scraping: {url}") - scrape_result = firecrawl_client.scrape( - url=url, - formats=formats - ) - - # Process the result - properly handle object serialization - metadata = {} - title = "" - content_markdown = None - content_html = None - - # Extract data from the scrape result - if hasattr(scrape_result, 'model_dump'): - # Pydantic model - use model_dump to get dict - result_dict = scrape_result.model_dump() - content_markdown = result_dict.get('markdown') - content_html = result_dict.get('html') - metadata = result_dict.get('metadata', {}) - elif hasattr(scrape_result, '__dict__'): - # Regular object with attributes - content_markdown = getattr(scrape_result, 'markdown', None) - content_html = getattr(scrape_result, 'html', None) - - # Handle metadata - convert to dict if it's an object - metadata_obj = getattr(scrape_result, 'metadata', {}) - if hasattr(metadata_obj, 'model_dump'): - metadata = metadata_obj.model_dump() - elif hasattr(metadata_obj, '__dict__'): - metadata = metadata_obj.__dict__ - elif isinstance(metadata_obj, dict): - metadata = metadata_obj - else: - metadata = {} - elif isinstance(scrape_result, dict): - # Already a dictionary - content_markdown = scrape_result.get('markdown') - content_html = scrape_result.get('html') - metadata = scrape_result.get('metadata', {}) - - # Ensure metadata is a dict (not an object) - if not isinstance(metadata, dict): - if hasattr(metadata, 'model_dump'): - metadata = metadata.model_dump() - elif hasattr(metadata, '__dict__'): - metadata = metadata.__dict__ - else: - metadata = {} - - # Get title from metadata - title = metadata.get("title", "") - - # Choose content based on requested format - chosen_content = content_markdown if (format == "markdown" or (format is None and content_markdown)) else content_html or content_markdown or "" - - results.append({ - "url": metadata.get("sourceURL", url), - "title": title, - "content": chosen_content, - "raw_content": chosen_content, - "metadata": metadata # Now guaranteed to be a dict - }) - - except Exception as scrape_err: - print(f" āŒ Error scraping {url}: {str(scrape_err)}") - results.append({ - "url": url, - "title": "", - "content": "", - "raw_content": "", - "error": str(scrape_err) - }) - - response = {"results": results} - - pages_extracted = len(response.get('results', [])) - print(f"āœ… Extracted content from {pages_extracted} pages") - - debug_call_data["pages_extracted"] = pages_extracted - debug_call_data["original_response_size"] = len(json.dumps(response)) - - # Process each result with LLM if enabled - if use_llm_processing and os.getenv("NOUS_API_KEY"): - print("šŸ§  Processing extracted content with LLM...") - debug_call_data["processing_applied"].append("llm_processing") - - for result in response.get('results', []): - url = result.get('url', 'Unknown URL') - title = result.get('title', '') - raw_content = result.get('raw_content', '') or result.get('content', '') - - if raw_content: - original_size = len(raw_content) - - # Process content with LLM - processed = await process_content_with_llm( - raw_content, url, title, model, min_length - ) - - if processed: - processed_size = len(processed) - compression_ratio = processed_size / original_size if original_size > 0 else 1.0 - - # Capture compression metrics - debug_call_data["compression_metrics"].append({ - "url": url, - "original_size": original_size, - "processed_size": processed_size, - "compression_ratio": compression_ratio, - "model_used": model - }) - - # Replace content with processed version - result['content'] = processed - # Keep raw content in separate field for reference - result['raw_content'] = raw_content - debug_call_data["pages_processed_with_llm"] += 1 - print(f" šŸ“ {url} (processed)") - else: - debug_call_data["compression_metrics"].append({ - "url": url, - "original_size": original_size, - "processed_size": original_size, - "compression_ratio": 1.0, - "model_used": None, - "reason": "content_too_short" - }) - print(f" šŸ“ {url} (no processing - content too short)") - else: - print(f" āš ļø {url} (no content to process)") - else: - if use_llm_processing and not os.getenv("NOUS_API_KEY"): - print("āš ļø LLM processing requested but NOUS_API_KEY not set, returning raw content") - debug_call_data["processing_applied"].append("llm_processing_unavailable") - - # Print summary of extracted pages for debugging (original behavior) - for result in response.get('results', []): - url = result.get('url', 'Unknown URL') - content_length = len(result.get('raw_content', '')) - print(f" šŸ“ {url} ({content_length} characters)") - - # Trim output to minimal fields per entry: title, content, error - trimmed_results = [ - { - "title": r.get("title", ""), - "content": r.get("content", ""), - "error": r.get("error") - } - for r in response.get("results", []) - ] - trimmed_response = {"results": trimmed_results} - - result_json = json.dumps(trimmed_response, indent=2) - # Clean base64 images from extracted content - cleaned_result = clean_base64_images(result_json) - - debug_call_data["final_response_size"] = len(cleaned_result) - debug_call_data["processing_applied"].append("base64_image_removal") - - # Log debug information - _log_debug_call("web_extract_tool", debug_call_data) - _save_debug_log() - - return cleaned_result - - except Exception as e: - error_msg = f"Error extracting content: {str(e)}" - print(f"āŒ {error_msg}") - - debug_call_data["error"] = error_msg - _log_debug_call("web_extract_tool", debug_call_data) - _save_debug_log() - - return json.dumps({"error": error_msg}) - - -async def web_crawl_tool( - url: str, - instructions: str = None, - depth: str = "basic", - use_llm_processing: bool = True, - model: str = DEFAULT_SUMMARIZER_MODEL, - min_length: int = DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION -) -> str: - """ - Crawl a website with specific instructions using available crawling API backend. - - This function provides a generic interface for web crawling that can work - with multiple backends. Currently uses Firecrawl. - - Args: - url (str): The base URL to crawl (can include or exclude https://) - instructions (str): Instructions for what to crawl/extract using LLM intelligence (optional) - depth (str): Depth of extraction ("basic" or "advanced", default: "basic") - use_llm_processing (bool): Whether to process content with LLM for summarization (default: True) - model (str): The model to use for LLM processing (default: gemini-2.5-flash) - min_length (int): Minimum content length to trigger LLM processing (default: 5000) - - Returns: - str: JSON string containing crawled content. If LLM processing is enabled and successful, - the 'content' field will contain the processed markdown summary instead of raw content. - Each page is processed individually. - - Raises: - Exception: If crawling fails or API key is not set - """ - debug_call_data = { - "parameters": { - "url": url, - "instructions": instructions, - "depth": depth, - "use_llm_processing": use_llm_processing, - "model": model, - "min_length": min_length - }, - "error": None, - "pages_crawled": 0, - "pages_processed_with_llm": 0, - "original_response_size": 0, - "final_response_size": 0, - "compression_metrics": [], - "processing_applied": [] - } - - try: - # Ensure URL has protocol - if not url.startswith(('http://', 'https://')): - url = f'https://{url}' - print(f" šŸ“ Added https:// prefix to URL: {url}") - - instructions_text = f" with instructions: '{instructions}'" if instructions else "" - print(f"šŸ•·ļø Crawling {url}{instructions_text}") - - # Use Firecrawl's v2 crawl functionality - # Docs: https://docs.firecrawl.dev/features/crawl - # The crawl() method automatically waits for completion and returns all data - - # Build crawl parameters - keep it simple - crawl_params = { - "limit": 20, # Limit number of pages to crawl - "scrape_options": { - "formats": ["markdown"] # Just markdown for simplicity - } - } - - # Note: The 'prompt' parameter is not documented for crawl - # Instructions are typically used with the Extract endpoint, not Crawl - if instructions: - print(f" ā„¹ļø Note: Instructions parameter ignored (not supported in crawl API)") - - # Use the crawl method which waits for completion automatically - try: - crawl_result = firecrawl_client.crawl( - url=url, - **crawl_params - ) - except Exception as e: - print(f" āŒ Crawl API call failed: {e}") - raise - - pages: List[Dict[str, Any]] = [] - - # Process crawl results - the crawl method returns a CrawlJob object with data attribute - data_list = [] - - # The crawl_result is a CrawlJob object with a 'data' attribute containing list of Document objects - if hasattr(crawl_result, 'data'): - data_list = crawl_result.data if crawl_result.data else [] - print(f" šŸ“Š Status: {getattr(crawl_result, 'status', 'unknown')}") - print(f" šŸ“„ Retrieved {len(data_list)} pages") - - # Debug: Check other attributes if no data - if not data_list: - print(f" šŸ” Debug - CrawlJob attributes: {[attr for attr in dir(crawl_result) if not attr.startswith('_')]}") - print(f" šŸ” Debug - Status: {getattr(crawl_result, 'status', 'N/A')}") - print(f" šŸ” Debug - Total: {getattr(crawl_result, 'total', 'N/A')}") - print(f" šŸ” Debug - Completed: {getattr(crawl_result, 'completed', 'N/A')}") - - elif isinstance(crawl_result, dict) and 'data' in crawl_result: - data_list = crawl_result.get("data", []) - else: - print(" āš ļø Unexpected crawl result type") - print(f" šŸ” Debug - Result type: {type(crawl_result)}") - if hasattr(crawl_result, '__dict__'): - print(f" šŸ” Debug - Result attributes: {list(crawl_result.__dict__.keys())}") - - for item in data_list: - # Process each crawled page - properly handle object serialization - page_url = "Unknown URL" - title = "" - content_markdown = None - content_html = None - metadata = {} - - # Extract data from the item - if hasattr(item, 'model_dump'): - # Pydantic model - use model_dump to get dict - item_dict = item.model_dump() - content_markdown = item_dict.get('markdown') - content_html = item_dict.get('html') - metadata = item_dict.get('metadata', {}) - elif hasattr(item, '__dict__'): - # Regular object with attributes - content_markdown = getattr(item, 'markdown', None) - content_html = getattr(item, 'html', None) - - # Handle metadata - convert to dict if it's an object - metadata_obj = getattr(item, 'metadata', {}) - if hasattr(metadata_obj, 'model_dump'): - metadata = metadata_obj.model_dump() - elif hasattr(metadata_obj, '__dict__'): - metadata = metadata_obj.__dict__ - elif isinstance(metadata_obj, dict): - metadata = metadata_obj - else: - metadata = {} - elif isinstance(item, dict): - # Already a dictionary - content_markdown = item.get('markdown') - content_html = item.get('html') - metadata = item.get('metadata', {}) - - # Ensure metadata is a dict (not an object) - if not isinstance(metadata, dict): - if hasattr(metadata, 'model_dump'): - metadata = metadata.model_dump() - elif hasattr(metadata, '__dict__'): - metadata = metadata.__dict__ - else: - metadata = {} - - # Extract URL and title from metadata - page_url = metadata.get("sourceURL", metadata.get("url", "Unknown URL")) - title = metadata.get("title", "") - - # Choose content (prefer markdown) - content = content_markdown or content_html or "" - - pages.append({ - "url": page_url, - "title": title, - "content": content, - "raw_content": content, - "metadata": metadata # Now guaranteed to be a dict - }) - - response = {"results": pages} - - pages_crawled = len(response.get('results', [])) - print(f"āœ… Crawled {pages_crawled} pages") - - debug_call_data["pages_crawled"] = pages_crawled - debug_call_data["original_response_size"] = len(json.dumps(response)) - - # Process each result with LLM if enabled - if use_llm_processing and os.getenv("NOUS_API_KEY"): - print("šŸ§  Processing crawled content with LLM...") - debug_call_data["processing_applied"].append("llm_processing") - - for result in response.get('results', []): - page_url = result.get('url', 'Unknown URL') - title = result.get('title', '') - content = result.get('content', '') - - if content: - original_size = len(content) - - # Process content with LLM - processed = await process_content_with_llm( - content, page_url, title, model, min_length - ) - - if processed: - processed_size = len(processed) - compression_ratio = processed_size / original_size if original_size > 0 else 1.0 - - # Capture compression metrics - debug_call_data["compression_metrics"].append({ - "url": page_url, - "original_size": original_size, - "processed_size": processed_size, - "compression_ratio": compression_ratio, - "model_used": model - }) - - # Keep original content in raw_content field - result['raw_content'] = content - # Replace content with processed version - result['content'] = processed - debug_call_data["pages_processed_with_llm"] += 1 - print(f" šŸŒ {page_url} (processed)") - else: - debug_call_data["compression_metrics"].append({ - "url": page_url, - "original_size": original_size, - "processed_size": original_size, - "compression_ratio": 1.0, - "model_used": None, - "reason": "content_too_short" - }) - print(f" šŸŒ {page_url} (no processing - content too short)") - else: - print(f" āš ļø {page_url} (no content to process)") - else: - if use_llm_processing and not os.getenv("NOUS_API_KEY"): - print("āš ļø LLM processing requested but NOUS_API_KEY not set, returning raw content") - debug_call_data["processing_applied"].append("llm_processing_unavailable") - - # Print summary of crawled pages for debugging (original behavior) - for result in response.get('results', []): - page_url = result.get('url', 'Unknown URL') - content_length = len(result.get('content', '')) - print(f" šŸŒ {page_url} ({content_length} characters)") - - # Trim output to minimal fields per entry: title, content, error - trimmed_results = [ - { - "title": r.get("title", ""), - "content": r.get("content", ""), - "error": r.get("error") - } - for r in response.get("results", []) - ] - trimmed_response = {"results": trimmed_results} - - result_json = json.dumps(trimmed_response, indent=2) - # Clean base64 images from crawled content - cleaned_result = clean_base64_images(result_json) - - debug_call_data["final_response_size"] = len(cleaned_result) - debug_call_data["processing_applied"].append("base64_image_removal") - - # Log debug information - _log_debug_call("web_crawl_tool", debug_call_data) - _save_debug_log() - - return cleaned_result - - except Exception as e: - error_msg = f"Error crawling website: {str(e)}" - print(f"āŒ {error_msg}") - - debug_call_data["error"] = error_msg - _log_debug_call("web_crawl_tool", debug_call_data) - _save_debug_log() - - return json.dumps({"error": error_msg}) - - -# Convenience function to check if API key is available -def check_firecrawl_api_key() -> bool: - """ - Check if the Firecrawl API key is available in environment variables. - - Returns: - bool: True if API key is set, False otherwise - """ - return bool(os.getenv("FIRECRAWL_API_KEY")) - - -def check_nous_api_key() -> bool: - """ - Check if the Nous Research API key is available in environment variables. - - Returns: - bool: True if API key is set, False otherwise - """ - return bool(os.getenv("NOUS_API_KEY")) - - -def get_debug_session_info() -> Dict[str, Any]: - """ - Get information about the current debug session. - - Returns: - Dict[str, Any]: Dictionary containing debug session information: - - enabled: Whether debug mode is enabled - - session_id: Current session UUID (if enabled) - - log_path: Path where debug logs are saved (if enabled) - - total_calls: Number of tool calls logged so far (if enabled) - """ - if not DEBUG_MODE or not DEBUG_DATA: - return { - "enabled": False, - "session_id": None, - "log_path": None, - "total_calls": 0 - } - - return { - "enabled": True, - "session_id": DEBUG_SESSION_ID, - "log_path": str(DEBUG_LOG_PATH / f"web_tools_debug_{DEBUG_SESSION_ID}.json"), - "total_calls": len(DEBUG_DATA["tool_calls"]) - } - - -if __name__ == "__main__": - """ - Simple test/demo when run directly - """ - print("šŸŒ Standalone Web Tools Module") - print("=" * 40) - - # Check if API keys are available - firecrawl_available = check_firecrawl_api_key() - nous_available = check_nous_api_key() - - if not firecrawl_available: - print("āŒ FIRECRAWL_API_KEY environment variable not set") - print("Please set your API key: export FIRECRAWL_API_KEY='your-key-here'") - print("Get API key at: https://firecrawl.dev/") - else: - print("āœ… Firecrawl API key found") - - if not nous_available: - print("āŒ NOUS_API_KEY environment variable not set") - print("Please set your API key: export NOUS_API_KEY='your-key-here'") - print("Get API key at: https://inference-api.nousresearch.com/") - print("āš ļø Without Nous API key, LLM content processing will be disabled") - else: - print("āœ… Nous Research API key found") - - if not firecrawl_available: - exit(1) - - print("šŸ› ļø Web tools ready for use!") - - if nous_available: - print("šŸ§  LLM content processing available with Gemini 2.5 Flash") - print(f" Default min length for processing: {DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION} chars") - - # Show debug mode status - if DEBUG_MODE: - print(f"šŸ› Debug mode ENABLED - Session ID: {DEBUG_SESSION_ID}") - print(f" Debug logs will be saved to: ./logs/web_tools_debug_{DEBUG_SESSION_ID}.json") - else: - print("šŸ› Debug mode disabled (set WEB_TOOLS_DEBUG=true to enable)") - - print("\nBasic usage:") - print(" from web_tools import web_search_tool, web_extract_tool, web_crawl_tool") - print(" import asyncio") - print("") - print(" # Search (synchronous)") - print(" results = web_search_tool('Python tutorials')") - print("") - print(" # Extract and crawl (asynchronous)") - print(" async def main():") - print(" content = await web_extract_tool(['https://example.com'])") - print(" crawl_data = await web_crawl_tool('example.com', 'Find docs')") - print(" asyncio.run(main())") - - if nous_available: - print("\nLLM-enhanced usage:") - print(" # Content automatically processed for pages >5000 chars (default)") - print(" content = await web_extract_tool(['https://python.org/about/'])") - print("") - print(" # Customize processing parameters") - print(" crawl_data = await web_crawl_tool(") - print(" 'docs.python.org',") - print(" 'Find key concepts',") - print(" model='gemini-2.5-flash',") - print(" min_length=3000") - print(" )") - print("") - print(" # Disable LLM processing") - print(" raw_content = await web_extract_tool(['https://example.com'], use_llm_processing=False)") - - print("\nDebug mode:") - print(" # Enable debug logging") - print(" export WEB_TOOLS_DEBUG=true") - print(" # Debug logs capture:") - print(" # - All tool calls with parameters") - print(" # - Original API responses") - print(" # - LLM compression metrics") - print(" # - Final processed results") - print(" # Logs saved to: ./logs/web_tools_debug_UUID.json") - - print(f"\nšŸ“ Run 'python test_web_tools_llm.py' to test LLM processing capabilities") +#!/usr/bin/env python3 +""" +Standalone Web Tools Module + +This module provides generic web tools that work with multiple backend providers. +Currently uses Firecrawl as the backend, and the interface makes it easy to swap +providers without changing the function signatures. + +Available tools: +- web_search_tool: Search the web for information +- web_extract_tool: Extract content from specific web pages +- web_crawl_tool: Crawl websites with specific instructions + +Backend compatibility: +- Firecrawl: https://docs.firecrawl.dev/introduction + +LLM Processing: +- Uses Nous Research API with Gemini 2.5 Flash for intelligent content extraction +- Extracts key excerpts and creates markdown summaries to reduce token usage + +Debug Mode: +- Set WEB_TOOLS_DEBUG=true to enable detailed logging +- Creates web_tools_debug_UUID.json in ./logs directory +- Captures all tool calls, results, and compression metrics + +Usage: + from web_tools import web_search_tool, web_extract_tool, web_crawl_tool + + # Search the web + results = web_search_tool("Python machine learning libraries", limit=3) + + # Extract content from URLs + content = web_extract_tool(["https://example.com"], format="markdown") + + # Crawl a website + crawl_data = web_crawl_tool("example.com", "Find contact information") +""" + +#TODO: Search Capabilities over the scraped pages +#TODO: Store the pages in something +#TODO: Tool to see what pages are available/saved to search over + +import json +import os +import re +import asyncio +import uuid +import datetime +from pathlib import Path +from typing import List, Dict, Any, Optional +from firecrawl import Firecrawl +from openai import AsyncOpenAI + +# Initialize Firecrawl client once at module level +firecrawl_client = Firecrawl(api_key=os.getenv("FIRECRAWL_API_KEY")) + +# Initialize Nous Research API client for LLM processing (async) +nous_client = AsyncOpenAI( + api_key=os.getenv("NOUS_API_KEY"), + base_url="https://inference-api.nousresearch.com/v1" +) + +# Configuration for LLM processing +DEFAULT_SUMMARIZER_MODEL = "gemini-2.5-flash" +DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION = 5000 + +# Debug mode configuration +DEBUG_MODE = os.getenv("WEB_TOOLS_DEBUG", "false").lower() == "true" +DEBUG_SESSION_ID = str(uuid.uuid4()) +DEBUG_LOG_PATH = Path("./logs") +DEBUG_DATA = { + "session_id": DEBUG_SESSION_ID, + "start_time": datetime.datetime.now().isoformat(), + "debug_enabled": DEBUG_MODE, + "tool_calls": [] +} if DEBUG_MODE else None + +# Create logs directory if debug mode is enabled +if DEBUG_MODE: + DEBUG_LOG_PATH.mkdir(exist_ok=True) + print(f"šŸ› Debug mode enabled - Session ID: {DEBUG_SESSION_ID}") + + +def _log_debug_call(tool_name: str, call_data: Dict[str, Any]) -> None: + """ + Log a debug call entry to the global debug data structure. + + Args: + tool_name (str): Name of the tool being called + call_data (Dict[str, Any]): Data about the call including parameters and results + """ + if not DEBUG_MODE or not DEBUG_DATA: + return + + call_entry = { + "timestamp": datetime.datetime.now().isoformat(), + "tool_name": tool_name, + **call_data + } + + DEBUG_DATA["tool_calls"].append(call_entry) + + +def _save_debug_log() -> None: + """ + Save the current debug data to a JSON file in the logs directory. + """ + if not DEBUG_MODE or not DEBUG_DATA: + return + + try: + debug_filename = f"web_tools_debug_{DEBUG_SESSION_ID}.json" + debug_filepath = DEBUG_LOG_PATH / debug_filename + + # Update end time + DEBUG_DATA["end_time"] = datetime.datetime.now().isoformat() + DEBUG_DATA["total_calls"] = len(DEBUG_DATA["tool_calls"]) + + with open(debug_filepath, 'w', encoding='utf-8') as f: + json.dump(DEBUG_DATA, f, indent=2, ensure_ascii=False) + + print(f"šŸ› Debug log saved: {debug_filepath}") + + except Exception as e: + print(f"āŒ Error saving debug log: {str(e)}") + + +async def process_content_with_llm( + content: str, + url: str = "", + title: str = "", + model: str = DEFAULT_SUMMARIZER_MODEL, + min_length: int = DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION +) -> Optional[str]: + """ + Process web content using LLM to create intelligent summaries with key excerpts. + + This function uses Gemini 2.5 Flash (or specified model) via Nous Research API + to intelligently extract key information and create markdown summaries, + significantly reducing token usage while preserving all important information. + + Args: + content (str): The raw content to process + url (str): The source URL (for context, optional) + title (str): The page title (for context, optional) + model (str): The model to use for processing (default: gemini-2.5-flash) + min_length (int): Minimum content length to trigger processing (default: 5000) + + Returns: + Optional[str]: Processed markdown content, or None if content too short or processing fails + """ + try: + # Skip processing if content is too short + if len(content) < min_length: + print(f"šŸ“ Content too short ({len(content)} < {min_length} chars), skipping LLM processing") + return None + + print(f"šŸ§  Processing content with LLM ({len(content)} characters)") + + # Create context information + context_info = [] + if title: + context_info.append(f"Title: {title}") + if url: + context_info.append(f"Source: {url}") + + context_str = "\n".join(context_info) + "\n\n" if context_info else "" + + # Simplified prompt for better quality markdown output + system_prompt = """You are an expert content analyst. Your job is to process web content and create a comprehensive yet concise summary that preserves all important information while dramatically reducing bulk. + +Create a well-structured markdown summary that includes: +1. Key excerpts (quotes, code snippets, important facts) in their original format +2. Comprehensive summary of all other important information +3. Proper markdown formatting with headers, bullets, and emphasis + +Your goal is to preserve ALL important information while reducing length. Never lose key facts, figures, insights, or actionable information. Make it scannable and well-organized.""" + + user_prompt = f"""Please process this web content and create a comprehensive markdown summary: + +{context_str}CONTENT TO PROCESS: +{content} + +Create a markdown summary that captures all key information in a well-organized, scannable format. Include important quotes and code snippets in their original formatting. Focus on actionable information, specific details, and unique insights.""" + + # Call the LLM asynchronously + response = await nous_client.chat.completions.create( + model=model, + messages=[ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt} + ], + temperature=0.1, # Low temperature for consistent extraction + max_tokens=4000 # Generous limit for comprehensive processing + ) + + # Get the markdown response directly + processed_content = response.choices[0].message.content.strip() + + # Calculate compression metrics for logging + original_length = len(content) + processed_length = len(processed_content) + compression_ratio = processed_length / original_length if original_length > 0 else 1.0 + + print(f"āœ… Content processed: {original_length} ā†’ {processed_length} chars ({compression_ratio:.1%})") + + return processed_content + + except Exception as e: + print(f"āŒ Error processing content with LLM: {str(e)}") + return None + + +def clean_base64_images(text: str) -> str: + """ + Remove base64 encoded images from text to reduce token count and clutter. + + This function finds and removes base64 encoded images in various formats: + - (data:image/png;base64,...) + - (data:image/jpeg;base64,...) + - (data:image/svg+xml;base64,...) + - data:image/[type];base64,... (without parentheses) + + Args: + text: The text content to clean + + Returns: + Cleaned text with base64 images replaced with placeholders + """ + # Pattern to match base64 encoded images wrapped in parentheses + # Matches: (data:image/[type];base64,[base64-string]) + base64_with_parens_pattern = r'\(data:image/[^;]+;base64,[A-Za-z0-9+/=]+\)' + + # Pattern to match base64 encoded images without parentheses + # Matches: data:image/[type];base64,[base64-string] + base64_pattern = r'data:image/[^;]+;base64,[A-Za-z0-9+/=]+' + + # Replace parentheses-wrapped images first + cleaned_text = re.sub(base64_with_parens_pattern, '[BASE64_IMAGE_REMOVED]', text) + + # Then replace any remaining non-parentheses images + cleaned_text = re.sub(base64_pattern, '[BASE64_IMAGE_REMOVED]', cleaned_text) + + return cleaned_text + + +def web_search_tool(query: str, limit: int = 5) -> str: + """ + Search the web for information using available search API backend. + + This function provides a generic interface for web search that can work + with multiple backends. Currently uses Firecrawl. + + Note: This function returns search result metadata only (URLs, titles, descriptions). + Use web_extract_tool to get full content from specific URLs. + + Args: + query (str): The search query to look up + limit (int): Maximum number of results to return (default: 5) + + Returns: + str: JSON string containing search results with the following structure: + { + "success": bool, + "data": { + "web": [ + { + "title": str, + "url": str, + "description": str, + "position": int + }, + ... + ] + } + } + + Raises: + Exception: If search fails or API key is not set + """ + debug_call_data = { + "parameters": { + "query": query, + "limit": limit + }, + "error": None, + "results_count": 0, + "original_response_size": 0, + "final_response_size": 0 + } + + try: + print(f"šŸ” Searching the web for: '{query}' (limit: {limit})") + + # Use Firecrawl's v2 search functionality WITHOUT scraping + # We only want search result metadata, not scraped content + # Docs: https://docs.firecrawl.dev/features/search + response = firecrawl_client.search( + query=query, + limit=limit + ) + + # The response is a SearchData object with web, news, and images attributes + # When not scraping, the results are directly in these attributes + web_results = [] + + # Check if response has web attribute (SearchData object) + if hasattr(response, 'web'): + # Response is a SearchData object with web attribute + if response.web: + # Convert each SearchResultWeb object to dict + for result in response.web: + if hasattr(result, 'model_dump'): + # Pydantic model - use model_dump + web_results.append(result.model_dump()) + elif hasattr(result, '__dict__'): + # Regular object - use __dict__ + web_results.append(result.__dict__) + elif isinstance(result, dict): + # Already a dict + web_results.append(result) + elif hasattr(response, 'model_dump'): + # Response has model_dump method - use it to get dict + response_dict = response.model_dump() + if 'web' in response_dict and response_dict['web']: + web_results = response_dict['web'] + elif isinstance(response, dict): + # Response is already a dictionary + if 'web' in response and response['web']: + web_results = response['web'] + + results_count = len(web_results) + print(f"āœ… Found {results_count} search results") + + # Build response with just search metadata (URLs, titles, descriptions) + response_data = { + "success": True, + "data": { + "web": web_results + } + } + + # Capture debug information + debug_call_data["results_count"] = results_count + + # Convert to JSON + result_json = json.dumps(response_data, indent=2) + + debug_call_data["final_response_size"] = len(result_json) + + # Log debug information + _log_debug_call("web_search_tool", debug_call_data) + _save_debug_log() + + return result_json + + except Exception as e: + error_msg = f"Error searching web: {str(e)}" + print(f"āŒ {error_msg}") + + debug_call_data["error"] = error_msg + _log_debug_call("web_search_tool", debug_call_data) + _save_debug_log() + + return json.dumps({"error": error_msg}) + + +async def web_extract_tool( + urls: List[str], + format: str = None, + use_llm_processing: bool = True, + model: str = DEFAULT_SUMMARIZER_MODEL, + min_length: int = DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION +) -> str: + """ + Extract content from specific web pages using available extraction API backend. + + This function provides a generic interface for web content extraction that + can work with multiple backends. Currently uses Firecrawl. + + Args: + urls (List[str]): List of URLs to extract content from + format (str): Desired output format ("markdown" or "html", optional) + use_llm_processing (bool): Whether to process content with LLM for summarization (default: True) + model (str): The model to use for LLM processing (default: gemini-2.5-flash) + min_length (int): Minimum content length to trigger LLM processing (default: 5000) + + Returns: + str: JSON string containing extracted content. If LLM processing is enabled and successful, + the 'content' field will contain the processed markdown summary instead of raw content. + + Raises: + Exception: If extraction fails or API key is not set + """ + debug_call_data = { + "parameters": { + "urls": urls, + "format": format, + "use_llm_processing": use_llm_processing, + "model": model, + "min_length": min_length + }, + "error": None, + "pages_extracted": 0, + "pages_processed_with_llm": 0, + "original_response_size": 0, + "final_response_size": 0, + "compression_metrics": [], + "processing_applied": [] + } + + try: + print(f"šŸ“„ Extracting content from {len(urls)} URL(s)") + + # Determine requested formats for Firecrawl v2 + formats: List[str] = [] + if format == "markdown": + formats = ["markdown"] + elif format == "html": + formats = ["html"] + else: + # Default: request markdown for LLM-readiness and include html as backup + formats = ["markdown", "html"] + + # Always use individual scraping for simplicity and reliability + # Batch scraping adds complexity without much benefit for small numbers of URLs + results: List[Dict[str, Any]] = [] + + for url in urls: + try: + print(f" šŸ“„ Scraping: {url}") + scrape_result = firecrawl_client.scrape( + url=url, + formats=formats + ) + + # Process the result - properly handle object serialization + metadata = {} + title = "" + content_markdown = None + content_html = None + + # Extract data from the scrape result + if hasattr(scrape_result, 'model_dump'): + # Pydantic model - use model_dump to get dict + result_dict = scrape_result.model_dump() + content_markdown = result_dict.get('markdown') + content_html = result_dict.get('html') + metadata = result_dict.get('metadata', {}) + elif hasattr(scrape_result, '__dict__'): + # Regular object with attributes + content_markdown = getattr(scrape_result, 'markdown', None) + content_html = getattr(scrape_result, 'html', None) + + # Handle metadata - convert to dict if it's an object + metadata_obj = getattr(scrape_result, 'metadata', {}) + if hasattr(metadata_obj, 'model_dump'): + metadata = metadata_obj.model_dump() + elif hasattr(metadata_obj, '__dict__'): + metadata = metadata_obj.__dict__ + elif isinstance(metadata_obj, dict): + metadata = metadata_obj + else: + metadata = {} + elif isinstance(scrape_result, dict): + # Already a dictionary + content_markdown = scrape_result.get('markdown') + content_html = scrape_result.get('html') + metadata = scrape_result.get('metadata', {}) + + # Ensure metadata is a dict (not an object) + if not isinstance(metadata, dict): + if hasattr(metadata, 'model_dump'): + metadata = metadata.model_dump() + elif hasattr(metadata, '__dict__'): + metadata = metadata.__dict__ + else: + metadata = {} + + # Get title from metadata + title = metadata.get("title", "") + + # Choose content based on requested format + chosen_content = content_markdown if (format == "markdown" or (format is None and content_markdown)) else content_html or content_markdown or "" + + results.append({ + "url": metadata.get("sourceURL", url), + "title": title, + "content": chosen_content, + "raw_content": chosen_content, + "metadata": metadata # Now guaranteed to be a dict + }) + + except Exception as scrape_err: + print(f" āŒ Error scraping {url}: {str(scrape_err)}") + results.append({ + "url": url, + "title": "", + "content": "", + "raw_content": "", + "error": str(scrape_err) + }) + + response = {"results": results} + + pages_extracted = len(response.get('results', [])) + print(f"āœ… Extracted content from {pages_extracted} pages") + + debug_call_data["pages_extracted"] = pages_extracted + debug_call_data["original_response_size"] = len(json.dumps(response)) + + # Process each result with LLM if enabled + if use_llm_processing and os.getenv("NOUS_API_KEY"): + print("šŸ§  Processing extracted content with LLM...") + debug_call_data["processing_applied"].append("llm_processing") + + for result in response.get('results', []): + url = result.get('url', 'Unknown URL') + title = result.get('title', '') + raw_content = result.get('raw_content', '') or result.get('content', '') + + if raw_content: + original_size = len(raw_content) + + # Process content with LLM + processed = await process_content_with_llm( + raw_content, url, title, model, min_length + ) + + if processed: + processed_size = len(processed) + compression_ratio = processed_size / original_size if original_size > 0 else 1.0 + + # Capture compression metrics + debug_call_data["compression_metrics"].append({ + "url": url, + "original_size": original_size, + "processed_size": processed_size, + "compression_ratio": compression_ratio, + "model_used": model + }) + + # Replace content with processed version + result['content'] = processed + # Keep raw content in separate field for reference + result['raw_content'] = raw_content + debug_call_data["pages_processed_with_llm"] += 1 + print(f" šŸ“ {url} (processed)") + else: + debug_call_data["compression_metrics"].append({ + "url": url, + "original_size": original_size, + "processed_size": original_size, + "compression_ratio": 1.0, + "model_used": None, + "reason": "content_too_short" + }) + print(f" šŸ“ {url} (no processing - content too short)") + else: + print(f" āš ļø {url} (no content to process)") + else: + if use_llm_processing and not os.getenv("NOUS_API_KEY"): + print("āš ļø LLM processing requested but NOUS_API_KEY not set, returning raw content") + debug_call_data["processing_applied"].append("llm_processing_unavailable") + + # Print summary of extracted pages for debugging (original behavior) + for result in response.get('results', []): + url = result.get('url', 'Unknown URL') + content_length = len(result.get('raw_content', '')) + print(f" šŸ“ {url} ({content_length} characters)") + + # Trim output to minimal fields per entry: title, content, error + trimmed_results = [ + { + "title": r.get("title", ""), + "content": r.get("content", ""), + "error": r.get("error"), + **({"llm_model": model} if use_llm_processing else {}) + } + for r in response.get("results", []) + ] + trimmed_response = {"results": trimmed_results} + # Include model name used for summarization when LLM processing was requested + if use_llm_processing: + trimmed_response["llm_model"] = model + + result_json = json.dumps(trimmed_response, indent=2) + # Clean base64 images from extracted content + cleaned_result = clean_base64_images(result_json) + + debug_call_data["final_response_size"] = len(cleaned_result) + debug_call_data["processing_applied"].append("base64_image_removal") + + # Log debug information + _log_debug_call("web_extract_tool", debug_call_data) + _save_debug_log() + + return cleaned_result + + except Exception as e: + error_msg = f"Error extracting content: {str(e)}" + print(f"āŒ {error_msg}") + + debug_call_data["error"] = error_msg + _log_debug_call("web_extract_tool", debug_call_data) + _save_debug_log() + + return json.dumps({"error": error_msg}) + + +async def web_crawl_tool( + url: str, + instructions: str = None, + depth: str = "basic", + use_llm_processing: bool = True, + model: str = DEFAULT_SUMMARIZER_MODEL, + min_length: int = DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION +) -> str: + """ + Crawl a website with specific instructions using available crawling API backend. + + This function provides a generic interface for web crawling that can work + with multiple backends. Currently uses Firecrawl. + + Args: + url (str): The base URL to crawl (can include or exclude https://) + instructions (str): Instructions for what to crawl/extract using LLM intelligence (optional) + depth (str): Depth of extraction ("basic" or "advanced", default: "basic") + use_llm_processing (bool): Whether to process content with LLM for summarization (default: True) + model (str): The model to use for LLM processing (default: gemini-2.5-flash) + min_length (int): Minimum content length to trigger LLM processing (default: 5000) + + Returns: + str: JSON string containing crawled content. If LLM processing is enabled and successful, + the 'content' field will contain the processed markdown summary instead of raw content. + Each page is processed individually. + + Raises: + Exception: If crawling fails or API key is not set + """ + debug_call_data = { + "parameters": { + "url": url, + "instructions": instructions, + "depth": depth, + "use_llm_processing": use_llm_processing, + "model": model, + "min_length": min_length + }, + "error": None, + "pages_crawled": 0, + "pages_processed_with_llm": 0, + "original_response_size": 0, + "final_response_size": 0, + "compression_metrics": [], + "processing_applied": [] + } + + try: + # Ensure URL has protocol + if not url.startswith(('http://', 'https://')): + url = f'https://{url}' + print(f" šŸ“ Added https:// prefix to URL: {url}") + + instructions_text = f" with instructions: '{instructions}'" if instructions else "" + print(f"šŸ•·ļø Crawling {url}{instructions_text}") + + # Use Firecrawl's v2 crawl functionality + # Docs: https://docs.firecrawl.dev/features/crawl + # The crawl() method automatically waits for completion and returns all data + + # Build crawl parameters - keep it simple + crawl_params = { + "limit": 20, # Limit number of pages to crawl + "scrape_options": { + "formats": ["markdown"] # Just markdown for simplicity + } + } + + # Note: The 'prompt' parameter is not documented for crawl + # Instructions are typically used with the Extract endpoint, not Crawl + if instructions: + print(f" ā„¹ļø Note: Instructions parameter ignored (not supported in crawl API)") + + # Use the crawl method which waits for completion automatically + try: + crawl_result = firecrawl_client.crawl( + url=url, + **crawl_params + ) + except Exception as e: + print(f" āŒ Crawl API call failed: {e}") + raise + + pages: List[Dict[str, Any]] = [] + + # Process crawl results - the crawl method returns a CrawlJob object with data attribute + data_list = [] + + # The crawl_result is a CrawlJob object with a 'data' attribute containing list of Document objects + if hasattr(crawl_result, 'data'): + data_list = crawl_result.data if crawl_result.data else [] + print(f" šŸ“Š Status: {getattr(crawl_result, 'status', 'unknown')}") + print(f" šŸ“„ Retrieved {len(data_list)} pages") + + # Debug: Check other attributes if no data + if not data_list: + print(f" šŸ” Debug - CrawlJob attributes: {[attr for attr in dir(crawl_result) if not attr.startswith('_')]}") + print(f" šŸ” Debug - Status: {getattr(crawl_result, 'status', 'N/A')}") + print(f" šŸ” Debug - Total: {getattr(crawl_result, 'total', 'N/A')}") + print(f" šŸ” Debug - Completed: {getattr(crawl_result, 'completed', 'N/A')}") + + elif isinstance(crawl_result, dict) and 'data' in crawl_result: + data_list = crawl_result.get("data", []) + else: + print(" āš ļø Unexpected crawl result type") + print(f" šŸ” Debug - Result type: {type(crawl_result)}") + if hasattr(crawl_result, '__dict__'): + print(f" šŸ” Debug - Result attributes: {list(crawl_result.__dict__.keys())}") + + for item in data_list: + # Process each crawled page - properly handle object serialization + page_url = "Unknown URL" + title = "" + content_markdown = None + content_html = None + metadata = {} + + # Extract data from the item + if hasattr(item, 'model_dump'): + # Pydantic model - use model_dump to get dict + item_dict = item.model_dump() + content_markdown = item_dict.get('markdown') + content_html = item_dict.get('html') + metadata = item_dict.get('metadata', {}) + elif hasattr(item, '__dict__'): + # Regular object with attributes + content_markdown = getattr(item, 'markdown', None) + content_html = getattr(item, 'html', None) + + # Handle metadata - convert to dict if it's an object + metadata_obj = getattr(item, 'metadata', {}) + if hasattr(metadata_obj, 'model_dump'): + metadata = metadata_obj.model_dump() + elif hasattr(metadata_obj, '__dict__'): + metadata = metadata_obj.__dict__ + elif isinstance(metadata_obj, dict): + metadata = metadata_obj + else: + metadata = {} + elif isinstance(item, dict): + # Already a dictionary + content_markdown = item.get('markdown') + content_html = item.get('html') + metadata = item.get('metadata', {}) + + # Ensure metadata is a dict (not an object) + if not isinstance(metadata, dict): + if hasattr(metadata, 'model_dump'): + metadata = metadata.model_dump() + elif hasattr(metadata, '__dict__'): + metadata = metadata.__dict__ + else: + metadata = {} + + # Extract URL and title from metadata + page_url = metadata.get("sourceURL", metadata.get("url", "Unknown URL")) + title = metadata.get("title", "") + + # Choose content (prefer markdown) + content = content_markdown or content_html or "" + + pages.append({ + "url": page_url, + "title": title, + "content": content, + "raw_content": content, + "metadata": metadata # Now guaranteed to be a dict + }) + + response = {"results": pages} + + pages_crawled = len(response.get('results', [])) + print(f"āœ… Crawled {pages_crawled} pages") + + debug_call_data["pages_crawled"] = pages_crawled + debug_call_data["original_response_size"] = len(json.dumps(response)) + + # Process each result with LLM if enabled + if use_llm_processing and os.getenv("NOUS_API_KEY"): + print("šŸ§  Processing crawled content with LLM...") + debug_call_data["processing_applied"].append("llm_processing") + + for result in response.get('results', []): + page_url = result.get('url', 'Unknown URL') + title = result.get('title', '') + content = result.get('content', '') + + if content: + original_size = len(content) + + # Process content with LLM + processed = await process_content_with_llm( + content, page_url, title, model, min_length + ) + + if processed: + processed_size = len(processed) + compression_ratio = processed_size / original_size if original_size > 0 else 1.0 + + # Capture compression metrics + debug_call_data["compression_metrics"].append({ + "url": page_url, + "original_size": original_size, + "processed_size": processed_size, + "compression_ratio": compression_ratio, + "model_used": model + }) + + # Keep original content in raw_content field + result['raw_content'] = content + # Replace content with processed version + result['content'] = processed + debug_call_data["pages_processed_with_llm"] += 1 + print(f" šŸŒ {page_url} (processed)") + else: + debug_call_data["compression_metrics"].append({ + "url": page_url, + "original_size": original_size, + "processed_size": original_size, + "compression_ratio": 1.0, + "model_used": None, + "reason": "content_too_short" + }) + print(f" šŸŒ {page_url} (no processing - content too short)") + else: + print(f" āš ļø {page_url} (no content to process)") + else: + if use_llm_processing and not os.getenv("NOUS_API_KEY"): + print("āš ļø LLM processing requested but NOUS_API_KEY not set, returning raw content") + debug_call_data["processing_applied"].append("llm_processing_unavailable") + + # Print summary of crawled pages for debugging (original behavior) + for result in response.get('results', []): + page_url = result.get('url', 'Unknown URL') + content_length = len(result.get('content', '')) + print(f" šŸŒ {page_url} ({content_length} characters)") + + # Trim output to minimal fields per entry: title, content, error + trimmed_results = [ + { + "title": r.get("title", ""), + "content": r.get("content", ""), + "error": r.get("error"), + **({"llm_model": model} if use_llm_processing else {}) + } + for r in response.get("results", []) + ] + trimmed_response = {"results": trimmed_results} + # Include model name used for summarization when LLM processing was requested + if use_llm_processing: + trimmed_response["llm_model"] = model + + result_json = json.dumps(trimmed_response, indent=2) + # Clean base64 images from crawled content + cleaned_result = clean_base64_images(result_json) + + debug_call_data["final_response_size"] = len(cleaned_result) + debug_call_data["processing_applied"].append("base64_image_removal") + + # Log debug information + _log_debug_call("web_crawl_tool", debug_call_data) + _save_debug_log() + + return cleaned_result + + except Exception as e: + error_msg = f"Error crawling website: {str(e)}" + print(f"āŒ {error_msg}") + + debug_call_data["error"] = error_msg + _log_debug_call("web_crawl_tool", debug_call_data) + _save_debug_log() + + return json.dumps({"error": error_msg}) + + +# Convenience function to check if API key is available +def check_firecrawl_api_key() -> bool: + """ + Check if the Firecrawl API key is available in environment variables. + + Returns: + bool: True if API key is set, False otherwise + """ + return bool(os.getenv("FIRECRAWL_API_KEY")) + + +def check_nous_api_key() -> bool: + """ + Check if the Nous Research API key is available in environment variables. + + Returns: + bool: True if API key is set, False otherwise + """ + return bool(os.getenv("NOUS_API_KEY")) + + +def get_debug_session_info() -> Dict[str, Any]: + """ + Get information about the current debug session. + + Returns: + Dict[str, Any]: Dictionary containing debug session information: + - enabled: Whether debug mode is enabled + - session_id: Current session UUID (if enabled) + - log_path: Path where debug logs are saved (if enabled) + - total_calls: Number of tool calls logged so far (if enabled) + """ + if not DEBUG_MODE or not DEBUG_DATA: + return { + "enabled": False, + "session_id": None, + "log_path": None, + "total_calls": 0 + } + + return { + "enabled": True, + "session_id": DEBUG_SESSION_ID, + "log_path": str(DEBUG_LOG_PATH / f"web_tools_debug_{DEBUG_SESSION_ID}.json"), + "total_calls": len(DEBUG_DATA["tool_calls"]) + } + + +if __name__ == "__main__": + """ + Simple test/demo when run directly + """ + print("šŸŒ Standalone Web Tools Module") + print("=" * 40) + + # Check if API keys are available + firecrawl_available = check_firecrawl_api_key() + nous_available = check_nous_api_key() + + if not firecrawl_available: + print("āŒ FIRECRAWL_API_KEY environment variable not set") + print("Please set your API key: export FIRECRAWL_API_KEY='your-key-here'") + print("Get API key at: https://firecrawl.dev/") + else: + print("āœ… Firecrawl API key found") + + if not nous_available: + print("āŒ NOUS_API_KEY environment variable not set") + print("Please set your API key: export NOUS_API_KEY='your-key-here'") + print("Get API key at: https://inference-api.nousresearch.com/") + print("āš ļø Without Nous API key, LLM content processing will be disabled") + else: + print("āœ… Nous Research API key found") + + if not firecrawl_available: + exit(1) + + print("šŸ› ļø Web tools ready for use!") + + if nous_available: + print("šŸ§  LLM content processing available with Gemini 2.5 Flash") + print(f" Default min length for processing: {DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION} chars") + + # Show debug mode status + if DEBUG_MODE: + print(f"šŸ› Debug mode ENABLED - Session ID: {DEBUG_SESSION_ID}") + print(f" Debug logs will be saved to: ./logs/web_tools_debug_{DEBUG_SESSION_ID}.json") + else: + print("šŸ› Debug mode disabled (set WEB_TOOLS_DEBUG=true to enable)") + + print("\nBasic usage:") + print(" from web_tools import web_search_tool, web_extract_tool, web_crawl_tool") + print(" import asyncio") + print("") + print(" # Search (synchronous)") + print(" results = web_search_tool('Python tutorials')") + print("") + print(" # Extract and crawl (asynchronous)") + print(" async def main():") + print(" content = await web_extract_tool(['https://example.com'])") + print(" crawl_data = await web_crawl_tool('example.com', 'Find docs')") + print(" asyncio.run(main())") + + if nous_available: + print("\nLLM-enhanced usage:") + print(" # Content automatically processed for pages >5000 chars (default)") + print(" content = await web_extract_tool(['https://python.org/about/'])") + print("") + print(" # Customize processing parameters") + print(" crawl_data = await web_crawl_tool(") + print(" 'docs.python.org',") + print(" 'Find key concepts',") + print(" model='gemini-2.5-flash',") + print(" min_length=3000") + print(" )") + print("") + print(" # Disable LLM processing") + print(" raw_content = await web_extract_tool(['https://example.com'], use_llm_processing=False)") + + print("\nDebug mode:") + print(" # Enable debug logging") + print(" export WEB_TOOLS_DEBUG=true") + print(" # Debug logs capture:") + print(" # - All tool calls with parameters") + print(" # - Original API responses") + print(" # - LLM compression metrics") + print(" # - Final processed results") + print(" # Logs saved to: ./logs/web_tools_debug_UUID.json") + + print(f"\nšŸ“ Run 'python test_web_tools_llm.py' to test LLM processing capabilities")