juanmaguitar commited on
Commit
e27aff8
·
1 Parent(s): a33f19e

more reliable way of getting images

Browse files
Files changed (2) hide show
  1. tools/image_handler.py +90 -43
  2. tools/visit_webpage.py +14 -16
tools/image_handler.py CHANGED
@@ -18,7 +18,12 @@ class ImageHandlerTool(Tool):
18
  },
19
  'style': {
20
  'type': 'string',
21
- 'description': 'Style for generated images (e.g., "photo", "artistic")',
 
 
 
 
 
22
  'nullable': True
23
  }
24
  }
@@ -33,9 +38,17 @@ class ImageHandlerTool(Tool):
33
  def _download_image(self, url: str, filename: str) -> Optional[str]:
34
  """Downloads an image from a URL and saves it to a temporary file"""
35
  try:
36
- response = requests.get(url, timeout=10)
 
 
 
37
  response.raise_for_status()
38
 
 
 
 
 
 
39
  # Ensure temp directory exists
40
  os.makedirs(self.temp_dir, exist_ok=True)
41
 
@@ -52,11 +65,12 @@ class ImageHandlerTool(Tool):
52
  """Attempts to find images via web search"""
53
  results = []
54
  try:
55
- # Try different search queries
56
  search_queries = [
57
- f"{query} high quality photo",
58
- f"{query} professional photograph",
59
- f"{query} travel photo"
 
60
  ]
61
 
62
  for search_query in search_queries:
@@ -64,23 +78,33 @@ class ImageHandlerTool(Tool):
64
  break
65
 
66
  time.sleep(2) # Rate limiting
67
- search_results = self.web_search.forward(
68
- query=search_query, max_results=num_images)
69
-
70
- for idx, result in enumerate(search_results):
71
- if len(results) >= num_images:
72
- break
73
-
74
- if 'image_url' in result:
75
- filename = f"{query.replace(' ', '_')}_{idx}.jpg"
76
- file_path = self._download_image(
77
- result['image_url'], filename)
78
- if file_path:
79
- results.append({
80
- 'file_path': file_path,
81
- 'source': 'web',
82
- 'url': result['image_url']
83
- })
 
 
 
 
 
 
 
 
 
 
84
 
85
  except Exception as e:
86
  print(f"Web search failed: {str(e)}")
@@ -91,42 +115,61 @@ class ImageHandlerTool(Tool):
91
  """Generates images using the image generation tool"""
92
  results = []
93
  try:
 
 
 
 
 
 
 
94
  for idx in range(num_images):
95
- prompt = f"Generate a {style} of {query}"
96
- response = self.image_gen.forward(prompt=prompt)
97
-
98
- if isinstance(response, dict) and 'image_path' in response:
99
- results.append({
100
- 'file_path': response['image_path'],
101
- 'source': 'generated',
102
- 'prompt': prompt
103
- })
104
- elif isinstance(response, str) and os.path.exists(response):
105
- results.append({
106
- 'file_path': response,
107
- 'source': 'generated',
108
- 'prompt': prompt
109
- })
 
 
 
 
 
 
 
 
 
 
110
 
111
  except Exception as e:
112
  print(f"Image generation failed: {str(e)}")
113
 
114
  return results
115
 
116
- def forward(self, query: str, num_images: int = 2, style: str = "photo") -> Dict:
117
  """Gets or generates images for the query
118
  Args:
119
  query: What to get images of
120
  num_images: How many images to get
121
  style: Style for generated images
 
122
  Returns:
123
  Dict containing results and status
124
  """
125
  all_results = []
126
 
127
- # First try web search
128
- web_results = self._try_web_search(query, num_images)
129
- all_results.extend(web_results)
 
130
 
131
  # If we don't have enough images, try generation
132
  if len(all_results) < num_images:
@@ -143,5 +186,9 @@ class ImageHandlerTool(Tool):
143
  return {
144
  "status": "success",
145
  "images": all_results,
146
- "total": len(all_results)
 
 
 
 
147
  }
 
18
  },
19
  'style': {
20
  'type': 'string',
21
+ 'description': 'Style for generated images (e.g., "photo", "artistic", "realistic")',
22
+ 'nullable': True
23
+ },
24
+ 'skip_web_search': {
25
+ 'type': 'boolean',
26
+ 'description': 'Whether to skip web search and go straight to generation',
27
  'nullable': True
28
  }
29
  }
 
38
  def _download_image(self, url: str, filename: str) -> Optional[str]:
39
  """Downloads an image from a URL and saves it to a temporary file"""
40
  try:
41
+ headers = {
42
+ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
43
+ }
44
+ response = requests.get(url, timeout=10, headers=headers)
45
  response.raise_for_status()
46
 
47
+ # Check if response is actually an image
48
+ content_type = response.headers.get('content-type', '')
49
+ if not content_type.startswith('image/'):
50
+ return None
51
+
52
  # Ensure temp directory exists
53
  os.makedirs(self.temp_dir, exist_ok=True)
54
 
 
65
  """Attempts to find images via web search"""
66
  results = []
67
  try:
68
+ # Try different search queries with better targeting
69
  search_queries = [
70
+ f"{query} high resolution photo",
71
+ f"{query} professional photography",
72
+ f"{query} best pictures",
73
+ f"{query} travel photography"
74
  ]
75
 
76
  for search_query in search_queries:
 
78
  break
79
 
80
  time.sleep(2) # Rate limiting
81
+ try:
82
+ search_results = self.web_search.forward(
83
+ query=search_query, max_results=num_images)
84
+
85
+ if isinstance(search_results, str): # Handle string responses
86
+ continue
87
+
88
+ for idx, result in enumerate(search_results):
89
+ if len(results) >= num_images:
90
+ break
91
+
92
+ # Try both image_url and direct URL fields
93
+ url = result.get('image_url') or result.get('url')
94
+ if url and url.lower().endswith(('.jpg', '.jpeg', '.png', '.gif')):
95
+ filename = f"{query.replace(' ', '_')}_{idx}.jpg"
96
+ file_path = self._download_image(url, filename)
97
+ if file_path:
98
+ results.append({
99
+ 'file_path': file_path,
100
+ 'source': 'web',
101
+ 'url': url,
102
+ 'title': result.get('title', ''),
103
+ 'attribution': result.get('source', '')
104
+ })
105
+ except Exception as search_error:
106
+ print(f"Search query failed: {str(search_error)}")
107
+ continue
108
 
109
  except Exception as e:
110
  print(f"Web search failed: {str(e)}")
 
115
  """Generates images using the image generation tool"""
116
  results = []
117
  try:
118
+ # Enhanced prompts for better generation
119
+ prompts = [
120
+ f"Generate a {style} style image of {query}, high quality, detailed",
121
+ f"Create a {style} representation of {query}, professional quality",
122
+ f"Make a {style} image showing {query}, realistic and clear"
123
+ ]
124
+
125
  for idx in range(num_images):
126
+ prompt = prompts[idx % len(prompts)].replace(
127
+ query, query + f" {idx+1}")
128
+ try:
129
+ response = self.image_gen.forward(prompt=prompt)
130
+
131
+ if isinstance(response, dict) and 'image_path' in response:
132
+ results.append({
133
+ 'file_path': response['image_path'],
134
+ 'source': 'generated',
135
+ 'prompt': prompt,
136
+ 'style': style
137
+ })
138
+ elif isinstance(response, str) and os.path.exists(response):
139
+ results.append({
140
+ 'file_path': response,
141
+ 'source': 'generated',
142
+ 'prompt': prompt,
143
+ 'style': style
144
+ })
145
+ except Exception as gen_error:
146
+ print(
147
+ f"Failed to generate image {idx+1}: {str(gen_error)}")
148
+ continue
149
+
150
+ time.sleep(1) # Brief pause between generations
151
 
152
  except Exception as e:
153
  print(f"Image generation failed: {str(e)}")
154
 
155
  return results
156
 
157
+ def forward(self, query: str, num_images: int = 2, style: str = "photo", skip_web_search: bool = False) -> Dict:
158
  """Gets or generates images for the query
159
  Args:
160
  query: What to get images of
161
  num_images: How many images to get
162
  style: Style for generated images
163
+ skip_web_search: Whether to skip web search
164
  Returns:
165
  Dict containing results and status
166
  """
167
  all_results = []
168
 
169
+ # Try web search first unless skipped
170
+ if not skip_web_search:
171
+ web_results = self._try_web_search(query, num_images)
172
+ all_results.extend(web_results)
173
 
174
  # If we don't have enough images, try generation
175
  if len(all_results) < num_images:
 
186
  return {
187
  "status": "success",
188
  "images": all_results,
189
+ "total": len(all_results),
190
+ "sources": {
191
+ "web": len([img for img in all_results if img['source'] == 'web']),
192
+ "generated": len([img for img in all_results if img['source'] == 'generated'])
193
+ }
194
  }
tools/visit_webpage.py CHANGED
@@ -1,42 +1,40 @@
 
1
  from typing import Any, Optional
2
  from smolagents.tools import Tool
3
  import requests
4
  import markdownify
5
  import smolagents
6
 
 
7
  class VisitWebpageTool(Tool):
8
  name = "visit_webpage"
9
  description = "Visits a webpage at the given url and reads its content as a markdown string. Use this to browse webpages."
10
- inputs = {'url': {'type': 'string', 'description': 'The url of the webpage to visit.'}}
 
11
  output_type = "string"
12
 
13
  def forward(self, url: str) -> str:
14
  try:
15
- import requests
16
- from markdownify import markdownify
17
- from requests.exceptions import RequestException
18
-
19
- from smolagents.utils import truncate_content
20
- except ImportError as e:
21
- raise ImportError(
22
- "You must install packages `markdownify` and `requests` to run this tool: for instance run `pip install markdownify requests`."
23
- ) from e
24
- try:
25
  # Send a GET request to the URL with a 20-second timeout
26
- response = requests.get(url, timeout=20)
27
- response.raise_for_status() # Raise an exception for bad status codes
28
 
29
  # Convert the HTML content to Markdown
30
- markdown_content = markdownify(response.text).strip()
31
 
32
  # Remove multiple line breaks
33
  markdown_content = re.sub(r"\n{3,}", "\n\n", markdown_content)
34
 
35
- return truncate_content(markdown_content, 10000)
36
 
37
  except requests.exceptions.Timeout:
38
  return "The request timed out. Please try again later or check the URL."
39
- except RequestException as e:
40
  return f"Error fetching the webpage: {str(e)}"
41
  except Exception as e:
42
  return f"An unexpected error occurred: {str(e)}"
 
1
+ import re
2
  from typing import Any, Optional
3
  from smolagents.tools import Tool
4
  import requests
5
  import markdownify
6
  import smolagents
7
 
8
+
9
  class VisitWebpageTool(Tool):
10
  name = "visit_webpage"
11
  description = "Visits a webpage at the given url and reads its content as a markdown string. Use this to browse webpages."
12
+ inputs = {'url': {'type': 'string',
13
+ 'description': 'The url of the webpage to visit.'}}
14
  output_type = "string"
15
 
16
  def forward(self, url: str) -> str:
17
  try:
18
+ # Add headers to mimic a browser
19
+ headers = {
20
+ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
21
+ }
22
+
 
 
 
 
 
23
  # Send a GET request to the URL with a 20-second timeout
24
+ response = requests.get(url, timeout=20, headers=headers)
25
+ response.raise_for_status()
26
 
27
  # Convert the HTML content to Markdown
28
+ markdown_content = markdownify.markdownify(response.text).strip()
29
 
30
  # Remove multiple line breaks
31
  markdown_content = re.sub(r"\n{3,}", "\n\n", markdown_content)
32
 
33
+ return markdown_content[:10000] # Limit content length
34
 
35
  except requests.exceptions.Timeout:
36
  return "The request timed out. Please try again later or check the URL."
37
+ except requests.exceptions.RequestException as e:
38
  return f"Error fetching the webpage: {str(e)}"
39
  except Exception as e:
40
  return f"An unexpected error occurred: {str(e)}"