Update app.py

app.py

@@ -1,61 +1,273 @@
-import gradio as gr
-from smolagents import Tool, CodeAgent
-import requests
+#!/usr/bin/env python3
+# coding: utf-8
+"""
+Minimal, robust app.py for the agent Space.
+- Uses smolagents CodeAgent + HfApiModel (free model google/flan-t5-small)
+- Exposes a few safe tools: web_search, visit_webpage, calculator, get_current_time_in_timezone, final_answer
+- Loads prompts.yaml and normalizes final_answer templates
+- Launches the existing Gradio_UI (Gradio_UI.py)
+"""
+
+import json
+import re
+import datetime
+import pytz
 import yaml
+import traceback
+from typing import Any
+
+from smolagents import CodeAgent, HfApiModel, tool
+
+# ---- Tools (each tool must have clear docstring describing args) ----
+
+@tool
+def final_answer(answer: str) -> str:
+    """
+    Receive the final answer and return it.
+
+    Args:
+        answer (str): The final answer string produced by the agent.
+
+    Returns:
+        str: The same answer (the framework captures it as the final result).
+    """
+    return str(answer)
+
+
+@tool
+def calculator(expr: str) -> str:
+    """
+    Safely evaluate a simple arithmetic expression.
+
+    Args:
+        expr (str): A math expression like "2 + 3 * (4 - 1)".
+
+    Returns:
+        str: JSON string: {"expression": expr, "result": value} or {"error": "..."}.
+    """
+    try:
+        # whitelist of AST nodes for safety
+        import ast, operator
+
+        allowed = {
+            ast.Add: operator.add,
+            ast.Sub: operator.sub,
+            ast.Mult: operator.mul,
+            ast.Div: operator.truediv,
+            ast.Pow: operator.pow,
+            ast.Mod: operator.mod,
+            ast.USub: operator.neg,
+        }
+
+        def _eval(node):
+            if isinstance(node, ast.Constant):
+                return node.value
+            if isinstance(node, ast.Num):
+                return node.n
+            if isinstance(node, ast.UnaryOp) and type(node.op) in allowed:
+                return allowed[type(node.op)](_eval(node.operand))
+            if isinstance(node, ast.BinOp) and type(node.op) in allowed:
+                return allowed[type(node.op)](_eval(node.left), _eval(node.right))
+            raise ValueError("Unsupported expression element: %s" % type(node))
+
+        tree = ast.parse(expr, mode="eval")
+        val = _eval(tree.body)
+        return json.dumps({"expression": expr, "result": float(val)})
+    except Exception as e:
+        return json.dumps({"error": f"Calc error: {e}"})
+
+
+@tool
+def get_current_time_in_timezone(timezone: str) -> str:
+    """
+    Get current local time for a timezone.
+
+    Args:
+        timezone (str): IANA timezone string (e.g. "Europe/Paris").
+
+    Returns:
+        str: JSON string {"timezone": tz, "local_time": "..."} or {"error":"..."}.
+    """
+    try:
+        tz = pytz.timezone(timezone)
+        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
+        return json.dumps({"timezone": timezone, "local_time": local_time}, ensure_ascii=False)
+    except Exception as e:
+        return json.dumps({"error": f"Timezone error: {e}"})
 
-# ---------------------------
-# TOOLS
-# ---------------------------
 
-@Tool
+@tool
 def web_search(query: str, max_results: int = 5) -> str:
-    """Perform a web search (dummy implementation)."""
-    return f"Search results for '{query}' (top {max_results} results)."
+    """
+    Search the web using duckduckgo_search (best-effort).
 
-@Tool
+    Args:
+        query (str): Search query string.
+        max_results (int): Number of results to return (default 5).
+
+    Returns:
+        str: JSON string list of results or {"error": "..."}.
+    """
+    try:
+        # local import so app can start even if ddg lib is missing
+        from duckduckgo_search import DDGS
+
+        with DDGS() as ddgs:
+            results = list(ddgs.text(query, max_results=max_results))
+        return json.dumps(results, ensure_ascii=False)
+    except Exception as e:
+        return json.dumps({"error": f"web_search failed: {str(e)}"})
+
+
+@tool
 def visit_webpage(url: str) -> str:
-    """Visit a webpage and return its HTML content."""
+    """
+    Download a webpage HTML using requests.
+
+    Args:
+        url (str): The page URL to fetch.
+
+    Returns:
+        str: The page HTML (string) or an error message.
+    """
     try:
-        response = requests.get(url, timeout=5)
-        return response.text[:2000]
+        import requests
+
+        headers = {
+            "User-Agent": "Mozilla/5.0 (compatible; InfoAgent/1.0; +https://example.com)"
+        }
+        r = requests.get(url, headers=headers, timeout=12)
+        r.raise_for_status()
+        return r.text
     except Exception as e:
-        return f"Error visiting {url}: {e}"
+        return json.dumps({"error": f"visit_webpage failed: {str(e)}"})
+
+
+# ---- load prompts.yaml and normalize final_answer template ----
+
+DEFAULT_PROMPTS = {
+    "system_prompt": (
+        "You are an assistant that solves tasks by producing short 'Thought:' then a 'Code:' block "
+        "wrapped as:\nCode:\n```py\n# python code\n```\n<end_code>\n"
+        "When finished call final_answer(...) with the result.\n"
+    ),
+    "final_answer": {
+        "pre_messages": "You are done. Produce a concise final answer.",
+        "post_messages": ""
+    },
+}
+
+def load_and_normalize_prompts(path: str = "prompts.yaml") -> dict:
+    try:
+        with open(path, "r", encoding="utf-8") as fh:
+            raw = yaml.safe_load(fh) or {}
+    except FileNotFoundError:
+        raw = {}
 
-# ---------------------------
-# LOAD CONFIG.YAML
-# ---------------------------
+    # extract system_prompt if present
+    system_prompt = raw.get("system_prompt", DEFAULT_PROMPTS["system_prompt"])
 
-with open("config.yaml", "r") as f:
-    config = yaml.safe_load(f)
+    # prepare prompt_templates dict: everything except system_prompt
+    templates = {k: v for k, v in raw.items() if k != "system_prompt"}
 
-system_prompt = config.get("system_prompt", "")
-prompt_templates = config.get("prompt_templates", {})
+    # ensure final_answer template exists and both fields are strings (Jinja expects strings)
+    fa = templates.get("final_answer")
+    if not isinstance(fa, dict):
+        fa = {}
+    pre = fa.get("pre_messages", DEFAULT_PROMPTS["final_answer"]["pre_messages"])
+    post = fa.get("post_messages", DEFAULT_PROMPTS["final_answer"]["post_messages"])
+    # If pre/post are lists, join them
+    if isinstance(pre, list):
+        pre = "\n".join(str(x) for x in pre)
+    if isinstance(post, list):
+        post = "\n".join(str(x) for x in post)
+    templates["final_answer"] = {"pre_messages": str(pre), "post_messages": str(post)}
+
+    # keep system_prompt included at top-level (CodeAgent expects system_prompt argument separately)
+    templates["_system_prompt"] = system_prompt
+    return templates
+
+
+prompt_templates_raw = load_and_normalize_prompts("prompts.yaml")
+system_prompt = prompt_templates_raw.pop("_system_prompt", DEFAULT_PROMPTS["system_prompt"])
+prompt_templates = prompt_templates_raw
+
+
+# ---- create model (HfApiModel) with fallback ----
+
+def create_model(preferred: str = "google/flan-t5-small"):
+    try:
+        model = HfApiModel(model_id=preferred, max_tokens=1024, temperature=0.0)
+        # ensure numeric token counters exist
+        try:
+            if getattr(model, "last_input_token_count", None) is None:
+                model.last_input_token_count = 0
+        except Exception:
+            pass
+        try:
+            if getattr(model, "last_output_token_count", None) is None:
+                model.last_output_token_count = 0
+        except Exception:
+            pass
+        return model
+    except Exception:
+        # fallback to a tiny local object implementing run()
+        class Fallback:
+            def __init__(self):
+                self.model_id = "fallback"
+                self.last_input_token_count = 0
+                self.last_output_token_count = 0
 
-# ---------------------------
-# AGENT SETUP
-# ---------------------------
+            def run(self, prompt: str, *args, **kwargs):
+                self.last_input_token_count = len(prompt.split()) if prompt else 0
+                reply = (
+                    "FALLBACK: remote model unavailable. I can still use tools: web_search, visit_webpage, "
+                    "calculator, get_current_time_in_timezone. Ask a concrete question."
+                )
+                self.last_output_token_count = len(reply.split())
+                return reply
+
+            def __call__(self, *args, **kwargs):
+                return self.run(args[0] if args else kwargs.get("prompt", ""))
+
+        return Fallback()
+
+
+model = create_model("google/flan-t5-small")
+
+# ensure numeric counters not None (protect Gradio_UI)
+try:
+    if getattr(model, "last_input_token_count", None) is None:
+        model.last_input_token_count = 0
+except Exception:
+    pass
+try:
+    if getattr(model, "last_output_token_count", None) is None:
+        model.last_output_token_count = 0
+except Exception:
+    pass
+
+# ---- create agent ----
+
+tools_list = [final_answer, calculator, get_current_time_in_timezone, web_search, visit_webpage]
 
 agent = CodeAgent(
-    model="google/flan-t5-small",
-    tools=[web_search, visit_webpage],
+    model=model,
+    tools=tools_list,
     max_steps=6,
+    verbosity_level=1,
     system_prompt=system_prompt,
     prompt_templates=prompt_templates,
+    name="MinimalInfoAgent",
+    description="Minimal, robust agent with small set of tools",
 )
 
-# ---------------------------
-# GRADIO UI
-# ---------------------------
-
-def chat_fn(message, history):
-    response = agent.run(message)
-    return str(response)
-
-demo = gr.ChatInterface(
-    chat_fn,
-    title="InfoAgent",
-    description="Posez vos questions factuelles, l'agent cherchera et répondra."
-)
-
+# ---- launch UI (uses Gradio_UI.py present in project) ----
 if __name__ == "__main__":
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+    try:
+        from Gradio_UI import GradioUI
+
+        GradioUI(agent).launch()
+    except Exception as exc:
+        print("Failed to launch Gradio UI:", exc)
+        traceback.print_exc()