Create app.py

app.py

@@ -0,0 +1,230 @@
+import gradio as gr
+import requests
+from transformers import pipeline
+import json
+import time
+from typing import Dict, List
+import os
+
+class FinanceAgent:
+    def __init__(self):
+        # Initialize the summarization pipeline
+        try:
+            self.summarizer = pipeline(
+                "summarization", 
+                model="facebook/bart-large-cnn",
+                max_length=512,
+                min_length=100
+            )
+        except Exception as e:
+            print(f"Error loading summarization model: {e}")
+            self.summarizer = None
+        
+    def search_web(self, query: str, num_results: int = 5) -> List[Dict]:
+        """
+        Search the web using DuckDuckGo API (free alternative)
+        """
+        try:
+            # Using DuckDuckGo Instant Answer API
+            url = "https://api.duckduckgo.com/"
+            params = {
+                'q': f"{query} finance financial",
+                'format': 'json',
+                'no_html': '1',
+                'skip_disambig': '1'
+            }
+            
+            response = requests.get(url, params=params, timeout=10)
+            data = response.json()
+            
+            results = []
+            
+            # Extract abstract if available
+            if data.get('Abstract'):
+                results.append({
+                    'title': data.get('Heading', 'Financial Information'),
+                    'content': data.get('Abstract', ''),
+                    'url': data.get('AbstractURL', '')
+                })
+            
+            # Extract related topics
+            for topic in data.get('RelatedTopics', [])[:num_results-1]:
+                if isinstance(topic, dict) and topic.get('Text'):
+                    results.append({
+                        'title': topic.get('Text', '')[:100] + '...',
+                        'content': topic.get('Text', ''),
+                        'url': topic.get('FirstURL', '')
+                    })
+            
+            return results
+            
+        except Exception as e:
+            print(f"Search error: {e}")
+            return [{'title': 'Search Error', 'content': f'Unable to search: {str(e)}', 'url': ''}]
+    
+    def get_financial_context(self, topic: str) -> str:
+        """
+        Get basic financial context for common topics
+        """
+        financial_contexts = {
+            'stock': 'Stocks represent ownership shares in publicly traded companies. Stock prices fluctuate based on company performance, market conditions, and investor sentiment.',
+            'bond': 'Bonds are debt securities where investors lend money to entities for a defined period at a fixed interest rate.',
+            'cryptocurrency': 'Cryptocurrencies are digital assets that use cryptography for security and operate on decentralized networks.',
+            'inflation': 'Inflation is the rate at which the general level of prices for goods and services rises, eroding purchasing power.',
+            'recession': 'A recession is a significant decline in economic activity across the economy lasting more than a few months.',
+            'fed': 'The Federal Reserve is the central banking system of the United States, responsible for monetary policy.',
+            'gdp': 'Gross Domestic Product (GDP) is the total monetary value of all goods and services produced within a country.',
+            'market': 'Financial markets are platforms where buyers and sellers trade financial securities, commodities, and other assets.'
+        }
+        
+        topic_lower = topic.lower()
+        for key, context in financial_contexts.items():
+            if key in topic_lower:
+                return context
+        
+        return "This is a financial topic that may involve various economic, market, or investment-related concepts."
+    
+    def summarize_content(self, content: str, max_length: int = 300) -> str:
+        """
+        Summarize content using the transformer model
+        """
+        if not self.summarizer or not content.strip():
+            return "Unable to generate summary."
+        
+        try:
+            # Truncate content if too long
+            if len(content) > 1000:
+                content = content[:1000] + "..."
+            
+            summary = self.summarizer(content, max_length=max_length, min_length=50, do_sample=False)
+            return summary[0]['summary_text']
+        except Exception as e:
+            print(f"Summarization error: {e}")
+            return f"Summary unavailable: {str(e)}"
+    
+    def process_finance_query(self, prompt: str) -> str:
+        """
+        Main function to process finance queries
+        """
+        if not prompt.strip():
+            return "Please enter a financial topic or question."
+        
+        # Add progress indicator
+        progress_msg = f"🔍 Searching for information about: {prompt}\n\n"
+        
+        # Search for relevant information
+        search_results = self.search_web(prompt)
+        
+        if not search_results:
+            return progress_msg + "❌ No search results found. Please try a different query."
+        
+        # Combine search results
+        combined_content = ""
+        sources = []
+        
+        for i, result in enumerate(search_results[:3], 1):  # Limit to top 3 results
+            if result.get('content'):
+                combined_content += f"\n{result['content']}\n"
+                if result.get('url'):
+                    sources.append(f"{i}. {result['url']}")
+        
+        # Get financial context
+        context = self.get_financial_context(prompt)
+        
+        # Create comprehensive content for summarization
+        full_content = f"{context}\n\nCurrent Information:\n{combined_content}"
+        
+        # Generate summary
+        summary = self.summarize_content(full_content)
+        
+        # Format final response
+        response = f"""## 📊 Financial Summary: {prompt.title()}
+
+### 🎯 Key Points:
+{summary}
+
+### 📋 Context:
+{context}
+
+### 🔗 Sources:
+"""
+        
+        if sources:
+            for source in sources:
+                response += f"\n{source}"
+        else:
+            response += "\nInformation compiled from web search results."
+        
+        response += f"\n\n⏰ *Generated on: {time.strftime('%Y-%m-%d %H:%M:%S')}*"
+        
+        return response
+
+# Initialize the agent
+agent = FinanceAgent()
+
+def finance_chat_interface(message, history):
+    """
+    Gradio chat interface function
+    """
+    try:
+        response = agent.process_finance_query(message)
+        return response
+    except Exception as e:
+        return f"❌ Error processing your request: {str(e)}\n\nPlease try again with a different query."
+
+# Create Gradio interface
+def create_interface():
+    with gr.Blocks(title="Financial AI Agent", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("""
+        # 🏦 Financial AI Agent
+        
+        Ask me about any financial topic and I'll search the web for current information and provide you with a comprehensive summary.
+        
+        **Examples:**
+        - "What is the current state of cryptocurrency market?"
+        - "Explain inflation and its impact on the economy"
+        - "Tell me about recent stock market trends"
+        - "What are bonds and how do they work?"
+        """)
+        
+        chatbot = gr.Chatbot(
+            height=500,
+            placeholder="Financial AI Agent is ready to help with your finance questions!"
+        )
+        
+        msg = gr.Textbox(
+            label="Ask a financial question",
+            placeholder="e.g., What is inflation and how does it affect the stock market?",
+            lines=2
+        )
+        
+        clear = gr.Button("Clear Chat")
+        
+        def user(user_message, history):
+            return "", history + [[user_message, None]]
+        
+        def bot(history):
+            if history and history[-1][1] is None:
+                user_message = history[-1][0]
+                bot_message = finance_chat_interface(user_message, history[:-1])
+                history[-1][1] = bot_message
+            return history
+        
+        msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then(
+            bot, chatbot, chatbot
+        )
+        
+        clear.click(lambda: None, None, chatbot, queue=False)
+        
+        gr.Markdown("""
+        ### 📝 Notes:
+        - This agent searches the web for current financial information
+        - Summaries are generated using AI and should be verified with official sources
+        - For investment decisions, always consult with qualified financial advisors
+        """)
+    
+    return demo
+
+if __name__ == "__main__":
+    demo = create_interface()
+    demo.launch(share=True)