Created
August 15, 2025 14:36
-
-
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
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 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