Spaces:
Running
Running
File size: 3,680 Bytes
c10da7d 9e0ec52 716a5c8 8e7d1a1 9e0ec52 c10da7d 4433b73 21e9b57 0208310 c10da7d 4433b73 0b689e4 355baca c10da7d 4433b73 0b689e4 355baca 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
def GoogleSearchTool(query: str) -> str:
"""Tool for performing Google searches using Custom Search JSON API
Args:
query (str): Search query string to look up
Returns:
str: Formatted search results (simulated or real)
"""
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.")
return f"Google search results for '{query}' (simulated)."
@tool
def ImageAnalysisTool(image_path: str) -> str:
"""Tool for analyzing images using computer vision
Args:
image_path: Path to image file
Returns:
Simulated image analysis results
"""
def analyze(self, image_path: str) -> str:
# Placeholder: Use Google Vision API or similar in real implementation
return f"Analyzed image at '{image_path}' (simulated description)."
@tool
def LocalFileAudioTool(file_path: str) -> str:
"""Tool for transcribing audio files
Args:
file_path: Path to audio file
Returns:
Simulated transcription text
"""
def transcribe(self, file_path: str) -> str:
# Placeholder: Use speech recognition library like SpeechRecognition in real setup
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 |