🎯 What You'll Learn Today

from typing import Annotated, TypedDict
#!pip install langchain-core
#!pip install langgraph
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

class ConversationMemory(TypedDict):
    """Advanced conversation context tracker.

    This class demonstrates sophisticated context management by tracking
    multiple aspects of the conversation state:

    Attributes:
        sentiment: Emotional context of the conversation
        question_count: Conversation depth tracking
        last_topic: Topic continuity management
        conversation_turns: Turn-taking management
        awaiting_human_input: Input state management

    Note:
        This multi-faceted approach allows for nuanced conversation control
        and context-aware response generation.
    """

    sentiment: str
    question_count: int
    last_topic: str
    conversation_turns: int
    awaiting_human_input: bool

class State(TypedDict):
    """Enhanced state container with memory integration.

    Combines LangGraph's message management with sophisticated
    conversation tracking:

    Attributes:
        messages: Annotated message list for LangGraph compatibility
        memory: Integrated conversation memory system
        current_path: Dynamic conversation path tracking
    """

    messages: Annotated[list[BaseMessage], add_messages]
    memory: ConversationMemory
    current_path: str

def analyze_sentiment(message: str) -> str:
    """Analyzes message sentiment for context-aware responses.

    Demonstrates integration of analytical capabilities into
    the conversation flow:

    Args:
        message: Input text to analyze

    Returns:
        str: Sentiment classification ('happy', 'sad', or 'neutral')

    Note:
        In production systems, this would typically integrate
        with more sophisticated NLP models.
    """
    positive_words = {"great", "good", "happy", "wonderful", "excellent"}
    negative_words = {"bad", "sad", "unhappy", "terrible", "awful"}

    message_words = set(message.lower().split())
    if any(word in message_words for word in positive_words):
        return "happy"
    elif any(word in message_words for word in negative_words):
        return "sad"
    return "neutral"

def process_human_input(state: State) -> State:
    """Processes human input with context awareness.

    Demonstrates advanced input processing with:
    1. State-based response selection
    2. Context-aware processing
    3. Turn management

    Args:
        state: Current conversation state with memory and path information

    Returns:
        State: Updated state with processed input and modified memory

    Note:
        This implementation simulates human responses for demonstration.
        In a real system, this would process actual user input.
    """
    memory = state["memory"]
    current_path = state["current_path"]

    # Context-aware response mapping
    human_responses = {
        "initial": "I'm feeling pretty good today, thanks for asking!",
        "goals": "I'd love to discuss my career development.",
        "support": "I've been feeling a bit overwhelmed lately.",
        "general": "I'm interested in learning about new technologies.",
        "closing": "Yes, thank you for the chat!",
    }

    response = human_responses.get(current_path, "Goodbye!")
    memory["awaiting_human_input"] = False

    return {
        "messages": [HumanMessage(content=response)],
        "memory": memory,
        "current_path": current_path,
    }

def conversation_node(state: State) -> State:
    """Manages the main conversation flow with advanced state handling.

    This node demonstrates sophisticated conversation management:
    1. Context-aware response generation
    2. Dynamic path selection
    3. Turn management
    4. State persistence

    Args:
        state: Current conversation state including memory and messages

    Returns:
        State: Updated state with new messages and modified memory

    Note:
        This implementation shows how to handle complex conversation
        flows while maintaining context and state.
    """
    memory = state["memory"]

    # Handle initial conversation setup
    if not state["messages"]:
        return {
            "messages": [
                SystemMessage(content="Conversation initialized"),
                AIMessage(content="How are you feeling today?"),
            ],
            "memory": {
                "sentiment": "unknown",
                "question_count": 1,
                "last_topic": "greeting",
                "conversation_turns": 1,
                "awaiting_human_input": True,
            },
            "current_path": "initial",
        }

    last_message = state["messages"][-1]

    # Implement turn-taking logic
    if memory.get("awaiting_human_input", True):
        return process_human_input(state)

    memory["conversation_turns"] = memory.get("conversation_turns", 0) + 1
    current_path = state["current_path"]

    # Handle conversation paths with context awareness
    if current_path == "initial":
        # Process initial response and determine conversation direction
        sentiment = analyze_sentiment(last_message.content)
        memory["sentiment"] = sentiment

        if sentiment == "happy":
            response = "That's wonderful! Would you like to talk about your goals?"
            next_path = "goals"
        elif sentiment == "sad":
            response = (
                "I'm here to listen. Would you like to share what's bothering you?"
            )
            next_path = "support"
        else:
            response = "Would you like to explore some interesting topics?"
            next_path = "general"

    elif current_path in ["goals", "support", "general"]:
        # Handle ongoing conversation with path-specific responses
        memory["question_count"] += 1

        if memory["question_count"] >= 3:
            response = (
                "Thank you for sharing. This has been a great conversation. Goodbye!"
            )
            next_path = "ended"
        else:
            response_map = {
                "goals": "Tell me more about your aspirations.",
                "support": "How can I help make things better?",
                "general": "What interests you the most right now?",
            }
            response = response_map[current_path]
            next_path = current_path

    else:
        # Handle conversation termination
        response = "Goodbye! Take care!"
        next_path = "ended"

    memory["awaiting_human_input"] = True

    return {
        "messages": [AIMessage(content=response)],
        "memory": memory,
        "current_path": next_path,
    }

def should_end(state: State) -> bool:
    """Determines conversation termination based on multiple conditions.

    This function demonstrates advanced flow control by evaluating
    multiple termination conditions:
    1. Explicit end state reached
    2. Maximum interaction limit reached
    3. Conversation length limit reached

    Args:
        state: Current conversation state with memory and path information

    Returns:
        bool: True if any termination condition is met, False otherwise

    Note:
        Multiple conditions provide flexible conversation control
        while preventing infinite loops.
    """
    conditions = [
        state["current_path"] == "ended",
        state["memory"]["question_count"] >= 3,
        state["memory"]["conversation_turns"] > 10,
    ]
    return any(conditions)

graph = StateGraph(State)

graph.add_node("chat", conversation_node)

graph.add_edge(START, "chat")
graph.add_conditional_edges(
    "chat",
    should_end,
    {
        True: END,  # End conversation when conditions are met

        False: "chat",  # Continue conversation

    },
)

chain = graph.compile()

def print_conversation(messages: list[BaseMessage]):
    """Formats and displays the conversation flow.

    This utility function provides a clear view of the conversation
    progress and message flow.

    Args:
        messages: List of conversation messages to display

    Note:
        Distinguishes between different message types (System/AI/Human)
        for clear conversation visualization.
    """
    print("\nConversation Flow:")
    print("-----------------")
    for idx, msg in enumerate(messages, 1):
        speaker = (
            "System"
            if isinstance(msg, SystemMessage)
            else "AI"
            if isinstance(msg, AIMessage)
            else "Human"
        )
        print(f"{idx}. {speaker}: {msg.content}")

initial_state = {
    "messages": [],
    "memory": {
        "sentiment": "unknown",
        "question_count": 0,
        "last_topic": "none",
        "conversation_turns": 0,
        "awaiting_human_input": False,
    },
    "current_path": "initial",
}

Configure execution parameters

```python
config = {"recursion_limit": 15}  # Prevent infinite recursion

final_state = chain.invoke(initial_state, config=config)

print_conversation(final_state["messages"])

1. System: Conversation initialized
2. AI: How are you feeling today?
3. Human: I'm feeling pretty good today, thanks for asking!
4. AI: That's wonderful! Would you like to talk about your goals?
5. Human: I'd love to discuss my career development.
6. AI: Tell me more about your aspirations.
7. Human: I'm interested in learning about new technologies.
8. AI: Thank you for sharing. This has been a great conversation. Goodbye!