| import os |
| import re |
| import json |
| import gradio as gr |
| from typing import List, Dict, Any, Generator |
| import requests |
| from datetime import datetime |
| import ast |
| import operator as op |
| import wikipedia |
| from transformers import AutoTokenizer, AutoModelForCausalLM |
| import torch |
|
|
| class Tool: |
| def __init__(self, name: str, description: str, func): |
| self.name = name |
| self.description = description |
| self.func = func |
| |
| def __call__(self, *args, **kwargs): |
| return self.func(*args, **kwargs) |
|
|
| def duckduckgo_search(query: str) -> str: |
| try: |
| url = "https://api.duckduckgo.com/" |
| 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 = [topic['Text'] for topic in data['RelatedTopics'][:3] if 'Text' in topic] |
| return f"Search results: {' | '.join(results)}" if results else "No results found." |
| return "No results found." |
| except Exception as e: |
| return f"Search error: {str(e)}" |
|
|
| def wikipedia_search(query: str) -> str: |
| 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. 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: |
| try: |
| url = f"https://wttr.in/{location}?format=j1" |
| response = requests.get(url, timeout=10) |
| data = response.json() |
| current = data['current_condition'][0] |
| 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: |
| 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): |
| return node.n |
| elif isinstance(node, ast.BinOp): |
| 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)) |
| raise TypeError(node) |
| |
| try: |
| result = eval_expr(ast.parse(expression.strip(), mode='eval').body) |
| return f"Result: {result}" |
| except Exception as e: |
| return f"Calculation error: {str(e)}" |
|
|
| def python_repl(code: str) -> str: |
| 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} |
| 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('_')} |
| 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("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 |
| model_loaded = False |
|
|
| def download_and_load_model(progress=gr.Progress()): |
| global model, tokenizer, model_loaded |
| |
| try: |
| progress(0, desc="Downloading tokenizer...") |
| tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=True) |
| |
| progress(0.4, desc="Downloading model (this may take several minutes)...") |
| model = AutoModelForCausalLM.from_pretrained( |
| MODEL_NAME, |
| trust_remote_code=True, |
| torch_dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32, |
| device_map="auto", |
| low_cpu_mem_usage=True, |
| ) |
| |
| progress(0.95, desc="Finalizing...") |
| model_loaded = True |
| progress(1.0, desc="Model loaded!") |
| return f"Model '{MODEL_NAME}' loaded successfully!" |
| except Exception as e: |
| return f"Error: {str(e)}" |
|
|
| def get_tool_descriptions() -> str: |
| return "\n".join([f"- {tool.name}: {tool.description}" for tool in TOOLS]) |
|
|
| THINK_ONLY_PROMPT = """You are an expert problem solver. Use your knowledge and reasoning to answer questions. |
| |
| You must show your complete reasoning process using this format: |
| Thought: [Explain what you're thinking and why] |
| Thought: [Continue your reasoning, breaking down the problem] |
| Thought: [Build toward the solution step by step] |
| Answer: [Your final, complete answer] |
| |
| Important: |
| - Show multiple thought steps |
| - Break down complex problems |
| - Explain your reasoning clearly |
| - Only provide the Answer when you're certain |
| |
| Question: {question} |
| |
| Let me think through this step by step: |
| |
| Thought:""" |
|
|
| ACT_ONLY_PROMPT = """You are an AI agent with access to external tools. You MUST use tools to find information. |
| |
| Available tools: |
| {tools} |
| |
| You MUST respond ONLY with actions - no thinking out loud: |
| Action: [exact tool name] |
| Action Input: [specific input for the tool] |
| |
| After receiving the Observation, you can: |
| - Call another tool if you need more information |
| - Provide the final Answer when you have enough information |
| |
| Format: |
| Action: tool_name |
| Action Input: input_string |
| |
| Then after observation: |
| Action: another_tool |
| Action Input: another_input |
| |
| OR |
| Answer: [final answer based on observations] |
| |
| Question: {question} |
| |
| Action:""" |
|
|
| REACT_PROMPT = """You are an expert AI agent that combines reasoning with tool usage (ReAct paradigm). |
| |
| Available tools: |
| {tools} |
| |
| You MUST alternate between thinking and acting: |
| |
| 1. Thought: [Reason about what information you need and which tool to use] |
| 2. Action: [exact tool name] |
| 3. Action Input: [specific input] |
| 4. Observation: [tool result - will be provided to you] |
| 5. Thought: [Analyze the observation and decide next steps] |
| 6. Repeat 2-5 until you have enough information |
| 7. Thought: [Final reasoning with all gathered information] |
| 8. Answer: [Complete final answer] |
| |
| Rules: |
| - ALWAYS start with a Thought explaining your strategy |
| - After each Observation, think about what you learned |
| - Use multiple tools if needed |
| - Only give Answer when you have sufficient information |
| - Be specific in your Action Inputs |
| |
| Question: {question} |
| |
| Thought:""" |
|
|
| def parse_action(text: str) -> tuple: |
| 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: |
| for tool in TOOLS: |
| if tool.name.lower() == tool_name.lower(): |
| return tool(tool_input) |
| return f"Error: Tool '{tool_name}' not found." |
|
|
| def call_llm(prompt: str, max_tokens: int = 500) -> str: |
| if not model_loaded: |
| return "Error: Model not loaded." |
| |
| try: |
| inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=2048) |
| |
| if torch.cuda.is_available(): |
| inputs = {k: v.to(model.device) for k, v in inputs.items()} |
| |
| with torch.no_grad(): |
| outputs = model.generate( |
| **inputs, |
| max_new_tokens=max_tokens, |
| temperature=0.7, |
| do_sample=True, |
| top_p=0.9, |
| pad_token_id=tokenizer.eos_token_id, |
| eos_token_id=tokenizer.eos_token_id, |
| ) |
| |
| response = tokenizer.decode(outputs[0][inputs['input_ids'].shape[1]:], skip_special_tokens=True) |
| return response.strip() |
| except Exception as e: |
| return f"Error during generation: {str(e)}" |
|
|
| def think_only_mode(question: str) -> Generator[str, None, None]: |
| if not model_loaded: |
| yield "❌ **Error: Model not loaded. Click 'Download & Load Model' first.**\n\n" |
| return |
| |
| yield "🧠 **Mode: Think-Only (Chain-of-Thought)**\n\n" |
| yield "💭 Generating reasoning steps...\n\n" |
| |
| response = call_llm(THINK_ONLY_PROMPT.format(question=question), max_tokens=800) |
| |
| if response.startswith("Error"): |
| yield f"❌ {response}\n\n" |
| return |
| |
| for line in response.split('\n'): |
| if line.strip(): |
| if line.strip().startswith('Thought:'): |
| yield f"💭 **{line.strip()}**\n\n" |
| elif line.strip().startswith('Answer:'): |
| yield f"✅ **{line.strip()}**\n\n" |
| else: |
| yield f"{line}\n\n" |
| |
| yield "\n---\n✓ **Completed**\n" |
|
|
| def act_only_mode(question: str, max_iterations: int = 5) -> Generator[str, None, None]: |
| if not model_loaded: |
| yield "❌ **Error: Model not loaded. Click 'Download & Load Model' first.**\n\n" |
| return |
| |
| yield "🔧 **Mode: Act-Only (Tool Use Only)**\n\n" |
| conversation = ACT_ONLY_PROMPT.format(question=question, tools=get_tool_descriptions()) |
| |
| for iteration in range(max_iterations): |
| yield f"🔄 **Iteration {iteration + 1}**\n\n" |
| |
| response = call_llm(conversation, max_tokens=300) |
| |
| if response.startswith("Error"): |
| yield f"❌ {response}\n\n" |
| return |
| |
| if 'Answer:' in response: |
| match = re.search(r'Answer:\s*(.+)', response, re.IGNORECASE | re.DOTALL) |
| if match: |
| yield f"✅ **Answer:** {match.group(1).strip()}\n\n" |
| break |
| |
| action_name, action_input = parse_action(response) |
| if action_name and action_input: |
| yield f"🔧 **Action:** `{action_name}`\n" |
| yield f"📝 **Action Input:** {action_input}\n\n" |
| |
| yield f"⏳ Executing tool...\n\n" |
| observation = call_tool(action_name, action_input) |
| |
| yield f"👁️ **Observation:** {observation}\n\n" |
| conversation += f"\n{response}\nObservation: {observation}\n\n" |
| else: |
| yield f"⚠️ No valid action found. Response: {response}\n\n" |
| break |
| |
| yield "\n---\n✓ **Completed**\n" |
|
|
| def react_mode(question: str, max_iterations: int = 5) -> Generator[str, None, None]: |
| if not model_loaded: |
| yield "❌ **Error: Model not loaded. Click 'Download & Load Model' first.**\n\n" |
| return |
| |
| yield "🤖 **Mode: ReAct (Reasoning + Acting)**\n\n" |
| conversation = REACT_PROMPT.format(question=question, tools=get_tool_descriptions()) |
| |
| for iteration in range(max_iterations): |
| yield f"🔄 **Step {iteration + 1}**\n\n" |
| |
| response = call_llm(conversation, max_tokens=400) |
| |
| if response.startswith("Error"): |
| yield f"❌ {response}\n\n" |
| return |
| |
| |
| for thought in re.findall(r'Thought:\s*(.+?)(?=\n(?:Action:|Answer:|$))', response, re.IGNORECASE | re.DOTALL): |
| yield f"💭 **Thought:** {thought.strip()}\n\n" |
| |
| |
| if 'Answer:' in response: |
| match = re.search(r'Answer:\s*(.+)', response, re.IGNORECASE | re.DOTALL) |
| if match: |
| yield f"✅ **Answer:** {match.group(1).strip()}\n\n" |
| break |
| |
| |
| action_name, action_input = parse_action(response) |
| if action_name and action_input: |
| yield f"🔧 **Action:** `{action_name}`\n" |
| yield f"📝 **Action Input:** {action_input}\n\n" |
| |
| yield f"⏳ Executing tool...\n\n" |
| observation = call_tool(action_name, action_input) |
| |
| yield f"👁️ **Observation:** {observation}\n\n" |
| conversation += f"\n{response}\nObservation: {observation}\n\nThought:" |
| else: |
| if 'Answer:' not in response: |
| yield f"⚠️ No action found. Response: {response}\n\n" |
| break |
| |
| yield "\n---\n✓ **Completed**\n" |
|
|
| EXAMPLES = [ |
| "What is 25 * 47?", |
| "What is the weather in Paris?", |
| "Who wrote 1984?", |
| "Calculate: 100 + 200", |
| ] |
|
|
| def run_comparison(question: str, mode: str): |
| """Run selected mode with real-time streaming.""" |
| if not question.strip(): |
| yield "Please enter a question.", "", "" |
| return |
| |
| if mode == "Think-Only": |
| think_out = "" |
| for chunk in think_only_mode(question): |
| think_out += chunk |
| yield think_out, "", "" |
| |
| elif mode == "Act-Only": |
| act_out = "" |
| for chunk in act_only_mode(question): |
| act_out += chunk |
| yield "", act_out, "" |
| |
| elif mode == "ReAct": |
| react_out = "" |
| for chunk in react_mode(question): |
| react_out += chunk |
| yield "", "", react_out |
| |
| else: |
| yield "Invalid mode selected.", "", "" |
|
|
| 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="Status", value="Click to download", interactive=False) |
| |
| with gr.Row(): |
| with gr.Column(scale=3): |
| question_input = gr.Textbox(label="Question", lines=3) |
| mode_dropdown = gr.Dropdown(choices=["Think-Only", "Act-Only", "ReAct"], value="ReAct", label="Mode") |
| submit_btn = gr.Button("Run", variant="primary", size="lg") |
| with gr.Column(scale=1): |
| 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(): |
| think_output = gr.Markdown(label="Think-Only") |
| act_output = gr.Markdown(label="Act-Only") |
| react_output = gr.Markdown(label="ReAct") |
| |
| 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) |
