Build a "Plan and Execute" AI Agent Workflow

LangGraph + Perplexity Sonar

This tutorial demonstrates how to build an AI agent that dynamically plans and executes tasks using LangGraph and Perplexity’s “Sonar” model, a large language model built on Meta's Llama 3.3 70b that is connected to the internet for real-time web search. You'll learn how to structure a stateful workflow that refines live music recommendations in real-time, ensuring relevant and up-to-date event suggestions.

LangGraph is a graph-based orchestration layer for LangChain, enabling developers to design, visualize, and execute complex, node-based workflows with AI agents, tools, and data sources.

The video covers key concepts in plan + execute agent workflows, AI orchestration, LangGraph (e.g. nodes and edges), and iterative workflows, showing how to break down complex tasks into modular steps.

Full Code Below the Video

Environment (Python 3.10): pip install langgraph==0.3.5 langchain==0.3.20 langchain-community==0.3.19 langchain-core==0.3.43 openai==1.65.5 typing_extensions==4.12.2 ipykernel==6.29.5

Full Code:

from typing import Annotated, List, Optional
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_community.chat_models.perplexity import ChatPerplexity
from langchain_core.messages import HumanMessage, AIMessage
import os
from datetime import datetime
from IPython.display import Image, display


os.environ["PERPLEXITY_API_KEY"] = "API KEY GOES HERE"

llm = ChatPerplexity(model="sonar",
                     temperature=0.5,
                     pplx_api_key=os.environ["PERPLEXITY_API_KEY"])


# 1. Define Agent State (with excluded_events) ------------------------------------
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    music_event_recommendations: Optional[List[str]]
    event_info_list: Optional[List[str]]
    next_node: Optional[str]
    excluded_events: Optional[List[str]] 


# 2. Define Agent Nodes ------------------------------------
def music_planner(state: AgentState):
    """Node 1: Planner - Generates plan."""
    print("\n[PLANNER NODE] - Generating plan...")
    user_preferences = state["messages"][0].content
    today_date = datetime.now().strftime("%B %d, %Y")

    prompt = (
        f"Today's date is *{today_date}*. "
        "You are a live music expert in Brooklyn, New York. "
        "Create a concise plan to recommend 3 live music events and find their details for tonight."
        "Keep the plan brief.\n\n"
        f"User Preferences:\n{user_preferences}"
    )
    print("[PLANNER NODE] - Music event plan generated.")
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"messages": [response], "music_event_recommendations": None, "event_info_list": None, "next_node": "recommend_music_event", "excluded_events": state.get("excluded_events", [])} # сохраняем excluded_events


def music_recommendation_node(state: AgentState):
    """Node 2: Music Event Recommendation - Recommends 3 events, excluding blocked events."""
    print("\n[MUSIC RECOMMENDATION NODE] - Recommending events, blocking out previous no-go's...")
    user_preferences = state["messages"][0].content
    plan_text = state["messages"][-1].content
    today_date = datetime.now().strftime("%B %d, %Y")
    excluded_events = state.get("excluded_events", []) # Get excluded events from state

    prompt = (
        f"Today's date is *{today_date}*. "
        "Using the plan and user preferences, recommend three *different* live music events for tonight in Brooklyn, New York. " # Emphasize different events
        f"Prioritize events happening tonight in Brooklyn, NY. "
        f"**Ensure that the recommendations are *not* from this list of events to avoid:** {excluded_events if excluded_events else 'None'}. " # Exclude specific events
        "Output a numbered list of just the names of the 3 music events.\n\n"
        f"User Preferences:\n{user_preferences}\n\n"
        f"Recommendation Plan:\n{plan_text}"
    )
    print("[MUSIC RECOMMENDATION NODE] - Prompt with exclusion list sent.")
    response = llm.invoke([HumanMessage(content=prompt)])
    recommended_events_text = response.content.strip()
    music_event_recommendations = [line.split(". ", 1)[1] for line in recommended_events_text.split('\n') if ". " in line]
    recommendation_message = AIMessage(content=recommended_events_text)
    print(f"[MUSIC RECOMMENDATION NODE] - Music Events Recommended (excluding blocked events): {music_event_recommendations}")
    return {"messages": [recommendation_message], "music_event_recommendations": music_event_recommendations, "event_info_list": None, "next_node": "event_lookup", "excluded_events": excluded_events} # Pass along excluded_events


def event_lookup_node(state: AgentState):
    """Node 3: Event Lookup - Finds details for events and updates excluded_events."""
    print("\n[EVENT LOOKUP NODE] - Looking up event details and updating blocked events...")
    music_event_recommendations = state["music_event_recommendations"]
    event_info_list = []
    today_date = datetime.now().strftime("%B %d, %Y")
    insufficient_info_found = False
    excluded_events = state.get("excluded_events", []) # Get current excluded events

    for event_name in music_event_recommendations:
        search_query = f"live music events in Brooklyn NY tonight for '{event_name}' {today_date}"
        event_prompt = (
            f"Today's date is *{today_date}*. "
            f"Find details for '{event_name}' tonight in Brooklyn, New York. "
            f"Search online event listing and local venue websites for events on *{today_date}*. "
            f"Provide event name, venue, time, and address, and ticket info (if available). "
            f"If no details found, say 'No details found for {event_name} in Brooklyn on {today_date} (tonight).' "
            f"If sold out, mention that in ticket information."
        )
        response = llm.invoke([HumanMessage(content=event_prompt)])
        event_info = response.content.strip()
        event_info_list.append(event_info)

        venue_present = "Venue:" in event_info or "Venue Name:" in event_info or "at" in event_info
        time_present = "Time:" in event_info or "Doors:" in event_info or ":" in event_info
        address_present = "Address:" in event_info or "Location:" in event_info or "," in event_info and any(word in event_info.lower() for word in ["brooklyn", "ny", "new york"])
        sold_out = "Sold Out" in event_info or "sold out" in event_info or "SOLD OUT" in event_info # Keep sold out detection
        no_details = "No details found" in event_info

        if no_details or not venue_present or not time_present or not address_present: 
            insufficient_info_found = True
            if event_name not in excluded_events:
                excluded_events.append(event_name)
            print(f"[EVENT LOOKUP NODE] - Insufficient info for '{event_name}'. Added to blocked events.")
        elif sold_out:
             print(f"[EVENT LOOKUP NODE] - Event '{event_name}' is sold out, but details found.") # Sold out but details found message
        else:
            print(f"[EVENT LOOKUP NODE] - Details found for '{event_name}'.")

    print("[EVENT LOOKUP NODE] - Event details retrieved and blocked events updated.")

    if insufficient_info_found:
        next_node = "recommend_music_event"
        print("[EVENT LOOKUP NODE] - Insufficient event info for at least one event. WORKFLOW GOING BACK to music_recommendation_node. Blocked events list updated.") # Workflow go back message
    else:
        next_node = "executor"
        print("[EVENT LOOKUP NODE] - Sufficient event info found for all events. WORKFLOW PROCEEDING to executor_node.") # Workflow proceed message - updated message

    return {"messages": [], "music_event_recommendations": state["music_event_recommendations"], "event_info_list": event_info_list, "next_node": next_node, "excluded_events": excluded_events} # Return updated excluded_events


def music_agent_executor(state: AgentState):
    """Node 4: Executor - Generates final output summary."""
    print("\n[EXECUTOR NODE] - Generating final output summary...")
    music_event_recommendations = state["music_event_recommendations"]
    event_info_list = state["event_info_list"]
    today_date = datetime.now().strftime("%B %d, %Y")

    final_output = f"Here are 3 live music events in Brooklyn for *tonight*, *{today_date}*:\n\n"

    for i, event_name in enumerate(music_event_recommendations[:3]):
        event_details = event_info_list[i] if event_info_list and i < len(event_info_list) else "Details not found."
        final_output += f"**{i+1}. {event_name}**\n"
        final_output += f"{event_details}\n\n"

    final_output = f"""--- Agent Output ---\nFinal Output Message: {final_output.strip()}"""

    output_message = AIMessage(content=final_output)
    print("[EXECUTOR NODE] - Final output summary generated.")
    return {"messages": [output_message], "music_event_recommendations": music_event_recommendations, "event_info_list": event_info_list, "next_node": "executor", "excluded_events": state.get("excluded_events", [])} # Pass along excluded_events


# 3. Build the LangGraph StateGraph  ------------------------------------
graph_builder = StateGraph(AgentState)
graph_builder.add_node("planner", music_planner)
graph_builder.add_node("recommend_music_event", music_recommendation_node)
graph_builder.add_node("event_lookup", event_lookup_node)
graph_builder.add_node("executor", music_agent_executor)

# Conditional Edge Definition
def decide_event_lookup_next_node(state):
    print("\n[DECISION NODE] - Deciding next node after event_lookup...")
    if state['next_node'] == 'recommend_music_event':
        print("[DECISION NODE] - Decided to go back to recommend_music_event.")
        return "recommend_music_event"
    else:
        print("[DECISION NODE] - Decided to proceed to executor.")
        return "executor"

# Graph Edges
graph_builder.add_edge(START, "planner")
graph_builder.add_edge("planner", "recommend_music_event")
graph_builder.add_edge("recommend_music_event", "event_lookup")
graph_builder.add_conditional_edges(
    "event_lookup",
    decide_event_lookup_next_node,
    {
        "recommend_music_event": "recommend_music_event",
        "executor": "executor"
    }
)
graph_builder.add_edge("executor", END)

# Compile the graph
live_music_agent_brooklyn_date_aware_verbose_conditional_block = graph_builder.compile() # Updated agent name


# 4. Run the Agent  ------------------------------------
user_input = "I'm looking for some jazz music tonight in Brooklyn."

inputs = {"messages": [HumanMessage(content=user_input)], "music_event_recommendations": None, "event_info_list": None, "next_node": None, "excluded_events": []} # Initialize excluded_events as empty list
result = live_music_agent_brooklyn_date_aware_verbose_conditional_block.invoke(inputs) # Use updated agent


# 5. Output  ------------------------------------
print("\n--- Agent Output ---")
print("Final Output Message:", result['messages'][-1].content)

Subscribe to the Deep Charts YouTube Channel for more informative AI and Machine Learning Tutorials and Workflow Ideas.