Spaces:
Build error
Build error
import os | |
import sys | |
import subprocess | |
import base64 | |
import json | |
from io import StringIO | |
from typing import Dict, List | |
import streamlit as st | |
import torch | |
from transformers import pipeline, AutoModelForSeq2SeqLM, AutoTokenizer, HfApi | |
from pylint import lint | |
import black | |
from setuptools import setup, Extension | |
from Cython.Build import cythonize | |
extensions = [ | |
Extension("app", ["app.pyx"]), | |
] | |
setup( | |
ext_modules=cythonize(extensions), | |
) | |
# Add your Hugging Face API token here | |
hf_token = st.secrets["huggingface"] | |
# Constants | |
PROJECT_ROOT = "./projects" | |
AGENT_DIRECTORY = "./agents" | |
AVAILABLE_CODE_GENERATIVE_MODELS = ["codegen", "gpt-neo", "codeparrot"] | |
# Global state management | |
if "chat_history" not in st.session_state: | |
st.session_state.chat_history = [] | |
if "terminal_history" not in st.session_state: | |
st.session_state.terminal_history = [] | |
if "workspace_projects" not in st.session_state: | |
st.session_state.workspace_projects = {} | |
if "available_agents" not in st.session_state: | |
st.session_state.available_agents = [] | |
# Load pre-trained models | |
def load_models(): | |
rag_retriever = pipeline("question-answering", model="facebook/rag-token-nq") | |
chat_model = AutoModelForSeq2SeqLM.from_pretrained("microsoft/DialoGPT-medium") | |
tokenizer = AutoTokenizer.from_pretrained("microsoft/DialoGPT-medium") | |
return rag_retriever, chat_model, tokenizer | |
rag_retriever, chat_model, tokenizer = load_models() | |
def process_input(user_input: str) -> str: | |
# Input pipeline: Tokenize and preprocess user input | |
input_ids = tokenizer(user_input, return_tensors="pt").input_ids | |
attention_mask = tokenizer(user_input, return_tensors="pt").attention_mask | |
# RAG model: Generate response | |
with torch.no_grad(): | |
rag_output = rag_retriever(question=user_input, context=user_input) | |
rag_answer = rag_output['answer'] | |
# Chat model: Refine response | |
chat_input = tokenizer(rag_answer, return_tensors="pt") | |
with torch.no_grad(): | |
chat_output = chat_model.generate(**chat_input) | |
refined_response = tokenizer.decode(chat_output[0], skip_special_tokens=True) | |
return refined_response | |
class AIAgent: | |
def __init__(self, name: str, description: str, skills: List[str], hf_api=None): | |
self.name = name | |
self.description = description | |
self.skills = skills | |
self._hf_api = hf_api | |
self._hf_token = hf_token | |
def hf_api(self): | |
if not self._hf_api and self.has_valid_hf_token(): | |
self._hf_api = HfApi(token=self._hf_token) | |
return self._hf_api | |
def has_valid_hf_token(self): | |
return bool(self._hf_token) | |
def create_agent_prompt(self): | |
return f"Name: {self.name}\nDescription: {self.description}\nSkills:\n" + "\n".join(self.skills) | |
async def autonomous_build(self, chat_history: List[str], workspace_projects: Dict[str, str], project_name: str, selected_model: str): | |
summary = "Chat History:\n" + "\n".join(chat_history) | |
summary += "\n\nWorkspace Projects:\n" + "\n".join(workspace_projects.items()) | |
next_step = "Based on the current state, the next logical step is to implement the main application logic." | |
project_path = os.path.join(PROJECT_ROOT, project_name) | |
if not os.path.exists(project_path): | |
os.makedirs(project_path) | |
requirements_file = os.path.join(project_path, "requirements.txt") | |
if not os.path.exists(requirements_file): | |
with open(requirements_file, "w") as f: | |
f.write("# Add your project's dependencies here\n") | |
app_file = os.path.join(project_path, "app.py") | |
if not os.path.exists(app_file): | |
with open(app_file, "w") as f: | |
f.write("# Your project's main application logic goes here\n") | |
if "create a gui" in summary.lower(): | |
gui_code = generate_code( | |
"Create a simple GUI for this application", selected_model) | |
with open(app_file, "a") as f: | |
f.write(gui_code) | |
build_command = "pip install -r requirements.txt && python app.py" | |
try: | |
result = subprocess.run( | |
build_command, shell=True, capture_output=True, text=True, cwd=project_path) | |
st.write(f"Build Output:\n{result.stdout}") | |
if result.stderr: | |
st.error(f"Build Errors:\n{result.stderr}") | |
except Exception as e: | |
st.error(f"Build Error: {e}") | |
return summary, next_step | |
def deploy_built_space_to_hf(self): | |
if not self.has_valid_hf_token(): | |
st.error("Invalid Hugging Face token. Please check your configuration.") | |
return | |
try: | |
files = get_built_space_files() | |
create_space_on_hugging_face(self.hf_api, self.name, self.description, True, files) | |
st.success(f"Successfully deployed {self.name} to Hugging Face Spaces!") | |
except Exception as e: | |
st.error(f"Error deploying to Hugging Face Spaces: {str(e)}") | |
def get_built_space_files() -> Dict[str, str]: | |
return { | |
"app.py": "# Your Streamlit app code here", | |
"requirements.txt": "streamlit\ntransformers" | |
} | |
def save_agent_to_file(agent: AIAgent): | |
if not os.path.exists(AGENT_DIRECTORY): | |
os.makedirs(AGENT_DIRECTORY) | |
file_path = os.path.join(AGENT_DIRECTORY, f"{agent.name}.txt") | |
with open(file_path, "w") as file: | |
file.write(agent.create_agent_prompt()) | |
st.session_state.available_agents.append(agent.name) | |
def load_agent_prompt(agent_name: str) -> str: | |
file_path = os.path.join(AGENT_DIRECTORY, f"{agent_name}.txt") | |
if os.path.exists(file_path): | |
with open(file_path, "r") as file: | |
agent_prompt = file.read() | |
return agent_prompt | |
else: | |
return None | |
def create_agent_from_text(name: str, text: str) -> str: | |
skills = text.split("\n") | |
agent = AIAgent(name, "AI agent created from text input.", skills) | |
save_agent_to_file(agent) | |
return agent.create_agent_prompt() | |
def chat_interface_with_agent(input_text: str, agent_name: str) -> str: | |
agent_prompt = load_agent_prompt(agent_name) | |
if agent_prompt is None: | |
return f"Agent {agent_name} not found." | |
model_name = "microsoft/DialoGPT-medium" | |
try: | |
generator = pipeline("text-generation", model=model_name) | |
generated_response = generator( | |
f"{agent_prompt}\n\nUser: {input_text}\nAgent:", max_length=100, do_sample=True, top_k=50)[0]["generated_text"] | |
return generated_response | |
except Exception as e: | |
return f"Error loading model: {e}" | |
def terminal_interface(command: str, project_name: str = None) -> str: | |
if project_name: | |
project_path = os.path.join(PROJECT_ROOT, project_name) | |
if not os.path.exists(project_path): | |
return f"Project {project_name} does not exist." | |
result = subprocess.run( | |
command, shell=True, capture_output=True, text=True, cwd=project_path) | |
else: | |
result = subprocess.run(command, shell=True, capture_output=True, text=True) | |
return result.stdout | |
def code_editor_interface(code: str) -> str: | |
try: | |
formatted_code = black.format_str(code, mode=black.FileMode()) | |
except black.NothingChanged: | |
formatted_code = code | |
result = StringIO() | |
sys.stdout = result | |
sys.stderr = result | |
lint.Run(['--rcfile=/dev/null', '-'], exit=False) | |
lint_message = result.getvalue() | |
sys.stdout = sys.__stdout__ | |
sys.stderr = sys.__stderr__ | |
return formatted_code, lint_message | |
def summarize_text(text: str) -> str: | |
summarizer = pipeline("summarization") | |
summary = summarizer(text, max_length=130, min_length=30, do_sample=False) | |
return summary[0]['summary_text'] | |
def sentiment_analysis(text: str) -> str: | |
analyzer = pipeline("sentiment-analysis") | |
result = analyzer(text) | |
return result[0]['label'] | |
def translate_code(code: str, source_language: str, target_language: str) -> str: | |
translator = pipeline("translation", model="Helsinki-NLP/opus-mt-en-ROMANCE") | |
translated_code = translator(code, max_length=512)[0]['translation_text'] | |
return translated_code | |
def generate_code(code_idea: str, model_name: str) -> str: | |
try: | |
generator = pipeline('text-generation', model=model_name) | |
generated_code = generator(code_idea, max_length=1000, num_return_sequences=1)[0]['generated_text'] | |
return generated_code | |
except Exception as e: | |
return f"Error generating code: {e}" | |
def chat_interface(input_text: str) -> str: | |
chatbot = pipeline("text-generation", model="microsoft/DialoGPT-medium") | |
response = chatbot(input_text, max_length=50, num_return_sequences=1)[0]['generated_text'] | |
return response | |
def workspace_interface(project_name: str) -> str: | |
project_path = os.path.join(PROJECT_ROOT, project_name) | |
if not os.path.exists(project_path): | |
os.makedirs(project_path) | |
st.session_state.workspace_projects[project_name] = {'files': []} | |
return f"Project '{project_name}' created successfully." | |
else: | |
return f"Project '{project_name}' already exists." | |
def add_code_to_workspace(project_name: str, code: str, file_name: str) -> str: | |
project_path = os.path.join(PROJECT_ROOT, project_name) | |
if not os.path.exists(project_path): | |
return f"Project '{project_name}' does not exist." | |
file_path = os.path.join(project_path, file_name) | |
with open(file_path, "w") as file: | |
file.write(code) | |
st.session_state.workspace_projects[project_name]['files'].append(file_name) | |
return f"Code added to '{file_name}' in project '{project_name}'." | |
def create_space_on_hugging_face(api, name, description, public, files, entrypoint="app.py"): | |
try: | |
repo = api.create_repo(name, exist_ok=True, private=not public) | |
for filename, content in files.items(): | |
api.upload_file( | |
path_or_fileobj=content.encode(), | |
path_in_repo=filename, | |
repo_id=repo.repo_id, | |
repo_type="space", | |
) | |
return repo | |
except Exception as e: | |
st.error(f"Error creating Hugging Face Space: {str(e)}") | |
return None | |
# Streamlit App | |
st.title("AI Agent Creator") | |
# Sidebar navigation | |
st.sidebar.title("Navigation") | |
app_mode = st.sidebar.selectbox( | |
"Choose the app mode", ["AI Agent Creator", "Tool Box", "Workspace Chat App"]) | |
if app_mode == "AI Agent Creator": | |
st.header("Create an AI Agent from Text") | |
st.subheader("From Text") | |
agent_name = st.text_input("Enter agent name:") | |
text_input = st.text_area("Enter skills (one per line):") | |
if st.button("Create Agent"): | |
agent_prompt = create_agent_from_text(agent_name, text_input) | |
st.success(f"Agent '{agent_name}' created and saved successfully.") | |
st.session_state.available_agents.append(agent_name) | |
elif app_mode == "Tool Box": | |
st.header("AI-Powered Tools") | |
st.subheader("Chat with CodeCraft") | |
chat_input = st.text_area("Enter your message:") | |
if st.button("Send"): | |
chat_response = chat_interface(chat_input) | |
st.session_state.chat_history.append((chat_input, chat_response)) | |
st.write(f"CodeCraft: {chat_response}") | |
st.subheader("Terminal") | |
terminal_input = st.text_input("Enter a command:") | |
if st.button("Run"): | |
terminal_output = terminal_interface(terminal_input) | |
st.session_state.terminal_history.append( | |
(terminal_input, terminal_output)) | |
st.code(terminal_output, language="bash") | |
st.subheader("Code Editor") | |
code_editor = st.text_area("Write your code:", height=300) | |
if st.button("Format & Lint"): | |
formatted_code, lint_message = code_editor_interface(code_editor) | |
st.code(formatted_code, language="python") | |
st.info(lint_message) | |
st.subheader("Summarize Text") | |
text_to_summarize = st.text_area("Enter text to summarize:") | |
if st.button("Summarize"): | |
summary = summarize_text(text_to_summarize) | |
st.write(f"Summary: {summary}") | |
st.subheader("Sentiment Analysis") | |
sentiment_text = st.text_area("Enter text for sentiment analysis:") | |
if st.button("Analyze Sentiment"): | |
sentiment = sentiment_analysis(sentiment_text) | |
st.write(f"Sentiment: {sentiment}") | |
st.subheader("Translate Code") | |
code_to_translate = st.text_area("Enter code to translate:") | |
source_language = st.text_input("Enter source language (e.g., 'Python'):") | |
target_language = st.text_input( | |
"Enter target language (e.g., 'JavaScript'):") | |
if st.button("Translate Code"): | |
translated_code = translate_code( | |
code_to_translate, source_language, target_language) | |
st.code(translated_code, language=target_language.lower()) | |
st.subheader("Code Generation") | |
code_idea = st.text_input("Enter your code idea:") | |
if st.button("Generate Code"): | |
generated_code = generate_code(code_idea, "gpt2") | |
st.code(generated_code, language="python") | |
elif app_mode == "Workspace Chat App": | |
st.header("Workspace Chat App") | |
st.subheader("Create a New Project") | |
project_name = st.text_input("Enter project name:") | |
if st.button("Create Project"): | |
workspace_status = workspace_interface(project_name) | |
st.success(workspace_status) | |
st.subheader("Add Code to Workspace") | |
code_to_add = st.text_area("Enter code to add to workspace:") |