tools/simple_tools.py

"""Simple synchronous tools for LlamaIndex ReActAgent."""

import os
import time
import hashlib
import json
from typing import Optional
from datetime import datetime
from tavily import TavilyClient  # Use sync client
from llama_index.core.tools import FunctionTool

# Global state store - simple in-memory storage
_workflow_state = {
    "research_notes": {},
    "report_content": "Not written yet.",
    "review": "Review required.",
    "structured_report": None
}

# Global cache to track recent tool calls
_tool_call_cache = {}
_cache_timeout = 30

def _generate_call_hash(tool_name: str, **kwargs) -> str:
    """Generate a hash for tool call deduplication."""
    call_data = {"tool": tool_name, "args": kwargs}
    call_str = json.dumps(call_data, sort_keys=True)
    return hashlib.md5(call_str.encode()).hexdigest()

def _should_execute_call(tool_name: str, **kwargs) -> bool:
    """Check if a tool call should be executed or if it's a duplicate."""
    current_time = time.time()
    call_hash = _generate_call_hash(tool_name, **kwargs)
    
    # Clean up old cache entries
    expired_keys = [k for k, v in _tool_call_cache.items() if current_time - v > _cache_timeout]
    for key in expired_keys:
        del _tool_call_cache[key]
    
    # Check if this call was made recently
    if call_hash in _tool_call_cache:
        return False
    
    # Record this call
    _tool_call_cache[call_hash] = current_time
    return True

def search_web(query: str) -> str:
    """Search the web for information on a given query."""
    try:
        print(f"DEBUG: search_web called with query: '{query}'")
        
        # Check for duplicate calls
        if not _should_execute_call("search_web", query=query):
            return f"Duplicate search call detected for query: '{query}'. Skipping to avoid redundant API calls."
        
        # Use synchronous Tavily client
        client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
        result = client.search(query)
        
        print(f"DEBUG: search_web executed successfully for query: '{query}'")
        return str(result)
    except Exception as e:
        error_msg = f"Search failed: {str(e)}"
        print(f"ERROR: search_web failed: {e}")
        return error_msg

def record_notes(notes: str, notes_title: str) -> str:
    """Record notes on a given topic with a title."""
    try:
        print(f"DEBUG: record_notes called with title: '{notes_title}', notes length: {len(notes)}")
        
        # Check for duplicate calls
        if not _should_execute_call("record_notes", notes=notes, notes_title=notes_title):
            return f"Duplicate notes recording detected for title: '{notes_title}'. Skipping to avoid redundant recording."
        
        # Store in global state
        _workflow_state["research_notes"][notes_title] = notes
        
        print(f"DEBUG: Notes stored. Total research notes: {len(_workflow_state['research_notes'])}")
        return f"Notes recorded successfully with title: '{notes_title}'. Total notes: {len(_workflow_state['research_notes'])}"
    except Exception as e:
        error_msg = f"Failed to record notes: {str(e)}"
        print(f"ERROR: record_notes failed: {e}")
        return error_msg

def write_report(report_content: str, title: str = "Research Report") -> str:
    """Write a structured report with the given content and title."""
    try:
        print(f"DEBUG: write_report FUNCTION ENTERED with title: '{title}', content length: {len(report_content)}")
        print(f"DEBUG: Function arguments - report_content type: {type(report_content)}, title type: {type(title)}")
        
        # Check for duplicate calls
        if not _should_execute_call("write_report", report_content=report_content, title=title):
            print("DEBUG: Duplicate call detected, returning early")
            return "Duplicate report writing detected. Skipping to avoid redundant report generation."
        
        print("DEBUG: Processing report content...")
        
        # Extract sections from markdown content
        import re
        sections = re.findall(r'^#{1,3}\s+(.+)$', report_content, re.MULTILINE)
        print(f"DEBUG: Found {len(sections)} sections: {sections}")
        
        # Calculate word count
        word_count = len(report_content.split())
        print(f"DEBUG: Word count: {word_count}")
        
        # Extract abstract (first paragraph after title)
        lines = report_content.split('\n')
        abstract = ""
        for line in lines:
            if line.strip() and not line.startswith('#'):
                abstract = line.strip()
                break
        print(f"DEBUG: Abstract: {abstract[:100]}...")
        
        # Create structured report
        structured_report = {
            "title": title,
            "abstract": abstract[:200] + "..." if len(abstract) > 200 else abstract,
            "content": report_content,
            "sections": sections,
            "word_count": word_count,
            "generated_at": datetime.now().isoformat(),
            "sources_used": list(_workflow_state["research_notes"].keys())
        }
        print("DEBUG: Structured report created")
        
        # Store in global state
        print("DEBUG: Storing in global state...")
        _workflow_state["report_content"] = report_content
        _workflow_state["structured_report"] = structured_report
        
        print(f"DEBUG: Report stored successfully. Word count: {word_count}, Sections: {len(sections)}")
        print(f"DEBUG: State keys now: {list(_workflow_state.keys())}")
        print(f"DEBUG: State report_content length: {len(_workflow_state['report_content'])}")
        
        result = f"Report written successfully! Title: '{title}', Word count: {word_count}, Sections: {len(sections)}"
        print(f"DEBUG: Returning result: {result}")
        return result
    except Exception as e:
        error_msg = f"Failed to write report: {str(e)}"
        print(f"ERROR: write_report failed: {e}")
        import traceback
        traceback.print_exc()
        return error_msg

def review_report(review: str) -> str:
    """Review a report and provide feedback."""
    try:
        print(f"DEBUG: review_report called with review: '{review[:100]}...'")
        
        # Check for duplicate calls
        if not _should_execute_call("review_report", review=review):
            return "Duplicate review detected. Skipping to avoid redundant review submission."
        
        # Store review in global state
        _workflow_state["review"] = review
        
        print(f"DEBUG: Review stored successfully")
        return f"Report reviewed successfully. Review: {review[:100]}{'...' if len(review) > 100 else ''}"
    except Exception as e:
        error_msg = f"Failed to review report: {str(e)}"
        print(f"ERROR: review_report failed: {e}")
        return error_msg

def get_workflow_state() -> dict:
    """Get the current workflow state."""
    return _workflow_state.copy()

def reset_workflow_state():
    """Reset the workflow state."""
    global _workflow_state
    _workflow_state = {
        "research_notes": {},
        "report_content": "Not written yet.",
        "review": "Review required.",
        "structured_report": None
    }

# Create LlamaIndex FunctionTool instances with better descriptions
search_web_tool = FunctionTool.from_defaults(
    fn=search_web,
    name="search_web",
    description=(
        "Search the web for information on any topic. "
        "Input: A search query string. "
        "Output: Search results containing relevant information. "
        "Use this to gather facts and information about your research topic."
    ),
)

record_notes_tool = FunctionTool.from_defaults(
    fn=record_notes,
    name="record_notes", 
    description=(
        "Record research notes with a descriptive title. "
        "Input: notes (string) - the content to save, notes_title (string) - a title for the notes. "
        "Output: Confirmation that notes were saved. "
        "Use this after searching to save important information you found."
    ),
)

write_report_tool = FunctionTool.from_defaults(
    fn=write_report,
    name="write_report",
    description=(
        "Write a comprehensive markdown report. "
        "Input: report_content (string) - full markdown report content, title (string, optional) - report title. "
        "Output: Confirmation that report was written. "
        "The report_content should be well-structured markdown with headers, sections, and detailed content."
    ),
)

review_report_tool = FunctionTool.from_defaults(
    fn=review_report,
    name="review_report",
    description=(
        "Review a written report and provide feedback. "
        "Input: review (string) - your review and feedback on the report. "
        "Output: Confirmation that review was recorded. "
        "Start with 'APPROVED:' if the report is satisfactory, otherwise provide specific improvement suggestions."
    ),
)