Spaces:
Sleeping
Sleeping
File size: 3,827 Bytes
c10da7d 9e0ec52 716a5c8 8e7d1a1 9e0ec52 c10da7d 4433b73 c10da7d a5d512e c10da7d a5d512e c10da7d 4433b73 c10da7d a5d512e c10da7d a5d512e c10da7d 4433b73 c10da7d a5d512e c10da7d a5d512e c10da7d 9e0ec52 3cf8730 d1568ce bc906fa 716a5c8 d1568ce 89d512b ea6e8d7 8e7d1a1 9e0ec52 ea6e8d7 9e0ec52 c10da7d 8e7d1a1 c10da7d 9e0ec52 89d512b 9e0ec52 3cf8730 9e0ec52 8e7d1a1 9e0ec52 8e7d1a1 9e0ec52 c10da7d 8e7d1a1 3cf8730 9e0ec52 8e7d1a1 9e0ec52 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 |
from smolagents import CodeAgent, WikipediaSearchTool, LiteLLMModel, tool, load_tool # HfApiModel, OpenAIServerModel
import asyncio
import os
import yaml
# Simulated additional tools (implementation depends on external APIs or setup)
@tool
class GoogleSearchTool:
"""Tool for performing Google searches using Custom Search JSON API"""
def __init__(self):
self.api_key = os.environ.get("GOOGLE_API_KEY")
self.cse_id = os.environ.get("GOOGLE_CSE_ID")
if not self.api_key or not self.cse_id:
raise ValueError("GOOGLE_API_KEY and GOOGLE_CSE_ID must be set in environment variables.")
def search(self, query: str) -> str:
"""Perform a Google search query
Args:
query: Search query string
Returns:
Simulated search results (replace with actual API call)
"""
return f"Google search results for '{query}' (simulated)."
@tool
class ImageAnalysisTool:
"""Tool for analyzing images using computer vision"""
def analyze(self, image_path: str) -> str:
# Placeholder: Use Google Vision API or similar in real implementation
"""Analyze an image file
Args:
image_path: Path to image file
Returns:
Simulated image analysis results
"""
return f"Analyzed image at '{image_path}' (simulated description)."
@tool
class LocalFileAudioTool:
"""Tool for transcribing audio files"""
def transcribe(self, file_path: str) -> str:
# Placeholder: Use speech recognition library like SpeechRecognition in real setup
"""Transcribe audio from file
Args:
file_path: Path to audio file
Returns:
Simulated transcription text
"""
return f"Transcribed audio from '{file_path}' (simulated transcription)."
class MagAgent:
def __init__(self):
"""Initialize the MagAgent with search tools."""
print("Initializing MagAgent with search tools...")
model = LiteLLMModel(
model_id="gemini/gemini-2.0-flash-lite",
api_key= os.environ.get("GEMINI_KEY"),
max_tokens=8192
)
# Load prompt templates
with open("prompts.yaml", 'r') as stream:
prompt_templates = yaml.safe_load(stream)
self.agent = CodeAgent(
model= model,
tools=[
GoogleSearchTool(),
WikipediaSearchTool(),
ImageAnalysisTool(),
LocalFileAudioTool()
]
)
print("MagAgent initialized.")
async def __call__(self, question: str) -> str:
"""Process a question asynchronously using the MagAgent."""
print(f"MagAgent received question (first 50 chars): {question[:50]}...")
try:
# Define a task with fallback search logic
task = (
f"Answer the following question accurately and concisely: {question}\n"
)
response = await asyncio.to_thread(
self.agent.run,
task=task
)
# Ensure response is a string, fixing the integer error
response = str(response) if response is not None else "No answer found."
if not response or "No Wikipedia page found" in response:
# Fallback response if search fails
response = "Unable to retrieve exact data. Please refine the question or check external sources."
print(f"MagAgent response: {response[:50]}...")
return response
except Exception as e:
error_msg = f"Error processing question: {str(e)}. Check API key or network connectivity."
print(error_msg)
return error_msg |