Spaces:
Sleeping
Sleeping
| 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) |