Create app.py

app.py

@@ -0,0 +1,623 @@
+from langchain_groq import ChatGroq
+from langgraph.graph import StateGraph, START, END
+from IPython.display import Image, display, Markdown
+from typing_extensions import TypedDict
+from langgraph.constants import Send
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_community.tools.tavily_search import TavilySearchResults
+import os
+import getpass
+from typing import Annotated, List, Dict, Any
+import operator
+from pydantic import BaseModel, Field
+from datetime import datetime
+import requests
+from bs4 import BeautifulSoup
+import re
+import json
+import gradio as gr
+from langdetect import detect
+
+# Define models for structured output
+class NewsItem(BaseModel):
+    title: str = Field(description="Title of the AI news article")
+    url: str = Field(description="URL of the news article")
+    source: str = Field(description="Source website of the news")
+    description: str = Field(description="Brief description of the news article")
+
+class NewsResults(BaseModel):
+    news_items: List[NewsItem] = Field(description="List of AI news articles found")
+
+class Subsection(BaseModel):
+    title: str = Field(description="Title of the subsection (based on news item title)")
+    source: str = Field(description="Source of the news item")
+    url: str = Field(description="URL of the news item")
+    content: str = Field(description="Content for this subsection")
+
+class Section(BaseModel):
+    name: str = Field(description="Name for this section of the blog")
+    description: str = Field(description="Description for this section of the blog")
+    information: str = Field(description="Information which should be included in this section of the blog")
+    subsections: List[Subsection] = Field(description="Subsections for each news item in this category", default=[])
+
+class Sections(BaseModel):
+    sections: List[Section] = Field(description="List of sections for this blog")
+
+# State definitions
+class NewsState(TypedDict):
+    query: str
+    date: str
+    search_results: List[Dict[str, Any]]
+    news_items: List[Dict[str, Any]]
+    
+class BlogState(TypedDict):
+    content: str
+    sections: List[Section]
+    completed_sections: Annotated[List, operator.add]
+    final_report: str
+
+class WorkerState(TypedDict):
+    section: Section
+    completed_sections: Annotated[List, operator.add]
+
+class ArticleScraperState(TypedDict):
+    url: str
+    article_content: str
+
+# Helper function to detect English language
+def is_english(text):
+    try:
+        return detect(text) == 'en'
+    except:
+        # If detection fails, check for common English words
+        common_english_words = ['the', 'and', 'in', 'to', 'of', 'is', 'for', 'with', 'on', 'that']
+        text_lower = text.lower()
+        english_word_count = sum(1 for word in common_english_words if f" {word} " in f" {text_lower} ")
+        return english_word_count >= 3  # If at least 3 common English words are found
+
+# News search functions
+def search_ai_news(state: NewsState):
+    """Search for the latest AI news using Tavily"""
+    search_tool = TavilySearchResults(max_results=10)
+    
+    # Format today's date
+    today = state.get("date", datetime.now().strftime("%Y-%m-%d"))
+    
+    # Create search query with date to get recent news
+    query = f"latest artificial intelligence news {today} english"
+    
+    # Execute search
+    search_results = search_tool.invoke({"query": query})
+    
+    # Filter out YouTube results and non-English content
+    filtered_results = []
+    for result in search_results:
+        if "youtube.com" not in result.get("url", "").lower():
+            # Check if content is in English
+            content = result.get("content", "") + " " + result.get("title", "")
+            if is_english(content):
+                filtered_results.append(result)
+    
+    return {"search_results": filtered_results}
+
+def parse_news_items(state: NewsState):
+    """Parse search results into structured news items using a more robust approach"""
+    search_results = state["search_results"]
+    
+    # Format results for the LLM
+    formatted_results = "\n\n".join([
+        f"Title: {result.get('title', 'No title')}\n"
+        f"URL: {result.get('url', 'No URL')}\n"
+        f"Content: {result.get('content', 'No content')}"
+        for result in search_results
+    ])
+    
+    # Use a direct prompt instead of structured output
+    system_prompt = """
+    Extract AI news articles from these search results. Filter out any that aren't about artificial intelligence.
+    
+    For each relevant AI news article, provide:
+    - title: The title of the article
+    - url: The URL of the article
+    - source: The source website of the news
+    - description: A brief description of the article
+    
+    Format your response as a JSON list of objects. Only include the relevant fields, nothing else.
+    Example format:
+    [
+      {
+        "title": "New AI Development",
+        "url": "https://example.com/news/ai-dev",
+        "source": "Example News",
+        "description": "Description of the AI development"
+      }
+    ]
+    """
+    
+    # Get the response as a string
+    response = llm.invoke([
+        SystemMessage(content=system_prompt),
+        HumanMessage(content=f"Here are the search results:\n\n{formatted_results}")
+    ])
+    
+    # Extract the JSON part from the response
+    response_text = response.content
+    
+    # Find JSON list in the response
+    json_match = re.search(r'\[\s*\{.*\}\s*\]', response_text, re.DOTALL)
+    
+    news_items = []
+    if json_match:
+        try:
+            # Parse the JSON text
+            news_items = json.loads(json_match.group(0))
+        except json.JSONDecodeError:
+            # Fallback: create a simple item if JSON parsing fails
+            news_items = [{
+                "title": "AI News Roundup",
+                "url": "https://example.com/ai-news",
+                "source": "Various Sources",
+                "description": "Compilation of latest AI news from various sources."
+            }]
+    else:
+        # Create a default item if no JSON found
+        news_items = [{
+            "title": "AI News Roundup",
+            "url": "https://example.com/ai-news",
+            "source": "Various Sources",
+            "description": "Compilation of latest AI news from various sources."
+        }]
+    
+    return {"news_items": news_items}
+
+# Article scraping function
+def scrape_article_content(state: ArticleScraperState):
+    """Scrape the content from a news article URL"""
+    url = state["url"]
+    
+    try:
+        headers = {
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
+        }
+        response = requests.get(url, headers=headers, timeout=10)
+        response.raise_for_status()
+        
+        soup = BeautifulSoup(response.text, 'html.parser')
+        
+        # Extract article content
+        article_text = ""
+        
+        # Try to find the main article content
+        article = soup.find('article')
+        if article:
+            paragraphs = article.find_all('p')
+        else:
+            # Fallback to all paragraphs
+            paragraphs = soup.find_all('p')
+        
+        # Extract text from paragraphs
+        article_text = "\n\n".join([p.get_text().strip() for p in paragraphs])
+        
+        # Clean up the text
+        article_text = re.sub(r'\s+', ' ', article_text).strip()
+        
+        # Trim to reasonable length for LLM processing
+        if len(article_text) > 10000:
+            article_text = article_text[:10000] + "..."
+        
+        # Verify the content is in English
+        if not is_english(article_text[:500]):  # Check first 500 chars to save processing time
+            return {"article_content": "Content not in English or insufficient text to analyze."}
+            
+        return {"article_content": article_text}
+    
+    except Exception as e:
+        return {"article_content": f"Error scraping article: {str(e)}"}
+
+# Blog generation functions
+def orchestrator(state: BlogState):
+    """Orchestrator that generates a plan for the blog based on news items"""
+    
+    try:
+        # Parse the content to extract news items
+        content_lines = state['content'].split('\n\n')
+        news_items = []
+        current_item = {}
+        
+        for content_block in content_lines:
+            if content_block.startswith('TITLE:'):
+                # Start of a new item
+                if current_item and 'title' in current_item:
+                    news_items.append(current_item)
+                current_item = {}
+                
+                lines = content_block.split('\n')
+                for line in lines:
+                    if line.startswith('TITLE:'):
+                        current_item['title'] = line.replace('TITLE:', '').strip()
+                    elif line.startswith('SOURCE:'):
+                        current_item['source'] = line.replace('SOURCE:', '').strip()
+                    elif line.startswith('URL:'):
+                        current_item['url'] = line.replace('URL:', '').strip()
+                    elif line.startswith('DESCRIPTION:'):
+                        current_item['description'] = line.replace('DESCRIPTION:', '').strip()
+                    elif line.startswith('CONTENT:'):
+                        current_item['content'] = line.replace('CONTENT:', '').strip()
+            elif 'content' in current_item:
+                # Add to existing content
+                current_item['content'] += ' ' + content_block
+        
+        # Add the last item
+        if current_item and 'title' in current_item:
+            news_items.append(current_item)
+        
+        # Group news items by category
+        ai_tech_items = []
+        ai_business_items = []
+        ai_research_items = []
+        
+        for item in news_items:
+            title = item.get('title', '').lower()
+            description = item.get('description', '').lower()
+            
+            # Simple categorization based on keywords
+            if any(kw in title + description for kw in ['business', 'market', 'company', 'investment', 'startup']):
+                ai_business_items.append(item)
+            elif any(kw in title + description for kw in ['research', 'study', 'paper', 'university']):
+                ai_research_items.append(item)
+            else:
+                ai_tech_items.append(item)
+        
+        # Create sections with subsections
+        sections = []
+        
+        # AI Technology section
+        if ai_tech_items:
+            tech_subsections = [
+                Subsection(
+                    title=item['title'],
+                    source=item['source'],
+                    url=item['url'],
+                    content=f"{item.get('description', '')} {item.get('content', '')[:500]}..."
+                ) for item in ai_tech_items
+            ]
+            
+            sections.append(Section(
+                name="AI Technology Developments",
+                description="Recent advancements in AI technology and applications",
+                information="Cover the latest developments in AI technology.",
+                subsections=tech_subsections
+            ))
+        
+        # AI Business section
+        if ai_business_items:
+            business_subsections = [
+                Subsection(
+                    title=item['title'],
+                    source=item['source'],
+                    url=item['url'],
+                    content=f"{item.get('description', '')} {item.get('content', '')[:500]}..."
+                ) for item in ai_business_items
+            ]
+            
+            sections.append(Section(
+                name="AI in Business",
+                description="How AI is transforming industries and markets",
+                information="Focus on business applications and market trends in AI.",
+                subsections=business_subsections
+            ))
+        
+        # AI Research section
+        if ai_research_items:
+            research_subsections = [
+                Subsection(
+                    title=item['title'],
+                    source=item['source'],
+                    url=item['url'],
+                    content=f"{item.get('description', '')} {item.get('content', '')[:500]}..."
+                ) for item in ai_research_items
+            ]
+            
+            sections.append(Section(
+                name="AI Research and Studies",
+                description="Latest research findings and academic work in AI",
+                information="Cover recent research papers and studies in AI.",
+                subsections=research_subsections
+            ))
+        
+        # If no items were categorized, create a general section
+        if not sections:
+            general_subsections = [
+                Subsection(
+                    title=item['title'],
+                    source=item['source'],
+                    url=item['url'],
+                    content=f"{item.get('description', '')} {item.get('content', '')[:500]}..."
+                ) for item in news_items
+            ]
+            
+            sections.append(Section(
+                name="Latest AI News",
+                description="Roundup of the latest AI news from around the web",
+                information="Cover a range of AI news topics.",
+                subsections=general_subsections
+            ))
+        
+        return {"sections": sections}
+    except Exception as e:
+        print(f"Error in orchestrator: {str(e)}")
+        # Fallback plan if structured output fails
+        fallback_sections = [
+            Section(
+                name="Latest AI Developments",
+                description="Overview of recent AI advancements and research",
+                information="Summarize the latest AI developments from the provided content.",
+                subsections=[]
+            )
+        ]
+        return {"sections": fallback_sections}
+
+def llm_call(state: WorkerState):
+    """Worker writes a section of the blog with subsections for each news item"""
+    
+    section = state['section']
+    
+    # Generate section header with ID for anchor linking
+    section_id = section.name.lower().replace(' ', '-')
+    section_header = f"## {section.name} {{#{section_id}}}\n\n{section.description}\n"
+    
+    # If there are subsections, process each one
+    subsections_content = ""
+    if section.subsections:
+        for idx, subsection in enumerate(section.subsections):
+            # Generate subsection using LLM
+            subsection_prompt = f"""
+Write a detailed subsection about this AI news item:
+Title: {subsection.title}
+Source: {subsection.source}
+URL: {subsection.url}
+
+Content to summarize and expand on:
+{subsection.content}
+
+Keep your response focused on the news item and make it engaging. Use markdown formatting.
+"""
+            
+            subsection_content = llm.invoke([
+                SystemMessage(content="You are writing a subsection for an AI news blog. Write in a professional but engaging style. Include key details and insights. Use markdown formatting."),
+                HumanMessage(content=subsection_prompt)
+            ])
+            
+            # Create a clean ID for the subsection
+            subsection_id = f"{section_id}-{idx+1}-{subsection.title.lower().replace(' ', '-').replace(':', '').replace('?', '').replace('!', '')}"
+            
+            # Format subsection with title and source
+            formatted_subsection = f"### {subsection.title} {{#{subsection_id}}}\n\n"
+            formatted_subsection += f"*Source: [{subsection.source}]({subsection.url})*\n\n"
+            formatted_subsection += subsection_content.content
+            
+            subsections_content += formatted_subsection + "\n\n"
+    else:
+        # If no subsections, generate the full section content
+        section_content = llm.invoke([
+            SystemMessage(content="Write a blog section following the provided name, description, and information. Include no preamble. Use markdown formatting."),
+            HumanMessage(content=f"Here is the section name: {section.name}\nDescription: {section.description}\nInformation: {section.information}")
+        ])
+        subsections_content = section_content.content
+    
+    # Combine section header and subsections
+    complete_section = section_header + subsections_content
+    
+    # Return the completed section
+    return {"completed_sections": [complete_section]}
+
+def synthesizer(state: BlogState):
+    """Synthesize full blog from sections with proper formatting and hierarchical TOC"""
+    
+    # List of completed sections
+    completed_sections = state["completed_sections"]
+    
+    # Format completed sections into a full blog post
+    completed_report = "\n\n".join(completed_sections)
+    
+    # Add title, date, and introduction
+    today = datetime.now().strftime("%Y-%m-%d")
+    blog_title = f"# AI News Roundup - {today}"
+    
+    # Generate a brief introduction
+    intro = llm.invoke([
+        SystemMessage(content="Write a brief introduction for an AI news roundup blog post. Keep it under 100 words. Be engaging and professional."),
+        HumanMessage(content=f"Today's date is {today}. Write a brief introduction for an AI news roundup.")
+    ])
+    
+    # Create hierarchical table of contents
+    table_of_contents = "## Table of Contents\n\n"
+    
+    # Find all section headings (## headings)
+    section_matches = re.findall(r'## (.*?) {#(.*?)}', completed_report)
+    
+    for i, (section_name, section_id) in enumerate(section_matches, 1):
+        # Add section to TOC
+        table_of_contents += f"{i}. [{section_name}](#{section_id})\n"
+        
+        # Find all subsections within this section
+        # Look for subsection headings (### headings) until the next section or end of text
+        section_start = completed_report.find(f"## {section_name}")
+        next_section_match = re.search(r'## ', completed_report[section_start+1:])
+        if next_section_match:
+            section_end = section_start + 1 + next_section_match.start()
+            section_text = completed_report[section_start:section_end]
+        else:
+            section_text = completed_report[section_start:]
+        
+        # Extract subsection headings and IDs
+        subsection_matches = re.findall(r'### (.*?) {#(.*?)}', section_text)
+        
+        for j, (subsection_name, subsection_id) in enumerate(subsection_matches, 1):
+            # Add subsection to TOC with proper indentation
+            table_of_contents += f"   {i}.{j}. [{subsection_name}](#{subsection_id})\n"
+    
+    final_report = f"{blog_title}\n\n{intro.content}\n\n{table_of_contents}\n\n---\n\n{completed_report}\n\n---\n\n*This AI News Roundup was automatically generated on {today}.*"
+    
+    return {"final_report": final_report}
+
+# Edge function to create workers for each section
+def assign_workers(state: BlogState):
+    """Assign a worker to each section in the plan"""
+    
+    # Kick off section writing in parallel
+    return [Send("llm_call", {"section": s}) for s in state["sections"]]
+
+# Main workflow functions
+def create_news_search_workflow():
+    """Create a workflow for searching and parsing AI news"""
+    workflow = StateGraph(NewsState)
+    
+    # Add nodes
+    workflow.add_node("search_ai_news", search_ai_news)
+    workflow.add_node("parse_news_items", parse_news_items)
+    
+    # Add edges
+    workflow.add_edge(START, "search_ai_news")
+    workflow.add_edge("search_ai_news", "parse_news_items")
+    workflow.add_edge("parse_news_items", END)
+    
+    return workflow.compile()
+
+def create_article_scraper_workflow():
+    """Create a workflow for scraping article content"""
+    workflow = StateGraph(ArticleScraperState)
+    
+    # Add node
+    workflow.add_node("scrape_article", scrape_article_content)
+    
+    # Add edges
+    workflow.add_edge(START, "scrape_article")
+    workflow.add_edge("scrape_article", END)
+    
+    return workflow.compile()
+
+def create_blog_generator_workflow():
+    """Create a workflow for generating the blog"""
+    workflow = StateGraph(BlogState)
+    
+    # Add nodes
+    workflow.add_node("orchestrator", orchestrator)
+    workflow.add_node("llm_call", llm_call)
+    workflow.add_node("synthesizer", synthesizer)
+    
+    # Add edges
+    workflow.add_edge(START, "orchestrator")
+    workflow.add_conditional_edges("orchestrator", assign_workers, ["llm_call"])
+    workflow.add_edge("llm_call", "synthesizer")
+    workflow.add_edge("synthesizer", END)
+    
+    return workflow.compile()
+
+def generate_ai_news_blog(groq_api_key=None, tavily_api_key=None, date=None):
+    """Main function to generate AI news blog"""
+    # Set API keys if provided
+    if groq_api_key:
+        os.environ["GROQ_API_KEY"] = groq_api_key
+    if tavily_api_key:
+        os.environ["TAVILY_API_KEY"] = tavily_api_key
+    
+    # Initialize LLM with the API key
+    global llm
+    llm = ChatGroq(model="qwen-2.5-32b")
+    
+    # Get date
+    if not date:
+        today = datetime.now().strftime("%Y-%m-%d")
+    else:
+        today = date
+    
+    # Step 1: Search for AI news
+    news_search = create_news_search_workflow()
+    news_results = news_search.invoke({"query": "latest artificial intelligence news", "date": today})
+    
+    print(f"Found {len(news_results['news_items'])} AI news items")
+    
+    # Step 2: Scrape content for each news item
+    article_scraper = create_article_scraper_workflow()
+    news_contents = []
+    
+    for item in news_results["news_items"]:
+        print(f"Scraping: {item['title']} from {item['source']}")
+        result = article_scraper.invoke({"url": item['url']})
+        
+        # Skip if not in English
+        if "not in English" in result["article_content"]:
+            print(f"Skipping non-English content: {item['title']}")
+            continue
+        
+        news_contents.append({
+            "title": item['title'],
+            "url": item['url'],
+            "source": item['source'],
+            "description": item['description'],
+            "content": result["article_content"]
+        })
+    
+    # Format news content for the blog generator
+    formatted_content = "\n\n".join([
+        f"TITLE: {item['title']}\nSOURCE: {item['source']}\nURL: {item['url']}\nDESCRIPTION: {item['description']}\nCONTENT: {item['content'][:2000]}..."
+        for item in news_contents
+    ])
+    
+    # Step 3: Generate the blog
+    blog_generator = create_blog_generator_workflow()
+    blog_result = blog_generator.invoke({
+        "content": formatted_content,
+        "completed_sections": []
+    })
+    
+    return blog_result["final_report"]
+
+# Gradio UI
+def create_gradio_interface():
+    """Create a Gradio interface for the AI News Blog Generator"""
+    
+    def run_generation(groq_key, tavily_key, selected_date):
+        if not groq_key or not tavily_key:
+            return "Please provide both API keys."
+        
+        try:
+            result = generate_ai_news_blog(groq_key, tavily_key, selected_date)
+            return result
+        except Exception as e:
+            return f"Error generating blog: {str(e)}"
+    
+    # Create the interface
+    with gr.Blocks(title="AI News Blog Generator") as demo:
+        gr.Markdown("# AI News Blog Generator")
+        gr.Markdown("Generate a daily roundup of AI news articles, categorized by topic.")
+        
+        with gr.Row():
+            with gr.Column():
+                groq_key = gr.Textbox(label="Groq API Key", placeholder="Enter your Groq API key", type="password")
+                tavily_key = gr.Textbox(label="Tavily API Key", placeholder="Enter your Tavily API key", type="password")
+                date_picker = gr.Textbox(label="Date (YYYY-MM-DD)", placeholder="Leave empty for today's date", 
+                                        value=datetime.now().strftime("%Y-%m-%d"))
+                generate_button = gr.Button("Generate AI News Blog")
+            
+            with gr.Column():
+                output_md = gr.Markdown("Your AI News Blog will appear here.")
+        
+        generate_button.click(
+            fn=run_generation,
+            inputs=[groq_key, tavily_key, date_picker],
+            outputs=output_md
+        )
+    
+    return demo
+
+# Run the entire pipeline
+if __name__ == "__main__":
+    try:
+        # Create and launch the Gradio interface
+        demo = create_gradio_interface()
+        demo.launch()
+        
+    except Exception as e:
+        print(f"Error running the pipeline: {str(e)}")