app.py

import gradio as gr
import json
import re
import math
import os
import time
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum

# ============================================================
# Monkey-patch Gradio 5.9.0 get_api_info crash
# Fixes: TypeError: argument of type 'bool' is not iterable
# in gradio_client/utils.py:887 (if "const" in schema:)
# ============================================================
import gradio_client.utils as gradio_utils
_orig_get_type = gradio_utils.get_type
def _patched_get_type(schema):
    if isinstance(schema, bool):
        return "boolean"
    return _orig_get_type(schema)
gradio_utils.get_type = _patched_get_type

# ============================================================
# Tool Definitions
# ============================================================

def safe_calc(expression: str) -> str:
    """Safe calculator using restricted eval."""
    import ast, operator as op
    ops = {
        ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul,
        ast.Div: op.truediv, ast.Pow: op.pow, ast.Mod: op.mod,
        ast.FloorDiv: op.floordiv, ast.USub: op.neg, ast.UAdd: op.pos,
    }
    funcs = {
        "abs": abs, "round": round, "int": int, "float": float,
        "min": min, "max": max, "sum": sum, "len": len,
        "sqrt": math.sqrt, "log": math.log, "sin": math.sin, "cos": math.cos,
        "pi": lambda: math.pi, "e": lambda: math.e,
    }
    consts = {"pi": math.pi, "e": math.e, "tau": math.tau}
    blocked = ["__", "import", "exec", "eval", "open", "os.", "subprocess", "sys."]
    
    for b in blocked:
        if b in expression.lower():
            return f"❌ Blocked pattern: {b}"
    
    def eval_node(node):
        if isinstance(node, ast.Expression):
            return eval_node(node.body)
        elif isinstance(node, ast.Constant):
            return node.value
        elif isinstance(node, ast.Num):
            return node.n
        elif isinstance(node, ast.Name):
            if node.id in consts:
                return consts[node.id]
            raise NameError(f"Unknown: {node.id}")
        elif isinstance(node, ast.UnaryOp):
            return ops[type(node.op)](eval_node(node.operand))
        elif isinstance(node, ast.BinOp):
            return ops[type(node.op)](eval_node(node.left), eval_node(node.right))
        elif isinstance(node, ast.Call):
            if isinstance(node.func, ast.Name) and node.func.id in funcs:
                args = [eval_node(a) for a in node.args]
                return funcs[node.func.id](*args) if callable(funcs[node.func.id]) else funcs[node.func.id]()
            raise NameError(f"Unknown function")
        raise ValueError(f"Unsupported: {type(node).__name__}")
    
    try:
        tree = ast.parse(expression.strip(), mode='eval')
        result = eval_node(tree.body)
        if isinstance(result, float):
            if result == int(result) and abs(result) < 1e15:
                return str(int(result))
            return f"{result:.4f}".rstrip("0").rstrip(".")
        return str(result)
    except Exception as e:
        return f"❌ Error: {e}"


MOCK_KB = {
    "python": "Python is a high-level, general-purpose programming language created by Guido van Rossum in 1991.",
    "langgraph": "LangGraph is a library for building stateful, multi-actor applications with LLMs. It extends LangChain by adding graph-based orchestration of agents.",
    "langchain": "LangChain is a framework for developing applications powered by language models. It provides tools, chains, and agents.",
    "qwen": "Qwen2.5 is Alibaba Cloud's LLM series. Qwen2.5-1.5B has 1.5B parameters and supports 32K context.",
    "gradio": "Gradio is a Python library for building ML web demos. It provides UI components for models.",
    "huggingface": "Hugging Face provides the Transformers library, model hub, and Spaces for ML demos.",
    "machine learning": "Machine learning enables systems to learn patterns from data without being explicitly programmed.",
    "transformer": "A deep learning architecture using self-attention, introduced in 'Attention Is All You Need' (2017).",
    "kaggle": "Kaggle is a data science community platform owned by Google, offering competitions, datasets, and notebooks.",
    "agent": "An AI agent perceives its environment, makes decisions, and takes actions to achieve goals. Tool-calling agents use external tools.",
}

def web_search(query: str) -> str:
    """Mock web search with knowledge base."""
    q = query.lower().strip()
    results = []
    for kw, info in MOCK_KB.items():
        if kw in q:
            results.append(f"📖 **{kw.title()}**: {info}")
    if results:
        return "\n\n".join(results[:3])
    return f"📭 No results found for '{query}'. Try rephrasing."


def read_file(path: str) -> str:
    """Read a text file safely."""
    if ".." in path:
        return "❌ Directory traversal blocked."
    try:
        if not os.path.exists(path):
            return f"❌ File not found: {path}"
        if os.path.isdir(path):
            items = os.listdir(path)
            return f"📁 Directory: {path}\n" + "\n".join(f"  {'📄' if os.path.isfile(os.path.join(path,f)) else '📁'} {f}" for f in items[:20])
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            content = f.read(2000)
        return f"📄 {path}\n\n{content}"
    except Exception as e:
        return f"❌ Error: {e}"


TOOLS = {
    "calculator": {"fn": safe_calc, "desc": "Evaluate math expressions (e.g., 2 + 2, sqrt(144), pi * 5^2)"},
    "web_search": {"fn": web_search, "desc": "Search the knowledge base for information"},
    "file_reader": {"fn": read_file, "desc": "Read a text file from the filesystem"},
}

# ============================================================
# Agent Engine
# ============================================================

class StepType(Enum):
    THOUGHT = "thought"
    TOOL_CALL = "tool_call" 
    TOOL_RESULT = "tool_result"
    FINAL = "final"

@dataclass
class AgentStep:
    type: StepType
    content: str
    tool_name: Optional[str] = None
    tool_input: Optional[str] = None
    tool_output: Optional[str] = None
    duration_ms: float = 0.0

def detect_intent(query: str) -> dict:
    """Detect what the user wants and route to appropriate tool."""
    q = query.lower().strip()
    
    # Calculator patterns
    calc_patterns = [
        r"(?:calculate|compute|what\s+is|evaluate|solve|how\s+much\s+is)\s+(.+)",
        r"(.+)\s*[+\-*/^%].+",  # contains math operators
    ]
    for pat in calc_patterns:
        m = re.search(pat, q)
        if m:
            expr = m.group(1) if m.lastindex else q
            # Clean up the expression
            expr = re.sub(r"^(?:calculate|compute|what\s+is|evaluate|solve|how\s+much\s+is)\s+", "", expr, flags=re.IGNORECASE).strip()
            if any(op in expr for op in ["+", "-", "*", "/", "^", "%", "sqrt", "log", "sin", "cos", "abs", "round", "pi", "e"]):
                return {"tool": "calculator", "input": expr}
    
    # File reader patterns
    if re.search(r"(?:read|open|show|list|cat|view|display|contents of)\s+(?:file\s+)?(.+)", q):
        m = re.search(r"(?:read|open|show|list|cat|view|display|contents of)\s+(?:file\s+)?(.+)", q)
        path = m.group(1).strip().strip('"\'')
        return {"tool": "file_reader", "input": path}
    
    if q.startswith("read ") or q.startswith("open ") or q.startswith("list "):
        parts = q.split(" ", 1)
        if len(parts) > 1:
            return {"tool": "file_reader", "input": parts[1].strip()}
    
    # Web search patterns (everything else with a question)
    if any(w in q for w in ["what", "who", "when", "where", "why", "how", "tell me", "explain", "about", "define"]):
        return {"tool": "web_search", "input": q}
    
    # Default: check for math operators
    if re.search(r"[\d\s]*[+\-*/^][\d\s]*", q):
        return {"tool": "calculator", "input": q}
    
    # Greetings - no tool needed
    greetings = ["hi", "hello", "hey", "greetings", "good morning", "good afternoon", "good evening"]
    if any(g in q for g in greetings):
        return {"tool": None, "input": q}
    
    # Fallback to web search
    return {"tool": "web_search", "input": q}


def run_agent(query: str) -> List[AgentStep]:
    """Run the agent pipeline and return all steps."""
    steps = []
    t_start = time.time()
    
    # Step 1: Thought
    thought_start = time.time()
    intent = detect_intent(query)
    thought_duration = (time.time() - thought_start) * 1000
    
    if intent["tool"] is None:
        # Direct response (no tool needed)
        steps.append(AgentStep(
            type=StepType.THOUGHT,
            content=f"The user said: '{query}'. This appears to be a greeting or simple statement — no tool needed.",
            duration_ms=thought_duration,
        ))
        steps.append(AgentStep(
            type=StepType.FINAL,
            content=f"Hello! I'm your Tool-Calling Agent. I can help you with:\n\n"
                    f"🔢 **Calculator** — evaluate math expressions\n"
                    f"🌐 **Web Search** — look up information\n"
                    f"📁 **File Reader** — read files\n\n"
                    f"Try asking me something like:\n"
                    f"• \"What is 25 * 4 + 10?\"\n"
                    f"• \"Tell me about LangGraph\"\n"
                    f"• \"Read /kaggle/working/somefile.txt\"",
            duration_ms=(time.time() - t_start) * 1000,
        ))
        return steps
    
    tool_name = intent["tool"]
    tool_input = intent["input"]
    tool_info = TOOLS[tool_name]
    
    # Step 2: Thought about which tool
    steps.append(AgentStep(
        type=StepType.THOUGHT,
        content=f"I need to answer: '{query}'\n\n"
                f"→ Detected intent requires **{tool_name}**\n"
                f"→ Tool description: {tool_info['desc']}\n"
                f"→ Input: {tool_input[:100]}",
        duration_ms=thought_duration,
    ))
    
    # Step 3: Tool call
    steps.append(AgentStep(
        type=StepType.TOOL_CALL,
        content=f"Calling **{tool_name}** with input: `{tool_input[:100]}`",
        tool_name=tool_name,
        tool_input=tool_input[:100],
    ))
    
    # Step 4: Execute tool
    tool_start = time.time()
    try:
        result = tool_info["fn"](tool_input)
    except Exception as e:
        result = f"❌ Error executing {tool_name}: {e}"
    tool_duration = (time.time() - tool_start) * 1000
    
    steps.append(AgentStep(
        type=StepType.TOOL_RESULT,
        content=f"**{tool_name}** completed in {tool_duration:.0f}ms",
        tool_name=tool_name,
        tool_output=str(result)[:500],
        duration_ms=tool_duration,
    ))
    
    # Step 5: Final answer
    is_error = result.startswith("❌")
    if is_error:
        final = f"⚠️ The **{tool_name}** tool encountered an issue:\n\n```\n{result}\n```\n\n**Recovery:** Double-check your input and try again."
    else:
        final = f"Here's what I found using **{tool_name}**:\n\n{result}"
    
    steps.append(AgentStep(
        type=StepType.FINAL,
        content=final,
        duration_ms=(time.time() - t_start) * 1000,
    ))
    
    return steps


def format_steps_as_html(steps: List[AgentStep]) -> str:
    """Format agent steps as nice HTML for Gradio."""
    html = ""
    colors = {
        StepType.THOUGHT: ("#f0f4f8", "#2c3e50", "🧠"),
        StepType.TOOL_CALL: ("#fff3e0", "#e65100", "🔧"),
        StepType.TOOL_RESULT: ("#e8f5e9", "#1b5e20", "📥"),
        StepType.FINAL: ("#e3f2fd", "#0d47a1", "💬"),
    }
    
    for i, step in enumerate(steps):
        bg, color, icon = colors[step.type]
        label = step.type.value.replace("_", " ").title()
        
        html += f"""
        <div style="background:{bg}; border-left:4px solid {color}; border-radius:8px; 
                    padding:14px 18px; margin:10px 0; font-family:'Segoe UI',system-ui,sans-serif;">
            <div style="display:flex; align-items:center; gap:8px; margin-bottom:6px;">
                <span style="font-size:18px;">{icon}</span>
                <strong style="color:{color}; font-size:14px;">Step {i+1}: {label}</strong>
                {f'<span style="margin-left:auto; color:#999; font-size:12px;">{step.duration_ms:.0f}ms</span>' if step.duration_ms > 0 else ''}
            </div>
            <div style="color:#333; font-size:14px; line-height:1.6;">
                {step.content}
            </div>
        """
        
        if step.tool_name and step.tool_input:
            html += f"""
            <div style="background:rgba(0,0,0,0.04); border-radius:4px; padding:8px 12px; margin-top:8px; font-family:monospace; font-size:13px;">
                <span style="color:#666;">Tool:</span> {step.tool_name} | 
                <span style="color:#666;">Input:</span> {step.tool_input}
            </div>
            """
        
        if step.tool_output:
            html += f"""
            <div style="background:#1e1e1e; color:#d4d4d4; border-radius:4px; padding:10px 14px; margin-top:8px; font-family:monospace; font-size:13px; white-space:pre-wrap; max-height:200px; overflow-y:auto;">
                {step.tool_output[:500]}
            </div>
            """
        
        html += "</div>"
    
    return html


# ============================================================
# Gradio UI
# ============================================================

def respond(message: str, history: list):
    """Process a user message and return the response."""
    if not message.strip():
        return "", history
    
    steps = run_agent(message)
    html_output = format_steps_as_html(steps)
    
    # Add to history (type="messages" format)
    history.append({"role": "user", "content": message})
    history.append({"role": "assistant", "content": html_output})
    return "", history


CSS = """
.gradio-container { max-width: 900px !important; margin: auto !important; }
.chatbot .user { background: #e3f2fd !important; }
.chatbot .assistant { background: transparent !important; }
footer { display: none !important; }
"""

with gr.Blocks(
    css=CSS,
    theme=gr.themes.Soft(
        primary_hue="blue",
        secondary_hue="indigo",
        neutral_hue="slate",
        font=gr.themes.GoogleFont("Inter"),
    ),
    title="Tool-Calling AI Agent",
) as demo:
    gr.Markdown(
        """
        # 🛠️ Tool-Calling AI Agent
        
        **Built with LangGraph architecture** — watch the agent think, call tools, and respond.
        
        The agent follows a structured pipeline: **Thought → Tool Call → Tool Result → Final Answer**.
        
        ### Available Tools:
        | Tool | Description | Example |
        |------|-------------|--------|
        | 🔢 Calculator | Safe math evaluation | `25 * 4 + 10` |
        | 🌐 Web Search | Knowledge base lookup | `Tell me about LangGraph` |
        | 📁 File Reader | Read text files | `Read README.md` |
        """,
    )
    
    chatbot = gr.Chatbot(
        label="Agent Conversation",
        height=600,
        show_label=False,
        bubble_full_width=False,
        avatar_images=(None, "🧠"),
        type="messages",
    )
    
    with gr.Row():
        msg = gr.Textbox(
            placeholder="Ask me anything... (e.g., 'What is 2+2?', 'Tell me about LangGraph', 'Read README.md')",
            show_label=False,
            container=False,
            scale=8,
        )
        send = gr.Button("Send", variant="primary", scale=1)
        clear = gr.ClearButton([msg, chatbot], scale=1)
    
    examples = gr.Examples(
        examples=[
            ["What is 2 + 2?"],
            ["Calculate the area of a circle with radius 7"],
            ["Tell me about LangGraph"],
            ["What is LangChain?"],
            ["Read app.py"],
            ["Calculate sqrt(144) + 50 * 3"],
            ["What is Hugging Face?"],
        ],
        inputs=[msg],
        label="Try these examples",
    )
    
    # Bind events
    msg.submit(respond, [msg, chatbot], [msg, chatbot])
    send.click(respond, [msg, chatbot], [msg, chatbot])

if __name__ == "__main__":
    demo.launch()