S-Dreamer commited on
Commit
6db3950
·
verified ·
1 Parent(s): 6f8e334

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +36 -273
app.py CHANGED
@@ -1,280 +1,43 @@
1
- import os
2
- from typing import Optional, List, Dict, Any
3
- import gradio as gr
4
- import requests
5
- from smolagents import CodeAgent, Tool
6
- from smolagents.models import HfApiModel
7
- from smolagents.monitoring import LogLevel
8
- from gradio import ChatMessage
9
- import logging
10
- from functools import lru_cache
11
-
12
- # Configure logging
13
- logging.basicConfig(level=logging.INFO)
14
- logger = logging.getLogger(__name__)
15
-
16
- DEFAULT_MODEL = "Qwen/Qwen2.5-Coder-32B-Instruct"
17
- HF_API_TOKEN = os.getenv("HF_TOKEN")
18
-
19
- # Tool descriptions for the UI
20
- TOOL_DESCRIPTIONS = {
21
- "Hub Collections": "Add tool collections from Hugging Face Hub.",
22
- "Spaces": "Add tools from Hugging Face Spaces.",
23
- }
24
-
25
- @lru_cache(maxsize=128)
26
- def search_spaces(query: str, limit: int = 1) -> Optional[Dict[str, str]]:
27
- """
28
- Search for Hugging Face Spaces using the API.
29
- Returns the first result or None if no results.
30
- """
31
- try:
32
- url = f"https://huggingface.co/api/spaces?search={query}&limit={limit}"
33
- response = requests.get(url, headers={"Authorization": f"Bearer {HF_API_TOKEN}"})
34
- response.raise_for_status()
35
- spaces = response.json()
36
- if not spaces:
37
- return None
38
- return extract_space_info(spaces[0])
39
- except requests.RequestException as e:
40
- logger.error(f"Error searching spaces: {e}")
41
- return None
42
-
43
- def extract_space_info(space: Dict[str, Any]) -> Dict[str, str]:
44
- """
45
- Extracts space information from the API response.
46
- """
47
- space_id = space["id"]
48
- title = space_id.split("/")[-1]
49
- description = f"Tool from {space_id}"
50
- if "title" in space:
51
- title = space["title"]
52
- elif "cardData" in space and "title" in space["cardData"]:
53
- title = space["cardData"]["title"]
54
- if "description" in space:
55
- description = space["description"]
56
- elif "cardData" in space and "description" in space["cardData"]:
57
- description = space["cardData"]["description"]
58
- return {"id": space_id, "title": title, "description": description}
59
-
60
- def get_space_metadata(space_id: str) -> Optional[Dict[str, str]]:
61
- """
62
- Get metadata for a specific Hugging Face Space.
63
- """
64
- try:
65
- url = f"https://huggingface.co/api/spaces/{space_id}"
66
- response = requests.get(url, headers={"Authorization": f"Bearer {HF_API_TOKEN}"})
67
- response.raise_for_status()
68
- space = response.json()
69
- return extract_space_info(space)
70
- except requests.RequestException as e:
71
- logger.error(f"Error getting space metadata: {e}")
72
- return None
73
-
74
- def create_agent(model_name: str, space_tools: Optional[List[Dict[str, str]]] = None) -> Optional[CodeAgent]:
75
- """
76
- Create a CodeAgent with the specified model and tools.
77
- """
78
- if not space_tools:
79
- space_tools = []
80
- try:
81
- tools = [
82
- Tool.from_space(
83
- tool_info["id"],
84
- name=tool_info.get("name", tool_info["id"]),
85
- description=tool_info.get("description", ""),
86
- ) for tool_info in space_tools
87
- ]
88
- model = HfApiModel(model_id=model_name, token=HF_API_TOKEN)
89
- agent = CodeAgent(
90
- tools=tools,
91
- model=model,
92
- additional_authorized_imports=["PIL", "requests"],
93
- verbosity_level=LogLevel.DEBUG,
94
- )
95
- logger.info(f"Agent created successfully with {len(tools)} tools")
96
- return agent
97
- except Exception as e:
98
- logger.error(f"Error creating agent: {e}")
99
- return create_fallback_agent(tools)
100
-
101
- def create_fallback_agent(tools: List[Tool]) -> Optional[CodeAgent]:
102
- """
103
- Create a fallback CodeAgent if the primary model fails.
104
- """
105
- try:
106
- logger.info("Trying fallback model...")
107
- fallback_model = HfApiModel(model_id="Qwen/Qwen2.5-Coder-7B-Instruct", token=HF_API_TOKEN)
108
- agent = CodeAgent(
109
- tools=tools,
110
- model=fallback_model,
111
- additional_authorized_imports=["PIL", "requests"],
112
- verbosity_level=LogLevel.DEBUG,
113
- )
114
- logger.info("Agent created successfully with fallback model")
115
- return agent
116
- except Exception as e:
117
- logger.error(f"Error creating agent with fallback model: {e}")
118
- return None
119
-
120
- # Event handler functions
121
- def on_search_spaces(query: str) -> tuple:
122
- if not query:
123
- return "Please enter a search term.", "", "", ""
124
- try:
125
- space_info = search_spaces(query)
126
- if space_info is None:
127
- return "No spaces found.", "", "", ""
128
- results_md = f"### Search Results:\n- ID: `{space_info['id']}`\n- Title: {space_info['title']}\n- Description: {space_info['description']}\n"
129
- return results_md, space_info["id"], space_info["title"], space_info["description"]
130
- except Exception as e:
131
- logger.error(f"Error in search: {e}")
132
- return f"Error: {str(e)}", "", "", ""
133
-
134
- def on_validate_space(space_id: str) -> tuple:
135
- if not space_id:
136
- return "Please enter a space ID or search term.", "", ""
137
- try:
138
- space_info = get_space_metadata(space_id)
139
- if space_info is None:
140
- space_info = search_spaces(space_id)
141
- if space_info is None:
142
- return f"No spaces found for '{space_id}'.", "", ""
143
- result_md = f"### Found Space via Search:\n- ID: `{space_info['id']}`\n- Title: {space_info['title']}\n- Description: {space_info['description']}\n"
144
- return result_md, space_info["title"], space_info["description"]
145
- result_md = f"### Space Validated Successfully:\n- ID: `{space_info['id']}`\n- Title: {space_info['title']}\n- Description: {space_info['description']}\n"
146
- return result_md, space_info["title"], space_info["description"]
147
- except Exception as e:
148
- logger.error(f"Error validating space: {e}")
149
- return f"Error: {str(e)}", "", ""
150
 
151
- def on_add_tool(space_id: str, space_name: str, space_description: str, current_tools: List[Dict[str, str]]) -> tuple:
152
- if not space_id:
153
- return current_tools, "Please enter a space ID."
154
- for tool in current_tools:
155
- if tool["id"] == space_id:
156
- return current_tools, f"Tool '{space_id}' is already added."
157
- new_tool = {
158
- "id": space_id,
159
- "name": space_name if space_name else space_id,
160
- "description": space_description if space_description else "No description",
161
- }
162
- updated_tools = current_tools + [new_tool]
163
- tools_md = "### Added Tools:\n"
164
- for i, tool in enumerate(updated_tools, 1):
165
- tools_md += f"{i}. **{tool['name']}** (`{tool['id']}`)\n {tool['description']}\n\n"
166
- return updated_tools, tools_md
167
 
168
- def on_create_agent(model: str, space_tools: List[Dict[str, str]]) -> tuple:
169
- if not space_tools:
170
- return None, [], "", "Please add at least one tool before creating an agent.", "No agent created yet."
171
  try:
172
- agent = create_agent(model, space_tools)
173
- if agent is None:
174
- return None, [], "", "Failed to create agent. Please try again with different tools or model.", "No agent created yet."
175
- tools_str = ", ".join([f"{tool['name']} ({tool['id']})" for tool in space_tools])
176
- agent_status = update_agent_status(agent)
177
- return agent, [], "", f"✅ Agent created successfully with {model}!\nTools: {tools_str}", agent_status
178
- except Exception as e:
179
- logger.error(f"Error creating agent: {e}")
180
- return None, [], "", f"Error creating agent: {str(e)}", "No agent created yet."
 
 
181
 
182
- def add_user_message(message: str, chat_history: List[ChatMessage]) -> tuple:
183
- if not message:
184
- return "", chat_history
185
- chat_history = chat_history + [ChatMessage(role="user", content=message)]
186
- return message, chat_history
187
 
188
- def stream_to_gradio(agent: CodeAgent, task: str, reset_agent_memory: bool = False, additional_args: Optional[dict] = None):
189
- from smolagents.gradio_ui import pull_messages_from_step, handle_agent_output_types
190
- from smolagents.agent_types import AgentAudio, AgentImage, AgentText
191
- for step_log in agent.run(task, stream=True, reset=reset_agent_memory, additional_args=additional_args):
192
- for message in pull_messages_from_step(step_log):
193
- yield message
194
- final_answer = step_log
195
- final_answer = handle_agent_output_types(final_answer)
196
- if isinstance(final_answer, AgentImage):
197
- yield gr.ChatMessage(role="assistant", content={"path": final_answer.to_string(), "mime_type": "image/png"})
198
- elif isinstance(final_answer, AgentText) and os.path.exists(final_answer.to_string()):
199
- yield gr.ChatMessage(role="assistant", content=gr.Image(final_answer.to_string()))
200
- elif isinstance(final_answer, AgentAudio):
201
- yield gr.ChatMessage(role="assistant", content={"path": final_answer.to_string(), "mime_type": "audio/wav"})
202
- else:
203
- yield gr.ChatMessage(role="assistant", content=f"**Final answer:** {str(final_answer)}")
204
 
205
- def stream_agent_response(agent: CodeAgent, message: str, chat_history: List[ChatMessage]):
206
- if not message or agent is None:
207
- return chat_history
208
- yield chat_history
209
- try:
210
- for msg in stream_to_gradio(agent, message):
211
- chat_history = chat_history + [msg]
212
- yield chat_history
213
  except Exception as e:
214
- error_msg = f"Error: {str(e)}"
215
- chat_history = chat_history + [ChatMessage(role="assistant", content=error_msg)]
216
- yield chat_history
217
-
218
- def on_clear(agent: Optional[CodeAgent] = None) -> tuple:
219
- return agent, [], "", "Agent cleared. Create a new one to continue.", "", gr.update(interactive=False)
220
-
221
- def update_agent_status(agent: Optional[CodeAgent]) -> str:
222
- if agent is None:
223
- return "No agent created yet. Add a Space tool to get started."
224
- tools = agent.tools if hasattr(agent, "tools") else []
225
- tool_count = len(tools)
226
- status = f"Agent ready with {tool_count} tools"
227
- return status
228
-
229
- # Create the Gradio app
230
- with gr.Blocks(title="AI Agent Builder") as app:
231
- gr.Markdown("# AI Agent Builder with smolagents")
232
- gr.Markdown("Build your own AI agent by selecting tools from Hugging Face Spaces.")
233
- agent_state = gr.State(None)
234
- last_message = gr.State("")
235
- space_tools_state = gr.State([])
236
- msg_store = gr.State("")
237
- with gr.Row():
238
- with gr.Column(scale=1):
239
- gr.Markdown("## Tool Configuration")
240
- gr.Markdown("Add multiple Hugging Face Spaces as tools for your agent:")
241
- model_input = gr.Textbox(value=DEFAULT_MODEL, label="Model", visible=False)
242
- with gr.Group():
243
- gr.Markdown("### Add Space as Tool")
244
- space_tool_input = gr.Textbox(
245
- label="Space ID or Search Term",
246
- placeholder="Enter a Space ID (username/space-name) or search term",
247
- info="Enter a Space ID (username/space-name) or search term"
248
- )
249
- space_name_input = gr.Textbox(
250
- label="Tool Name (optional)",
251
- placeholder="Enter a name for this tool"
252
- )
253
- space_description_input = gr.Textbox(
254
- label="Tool Description (optional)",
255
- placeholder="Enter a description for this tool",
256
- lines=2
257
- )
258
- add_tool_button = gr.Button("Add Tool", variant="primary")
259
- gr.Markdown("### Added Tools")
260
- tools_display = gr.Markdown("No tools added yet. Add at least one tool before creating an agent.")
261
- create_button = gr.Button("Create Agent with Selected Tools", variant="secondary", size="lg")
262
- status_msg = gr.Markdown("")
263
- agent_status = gr.Markdown("No agent created yet.")
264
- with gr.Column(scale=2):
265
- chatbot = gr.Chatbot(label="Agent Chat", height=600, show_copy_button=True, avatar_images=("👤", "🤖"), type="messages")
266
- msg = gr.Textbox(label="Your message", placeholder="Type a message to your agent...", interactive=True)
267
- with gr.Row():
268
- with gr.Column(scale=1, min_width=60):
269
- clear = gr.Button("🗑️", scale=1)
270
- with gr.Column(scale=8):
271
- pass
272
- space_tool_input.submit(on_validate_space, inputs=[space_tool_input], outputs=[status_msg, space_name_input, space_description_input])
273
- add_tool_button.click(on_add_tool, inputs=[space_tool_input, space_name_input, space_description_input, space_tools_state], outputs=[space_tools_state, tools_display])
274
- create_button.click(on_create_agent, inputs=[model_input, space_tools_state], outputs=[agent_state, chatbot, msg, status_msg, agent_status])
275
- msg.submit(lambda message: (message, message, ""), inputs=[msg], outputs=[msg_store, msg, msg], queue=False)\
276
- .then(add_user_message, inputs=[msg_store, chatbot], outputs=[msg_store, chatbot], queue=False)\
277
- .then(stream_agent_response, inputs=[agent_state, msg_store, chatbot], outputs=chatbot, queue=True)
278
-
279
- if __name__ == "__main__":
280
- app.queue().launch()
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
2
+ import gradio as gr
3
+ import ast
4
+ import traceback
 
 
 
 
 
 
 
 
 
 
 
 
 
5
 
6
+ def optimize_and_debug(code):
7
+ """Attempts to optimize and debug the given Python code."""
 
8
  try:
9
+ tree = ast.parse(code)
10
+ # Rudimentary optimization: This is a placeholder. Real optimization is complex.
11
+ # Here we just check for unnecessary nested loops (a very simple example).
12
+ nested_loops = False
13
+ for node in ast.walk(tree):
14
+ if isinstance(node, ast.For) and any(isinstance(subnode, ast.For) for subnode in ast.walk(node)):
15
+ nested_loops = True
16
+ break
17
+ optimization_suggestions = []
18
+ if nested_loops:
19
+ optimization_suggestions.append("Consider optimizing nested loops. They can significantly impact performance.")
20
 
 
 
 
 
 
21
 
22
+ # Execution and error handling
23
+ exec(code, {}) # Execute the code (Caution: security implications for untrusted input)
24
+ return "Code executed successfully.", optimization_suggestions, ""
 
 
 
 
 
 
 
 
 
 
 
 
 
25
 
 
 
 
 
 
 
 
 
26
  except Exception as e:
27
+ error_message = traceback.format_exc()
28
+ return "Code execution failed.", [], error_message
29
+
30
+
31
+ iface = gr.Interface(
32
+ fn=optimize_and_debug,
33
+ inputs=gr.Textbox(lines=10, label="Enter your Python code here:"),
34
+ outputs=[
35
+ gr.Textbox(label="Result"),
36
+ gr.Textbox(label="Optimization Suggestions"),
37
+ gr.Textbox(label="Error Messages (if any)"),
38
+ ],
39
+ title="Python Code Optimizer & Debugger (Simplified)",
40
+ description="Paste your Python code below. This tool provides basic optimization suggestions and error reporting. Note: This is a simplified demonstration and does not replace a full debugger.",
41
+ )
42
+
43
+ iface.launch()