check_tool_usage.py

"""
Script to check if CoordinatorAgent ever uses tools.

This script analyzes the code and runtime behavior to determine if tools
are actually being invoked during agent execution.
"""

import ast
import inspect
from typing import List, Dict, Set
import sys
import os

# Add the project root to the path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

# Try to import, but handle import errors gracefully
try:
    from src.coordinator_agent import CoordinatorAgent
    CAN_IMPORT_AGENT = True
except ImportError as e:
    CAN_IMPORT_AGENT = False
    IMPORT_ERROR = str(e)


class ToolUsageChecker:
    """Checks if tools are used in the CoordinatorAgent."""
    
    def __init__(self):
        self.tool_names = ["ask_question", "fail_match", "success_match"]
        self.tool_usage_found = {
            "ask_question": False,
            "fail_match": False,
            "success_match": False
        }
        self.tool_methods_called = {
            "ask_question": False,
            "fail_match": False,
            "success_match": False
        }
    
    def check_code_structure(self) -> Dict:
        """Analyze the code structure to see if tools are bound to LLM."""
        results = {
            "tools_initialized": False,
            "tools_bound_to_llm": False,
            "agent_framework_used": False,
            "direct_llm_call": False,
            "tool_invocation_found": False
        }
        
        # Read the coordinator_agent.py file
        file_path = "src/coordinator_agent.py"
        with open(file_path, "r") as f:
            code = f.read()
        
        tree = ast.parse(code)
        
        # Check if tools are initialized
        for node in ast.walk(tree):
            if isinstance(node, ast.Assign):
                for target in node.targets:
                    if isinstance(target, ast.Attribute):
                        if isinstance(target.value, ast.Name) and target.value.id == "self":
                            if target.attr == "tools":
                                results["tools_initialized"] = True
                    elif isinstance(target, ast.Name) and target.id == "self.tools":
                        results["tools_initialized"] = True
            
            # Also check for _initialize_tools method
            if isinstance(node, ast.FunctionDef) and node.name == "_initialize_tools":
                results["tools_initialized"] = True
            
            # Check if LLM is called with tools
            if isinstance(node, ast.Call):
                if isinstance(node.func, ast.Attribute):
                    # Check for .bind_tools() or similar
                    if node.func.attr in ["bind_tools", "bind", "with_tools"]:
                        results["tools_bound_to_llm"] = True
                    
                    # Check for direct LLM invoke
                    if node.func.attr == "invoke":
                        results["direct_llm_call"] = True
            
            # Check for agent framework usage
            if isinstance(node, ast.Import):
                for alias in node.names:
                    if "agent" in alias.name.lower() or "chain" in alias.name.lower():
                        if "langchain" in alias.name or "langgraph" in alias.name:
                            results["agent_framework_used"] = True
        
        return results
    
    def check_runtime_behavior(self) -> Dict:
        """Check runtime behavior by inspecting the agent instance."""
        results = {
            "tools_exist": False,
            "tools_count": 0,
            "llm_has_tools": False,
            "methods_exist": False
        }
        
        if not CAN_IMPORT_AGENT:
            results["error"] = f"Cannot import CoordinatorAgent: {IMPORT_ERROR}"
            results["note"] = "Skipping runtime checks - run with venv activated"
            return results
        
        try:
            agent = CoordinatorAgent()
            
            # Check if tools exist
            if hasattr(agent, "tools"):
                results["tools_exist"] = True
                results["tools_count"] = len(agent.tools) if agent.tools else 0
            
            # Check if LLM has tools bound
            if hasattr(agent, "llm"):
                llm = agent.llm
                # Check various ways tools might be bound
                if hasattr(llm, "bound_tools"):
                    results["llm_has_tools"] = bool(llm.bound_tools)
                elif hasattr(llm, "tools"):
                    results["llm_has_tools"] = bool(llm.tools)
                elif hasattr(llm, "lc_kwargs"):
                    kwargs = llm.lc_kwargs if hasattr(llm, "lc_kwargs") else {}
                    if "tools" in kwargs:
                        results["llm_has_tools"] = True
            
            # Check if llm_with_tools exists and has tools
            if hasattr(agent, "llm_with_tools"):
                llm_with_tools = agent.llm_with_tools
                # Check if tools are bound to llm_with_tools
                # RunnableBinding typically has bound_tools attribute
                if hasattr(llm_with_tools, "bound_tools"):
                    bound_tools = llm_with_tools.bound_tools
                    results["llm_has_tools"] = bool(bound_tools) if bound_tools else False
                elif hasattr(llm_with_tools, "tools"):
                    results["llm_has_tools"] = bool(llm_with_tools.tools)
                elif hasattr(llm_with_tools, "kwargs"):
                    kwargs = llm_with_tools.kwargs if hasattr(llm_with_tools, "kwargs") else {}
                    if "tools" in kwargs:
                        results["llm_has_tools"] = True
                # If it's a RunnableBinding, tools are definitely bound
                if "RunnableBinding" in type(llm_with_tools).__name__:
                    results["llm_has_tools"] = True
                    results["note"] = "Tools are bound to llm_with_tools (RunnableBinding)"
            
            # Check if tool methods exist
            results["methods_exist"] = all(
                hasattr(agent, method) 
                for method in ["ask_question", "fail_match", "success_match"]
            )
            
        except Exception as e:
            results["error"] = str(e)
        
        return results
    
    def trace_tool_methods(self) -> Dict:
        """Trace if tool methods are ever called."""
        # This would require runtime tracing/hooking
        # For now, we'll check if they're called in the codebase
        results = {
            "ask_question_called": False,
            "fail_match_called": False,
            "success_match_called": False
        }
        
        # Search for method calls in the codebase
        codebase_files = [
            "src/coordinator_agent.py",
            "main.py",
            "tests/Coordinator_test.py"
        ]
        
        for file_path in codebase_files:
            if not os.path.exists(file_path):
                continue
                
            with open(file_path, "r") as f:
                code = f.read()
            
            # Simple string search for method calls
            if ".ask_question(" in code or "ask_question(" in code:
                results["ask_question_called"] = True
            if ".fail_match(" in code or "fail_match(" in code:
                results["fail_match_called"] = True
            if ".success_match(" in code or "success_match(" in code:
                results["success_match_called"] = True
        
        return results
    
    def check_llm_integration(self) -> Dict:
        """Check how the LLM is integrated and if it can use tools."""
        results = {
            "llm_type": None,
            "has_structured_output": False,
            "can_use_tools": False
        }
        
        if not CAN_IMPORT_AGENT:
            results["error"] = f"Cannot import CoordinatorAgent: {IMPORT_ERROR}"
            results["note"] = "Skipping runtime checks - run with venv activated"
            return results
        
        try:
            agent = CoordinatorAgent()
            
            # Check LLM type
            llm = agent.llm
            results["llm_type"] = type(llm).__name__
            
            # Check if structured output is used
            if hasattr(llm, "with_structured_output"):
                results["has_structured_output"] = True
            
            # Check if LLM supports tool calling
            # Check if llm_with_tools exists (tools are bound to a separate LLM instance)
            if hasattr(agent, "llm_with_tools"):
                results["can_use_tools"] = True
                results["note"] = "Tools are bound to llm_with_tools (separate from structured output LLM)"
            elif "structured_output" in str(type(llm)).lower():
                results["can_use_tools"] = False  # Structured output bypasses tool calling
            elif hasattr(llm, "bind_tools") or hasattr(llm, "with_tools"):
                results["can_use_tools"] = True
            
        except Exception as e:
            results["error"] = str(e)
        
        return results
    
    def generate_report(self) -> str:
        """Generate a comprehensive report."""
        print("=" * 70)
        print("COORDINATOR AGENT TOOL USAGE ANALYSIS")
        print("=" * 70)
        print()
        
        # Code structure analysis
        print("1. CODE STRUCTURE ANALYSIS")
        print("-" * 70)
        code_analysis = self.check_code_structure()
        for key, value in code_analysis.items():
            status = "✓" if value else "✗"
            print(f"  {status} {key}: {value}")
        print()
        
        # Runtime behavior
        print("2. RUNTIME BEHAVIOR ANALYSIS")
        print("-" * 70)
        runtime_analysis = self.check_runtime_behavior()
        for key, value in runtime_analysis.items():
            if key not in ["error", "note"]:
                status = "✓" if value else "✗"
                print(f"  {status} {key}: {value}")
            elif key == "error":
                print(f"  ⚠ ERROR: {value}")
        # Print note if it exists (only once)
        if "note" in runtime_analysis:
            print(f"  ℹ NOTE: {runtime_analysis['note']}")
        print()
        
        # Method call tracing
        print("3. METHOD CALL TRACING")
        print("-" * 70)
        method_tracing = self.trace_tool_methods()
        for key, value in method_tracing.items():
            status = "✓" if value else "✗"
            print(f"  {status} {key}: {value}")
        print()
        
        # LLM integration
        print("4. LLM INTEGRATION ANALYSIS")
        print("-" * 70)
        llm_analysis = self.check_llm_integration()
        for key, value in llm_analysis.items():
            if key not in ["error", "note"]:
                print(f"  {key}: {value}")
            elif key == "error":
                print(f"  ⚠ ERROR: {value}")
            elif key == "note":
                print(f"  ℹ NOTE: {value}")
        print()
        
        # Conclusion
        print("5. CONCLUSION")
        print("-" * 70)
        
        tools_initialized = code_analysis.get("tools_initialized", False)
        tools_bound = code_analysis.get("tools_bound_to_llm", False)
        direct_call = code_analysis.get("direct_llm_call", False)
        has_structured_output = llm_analysis.get("has_structured_output", False)
        runtime_error = runtime_analysis.get("error", "")
        can_use_tools = llm_analysis.get("can_use_tools", False)
        tools_exist_runtime = runtime_analysis.get("tools_exist", False)
        llm_has_tools_runtime = runtime_analysis.get("llm_has_tools", False)
        
        # Check if we have runtime verification
        has_runtime_verification = not runtime_error and CAN_IMPORT_AGENT
        
        if not has_runtime_verification:
            print("  ⚠ PARTIAL ANALYSIS: Cannot verify runtime behavior")
            print()
            print("  Code Analysis Results:")
            print(f"  - ✓ Tools initialized: {tools_initialized}")
            print(f"  - ✓ Tools bound to LLM: {tools_bound}")
            print(f"  - ✓ Direct LLM call: {direct_call}")
            print()
            print("  Note: Runtime checks skipped due to import error.")
            print("  To verify tool usage at runtime, run with venv activated:")
            print("    source venv/bin/activate && python check_tool_usage.py")
            print()
            if tools_initialized and tools_bound:
                print("  Based on code analysis:")
                print("  ✓ Tools are defined and bound to llm_with_tools")
                print("  ✓ Tool-calling implementation exists in __organize_match()")
                print("  ⚠ Cannot verify actual tool execution without runtime checks")
        elif tools_initialized and tools_bound and direct_call:
            if tools_exist_runtime and llm_has_tools_runtime and can_use_tools:
                print("  ✓ CONFIRMED: Tools are properly bound and can be used!")
                print()
                print("  Verification:")
                print(f"  - ✓ Tools exist: {tools_exist_runtime} ({runtime_analysis.get('tools_count', 0)} tools)")
                print(f"  - ✓ Tools bound to LLM: {llm_has_tools_runtime}")
                print(f"  - ✓ Can use tools: {can_use_tools}")
                print(f"  - ✓ Tool methods exist: {runtime_analysis.get('methods_exist', False)}")
                print()
                print("  Implementation:")
                print("  - Tools are bound to self.llm_with_tools")
                print("  - __organize_match() uses tool-calling loop")
                print("  - Tools can be executed: ask_question, fail_match, success_match")
            else:
                print("  ⚠ WARNING: Tools are defined but may not be properly configured!")
                print()
                print("  Analysis:")
                print(f"  - Tools exist: {tools_exist_runtime}")
                print(f"  - Tools bound: {llm_has_tools_runtime}")
                print(f"  - Can use tools: {can_use_tools}")
        elif tools_bound:
            print("  ✓ Tools appear to be bound to the LLM")
        else:
            print("  ✗ Tools are not being used")
        
        print()
        print("=" * 70)
        
        return "Report generated"


def main():
    """Main function to run the tool usage checker."""
    checker = ToolUsageChecker()
    report = checker.generate_report()
    return checker


if __name__ == "__main__":
    main()