Update app.py

app.py

@@ -1,84 +1,47 @@
+from __future__ import annotations
+from typing import List, Dict, Union
 import os
+import re
 import subprocess
-from huggingface_hub import InferenceClient
-import gradio as gr
-import random
-import prompts
+import streamlit as st
 import time
-from typing import List, Dict
-from .dictionary import corpora, models
-from gensim.summarization import summarize
-
-# Simulated agent and tool libraries
-AGENT_TYPES = [
-    "Task Executor",
-    "Information Retriever",
-    "Decision Maker",
-    "Data Analyzer",
-]
-TOOL_TYPES = [
-    "Web Scraper",
-    "Database Connector",
-    "API Caller",
-    "File Handler",
-    "Text Processor",
-]
-
-# Initialize Hugging Face client
-client = InferenceClient("mistralai/Mixtral-8x7B-Instruct-v0.1")
-
-VERBOSE = False
-MAX_HISTORY = 100
-MODEL = "mistralai/Mixtral-8x7B-Instruct-v0.1"
-
-# Import necessary prompts and functions from the existing code
-from prompts import (
-    ACTION_PROMPT,
-    ADD_PROMPT,
-    COMPRESS_HISTORY_PROMPT,
-    LOG_PROMPT,
-    LOG_RESPONSE,
-    MODIFY_PROMPT,
-    PREFIX,
-    READ_PROMPT,
-    TASK_PROMPT,
-    UNDERSTAND_TEST_RESULTS_PROMPT,
-)
-from .utils import parse_action, parse_file_content, read_python_module_structure
-from flask import Flask, request, jsonify
+from langchain_core.prompts import PromptTemplate
+from langchain_community.llms import HuggingFaceEndpoint
+from huggingface_hub.inference_api import InferenceApi as InferenceClient
 
+# Load LLM
+llm = HuggingFaceHub(repo_id="tiiuae/falcon-7b-instruct", model_kwargs={"temperature": 0.1, "max_new_tokens": 500})
 
 class Agent:
     def __init__(self, name: str, agent_type: str, complexity: int):
-        self.name = name
-        self.type = agent_type
-        self.complexity = complexity
-        self.tools = []
+        self.name: str = name
+        self.agent_type: str = agent_type
+        self.complexity: int = complexity
+        self.tools: List[Tool] = []
 
-    def add_tool(self, tool):
+    def add_tool(self, tool: Tool):
         self.tools.append(tool)
 
     def __str__(self):
-        return f"{self.name} ({self.type}) - Complexity: {self.complexity}"
-
+        return f"{self.name} ({self.agent_type}) - Complexity: {self.complexity}"
 
 class Tool:
     def __init__(self, name: str, tool_type: str):
-        self.name = name
-        self.type = tool_type
+        self.name: str = name
+        self.tool_type: str = tool_type
 
     def __str__(self):
-        return f"{self.name} ({self.type})"
-
+        return f"{self.name} ({self.tool_type})"
 
 class Pypelyne:
     def __init__(self):
         self.agents: List[Agent] = []
         self.tools: List[Tool] = []
-        self.history = ""
-        self.task = None
-        self.purpose = None
-        self.directory = None
+        self.history: str = ""
+        self.task: str = ""
+        self.purpose: str = ""
+        self.directory: str = ""
+        self.task_queue: list = []
 
     def add_agent(self, agent: Agent):
         self.agents.append(agent)
@@ -86,28 +49,29 @@ class Pypelyne:
     def add_tool(self, tool: Tool):
         self.tools.append(tool)
 
-    def generate_chat_app(self):
-        time.sleep(2)  # Simulate processing time
+    def generate_chat_app(self) -> str:
+        time.sleep(2)
         return f"Chat app generated with {len(self.agents)} agents and {len(self.tools)} tools."
 
-    def run_gpt(self, prompt_template, stop_tokens, max_tokens, **prompt_kwargs):
-        content = PREFIX.format(
-            module_summary=read_python_module_structure(self.directory)[0],
-            purpose=self.purpose,
-        ) + prompt_template.format(**prompt_kwargs)
+    def run_gpt(
+        self,
+        prompt_template: PromptTemplate,
+        stop_tokens: List[str],
+        max_tokens: int,
+        **prompt_kwargs,
+    ) -> str:
+        content = f"""{PREFIX}
+{prompt_template.format(**prompt_kwargs)}"""
 
         if VERBOSE:
             print(LOG_PROMPT.format(content))
 
-        stream = client.text_generation(
-            prompt=content,
-            max_new_tokens=max_tokens,
-            stop_sequences=stop_tokens if stop_tokens else None,
-            do_sample=True,
-            temperature=0.7,
-        )
-
-        resp = "".join(token for token in stream)
+        try:
+            stream = llm.predict(content)
+            resp = "".join(stream)
+        except Exception as e:
+            print(f"Error in run_gpt: {e}")
+            resp = f"Error: {e}"
 
         if VERBOSE:
             print(LOG_RESPONSE.format(resp))
@@ -123,37 +87,29 @@ class Pypelyne:
         )
         self.history = f"observation: {resp}\n"
 
-    def run_action(self, action_name, action_input):
+    def run_action(self, action_name: str, action_input: Union[str, List[str]], tools: List[Tool] = None) -> str:
         if action_name == "COMPLETE":
             return "Task completed."
 
         if len(self.history.split("\n")) > MAX_HISTORY:
-            if VERBOSE:
-                print("COMPRESSING HISTORY")
             self.compress_history()
 
-        action_funcs = {
-            "MAIN": self.call_main,
-            "UPDATE-TASK": self.call_set_task,
-            "MODIFY-FILE": self.call_modify,
-            "READ-FILE": self.call_read,
-            "ADD-FILE": self.call_add,
-            "TEST": self.call_test,
-        }
+        if action_name not in self.task_queue:
+            self.task_queue.append(action_name)
 
-        if action_name not in action_funcs:
-            return f"Unknown action: {action_name}"
+        task_function = getattr(self, f"call_{action_name.lower()}")
+        result = task_function(action_input, tools)
+        self.task_queue.pop(0)
+        return result
 
-        print(f"RUN: {action_name} {action_input}")
-        return action_funcs[action_name](action_input)
-
-    def call_main(self, action_input):
+    def call_main(self, action_input: List[str]) -> str:
         resp = self.run_gpt(
-            ACTION_PROMPT,
+            f"{ACTION_PROMPT}",
             stop_tokens=["observation:", "task:"],
             max_tokens=256,
             task=self.task,
             history=self.history,
+            actions=action_input,
         )
         lines = resp.strip().strip("\n").split("\n")
         for line in lines:
@@ -161,276 +117,134 @@ class Pypelyne:
                 continue
             if line.startswith("thought: "):
                 self.history += f"{line}\n"
-            elif line.startswith("action: "):
                 action_name, action_input = parse_action(line)
-                self.history += f"{line}\n"
-                return self.run_action(action_name, action_input)
+                self.run_action(action_name, action_input)
         return "No valid action found."
 
-    def call_set_task(self, action_input):
-        self.task = self.run_gpt(
-            TASK_PROMPT,
-            stop_tokens=[],
-            max_tokens=64,
-            task=self.task,
-            history=self.history,
-        ).strip("\n")
-        self.history += f"observation: task has been updated to: {self.task}\n"
+    def call_set_task(self, action_input: str) -> str:
+        self.task = action_input
         return f"Task updated: {self.task}"
 
-    def call_modify(self, action_input):
-        if not os.path.exists(action_input):
-            self.history += "observation: file does not exist\n"
-            return "File does not exist."
-
-        content = read_python_module_structure(self.directory)[1]
-        f_content = (
-            content[action_input] if content[action_input] else "< document is empty >"
-        )
+    def call_modify(self, action_input: str, agent: Agent) -> str:
+        with open(action_input, "r") as file:
+            file_content = file.read()
 
         resp = self.run_gpt(
-            MODIFY_PROMPT,
+            f"{MODIFY_PROMPT}",
             stop_tokens=["action:", "thought:", "observation:"],
             max_tokens=2048,
             task=self.task,
             history=self.history,
             file_path=action_input,
-            file_contents=f_content,
+            file_contents=file_content,
+            agent=agent,
         )
-        new_contents, description = parse_file_content(resp)
-        if new_contents is None:
-            self.history += "observation: failed to modify file\n"
-            return "Failed to modify file."
+        new_contents = resp.strip()
 
-        with open(action_input, "w") as f:
-            f.write(new_contents)
+        with open(action_input, "w") as file:
+            file.write(new_contents)
 
         self.history += f"observation: file successfully modified\n"
-        self.history += f"observation: {description}\n"
         return f"File modified: {action_input}"
 
-    def call_read(self, action_input):
-        if not os.path.exists(action_input):
-            self.history += "observation: file does not exist\n"
-            return "File does not exist."
+    def call_read(self, action_input: str) -> str:
+        with open(action_input, "r") as file:
+            file_content = file.read()
 
-        content = read_python_module_structure(self.directory)[1]
-        f_content = (
-            content[action_input] if content[action_input] else "< document is empty >"
-        )
+        self.history += f"observation: {file_content}\n"
+        return file_content
 
-        resp = self.run_gpt(
-            READ_PROMPT,
-            stop_tokens=[],
-            max_tokens=256,
-            task=self.task,
-            history=self.history,
-            file_path=action_input,
-            file_contents=f_content,
-        ).strip("\n")
-        self.history += f"observation: {resp}\n"
-        return f"File read: {action_input}"
-
-    def call_add(self, action_input):
-        d = os.path.dirname(action_input)
-        if not d.startswith(self.directory):
-            self.history += (
-                f"observation: files must be under directory {self.directory}\n"
-            )
-            return f"Invalid directory: {d}"
-        elif not action_input.endswith(".py"):
-            self.history += "observation: can only write .py files\n"
-            return "Only .py files are allowed."
-        else:
-            if d and not os.path.exists(d):
-                os.makedirs(d)
-            if not os.path.exists(action_input):
-                resp = self.run_gpt(
-                    ADD_PROMPT,
-                    stop_tokens=["action:", "thought:", "observation:"],
-                    max_tokens=2048,
-                    task=self.task,
-                    history=self.history,
-                    file_path=action_input,
-                )
-                new_contents, description = parse_file_content(resp)
-                if new_contents is None:
-                    self.history += "observation: failed to write file\n"
-                    return "Failed to write file."
-
-                with open(action_input, "w") as f:
-                    f.write(new_contents)
-
-                self.history += "observation: file successfully written\n"
-                self.history += f"observation: {description}\n"
-                return f"File added: {action_input}"
-            else:
-                self.history += "observation: file already exists\n"
-                return "File already exists."
-
-    def call_test(self, action_input):
-        result = subprocess.run(
-            ["python", "-m", "pytest", "--collect-only", self.directory],
-            capture_output=True,
-            text=True,
-        )
-        if result.returncode != 0:
-            self.history += f"observation: there are no tests! Test should be written in a test folder under {self.directory}\n"
-            return "No tests found."
-        result = subprocess.run(
-            ["python", "-m", "pytest", self.directory], capture_output=True, text=True
-        )
-        if result.returncode == 0:
-            self.history += "observation: tests pass\n"
-            return "All tests passed."
+    def call_add(self, action_input: str) -> str:
+        if not os.path.exists(self.directory):
+            os.makedirs(self.directory)
 
-        resp = self.run_gpt(
-            UNDERSTAND_TEST_RESULTS_PROMPT,
-            stop_tokens=[],
-            max_tokens=256,
-            task=self.task,
-            history=self.history,
-            stdout=result.stdout[:5000],
-            stderr=result.stderr[:5000],
-        )
-        self.history += f"observation: tests failed: {resp}\n"
-        return f"Tests failed: {resp}"
+        with open(os.path.join(self.directory, action_input), "w") as file:
+            file.write("")
 
+        self.history += f"observation: file created: {action_input}\n"
+        return f"File created: {action_input}"
 
-pypelyne = Pypelyne()
+    def call_test(self, action_input: str) -> str:
+        result = subprocess.run(["python", os.path.join(self.directory, action_input)], capture_output=True, text=True)
+        error_message = result.stderr.strip()
 
+        self.history += f"observation: tests {('passed' if error_message == '' else 'failed')}\n"
+        return f"Tests {'passed' if error_message == '' else 'failed'}: {error_message}"
 
-def create_agent(name: str, agent_type: str, complexity: int) -> str:
+# Global Pypelyne Instance
+pypelyne = Pypelyne()
+
+# Helper Functions
+def create_agent(name: str, agent_type: str, complexity: int) -> Agent:
     agent = Agent(name, agent_type, complexity)
     pypelyne.add_agent(agent)
-    return f"Agent created: {agent}"
+    return agent
 
-def create_tool(name: str, tool_type: str) -> str:
+def create_tool(name: str, tool_type: str) -> Tool:
     tool = Tool(name, tool_type)
     pypelyne.add_tool(tool)
-    return f"Tool created: {tool}"
-
-def assign_tool(agent_name: str, tool_name: str) -> str:
-    agent = next((a for a in pypelyne.agents if a.name == agent_name), None)
-    tool = next((t for t in pypelyne.tools if t.name == tool_name), None)
+    return tool
 
-    if agent and tool:
-        agent.add_tool(tool)
-        return f"Tool '{tool.name}' assigned to agent '{agent.name}'"
-    else:
-        return "Agent or tool not found."
+# Streamlit App Code
+def main():
+    st.title("🧠 Pypelyne: Your AI-Powered Coding Assistant")
 
-def generate_chat_app() -> str:
-    return pypelyne.generate_chat_app()
-
-def list_agents() -> str:
-    return (
-        "\n".join(str(agent) for agent in pypelyne.agents) or "No agents created yet."
+    # Settings
+    st.sidebar.title("⚙️ Settings")
+    directory = st.sidebar.text_input(
+        "Project Directory:", value=pypelyne.directory, help="Path to your coding project"
     )
-
-def list_tools() -> str:
-    return "\n".join(str(tool) for tool in pypelyne.tools) or "No tools created yet."
-
-def chat_with_pypelyne(message: str) -> str:
-    return pypelyne.run_action("MAIN", message)
-
-def set_purpose_and_directory(purpose: str, directory: str) -> str:
-    pypelyne.purpose = purpose
     pypelyne.directory = directory
-    return f"Purpose set to: {purpose}\nWorking directory set to: {directory}"
-
-with gr.Blocks() as app:
-    gr.Markdown("# Welcome to Pypelyne")
-    gr.Markdown("Create your custom pipeline with agents and tools, then chat with it!")
-
-    with gr.Tab("Setup"):
-        purpose_input = gr.Textbox(label="Set Purpose")
-        directory_input = gr.Textbox(label="Set Working Directory")
-        setup_btn = gr.Button("Set Purpose and Directory")
-        setup_output = gr.Textbox(label="Setup Output")
-        setup_btn.click(
-            set_purpose_and_directory,
-            inputs=[purpose_input, directory_input],
-            outputs=setup_output,
-        )
 
-    with gr.Tab("Create Agents"):
-        agent_name = gr.Textbox(label="Agent Name")
-        agent_type = gr.Dropdown(choices=AGENT_TYPES, label="Agent Type")
-        agent_complexity = gr.Slider(
-            minimum=1, maximum=10, step=1, label="Agent Complexity"
-        )
-        create_agent_btn = gr.Button("Create Agent")
-        agent_output = gr.Textbox(label="Output")
-        create_agent_btn.click(
-            create_agent,
-            inputs=[agent_name, agent_type, agent_complexity],
-            outputs=agent_output,
-        )
-
-    with gr.Tab("Create Tools"):
-        tool_name = gr.Textbox(label="Tool Name")
-        tool_type = gr.Dropdown(choices=TOOL_TYPES, label="Tool Type")
-        create_tool_btn = gr.Button("Create Tool")
-        tool_output = gr.Textbox(label="Output")
-        create_tool_btn.click(
-            create_tool, inputs=[tool_name, tool_type], outputs=tool_output
-        )
-
-    with gr.Tab("Assign Tools"):
-        agent_select = gr.Dropdown(choices=[], label="Select Agent")
-        tool_select = gr.Dropdown(choices=[], label="Select Tool")
-        assign_tool_btn = gr.Button("Assign Tool")
-        assign_output = gr.Textbox(label="Output")
-        assign_tool_btn.click(
-            assign_tool, inputs=[agent_select, tool_select], outputs=assign_output
-        )
-
-    with gr.Tab("Generate Chat App"):
-        generate_btn = gr.Button("Generate Chat App")
-        generate_output = gr.Textbox(label="Output")
-        generate_btn.click(generate_chat_app, outputs=generate_output)
-
-    with gr.Tab("Chat with Pypelyne"):
-        chat_input = gr.Textbox(label="Your Message")
-        chat_output = gr.Textbox(label="Pypelyne's Response")
-        chat_btn = gr.Button("Send")
-        chat_btn.click(chat_with_pypelyne, inputs=chat_input, outputs=chat_output)
+    purpose = st.sidebar.text_area(
+        "Project Purpose:",
+        value=pypelyne.purpose,
+        help="Describe the purpose of your coding project.",
+    )
+    pypelyne.purpose = purpose
 
-    with gr.Tab("View Pypelyne"):
-        view_agents_btn = gr.Button("View Agents")
-        view_tools_btn = gr.Button("View Tools")
-        view_output = gr.Textbox(label="Pypelyne Components")
-        view_agents_btn.click(list_agents, outputs=view_output)
-        view_tools_btn.click(list_tools, outputs=view_output)
+    # Agent and Tool Management
+    st.sidebar.header("🤖 Agents")
+    agents = st.sidebar.column(2)
+    tools = st.sidebar.column(1)
+
+    for agent in pypelyne.agents:
+        agents.write(f"- {agent}")
+
+    if st.sidebar.button("Create New Agent"):
+        agent_name = st.sidebar.text_input("Agent Name:")
+        agent_type = st.sidebar.selectbox("Agent Type:", ["Task Executor", "Information Retriever", "Decision Maker", "Data Analyzer"])
+        agent_complexity = st.sidebar.slider("Complexity (1-5):", 1, 5, 3)
+        new_agent = create_agent(agent_name, agent_type, agent_complexity)
+        pypelyne.agents = pypelyne.agents + [new_agent]
+
+    st.sidebar.header("🛠️ Tools")
+    for tool in pypelyne.tools:
+        tools.write(f"- {tool}")
+
+    if st.sidebar.button("Create New Tool"):
+        tool_name = st.sidebar.text_input("Tool Name:")
+        tool_type = st.sidebar.selectbox("Tool Type:", ["Web Scraper", "Database Connector", "API Caller", "File Handler", "Text Processor"])
+        new_tool = create_tool(tool_name, tool_type)
+        pypelyne.tools = pypelyne.tools + [new_tool]
+
+    # Main Content Area
+    st.header("💻 Code Interaction")
+
+    task = st.text_area(
+        "🎯 Task:",
+        value=pypelyne.task,
+        help="Describe the coding task you want to perform.",
+    )
+    if task:
+        pypelyne.task = task
 
-    def update_dropdowns():
-        return gr.Dropdown.update(
-            choices=[agent.name for agent in pypelyne.agents]
-        ), gr.Dropdown.update(choices=[tool.name for tool in pypelyne.tools])
+    user_input = st.text_input("💬 Your Input:")
 
-    create_agent_btn.click(update_dropdowns, outputs=[agent_select, tool_select])
-    create_tool_btn.click(update_dropdowns, outputs=[agent_select, tool_select])
+    if st.button("Execute"):
+        if user_input:
+            response = pypelyne.run_action("main", [user_input])
+            st.write("Pypelyne Says: ", response)
 
 if __name__ == "__main__":
-    app.launch()
-    app = Flask(__name__)
-
-    @app.route("/chat", methods=["POST"])
-    def chat():
-        message = request.json["message"]
-        response = chat_with_pypelyne(message)
-        return jsonify({"response": response})
-
-    @app.route("/agents", methods=["GET"])
-    def get_agents():
-        agents = list_agents()
-        return jsonify({"agents": agents})
-
-    @app.route("/tools", methods=["GET"])
-    def get_tools():
-        tools = list_tools()
-        return jsonify({"tools": tools})
-
-    if __name__ == "__main__":
-        app.run()
+    main()