LangGraph Tutorial – A step-by-step guide for creating a Text Analysis pipeline

Install all required packages
You can install Langgraph and other Python modules using!pip.

Import os
From dotenv, import load_dotenv

Create a.env environment file with your API Key and load the variables.
load_dotenv()

OpenAI API key #
os.environ["OPENAI_API_KEY"] = os.getenv('OPENAI_API_KEY')

Import ChatOpenAI from Langchain_openai

ChatOpenAI: Initialize this instance
Model = llm (ChatOpenAI)"gpt-4o-mini")

Test your setup
response = llm.invoke("Hello! Are you working?")
print(response.content)

Import os
Import TypedDict List and Annotated
StateGraph can be imported from langgraph.graph.
Import PromptTemplate from langchain.prompts
Import ChatOpenAI from Langchain_openai
Import HumanMessage from langchain.schema
from langchain_core.runnables.graph import MermaidDrawMethod
Import Image from IPython.display

Class State (TypedDict),
 Text
 Classification:
 Entities: List[str]
 Summary:

# Initialize the temperature of our language model to zero for more deterministic results
Model = llm (ChatOpenAI)"gpt-4o-mini", temperature=0)

def classification_node(state: State):
 "''Classify this text in one of these categories: Research, News, or Blog''Return
    prompt = PromptTemplate(
        input_variables=["text"],
        template="Classify the following text into one of the categories: News, Blog, Research, or Other.nnText:{text}nnCategory:"
    )
    message = HumanMessage(content=prompt.format(text=state["text"]))
    classification = llm.invoke([message]).content.strip()
    return {"classification": classification}

def entity_extraction_node(state: State):
 "''Extract the Entities (Person/Organization/Location)'' from the Text''Return
    prompt = PromptTemplate(
        input_variables=["text"],
        template="Extract all the entities (Person, Organization, Location) from the following text. Provide the result as a comma-separated list.nnText:{text}nnEntities:"
    )
    message = HumanMessage(content=prompt.format(text=state["text"]))
    entities = llm.invoke([message]).content.strip().split(", ")
    return {"entities": entities}

def summarization_node(state: State):
 Summarize the entire text with one simple sentence
    prompt = PromptTemplate(
        input_variables=["text"],
        template="Summarize the following text in one short sentence.nnText:{text}nnSummary:"
    )
    message = HumanMessage(content=prompt.format(text=state["text"]))
 () summary = llm.invokeReturn[message]).content.strip()
    return {"summary": summary}

StateGraph: Create your own StateGraph
Workflow = StateGraph (State)

Add more nodes to your graph
workflow.add_node("classification_node", classification_node)
workflow.add_node("entity_extraction", entity_extraction_node)
workflow.add_node("summarization", summarization_node)

# Add edges on the graph
workflow.set_entry_point("classification_node"(# Sets the entry point in the graph
workflow.add_edge("classification_node", "entity_extraction")
workflow.add_edge("entity_extraction", "summarization")
workflow.add_edge("summarization", END)

Compose the graph
The app is a workflow.compile()

sample_text = """ OpenAI has announced the GPT-4 model, which is a large multimodal model that exhibits human-level performance on various professional benchmarks. It is developed to improve the alignment and safety of AI systems. Additionally, the model is designed to be more efficient and scalable than its predecessor, GPT-3. The GPT-4 model is expected to be released in the coming months and will be available to the public for research and development purposes. """ 
state_input = {"text": sample_text} 
result = app.invoke(state_input) 
print("Classification:"Results["classification"]) 
print("nEntities:"Results["entities"]) 
print("nSummary:"Results["summary"])
Classification of News Entities ['OpenAI', 'GPT-4', 'GPT-3'] OpenAI’s GPT-4 is an upcoming multimodal AI model that will aim for performance at the level of a human and improve safety, efficiency and scalability over GPT-3.

Please replace the text below with your own to get a better understanding of your text.# Run the pipeline to process your text. """ The recent advancements in quantum computing have opened new possibilities for cryptography and data security. Researchers at MIT and Google have demonstrated quantum algorithms that could potentially break current encryption methods. However, they are also developing new quantum-resistant encryption techniques to protect data in the future. """ 

# Process the text through our pipeline your_result = app.invoke({"text": your_text}) print("Classification:", your_result["classification"]) 

print("nEntities:", your_result["entities"]) 
print("nSummary:", your_result["summary"])

Researchers: Classification ['MIT', 'Google'] The recent advances in quantum computing could threaten existing encryption techniques while also leading to new quantum-resistant methods.

Update our state to include sentiment
class EnhancedState(TypedDict):
 Text
 Classification:
 Entities: List[str]
 Summary
 Sentiment:

Create our sentiment node
def sentiment_node(state: EnhancedState):
 Analyze whether the message is positive, negative, or neutral.Return
    prompt = PromptTemplate(
        input_variables=["text"],
        template="Analyze the sentiment of the following text. Is it Positive, Negative, or Neutral?nnText:{text}nnSentiment:"
    )
    message = HumanMessage(content=prompt.format(text=state["text"]))
    sentiment = llm.invoke([message]).content.strip()
    return {"sentiment": sentiment}

# Create a workflow that includes the enhanced state
enhanced_workflow = StateGraph(EnhancedState)

# Add nodes to the list
enhanced_workflow.add_node("classification_node", classification_node)
enhanced_workflow.add_node("entity_extraction", entity_extraction_node)
enhanced_workflow.add_node("summarization", summarization_node)

# Add new node for sentiment
enhanced_workflow.add_node("sentiment_analysis", sentiment_node)

# Create more complex workflows with branches
enhanced_workflow.set_entry_point("classification_node")
enhanced_workflow.add_edge("classification_node", "entity_extraction")
enhanced_workflow.add_edge("entity_extraction", "summarization")
enhanced_workflow.add_edge("summarization", "sentiment_analysis")
enhanced_workflow.add_edge("sentiment_analysis", END)

# Create the enhanced graph
enhanced_app = enhanced_workflow.compile()

# Try the enhanced version of the pipeline using the same text
enhanced_result = enhanced_app.invoke({"text": sample_text})

print("Classification:", enhanced_result["classification"])
print("nEntities:", enhanced_result["entities"])
print("nSummary:", enhanced_result["summary"])
print("nSentiment:", enhanced_result["sentiment"])

# Route after classification
def route_after_classification(state: EnhancedState) -> str:
 Category = state["classification"].lower() # returns: "news", "blog", "research", "other"
 Please return the category. ["news", "research"]

StateGraph can be imported from langgraph.graph.

conditional_workflow = StateGraph(EnhancedState)

# Add nodes
conditional_workflow.add_node("classification_node", classification_node)
conditional_workflow.add_node("entity_extraction", entity_extraction_node)
conditional_workflow.add_node("summarization", summarization_node)
conditional_workflow.add_node("sentiment_analysis", sentiment_node)

Set the entry point
conditional_workflow.set_entry_point("classification_node")

Add conditional edge
conditional_workflow.add_conditional_edges("classification_node", route_after_classification, path_map={
    True: "entity_extraction",
    False: "summarization"
})

# Remove remaining static edge
conditional_workflow.add_edge("entity_extraction", "summarization")
conditional_workflow.add_edge("summarization", "sentiment_analysis")
conditional_workflow.add_edge("sentiment_analysis", END)

# Compile
conditional_app = conditional_workflow.compile()

test_text = """
OpenAI launched the GPT-4 with improved performance for academic and professional task. This is viewed as a breakthrough in reasoning and alignment capabilities.
"""

result = conditional_app.invoke({"text": test_text})

print("Classification:"Results["classification"])
print("Entities:"Results.get("entities", "Skipped"))
print("Summary:"Results["summary"])
print("Sentiment:", result["sentiment"])

blog_text = """
This is what I discovered after spending a whole week in silent meditation. No phones, no talking—just me, my breath, and some deep realizations.
"""

result = conditional_app.invoke({"text": blog_text})

print("Classification:"Results["classification"])
print("Entities:"Results.get("entities", "Skipped (not applicable)"))
print("Summary:"Results["summary"])
print("Sentiment:", result["sentiment"])