Update app.py

app.py

@@ -8,8 +8,8 @@ from datetime import datetime
 import ast
 import operator as op
 import wikipedia
+from transformers import pipeline
 import torch
-from transformers import AutoTokenizer, AutoModelForCausalLM
 
 class Tool:
     def __init__(self, name: str, description: str, func):
@@ -21,73 +21,45 @@ class Tool:
         return self.func(*args, **kwargs)
 
 def duckduckgo_search(query: str) -> str:
-    """Search DuckDuckGo for information."""
     try:
         url = "https://api.duckduckgo.com/"
-        params = {
-            'q': query,
-            'format': 'json',
-            'no_html': 1,
-            'skip_disambig': 1
-        }
+        params = {'q': query, 'format': 'json', 'no_html': 1, 'skip_disambig': 1}
         response = requests.get(url, params=params, timeout=10)
         data = response.json()
         
         if data.get('Abstract'):
             return f"Search result: {data['Abstract']}"
         elif data.get('RelatedTopics') and len(data['RelatedTopics']) > 0:
-            results = []
-            for topic in data['RelatedTopics'][:3]:
-                if 'Text' in topic:
-                    results.append(topic['Text'])
+            results = [topic['Text'] for topic in data['RelatedTopics'][:3] if 'Text' in topic]
             return f"Search results: {' | '.join(results)}" if results else "No results found."
-        else:
-            return "No results found."
+        return "No results found."
     except Exception as e:
         return f"Search error: {str(e)}"
 
 def wikipedia_search(query: str) -> str:
-    """Search Wikipedia for information."""
     try:
         wikipedia.set_lang("en")
         summary = wikipedia.summary(query, sentences=3, auto_suggest=True)
         return f"Wikipedia: {summary}"
     except wikipedia.exceptions.DisambiguationError as e:
-        return f"Wikipedia: Multiple results found. Please be more specific. Options: {', '.join(e.options[:5])}"
+        return f"Wikipedia: Multiple results found. Options: {', '.join(e.options[:5])}"
     except wikipedia.exceptions.PageError:
         return f"Wikipedia: No page found for '{query}'."
     except Exception as e:
         return f"Wikipedia error: {str(e)}"
 
 def get_weather(location: str) -> str:
-    """Get current weather for a location using wttr.in."""
     try:
         url = f"https://wttr.in/{location}?format=j1"
         response = requests.get(url, timeout=10)
         data = response.json()
-        
         current = data['current_condition'][0]
-        temp_c = current['temp_C']
-        temp_f = current['temp_F']
-        desc = current['weatherDesc'][0]['value']
-        humidity = current['humidity']
-        wind_speed = current['windspeedKmph']
-        
-        return f"Weather in {location}: {desc}, {temp_c}°C ({temp_f}°F), Humidity: {humidity}%, Wind: {wind_speed} km/h"
+        return f"Weather in {location}: {current['weatherDesc'][0]['value']}, {current['temp_C']}°C, Humidity: {current['humidity']}%"
     except Exception as e:
         return f"Weather error: {str(e)}"
 
 def calculate(expression: str) -> str:
-    """Safely evaluate mathematical expressions."""
-    operators = {
-        ast.Add: op.add,
-        ast.Sub: op.sub,
-        ast.Mult: op.mul,
-        ast.Div: op.truediv,
-        ast.Pow: op.pow,
-        ast.USub: op.neg,
-        ast.Mod: op.mod,
-    }
+    operators = {ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv, ast.Pow: op.pow, ast.USub: op.neg, ast.Mod: op.mod}
     
     def eval_expr(node):
         if isinstance(node, ast.Num):
@@ -96,350 +68,212 @@ def calculate(expression: str) -> str:
             return operators[type(node.op)](eval_expr(node.left), eval_expr(node.right))
         elif isinstance(node, ast.UnaryOp):
             return operators[type(node.op)](eval_expr(node.operand))
-        elif isinstance(node, ast.Call):
-            if node.func.id == 'abs':
-                return abs(eval_expr(node.args[0]))
-            elif node.func.id == 'round':
-                return round(eval_expr(node.args[0]))
-        else:
-            raise TypeError(node)
+        raise TypeError(node)
     
     try:
-        expression = expression.strip()
-        node = ast.parse(expression, mode='eval')
-        result = eval_expr(node.body)
+        result = eval_expr(ast.parse(expression.strip(), mode='eval').body)
         return f"Result: {result}"
     except Exception as e:
-        return f"Calculation error: {str(e)}. Please use basic arithmetic operators (+, -, *, /, **, %)."
+        return f"Calculation error: {str(e)}"
 
 def python_repl(code: str) -> str:
-    """Execute safe Python code (limited to basic operations)."""
     try:
-        safe_builtins = {
-            'abs': abs, 'round': round, 'min': min, 'max': max,
-            'sum': sum, 'len': len, 'range': range, 'list': list,
-            'dict': dict, 'str': str, 'int': int, 'float': float,
-            'print': print, 'enumerate': enumerate, 'zip': zip,
-            'sorted': sorted, 'reversed': reversed,
-        }
-        
+        safe_builtins = {'abs': abs, 'round': round, 'min': min, 'max': max, 'sum': sum, 'len': len, 'range': range, 'list': list, 'dict': dict, 'str': str, 'int': int, 'float': float, 'print': print}
         namespace = {'__builtins__': safe_builtins}
         
         from io import StringIO
         import sys
         old_stdout = sys.stdout
         sys.stdout = StringIO()
-        
         exec(code, namespace)
-        
         output = sys.stdout.getvalue()
         sys.stdout = old_stdout
         
         result_vars = {k: v for k, v in namespace.items() if k != '__builtins__' and not k.startswith('_')}
-        
-        result = output if output else str(result_vars) if result_vars else "Code executed successfully (no output)"
-        return f"Python output: {result}"
+        return f"Python output: {output if output else (str(result_vars) if result_vars else 'Code executed')}"
     except Exception as e:
         return f"Python error: {str(e)}"
 
 TOOLS = [
-    Tool(
-        name="duckduckgo_search",
-        description="Search the web using DuckDuckGo. Use this when you need current information or facts. Input should be a search query string.",
-        func=duckduckgo_search
-    ),
-    Tool(
-        name="wikipedia_search",
-        description="Search Wikipedia for detailed information about topics, people, places, etc. Input should be a search query string.",
-        func=wikipedia_search
-    ),
-    Tool(
-        name="get_weather",
-        description="Get current weather information for a location. Input should be a city name or location string.",
-        func=get_weather
-    ),
-    Tool(
-        name="calculate",
-        description="Perform mathematical calculations. Input should be a mathematical expression like '5 + 3 * 2' or '2 ** 10'.",
-        func=calculate
-    ),
-    Tool(
-        name="python_repl",
-        description="Execute Python code for data processing or calculations. Input should be valid Python code. Only basic operations are allowed.",
-        func=python_repl
-    ),
+    Tool("duckduckgo_search", "Search the web. Input: search query.", duckduckgo_search),
+    Tool("wikipedia_search", "Search Wikipedia. Input: search query.", wikipedia_search),
+    Tool("get_weather", "Get weather for location. Input: city name.", get_weather),
+    Tool("calculate", "Calculate math expression. Input: expression.", calculate),
+    Tool("python_repl", "Execute Python code. Input: code.", python_repl),
 ]
 
 MODEL_NAME = "openai/gpt-oss-20b"
-model = None
-tokenizer = None
+pipe = None
 model_loaded = False
 
 def download_and_load_model(progress=gr.Progress()):
-    """Download and load the model."""
-    global model, tokenizer, model_loaded
+    global pipe, model_loaded
     
     try:
-        progress(0, desc="Starting model download...")
+        progress(0, desc="Downloading model...")
+        progress(0.5, desc="Loading model (this may take several minutes)...")
         
-        progress(0.3, desc="Downloading tokenizer...")
-        tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
-        
-        progress(0.5, desc="Downloading model (this may take several minutes)...")
-        model = AutoModelForCausalLM.from_pretrained(
-            MODEL_NAME,
-            torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
-            device_map="auto" if torch.cuda.is_available() else None,
-            low_cpu_mem_usage=True
+        pipe = pipeline(
+            "text-generation",
+            model=MODEL_NAME,
+            torch_dtype="auto",
+            device_map="auto",
         )
         
         progress(0.95, desc="Finalizing...")
         model_loaded = True
-        
-        progress(1.0, desc="Model loaded successfully!")
+        progress(1.0, desc="Model loaded!")
         return f"Model '{MODEL_NAME}' loaded successfully!"
-    
     except Exception as e:
-        return f"Error loading model: {str(e)}"
+        return f"Error: {str(e)}"
 
 def get_tool_descriptions() -> str:
-    descriptions = []
-    for tool in TOOLS:
-        descriptions.append(f"- {tool.name}: {tool.description}")
-    return "\n".join(descriptions)
-
-THINK_ONLY_PROMPT = """You are a helpful AI assistant. You solve problems by thinking through them step-by-step.
+    return "\n".join([f"- {tool.name}: {tool.description}" for tool in TOOLS])
 
-For each question:
-1. Think through the problem carefully in your internal monologue
-2. Show your reasoning process using "Thought: ..." format
-3. Provide a final answer using "Answer: ..." format
+THINK_ONLY_PROMPT = """You are a helpful AI assistant. Solve problems step-by-step.
 
-You do NOT have access to any tools. Rely only on your knowledge and reasoning.
+Format:
+Thought: your reasoning
+Answer: your final answer
 
 Question: {question}
 
-Let's think step by step:"""
+Think step by step:"""
 
-ACT_ONLY_PROMPT = """You are a helpful AI assistant with access to tools. You solve problems by using tools.
+ACT_ONLY_PROMPT = """You are a helpful AI assistant with tools.
 
 Available tools:
 {tools}
 
-For each question, you must use tools to find information. Do NOT think or reason - just use tools.
-
-Format your response as:
+Format:
 Action: tool_name
-Action Input: input_for_tool
-
-After receiving the observation, you can call another tool or provide the final answer:
-Answer: your final answer
+Action Input: input
 
 Question: {question}
 
 Action:"""
 
-REACT_PROMPT = """You are a helpful AI assistant that can think and use tools. You solve problems by alternating between Thought, Action, and Observation.
+REACT_PROMPT = """You are a helpful AI assistant with tools.
 
 Available tools:
 {tools}
 
-For each question, follow this pattern:
-Thought: Think about what you need to do next
+Pattern:
+Thought: what to do next
 Action: tool_name
-Action Input: input_for_tool
-Observation: [tool result will be provided]
-... (repeat Thought/Action/Observation as needed)
-Thought: I now know the final answer
-Answer: your final answer
+Action Input: input
+Observation: [result]
+... repeat as needed
+Thought: I know the answer
+Answer: final answer
 
 Question: {question}
 
 Thought:"""
 
 def parse_action(text: str) -> tuple:
-    """Parse action and action input from model output."""
-    action_pattern = r'Action:\s*(\w+)'
-    input_pattern = r'Action Input:\s*(.+?)(?=\n(?:Thought:|Action:|Answer:|$))'
-    
-    action_match = re.search(action_pattern, text, re.IGNORECASE)
-    input_match = re.search(input_pattern, text, re.IGNORECASE | re.DOTALL)
-    
-    if action_match and input_match:
-        action_name = action_match.group(1).strip()
-        action_input = input_match.group(1).strip()
-        return action_name, action_input
-    return None, None
+    action_match = re.search(r'Action:\s*(\w+)', text, re.IGNORECASE)
+    input_match = re.search(r'Action Input:\s*(.+?)(?=\n(?:Thought:|Action:|Answer:|$))', text, re.IGNORECASE | re.DOTALL)
+    return (action_match.group(1).strip(), input_match.group(1).strip()) if action_match and input_match else (None, None)
 
 def call_tool(tool_name: str, tool_input: str) -> str:
-    """Call a tool by name."""
     for tool in TOOLS:
         if tool.name.lower() == tool_name.lower():
             return tool(tool_input)
-    return f"Error: Tool '{tool_name}' not found. Available tools: {', '.join([t.name for t in TOOLS])}"
+    return f"Error: Tool '{tool_name}' not found."
 
-def call_llm(prompt: str, temperature: float = 0.7, max_tokens: int = 500) -> str:
-    """Call the local LLM."""
-    global model, tokenizer, model_loaded
-    
+def call_llm(prompt: str, max_tokens: int = 500) -> str:
     if not model_loaded:
-        return "Error: Model not loaded. Please click 'Download & Load Model' first."
+        return "Error: Model not loaded."
     
     try:
         messages = [{"role": "user", "content": prompt}]
-        
-        inputs = tokenizer.apply_chat_template(
-            messages,
-            add_generation_prompt=True,
-            tokenize=True,
-            return_dict=True,
-            return_tensors="pt",
-        ).to(model.device)
-        
-        with torch.no_grad():
-            outputs = model.generate(
-                **inputs,
-                max_new_tokens=max_tokens,
-                temperature=temperature,
-                do_sample=True,
-                top_p=0.9
-            )
-        
-        response = tokenizer.decode(outputs[0][inputs["input_ids"].shape[-1]:], skip_special_tokens=True)
-        return response.strip()
-    
+        outputs = pipe(messages, max_new_tokens=max_tokens)
+        return outputs[0]["generated_text"][-1]["content"]
     except Exception as e:
-        return f"Error calling model: {str(e)}"
+        return f"Error: {str(e)}"
 
 def think_only_mode(question: str) -> str:
-    """Think-Only mode: Chain-of-Thought only, no tools."""
     if not model_loaded:
-        return "Error: Model not loaded. Please click 'Download & Load Model' first."
+        return "Error: Model not loaded."
     
-    prompt = THINK_ONLY_PROMPT.format(question=question)
+    output = "**Mode: Think-Only**\n\n"
+    response = call_llm(THINK_ONLY_PROMPT.format(question=question), max_tokens=800)
     
-    output = "**Mode: Think-Only (Chain-of-Thought)**\n\n"
-    output += "Generating thoughts...\n\n"
-    
-    response = call_llm(prompt, temperature=0.7, max_tokens=800)
-    
-    lines = response.split('\n')
-    for line in lines:
+    for line in response.split('\n'):
         if line.strip():
-            if line.strip().startswith('Thought:'):
-                output += f"**{line.strip()}**\n\n"
-            elif line.strip().startswith('Answer:'):
-                output += f"**{line.strip()}**\n\n"
-            else:
-                output += f"{line}\n\n"
+            output += f"**{line.strip()}**\n\n" if line.strip().startswith(('Thought:', 'Answer:')) else f"{line}\n\n"
     
-    output += "\n---\n**Mode completed**\n"
-    return output
+    return output + "\n---\n"
 
 def act_only_mode(question: str, max_iterations: int = 5) -> str:
-    """Act-Only mode: Tool use only, no explicit thinking."""
     if not model_loaded:
-        return "Error: Model not loaded. Please click 'Download & Load Model' first."
-    
-    tool_descriptions = get_tool_descriptions()
-    prompt = ACT_ONLY_PROMPT.format(question=question, tools=tool_descriptions)
+        return "Error: Model not loaded."
     
-    output = "**Mode: Act-Only (Tool Use Only)**\n\n"
+    output = "**Mode: Act-Only**\n\n"
+    conversation = ACT_ONLY_PROMPT.format(question=question, tools=get_tool_descriptions())
     
-    iteration = 0
-    conversation_history = prompt
-    
-    while iteration < max_iterations:
-        iteration += 1
-        
-        response = call_llm(conversation_history, temperature=0.5, max_tokens=300)
+    for _ in range(max_iterations):
+        response = call_llm(conversation, max_tokens=300)
         
         if 'Answer:' in response:
-            answer_match = re.search(r'Answer:\s*(.+)', response, re.IGNORECASE | re.DOTALL)
-            if answer_match:
-                output += f"**Answer:** {answer_match.group(1).strip()}\n\n"
+            match = re.search(r'Answer:\s*(.+)', response, re.IGNORECASE | re.DOTALL)
+            if match:
+                output += f"**Answer:** {match.group(1).strip()}\n\n"
                 break
         
         action_name, action_input = parse_action(response)
-        
         if action_name and action_input:
-            output += f"**Action:** {action_name}\n"
-            output += f"**Action Input:** {action_input}\n\n"
-            
+            output += f"**Action:** {action_name}\n**Input:** {action_input}\n\n"
             observation = call_tool(action_name, action_input)
             output += f"**Observation:** {observation}\n\n"
-            
-            conversation_history += f"\n{response}\nObservation: {observation}\n\nContinue with another action or provide the final answer.\n"
+            conversation += f"\n{response}\nObservation: {observation}\n\n"
         else:
-            output += f"Could not parse action from response. Response: {response}\n\n"
+            output += f"Could not parse action.\n\n"
             break
     
-    if iteration >= max_iterations:
-        output += "**Reached maximum iterations.**\n\n"
-    
-    output += "\n---\n**Mode completed**\n"
-    return output
+    return output + "\n---\n"
 
 def react_mode(question: str, max_iterations: int = 5) -> str:
-    """ReAct mode: Interleaving Thought, Action, Observation."""
     if not model_loaded:
-        return "Error: Model not loaded. Please click 'Download & Load Model' first."
-    
-    tool_descriptions = get_tool_descriptions()
-    prompt = REACT_PROMPT.format(question=question, tools=tool_descriptions)
+        return "Error: Model not loaded."
     
-    output = "**Mode: ReAct (Thought + Action + Observation)**\n\n"
+    output = "**Mode: ReAct**\n\n"
+    conversation = REACT_PROMPT.format(question=question, tools=get_tool_descriptions())
     
-    iteration = 0
-    conversation_history = prompt
-    
-    while iteration < max_iterations:
-        iteration += 1
-        
-        response = call_llm(conversation_history, temperature=0.7, max_tokens=400)
+    for _ in range(max_iterations):
+        response = call_llm(conversation, max_tokens=400)
         
-        thought_matches = re.findall(r'Thought:\s*(.+?)(?=\n(?:Action:|Answer:|$))', response, re.IGNORECASE | re.DOTALL)
-        for thought in thought_matches:
+        for thought in re.findall(r'Thought:\s*(.+?)(?=\n(?:Action:|Answer:|$))', response, re.IGNORECASE | re.DOTALL):
             output += f"**Thought:** {thought.strip()}\n\n"
         
         if 'Answer:' in response:
-            answer_match = re.search(r'Answer:\s*(.+)', response, re.IGNORECASE | re.DOTALL)
-            if answer_match:
-                output += f"**Answer:** {answer_match.group(1).strip()}\n\n"
+            match = re.search(r'Answer:\s*(.+)', response, re.IGNORECASE | re.DOTALL)
+            if match:
+                output += f"**Answer:** {match.group(1).strip()}\n\n"
                 break
         
         action_name, action_input = parse_action(response)
-        
         if action_name and action_input:
-            output += f"**Action:** {action_name}\n"
-            output += f"**Action Input:** {action_input}\n\n"
-            
+            output += f"**Action:** {action_name}\n**Input:** {action_input}\n\n"
             observation = call_tool(action_name, action_input)
             output += f"**Observation:** {observation}\n\n"
-            
-            conversation_history += f"\n{response}\nObservation: {observation}\n\nThought:"
+            conversation += f"\n{response}\nObservation: {observation}\n\nThought:"
         else:
             if 'Answer:' not in response:
-                output += f"No action found. Response: {response}\n\n"
+                output += "No action found.\n\n"
             break
     
-    if iteration >= max_iterations:
-        output += "**Reached maximum iterations.**\n\n"
-    
-    output += "\n---\n**Mode completed**\n"
-    return output
+    return output + "\n---\n"
 
 EXAMPLES = [
-    "What is the capital of France and what's the current weather there?",
-    "Who wrote 'To Kill a Mockingbird' and when was it published?",
-    "Calculate the compound interest on $1000 at 5% annual rate for 3 years using the formula A = P(1 + r)^t",
-    "What is the population of Tokyo and how does it compare to New York City?",
-    "If I have a list of numbers [15, 23, 8, 42, 16], what is the average and which number is closest to it?",
-    "What are the main causes of climate change according to scientific consensus?",
+    "What is the capital of France and its weather?",
+    "Who wrote 'To Kill a Mockingbird'?",
+    "Calculate: 1000 * (1.05 ** 3)",
+    "What is Tokyo's population?",
 ]
 
 def run_comparison(question: str, mode: str):
-    """Run the selected mode(s)."""
     if mode == "Think-Only":
         return think_only_mode(question), "", ""
     elif mode == "Act-Only":
@@ -448,57 +282,34 @@ def run_comparison(question: str, mode: str):
         return "", "", react_mode(question)
     elif mode == "All (Compare)":
         return think_only_mode(question), act_only_mode(question), react_mode(question)
-    else:
-        return "Invalid mode selected.", "", ""
+    return "Invalid mode.", "", ""
 
-with gr.Blocks(title="LLM Reasoning Modes Comparison") as demo:
+with gr.Blocks(title="LLM Reasoning Modes") as demo:
+    gr.Markdown("# LLM Reasoning Modes Comparison\n\n**Model:** openai/gpt-oss-20b\n\n**Tools:** DuckDuckGo | Wikipedia | Weather | Calculator | Python")
     
     with gr.Row():
         download_btn = gr.Button("Download & Load Model", variant="primary", size="lg")
-        model_status = gr.Textbox(label="Model Status", value="Model not loaded. Click to download openai/gpt-oss-20b", interactive=False)
+        model_status = gr.Textbox(label="Status", value="Click to download", interactive=False)
     
     with gr.Row():
         with gr.Column(scale=3):
-            question_input = gr.Textbox(
-                label="Enter your question",
-                placeholder="Ask a question that might require tools or reasoning...",
-                lines=3
-            )
-            mode_dropdown = gr.Dropdown(
-                choices=["Think-Only", "Act-Only", "ReAct", "All (Compare)"],
-                value="All (Compare)",
-                label="Select Mode"
-            )
+            question_input = gr.Textbox(label="Question", lines=3)
+            mode_dropdown = gr.Dropdown(choices=["Think-Only", "Act-Only", "ReAct", "All (Compare)"], value="All (Compare)", label="Mode")
             submit_btn = gr.Button("Run", variant="primary", size="lg")
-        
         with gr.Column(scale=1):
-            gr.Markdown("**Example Questions**")
-            for idx, example in enumerate(EXAMPLES):
-                gr.Button(f"Example {idx+1}", size="sm").click(
-                    fn=lambda ex=example: ex,
-                    outputs=question_input
-                )
+            gr.Markdown("**Examples**")
+            for idx, ex in enumerate(EXAMPLES):
+                gr.Button(f"Ex {idx+1}", size="sm").click(fn=lambda e=ex: e, outputs=question_input)
     
     gr.Markdown("---")
     
     with gr.Row():
-        with gr.Column():
-            think_output = gr.Markdown(label="Think-Only Output")
-        with gr.Column():
-            act_output = gr.Markdown(label="Act-Only Output")
-        with gr.Column():
-            react_output = gr.Markdown(label="ReAct Output")
-    
-    download_btn.click(
-        fn=download_and_load_model,
-        outputs=model_status
-    )
+        think_output = gr.Markdown(label="Think-Only")
+        act_output = gr.Markdown(label="Act-Only")
+        react_output = gr.Markdown(label="ReAct")
     
-    submit_btn.click(
-        fn=run_comparison,
-        inputs=[question_input, mode_dropdown],
-        outputs=[think_output, act_output, react_output]
-    )
+    download_btn.click(fn=download_and_load_model, outputs=model_status)
+    submit_btn.click(fn=run_comparison, inputs=[question_input, mode_dropdown], outputs=[think_output, act_output, react_output])
 
 if __name__ == "__main__":
     demo.launch(share=True)