Created
August 15, 2025 14:35
-
-
Save KrishnanSriram/460fc2940bdd894badca1f2be088101b to your computer and use it in GitHub Desktop.
A simple sequential direction implementation of LangGraph solution
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from typing import List, Dict, Any, Optional, TypedDict | |
import re | |
from langchain.chains.question_answering.map_reduce_prompt import messages | |
from langchain_core.output_parsers import StrOutputParser | |
from langchain_core.prompts import PromptTemplate | |
from langchain_core.messages import HumanMessage | |
from langchain_core.runnables import RunnablePassthrough | |
from langchain_ollama import ChatOllama | |
from langgraph.graph import StateGraph, END, START | |
from langgraph.prebuilt import ToolNode | |
# -------------------------- | |
# TOOL IMPLEMENTATIONS | |
# -------------------------- | |
def extract_positive_numbers(sentence: str) -> Dict[str, Any]: | |
"""Extract positive integers from the sentence.""" | |
numbers = [int(n) for n in re.findall(r"\b\d+\b", sentence) if int(n) > 0] | |
print(f"Extracted numbers - {numbers}") | |
if not numbers: | |
return {"numbers": [], "message": "No positive numbers found. Cannot proceed with operations."} | |
return {"numbers": numbers, "message": None} | |
def add_two_numbers(numbers: List[int]) -> Dict[str, Any]: | |
if len(numbers) < 2: | |
return {"add": None, "message": "Need at least 2 numbers for addition."} | |
return {"add": numbers[0] + numbers[1], "message": None} | |
def subtract_two_numbers(numbers: List[int]) -> Dict[str, Any]: | |
if len(numbers) < 2: | |
return {"subtract": None, "message": "Need at least 2 numbers for subtraction."} | |
return {"subtract": numbers[0] - numbers[1], "message": None} | |
def multiply_two_numbers(numbers: List[int]) -> Dict[str, Any]: | |
if len(numbers) < 2: | |
return {"multiply": None, "message": "Need at least 2 numbers for multiplication."} | |
return {"multiply": numbers[0] * numbers[1], "message": None} | |
# -------------------------- | |
# LANGGRAPH STATE | |
# -------------------------- | |
class CalcState(TypedDict): | |
sentence: str | |
message: str | |
stop_message: str | |
numbers: [] | |
add: int | |
subtract: int | |
multiply: int | |
final_result: {} | |
# -------------------------- | |
# WRAPPER FUNCTIONS FOR TOOLS | |
# -------------------------- | |
def tool1_node(state: CalcState) -> CalcState: | |
print(f"Current state is - {state}") | |
result = extract_positive_numbers(state["sentence"]) | |
if result["message"]: | |
state["stop_message"] = result["message"] | |
state["numbers"] = [] | |
return state | |
state["numbers"] = result["numbers"] | |
return state | |
def tool2_node(state: CalcState) -> CalcState: | |
if "stop_message" in state: | |
return state | |
result = add_two_numbers(state["numbers"]) | |
state["add"] = result["add"] | |
return state | |
def tool3_node(state: CalcState) -> CalcState: | |
if "stop_message" in state: | |
return state | |
result = subtract_two_numbers(state["numbers"]) | |
state["subtract"] = result["subtract"] | |
return state | |
def tool4_node(state: CalcState) -> CalcState: | |
if "stop_message" in state: | |
return state | |
result = multiply_two_numbers(state["numbers"]) | |
state["multiply"] = result["multiply"] | |
return state | |
def final_node(state: CalcState) -> CalcState: | |
if "stop_message" in state: | |
state["final_result"] = state["stop_message"] | |
else: | |
state["final_result"] = { | |
"numbers": state["numbers"], | |
"add": state["add"], | |
"subtract": state["subtract"], | |
"multiply": state["multiply"] | |
} | |
return state | |
# -------------------------- | |
# BUILD GRAPH | |
# -------------------------- | |
graph = StateGraph(CalcState) | |
graph.add_node("tool1", tool1_node) | |
graph.add_node("tool2", tool2_node) | |
graph.add_node("tool3", tool3_node) | |
graph.add_node("tool4", tool4_node) | |
graph.add_node("final", final_node) | |
graph.add_edge(START, "tool1") | |
graph.add_edge("tool1", "tool2") | |
graph.add_edge("tool2", "tool3") | |
graph.add_edge("tool3", "tool4") | |
graph.add_edge("tool4", "final") | |
graph.add_edge("final", END) | |
app = graph.compile() | |
# -------------------------- | |
# TEST RUNS | |
# -------------------------- | |
print("=== Example 1: With numbers ===") | |
input_prompt = "I have 8 apples and 3 oranges" | |
result1 = app.invoke({"sentence": "I have 8 apples and 3 oranges"}) | |
print(result1["final_result"]) | |
# print("\n=== Example 2: Without numbers ===") | |
# result2 = app.invoke({"sentence": "I have apples and oranges"}) | |
# print(result2["final_result"]) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment