app.py

import streamlit as st
from dotenv import load_dotenv
import os
from langchain_groq import ChatGroq
from typing_extensions import TypedDict
from langgraph.graph import add_messages, StateGraph, END, START
from langchain_core.messages import AIMessage
from typing import Annotated, List, Dict, Any
from langdetect import detect
from langchain_community.tools.tavily_search import TavilySearchResults

## Langsmith Tracking
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"]="Blog Generator Agent"

# Define BlogState TypedDict
class BlogState(TypedDict):
    topic: str
    title: str
    search_results: Annotated[List[Dict[str, Any]], add_messages]
    blog_content: Annotated[List, add_messages]
    reviewed_content: Annotated[List, add_messages]
    is_blog_ready: str

# Initialize session state
if 'blog_state' not in st.session_state:
    st.session_state.blog_state = None
if 'graph' not in st.session_state:
    st.session_state.graph = None
if 'graph_image' not in st.session_state:
    st.session_state.graph_image = None

# Helper function to detect English language
def is_english(text):
    # Ensure we have enough text to analyze
    if not text or len(text.strip()) < 50:
        return False
        
    try:
        # Try primary language detection
        return detect(text) == 'en'
    except:
        # If detection fails, use a more robust approach
        common_english_words = ['the', 'and', 'in', 'to', 'of', 'is', 'for', 'with', 'on', 'that', 
                              'this', 'are', 'was', 'be', 'have', 'it', 'not', 'they', 'by', 'from']
        text_lower = text.lower()
        # Count occurrences of common English words
        english_word_count = sum(1 for word in common_english_words if f" {word} " in f" {text_lower} ")
        # Calculate ratio of English words to text length
        text_words = len(text_lower.split())
        if text_words == 0:  # Avoid division by zero
            return False
            
        english_ratio = english_word_count / min(20, text_words)  # Cap at 20 to avoid skew
        return english_word_count >= 5 or english_ratio > 0.25  # More stringent criteria
    
def init_graph(api_key: str):
    
    global llm
    llm  = ChatGroq(model="qwen-2.5-32b", api_key=api_key)
    
    builder = StateGraph(BlogState)
    
    builder.add_node("title_generator", generate_title) ## Generate Title
    builder.add_node("search_web", search_web) ## Search Web using Tavily based in the topic
    builder.add_node("content_generator", generate_content) ## Generate Content using the output of title_generator and search_web
    builder.add_node("content_reviewer", review_content) ## Review Content and generate feedback
    builder.add_node("quality_check", evaluate_content) ## Validate the content based on feedback and generate verdict

    builder.add_edge(START, "title_generator")
    builder.add_edge(START, "search_web")
    builder.add_edge("title_generator", "content_generator")
    builder.add_edge("search_web", "content_generator")
    builder.add_edge("content_generator", "content_reviewer")
    builder.add_edge("content_reviewer", "quality_check")
    
    builder.add_conditional_edges(
        "quality_check",
        route_based_on_verdict,
        {"Pass": END, "Fail": "content_generator"}
    )
    return builder.compile()

# Node functions with state management
def generate_title(state: BlogState):
    prompt = f"""Generate compelling blog title options about {state["topic"]} that are:
    - SEO-friendly
    - Attention-grabbing
    - Between 6-12 words"""
    
    
    response = llm.invoke(prompt)
    state["title"] = response.content.split("\n")[0].strip('"')
    return state

def search_web(state: BlogState):
    
    search_tool = TavilySearchResults(max_results=2)
    
    # Create search query with date to get recent news
    query = f"Latest data on {state['topic']}"
    
    # 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": [
        {
            "role": "system",
            "content": f"{result['title']}\n{result['content']}\n(Source: {result['url']})"
        }
        for result in filtered_results
        ]
    }


def generate_content(state: BlogState):
    prompt = f"""Write a comprehensive blog post titled "{state['title']}" and based on the web search results {state['search_results']} with:
    1. Engaging introduction with hook
    2. 3-5 subheadings with detailed content
    3. Practical examples/statistics
    4. Clear transitions between sections
    5. Actionable conclusion
    Style: Professional yet conversational (Flesch-Kincaid 60-70). Use markdown formatting"""
    
    with st.status("📝 Generating Content..."):
        response = llm.invoke(prompt)
        state["blog_content"].append(AIMessage(content=response.content))
        st.markdown(response.content)
    return state

def review_content(state: BlogState):
    content = state["blog_content"][-1].content
    prompt = f"""Critically review this blog content:
    - Clarity & Structure
    - Grammar & Style
    - SEO optimization
    - Reader engagement
    Provide specific improvement suggestions. Content:\n{content}"""
    
    with st.status("🔍 Reviewing Content..."):
        feedback = llm.invoke(prompt)
        state["reviewed_content"].append(AIMessage(content=feedback.content))
        st.write(feedback.content)
    return state

def evaluate_content(state: BlogState):
    content = state["blog_content"][-1].content
    feedback = state["reviewed_content"][-1].content
    
    prompt = f"""Evaluate blog content against editorial feedback (Pass/Fail):
    Content: {content}
    Feedback: {feedback}
    Answer only Pass or Fail:"""
    
    with st.status("✅ Evaluating Quality..."):
        response = llm.invoke(prompt)
        verdict = response.content.strip().upper()
        state["is_blog_ready"] = "Pass" if "PASS" or "Pass" in verdict else "Fail"
        state["reviewed_content"].append(AIMessage(
            content=f"Verdict: {response.content}"
        ))
        st.write(f"Final Verdict: **{state['is_blog_ready']}**")
    return state

def route_based_on_verdict(state: BlogState):
    return "Pass" if state["is_blog_ready"] == "Pass" else "Fail"

# Streamlit UI components
st.title("🚀 BlogForge Pro Agent")
st.markdown("""
**Smart Blog Generation with Auto-Refinement**  
*From first draft to final edit - AI-assisted writing meets professional standards*  
✓ SEO-Optimized Structure ✓ Grammar & Style Checks ✓ Feedback-Driven Revisions  
""")

# Sidebar components
with st.sidebar:

    st.subheader("Configuration")
        
    # Groq API Key Input
    api_key = st.text_input("Groq API Key:", 
                          type="password",
                          value=os.getenv("GROQ_API_KEY", ""))
    
     # Validate API key
    if not api_key:
        st.warning("⚠️ Please enter your GROQ API key to proceed. Don't have? refer : https://console.groq.com/keys ")
        
    # Groq API Key Input
    tavily_api_key = os.environ["TAVILY_API_KEY"] =  st.session_state["TAVILY_API_KEY"] = st.text_input("Tavily API Key:", 
                          type="password",
                          value=os.getenv("TAVILY_API_KEY", ""))
    
    # Validate API key
    if not tavily_api_key:
        st.warning("⚠️ Please enter your TAVILY_API_KEY key to proceed. Don't have? refer : https://app.tavily.com/home")
                    
    if st.button("Reset Session"):
        st.session_state.clear()
        st.rerun()
        
        
    st.subheader("Workflow Overview")
    st.image("workflow_graph.png")
        
# Main content
topic = st.text_input("Enter your blog topic:", placeholder="Generative AI in Healthcare")
generate_btn = st.button("Generate Blog Post")

if generate_btn:
    if not api_key:
        st.error("Please provide a Groq API key in the sidebar!")
        st.stop()
    
    if not topic:
        st.error("Please enter a blog topic!")
        st.stop()
    
    try:
        
        # Initialize and run graph
        st.session_state.graph = init_graph(api_key)
        
        st.session_state.blog_state = BlogState(
            topic=topic,
            title="",
            search_results=[],
            blog_content=[],
            reviewed_content=[],
            is_blog_ready=""
        )
        
        # Execute the graph
        final_state = st.session_state.graph.invoke(st.session_state.blog_state)
        st.session_state.blog_state = final_state
        
        
        # Display results
        st.success("Blog post generation complete!")
        st.markdown("---")
        st.subheader("Final Blog Post")
        st.markdown(final_state["blog_content"][-1].content)
        
        st.markdown("---")
        st.subheader("Generated Title")
        st.write(final_state["title"])
        
        st.markdown("---")
        st.subheader("Web Search Results")
        st.write(final_state["search_results"][-1].content)
        
        st.markdown("---")
        st.subheader("Quality Assurance Report")
        st.write(final_state["reviewed_content"][-1].content)
        
        if st.session_state.blog_state:
            st.markdown("---")
            st.subheader("Generation Status")
            st.write(f"**Topic:** {st.session_state.blog_state['topic']}")
            st.write(f"**Status**: {'✅ Approved' if st.session_state.blog_state['is_blog_ready'] == 'Pass' else '❌ Needs Revision'}")
            st.write(f"**Review Cycles**: {len(st.session_state.blog_state['reviewed_content']) - 1}")
        
    except Exception as e:
        st.error(f"Error in blog generation: {str(e)}")