Upload 2 files

app.py

@@ -0,0 +1,323 @@
+import gradio as gr
+import os
+import json
+import requests
+import re
+import traceback
+from datetime import datetime
+
+# ── Configuration ──
+OPENROUTER_API_KEY = os.environ.get("OPENROUTER_API_KEY")
+MODEL = "deepseek/deepseek-chat-v3-0324"  # Free on OpenRouter
+SITE_URL = "https://huggingface.co/spaces/YOUR_USERNAME/autonomous-agent"
+SITE_NAME = "Autonomous Agent"
+MAX_ITERATIONS = 10  # Safety limit for agent loop
+
+# ── OpenRouter Client ──
+def call_llm(messages, tools=None):
+    """Call OpenRouter API with optional tool definitions."""
+    headers = {
+        "Authorization": f"Bearer {OPENROUTER_API_KEY}",
+        "Content-Type": "application/json",
+        "HTTP-Referer": SITE_URL,
+        "X-Title": SITE_NAME,
+    }
+    payload = {
+        "model": MODEL,
+        "messages": messages,
+        "max_tokens": 2048,
+        "temperature": 0.6,
+    }
+    if tools:
+        payload["tools"] = tools
+        payload["tool_choice"] = "auto"
+
+    resp = requests.post(
+        "https://openrouter.ai/api/v1/chat/completions",
+        headers=headers,
+        json=payload,
+        timeout=60,
+    )
+    resp.raise_for_status()
+    return resp.json()
+
+
+# ── Tool Definitions (OpenAI function-calling format) ──
+TOOL_DEFINITIONS = [
+    {
+        "type": "function",
+        "function": {
+            "name": "web_search",
+            "description": "Search the web for current information. Returns top results with titles and snippets.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {"type": "string", "description": "Search query"}
+                },
+                "required": ["query"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "read_webpage",
+            "description": "Fetch and extract text content from a URL. Useful for reading articles, docs, or API references.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "url": {"type": "string", "description": "URL to fetch"}
+                },
+                "required": ["url"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "execute_python",
+            "description": "Execute Python code and return stdout. Use for calculations, data processing, file manipulation. No network access in sandbox.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "code": {"type": "string", "description": "Python code to execute"}
+                },
+                "required": ["code"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "write_file",
+            "description": "Write content to a file in the /data workspace. Returns confirmation.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "File path (relative to /data/)"},
+                    "content": {"type": "string", "description": "File content to write"}
+                },
+                "required": ["path", "content"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "read_file",
+            "description": "Read content from a file in the /data workspace.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "File path (relative to /data/)"}
+                },
+                "required": ["path"],
+            },
+        },
+    },
+]
+
+
+# ── Tool Implementations ──
+def tool_web_search(query):
+    """Search using DuckDuckGo HTML (no API key needed)."""
+    try:
+        resp = requests.get(
+            "https://html.duckduckgo.com/html/",
+            params={"q": query},
+            headers={"User-Agent": "Mozilla/5.0"},
+            timeout=10,
+        )
+        results = []
+        # Extract result snippets from HTML
+        snippets = re.findall(
+            r'<a rel="nofollow" class="result__a"[^>]*>(.*?)</a>.*?'
+            r'<a class="result__snippet"[^>]*>(.*?)</a>',
+            resp.text, re.DOTALL,
+        )
+        for title, snippet in snippets[:5]:
+            clean_title = re.sub(r"<.*?>", "", title).strip()
+            clean_snippet = re.sub(r"<.*?>", "", snippet).strip()
+            results.append(f"**{clean_title}**\n{clean_snippet}")
+        return "\n\n".join(results) if results else "No results found."
+    except Exception as e:
+        return f"Search error: {e}"
+
+
+def tool_read_webpage(url):
+    """Fetch and extract text from a URL."""
+    try:
+        resp = requests.get(
+            url,
+            headers={"User-Agent": "Mozilla/5.0"},
+            timeout=15,
+        )
+        resp.raise_for_status()
+        # Strip HTML tags for plain text
+        text = re.sub(r"<script[^>]*>.*?</script>", "", resp.text, flags=re.DOTALL)
+        text = re.sub(r"<style[^>]*>.*?</style>", "", text, flags=re.DOTALL)
+        text = re.sub(r"<.*?>", " ", text)
+        text = re.sub(r"\s+", " ", text).strip()
+        return text[:5000]  # Limit to 5000 chars
+    except Exception as e:
+        return f"Error fetching URL: {e}"
+
+
+def tool_execute_python(code):
+    """Execute Python code in a restricted namespace."""
+    import io
+    import contextlib
+
+    output = io.StringIO()
+    namespace = {"__builtins__": __builtins__}
+    try:
+        with contextlib.redirect_stdout(output):
+            exec(code, namespace)
+        result = output.getvalue()
+        return result if result else "(Code executed successfully, no output)"
+    except Exception as e:
+        return f"Error: {traceback.format_exc()}"
+
+
+def tool_write_file(path, content):
+    """Write content to /data/ directory."""
+    try:
+        full_path = os.path.join("/data", path)
+        os.makedirs(os.path.dirname(full_path), exist_ok=True)
+        with open(full_path, "w") as f:
+            f.write(content)
+        return f"File written: /data/{path} ({len(content)} chars)"
+    except Exception as e:
+        return f"Error writing file: {e}"
+
+
+def tool_read_file(path):
+    """Read content from /data/ directory."""
+    try:
+        full_path = os.path.join("/data", path)
+        with open(full_path, "r") as f:
+            return f.read()[:5000]
+    except Exception as e:
+        return f"Error reading file: {e}"
+
+
+# Map tool names to implementations
+TOOL_MAP = {
+    "web_search": tool_web_search,
+    "read_webpage": tool_read_webpage,
+    "execute_python": tool_execute_python,
+    "write_file": tool_write_file,
+    "read_file": tool_read_file,
+}
+
+
+# ── Agent Loop (ReAct) ──
+SYSTEM_PROMPT = """You are an autonomous AI agent. You have access to tools that let you search the web, 
+read webpages, execute Python code, and read/write files. 
+
+When given a task:
+1. Think step-by-step about what you need to do
+2. Use tools to gather information or perform actions
+3. Synthesize the results into a clear answer
+4. If one approach fails, try another
+
+Always be thorough. If a task requires multiple steps, complete all of them.
+When writing code, always show the code and its output.
+"""
+
+def run_agent(user_message, history):
+    """Run the autonomous agent loop with streaming."""
+    # Build conversation history
+    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
+    for user_msg, asst_msg in history:
+        messages.append({"role": "user", "content": user_msg})
+        messages.append({"role": "assistant", "content": asst_msg})
+    messages.append({"role": "user", "content": user_message})
+
+    full_response = ""
+    iteration = 0
+
+    while iteration < MAX_ITERATIONS:
+        iteration += 1
+
+        # Call LLM
+        try:
+            result = call_llm(messages, tools=TOOL_DEFINITIONS)
+        except Exception as e:
+            error_msg = f"⚠️ API Error: {e}"
+            full_response += f"\n{error_msg}"
+            yield full_response.strip()
+            break
+
+        choice = result["choices"][0]
+        msg = choice["message"]
+
+        # If LLM wants to call tools
+        if msg.get("tool_calls"):
+            # Add assistant message with tool calls to history
+            messages.append(msg)
+
+            for tool_call in msg["tool_calls"]:
+                fn_name = tool_call["function"]["name"]
+                fn_args = json.loads(tool_call["function"]["arguments"])
+                tool_call_id = tool_call["id"]
+
+                # Show the user what the agent is doing
+                args_str = ", ".join(f"{k}={v!r}" for k, v in fn_args.items())
+                action_text = f"\n🔧 **{fn_name}**({args_str})\n"
+                full_response += action_text
+                yield full_response.strip()
+
+                # Execute the tool
+                if fn_name in TOOL_MAP:
+                    tool_result = TOOL_MAP[fn_name](**fn_args)
+                else:
+                    tool_result = f"Unknown tool: {fn_name}"
+
+                # Show result summary
+                result_preview = tool_result[:300] + ("..." if len(tool_result) > 300 else "")
+                full_response += f"📋 {result_preview}\n"
+                yield full_response.strip()
+
+                # Feed result back to LLM
+                messages.append({
+                    "role": "tool",
+                    "tool_call_id": tool_call_id,
+                    "content": tool_result,
+                })
+            # Continue loop — LLM will reason about tool results
+
+        else:
+            # LLM gave a final answer (no tool calls)
+            final_text = msg.get("content", "")
+            if full_response:
+                full_response += f"\n\n---\n\n{final_text}"
+            else:
+                full_response = final_text
+            yield full_response.strip()
+            break
+
+    else:
+        # Hit iteration limit
+        full_response += "\n\n⚠️ Reached maximum iterations. Task may be incomplete."
+        yield full_response.strip()
+
+
+# ── Gradio Interface ──
+demo = gr.ChatInterface(
+    fn=run_agent,
+    title="🤖 Autonomous Agent",
+    description=(
+        "An autonomous AI agent powered by DeepSeek V3 (free via OpenRouter). "
+        "It can search the web, read pages, execute Python code, and manage files.\n\n"
+        "Try: *'Research the latest news about AI agents and write a summary to a file'*"
+    ),
+    type="messages",
+    chatbot=gr.Chatbot(height=600, show_copy_button=True),
+    textbox=gr.Textbox(placeholder="Give me a task...", scale=7),
+    theme=gr.themes.Soft(),
+)
+
+if __name__ == "__main__":
+    demo.queue(max_size=20)
+    demo.launch()

requirements.txt

@@ -0,0 +1,2 @@
+gradio==4.44.0
+requests==2.32.3