Add initial implementation of news agent with llama_index integration

.gitignore

@@ -0,0 +1,4 @@
+venv
+.env
+.vscode
+__pycache__

app.py

@@ -1,11 +1,48 @@
+import os
+from typing import Dict, List
+
 import gradio as gr
+import requests
 from huggingface_hub import InferenceClient
+from smolagents import CodeAgent, OpenAIServerModel, tool, AzureOpenAIServerModel
+import openai
+from transformers import pipeline
+
+from tools import (add_topic, analyze_sentiment, generate_implications,
+                   get_lead_up_events, get_news, get_social_media_opinions,
+                   recognize_entities, remove_topic)
 
 """
 For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
 """
 client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
 
+model = AzureOpenAIServerModel(
+    model_id = os.environ.get("AZURE_OPENAI_MODEL"),
+    azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
+    api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
+    api_version=os.environ.get("OPENAI_API_VERSION")    
+)
+
+
+# import openai
+
+# client = openai.OpenAI(
+#     api_key=os.environ.get("SAMBANOVA_API_KEY"),
+#     base_url="https://api.sambanova.ai/v1",
+# )
+
+SAMBANOVA_API_KEY = os.environ.get("SAMBANOVA_API_KEY")
+SAMBANOVA_API_BASE_URL = "https://api.sambanova.ai/v1"
+
+
+def get_model(model_id):
+    return OpenAIServerModel(  # Using OpenAIServerModel for SambaNova
+        model_id=model_id,
+        api_key=SAMBANOVA_API_KEY,
+        api_base=SAMBANOVA_API_BASE_URL
+    )
+
 
 def respond(
     message,
@@ -40,15 +77,82 @@ def respond(
         yield response
 
 
+# Initialize the agent
+agent = CodeAgent(
+    # model=get_model('Llama-4-Maverick-17B-128E-Instruct'),
+    model=model,
+    system_prompt="""
+    You are a news assistant. 
+    You can add topics with 'add: topic', remove with 'remove: topic', and get news with 'get news'. 
+    
+    When getting news, fetch the latest articles for the current topics, analyze their sentiment and extract entities, then present them. 
+    After presenting news, ask if the user wants to know implications (e.g., 'implications for 1'), why it happened (e.g., 'why happened for 1'), or social media reactions (e.g., 'social media reaction for 1'). 
+    
+    For 'implications', generate possible implications. 
+    
+    For 'why happened', extract the key event, search Wikipedia for its history, and present a chronological list. 
+    
+    For 'social media reaction', search X for related posts, analyze their sentiment, and present the level of positive and negative opinions as 'low', 'medium', or 'high'.  
+    
+    Authorized imports:
+    {{authorized_imports}}
+    
+    Tools:
+    {{tool_descriptions}}
+
+    Managed Agents:
+    {{managed_agents_descriptions}}""",
+    # tools = []
+    tools=[add_topic, remove_topic, get_news, analyze_sentiment, recognize_entities,
+           generate_implications, get_lead_up_events, get_social_media_opinions],
+)
+
+# Gradio chat interface
+
+
+def chat_with_agent(user_input,
+                    history: list[tuple[str, str]],
+                    system_message,
+                    max_tokens,
+                    temperature,
+                    top_p):
+    response = agent.run(user_input)
+    # Format news response for clarity
+    if user_input.lower() == "get news" and isinstance(response, list):
+        formatted_response = "Here are the latest news for your topics:\n"
+        for article in response:
+            if "error" in article:
+                formatted_response += article["error"] + "\n"
+            else:
+                formatted_response += (
+                    f"{article['index']}. Title: '{article['title']}'\n"
+                    f"   Summary: '{article['summary']}'\n"
+                    f"   Sentiment: {article['sentiment']}\n"
+                    f"   Entities: {', '.join(article['entities']) if article['entities'] else 'None'}\n"
+                )
+        formatted_response += "Would you like to know implications, why it happened, or social media reactions for any of these articles? (e.g., 'implications for 1', 'why happened for 1', 'social media reaction for 1')"
+        return formatted_response
+    return response
+
+
+demo = gr.ChatInterface(
+    fn=chat_with_agent,
+    title="Personalized News Agent",
+    description="An agent that helps you manage your interests and get personalized news with sentiment, entity analysis, implications, background events, and social media opinion levels."
+)
+
 """
 For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
 """
-demo = gr.ChatInterface(
+demo2 = gr.ChatInterface(
     respond,
     additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
+        gr.Textbox(value="You are a friendly Chatbot.",
+                   label="System message"),
+        gr.Slider(minimum=1, maximum=2048, value=512,
+                  step=1, label="Max new tokens"),
+        gr.Slider(minimum=0.1, maximum=4.0, value=0.7,
+                  step=0.1, label="Temperature"),
         gr.Slider(
             minimum=0.1,
             maximum=1.0,

app_llamaindex.py

@@ -0,0 +1,55 @@
+import os
+from typing import Dict, List
+
+import gradio as gr
+import requests
+from huggingface_hub import InferenceClient
+# from smolagents import CodeAgent, OpenAIServerModel, tool, AzureOpenAIServerModel
+# import openai
+from transformers import pipeline
+
+from tools import (add_topic, analyze_sentiment, generate_implications,
+                   get_lead_up_events, get_news, get_social_media_opinions,
+                   recognize_entities, remove_topic)
+from llama_index.llms.azure_openai import AzureOpenAI
+from llama_index.core.agent.workflow import FunctionAgent
+from basic_llama_agent import BasicLammaAgent
+
+agent_instance = BasicLammaAgent()
+
+
+async def chat_with_agent(user_input,
+                    history: list[tuple[str, str]],
+                    system_message,
+                    max_tokens,
+                    temperature,
+                    top_p):
+    # Use llama_index agent instead of smolagents agent
+    response = await agent_instance(user_input)
+    # Format news response for clarity
+    if user_input.lower() == "get news" and isinstance(response, list):
+        formatted_response = "Here are the latest news for your topics:\n"
+        for article in response:
+            if "error" in article:
+                formatted_response += article["error"] + "\n"
+            else:
+                formatted_response += (
+                    f"{article['index']}. Title: '{article['title']}'\n"
+                    f"   Summary: '{article['summary']}'\n"
+                    f"   Sentiment: {article['sentiment']}\n"
+                    f"   Entities: {', '.join(article['entities']) if article['entities'] else 'None'}\n"
+                )
+        formatted_response += "Would you like to know implications, why it happened, or social media reactions for any of these articles? (e.g., 'implications for 1', 'why happened for 1', 'social media reaction for 1')"
+        return formatted_response
+    return response
+
+demo = gr.ChatInterface(
+    fn=chat_with_agent,
+    title="Personalized News Agent (llama_index)",
+    description="An agent that helps you manage your interests and get personalized news with sentiment, entity analysis, implications, background events, and social media opinion levels. (llama_index version)",
+    type="messages"
+)
+
+
+if __name__ == "__main__":
+    demo.launch()

basic_llama_agent.py

@@ -0,0 +1,49 @@
+import os
+
+from llama_index.llms.azure_openai import AzureOpenAI
+from llama_index.core.agent.workflow import FunctionAgent
+from tools import (add_topic, remove_topic, get_news, analyze_sentiment,
+                    recognize_entities, generate_implications, get_lead_up_events,
+                    get_social_media_opinions)
+from llms import llm_azure_openai
+
+SYSTEM_PROMPT = """
+You are a news assistant. 
+You can add topics with 'add: topic', remove with 'remove: topic', and get news with 'get news'. 
+
+When getting news, fetch the latest articles for the current topics, analyze their sentiment and extract entities, then present them. 
+After presenting news, ask if the user wants to know implications (e.g., 'implications for 1'), why it happened (e.g., 'why happened for 1'), or social media reactions (e.g., 'social media reaction for 1'). 
+
+For 'implications', generate possible implications. 
+
+For 'why happened', extract the key event, search Wikipedia for its history, and present a chronological list. 
+
+For 'social media reaction', search X for related posts, analyze their sentiment, and present the level of positive and negative opinions as 'low', 'medium', or 'high'.  
+"""
+
+
+class BasicLammaAgent:
+    def __init__(self):
+        self.llm = llm_azure_openai
+
+        self.agent = FunctionAgent(
+            llm=self.llm,
+            system_prompt=SYSTEM_PROMPT,
+            # Add tools if llama_index agent supports tool usage
+            # tools=[]
+            tools=[add_topic, remove_topic, get_news, analyze_sentiment, recognize_entities,
+               generate_implications, get_lead_up_events, get_social_media_opinions],
+        )
+
+    async def __call__(self, q: str):
+        response = await self.agent.run(user_msg=q)
+
+        # Extract final output message from AgentOutput
+        if hasattr(response, "final_output") and hasattr(response.final_output, "content"):
+            return response.final_output.content
+
+        # Fallbacks for unexpected formats
+        if isinstance(response, str):
+            return response
+
+        return str(response)

llms.py

@@ -0,0 +1,12 @@
+import os
+
+from llama_index.llms.azure_openai import AzureOpenAI
+
+llm_azure_openai = AzureOpenAI(
+    engine=os.environ.get("AZURE_OPENAI_MODEL"),
+    model="gpt-4o-mini",
+    temperature=0.0,
+    azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
+    api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
+    api_version=os.environ.get("AZURE_OPENAI_API_VERSION"),
+)

requirements.txt

@@ -1 +1,8 @@
-huggingface_hub==0.25.2
+huggingface_hub==0.25.2
+gradio
+smolagents
+transformers
+requests
+openai
+llama-index
+llama-index-llms-azure-openai

tools.py

@@ -0,0 +1,309 @@
+import os
+from typing import Dict, List
+
+import requests
+
+from llms import llm_azure_openai
+
+from llama_index.core.llms import ChatMessage
+
+# Global state for simplicity (use database in production)
+topics = []
+last_news = []
+
+# Initialize pipelines for sentiment analysis, NER, and generative QA
+
+# Tool to add a topic to the interest list
+
+
+def add_topic(new_topic: str) -> str:
+    """
+    Adds a new topic to the global topics list.
+
+    Args:
+        new_topic: The topic to be added.
+
+    Returns:
+        str: A confirmation message indicating the topic was added.
+    """
+    global topics
+    topics.append(new_topic)
+    return f"Added {new_topic} to topics."
+
+# Tool to remove a topic from the interest list
+
+
+def remove_topic(topic: str) -> str:
+    """
+    Removes a specified topic from the global topics list.
+
+    Args:
+        topic: The topic to be removed from the topics list.
+
+    Returns:
+        str: A message indicating whether the topic was removed or not found.
+    """
+    global topics
+    if topic in topics:
+        topics.remove(topic)
+        return f"Removed {topic} from topics."
+    else:
+        return f"{topic} not found in topics."
+
+# Tool to fetch news articles using NewsAPI
+
+
+def get_news(topics: List[str]) -> List[Dict]:
+    """
+    Fetches news articles related to the specified topics using the NewsAPI, analyzes their sentiment and named entities, and returns a list of processed news items.
+
+    Args:
+        topics: A list of topics to search for in news articles.
+
+    Returns:
+        List[Dict]: A list of dictionaries, each containing:
+            - index (int): The position of the article in the result list.
+            - title (str): The title of the news article.
+            - summary (str): The description or summary of the article.
+            - sentiment: The sentiment analysis result of the summary.
+            - entities: The named entities recognized in the summary.
+        If an error occurs during the API request, returns a list with a single dictionary containing an "error" key and the error message.
+
+    Raises:
+        None. All exceptions are caught and returned as error messages in the result.
+    """
+    global last_news
+    api_key = os.environ.get("NEWS_API_KEY")
+    base_url = "https://newsapi.org/v2/everything"
+    queries = " OR ".join(topics)
+    params = {
+        "q": queries,
+        "apiKey": api_key,
+        "pageSize": 10,
+    }
+    try:
+        response = requests.get(base_url, params=params)
+        response.raise_for_status()
+        data = response.json()
+        articles = data.get("articles", [])
+        last_news = []
+        for idx, article in enumerate(articles, 1):
+            summary = article.get("description", "No description available")
+            sentiment = analyze_sentiment(summary)
+            entities = recognize_entities(summary)
+            last_news.append({
+                "index": idx,
+                "title": article.get("title", "No title"),
+                "summary": summary,
+                "sentiment": sentiment,
+                "entities": entities
+            })
+        return last_news
+    except requests.RequestException as e:
+        return [{"error": f"Failed to fetch news: {str(e)}"}]
+
+# Tool for sentiment analysis
+
+
+def analyze_sentiment(text: str) -> str:
+    """
+    Analyzes the sentiment of the given text and returns the sentiment label.
+
+    Args:
+        text: The input text to analyze.
+
+    Returns:
+        str: The sentiment label, such as 'positive', 'negative', or 'neutral'.
+    """
+    return 'positive'
+
+# Tool for named entity recognition
+
+
+def recognize_entities(text: str) -> List[str]:
+    """
+    Recognizes named entities in the given text.
+
+    Args:
+        text: The input text in which to recognize entities.
+
+    Returns:
+        List[str]: A list of recognized entity names as strings.
+    """
+    dummy_entities = ["Entity1", "Entity2",
+                      "Entity3"]  # Dummy entities for testing
+    return dummy_entities
+
+# Tool to generate implications for an article
+
+
+def generate_implications(article_index: int) -> str:
+    """
+    Generates a string describing the possible implications of a news article based on its index.
+
+    Args:
+        article_index: The 1-based index of the article in the global `last_news` list.
+
+    Returns:
+        str: A message containing the implications for the specified article, or an error message if the index is invalid.
+    """
+    global last_news
+    if not (1 <= article_index <= len(last_news)):
+        return "Invalid article index."
+    article = last_news[article_index - 1]
+    summary = article["summary"]
+    prompt = f"question: What are the possible implications of this news? context: {summary}"
+    try:
+        result = llm_azure_openai.chat(
+            messages=[ChatMessage(role="user", content=prompt)]
+        )
+    except Exception as e:
+        return f"Error generating implications: {str(e)}"
+    return f"Implications for article {article_index}: {result.message.content}"
+
+
+def web_search(query: str) -> List[Dict]:
+    """
+    Performs a web search and returns a list of results.
+
+    Args:
+        query: The search query string.
+
+    Returns:
+        List[Dict]: A list of dictionaries containing search results, each with 'title', 'snippet', and 'url'.
+    """
+    # Dummy implementation for testing
+    return [
+        {"title": "Example Result 1", "snippet": "This is an example snippet.",
+            "url": "http://example.com/1"},
+        {"title": "Example Result 2", "snippet": "This is another example snippet.",
+            "url": "http://example.com/2"},
+    ]
+
+
+def browse_page(url: str, query: str) -> str:
+    """
+    Fetches the content of a web page and returns it as a string.
+
+    Args:
+        url: The URL of the web page to fetch.
+        query: A query string to search within the page content.
+
+    Returns:
+        str: The content of the web page, or an error message if the page cannot be fetched.
+    """
+    try:
+        response = requests.get(url)
+        response.raise_for_status()
+        return response.text  # Return the raw HTML content for simplicity
+    except requests.RequestException as e:
+        return f"Error fetching page: {str(e)}"
+
+# Tool to get chronological events leading to an article
+
+
+def get_lead_up_events(article_index: int) -> str:
+    """
+    Retrieves a brief timeline or background of events leading up to a news article's topic.
+
+    Given the index of an article, this function attempts to find a Wikipedia page related to the article's title,
+    extracts historical events (preferably from a "history" section), and returns a summary of up to five relevant events.
+    If no Wikipedia timeline is found, it falls back to a general web search and provides background snippets.
+
+    Args:
+        article_index: The 1-based index of the article in the global `last_news` list.
+
+    Returns:
+        str: A formatted string summarizing the lead-up events or background information for the article's topic.
+    """
+    global last_news
+    if not (1 <= article_index <= len(last_news)):
+        return "Invalid article index."
+    article = last_news[article_index - 1]
+    title = article["title"]
+    # Search Wikipedia first
+    search_results = web_search(f"{title} Wikipedia")
+    wiki_url = None
+    for result in search_results:
+        if "wikipedia.org" in result["url"]:
+            wiki_url = result["url"]
+            break
+    if wiki_url:
+        history_content = browse_page(wiki_url, query="history")
+        # Simple parsing for timeline (assumes content contains chronological events)
+        events = []
+        for line in history_content.split("\n"):
+            if line.strip() and any(year in line for year in [str(y) for y in range(2000, 2026)]):
+                events.append(line.strip())
+        if events:
+            return f"Events leading up to '{title}':\n" + "\n".join(f"- {event}" for event in events[:5]) + f"\n(Source: {wiki_url})"
+        else:
+            return f"No clear timeline found on Wikipedia for '{title}'."
+    else:
+        # Fallback to general web search
+        search_results = web_search(f"background of {title}")
+        results = [
+            f"- {result['title']}: {result['snippet']}" for result in search_results[:3]]
+        return f"No Wikipedia timeline found for '{title}'. General background:\n" + "\n".join(results)
+
+
+def x_search(query: str) -> List[Dict]:
+    """
+    Searches social media (e.g., X/Twitter) for posts related to a query.
+
+    Args:
+        query: The search query string.
+
+    Returns:
+        List[Dict]: A list of dictionaries containing social media posts, each with 'content'.
+    """
+    # Dummy implementation for testing
+    return [
+        {"content": "This is a positive post about the event."},
+        {"content": "This is a negative post about the event."},
+        {"content": "Neutral post with no strong sentiment."},
+    ]
+
+# Tool to get social media opinion levels
+
+
+def get_social_media_opinions(article_index: int) -> str:
+    """
+    Analyzes social media opinions related to a news article by its index.
+
+    Given the index of a news article and a search function, this function retrieves relevant social media posts,
+    analyzes their sentiment, and categorizes the number of positive and negative opinions as 'low', 'medium', or 'high'.
+
+    Args:
+        article_index: The 1-based index of the article in the global `last_news` list.
+
+    Returns:
+        str: A summary string indicating the categorized number of positive and negative opinions about the event.
+    """
+    global last_news
+    if not (1 <= article_index <= len(last_news)):
+        return "Invalid article index."
+    article = last_news[article_index - 1]
+    title = article["title"]
+    # Use embedding-based search for relevant posts
+    posts = x_search(title)
+    positive_count = 0
+    negative_count = 0
+    for post in posts:
+        sentiment = analyze_sentiment(post["content"])
+        if sentiment == "POSITIVE":
+            positive_count += 1
+        elif sentiment == "NEGATIVE":
+            negative_count += 1
+    # Categorize counts
+
+    def categorize(count):
+        if count <= 10:
+            return "low"
+        elif count <= 30:
+            return "medium"
+        else:
+            return "high"
+    positive_level = categorize(positive_count)
+    negative_level = categorize(negative_count)
+    return f"There are a {positive_level} number of positive opinions and a {negative_level} number of negative opinions about this event."