LangGraph is an open-source framework(built by LangChain team) for building AI agents and workflows using a graph structure instead of simple linear steps. Most of the enterprise applications use LangGraph which is stateful.
is taking a call to decide what is the right way to pass request. In this scenario, we have tool which will tell us about weather information. Incase if the request to this agent is about weather, then no need to use llm, agent can simply use tool built for getting weather information. But if the request is not about weather, then agent can pass it to LLM, to get the desired output/data. This is a simple example on building a Agent, with tools and llm calls in it. If you see dotted line, then it is a conditional edge.
Above image represents a complex flow in real time where we are diverting requests to RAG if the query is related to our project which is indexed in RAG. If not, Query Analysis divert query to web to generate response.
- TypedDict comes from typing_extensions
- It lets you define a dictionary with a fixed structure & types
- So, instead of normal dict, you can enforce that :
- key must be graph_info and type must be string
class State(TypedDict):
graph_info:str
- Above code defines structure of the shared state that flows through a LangGraph workflow.
Then, we defined 3 definitions, these are the actions performed by respective nodes :
- start_play()
- cricket()
- badminton()
import random
from typing import Literal
def random_play(state:State)-> Literal['cricket','badminton']:
graph_info=state['graph_info']
if random.random()>0.5:
return "cricket"
else:
return "badminton"
Above definition will return either 'cricket' or 'badminton' and nothing else is allowed - because we declared above 2 type hints
- random_play(state:State) will take input which should be of type State
- and it should return a Literal, either cricket or badminton(just restricting the return statements)
Rest of the code is about state graph construction :
- import statements
- creating the instance graph for StateGraph() class - input as State class object which is graph_info
- adding nodes
- adding edges
- compleling graph
- finally printing the state graph using display() and Image class
Note that dotted line in the output image represents a conditional edge.
If we invoke, graph_builder.invoke() with some graph_info value as
graph_builder.invoke({"graph_info":"Hey My name is Arun"})
start_play node has been called
badminton node has been called
{'graph_info': 'Hey My name is Arun'}graph_builder.invoke({"graph_info":123})
start_play node has been called
criket node has been called
{'graph_info': 123}
Note that, we declared graph_info as string but even if we are invoking it with a value of type int, it is still getting invoked. This is the issue with TypedDict. It won't restrict the type during run time.
Implementing a simple chat bot using LangGraph :
Reducer :
We need to understand a concept called Reducer. In LangGrpah, a Reducer is what controls how the state getting updated when multiple nodes write to it.
Whenever a node returns something like this : {'graph_info' : 'some value'} then LangGraph doesn't just overwrite blindly, it uses a reducer to decide :
- Replace ?
- Append ?
- Merge ?
- Combine ?
Why reducers are needed ?
In LangGraph :
- Multiple nodes can update the same state key
- Execution can be parallel or iterative
- You need a consistent way to combine results
That's where reducers comes in.
from langgraph.graph.message import add_messages
Whenever we use add_messages(), internally we use reducer to handle the state getting updated by multiple nodes.
Steps to implement a LangGraph
- State
- Define functions (nodes)
- State Graph object
- Adding nodes
- Adding Edges
- Compile the Graph
- graph.invoke() - invoke state graph object
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
## Reducers
from typing import Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages:Annotated[list,add_messages]
import os
from dotenv import load_dotenv
load_dotenv()
# os.environ["OPENAI_API_KEY"]=os.getenv("OPENAI_API_KEY")
os.environ["GROQ_API_KEY"]=os.getenv("GROQ_API_KEY")
from langchain_groq import ChatGroq
llm_groq=ChatGroq(model="qwen/qwen3-32b")
llm_groq.invoke("Hey I am Arun Mathe and i like to play cricket")
def superbot(state:State):
return {"messages":[llm_groq.invoke(state['messages'])]}
graph=StateGraph(State)
## node
graph.add_node("Bot",superbot)
## Edges
graph.add_edge(START,"Bot")
graph.add_edge("Bot",END)
graph_builder=graph.compile()
## Display
from IPython.display import Image, display
display(Image(graph_builder.get_graph().draw_mermaid_png()))
Output :
## Invocation
graph_builder.invoke({'messages':"Hi"})
{'messages': [HumanMessage(content='Hi', additional_kwargs={}, response_metadata={},
id='77d19e20-a1ac-46db-95fa-a1cb355cae36'),
AIMessage(content='<think>\nOkay, the user said "Hi".
I need to respond appropriately. Since it\'s a greeting,
I should greet them back and offer assistance.
Let me keep it friendly and open-ended.
Maybe ask how I can help them today.
That should encourage them to elaborate on their needs.
I should make sure the tone is welcoming and professional.
Let me check for any typos or errors.
Alright, that looks good.
\n</think>\n\nHello! How can I assist you today? 😊',
additional_kwargs={}, response_metadata={'token_usage':
{'completion_tokens': 100, 'prompt_tokens': 9, 'total_tokens': 109,
'completion_time': 0.184991031, 'completion_tokens_details': None,
'prompt_time': 0.000199107, 'prompt_tokens_details': None, 'queue_time': 0.047408892,
'total_time': 0.185190138}, 'model_name': 'qwen/qwen3-32b',
'system_fingerprint': 'fp_5cf921caa2', 'service_tier': 'on_demand',
'finish_reason': 'stop', 'logprobs': None, 'model_provider': 'groq'},
id='lc_run--019d543a-0cb3-7950-8f9b-924ee100dded-0',
tool_calls=[], invalid_tool_calls=[],
usage_metadata={'input_tokens': 9, 'output_tokens': 100, 'total_tokens': 109})]}
Similar, we can invoke 'n' number of user messages to LLM and get answers.
Binding Tools with LLM :
We need to understand how to bind tools like wikipedia, Anxiv, Tavily and also custom tools like example addition, subtraction, multiplication, division etc. This concept of building LLM + custom tools is nothing but Binding.
What is the advantage with this ?
Suppose user requested
- "what is the top 10 AI news in April 2026 ?" - Sometimes hallucinated answers will come
- "What is LangGraph ?" - May be diverted to wikipedia
- "What is Attention all you need paper ?" - Arxiv
- "What is 4 + 5 ?" - It will got to add tool
- "What is 10 divide by 2 ?" - It will go to division tool
User request ---> LLM + Tools ---> redirect to appropriate tools if no tools available then only we have to use LLM. Once tools respond, LLM will articulate the response and finally produce the output.
Tools :
- Wikipedia
- Arxiv - it is for research related information
- Tavily - it is for latest information
- add
- sub
- mul
- div
We need to have a way to combine LLM and above tools. This is called Binding tools with LLM.
Implementing a Chatbot with multiple tools using LangGraph :
from langchain_community.tools import ArxivQueryRun,WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper,ArxivAPIWrapper
api_wrapper_arxiv=ArxivAPIWrapper(top_k_results=2,doc_content_chars_max=500)
arxiv=ArxivQueryRun(api_wrapper=api_wrapper_arxiv)
print(arxiv.name)
arxiv.invoke("Attention is all you need")
api_wrapper_wiki=WikipediaAPIWrapper(top_k_results=1,doc_content_chars_max=500)
wiki=WikipediaQueryRun(api_wrapper=api_wrapper_wiki)
wiki.name
wiki.invoke("What is machine learning")
from dotenv import load_dotenv
load_dotenv()
import os
os.environ["TAVILY_API_KEY"]=os.getenv("TAVILY_API_KEY")
os.environ["GROQ_API_KEY"]=os.getenv("GROQ_API_KEY")
### Tavily Search Tool
from langchain_community.tools.tavily_search import TavilySearchResults
tavily = TavilySearchResults()
tavily.invoke("Provide me latest movie reviews of Dhurandhar-2")
### Combine all the tools in the list
tools=[arxiv,wiki,tavily]
## Initialize my LLM model
from langchain_groq import ChatGroq
llm=ChatGroq(model="qwen/qwen3-32b")
llm_with_tools=llm.bind_tools(tools)
from pprint import pprint
from langchain_core.messages import AIMessage, HumanMessage
llm_with_tools.invoke([HumanMessage(content=f"What is the recent AI News")])
llm_with_tools.invoke([HumanMessage(content=f"What is the recent AI News")]).tool_calls
## State Schema
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage
from typing import Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages:Annotated[list[AnyMessage],add_messages]
### Entire Chatbot With LangGraph
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langgraph.prebuilt import tools_condition
### Node definition
def tool_calling_llm(state:State):
return {"messages":[llm_with_tools.invoke(state["messages"])]}
# Build graph
builder = StateGraph(State)
builder.add_node("tool_calling_llm", tool_calling_llm)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "tool_calling_llm")
builder.add_conditional_edges(
"tool_calling_llm",
# If the latest message (result) from assistant is a tool call -> tools_condition routes to tools
# If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END
tools_condition,
)
builder.add_edge("tools", END)
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
messages=graph.invoke({"messages":HumanMessage(content="1706.03762")})
for m in messages['messages']:
m.pretty_print()
messages=graph.invoke({"messages":HumanMessage(content="Provide me the top 10 recent AI news for March 2026")})
for m in messages['messages']:
m.pretty_print()
messages=graph.invoke({"messages":HumanMessage(content="What is machine learning")})
for m in messages['messages']:
m.pretty_print()
messages=graph.invoke({"messages":HumanMessage(content="What is the squre root of 123456789")})
for m in messages['messages']:
m.pretty_print()
messages=graph.invoke({"messages":HumanMessage(content="What is current status petrol and LPG GAS in India year 2026?")})
for m in messages['messages']:
m.pretty_print()
Explanation of code:
As shown in the above image, once user send a request to LLM, it will analyze to find if it have a tool to respond to this user request. If tool is available, then it will send the request to tool and after processing the request, tool will send response back to LLM, then it will articulate the response and finally produce the output. Lets see how to construct this mechanism using LangGraph.
- As we are using external tools like Arxiv, Wikipedia, we need to import libraries & wrappers related to them
- Observe that we are printing only top 2 results (top_k_results=2) & 500 characters only (doc_content_arxiv=500)
from langchain_community.tools import ArxivQueryRun,WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper,ArxivAPIWrapper
api_wrapper_arxiv=ArxivAPIWrapper(top_k_results=2,doc_content_chars_max=500)
arxiv=ArxivQueryRun(api_wrapper=api_wrapper_arxiv)
print(arxiv.name)
Output :
Arxiv
arxiv.invoke("Attention is all you need")
api_wrapper_wiki=WikipediaAPIWrapper(top_k_results=1,doc_content_chars_max=500)
wiki=WikipediaQueryRun(api_wrapper=api_wrapper_wiki)
wiki.name
Output :
wikipedia
wiki.invoke("What is machine learning")
Output :
'Page: Machine learning\nSummary: Machine learning (ML) is a field of study in
artificial intelligence concerned with the development and study of statistical
algorithms that can learn from data and generalize to unseen data,
and thus perform tasks without explicit programming language instructions.
Within a subdiscipline of machine learning, advances in the field of deep learning
have allowed neural networks, a class of statistical algorithms,
to surpass many previous machine learning approaches '
from dotenv import load_dotenv
load_dotenv()
import os
os.environ["TAVILY_API_KEY"]=os.getenv("TAVILY_API_KEY")
os.environ["GROQ_API_KEY"]=os.getenv("GROQ_API_KEY")
### Tavily Search Tool
from langchain_community.tools.tavily_search import TavilySearchResults
tavily = TavilySearchResults()
tavily.invoke("Provide me latest movie reviews of Dhurandhar-2")
- As shown above, we have seen using tools like Arxiv, Wikipedia, Tavily.
- Now we are placing all these 3 tools in a list
### Combine all the tools in the list
tools=[arxiv,wiki,tavily]
## Initialize my LLM model
from langchain_groq import ChatGroq
llm=ChatGroq(model="qwen/qwen3-32b")
llm_with_tools=llm.bind_tools(tools)
- first two lines in the above code are importing ChatGroq class from langchain_groq and created instances for ChatGroq() using groq model "qwen/qwen3-32b"
- third line is important step; where we are binding tools with LLM. This is how we bind tools with LLM
This is called Binding. It is important to note this feature, LLM now can use these tools incase if any user request comes which is related to these tools. Incase if that request is not related to tools, then LLM itself will process it.
Now look at below code :
from pprint import pprint
from langchain_core.messages import AIMessage, HumanMessage
llm_with_tools.invoke([HumanMessage(content=f"What is the recent AI News")])
We invoked llm and asked "What is the recent AI news" - ideally this is the latest news, hence Tavily should answer and AI should not answer as we bind tools to LLM.
Look at the response : AI message is empty(content=""), because request routed to Tavily which can bring latest news.
AIMessage(content=''", additional_kwargs={'reasoning_content': 'Okay, the user is asking for recent AI news. I need to figure out which tool to use here. Let me check the available functions.\n\nFirst, there\'s the arxiv function, which is for scientific papers in fields like physics and computer science. But the user is asking for news, not research papers. So maybe arxiv isn\'t the best fit here.\n\nNext, the wikipedia function. Wikipedia has general information, but it might not have the most up-to-date news. Plus, Wikipedia articles are more about established topics rather than recent events.\n\nThen there\'s the tavily_search_results_json function, which is described as a search engine optimized for current events. That sounds perfect for finding recent news on AI. The user wants the latest updates, so using a search engine that focuses on current information makes sense.\n\nI should call tavily_search_results_json with the query "recent AI news". That should retrieve the most relevant and up-to-date articles. Let me make sure the parameters are correct. The function requires a query, and I\'m providing "recent AI news" as the search term. Yep, that\'s appropriate.\n\nNo need to use the other functions since they don\'t fit the news aspect. Alright, I\'ll proceed with the tavily_search_results_json function.\n', 'tool_calls': [{'id': 'v31wp2yk3', 'function': {'arguments': '{"query":"recent AI news"}', 'name': 'tavily_search_results_json'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 287, 'prompt_tokens': 399, 'total_tokens': 686, 'completion_time': 1.014928323, 'prompt_time': 0.018003417, 'queue_time': 0.046043432, 'total_time': 1.03293174, 'completion_tokens_details': {'reasoning_tokens': 257}}, 'model_name': 'qwen/qwen3-32b', 'system_fingerprint': 'fp_5cf921caa2', 'service_tier': 'on_demand', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run--0c3e04f8-af0d-437b-a544-e0614d314e01-0', tool_calls=[{'name': 'tavily_search_results_json', 'args': {'query': 'recent AI news'}, 'id': 'v31wp2yk3', 'type': 'tool_call'}], usage_metadata={'input_tokens': 399, 'output_tokens': 287, 'total_tokens': 686})
- LLM is thinking and using tools as per above output !!
llm_with_tools.invoke([HumanMessage(content=f"What is the recent AI News")]).tool_calls
[{'name': 'tavily_search_results_json',
'args': {'query': 'recent AI news'},
'id': 'v74nfryb6',
'type': 'tool_call'}]
We can even see which tool is used as shown above using tool_calls.
Now, we are done with constructing tools. Lets construct a LangGraph.
## State Schema
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage
from typing import Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages:Annotated[list[AnyMessage],add_messages]
### Entire Chatbot With LangGraph
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langgraph.prebuilt import tools_condition
### Node definition
def tool_calling_llm(state:State):
return {"messages":[llm_with_tools.invoke(state["messages"])]}
# Build graph
builder = StateGraph(State)
builder.add_node("tool_calling_llm", tool_calling_llm)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "tool_calling_llm")
builder.add_conditional_edges(
"tool_calling_llm",
# If the latest message (result) from assistant is a tool call -> tools_condition routes to tools
# If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END
tools_condition,
)
builder.add_edge("tools", END)
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
Please spend some time and understand the code about initiating nodes, edges & conditional edges - conditional_edge(). Because in this case, tools is a list and we created it as a node using ToolNode class and also during conditional_edge(), we built and edge from tool_calling_llm to tools using tools_condition
Output :
messages=graph.invoke({"messages":HumanMessage(content="1706.03762")})
for m in messages['messages']:
m.pretty_print()
Point to remember : It is important to slow down here and understand how above code work line-by-line. It will give strong foundation on how to build state, nodes, edges, stategraph etc.
Pending concepts in LangGraph that we are going to see in the rest of the blog :
- State Schema
- TypedDict
- DataClass
- Pydantic
- ReAct - Reasoning + Observe + Action
- HILP - Human in the Loop
- Memory
- Streaming
- stream
- astream
- update/value
- Workflow + Agent
- Prompt Chaining
- Orchestration
- Parallelization
- Routing
- Evaluator - Optimizer
All the above concepts will be helping when we are designing the complex applications using LangGraph.
Official LangGraph documentation can be found under docs tab : https://www.langchain.com/langgraph
State
We can define State in below 3 ways:
- TypedDict - Basic level
- DataClass - Intermediate level
- Pydantic - Production grade
So far, we have been using TypedDict. We will now see the use of DataClass and Pydantic.
1) TypedDict is a python typing feature. It allows us to create dictionaries with a fixed structure, where each key has a specified type.
- Normally python dictionaries are very flexible:
- data = {'name':'Arun', 'age':36, 'skills':['Python', "LangGraph"]}
- Nothing stops us from writing:
- data = {'name':'1234', age:'thirtysix'}, because dictionaries don't enforce types
- which makes code less predictable
from typing_extensions import TypedDict
class State(TypedDict):
graph_info:str
In the above code graph_info is a string but below llm invoke still work:
graph_builder.invoke({"graph_info":123})
This is the issue with TypedDict. Dictionaries are not enforcing datatype at run time.
Limitations :
- No runtime validation
- No methods
2) DataClass in python provide another way to define structured data. Dataclasses uses a concise syntax for creating classes that primarily used to store data.
A dataclass is a feature introduced in python 3.7 (dataclasses module) that lets you create lightweight classes for holding data.
Instead of writing :
class Employee:
def __init__(self, id: int, name: str, skills: list[str]):
self.id = id
self.name = name
self.skills = skills
We can write :
from dataclasses import dataclass
# We can use below decorator
@dataclass
class Employee:
id: int
name: str
skills: list[str]
Even Dataclasses don't enforce datatype at runtime like TypedDict but the implementation is very easy.
Limitations of DataClasses :
- No runtime validation for datatype
- No built-in JSON serialization
Implementation of DataClass as State schema:
### DataClasses
from dataclasses import dataclass
@dataclass
class DataClassState:
name:str
game:Literal["badminton","cricket"]
def play_game(state:DataClassState):
print("---Play Game node has been called--")
return {"name":state.name }
def cricket(state:DataClassState):
print("-- Cricket node has been called--")
return {"name":state.name ,"game":"cricket"}
def badminton(state:DataClassState):
print("-- badminton node has been called--")
return {"name":state.name ,"game":"badminton"}
import random
def decide_play(state:DataClassState)->Literal["cricket","badminton"]:
# Here, let's just do a 50 / 50 split between nodes 2, 3
if random.random() < 0.5:
return "cricket"
else:
return "badminton"
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END
builder=StateGraph(DataClassState)
builder.add_node("playgame",play_game)
builder.add_node("cricket",cricket)
builder.add_node("badminton",badminton)
## Flow of the graph
builder.add_edge(START,"playgame")
builder.add_conditional_edges("playgame",decide_play)
builder.add_edge("cricket",END)
builder.add_edge("badminton",END)
# Add
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
graph.invoke(DataClassState(name="Arun Mathe",game="cricket"))
---Play Game node has been called--
-- Cricket node has been called--
{'name': 'Arun Mathe', 'game': 'cricket'}
graph.invoke(DataClassState(name=123, game=456))
---Play Game node has been called--
-- badminton node has been called--
{'name': 123, 'game': 'badminton'}
So, both TypedDict and DataClass don't enforce datatype during runtime but DataClass is better than TypedDict in terms of implementation. Hence Pydantic comes into picture.
3) Pydantic is a python library for data validation and parsing. It lets you define classes(called models) where:
- Each field has a type (ex: str, list[str] )
- Pydantic will check the datatype at runtime
- It also provides serialization methods(model_dump, dump_json)
xample:
from pydantic import BaseModel
class Employee(BaseModel):
id: int
name: str
skills: list[str]
If you do:
emp = Employee(id="123", name="Anil", skills=["Python"])
print(emp.id) # ✅ 123 (coerced from str to int)
Pydantic validates and even converts types automatically.
When you define your state (or node input/output) with Pydantic, LangGraph gains:
✅ Strong runtime validation of inputs and outputs.
✅ Automatic serialization (great when exporting to JSON or logs).
✅ Easier debugging (invalid/missing fields raise clear errors).
✅ Safer for production-grade workflows.
✅ In short:
In LangGraph, Pydantic is the most powerful way to define state schemas.
It ensures that every piece of data flowing between nodes is valid, structured,
and serializable, making it ideal for real-world, production-ready LangGraph
applications.
Note :
Hence Pydantic is best for production grade applications.
Sample code :
class State(BaseModel):
sal:float
Next concept is ReACT.
ReACT Agent
When the user sent a Query, it goes to LLM(
brain). LLM start thinking(
reasoning), then it will search for a tool then it will redirect query to tool(
action). Tool start executing actual task and it will produce the output, then it will send output back to LLM, then LLM will articulate the output and produce the final answer.
Try to understand below important concept : This save money will implementing tools
# Normal way of defining a method
def add(a: Int, b: Int) -> Int:
return a+b
# While using AI especially while implementing tools in Agentic AI
@Tool
def addition(a: Int, b: Int) -> Int:
"""
This function mainly used for adding two numbers.
Ex : 2 + 2 = 4
"""
return a + b
Generally, almost everyone write the first way of implementation BUT while using AI, we should write prompt(as shown in second way) as well to let LLM know that this tool is for adding 2 numbers, otherwise LLM will keep process this definition every time it received the user request to add two numbers. Incase if it is running in a loop for 1000 times and assume 3 different users requesting for an addition, it will be disaster as LLM keep consume tokens and it impact business. It cost us more $.
Suppose we have 2 tools, first one a+b, second one a+b+c ; then every time LLM process this request, it will try to analyze both the tools if tool doesn't have a doc string. Hence it is extremely important to write a doc string while implementing tools.
SLA + Cost + Latency + Efficiency everything will impact.
ReACT means :
- act - let model call specific tools
- observe - pass the tool output back to model
- reason - let the model reason about the tool output to decide what to do next
- call another tool or just respond directly
### Combine all the tools in the list
tools=[arxiv,wiki,tavily,add,divide,multiply]
## Initialize my LLM model
from langchain_groq import ChatGroq
llm=ChatGroq(model="qwen/qwen3-32b",callbacks=[tracer])
llm_with_tools=llm.bind_tools(tools)
from pprint import pprint
from langchain_core.messages import AIMessage, HumanMessage
llm_with_tools.invoke([HumanMessage(content=f"What is the recent AI News")])
llm_with_tools.invoke([HumanMessage(content=f"What is the recent AI News")]).tool_calls
## State Schema
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage
from typing import Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages:Annotated[list[AnyMessage],add_messages]
### Entire Chatbot With LangGraph
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langgraph.prebuilt import tools_condition
### Node definition
def tool_calling_llm(state:State):
return {"messages":[llm_with_tools.invoke(state["messages"])]}
# Build graph
builder = StateGraph(State)
builder.add_node("tool_calling_llm", tool_calling_llm)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "tool_calling_llm")
builder.add_conditional_edges(
"tool_calling_llm",
# If the latest message (result) from assistant is a tool call -> tools_condition routes to tools
# If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END
tools_condition,
)
# This is where we are sending the output from tools to LLM
builder.add_edge("tools","tool_calling_llm")
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
messages=graph.invoke({"messages":HumanMessage(content="Provide me the top 10 recent AI news in March 2026,add 5 plus 5 and then multiply by 10")})
for m in messages['messages']:
m.pretty_print()
messages=graph.invoke({"messages":HumanMessage(content="What is machine learning")})
for m in messages['messages']:
m.pretty_print()
messages=graph.invoke({"messages":HumanMessage(content="What is 5 plus 5")})
for m in messages['messages']:
m.pretty_print()
messages=graph.invoke({"messages":HumanMessage(content="mulitiply the number with 5")})
for m in messages['messages']:
m.pretty_print()
================================[1m Human Message [0m=================================
mulitiply the number with 5
==================================[1m Ai Message [0m==================================
The query is missing the specific number to multiply by 5.
Please provide the number you want to multiply. For example, if the number is 7,
the calculation would be 7 × 5.
See above 2 blocks of code where we add 2 numbers and then multiply the result with another number using another tool. Understand that, output from add tool is 10(5+5), this must store somewhere but LLMs are stateless. Hence we need to use a concept called memory.
Agent Memory - Lets introduce Agent with memory
LangGraph can use a checkpointer to automatically save the graph state after each step. This built-in layer gives us memory, allowing LangGraph to pick up from the last state update.
One of the easiest checkpointers to use is the MemorySaver, an in-memory key-value store for graph state. All we need to do is simply compile the graph with a checkpointer, and our graph has memory.
From the above example, Thread is nothing but collection of all checkpoints which is the last state value.
Implementation of MemorySaver :
┌──────────────┐
│ User Input │
└──────┬───────┘
↓
┌──────────────┐
│ Load State │ ← (MemorySaver loads previous state)
└──────┬───────┘
↓
┌──────────────┐
│ Node 1 │
└──────┬───────┘
↓
┌──────────────┐
│ Node 2 │
└──────┬───────┘
↓
┌──────────────┐
│ Update State │
└──────┬───────┘
↓
┌──────────────┐
│ Save State │ ← (MemorySaver saves checkpoint)
└──────────────┘
from langgraph.checkpoint.memory import MemorySaver
memory=MemorySaver()
graph_memory = builder.compile(checkpointer=memory)
# View
display(Image(graph_memory.get_graph().draw_mermaid_png()))
As shown in the above code, we implemented checkpointer but we have to implement thread as well to connect all checkpointers across state graph.
Let us say we are developing some productivity tool using GitHub copilot, and we mentioned a required to it saying - develop a multi agent system using LangGraph. Once copilot generated the code, do we deploy it into production without verification ? what about validation ? This can be achieved using Human-in-the-loop.
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict
from IPython.display import Image, display
# Define state with explicit keys
class State(TypedDict):
query: str
draft: str
final: str
# Step 1: Generation node
def generate_node(state: State):
query = state.get("query", "") # use .get() to avoid KeyError
draft = f"Draft answer for: {query}"
return {"draft": draft}
# Step 2: Human review node
def human_review_node(state: State):
print("Draft Answer:", state["draft"])
human_input = input("Type 'approve' or enter your own edit: ")
if human_input.lower().strip() == "approve":
return {"final": state["draft"]}
else:
return {"final": human_input} # take edited answer
# Build the graph
graph = StateGraph(State)
graph.add_node("generate", generate_node)
graph.add_node("review", human_review_node)
graph.set_entry_point("generate")
graph.add_edge("generate", "review")
graph.add_edge("review", END)
# Compile
app = graph.compile()
# View
display(Image(app.get_graph().draw_mermaid_png()))
Human in the loop is just a piece of python code to pause the workflow and analyze the response from LLM. This approach will be very useful to avoid hallucination in real time. We can just route the response from Bot/Node to another node called Human-in-the-loop, review and approve for a safe output.
Below piece of code implements this feature :
# Step 2: Human review node
def human_review_node(state: State):
print("Draft Answer:", state["draft"])
human_input = input("Type 'approve' or enter your own edit: ")
if human_input.lower().strip() == "approve":
return {"final": state["draft"]}
else:
return {"final": human_input} # take edited answer
When we run below code, a prompt will open and we can enter either "approve" or some other text.
# ✅ Pass query explicitly here
result = app.invoke({"query": "Explain photosynthesis in simple words"})
print("\n✅ Final Answer:", result["final"])
Output :
Draft Answer: Draft answer for: Explain photosynthesis in simple words
✅ Final Answer: Draft answer for: Explain photosynthesis in simple words
Understand, here no LLM call.
Below code have a feedback option and user can also enter his custom feedback :
from langgraph.graph import StateGraph, END
from IPython.display import Image, display
# Define state
class State(dict):
query: str
draft: str
final: str
feedback: str
# Step 1: Generation node (LLM draft simulation)
def generate_node(state: State):
query = state.get("query", "")
feedback = state.get("feedback", "")
if feedback:
draft = f"Revised draft for: {query}\n(Incorporating feedback: {feedback})"
else:
draft = f"Draft answer to: {query}"
return {"draft": draft, "feedback": ""} # reset feedback after use
# Step 2: Human review node (feedback loop)
def human_review_node(state: State):
print("\n--- Human Review Step ---")
print("Draft Answer:", state["draft"])
print("Options: ")
print(" 1. approve → accept as final")
print(" 2. edit → provide a new version")
print(" 3. revise → send back to generation with feedback")
choice = input("Enter choice (approve/edit/revise): ").strip().lower()
if choice == "approve":
return {"final": state["draft"]}
elif choice == "edit":
new_text = input("Enter your edited version: ")
return {"final": new_text}
elif choice == "revise":
feedback = input("Enter feedback for revision: ")
return {"feedback": feedback}
else:
print("Invalid choice → defaulting to draft.")
return {"final": state["draft"]}
# Build the graph
graph = StateGraph(State)
graph.add_node("generate", generate_node)
graph.add_node("review", human_review_node)
graph.set_entry_point("generate")
graph.add_edge("generate", "review")
graph.add_conditional_edges(
"review",
lambda state: "generate" if "feedback" in state and state["feedback"] else END,
{
"generate": "generate", # loop if feedback given
END: END # exit if final approved
}
)
# Compile
app = graph.compile()
# View
display(Image(app.get_graph().draw_mermaid_png()))
# Run
result = app.invoke({"query": "Explain black holes in simple terms"})
print("\n✅ Final Answer:", result["final"])
--- Human Review Step ---
Draft Answer: Draft answer to: Explain black holes in simple terms
Options:
1. approve → accept as final
2. edit → provide a new version
3. revise → send back to generation with feedback
✅ Final Answer: Draft answer to: Explain black holes in simple terms
Same code but implemented LLM call :
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict
from IPython.display import Image, display
from dotenv import load_dotenv
load_dotenv()
import os
os.environ["GROQ_API_KEY"]=os.getenv("GROQ_API_KEY")
# 1. Define state
class State(TypedDict):
query: str
draft: str
feedback: str
final: str
# 2. LLM
from langchain_groq import ChatGroq
llm=ChatGroq(model="qwen/qwen3-32b")
# 3. Node: Generate draft answer
def generate_answer(state: State):
response = llm.invoke(f"Answer clearly: {state['query']}")
return {"draft": response.content}
# 4. Node: Collect human feedback (runtime)
def get_human_feedback(state: State):
print("\n🤖 Draft Answer:", state["draft"])
feedback = input("📝 Enter your feedback (or press Enter to accept): ")
return {"feedback": feedback}
# 5. Node: Refine answer with feedback
def refine_answer(state: State):
if state["feedback"].strip():
prompt = f"Revise the answer based on this feedback: {state['feedback']}\n\nOriginal: {state['draft']}"
response = llm.invoke(prompt)
return {"final": response.content}
else:
return {"final": state["draft"]}
# 6. Graph setup
workflow = StateGraph(State)
workflow.add_node("generate", generate_answer)
workflow.add_node("human_feedback", get_human_feedback)
workflow.add_node("refine", refine_answer)
workflow.set_entry_point("generate")
workflow.add_edge("generate", "human_feedback")
workflow.add_edge("human_feedback", "refine")
workflow.add_edge("refine", END)
# 7. Compile
app = workflow.compile()
# View
display(Image(app.get_graph().draw_mermaid_png()))
# 8. Run
result = app.invoke({"query": "Explain quantum entanglement in simple terms"})
print("\n✅ Final Answer:", result["final"])
Please see below image, that's how a real time LangGraph looks : End-to-end flow
Now, we are going to see how to build a workflow using below techniques :
- Prompt Chaining
- Parallelization
- Orchestrator - worker
- Evaluator-Optimizer
- Routing
Once we build a workflow, we can integrate that with an Agent.
Prompt Chaining
Prompt chaining means breaking a complex task into smaller prompts and connecting them in sequence. Each steps output becomes the input to next step.
Also, we have some technique available which is called Query Decomposition which will help us to divide complex user input into multiple simple queries. We will discuss about it when we discuss RAG.
Implementation of prompt chaining :
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict
from IPython.display import Image, display
from dotenv import load_dotenv
load_dotenv()
import os
# 1. Define State
class State(TypedDict):
topic: str
outline: str
essay: str
# 2. LLM
from langchain_groq import ChatGroq
llm=ChatGroq(model="qwen/qwen3-32b")
# llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 3. Node: Create outline
def generate_outline(state: State):
response = llm.invoke(f"Create a simple outline for an essay on: {state['topic']}")
return {"outline": response.content}
# 4. Node: Expand outline into essay
def expand_outline(state: State):
response = llm.invoke(f"Write a detailed essay based on this outline:\n{state['outline']}")
return {"essay": response.content}
# 5. Graph Setup
workflow = StateGraph(State)
workflow.add_node("outline", generate_outline)
workflow.add_node("essay", expand_outline)
workflow.set_entry_point("outline")
workflow.add_edge("outline", "essay")
workflow.add_edge("essay", END)
# 6. Compile graph
app = workflow.compile()
# View
display(Image(app.get_graph().draw_mermaid_png()))
# 7. Run workflow
result = app.invoke({"topic": "The Importance of Renewable Energy"})
print("\n📌 Outline:", result["outline"])
print("\n✅ Final Essay:", result["essay"])
It will print both outline and essay.
Parallelization
Parallelization means running multiple tasks at the same time instead of sequentially. In LangGraph, you can design a workflow where multiple nodes branch out in parallel in one step. Once all are done, results can be combined downstream.
Implementation of Parallelization :
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict
from dotenv import load_dotenv
from IPython.display import Image, display
load_dotenv()
import os
# 1. Define state
class State(TypedDict):
query: str
short: str
detailed: str
combined: str
# 2. LLM
from langchain_groq import ChatGroq
llm=ChatGroq(model="qwen/qwen3-32b")
# 3. Node: Start (fan-out to parallel tasks)
def start_node(state: State):
return {} # nothing to add, just trigger next steps
# 4. Parallel node: Short answer
def short_answer(state: State):
resp = llm.invoke(f"Answer briefly: {state['query']}")
return {"short": resp.content}
# 5. Parallel node: Detailed answer
def detailed_answer(state: State):
resp = llm.invoke(f"Explain in detail: {state['query']}")
return {"detailed": resp.content}
# 6. Combine results
def combine(state: State):
return {
"combined": f"Short Answer:\n{state.get('short','')}\
\n\nDetailed Answer:\n{state.get('detailed','')}"
}
# 7. Build graph
workflow = StateGraph(State)
workflow.add_node("start", start_node)
workflow.add_node("short", short_answer)
workflow.add_node("detailed", detailed_answer)
workflow.add_node("combine", combine)
workflow.set_entry_point("start")
# Parallel fan-out from "start"
workflow.add_edge("start", "short")
workflow.add_edge("start", "detailed")
# Both feed into combine
workflow.add_edge("short", "combine")
workflow.add_edge("detailed", "combine")
workflow.add_edge("combine", END)
# 8. Compile
app = workflow.compile()
# View
display(Image(app.get_graph().draw_mermaid_png()))
# 9. Run
result = app.invoke({"query": "What is machine learning?"})
print("\n✅ Final Combined Answer:\n", result["combined"])
Output :
Routing
Routing is the mechanism that decides which node to go next depending on the current state or some condition. Normally, edges in a graph connect nodes in a fixed order(Node A - Node B - Node C). With routing, instead of always going to a single fixed next node, the graph can dynamically decide :
- Go to Node B if the answer looks correct
- Or Go to Node C if the answer looks incorrect
- Or Finish if the work is already done
Its like adding intelligence to the edges of your workflow graph.
Implementation of routing :
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict
from IPython.display import Image, display
# -------------------------------
# Define the State
# -------------------------------
class State(TypedDict):
query: str
draft: str
final: str
# -------------------------------
# Node 1: Generate Draft
# -------------------------------
def generate_node(state: State):
query = state["query"]
return {"draft": f"Draft answer for: {query}"}
# -------------------------------
# Node 2: Review Draft
# -------------------------------
def review_node(state: State):
draft = state["draft"]
if "error" in draft.lower():
return {"route": "fix"} # Route to fix node
elif "good" in draft.lower():
return {"route": "approve"} # Route to approval
else:
return {"route": "end"} # Route to end
# -------------------------------
# Node 3: Fix Draft
# -------------------------------
def fix_node(state: State):
return {"final": state["draft"].replace("error", "corrected")}
# -------------------------------
# Node 4: Approve Draft
# -------------------------------
def approve_node(state: State):
return {"final": f"✅ Approved: {state['draft']}"}
# -------------------------------
# Build Graph with Routing
# -------------------------------
graph = StateGraph(State)
graph.add_node("generate", generate_node)
graph.add_node("review", review_node)
graph.add_node("fix", fix_node)
graph.add_node("approve", approve_node)
graph.set_entry_point("generate")
graph.add_edge("generate", "review")
# Routing edges
graph.add_conditional_edges(
"review",
lambda state: state["route"], # Routing function
{
"fix": "fix",
"approve": "approve",
"end": END
}
)
graph.add_edge("fix", END)
graph.add_edge("approve", END)
app = graph.compile()
# View
display(Image(app.get_graph().draw_mermaid_png()))
# -------------------------------
# Run Example
# -------------------------------
print(app.invoke({"query": "This has an error"}))
# → {'query': 'This has an error', 'draft': 'Draft answer for: This has an error',
'route': 'fix', 'final': 'Draft answer for: This has an corrected'}
print(app.invoke({"query": "This is good"}))
# → {'query': 'This is good', 'draft': 'Draft answer for: This is good',
'route': 'approve', 'final': '✅ Approved: Draft answer for: This is good'}
Output :
{'query': 'This has an error', 'draft': 'Draft answer for: This has an error',
'final': 'Draft answer for: This has an corrected'}
{'query': 'This is good', 'draft': 'Draft answer for: This is good',
'final': '✅ Approved: Draft answer for: This is good'}
Key idea behind routing :
- add_conditional_edges() where routing happens
- The function (lambda state: state['route']) decides which edge to follow
- Output routes mapped to nodes(fix, approve, end)
Why routing is powerful ?
- Enables branching logic in workflows
- Helps build complex agent workflows(e.g., human approval, retrial loops, tool selection)
- Reduces wasted compute by skipping un-neccesary nodes.
In short, routing in LangGraph = traffic signals in your workflow. It directs the flow dynamically to the right lane (node) instead of always following a straight line.
Orchestrator - Worker
Orchestrator is the manager node that decides what needs to be done, breaks a problem into subtasks, and assigns them to the right worker.
Workers - are specialized nodes/agents tat perform the actual subtasks(ex : researcher, writter, summarizer)
The Orchestrator doesn't solve the problem itself - it delegates. Workers don't make global decisions - they just execute tasks given by the Orchestrator.
Implementing Orchestrator - worker :
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict
from IPython.display import Image, display
# -------------------------------
# Define the State
# -------------------------------
class State(TypedDict):
query: str
task: str
result: str
# -------------------------------
# Worker Nodes
# -------------------------------
def researcher_node(state: State):
return {"result": f"📚 Research findings about {state['query']}"}
def writer_node(state: State):
return {"result": f"✍️ Written article draft on {state['query']}"}
def summarizer_node(state: State):
return {"result": f"📝 Concise summary of {state['query']}"}
# -------------------------------
# Orchestrator Node
# -------------------------------
def orchestrator_node(state: State):
query = state["query"].lower()
if "research" in query:
return {"task": "researcher"}
elif "write" in query:
return {"task": "writer"}
elif "summary" in query or "summarize" in query:
return {"task": "summarizer"}
else:
return {"task": "unknown"}
# -------------------------------
# Build the Graph
# -------------------------------
graph = StateGraph(State)
graph.add_node("orchestrator", orchestrator_node)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)
graph.add_node("summarizer", summarizer_node)
graph.set_entry_point("orchestrator")
graph.add_conditional_edges(
"orchestrator",
lambda state: state["task"], # routing decision
{
"researcher": "researcher",
"writer": "writer",
"summarizer": "summarizer",
"unknown": END
}
)
# All workers go to END
graph.add_edge("researcher", END)
graph.add_edge("writer", END)
graph.add_edge("summarizer", END)
# Compile
app = graph.compile()
# View
display(Image(app.get_graph().draw_mermaid_png()))
# -------------------------------
# Run Examples
# -------------------------------
print(app.invoke({"query": "Do research on climate change"}))
# → {'query': 'Do research on climate change', 'task': 'researcher',
'result': '📚 Research findings about Do research on climate change'}
print(app.invoke({"query": "Write an article on AI"}))
# → {'query': 'Write an article on AI', 'task': 'writer',
'result': '✍️ Written article draft on Write an article on AI'}
print(app.invoke({"query": "Summarize the history of blockchain"}))
# → {'query': 'Summarize the history of blockchain', 'task': 'summarizer',
'result': '📝 Concise summary of Summarize the history of blockchain'}
Output :
{'query': 'Do research on climate change', 'task': 'researcher', 'result': '📚 Research findings about Do research on climate change'}
{'query': 'Write an article on AI', 'task': 'writer', 'result': '✍️ Written article draft on Write an article on AI'}
{'query': 'Summarize the history of blockchain', 'task': 'summarizer', 'result': '📝 Concise summary of Summarize the history of blockchain'}
In short: Orchestrator - worker in LangGraph is like a project manager assigning tasks to a specialized team. The Orchestrator keeps the big picture, an workers execute their expertise.
Also, these supervisor levels are classified into 6 levels. Currently, we are using Level-4 supervisor. There are Level-5, Level-6 agents also available.
Swarm multi-agent design pattern is a decentralized approach where many autonomous agents collaborate without a central supervisor, similar to how ants, bees, or birds operate in nature. This is the opposite of Supervisor Architecture. Here, intelligence emerges from collective behavior, not centralized control. This comes under Level-5 Orchestrator.
Level-6 Orchestrators are conflict resolvers. If there is a conflict between 2 or more agents, then a Level-6 classified Orchestrator can fix it. These comes under design patterns.
Evaluator - Optimizer
It's a feedback loop pattern where :
- Optimizer generates an output (solution, draft, plan)
- Evaluator inspects the output, checks for quality, correctness, or alignment.
- Based on Evaluator judgement, the process either :
- Accepts the output OR
- Sends it back to optimizer for refinement.
- This forms a closed loop until the evaluator is satisfied or a retry limit is reached.
Implementation of Evaluator - Optimizer :
from langgraph.graph import StateGraph, END
from typing import TypedDict
from IPython.display import Image, display
# Define the state
class State(TypedDict):
text: str
summary: str
feedback: str
attempts: int
# Optimizer node - generates/updates summary
def optimizer(state: State) -> State:
if state["attempts"] == 0:
# First draft
summary = f"Draft Summary: {state['text'][:50]}..."
else:
# Refined version with feedback applied
summary = f"Refined Summary after feedback: {state['feedback']}"
return {
"summary": summary,
"attempts": state["attempts"] + 1,
}
# Evaluator node - checks summary quality
def evaluator(state: State) -> State:
summary = state["summary"]
feedback = ""
# Simple evaluation rule (in real case use LLM or metrics)
if len(summary.split()) < 8:
feedback = "Too short. Please expand with more details."
elif "Refined" not in summary and state["attempts"] < 2:
feedback = "Needs refinement for clarity."
else:
feedback = "Looks good!"
return {"feedback": feedback}
# Router: Decide next step
def router(state: State):
if state["feedback"] == "Looks good!" or state["attempts"] >= 3:
return END
else:
return "optimizer"
# Build graph
workflow = StateGraph(State)
workflow.add_node("optimizer", optimizer)
workflow.add_node("evaluator", evaluator)
workflow.set_entry_point("optimizer")
workflow.add_edge("optimizer", "evaluator")
workflow.add_conditional_edges("evaluator", router, {"optimizer": "optimizer", END: END})
graph = workflow.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
# Run it
final_state = graph.invoke({"text": "LangGraph helps in orchestrating LLM workflows.", "summary": "", "feedback": "", "attempts": 0})
print("Final State:", final_state)
Output :
Final State: {'text': 'LangGraph helps in orchestrating LLM workflows.',
'summary': 'Refined Summary after feedback: Needs refinement for clarity.',
'feedback': 'Looks good!', 'attempts': 2}
Why its useful ?
- Helps build self-correcting AI workflows.
- Can implement automated grading, AQ, hallucination reduction.
- Mimics human processes of drafting + review.
We are done with all the concepts of LangGraph.
Thank you for reading this blog !
Comments
Post a Comment