Skip to content

Instantly share code, notes, and snippets.

@KrishnanSriram
Created August 15, 2025 14:36
Show Gist options
  • Save KrishnanSriram/83022908921de6e5cac7a4f545bbb680 to your computer and use it in GitHub Desktop.
Save KrishnanSriram/83022908921de6e5cac7a4f545bbb680 to your computer and use it in GitHub Desktop.
A simple example of agent like solution with LangGraph that does a sequential execution of multiple tasks
from typing import Dict, Any, List, TypedDict, Literal
import re
from langchain_ollama import ChatOllama
from langchain.tools import tool
from langgraph.graph import StateGraph, END, START
from numpy.matlib import empty
# --------------------------
# TOOLS
# --------------------------
@tool
def extract_positive_numbers(sentence: str) -> Dict[str, Any]:
"""Extract positive integers from the given sentence."""
numbers = [int(n) for n in re.findall(r"\b\d+\b", sentence) if int(n) > 0]
if not numbers:
return {"numbers": [], "message": "No positive numbers found. Cannot proceed."}
return {"numbers": numbers, "message": None}
@tool
def add_two_numbers(numbers: List[int]) -> Dict[str, Any]:
"""Add the first two numbers."""
if len(numbers) < 2:
return {"add": None, "message": "Need at least 2 numbers for addition."}
return {"add": numbers[0] + numbers[1], "message": None}
@tool
def subtract_two_numbers(numbers: List[int]) -> Dict[str, Any]:
"""Subtract the second number from the first."""
if len(numbers) < 2:
return {"subtract": None, "message": "Need at least 2 numbers for subtraction."}
return {"subtract": numbers[0] - numbers[1], "message": None}
@tool
def multiply_two_numbers(numbers: List[int]) -> Dict[str, Any]:
"""Multiply the first two numbers."""
if len(numbers) < 2:
return {"multiply": None, "message": "Need at least 2 numbers for multiplication."}
return {"multiply": numbers[0] * numbers[1], "message": None}
# --------------------------
# STATE
# --------------------------
class CalcState(TypedDict):
"""Custom state for the calculator pipeline."""
sentence: str
message: str
stop_message: str
numbers: [int]
add: int
subtract: int
multiply: int
already_run: []
next_tool: {}
# --------------------------
# AGENT NODE
# --------------------------
llm = ChatOllama(model="llama3.2", temperature=0)
TOOLS = [
extract_positive_numbers,
add_two_numbers,
subtract_two_numbers,
multiply_two_numbers
]
def agent_node(state: CalcState) -> CalcState:
"""Decide which tool to run next, or end."""
already_run = state.get("already_run", [])
# If Tool 1 failed, end immediately
if "extract_positive_numbers" in already_run and len(state.get("numbers", [])) <= 0:
state["stop_message"] = "No numbers found. Cannot execute operations."
return state
# If there are still tools left, pick the next
if len(already_run) < len(TOOLS):
next_tool = TOOLS[len(already_run)]
state["next_tool"] = next_tool
return state
# Otherwise, we’re done
return state
def tool_runner_node(state: CalcState) -> CalcState:
tool = state["next_tool"]
# Build inputs
if tool.name == "extract_positive_numbers":
input_data = {"sentence": state["sentence"]}
else:
input_data = {"numbers": state.get("numbers", [])}
# Run the tool
result = tool.invoke(input_data)
# Record results
state["last_result"] = result
if "numbers" in result:
# Also reassign to avoid sharing the tool's internal list
state["numbers"] = list(result["numbers"])
if "add" in result:
state["add"] = result["add"]
if "subtract" in result:
state["subtract"] = result["subtract"]
if "multiply" in result:
state["multiply"] = result["multiply"]
# ❌ avoid in-place append; ✅ reassign a new list
already_run = state.get("already_run", [])
already_run.append(tool.name)
state["already_run"] = already_run
# Clear the consumed pointer (optional but tidy)
state.pop("next_tool", None)
return state
def decide_next(state: CalcState) -> Literal["tool_runner", "end"]:
if len(state.get("already_run", [])) == len(TOOLS) or len(state.get("stop_message", '')) > 0:
return "end"
return "tool_runner"
# --------------------------
# GRAPH
# --------------------------
graph = StateGraph(CalcState)
graph.add_node("agent", agent_node)
graph.add_node("tool_runner", tool_runner_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", decide_next, {"tool_runner": "tool_runner", "end": END})
graph.add_edge("tool_runner", "agent")
app = graph.compile()
# --------------------------
# TEST
# --------------------------
print("=== Example 1: With numbers ===")
res1 = app.invoke({"sentence": "I have 8 apples and 3 oranges"})
print(res1)
# print("\n=== Example 2: Without numbers ===")
# res2 = app.invoke({"sentence": "I have apples and oranges"})
# print(res2)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment