My Journey into Agentic AI Development: AI Newsroom

from typing import List, Dict, Annotated

from langgraph.graph import add_messages
from typing_extensions import TypedDict


class NewsroomState(TypedDict):
    trending_topics: List[Dict[str, str]]  # List of trendy topics from Researcher
    editors: List[str]  # Editor names to persona prompt content
    topic_claims: Dict[str, List[str]]  # Topic to list of editors claiming it
    assignments: Dict[str, str]  # Editor to assigned topic
    articles: Dict[str, str]  # Editor to written article
    messages: Annotated[list, add_messages]

def build_workflow(use_mock_topics:bool=False):
    # Create agents
    advanced_research_llm = AdvancedResearcherLlm(prompt_path='prompts/advanced_researcher.txt')
    search_tools = get_search_tools()
    # Use the new bind_tools method for cleaner syntax
    advanced_research_llm.bind_tools(search_tools)
    # Create standardized agent nodes with explicit data flow
    researcher_node = advanced_research_llm.create_node(
        output_field='trending_topics'
    )
    search_tool_node = ToolNode(search_tools)

    graph_builder = StateGraph(NewsroomState)

    graph_builder.add_node(ADVANCED_RESEARCH_NODE, researcher_node)
    graph_builder.add_node(ADVANCED_WEB_SEARCH_NODE, search_tool_node)
    graph_builder.add_node(CHIEF_EDITOR_NODE, chief_editor_node)
    graph_builder.add_node(EDITOR_WRITE_NODE, editor_write_node)

    if not use_mock_topics:
        graph_builder.add_edge(START, ADVANCED_RESEARCH_NODE)
    else:
        graph_builder.add_edge(START, CHIEF_EDITOR_NODE)

    def research_route(state: NewsroomState) -> str:
        """Route after research based on tool calls"""
        try:
            last_message = state["messages"][-1] if state["messages"] else None
            logger.info(f"ROUTE CHECK: Last message type: {type(last_message)}")

            # Pretty print the message content if it exists
            if hasattr(last_message, 'content') and last_message.content:
                logger.info(f"ROUTE CHECK: Message content: {last_message.content}")

            logger.info(f"ROUTE CHECK: Trending topics so far: {json.dumps(state.get('trending_topics'), indent=2, default=str)}")

            if hasattr(last_message, "tool_calls") and last_message.tool_calls:
                logger.info(f"Tool calls detected, routing to web search")
                logger.info(f"Tool Calls:n{prettify_tool_calls(last_message.tool_calls)}")
                return ADVANCED_WEB_SEARCH_NODE
            return CHIEF_EDITOR_NODE
        except (IndexError, AttributeError) as e:
            logger.warning(f"Routing error: {str(e)}, defaulting to chief editor node")
            return CHIEF_EDITOR_NODE

    # Add conditional routing from researcher
    graph_builder.add_conditional_edges(
        ADVANCED_RESEARCH_NODE,
        research_route,
        {
            ADVANCED_WEB_SEARCH_NODE: ADVANCED_WEB_SEARCH_NODE,
            CHIEF_EDITOR_NODE: CHIEF_EDITOR_NODE
        }
    )
    # Add edge from web search back to researcher (creates the loop)
    graph_builder.add_edge(ADVANCED_WEB_SEARCH_NODE, ADVANCED_RESEARCH_NODE)
    graph_builder.add_edge(CHIEF_EDITOR_NODE, EDITOR_WRITE_NODE)

    # Add edge from editor_write to end
    graph_builder.add_edge(EDITOR_WRITE_NODE, END)

    # Compile the graph before returning
    return graph_builder.compile()

from langchain_tavily import TavilySearch

from tools.tool_secrets import PROD_TAVILY_API_KEY


def get_search_tools():
    """Get optimized search tools for different use cases"""
    # Default tool for general viral content discovery
    tavily_search_tool = _get_tavily_search_tool()

    # Specialized tools for different trending content types
    viral_social_tool = _get_viral_search_tool("social")
    trending_news_tool = _get_viral_search_tool("news")
    tech_trends_tool = _get_viral_search_tool("tech")
    finance_trends_tool = _get_viral_search_tool("finance")
    celebrities_trends_tool = _get_viral_search_tool("celebrities")
    entertainment_trends_tool = _get_viral_search_tool("entertainment")

    tools = [tavily_search_tool, viral_social_tool, trending_news_tool, tech_trends_tool, finance_trends_tool,
             celebrities_trends_tool, entertainment_trends_tool]
    return tools


def _get_tavily_search_tool():
    """Optimized default search tool for viral and trending content"""
    return TavilySearch(
        max_results=10,  # Reduced from 20 for efficiency
        tavily_api_key=PROD_TAVILY_API_KEY,
        search_depth="advanced",  # Changed from basic for better context
        description="Search the web for viral and trending topics. Optimized for real-time discovery of trending content, viral stories, and emerging topics with time-sensitive filtering.",
        include_images=True,  # Visual content crucial for viral trends
        include_answer=True,  # Get direct answers for trending topics
        time_range="day",  # Focus on most recent viral content
        topic="general"  # Best for discovering diverse trending topics
    )


def _get_viral_search_tool(
        trend_focus: str = "general",
        max_results: int = 10,
        time_sensitivity: str = "day"
):
    """
    Specialized Tavily search tool for different types of viral content

    Args:
        trend_focus: "general", "news", "social", "tech", "finance"
        max_results: Number of results (5-15 recommended for efficiency)
        time_sensitivity: "day", "week", "month"
    """

    # Base configuration optimized for viral content
    config = {
        "max_results": max_results,
        "tavily_api_key": PROD_TAVILY_API_KEY,
        "search_depth": "advanced",
        "include_images": True,
        "include_answer": True,
        "time_range": time_sensitivity,
    }

    # Trend-specific optimizations
    trend_configs = {
        "social": {
            "include_domains": ["tiktok.com", "instagram.com", "youtube.com", "twitter.com", "reddit.com"],
            "description": "Discover viral social media trends, trending hashtags, and viral social content from major platforms"
        },
        "news": {
            "topic": "news",
            "time_range": "day",
            "description": "Find breaking news, viral news stories, and trending current events"
        },
        "tech": {
            "include_domains": ["techcrunch.com", "theverge.com", "reddit.com", "hackernews.com"],
            "description": "Track viral tech trends, product launches, and trending technology discussions"
        },
        "finance": {
            "topic": "finance",
            "include_domains": ["bloomberg.com", "reuters.com", "reddit.com"],
            "description": "Find trending financial news, viral market movements, and trending investment topics"
        },
        "celebrities": {
            "include_domains": ["tmz.com", "people.com", "eonline.com", "usmagazine.com", "extratv.com",
                                "instagram.com", "twitter.com", "reddit.com"],
            "time_range": "day",
            "description": "Track celebrity news, viral celebrity moments, trending celebrity gossip, and celebrity social media activity"
        },
        "entertainment": {
            "include_domains": ["variety.com", "hollywoodreporter.com", "deadline.com", "ew.com", "imdb.com",
                                "youtube.com", "netflix.com", "reddit.com"],
            "time_range": "day",
            "description": "Discover trending entertainment news, viral movie/TV content, music trends, streaming hits, and entertainment industry buzz"
        },
        "general": {
            "topic": "general",
            "description": "Search for viral content and trending topics across all categories"
        }
    }

    config.update(trend_configs.get(trend_focus, trend_configs["general"]))

    return TavilySearch(**config)

graph_builder = StateGraph(NewsroomState)

graph_builder.add_node(ADVANCED_RESEARCH_NODE, researcher_node)
graph_builder.add_node(ADVANCED_WEB_SEARCH_NODE, search_tool_node)
graph_builder.add_node(CHIEF_EDITOR_NODE, chief_editor_node)
graph_builder.add_node(EDITOR_WRITE_NODE, editor_write_node)

if not use_mock_topics:
    graph_builder.add_edge(START, ADVANCED_RESEARCH_NODE)
else:
    graph_builder.add_edge(START, CHIEF_EDITOR_NODE)
.......
..#(skip the conditional edge part as it is same as before)
.......
graph_builder.add_edge(ADVANCED_WEB_SEARCH_NODE, ADVANCED_RESEARCH_NODE)
graph_builder.add_edge(CHIEF_EDITOR_NODE, EDITOR_WRITE_NODE)

# Add edge from editor_write to end
graph_builder.add_edge(EDITOR_WRITE_NODE, END)

def chief_editor_node(state: NewsroomState) -> NewsroomState:
    """
    Enhanced chief editor node that processes the state and handles topic distribution.
    This node analyzes the current state, creates editor LLMs, and lets them claim topics.
    """
    logger.info("👔 CHIEF EDITOR NODE: Processing state and distributing topics")

    try:
        # Extract data from state
        trending_topics = state.get("trending_topics", [])
        editors = state.get("editors", [])

        # Check if we have trending topics
        if not trending_topics:
            logger.warning("❌ No trending topics found - workflow incomplete")
            return state

        # Check if we have editors
        if not editors:
            logger.warning("❌ No editors found - workflow incomplete")
            return state

        # If we have topics and no tool calls, we're ready for distribution
        logger.info("✅ State ready for topic distribution")

        # Prepare topic titles for distribution
        topic_titles = [topic.get("title", "") for topic in trending_topics if topic.get("title")]

        # Initialize topic claims tracking
        topic_claims = {title: [] for title in topic_titles}

        # Loop through all editors and let them claim topics
        logger.info("🤖 Starting topic distribution to editors...")

        llm = init_chat_model(model_provider="azure_openai", model=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"])

        for editor_name in editors:
            try:
                logger.info(f"🤖 Creating agent for editor: {editor_name}")

                # Create dynamic prompt for topic selection
                selection_prompt = get_editor_prompt(
                    editor_name=editor_name.lower(),
                    task_type='topic_selection',
                    trending_topics=topic_titles
                )

                # Get editor's response
                response = llm.invoke([HumanMessage(content=selection_prompt)])

                if response.content.startswith("CLAIM|"):
                    parts = response.content.split("|")
                    if len(parts) >= 3:
                        claimed_topic = parts[1].strip()
                        claiming_editor = parts[2].strip()

                        if claimed_topic in topic_claims:
                            topic_claims[claimed_topic].append(claiming_editor)
                            logger.info(f"✅ Editor {claiming_editor} claimed topic: {claimed_topic}")
                        else:
                            logger.warning(f"❌ Editor {claiming_editor} claimed invalid topic: {claimed_topic}")
                    else:
                        logger.warning(f"❌ Invalid CLAIM format from {editor_name}: {response.content}")
                else:
                    logger.warning(f"❌ {editor_name} did not use CLAIM format: {response.content}")

            except Exception as e:
                logger.error(f"❌ Failed to get response from editor {editor_name}: {str(e)}")

llm = init_chat_model(model_provider="azure_openai", model=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"])

for editor_name in editors:
    try:
        logger.info(f"🤖 Creating agent for editor: {editor_name}")

        # Create dynamic prompt for topic selection
        selection_prompt = get_editor_prompt(
            editor_name=editor_name.lower(),
            task_type='topic_selection',
            trending_topics=topic_titles
        )

        # Get editor's response
        response = llm.invoke([HumanMessage(content=selection_prompt)])

if response.content.startswith("CLAIM|"):
    parts = response.content.split("|")
    if len(parts) >= 3:
        claimed_topic = parts[1].strip()
        claiming_editor = parts[2].strip()

        if claimed_topic in topic_claims:
            topic_claims[claimed_topic].append(claiming_editor)
            logger.info(f"✅ Editor {claiming_editor} claimed topic: {claimed_topic}")
        else:
            logger.warning(f"❌ Editor {claiming_editor} claimed invalid topic: {claimed_topic}")
    else:
        logger.warning(f"❌ Invalid CLAIM format from {editor_name}: {response.content}")
else:
    logger.warning(f"❌ {editor_name} did not use CLAIM format: {response.content}")

# First pass: assign single claims
for topic, claiming_editors in topic_claims.items():
    if len(claiming_editors) == 1:
        editor = claiming_editors[0]
        assignments[editor] = topic
        unassigned_editors.discard(editor)
        logger.info(f"✅ Assigned {topic} to {editor}")

# Second pass: handle conflicts by keeping the first editor who claimed
for topic, claiming_editors in topic_claims.items():
    if len(claiming_editors) > 1:
        # Multiple editors claimed the same topic - assign to the first one
        first_editor = claiming_editors[0]
        if first_editor not in assignments:  # Only if not already assigned
            assignments[first_editor] = topic
            unassigned_editors.discard(first_editor)
            logger.info(f"✅ Resolved conflict: assigned {topic} to {first_editor} (first to claim)")
            # Remove this topic from other editors' claims
            for other_editor in claiming_editors[1:]:
                if other_editor in unassigned_editors:
                    logger.info(f"📝 {other_editor} will be assigned a different topic")

# Third pass: assign remaining editors to unclaimed topics
unclaimed_topics = [topic for topic in topic_titles if topic not in assignments.values()]

logger.info(f"📊 Assignment Status:")
logger.info(f"   - Assigned editors: {len(assignments)}")
logger.info(f"   - Unassigned editors: {len(unassigned_editors)}")
logger.info(f"   - Unclaimed topics: {len(unclaimed_topics)}")

# Ensure every editor gets at least one topic
for editor in unassigned_editors:
    if unclaimed_topics:
        # Assign to first available unclaimed topic
        topic = unclaimed_topics.pop(0)
        assignments[editor] = topic
        logger.info(f"✅ Assigned unclaimed topic {topic} to {editor}")
    else:
        # No unclaimed topics left - assign from already assigned topics
        # Find an editor who has multiple topics or redistribute
        logger.warning(f"⚠️ No unclaimed topics left for {editor} - will redistribute")
        # For now, we'll assign a random topic (in a real scenario, you might want more sophisticated logic)
        if topic_titles:
            topic = topic_titles[0]  # Assign the first topic as fallback
            assignments[editor] = topic
            logger.info(f"🔄 Fallback assignment: {topic} to {editor}")

# Final verification: ensure every editor has a topic
missing_editors = set(editors) - set(assignments.keys())
if missing_editors:
    logger.error(f"❌ CRITICAL: Editors without topics: {missing_editors}")
    # Emergency assignment - assign any available topic
    for editor in missing_editors:
        if topic_titles:
            topic = topic_titles[0]
            assignments[editor] = topic
            logger.info(f"🚨 Emergency assignment: {topic} to {editor}")

logger.info(f"✅ Final assignment count: {len(assignments)} editors assigned")
logger.info(f"✅ Unclaimed topics remaining: {len([t for t in topic_titles if t not in assignments.values()])}")

# Update state with analysis results and assignments
state["topic_claims"] = topic_claims
state["assignments"] = assignments

logger.info(f"✅ Topic distribution completed. Final assignments: {assignments}")

return state

def editor_write_node(state: NewsroomState) -> NewsroomState:
.........
.........

  # 🔍 PHASE 1: IN-DEPTH RESEARCH
  # Conduct comprehensive research on the assigned topic using specialized tools
  logger.info(f"🔍 Conducting in-depth research for topic: {assigned_topic}")

  # Get specialized in-depth search tools
  # These tools are optimized for comprehensive research from authoritative sources
  in_depth_tools = get_in_depth_tools()

  # Load and format the comprehensive research prompt template
  # This prompt contains detailed instructions for thorough research methodology
  research_prompt_template = load_prompt('prompts/in_depth_researcher.txt')
  formatted_research_prompt = research_prompt_template.format(
      topic=assigned_topic,
      description=topic_details.get('description', 'N/A'),
      category=topic_details.get('category', 'N/A'),
      keywords=', '.join(topic_details.get('keywords', [])) if topic_details.get('keywords') else 'N/A'
  )

  # Create the agent prompt template with required LangChain structure
  # agent_scratchpad is required by LangChain for tool execution and conversation memory
  agent_prompt = ChatPromptTemplate.from_messages([
      ("system", formatted_research_prompt),  # Research instructions and methodology
      ("human", "{input}"),                  # User input (topic to research)
      MessagesPlaceholder(variable_name="agent_scratchpad"),  # Required for tool execution
  ])

  # Create the research agent with automatic tool execution capabilities
  # This agent can automatically decide which search tools to use and execute them
  # Using the same LLM instance for consistency and efficiency
  agent = create_openai_tools_agent(llm, in_depth_tools, agent_prompt)
  agent_executor = AgentExecutor(agent=agent, tools=in_depth_tools, verbose=True)

  # Execute the automated research process
  # The agent will automatically use the search tools to gather comprehensive information
  try:
      logger.info(f"🔍 Starting automated in-depth research for: {assigned_topic}")

      # Invoke the research agent with the topic information
      # The agent will automatically use tools and synthesize the results
      result = agent_executor.invoke({
          "input": f"Research topic: {assigned_topic}. Description: {topic_details.get('description', 'N/A')}"
      })

      # Extract the research results from the agent's output
      in_depth_research = result["output"]
      logger.info(f"✅ Automated in-depth research completed successfully")

  except Exception as e:
      logger.error(f"❌ Automated research failed: {str(e)}")
      in_depth_research = f"Research error: {str(e)}"

  logger.info(f"✅ In-depth research completed for {assigned_topic}")

# 🔍 PHASE 2: ARTICLE GENERATION
  # Generate personalized article using the research results and editor persona

  # Create dynamic article prompt combining:
  # - Editor's persona and writing style
  # - Topic details and context
  # - Comprehensive research results
  article_prompt = get_editor_prompt(
      editor_name=editor_name.lower(),
      task_type='article_writing',
      assigned_topic=assigned_topic,
      topic_details=topic_details,
      in_depth_research=in_depth_research
  )

  # Generate the article using the main LLM
  # This creates the final article with the editor's unique style and voice
  response = llm.invoke([HumanMessage(content=article_prompt)])

  # Process successful article generation
  if response and response.content:
      # Store article with comprehensive metadata
      articles[editor_name] = {
          "topic": assigned_topic,                    # The assigned topic
          "content": response.content,                # The generated article text
          "word_count": len(response.content.split()), # Article length
          "status": "completed",                      # Success status
          "research_used": bool(in_depth_research)    # Whether research was used
      }
      logger.info(f"✅ Article completed for {editor_name}: {len(response.content.split())} words")
  else:
      # Handle article generation failure
      articles[editor_name] = {
          "topic": assigned_topic,
          "content": "Article generation failed",
          "word_count": 0,
          "status": "failed",
          "research_used": False
      }
      logger.error(f"❌ Article generation failed for {editor_name}")

# (sample code for manual tool execution)

@tool
def web_search(query: str) -> dict:
    """Perform a web search for the given query."""
    # Placeholder: Replace with actual web search logic (e.g., SerpAPI)
    return {"results": f"Web search results for '{query}'"}

@tool
def generate_article(topic: str, style: str) -> dict:
    """Generate an article for the given topic in the specified style."""
    # Placeholder: Replace with LLM-based article generation
    return {"article": f"Generated article on '{topic}' in {style} style"}

# List of available tools
tools = [web_search, generate_article]
tool_map = {tool.name: tool for tool in tools}

# Node to check for tool calls and invoke tools
def process_tool_calls(state: GraphState) -> GraphState:
    # Get the last message
    last_message = state["messages"][-1]
    tool_results = state.get("tool_results", [])

    # Check if the last message has tool_calls
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        for tool_call in last_message.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]

            # Check if the tool exists in the tool_map
            if tool_name in tool_map:
                try:
                    # Invoke the tool with the provided arguments
                    result = tool_map[tool_name].invoke(**tool_args)
                    tool_results.append({
                        "tool_call_id": tool_call["id"],
                        "tool_name": tool_name,
                        "result": result
                    })
                except Exception as e:
                    tool_results.append({
                        "tool_call_id": tool_call["id"],
                        "tool_name": tool_name,
                        "error": str(e)
                    })
            else:
                tool_results.append({
                    "tool_call_id": tool_call["id"],
                    "tool_name": tool_name,
                    "error": f"Tool '{tool_name}' not found"
                })

    # Update the state with tool results
    state["tool_results"] = tool_results
    return state