Spaces:
Sleeping
Sleeping
| import json | |
| import os | |
| import requests | |
| import gradio as gr | |
| from smolagents import Tool, CodeAgent, InferenceClientModel | |
| class SentimentAnalysisTool(Tool): | |
| """ | |
| Tool that connects to the MCP sentiment analysis server. | |
| Analyzes the sentiment of text using the remote MCP server. | |
| """ | |
| name = "sentiment_analysis" | |
| description = """ | |
| Analyze the sentiment of text to determine if it's positive, negative, or neutral. | |
| Use this tool when the user asks about sentiment, emotion, feeling, or opinion in text. | |
| Returns polarity score (-1 to 1), subjectivity (0 to 1), and assessment (positive/negative/neutral). | |
| This tool connects to the MCP sentiment analysis server. | |
| """ | |
| inputs = { | |
| "text": { | |
| "type": "string", | |
| "description": "The text to analyze for sentiment", | |
| } | |
| } | |
| output_type = "string" | |
| def __init__(self, mcp_server_url: str = "https://s2v-general-mcp-sentiment.hf.space/gradio_api/mcp/"): | |
| """ | |
| Initialize the sentiment analysis tool with MCP server URL. | |
| Args: | |
| mcp_server_url: The URL of the MCP sentiment analysis server | |
| """ | |
| super().__init__() | |
| self.mcp_server_url = mcp_server_url | |
| def forward(self, text: str) -> str: | |
| """ | |
| Call the MCP sentiment analysis server to analyze text. | |
| Args: | |
| text: The text to analyze | |
| Returns: | |
| JSON string with sentiment analysis results | |
| """ | |
| try: | |
| # Call the Gradio MCP endpoint | |
| # Gradio MCP endpoints typically use SSE or JSON-RPC over HTTP | |
| # Try calling the function directly via the Gradio API | |
| api_url = self.mcp_server_url.rstrip('/') | |
| if not api_url.endswith('/api'): | |
| # Try the Gradio predict endpoint | |
| predict_url = api_url.replace('/gradio_api/mcp/', '/api/predict/') | |
| else: | |
| predict_url = api_url | |
| # Try calling via Gradio's predict API | |
| payload = { | |
| "data": [text], | |
| "fn_index": 0 | |
| } | |
| response = requests.post( | |
| predict_url, | |
| json=payload, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=10 | |
| ) | |
| if response.status_code == 200: | |
| result = response.json() | |
| # Gradio returns {"data": [result]} | |
| if "data" in result and len(result["data"]) > 0: | |
| return result["data"][0] | |
| else: | |
| return json.dumps(result, indent=2) | |
| else: | |
| # Fallback: Use TextBlob directly (same as MCP server) | |
| from textblob import TextBlob | |
| blob = TextBlob(text) | |
| sentiment = blob.sentiment | |
| result = { | |
| "polarity": round(sentiment.polarity, 2), | |
| "subjectivity": round(sentiment.subjectivity, 2), | |
| "assessment": "positive" if sentiment.polarity > 0 else "negative" if sentiment.polarity < 0 else "neutral" | |
| } | |
| return json.dumps(result, indent=2) | |
| except requests.exceptions.RequestException: | |
| # Fallback: Use TextBlob directly (same as MCP server) | |
| try: | |
| from textblob import TextBlob | |
| blob = TextBlob(text) | |
| sentiment = blob.sentiment | |
| result = { | |
| "polarity": round(sentiment.polarity, 2), | |
| "subjectivity": round(sentiment.subjectivity, 2), | |
| "assessment": "positive" if sentiment.polarity > 0 else "negative" if sentiment.polarity < 0 else "neutral" | |
| } | |
| return json.dumps(result, indent=2) | |
| except Exception as e: | |
| return f"Error analyzing sentiment: {str(e)}" | |
| except Exception as e: | |
| return f"Error: {str(e)}" | |
| # Initialize the sentiment analysis tool with MCP server connection | |
| sentiment_tool = SentimentAnalysisTool() | |
| # Create the agent with the MCP sentiment analysis tool | |
| # ============================================================================ | |
| # LLM Configuration - Current: Qwen/Qwen2.5-72B-Instruct via HuggingFace Inference API | |
| # ============================================================================ | |
| # | |
| # API KEY SETUP: | |
| # | |
| # HUGGINGFACE SPACE DEPLOYMENT: | |
| # - HF_TOKEN is automatically available in HuggingFace Spaces | |
| # - You can also set secrets in Space Settings → Secrets | |
| # - Add secret: HF_TOKEN with your token value | |
| # | |
| # LOCAL SETUP: | |
| # 1. Environment variable: export HF_TOKEN="your_huggingface_token" | |
| # 2. Or login via: huggingface-cli login | |
| # | |
| # Get your HuggingFace token from: https://huggingface.co/settings/tokens | |
| # | |
| # PRIMARY MODEL (Currently Active): | |
| # Model: Qwen/Qwen2.5-72B-Instruct | |
| # Provider: HuggingFace Inference API (default, no provider specified) | |
| # Usage: High-performance model via HuggingFace Inference API | |
| # | |
| # ALTERNATIVE MODELS (for rate limit fallbacks): | |
| # If you hit rate limits, uncomment one of these alternatives: | |
| # | |
| # Option 1: Other HuggingFace Inference API models (uses HF_TOKEN) | |
| # model=InferenceClientModel(model_id="meta-llama/Meta-Llama-3-8B-Instruct", token=api_token) | |
| # model=InferenceClientModel(model_id="mistralai/Mistral-7B-Instruct-v0.2", token=api_token) | |
| # model=InferenceClientModel(model_id="Qwen/Qwen2.5-7B-Instruct", token=api_token) | |
| # model=InferenceClientModel(model_id="google/gemma-2-2b-it", token=api_token) | |
| # model=InferenceClientModel(model_id="Qwen/Qwen2.5-32B-Instruct", token=api_token) | |
| # model=InferenceClientModel(model_id="meta-llama/Llama-3.1-70B-Instruct", token=api_token) | |
| # | |
| # Option 2: OpenAI (requires API key: export OPENAI_API_KEY="your_key") | |
| # from smolagents import OpenAIModel | |
| # model=OpenAIModel(model="gpt-4o-mini") # or "gpt-4o", "gpt-3.5-turbo" | |
| # | |
| # Option 3: Anthropic Claude (requires API key: export ANTHROPIC_API_KEY="your_key") | |
| # from smolagents import AnthropicModel | |
| # model=AnthropicModel(model="claude-3-5-sonnet-20241022") # or "claude-3-opus-20240229" | |
| # | |
| # Option 4: Local Ollama (for offline/local use) | |
| # from smolagents import OllamaModel | |
| # model=OllamaModel(model="llama3.1") # or "mistral", "qwen", etc. | |
| # | |
| # ============================================================================ | |
| # Get API token from environment variables | |
| # Priority: HF_TOKEN > HUGGINGFACE_TOKEN | |
| # In HuggingFace Spaces, HF_TOKEN is automatically available | |
| api_token = ( | |
| os.getenv("HF_TOKEN") or | |
| os.getenv("HUGGINGFACE_TOKEN") | |
| ) | |
| # Check if running in HuggingFace Space | |
| is_hf_space = os.getenv("SPACE_ID") is not None | |
| # Initialize the model with API token | |
| # Using HuggingFace Inference API (no provider specified = default HF Inference API) | |
| if api_token: | |
| try: | |
| agent = CodeAgent( | |
| tools=[sentiment_tool], | |
| model=InferenceClientModel( | |
| model_id="Qwen/Qwen2.5-72B-Instruct", | |
| token=api_token # Uses HuggingFace Inference API by default | |
| ) | |
| ) | |
| if is_hf_space: | |
| print("✅ Agent initialized successfully using HuggingFace Space token") | |
| else: | |
| print("✅ Agent initialized successfully using local API token") | |
| except Exception as e: | |
| print(f"❌ Error initializing agent with token: {e}") | |
| print("Falling back to tokenless initialization...") | |
| agent = CodeAgent( | |
| tools=[sentiment_tool], | |
| model=InferenceClientModel( | |
| model_id="Qwen/Qwen2.5-72B-Instruct" | |
| ) | |
| ) | |
| else: | |
| # Try without explicit token (will use HuggingFace login if available) | |
| if is_hf_space: | |
| print("⚠️ Warning: No API token found in HuggingFace Space secrets.") | |
| print("Please add HF_TOKEN in Space Settings → Secrets") | |
| else: | |
| print("⚠️ Warning: No API token found. Attempting to use HuggingFace login...") | |
| print("If this fails, set one of these environment variables:") | |
| print(" - HF_TOKEN (HuggingFace token)") | |
| print(" - HUGGINGFACE_TOKEN (alternative HuggingFace token)") | |
| print("\nOr login via: huggingface-cli login") | |
| agent = CodeAgent( | |
| tools=[sentiment_tool], | |
| model=InferenceClientModel( | |
| model_id="Qwen/Qwen2.5-72B-Instruct" | |
| ) | |
| ) | |
| def chat_with_agent(message, history, request: gr.Request): | |
| """ | |
| Wrapper function to use the agent's run method with Gradio. | |
| The agent is stateful and manages history internally. | |
| """ | |
| try: | |
| # Run the agent and get the response | |
| result = agent.run(message) | |
| # Convert result to string if it's not already | |
| if isinstance(result, dict): | |
| # Format dictionary results nicely | |
| if "polarity" in result and "subjectivity" in result and "assessment" in result: | |
| formatted_result = f"""Sentiment Analysis Results: | |
| - **Polarity**: {result['polarity']} (range: -1 to 1) | |
| - **Subjectivity**: {result['subjectivity']} (range: 0 to 1) | |
| - **Assessment**: {result['assessment'].upper()}""" | |
| return formatted_result | |
| else: | |
| # Format other dictionaries as JSON | |
| import json | |
| return json.dumps(result, indent=2) | |
| elif isinstance(result, (list, tuple)): | |
| # Convert lists/tuples to string | |
| return str(result) | |
| else: | |
| # Already a string or other type that can be converted | |
| return str(result) | |
| except Exception as e: | |
| return f"Error: {str(e)}" | |
| # Create and launch the Gradio Chat Interface | |
| demo = gr.ChatInterface( | |
| fn=chat_with_agent, | |
| title="MCP Sentiment Analysis Agent", | |
| description="Chat with an AI agent that can analyze sentiment using the MCP sentiment analysis server.", | |
| chatbot=gr.Chatbot(layout="bubble", show_copy_button=True, type="messages"), | |
| type="messages" | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch() | |