Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| import pandas as pd | |
| from typing import Union, List, Dict | |
| from groq import Groq | |
| import os | |
| from duckduckgo_search import DDGS | |
| class DuckDuckGoSearch: | |
| """ | |
| Custom DuckDuckGo search implementation with robust error handling and result processing. | |
| Uses the duckduckgo_search library to fetch and format news results. | |
| """ | |
| def __init__(self): | |
| # Initialize the DuckDuckGo search session | |
| self.ddgs = DDGS() | |
| def __call__(self, query: str, max_results: int = 5) -> str: | |
| try: | |
| # Perform the search and get results | |
| # The news method is more appropriate for recent news analysis | |
| search_results = list(self.ddgs.news( | |
| query, | |
| max_results=max_results, | |
| region='wt-wt', # Worldwide results | |
| safesearch='on' | |
| )) | |
| if not search_results: | |
| return "No results found. Try modifying your search query." | |
| # Format the results into a readable string | |
| formatted_results = [] | |
| for idx, result in enumerate(search_results, 1): | |
| # Extract available fields with fallbacks for missing data | |
| title = result.get('title', 'No title available') | |
| snippet = result.get('body', result.get('snippet', 'No description available')) | |
| source = result.get('source', 'Unknown source') | |
| url = result.get('url', result.get('link', 'No link available')) | |
| date = result.get('date', 'Date not available') | |
| # Format each result with available information | |
| formatted_results.append( | |
| f"{idx}. Title: {title}\n" | |
| f" Date: {date}\n" | |
| f" Source: {source}\n" | |
| f" Summary: {snippet}\n" | |
| f" URL: {url}\n" | |
| ) | |
| return "\n".join(formatted_results) | |
| except Exception as e: | |
| # Provide detailed error information for debugging | |
| error_msg = f"Search error: {str(e)}\nTry again with a different search term or check your internet connection." | |
| print(f"DuckDuckGo search error: {str(e)}") # For logging | |
| return error_msg | |
| class GroqLLM: | |
| """ | |
| LLM interface using Groq's LLama model. | |
| Handles API communication and response processing. | |
| """ | |
| def __init__(self, model_name="llama-3.1-8B-Instant"): | |
| self.client = Groq(api_key=os.environ.get("GROQ_API_KEY")) | |
| self.model_name = model_name | |
| def __call__(self, prompt: Union[str, dict, List[Dict]]) -> str: | |
| try: | |
| # Convert prompt to string if it's a complex structure | |
| prompt_str = str(prompt) if isinstance(prompt, (dict, list)) else prompt | |
| # Make API call to Groq | |
| completion = self.client.chat.completions.create( | |
| model=self.model_name, | |
| messages=[{ | |
| "role": "user", | |
| "content": prompt_str | |
| }], | |
| temperature=0.7, | |
| max_tokens=1024, | |
| stream=False | |
| ) | |
| return completion.choices[0].message.content if completion.choices else "Error: No response generated" | |
| except Exception as e: | |
| error_msg = f"Error generating response: {str(e)}" | |
| print(error_msg) # For logging | |
| return error_msg | |
| def create_analysis_prompt(topic: str, search_results: str) -> str: | |
| """ | |
| Creates a detailed prompt for news analysis, structuring the request | |
| to get comprehensive and well-organized results from the LLM. | |
| """ | |
| return f"""Analyze the following news information about {topic}. | |
| Search Results: {search_results} | |
| Please provide a comprehensive analysis including: | |
| 1. Key Points Summary: | |
| - Main events and developments | |
| - Critical updates and changes | |
| 2. Stakeholder Analysis: | |
| - Primary parties involved | |
| - Their roles and positions | |
| 3. Impact Assessment: | |
| - Immediate implications | |
| - Potential long-term effects | |
| - Broader context and significance | |
| 4. Multiple Perspectives: | |
| - Different viewpoints on the issue | |
| - Areas of agreement and contention | |
| 5. Fact Check & Reliability: | |
| - Verification of major claims | |
| - Consistency across sources | |
| - Source credibility assessment | |
| Please format the analysis in a clear, journalistic style with section headers.""" | |
| def log_agent_activity(prompt: str, result: str, agent_name: str): | |
| """ | |
| Creates an expandable log of agent activities in the Streamlit interface | |
| for transparency and debugging purposes. | |
| """ | |
| with st.expander("View Agent Activity Log"): | |
| st.write(f"### Agent Activity ({agent_name}):") | |
| st.write("**Input Prompt:**") | |
| st.code(prompt, language="text") | |
| st.write("**Analysis Output:**") | |
| st.code(result, language="text") | |
| # Initialize Streamlit app | |
| st.set_page_config(page_title="News Analysis Tool", layout="wide") | |
| # Title and description | |
| st.title("Z-Agent🔍 AI News Analysis Tool") | |
| st.write(""" | |
| This tool combines the power of Ai Agent. Get comprehensive insights and multiple | |
| perspectives on any news topic. | |
| """) | |
| # Initialize the components | |
| try: | |
| # Initialize LLM and search tool | |
| llm = GroqLLM() | |
| search_tool = DuckDuckGoSearch() | |
| # Input section | |
| news_topic = st.text_input( | |
| "Enter News Topic or Query:", | |
| placeholder="E.g., Recent developments in renewable energy" | |
| ) | |
| # Analysis options | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| search_depth = st.slider( | |
| "Search Depth (number of results)", | |
| min_value=3, | |
| max_value=10, | |
| value=5 | |
| ) | |
| with col2: | |
| analysis_type = st.selectbox( | |
| "Analysis Type", | |
| ["Comprehensive", "Quick Summary", "Technical", "Simplified"] | |
| ) | |
| # Generate analysis button | |
| if st.button("Analyze News"): | |
| if news_topic: | |
| with st.spinner("Gathering information and analyzing..."): | |
| try: | |
| # Show search progress | |
| search_placeholder = st.empty() | |
| search_placeholder.info("Searching for recent news...") | |
| # Perform search | |
| search_results = search_tool( | |
| f"Latest news about {news_topic} last 7 days", | |
| max_results=search_depth | |
| ) | |
| if not search_results.startswith(("Search error", "No results")): | |
| # Update progress | |
| search_placeholder.info("Analyzing search results...") | |
| # Create analysis prompt | |
| analysis_prompt = create_analysis_prompt(news_topic, search_results) | |
| # Get analysis from LLM | |
| analysis_result = llm(analysis_prompt) | |
| # Clear progress messages | |
| search_placeholder.empty() | |
| # Display results | |
| st.subheader("📊 Analysis Results") | |
| st.markdown(analysis_result) | |
| # Log the activity | |
| log_agent_activity( | |
| analysis_prompt, | |
| analysis_result, | |
| "News Analysis Agent" | |
| ) | |
| else: | |
| search_placeholder.empty() | |
| st.error(search_results) | |
| except Exception as e: | |
| st.error(f"An error occurred during analysis: {str(e)}") | |
| else: | |
| st.warning("Please enter a news topic to analyze.") | |
| # Add helpful tips | |
| with st.expander("💡 Tips for Better Results"): | |
| st.write(""" | |
| - Be specific with your topic for more focused analysis | |
| - Use keywords related to recent events for timely information | |
| - Consider including timeframes in your query | |
| - Try different analysis types for various perspectives | |
| - For complex topics, start with a broader search and then narrow down | |
| """) | |
| except Exception as e: | |
| st.error(f""" | |
| Failed to initialize the application: {str(e)} | |
| Please ensure: | |
| 1. Your GROQ_API_KEY is properly set in environment variables | |
| 2. All required packages are installed: | |
| - pip install streamlit groq duckduckgo-search | |
| 3. You have internet connectivity for DuckDuckGo searches | |
| """) | |
| # Footer | |
| st.markdown("---") | |
| st.caption( | |
| "Powered by Groq LLama 3.1 8B Instant, DuckDuckGo, and Streamlit | " | |
| "Created for news analysis and research purposes" | |
| ) |