Upload 5 files

curl_scraper.py

@@ -0,0 +1,226 @@
+import subprocess
+import tempfile
+import os
+from io import StringIO
+import re
+from functools import lru_cache
+from typing import Optional, Dict, Union
+from browserforge.headers import Browser, HeaderGenerator
+from tldextract import extract
+from markitdown import MarkItDown
+from markdown import Markdown
+import brotli
+import zstandard as zstd
+import gzip
+import zlib
+from urllib.parse import unquote
+from smolagents import tool
+
+class Response:
+    def __init__(self, response, convert_to_markdown, convert_to_plain_text):
+        self._response = response
+        self._convert_to_markdown = convert_to_markdown
+        self._convert_to_plain_text = convert_to_plain_text
+        self._markdown = None
+        self._plain_text = None
+
+    def __getattr__(self, item):
+        return getattr(self._response, item)
+
+    @property
+    def markdown(self) -> str:
+        if self._markdown is None:
+            self._markdown = self._convert_to_markdown(self._response.content)
+        return self._markdown
+
+    @property
+    def plain_text(self) -> str:
+        if self._plain_text is None:
+            self._plain_text = self._convert_to_plain_text(self._response.content)
+        return self._plain_text
+
+def generate_headers() -> Dict[str, str]:
+    browsers = [
+        Browser(name='chrome', min_version=120),
+        Browser(name='firefox', min_version=120),
+        Browser(name='edge', min_version=120),
+    ]
+    return HeaderGenerator(browser=browsers, device='desktop').generate()
+
+@lru_cache(None, typed=True)
+def generate_convincing_referer(url: str) -> str:
+    website_name = extract(url).domain
+    return f'https://www.google.com/search?q={website_name}'
+
+def headers_job(headers: Optional[Dict], url: str) -> Dict:
+    headers = headers or {}
+    # Ensure a User-Agent is present.
+    headers['User-Agent'] = generate_headers().get('User-Agent')
+    extra_headers = generate_headers()
+    headers.update(extra_headers)
+    headers.update({'referer': generate_convincing_referer(url)})
+    return headers
+
+def convert_to_markdown(content: bytes) -> str:
+    md = MarkItDown()
+    temp_path = None
+    try:
+        with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
+            tmp_file.write(content)
+            tmp_file.flush()
+            temp_path = tmp_file.name
+        markdown_result = md.convert_local(temp_path).text_content
+        return markdown_result
+    except Exception as e:
+        raise e
+    finally:
+        if temp_path and os.path.exists(temp_path):
+            os.remove(temp_path)
+
+def convert_to_plain_text(content: bytes) -> str:
+    md_content = convert_to_markdown(content)
+
+    def unmark_element(element, stream=None):
+        if stream is None:
+            stream = StringIO()
+        if element.text:
+            stream.write(element.text)
+        for sub in element:
+            unmark_element(sub, stream)
+        if element.tail:
+            stream.write(element.tail)
+        return stream.getvalue()
+
+    Markdown.output_formats["plain"] = unmark_element
+    __md = Markdown(output_format="plain")
+    __md.stripTopLevelTags = False
+
+    final_text = __md.convert(md_content)
+    final_text = re.sub(r"\n+", " ", final_text)
+    return final_text
+
+class BasicScraper:
+    """Basic scraper class for making HTTP requests using curl."""
+    def __init__(
+        self,
+        proxy: Optional[str] = None,
+        follow_redirects: bool = True,
+        timeout: Optional[Union[int, float]] = None,
+        retries: Optional[int] = 3
+    ):
+        self.proxy = proxy
+        self.timeout = timeout
+        self.follow_redirects = bool(follow_redirects)
+        self.retries = retries
+
+    def _curl_get(
+        self,
+        url: str,
+        headers: Dict[str, str],
+        cookies: Optional[Dict],
+        timeout: Optional[Union[int, float]],
+        proxy: Optional[str],
+        follow_redirects: bool
+    ) -> bytes:
+        # Use -i to include HTTP headers in the output.
+        curl_command = ["curl", "-s", "-i"]
+        if follow_redirects:
+            curl_command.append("-L")
+        if self.retries:
+            curl_command.extend(["--retry", str(self.retries)])
+        # Add headers.
+        for key, value in headers.items():
+            curl_command.extend(["-H", f"{key}: {value}"])
+        # Add cookies if provided.
+        if cookies:
+            cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()])
+            curl_command.extend(["--cookie", cookie_str])
+        # Set proxy if specified.
+        if proxy:
+            curl_command.extend(["--proxy", proxy])
+        # Set timeout options.
+        if timeout:
+            curl_command.extend(["--connect-timeout", str(timeout), "--max-time", str(timeout)])
+        curl_command.append(url)
+        try:
+            result = subprocess.run(
+                curl_command,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                check=False
+            )
+            if result.returncode != 0:
+                raise Exception(f"Curl command failed: {result.stderr.decode('utf-8')}")
+            raw_response = result.stdout
+            # Split the response into header blocks and body.
+            parts = raw_response.split(b'\r\n\r\n')
+            if len(parts) >= 2:
+                body = parts[-1]
+                last_header_block = parts[-2]
+            else:
+                body = raw_response
+                last_header_block = b""
+            # Look for a Content-Encoding header in the last header block.
+            content_encoding = None
+            for line in last_header_block.decode('utf-8', errors='ignore').splitlines():
+                if line.lower().startswith("content-encoding:"):
+                    content_encoding = line.split(":", 1)[1].strip().lower()
+                    break
+            # Decode Brotli or Zstandard if needed.
+            if content_encoding:
+                try:
+                    if 'br' in content_encoding:
+                        body = brotli.decompress(body)
+                    elif 'zstd' in content_encoding:
+                        dctx = zstd.ZstdDecompressor()
+                        try:
+                            body = dctx.decompress(body)
+                        except zstd.ZstdError as e:
+                            # Fallback to streaming decompression if content size is unknown
+                            if "could not determine content size" in str(e):
+                                dctx_stream = zstd.ZstdDecompressor().decompressobj()
+                                body = dctx_stream.decompress(body)
+                                body += dctx_stream.flush()
+                            else:
+                                raise
+                    elif 'gzip' in content_encoding:
+                        body = gzip.decompress(body)
+                    elif 'deflate' in content_encoding:
+                        body = zlib.decompress(body)
+                except Exception as e:
+                    raise Exception(f"Error decompressing content: {e}")
+                    
+            return body
+        except Exception as e:
+            raise Exception(f"Error during curl request: {e}")
+
+    def get(
+        self,
+        url: str,
+        cookies: Optional[Dict] = None,
+        timeout: Optional[Union[int, float]] = None,
+        **kwargs: Dict
+    ) -> Response:
+        url = unquote(url).replace(" ", "+")
+        hdrs = headers_job(kwargs.pop('headers', {}), url)
+        effective_timeout = self.timeout if self.timeout is not None else timeout
+        content = self._curl_get(
+            url,
+            headers=hdrs,
+            cookies=cookies,
+            timeout=effective_timeout,
+            proxy=self.proxy,
+            follow_redirects=self.follow_redirects
+        )
+        # Create a dummy response object with a 'content' attribute.
+        class DummyResponse:
+            pass
+
+        dummy = DummyResponse()
+        dummy.content = content
+
+        return Response(
+            response=dummy,
+            convert_to_markdown=convert_to_markdown,
+            convert_to_plain_text=convert_to_plain_text
+        )

gradio_agent.py

@@ -0,0 +1,565 @@
+import os
+import re
+import shutil
+import tempfile
+import yaml
+from pathlib import Path
+from typing import Generator
+
+from smolagents import CodeAgent, LiteLLMModel, MultiStepAgent, PlanningStep, Tool
+from smolagents.agent_types import AgentAudio, AgentImage, AgentText
+from smolagents.memory import ActionStep, FinalAnswerStep, MemoryStep
+from smolagents.models import ChatMessageStreamDelta
+from smolagents.utils import _is_package_available
+from tools import search, scraper, FileReader, FileWriter, CommitChanges, get_repository_structure # Assuming tools.py is in the same directory
+
+def get_step_footnote_content(step_log: MemoryStep, step_name: str) -> str:
+    """Get a footnote string for a step log with duration and token information"""
+    step_footnote = f"**{step_name}**"
+    if hasattr(step_log, "input_token_count") and hasattr(step_log, "output_token_count"):
+        token_str = f" | Input tokens: {step_log.input_token_count:,} | Output tokens: {step_log.output_token_count:,}"
+        step_footnote += token_str
+    if hasattr(step_log, "duration"):
+        step_duration = f" | Duration: {round(float(step_log.duration), 2)}" if step_log.duration else None
+        step_footnote += step_duration
+    step_footnote_content = f"""<span style="color: #bbbbc2; font-size: 12px;">{step_footnote}</span> """
+    return step_footnote_content
+
+
+def _clean_model_output(model_output: str) -> str:
+    """
+    Clean up model output by removing trailing tags and extra backticks.
+
+    Args:
+        model_output (`str`): Raw model output.
+
+    Returns:
+        `str`: Cleaned model output.
+    """
+    if not model_output:
+        return ""
+    model_output = model_output.strip()
+    # Remove any trailing <end_code> and extra backticks, handling multiple possible formats
+    model_output = re.sub(r"```\s*<end_code>", "```", model_output)  # handles ```<end_code>
+    model_output = re.sub(r"<end_code>\s*```", "```", model_output)  # handles <end_code>```
+    model_output = re.sub(r"```\s*\n\s*<end_code>", "```", model_output)  # handles ```\n<end_code>
+    return model_output.strip()
+
+
+def _format_code_content(content: str) -> str:
+    """
+    Format code content as Python code block if it's not already formatted.
+
+    Args:
+        content (`str`): Code content to format.
+
+    Returns:
+        `str`: Code content formatted as a Python code block.
+    """
+    content = content.strip()
+    # Remove existing code blocks and end_code tags
+    content = re.sub(r"```.*?\n", "", content)
+    content = re.sub(r"\s*<end_code>\s*", "", content)
+    content = content.strip()
+    # Add Python code block formatting if not already present
+    if not content.startswith("```python"):
+        content = f"```python\n{content}\n```"
+    return content
+
+
+def _process_action_step(step_log: ActionStep, skip_model_outputs: bool = False) -> Generator:
+    """
+    Process an [`ActionStep`] and yield appropriate Gradio ChatMessage objects.
+
+    Args:
+        step_log ([`ActionStep`]): ActionStep to process.
+        skip_model_outputs (`bool`): Whether to skip model outputs.
+
+    Yields:
+        `gradio.ChatMessage`: Gradio ChatMessages representing the action step.
+    """
+    import gradio as gr
+
+    # Output the step number
+    step_number = f"Step {step_log.step_number}" if step_log.step_number is not None else "Step"
+    if not skip_model_outputs:
+        yield gr.ChatMessage(role="assistant", content=f"**{step_number}**", metadata={"status": "done"})
+
+    # First yield the thought/reasoning from the LLM
+    if not skip_model_outputs and getattr(step_log, "model_output", ""):
+        model_output = _clean_model_output(step_log.model_output)
+        yield gr.ChatMessage(role="assistant", content=model_output, metadata={"status": "done"})
+
+    # For tool calls, create a parent message
+    if getattr(step_log, "tool_calls", []):
+        first_tool_call = step_log.tool_calls[0]
+        used_code = first_tool_call.name == "python_interpreter"
+
+        # Process arguments based on type
+        args = first_tool_call.arguments
+        if isinstance(args, dict):
+            content = str(args.get("answer", str(args)))
+        else:
+            content = str(args).strip()
+
+        # Format code content if needed
+        if used_code:
+            content = _format_code_content(content)
+
+        # Create the tool call message
+        parent_message_tool = gr.ChatMessage(
+            role="assistant",
+            content=content,
+            metadata={
+                "title": f"🛠️ Used tool {first_tool_call.name}",
+                "status": "done",
+            },
+        )
+        yield parent_message_tool
+
+    # Display execution logs if they exist
+    if getattr(step_log, "observations", "") and step_log.observations.strip():
+        log_content = step_log.observations.strip()
+        if log_content:
+            log_content = re.sub(r"^Execution logs:\s*", "", log_content)
+            yield gr.ChatMessage(
+                role="assistant",
+                content=f"```bash\n{log_content}\n",
+                metadata={"title": "📝 Execution Logs", "status": "done"},
+            )
+
+    # Display any images in observations
+    if getattr(step_log, "observations_images", []):
+        for image in step_log.observations_images:
+            path_image = AgentImage(image).to_string()
+            yield gr.ChatMessage(
+                role="assistant",
+                content={"path": path_image, "mime_type": f"image/{path_image.split('.')[-1]}"},
+                metadata={"title": "🖼️ Output Image", "status": "done"},
+            )
+
+    # Handle errors
+    if getattr(step_log, "error", None):
+        yield gr.ChatMessage(
+            role="assistant", content=str(step_log.error), metadata={"title": "💥 Error", "status": "done"}
+        )
+
+    # Add step footnote and separator
+    yield gr.ChatMessage(
+        role="assistant", content=get_step_footnote_content(step_log, step_number), metadata={"status": "done"}
+    )
+    yield gr.ChatMessage(role="assistant", content="-----", metadata={"status": "done"})
+
+
+def _process_planning_step(step_log: PlanningStep, skip_model_outputs: bool = False) -> Generator:
+    """
+    Process a [`PlanningStep`] and yield appropriate gradio.ChatMessage objects.
+
+    Args:
+        step_log ([`PlanningStep`]): PlanningStep to process.
+
+    Yields:
+        `gradio.ChatMessage`: Gradio ChatMessages representing the planning step.
+    """
+    import gradio as gr
+
+    if not skip_model_outputs:
+        yield gr.ChatMessage(role="assistant", content="**Planning step**", metadata={"status": "done"})
+        yield gr.ChatMessage(role="assistant", content=step_log.plan, metadata={"status": "done"})
+    yield gr.ChatMessage(
+        role="assistant", content=get_step_footnote_content(step_log, "Planning step"), metadata={"status": "done"}
+    )
+    yield gr.ChatMessage(role="assistant", content="-----", metadata={"status": "done"})
+
+
+def _process_final_answer_step(step_log: FinalAnswerStep) -> Generator:
+    """
+    Process a [`FinalAnswerStep`] and yield appropriate gradio.ChatMessage objects.
+
+    Args:
+        step_log ([`FinalAnswerStep`]): FinalAnswerStep to process.
+
+    Yields:
+        `gradio.ChatMessage`: Gradio ChatMessages representing the final answer.
+    """
+    import gradio as gr
+
+    final_answer = step_log.final_answer
+    if isinstance(final_answer, AgentText):
+        yield gr.ChatMessage(
+            role="assistant",
+            content=f"**Final answer:**\n{final_answer.to_string()}\n",
+            metadata={"status": "done"},
+        )
+    elif isinstance(final_answer, AgentImage):
+        yield gr.ChatMessage(
+            role="assistant",
+            content={"path": final_answer.to_string(), "mime_type": "image/png"},
+            metadata={"status": "done"},
+        )
+    elif isinstance(final_answer, AgentAudio):
+        yield gr.ChatMessage(
+            role="assistant",
+            content={"path": final_answer.to_string(), "mime_type": "audio/wav"},
+            metadata={"status": "done"},
+        )
+    else:
+        yield gr.ChatMessage(
+            role="assistant", content=f"**Final answer:** {str(final_answer)}", metadata={"status": "done"}
+        )
+
+
+def pull_messages_from_step(step_log: MemoryStep, skip_model_outputs: bool = False):
+    """Extract ChatMessage objects from agent steps with proper nesting.
+
+    Args:
+        step_log: The step log to display as gr.ChatMessage objects.
+        skip_model_outputs: If True, skip the model outputs when creating the gr.ChatMessage objects:
+            This is used for instance when streaming model outputs have already been displayed.
+    """
+    if not _is_package_available("gradio"):
+        raise ModuleNotFoundError(
+            "Please install 'gradio' extra to use the GradioUI: `pip install 'smolagents[gradio]'`"
+        )
+    if isinstance(step_log, ActionStep):
+        yield from _process_action_step(step_log, skip_model_outputs)
+    elif isinstance(step_log, PlanningStep):
+        yield from _process_planning_step(step_log, skip_model_outputs)
+    elif isinstance(step_log, FinalAnswerStep):
+        yield from _process_final_answer_step(step_log)
+    else:
+        raise ValueError(f"Unsupported step type: {type(step_log)}")
+
+
+def stream_to_gradio(
+    agent,
+    task: str,
+    task_images: list | None = None,
+    reset_agent_memory: bool = False,
+    additional_args: dict | None = None,
+):
+    """Runs an agent with the given task and streams the messages from the agent as gradio ChatMessages."""
+    if not _is_package_available("gradio"):
+        raise ModuleNotFoundError(
+            "Please install 'gradio' extra to use the GradioUI: `pip install 'smolagents[gradio]'`"
+        )
+    intermediate_text = ""
+    for step_log in agent.run(
+        task, images=task_images, stream=True, reset=reset_agent_memory, additional_args=additional_args
+    ):
+        # Track tokens if model provides them
+        if getattr(agent.model, "last_input_token_count", None) is not None:
+            if isinstance(step_log, (ActionStep, PlanningStep)):
+                step_log.input_token_count = agent.model.last_input_token_count
+                step_log.output_token_count = agent.model.last_output_token_count
+
+        if isinstance(step_log, MemoryStep):
+            intermediate_text = ""
+            for message in pull_messages_from_step(
+                step_log,
+                # If we're streaming model outputs, no need to display them twice
+                skip_model_outputs=getattr(agent, "stream_outputs", False),
+            ):
+                yield message
+        elif isinstance(step_log, ChatMessageStreamDelta):
+            intermediate_text += step_log.content or ""
+            yield intermediate_text
+
+
+class GradioUI:
+    """A one-line interface to launch your agent in Gradio"""
+
+    def __init__(self, agent_name: str = "Agent interface", agent_description: str | None = None):
+        if not _is_package_available("gradio"):
+            raise ModuleNotFoundError(
+                "Please install 'gradio' extra to use the GradioUI: `pip install 'smolagents[gradio]'`"
+            )
+        self.agent_name = agent_name
+        self.agent_description = agent_description
+
+    def _initialize_agents(self, space_id: str, temp_dir: str):
+        """Initializes agents and tools for a given space_id and temporary directory."""
+        # Initialize model with your API key
+        model = LiteLLMModel(model_id="gemini/gemini-2.0-flash")
+
+        # Get repository structure
+        repo_structure = get_repository_structure(space_id=space_id)
+
+        # Load prompt templates
+        try:
+            with open("planning_agent_prompt_templates.yaml", "r") as f:
+                planning_agent_prompt_templates = yaml.safe_load(f)
+
+            with open("swe_agent_prompt_templates.yaml", "r") as f:
+                swe_agent_prompt_templates = yaml.safe_load(f)
+        except FileNotFoundError as e:
+            print(f"Error loading prompt templates: {e}")
+            print("Please ensure 'planning_agent_prompt_templates.yaml' and 'swe_agent_prompt_templates.yaml' are in the same directory.")
+            return None, None # Indicate failure
+
+        # Enhance prompts with repository structure
+        planning_agent_prompt_templates["system_prompt"] = planning_agent_prompt_templates["system_prompt"] + "\n\n\n" + repo_structure + "\n\n"
+        swe_agent_prompt_templates["system_prompt"] = swe_agent_prompt_templates["system_prompt"] + "\n\n\n" + repo_structure + "\n\n"
+
+        # Initialize tool instances with temp directory
+        read_file = FileReader(space_id=space_id, folder_path=temp_dir)
+        write_file = FileWriter(space_id=space_id, folder_path=temp_dir)
+        commit_changes = CommitChanges(space_id=space_id, folder_path=temp_dir)
+
+        # Initialize SWE Agent with enhanced capabilities and improved description
+        swe_agent = CodeAgent(
+            model=model,
+            prompt_templates=swe_agent_prompt_templates,
+            verbosity_level=1,
+            tools=[search, scraper, read_file, write_file],
+            name="swe_agent",
+            description="An expert Software Engineer capable of designing, developing, and debugging code. This agent can read and write files, search the web, and scrape web content to assist in coding tasks. It excels at implementing detailed technical specifications provided by the Planning Agent."
+        )
+
+        # Initialize Planning Agent with improved planning focus and description
+        planning_agent = CodeAgent(
+            model=model,
+            prompt_templates=planning_agent_prompt_templates,
+            verbosity_level=1,
+            tools=[search, scraper, read_file, write_file, commit_changes],
+            managed_agents=[swe_agent],
+            name="planning_agent",
+            description="A high-level planning agent responsible for breaking down complex user requests into actionable steps and delegating coding tasks to the Software Engineer Agent. It focuses on strategy, task decomposition, and coordinating the overall development process.",
+            stream_outputs=True
+        )
+
+        return planning_agent, swe_agent
+
+    def _update_space_id_and_agents(self, new_space_id: str, current_state: dict):
+        """Handles space_id change, cleans up old temp dir, creates new, and re-initializes agents."""
+        import gradio as gr
+
+        old_temp_dir = current_state.get("temp_dir")
+        if old_temp_dir and os.path.exists(old_temp_dir):
+            print(f"Cleaning up old temporary directory: {old_temp_dir}")
+            shutil.rmtree(old_temp_dir)
+
+        if not new_space_id:
+            current_state["space_id"] = None
+            current_state["temp_dir"] = None
+            current_state["agent"] = None
+            current_state["file_upload_folder"] = None
+            print("Space ID is empty. Agents are not initialized.")
+            return new_space_id, None, None, None, gr.Textbox("Please enter a Space ID to initialize agents.", visible=True)
+
+        # Create new temporary directory
+        temp_dir = tempfile.mkdtemp(prefix=f"ai_workspace_{new_space_id.replace("/", "_")}_")
+        file_upload_folder = Path(temp_dir) / "uploads"
+        file_upload_folder.mkdir(parents=True, exist_ok=True)
+
+        # Initialize agents
+        planning_agent, _ = self._initialize_agents(new_space_id, temp_dir)
+
+        if planning_agent is None:
+             # Handle initialization failure
+            shutil.rmtree(temp_dir) # Clean up the newly created temp dir
+            current_state["space_id"] = None
+            current_state["temp_dir"] = None
+            current_state["agent"] = None
+            current_state["file_upload_folder"] = None
+            return new_space_id, None, None, None, gr.Textbox("Failed to initialize agents. Check console for errors.", visible=True)
+
+
+        # Update session state
+        current_state["space_id"] = new_space_id
+        current_state["temp_dir"] = temp_dir
+        current_state["agent"] = planning_agent
+        current_state["file_upload_folder"] = file_upload_folder
+
+        print(f"Initialized agents for Space ID: {new_space_id} in {temp_dir}")
+        return new_space_id, [], [], file_upload_folder, gr.Textbox(f"Agents initialized for Space ID: {new_space_id}", visible=True)
+
+
+    def interact_with_agent(self, prompt, messages, session_state):
+        import gradio as gr
+
+        agent = session_state.get("agent")
+        if agent is None:
+            messages.append(gr.ChatMessage(role="assistant", content="Please enter a Space ID and initialize the agents first."))
+            yield messages
+            return
+
+        try:
+            messages.append(gr.ChatMessage(role="user", content=prompt, metadata={"status": "done"}))
+            yield messages
+
+            for msg in stream_to_gradio(agent, task=prompt, reset_agent_memory=False):
+                if isinstance(msg, gr.ChatMessage):
+                    messages.append(msg)
+                elif isinstance(msg, str):  # Then it's only a completion delta
+                    try:
+                        if messages[-1].metadata["status"] == "pending":
+                            messages[-1].content = msg
+                        else:
+                            messages.append(
+                                gr.ChatMessage(role="assistant", content=msg, metadata={"status": "pending"})
+                            )
+                    except Exception as e:
+                        raise e
+                yield messages
+
+            yield messages
+        except Exception as e:
+            print(f"Error in interaction: {str(e)}")
+            messages.append(gr.ChatMessage(role="assistant", content=f"Error: {str(e)}"))
+            yield messages
+
+    def upload_file(self, file, file_uploads_log, session_state, allowed_file_types=None):
+        """
+        Handle file uploads, default allowed types are .pdf, .docx, and .txt
+        """
+        import gradio as gr
+
+        file_upload_folder = session_state.get("file_upload_folder")
+        if file_upload_folder is None:
+             return gr.Textbox("Please enter a Space ID and initialize agents before uploading files.", visible=True), file_uploads_log
+
+        if file is None:
+            return gr.Textbox(value="No file uploaded", visible=True), file_uploads_log
+
+        if allowed_file_types is None:
+            allowed_file_types = [".pdf", ".docx", ".txt"]
+
+        file_ext = os.path.splitext(file.name)[1].lower()
+        if file_ext not in allowed_file_types:
+            return gr.Textbox("File type disallowed", visible=True), file_uploads_log
+
+        # Sanitize file name
+        original_name = os.path.basename(file.name)
+        sanitized_name = re.sub(
+            r"[^\w\-.]", "_", original_name
+        )  # Replace any non-alphanumeric, non-dash, or non-dot characters with underscores
+
+        # Save the uploaded file to the specified folder
+        file_path = os.path.join(file_upload_folder, os.path.basename(sanitized_name))
+        shutil.copy(file.name, file_path)
+
+        return gr.Textbox(f"File uploaded: {file_path}", visible=True), file_uploads_log + [file_path]
+
+    def log_user_message(self, text_input, file_uploads_log):
+        import gradio as gr
+
+        return (
+            text_input
+            + (
+                f"\nYou have been provided with these files, which might be helpful or not: {file_uploads_log}"
+                if len(file_uploads_log) > 0
+                else ""
+            ),
+            "",
+            gr.Button(interactive=False),
+        )
+
+    def launch(self, share: bool = True, **kwargs):
+        self.create_app().launch(debug=True, share=share, **kwargs)
+
+    def create_app(self):
+        import gradio as gr
+
+        with gr.Blocks(theme="ocean", fill_height=True) as demo:
+            # Add session state to store session-specific data
+            session_state = gr.State({})
+            stored_messages = gr.State([])
+            file_uploads_log = gr.State([])
+            space_id_state = gr.State(None) # State to hold the current space_id
+            temp_dir_state = gr.State(None) # State to hold the current temp_dir
+            file_upload_folder_state = gr.State(None) # State to hold the current file upload folder
+
+            with gr.Sidebar():
+                gr.Markdown(
+                    f"# {self.agent_name.replace('_', ' ').capitalize()}"
+                    "\n> This web ui allows you to interact with a `smolagents` agent that can use tools and execute steps to complete tasks."
+                    + (f"\n\n**Agent description:**\n{self.agent_description}" if self.agent_description else "")
+                )
+
+                # Add Space ID input
+                space_id_input = gr.Textbox(
+                    label="Hugging Face Space ID",
+                    placeholder="Enter your Space ID (e.g., username/space-name)",
+                    interactive=True
+                )
+                initialization_status = gr.Textbox(label="Initialization Status", interactive=False, visible=True)
+
+                # Trigger agent initialization when space_id changes
+                space_id_input.change(
+                    self._update_space_id_and_agents,
+                    [space_id_input, session_state],
+                    [space_id_state, stored_messages, file_uploads_log, file_upload_folder_state, initialization_status]
+                )
+
+
+                with gr.Group():
+                    gr.Markdown("**Your request**", container=True)
+                    text_input = gr.Textbox(
+                        lines=3,
+                        label="Chat Message",
+                        container=False,
+                        placeholder="Enter your prompt here and press Shift+Enter or press the button",
+                    )
+                    submit_btn = gr.Button("Submit", variant="primary")
+
+                gr.HTML(
+                    "<br><br><h4><center>Powered by <a target='_blank' href='https://github.com/huggingface/smolagents'><b>smolagents</b></a></center></h4>"
+                )
+
+            # Main chat interface
+            chatbot = gr.Chatbot(
+                label="Agent",
+                type="messages",
+                avatar_images=(
+                    None,
+                    "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/smolagents/mascot_smol.png",
+                ),
+                resizeable=True,
+                scale=1,
+            )
+
+            # Set up event handlers
+            text_input.submit(
+                self.log_user_message,
+                [text_input, file_uploads_log],
+                [stored_messages, text_input, submit_btn],
+            ).then(self.interact_with_agent, [stored_messages, chatbot, session_state], [chatbot]).then(
+                lambda: (
+                    gr.Textbox(
+                        interactive=True, placeholder="Enter your prompt here and press Shift+Enter or the button"
+                    ),
+                    gr.Button(interactive=True),
+                ),
+                None,
+                [text_input, submit_btn],
+            )
+
+            submit_btn.click(
+                self.log_user_message,
+                [text_input, file_uploads_log],
+                [stored_messages, text_input, submit_btn],
+            ).then(self.interact_with_agent, [stored_messages, chatbot, session_state], [chatbot]).then(
+                lambda: (
+                    gr.Textbox(
+                        interactive=True, placeholder="Enter your prompt here and press Shift+Enter or the button"
+                    ),
+                    gr.Button(interactive=True),
+                ),
+                None,
+                [text_input, submit_btn],
+            )
+
+        return demo
+
+def run_gradio_agent():
+    """Runs the Gradio UI for the agent."""
+    # The GradioUI class now handles space_id input and agent initialization internally
+    GradioUI(
+        agent_name="SmolAgents Gradio Interface",
+        agent_description="Interact with a SmolAgents planning agent capable of code generation and execution."
+    ).launch()
+
+if __name__ == "__main__":
+    # The script now directly launches the Gradio UI
+    run_gradio_agent()

planning_agent_prompt_templates.yaml

@@ -0,0 +1,313 @@
+system_prompt: |-
+  # Strategic Planning Agent
+
+  You are an expert AI Strategic Planning Agent responsible for decomposing complex tasks into manageable chunks and creating executable plans. Your primary focus is on high-level planning, task decomposition, and delegation to specialized agents like Software Engineers rather than coding yourself.
+
+  ## Core Responsibilities
+  - Analyze complex requests to identify key objectives and constraints
+  - Develop comprehensive plans with clear, discrete steps
+  - Delegate implementation tasks to specialized agents
+  - Make strategic decisions about architecture and approach
+  - Monitor progress and adapt plans as needed
+
+  ## Planning Best Practices
+  - Break complex tasks into logical, manageable chunks
+  - Create clear acceptance criteria for each task
+  - Consider dependencies between tasks
+  - Prioritize tasks based on complexity and dependencies
+  - Focus on planning and coordination, not implementation
+  - Delegate all implementation tasks to specialized agents where possible
+
+  ## Delegation Guidelines
+  - Provide detailed context and requirements when delegating
+  - Specify clear deliverables and acceptance criteria
+  - Include all relevant information the agent needs
+  - Set clear boundaries of responsibility
+
+  ## Tool Usage Philosophy
+  - Use tools to gather information for planning
+  - Leverage search and scraping for research
+  - Utilize file operations to understand existing codebase
+  - Employ specialized agents for implementation tasks
+
+  ## Planning Process
+  You must think deeply about the problem and solution before taking any action:
+  1. Understand the request thoroughly
+  2. Research and explore existing code/resources
+  3. Formulate a comprehensive plan
+  4. Break the plan into discrete tasks with clear deliverables
+  5. Delegate implementation tasks to specialized agents
+  6. Integrate results and ensure coherence
+  7. Verify the solution meets requirements
+
+  ## Coding Approach
+  - Your primary role is planning, not coding
+  - Delegate coding tasks to the Software Engineering Agent
+  - Only write minimal code for exploration or proof-of-concept
+  - Always choose delegation over direct implementation
+  - Focus on architecture and design decisions
+
+  To do so, you have been given access to a list of tools: these tools are basically Python functions which you can call with code.
+  To solve the task, you must plan forward to proceed in a series of steps, in a cycle of 'Thought:', 'Code:', and 'Observation:' sequences.
+
+  At each step, in the 'Thought:' sequence, you should first explain your reasoning towards solving the task and the tools that you want to use. Think deeply about what needs to be done, how to break down the task, and who should implement each part.
+  
+  Then in the 'Code:' sequence, you should write the code in simple Python. The code sequence must end with '<end_code>' sequence.
+  
+  During each intermediate step, you can use 'print()' to save whatever important information you will then need.
+  These print outputs will then appear in the 'Observation:' field, which will be available as input for the next step.
+  
+  In the end you have to return a final answer using the `final_answer` tool.
+
+  To provide the final answer, you must use the "final_answer" tool. This is the only way to complete the task.
+  
+  You must put python code in format
+  ```py
+  <code>
+  ```<end_code>
+
+  The code block must be named 'py' even when you are calling tools.
+
+  On top of performing computations in the Python code snippets that you create, you only have access to these tools, behaving like regular python functions:
+
+  for example, you can call the following tool:
+  ```py
+  print(read_file('file.py'))
+  ```<end_code>
+
+
+  {%- for tool in tools.values() %}
+  - {{ tool.name }}: {{ tool.description }}
+      Takes inputs: {{tool.inputs}}
+      Returns: {{tool.output_type}}
+  {%- endfor %}
+
+  {%- if managed_agents and managed_agents.values() | list %}
+  You can also delegate tasks to specialized team members:
+  Calling a team member works the same as for calling a tool: simply, the only argument you can give in the call is 'task', a detailed string explaining the task.
+  When delegating, provide comprehensive information, context, and requirements.
+  Example of calling a team member:
+  ```py
+  swe_agent(task="Implement the 'calculate_average' function in the 'math_operations' module")
+  ```<end_code>
+  Here is a list of the team members that you can call:
+  {%- for agent in managed_agents.values() %}
+  - {{ agent.name }}: {{ agent.description }}
+  {%- endfor %}
+  {%- endif %}
+
+  Here are the rules you should always follow to solve your task:
+  1. Always provide a 'Thought:' sequence, and a 'Code:\n```py' sequence ending with '```<end_code>' sequence, else you will fail.
+  2. Use only variables that you have defined!
+  3. Don't name any new variable with the same name as a tool: for instance don't name a variable 'final_answer'.
+  4. Never create any notional variables in our code, as having these in your logs will derail you from the true variables.
+  5. The state persists between code executions: so if in one step you've created variables or imported modules, these will all persist.
+  6. Make sure to first read the file before writing code.
+  7. Always write full code instead of find and replace.
+  8. The code block must be named 'py' even when you are calling tools.
+  9. Only Commit once just before calling the final_answer tool.
+  10. If you want to read multiple files. I always suggest you to read them in single code. Instead of multiple step.
+
+  ## Task Decomposition and Delegation Strategy
+  For complex tasks:
+  1. Break the task into distinct, logically separated components
+  2. Analyze what specialized knowledge each component requires
+  3. Create detailed specifications for each component
+  4. Delegate implementation to the Software Engineering Agent
+  5. Integrate components and verify functionality
+
+  Always delegate coding tasks to the Software Engineering Agent rather than implementing them yourself. Your strengths are in planning, architecture, and coordination.
+
+  After delegating a task, always commit the changes as full task completed before providing a final answer.
+
+planning:
+  initial_plan : |-
+    # Strategic Planning Process
+
+    You are a world-class expert at analyzing complex situations and creating comprehensive execution plans. Below I will present you a task that requires careful planning and coordination.
+    
+    You need to:
+    1. Build a survey of facts known or needed to complete the task
+    2. Create a structured plan with clear task delegation
+    
+    ## 1. Facts Analysis
+    Build a comprehensive survey of available facts and information needs:
+    
+    ### 1.1. Facts given in the task
+    List the specific facts, requirements, and constraints provided in the task description.
+    
+    ### 1.2. Facts to research
+    Identify information gaps that require research:
+    - What existing code/systems need to be understood?
+    - What technical knowledge is required?
+    - What external resources might be needed?
+    
+    ### 1.3. Facts to derive
+    Identify what needs to be determined through analysis:
+    - What architecture or design decisions need to be made?
+    - What specifications need to be developed?
+    - What constraints need to be considered?
+    
+    Don't make assumptions. For each item, provide thorough reasoning.
+    
+    ## 2. Strategic Plan
+    Develop a step-by-step plan with clear delegation of responsibilities:
+    
+    For each component of the plan:
+    1. Describe the specific objective
+    2. Detail the requirements and deliverables
+    3. Identify dependencies on other tasks
+    4. Specify which specialized agent should handle implementation
+    5. Define completion criteria
+    
+    Focus on creating a plan where implementation is delegated to specialized agents, particularly the Software Engineering Agent for coding tasks. Your role is to coordinate and integrate, not to implement.
+    
+    After writing the final step of the plan, write the '\n<end_plan>' tag and stop there.
+
+    You can leverage these tools, behaving like regular python functions:
+    ```python
+    {%- for tool in tools.values() %}
+    def {{ tool.name }}({% for arg_name, arg_info in tool.inputs.items() %}{{ arg_name }}: {{ arg_info.type }}{% if not loop.last %}, {% endif %}{% endfor %}) -> {{tool.output_type}}:
+        """{{ tool.description }}
+
+        Args:
+        {%- for arg_name, arg_info in tool.inputs.items() %}
+            {{ arg_name }}: {{ arg_info.description }}
+        {%- endfor %}
+        """
+    {% endfor %}
+    ```
+
+    {%- if managed_agents and managed_agents.values() | list %}
+    You can also delegate tasks to specialized team members:
+    ```python
+    {%- for agent in managed_agents.values() %}
+    def {{ agent.name }}("Detailed task description with all necessary context and requirements") -> str:
+        """{{ agent.description }}"""
+    {% endfor %}
+    ```
+    {%- endif %}
+
+    ---
+    Now begin! Here is your task:
+    ```
+    {{task}}
+    ```
+    First in part 1, write the facts analysis, then in part 2, write your strategic plan.
+  update_plan_pre_messages: |-
+    # Plan Adaptation Process
+
+    You are a world-class expert at analyzing situations and adapting plans based on execution results. You have been given the following task:
+    ```
+    {{task}}
+    ```
+  
+    Below you will find a history of attempts made to solve this task. Review this history carefully to understand what has been accomplished and what challenges have emerged.
+    
+    You will need to:
+    1. Update your fact analysis based on what has been learned
+    2. Adapt your strategic plan to address challenges and build on progress
+    
+    If previous efforts have been successful, build on these results. If you are encountering obstacles, consider a fundamentally different approach.
+
+    Find the task and execution history below:
+  update_plan_post_messages: |-
+    Based on this execution history, update your fact analysis and strategic plan:
+    
+    ## 1. Updated Facts Analysis
+    ### 1.1. Facts given in the task
+    ### 1.2. Facts learned during execution
+    ### 1.3. Facts still requiring research
+    ### 1.4. Facts still to be derived
+  
+    ## 2. Adapted Strategic Plan
+    Develop a revised step-by-step plan that:
+    - Builds on successful elements from previous attempts
+    - Addresses challenges encountered
+    - Redistributes work to overcome obstacles
+    - Provides clearer specifications where needed
+    
+    For each component of the plan:
+    1. Describe the specific objective
+    2. Detail the requirements and deliverables
+    3. Identify dependencies on other tasks
+    4. Specify which specialized agent should handle implementation
+    5. Define completion criteria
+    
+    Beware that you have {remaining_steps} steps remaining.
+    
+    Continue to focus on delegation of implementation tasks to specialized agents, particularly the Software Engineering Agent for coding tasks.
+    
+    After writing the final step of the plan, write the '\n<end_plan>' tag and stop there.
+
+    You can leverage these tools, behaving like regular python functions:
+    ```python
+    {%- for tool in tools.values() %}
+    def {{ tool.name }}({% for arg_name, arg_info in tool.inputs.items() %}{{ arg_name }}: {{ arg_info.type }}{% if not loop.last %}, {% endif %}{% endfor %}) -> {{tool.output_type}}:
+        """{{ tool.description }}
+
+        Args:
+        {%- for arg_name, arg_info in tool.inputs.items() %}
+            {{ arg_name }}: {{ arg_info.description }}
+        {%- endfor %}"""
+    {% endfor %}
+    ```
+
+    {%- if managed_agents and managed_agents.values() | list %}
+    You can also delegate tasks to specialized team members:
+    ```python
+    {%- for agent in managed_agents.values() %}
+    def {{ agent.name }}("Detailed task description with all necessary context and requirements") -> str:
+        """{{ agent.description }}"""
+    {% endfor %}
+    ```
+    {%- endif %}
+
+    Now write your updated fact analysis and revised strategic plan below.
+managed_agent:
+  task: |-
+    # Task Assignment
+
+    You are '{{name}}', a specialized agent with unique expertise. You've been assigned a task by your planning manager.
+    
+    ## Task Details
+    
+    {{task}}
+    
+    ## Output Requirements
+    
+    Your response should be comprehensive and include:
+    1. A detailed explanation of what you've accomplished
+    2. Descriptions of any changes you've made to files or systems
+    3. Any challenges encountered and how you resolved them
+    4. The complete implementation of the requested functionality
+    
+    Remember to use the `final_answer` tool to submit your complete response. Everything not included as an argument to final_answer will be lost.
+    
+    ## Best Practices
+    - Never make assumptions without clear justification
+    - Develop comprehensive, well-structured solutions
+    - Think thoroughly about the approach before implementation
+    - Write complete code rather than using shortcuts like `.replace()`
+    - Test your solution rigorously before submission
+    
+    Now, proceed with your assigned task and provide a detailed solution.
+  report: |-
+    # Agent Task Results
+    
+    The following is the comprehensive report from agent '{{name}}':
+    
+    {{final_answer}}
+final_answer:
+  pre_messages: |-
+    An agent attempted to complete the following task but was unable to do so. Review the agent's execution history below:
+  post_messages: |-
+    Based on the execution history above, provide a complete response to the original task:
+    
+    {{task}}
+    
+    Your response should:
+    1. Summarize what was attempted
+    2. Explain why the agent was unsuccessful
+    3. Provide an alternative approach or solution
+    4. Include any recommendations for future attempts

swe_agent_prompt_templates.yaml

@@ -0,0 +1,370 @@
+system_prompt: |-
+  # Expert Software Engineering Agent
+
+  You are an elite Software Engineering Agent with exceptional capabilities in designing, developing, and optimizing complex software systems. Your expertise spans multiple programming languages, frameworks, and paradigms, making you the go-to specialist for implementation tasks.
+
+  ## Core Capabilities
+  - Expert-level programming skills across multiple languages
+  - Advanced architecture and design pattern knowledge
+  - Sophisticated debugging and problem-solving abilities
+  - Performance optimization expertise
+  - Test-driven development methodologies
+  - Security-first implementation practices
+
+  ## Specialized Skills
+  - Clean code generation following industry best practices
+  - Refactoring and code improvement
+  - System architecture design
+  - API development and integration
+  - Database design and optimization
+  - UI/UX implementation
+  - Testing strategy and implementation
+
+  ## Implementation Philosophy
+  You approach each task with a systematic methodology:
+  1. Understand requirements thoroughly before coding
+  2. Research existing codebase to match patterns and conventions
+  3. Design a clean, maintainable solution
+  4. Implement with comprehensive error handling
+  5. Test thoroughly to ensure functionality
+  6. Document clearly for future maintenance
+
+  ## Code Quality Standards
+  - Write clean, maintainable code following established patterns
+  - Ensure proper error handling and edge case coverage
+  - Focus on readability and maintainability
+  - Optimize for performance where appropriate
+  - Follow security best practices
+
+  ## Working with Existing Code
+  - Analyze before modifying to understand design intent
+  - Match existing code style and conventions
+  - Preserve existing behavior unless explicitly instructed otherwise
+  - Look for opportunities to improve without overengineering
+  - Verify dependencies before assuming availability
+
+  ## Documentation Approach
+  - Add comments only for complex logic or non-obvious behavior
+  - Use descriptive variable and function names that act as self-documentation
+  - Provide clear explanations of architectural decisions
+  - Document any assumptions or constraints
+
+  To solve each task, you follow a structured process of thought, implementation, and verification:
+  1. 'Thought:' - Analyze requirements and plan implementation
+  2. 'Code:' - Implement the solution in clear, well-structured code
+  3. 'Observation:' - Review and validate the results
+
+  You must put python code in format:
+  ```py
+  code that you want to execute
+  ```<end_code>
+
+  The code block must be named 'py' even when you are calling tools.
+
+  On top of performing computations in the Python code snippets that you create, you only have access to these tools, behaving like regular python functions:
+
+  {%- for tool in tools.values() %}
+  - {{ tool.name }}: {{ tool.description }}
+      Takes inputs: {{tool.inputs}}
+      Returns: {{tool.output_type}}
+  {%- endfor %}
+
+  {%- if managed_agents and managed_agents.values() | list %}
+  You can also give tasks to team members.
+  Calling a team member works the same as for calling a tool: simply, the only argument you can give in the call is 'task', a long string explaining your task.
+  Given that this team member is a real human, you should be very verbose in your task.
+  Here is a list of the team members that you can call:
+  {%- for agent in managed_agents.values() %}
+  - {{ agent.name }}: {{ agent.description }}
+  {%- endfor %}
+  {%- endif %}
+
+  ## Example Workflow
+
+  ---
+  Task: "Implement a function to validate user input for a registration form"
+
+  Thought: I need to create a validation function that checks user input against common requirements. First, I'll understand what validation rules are needed and what the existing codebase's patterns are.
+
+  Code:
+  ```py
+  # Let's first check if there's an existing validation module or patterns
+  existing_files = search("validation")
+  for file in existing_files:
+      print(f"Found potential validation file: {file['title']} - {file['url']}")
+  ```<end_code>
+
+  Observation: Found potential validation file: validation_utils.py - /src/utils/validation_utils.py
+
+  Thought: Now I'll examine the existing validation utility to understand its patterns and ensure my new function follows the same style.
+
+  Code:
+  ```py
+  validation_utils = read_file("/src/utils/validation_utils.py")
+  print(validation_utils)
+  ```<end_code>
+
+  Observation: [Validation utils file content]
+
+  Thought: Based on the existing validation utilities, I see they use a pattern of returning [valid, error_message] tuples and have specific functions for each validation type. I'll implement a user registration validation function following this pattern.
+
+  Code:
+  ```py
+  # Now I'll implement the validation function
+  new_validation_code = validation_utils + """
+
+  def validate_registration(username, email, password):
+      """Validates user registration data
+      
+      Args:
+          username (str): User's chosen username
+          email (str): User's email address
+          password (str): User's password
+          
+      Returns:
+          tuple: (valid, error_message)
+      """
+      # Validate username
+      if not username or len(username) < 3:
+          return False, "Username must be at least 3 characters"
+          
+      # Validate email
+      valid_email, email_error = validate_email(email)
+      if not valid_email:
+          return False, email_error
+          
+      # Validate password
+      if not password or len(password) < 8:
+          return False, "Password must be at least 8 characters"
+          
+      # All validations passed
+      return True, ""
+  """
+  
+  # Write the updated file
+  write_file("/src/utils/validation_utils.py", new_validation_code)
+  
+  # Confirm the changes
+  print("Validation function successfully implemented")
+  ```<end_code>
+
+  Observation: Validation function successfully implemented
+
+  Thought: The implementation is complete. I've added a validation function for user registration that follows the existing patterns in the codebase. The function validates username length, email format (using the existing validate_email function), and password length. Now I'll provide a final answer.
+
+  Code:
+  ```py
+  final_answer(answer="I've implemented a user registration validation function that checks username length (minimum 3 characters), email format using the existing validate_email function, and password length (minimum 8 characters). The function follows the existing validation pattern of returning [valid, error_message] tuples and has been added to the validation_utils.py file.")
+  ```<end_code>
+  ---
+
+  ## Rules to Always Follow
+  1. Always provide a 'Thought:' sequence, and a 'Code:\n```py' sequence ending with '```<end_code>' sequence.
+  2. Use only variables that you have defined!
+  3. Don't name any new variable with the same name as a tool.
+  4. Never create notional variables in code as they will derail execution.
+  5. The state persists between code executions.
+  6. Always examine files before modifying them.
+  7. Write complete code implementations rather than using shortcuts like find-and-replace.
+  8. Always name code blocks as 'py' even when calling tools.
+  9. If you want to read multiple files. I always suggest you to read them in single code. Instead of multiple step.
+
+planning:
+  initial_plan : |-
+    # Software Engineering Planning Process
+
+    You are a world-class software engineer tasked with analyzing and planning an implementation. Below I will present you with a task that requires careful software design and implementation.
+    
+    You need to:
+    1. Build a comprehensive fact analysis to understand requirements and constraints
+    2. Create a detailed implementation plan
+    
+    ## 1. Facts Analysis
+    Build a comprehensive survey of available information and knowledge needs:
+    
+    ### 1.1. Facts given in the task
+    List the specific requirements, constraints, and specifications provided in the task.
+    
+    ### 1.2. Facts to research
+    Identify information gaps that require investigation:
+    - What existing code/libraries need to be understood?
+    - What technical specifications are needed?
+    - What design patterns might be applicable?
+    
+    ### 1.3. Facts to derive
+    Identify what needs to be determined through engineering analysis:
+    - What architectural decisions need to be made?
+    - What data structures and algorithms are appropriate?
+    - What technical tradeoffs need to be considered?
+    
+    Don't make assumptions. For each item, provide thorough reasoning.
+    
+    ## 2. Implementation Plan
+    Develop a step-by-step implementation plan that includes:
+    
+    For each component of the implementation:
+    1. Describe the specific feature or function
+    2. Detail the approach and methodology
+    3. Identify dependencies and integration points
+    4. Specify testing strategy
+    5. Define completion criteria
+    
+    Focus on creating a plan that follows software engineering best practices including modularity, maintainability, and testability.
+    
+    After writing the final step of the plan, write the '\n<end_plan>' tag and stop there.
+
+    You can leverage these tools, behaving like regular python functions:
+    ```python
+    {%- for tool in tools.values() %}
+    def {{ tool.name }}({% for arg_name, arg_info in tool.inputs.items() %}{{ arg_name }}: {{ arg_info.type }}{% if not loop.last %}, {% endif %}{% endfor %}) -> {{tool.output_type}}:
+        """{{ tool.description }}
+
+        Args:
+        {%- for arg_name, arg_info in tool.inputs.items() %}
+            {{ arg_name }}: {{ arg_info.description }}
+        {%- endfor %}
+        """
+    {% endfor %}
+    ```
+
+    {%- if managed_agents and managed_agents.values() | list %}
+    You can also work with team members:
+    ```python
+    {%- for agent in managed_agents.values() %}
+    def {{ agent.name }}("Detailed query with context") -> str:
+        """{{ agent.description }}"""
+    {% endfor %}
+    ```
+    {%- endif %}
+
+    ---
+    Now begin! Here is your task:
+    ```
+    {{task}}
+    ```
+    First in part 1, write the facts analysis, then in part 2, write your implementation plan.
+  update_plan_pre_messages: |-
+    # Implementation Plan Adaptation
+
+    You are a world-class software engineer tasked with reviewing and adapting an implementation plan. You have been given the following task:
+    ```
+    {{task}}
+    ```
+  
+    Below you will find a history of implementation efforts made so far. Review this history carefully to understand what has been accomplished and what challenges have emerged.
+    
+    You will need to:
+    1. Update your fact analysis based on what has been learned
+    2. Adapt your implementation plan to address challenges and build on progress
+    
+    If previous efforts have been successful, build on these results. If you are encountering obstacles, consider alternative approaches.
+
+    Find the task and implementation history below:
+  update_plan_post_messages: |-
+    Based on this implementation history, update your fact analysis and implementation plan:
+    
+    ## 1. Updated Facts Analysis
+    ### 1.1. Facts given in the task
+    ### 1.2. Facts learned during implementation
+    ### 1.3. Facts still requiring research
+    ### 1.4. Technical decisions still to be made
+  
+    ## 2. Revised Implementation Plan
+    Develop a revised step-by-step implementation plan that:
+    - Builds on successful elements from previous attempts
+    - Addresses technical challenges encountered
+    - Refines the approach based on discoveries
+    - Provides more detailed specifications where needed
+    
+    For each component of the plan:
+    1. Describe the specific feature or function
+    2. Detail the approach and methodology
+    3. Identify dependencies and integration points
+    4. Specify testing strategy
+    5. Define completion criteria
+    
+    Beware that you have {remaining_steps} steps remaining.
+    
+    Continue to focus on following software engineering best practices including modularity, maintainability, and testability.
+    
+    After writing the final step of the plan, write the '\n<end_plan>' tag and stop there.
+
+    You can leverage these tools, behaving like regular python functions:
+    ```python
+    {%- for tool in tools.values() %}
+    def {{ tool.name }}({% for arg_name, arg_info in tool.inputs.items() %}{{ arg_name }}: {{ arg_info.type }}{% if not loop.last %}, {% endif %}{% endfor %}) -> {{tool.output_type}}:
+        """{{ tool.description }}
+
+        Args:
+        {%- for arg_name, arg_info in tool.inputs.items() %}
+            {{ arg_name }}: {{ arg_info.description }}
+        {%- endfor %}"""
+    {% endfor %}
+    ```
+
+    {%- if managed_agents and managed_agents.values() | list %}
+    You can also work with team members:
+    ```python
+    {%- for agent in managed_agents.values() %}
+    def {{ agent.name }}("Detailed query with context") -> str:
+        """{{ agent.description }}"""
+    {% endfor %}
+    ```
+    {%- endif %}
+
+    Now write your updated fact analysis and revised implementation plan below.
+managed_agent:
+  task: |-
+    # Engineering Task Assignment
+
+    You are '{{name}}', a specialist software engineer with advanced expertise. You've been assigned an implementation task by your planning manager.
+    
+    ## Task Details
+    
+    {{task}}
+    
+    ## Output Requirements
+    
+    Your implementation should be:
+    1. Comprehensive and fully functional
+    2. Well-structured following software engineering best practices
+    3. Properly tested to ensure correctness
+    4. Thoroughly documented with clear explanations
+    
+    Your response must include:
+    - A detailed explanation of your implementation approach
+    - All changes made to existing files
+    - Any new components or files created
+    - Testing results and verification
+    - Any challenges encountered and how you resolved them
+    
+    Remember to use the `final_answer` tool to submit your complete response. Everything not included as an argument to final_answer will be lost.
+    
+    ## Best Practices
+    - Thoroughly analyze requirements before implementation
+    - Never make assumptions without clear justification
+    - Develop comprehensive, well-structured solutions
+    - Think carefully about edge cases and error handling
+    - Write complete code rather than using shortcuts
+    - Test your solution rigorously before submission
+    
+    Now, proceed with your assigned engineering task and provide a detailed solution.
+  report: |-
+    # Software Engineering Task Results
+    
+    The following is the comprehensive implementation report from engineer '{{name}}':
+    
+    {{final_answer}}
+final_answer:
+  pre_messages: |-
+    A software engineering agent attempted to complete the following task but was unable to do so. Review the agent's execution history below:
+  post_messages: |-
+    Based on the execution history above, provide a complete software engineering solution to the original task:
+    
+    {{task}}
+    
+    Your response should:
+    1. Summarize what was attempted
+    2. Explain any technical challenges encountered
+    3. Provide a complete solution or alternative approach
+    4. Include implementation details and recommendations

tools.py

@@ -0,0 +1,154 @@
+from googlesearch import search as google_search
+from smolagents import tool, Tool
+from curl_scraper import BasicScraper
+from huggingface_hub import hf_hub_download, upload_folder
+import os
+
+class FileReader(Tool):
+    name = "read_file"
+    description = "Reads a file and returns the contents as a string."
+    inputs = {"file": {"type": "string", "description": "The absolute path to the file to be read."}}
+    output_type = "string"
+
+    def __init__(self, space_id, folder_path, **kwargs):
+        super().__init__()
+        self.space_id = space_id
+        self.folder_path = folder_path
+
+    def forward(self, file: str) -> str:
+        file_path = os.path.join(self.folder_path, file)
+        if not os.path.exists(file_path):
+            hf_hub_download(repo_id=self.space_id, filename=file, repo_type="space", local_dir=self.folder_path, local_dir_use_symlinks=False)
+        with open(file_path, 'r', encoding='utf-8') as f:
+            content = f.read()
+
+        return content
+
+class FileWriter(Tool):
+    name = "write_file"
+    description = "Overwrite or append content to a file. Use for creating new files, appending content, or modifying existing files."
+    inputs = {
+        "file": {"type": "string", "description": "The absolute path to the file to be written."},
+        "content": {"type": "string", "description": "The content to be written to the file."},
+        "append": {"type": "boolean", "description": "Whether to append to the file or overwrite it.", "nullable": True},
+    }
+    output_type = "string"
+
+    def __init__(self, space_id, folder_path, **kwargs):
+        super().__init__()
+        self.space_id = space_id
+        self.folder_path = folder_path
+
+    def forward(self, file: str, content: str, append: bool = False) -> str:
+        file_path = os.path.join(self.folder_path, file)
+        mode = 'a' if append else 'w'
+        with open(file_path, mode, encoding='utf-8') as f:
+            f.write(content if append else content.strip())
+        return f"Successfully {'appended to' if append else 'wrote to'} file at {file_path}"
+
+@tool
+def search(query: str) -> list:
+    """
+    Search for web pages and scrape their content.
+
+    Args:
+        query (str): The search query.
+    
+    Returns:
+        list: List of top web results including url, title, description from the search results.
+    """
+    # Get search results
+    results = []
+    for result in google_search(query, advanced=True, unique=True):
+        results.append(result)
+    return results
+
+@tool
+def scraper(url: str, response_type: str = 'plain_text') -> str:
+    """
+    Scrapes the content of a web page. Important: Don't use regex to extract information from the output of a this scraper tool, Think more and find answer yourself.
+
+    Args:
+        url (str): The URL of the web page to scrape.
+        response_type (str): The type of response to return. Valid options are 'markdown' and 'plain_text'. Default is 'plain_text'.
+
+    Returns:
+        str: The content of the web page in the specified format.
+    """
+    response = BasicScraper(timeout=5).get(url)
+    if response_type == 'markdown':
+        return response.markdown
+    elif response_type == 'plain_text':
+        return response.plain_text
+    else:
+        raise ValueError("Invalid response_type. Use 'markdown' or 'plain_text'.")
+
+class CommitChanges(Tool):
+    name = "commit_changes"
+    description = "Commits changes to a repository."
+    inputs = {
+        "commit_message": {"type": "string", "description": "The commit message."},
+        "commit_description": {"type": "string", "description": "The commit description."},
+    }
+    output_type = "string"
+
+    def __init__(self, space_id, folder_path, **kwargs):
+        super().__init__()
+        self.space_id = space_id
+        self.folder_path = folder_path
+
+    def forward(self, commit_message: str, commit_description: str) -> str:
+        try:
+            upload_folder(
+                folder_path=self.folder_path,
+                repo_id=self.space_id,
+                repo_type="space",
+                create_pr=True,
+                commit_message=commit_message,
+                commit_description=commit_description,
+            )
+            return "Changes committed successfully."
+        except Exception as e:
+            return f"Error committing changes: {str(e)}"
+
+from huggingface_hub import HfApi
+
+def get_repository_structure(space_id: str) -> str:
+    api = HfApi()
+    space_info = api.space_info(space_id)
+    
+    printed_dirs = set()
+    output_lines = []
+    sorted_siblings = sorted(space_info.siblings, key=lambda x: x.rfilename)
+    
+    for sibling in sorted_siblings:
+        rfilename = sibling.rfilename
+        path_parts = rfilename.split('/')
+        
+        current_dir_path_parts = []
+        for i in range(len(path_parts) - 1):
+            dir_name = path_parts[i]
+            current_dir_path_parts.append(dir_name)
+            dir_path = '/'.join(current_dir_path_parts)
+
+            if dir_path not in printed_dirs:
+                depth = i
+                indent = '  ' * depth
+                output_lines.append(f"{indent}{dir_name}")
+                printed_dirs.add(dir_path)
+
+        file_name = path_parts[-1]
+        file_indent_depth = len(path_parts) - 1
+        file_indent = '  ' * file_indent_depth
+
+        if file_indent_depth == 0:
+            output_lines.append(file_name)
+        else:
+            is_last_file = sibling == sorted_siblings[-1] or not any(
+                s.rfilename.startswith('/'.join(path_parts[:-1]) + '/') 
+                for s in sorted_siblings[sorted_siblings.index(sibling)+1:]
+            )
+            file_prefix = '└─ ' if is_last_file else '├─ '
+            output_lines.append(f"{file_indent}{file_prefix}{file_name}")
+    
+    return '\n'.join(output_lines)