Continue to read this article if you want to experiment with Lang Graph. LangGraph is a Python framework designed to build cyclical, state-driven AI applications (often called AI Agents).
┌────────────────────────┐
│ STATE |
| (The Shared Memory) │
└───────────┬────────────┘
│
┌─────────┴─────────┐
▼ ▼
┌───────────┐ ┌───────────┐
│ NODE │ ─────►│ EDGE │
│ (Actions) │ │(Routing/If)
└───────────┘ └───────────┘
Let's work with a very basic example, not to complicate things, we don't make any external call(no llm calls etc)
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
# 1. Define the Shared State structure
class State(TypedDict):
value: str
def First_Node(state: State):
print("I am from first Node")
return {"value": "First Node done"}
def Second_Node(state: State):
print("I am from Second Node")
return {"value": "Second Node done"}
workflow = StateGraph(State)
workflow.add_node(First_Node)
workflow.add_node(Second_Node)
workflow.add_edge(START, "First_Node")
workflow.add_edge("First_Node", "Second_Node")
workflow.add_edge("Second_Node", END)
graph = workflow.compile()
final_response = graph.invoke({"value": []})
print(graph.get_graph().draw_mermaid())
print(f"the final value is : {final_response}")
Here is the output we get
I am from first Node
I am from Second Node
---
config:
flowchart:
curve: linear
---
graph TD;
__start__([__start__
]):::first
First_Node(First_Node)
Second_Node(Second_Node)
__end__([__end__
]):::last
First_Node --> Second_Node;
__start__ --> First_Node;
Second_Node --> __end__;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
the final value is : {'value': 'Second Node done'}
Going little further into the topic. Use an annotated list to preserve the history
class State(TypedDict):
value: str
history: Annotated[list[str], operator.add]
def First_Node(state: State):
print("I am from first Node")
return {"value": "First Node done", "history": ["First Node Activated"]}
def Second_Node(state: State):
print("I am from Second Node")
return {"value": "Second Node done", "history": ["Second Node Activated"]}
...
final_response = graph.invoke({"value": [], "history": ["User initiated session"]})
You would see an output like this.
workflow.add_edge(START, "Node_A")
workflow.add_edge(START, "Node_B")
Now, apart from nodes being python functions, we can add variety of things like LCEL (Lang Chain Expression Language), sub graphs etc. Consider the following:
..
model = ChatOpenAI(model="..")
prompt = ChatPromptTemplate.from_template("Summarize this text in one sentence: {input_text}")
lcel_chain = prompt | model | (lambda msg: {"summary": msg.content})
we can pass this as a node to a lang graph.
workflow.add_node("summarizer_node", lcel_chain)
Or sometimes, we may need a placeholder node, something like below
def DUMMY(state: State):
pass
workflow.add_node(DUMMY)
You may also try to pass another workflow graph as subgraph.