AI EngineeringLangchain ai agents
LangGraph Workflow with and without LLM
Introduction to LangGraph
- Graph-based workflow: Build pipelines as a visual + executable Graph for clear step-by-step flow.
- Nodes & Edges: Use Nodes (functions) for work and Edges (connections) to control execution order.
- Shared State: Store and pass all outputs in a single State object across the workflow.
- Flexible pipelines: Works for non-LLM logic workflows and also LLM-powered workflows inside nodes.
Important Terms:
- Graph: The main workflow container that holds the State (TypedDict), Nodes (functions), and Edges (flow) (created using
graph = StateGraph(StateType)). - Node (Function): A Python function step that takes the state, updates it, and returns it (e.g.,
calc_eligibility,label_category,startup_comedian). - Edge (Relation): The connection that defines execution order between steps like START → node → node → END (added using
graph.add_edge(source, destination)). - START: A built-in LangGraph constant that marks the entry point of the workflow and connects to the first node (e.g.,
graph.add_edge(START, "calc_eligibility")). - END: A built-in LangGraph constant that marks the exit point of the workflow and connects from the last node (e.g.,
graph.add_edge("label_category", END)).
State in LangGraph
- Shared dictionary-like object passed through all nodes.
- Stores initial inputs and updated outputs, often defined using TypedDict for structure and type clarity.
Workflow in LangGraph
- Compiled runnable version of the graph created using workflow = graph.compile().
- Runs using workflow.invoke(initial_state) and returns the final updated state.
State vs Workflow
| Aspect | State | Workflow |
|---|---|---|
| Meaning | Shared data object passed through nodes | Executable compiled graph |
| Purpose | Holds inputs + node outputs | Runs nodes in edge order |
| In Code | class LoanState(TypedDict): ... | workflow = graph.compile() |
Implementation Without LLM Workflow (Loan Eligibility)
Overview of this workflow
-
This workflow calculates:
- Loan eligibility
- Max loan amount
- Customer category
-
Nodes in flow:
calc_eligibility → label_category -
Execution path:
START -> calc_eligibility -> label_category -> END
Step 1: Import required packages
- StateGraph, START, and END are used to build the graph
- TypedDict is used to define the structure of the state
from langgraph.graph import StateGraph , START, END
from typing import TypedDictStep 2: Define the State using TypedDict
- This defines what keys will exist in the workflow state
- Keys include both input fields and fields that will be generated by nodes
class LoanState(TypedDict):
income: float
age: int
eligible: bool
max_loan: float
category: strStep3: Create Node 1 - calc_eligibilty
def calc_eligibilty(state: LoanState) -> LoanState:
income = state['income']
age = state['age']
eligible = income >= 30000 and 21 <= age <= 65
max_loan = income * 5 if eligible else 0
state['eligible'] = eligible
state['max_loan'] = max_loan
return stateStep 4: Create Node 2 - label_category
def label_category(state: LoanState) -> LoanState:
income = state['income']
eligible = state['eligible']
if not eligible:
category = 'Not Eligible'
elif income < 50000:
category = 'Basic'
elif income < 100000:
category = 'Premium'
else:
category = 'Elite'
state['category'] = category
return stateStep 5: Create the Graph with the State type
- Graph is created by giving it the state schema
graph = StateGraph(LoanState)Step 6: Add nodes to the graph
- Register your functions as nodes with a node name
graph.add_node('calc_eligibility', calc_eligibilty)
graph.add_node('label_category', label_category)Step 7: Add edges to define workflow order
graph.add_edge(START, 'calc_eligibility')
graph.add_edge('calc_eligibility', 'label_category')
graph.add_edge('label_category', END)Step 8: Compile the graph into a runnable workflow
- compile() converts your definition into an executable workflow
workflow = graph.compile()Step 9: Provide initial state and run the workflow
intial_state : LoanState = {
'income': 5000,
'age': 30
}
final_state = workflow.invoke(intial_state)
print(final_state)Step 10: Generate Mermaid workflow image (PNG)
png = workflow.get_graph().draw_mermaid_png()
with open('loan_eligibility_workflow.png', 'wb') as f:
f.write(png)Implementation With LLM Workflow (Joke Generator)
Step 1: Import required packages
from langgraph.graph import StateGraph , START, END
from typing import TypedDict
from langchain_openai import ChatOpenAI
from dotenv import load_dotenvStep 2: Load environment variables and initialize the model
load_dotenv()
model = ChatOpenAI(model_name='gpt-4o')Step 3: Define the State for LLM workflow
class JokeState(TypedDict):
topic: str
joke: strStep 4: Create the Node - startup_comedian
def startup_comedian(state: JokeState) -> JokeState:
prompt = f"Tell me a funny joke about {state['topic']}."
joke = model.invoke(prompt)
state['joke'] = joke
return stateStep 5: Create graph and add the node
graph = StateGraph(JokeState)
graph.add_node('startup_comedian', startup_comedian)Step 6: Add edges (START → node → END)
graph.add_edge(START, 'startup_comedian')
graph.add_edge('startup_comedian', END)Step 7: Compile graph into workflow
workflow = graph.compile()Step 8: Provide initial state and invoke workflow
intial_state = {
"topic" : "startups"
}
final_state = workflow.invoke(intial_state)
print(final_state)Step 9: Generate Mermaid workflow image (PNG)
png = workflow.get_graph().draw_mermaid_png()
with open('llm_workflow.png', 'wb') as f:
f.write(png)Last updated on