On this tutorial, we construct a production-style Route Optimizer Agent for a logistics dispatch heart utilizing the newest LangChain agent APIs. We design a tool-driven workflow through which the agent reliably computes distances, ETAs, and optimum routes relatively than guessing, and we implement structured outputs to make the outcomes instantly usable in downstream techniques. We combine geographic calculations, configurable pace profiles, site visitors buffers, and multi-stop route optimization, making certain the agent behaves deterministically whereas nonetheless reasoning flexibly by way of instruments.
!pip -q set up -U langchain langchain-openai pydantic
import os
from getpass import getpass
if not os.environ.get("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = getpass("Enter OPENAI_API_KEY (enter hidden): ")
from typing import Dict, Record, Elective, Tuple, Any
from math import radians, sin, cos, sqrt, atan2
from pydantic import BaseModel, Subject, ValidationError
from langchain_openai import ChatOpenAI
from langchain.instruments import instrument
from langchain.brokers import create_agentWe arrange the execution surroundings and guarantee all required libraries are put in and imported appropriately. We securely load the OpenAI API key so the agent can work together with the language mannequin with out hardcoding credentials. We additionally put together the core dependencies that energy instruments, brokers, and structured outputs.
SITES: Dict[str, Dict[str, Any]] = {
"Rig_A": {"lat": 23.5880, "lon": 58.3829, "kind": "rig"},
"Rig_B": {"lat": 23.6100, "lon": 58.5400, "kind": "rig"},
"Rig_C": {"lat": 23.4500, "lon": 58.3000, "kind": "rig"},
"Yard_Main": {"lat": 23.5700, "lon": 58.4100, "kind": "yard"},
"Depot_1": {"lat": 23.5200, "lon": 58.4700, "kind": "depot"},
"Depot_2": {"lat": 23.6400, "lon": 58.4300, "kind": "depot"},
}
SPEED_PROFILES: Dict[str, float] = {
"freeway": 90.0,
"arterial": 65.0,
"native": 45.0,
}
DEFAULT_TRAFFIC_MULTIPLIER = 1.10
def haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
R = 6371.0
dlat = radians(lat2 - lat1)
dlon = radians(lon2 - lon1)
a = sin(dlat / 2) ** 2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon / 2) ** 2
return R * cWe outline the core area information representing rigs, yards, and depots together with their geographic coordinates. We set up pace profiles and a default site visitors multiplier to replicate life like driving situations. We additionally implement the Haversine distance operate, which serves because the mathematical spine of all routing choices.
def _normalize_site_name(identify: str) -> str:
return identify.strip()
def _assert_site_exists(identify: str) -> None:
if identify not in SITES:
increase ValueError(f"Unknown web site '{identify}'. Use list_sites() or suggest_site().")
def _distance_between(a: str, b: str) -> float:
_assert_site_exists(a)
_assert_site_exists(b)
sa, sb = SITES[a], SITES[b]
return float(haversine_km(sa["lat"], sa["lon"], sb["lat"], sb["lon"]))
def _eta_minutes(distance_km: float, speed_kmph: float, traffic_multiplier: float) -> float:
pace = max(float(speed_kmph), 1e-6)
base_minutes = (distance_km / pace) * 60.0
return float(base_minutes * max(float(traffic_multiplier), 0.0))
def compute_route_metrics(path: Record[str], speed_kmph: float, traffic_multiplier: float) -> Dict[str, Any]:
if len(path) < 2:
increase ValueError("Route path should embrace at the least origin and vacation spot.")
for s in path:
_assert_site_exists(s)
legs = []
total_km = 0.0
total_min = 0.0
for i in vary(len(path) - 1):
a, b = path[i], path[i + 1]
d_km = _distance_between(a, b)
t_min = _eta_minutes(d_km, speed_kmph, traffic_multiplier)
legs.append({"from": a, "to": b, "distance_km": d_km, "eta_minutes": t_min})
total_km += d_km
total_min += t_min
return {"route": path, "distance_km": float(total_km), "eta_minutes": float(total_min), "legs": legs}We construct the low-level utility capabilities that validate web site names and compute distances and journey occasions. We implement logic to calculate per-leg and whole route metrics deterministically. This ensures that each ETA and distance returned by the agent relies on specific computation relatively than inference.
def _all_paths_with_waypoints(origin: str, vacation spot: str, waypoints: Record[str], max_stops: int) -> Record[List[str]]:
from itertools import permutations
waypoints = [w for w in waypoints if w not in (origin, destination)]
max_stops = int(max(0, max_stops))
candidates = []
for ok in vary(0, min(len(waypoints), max_stops) + 1):
for perm in permutations(waypoints, ok):
candidates.append([origin, *perm, destination])
if [origin, destination] not in candidates:
candidates.insert(0, [origin, destination])
return candidates
def find_best_route(origin: str, vacation spot: str, allowed_waypoints: Elective[List[str]], max_stops: int, speed_kmph: float, traffic_multiplier: float, goal: str, top_k: int) -> Dict[str, Any]:
origin = _normalize_site_name(origin)
vacation spot = _normalize_site_name(vacation spot)
_assert_site_exists(origin)
_assert_site_exists(vacation spot)
allowed_waypoints = allowed_waypoints or []
for w in allowed_waypoints:
_assert_site_exists(_normalize_site_name(w))
goal = (goal or "eta").strip().decrease()
if goal not in {"eta", "distance"}:
increase ValueError("goal should be certainly one of: 'eta', 'distance'")
top_k = max(1, int(top_k))
candidates = _all_paths_with_waypoints(origin, vacation spot, allowed_waypoints, max_stops=max_stops)
scored = []
for path in candidates:
metrics = compute_route_metrics(path, speed_kmph=speed_kmph, traffic_multiplier=traffic_multiplier)
rating = metrics["eta_minutes"] if goal == "eta" else metrics["distance_km"]
scored.append((rating, metrics))
scored.type(key=lambda x: x[0])
greatest = scored[0][1]
alternate options = [m for _, m in scored[1:top_k]]
return {"greatest": greatest, "alternate options": alternate options, "goal": goal}We introduce multi-stop routing logic by producing candidate paths with non-compulsory waypoints. We consider every candidate route in opposition to a transparent optimization goal, equivalent to ETA or distance. We then rank routes and extract the best choice together with a set of robust alternate options.
@instrument
def list_sites(site_type: Elective[str] = None) -> Record[str]:
if site_type:
st = site_type.strip().decrease()
return sorted([k for k, v in SITES.items() if str(v.get("type", "")).lower() == st])
return sorted(SITES.keys())
@instrument
def get_site_details(web site: str) -> Dict[str, Any]:
s = _normalize_site_name(web site)
_assert_site_exists(s)
return {"web site": s, **SITES[s]}
@instrument
def suggest_site(question: str, max_suggestions: int = 5) -> Record[str]:
q = (question or "").strip().decrease()
max_suggestions = max(1, int(max_suggestions))
scored = []
for identify in SITES.keys():
n = identify.decrease()
frequent = len(set(q) & set(n))
bonus = 5 if q and q in n else 0
scored.append((frequent + bonus, identify))
scored.type(key=lambda x: x[0], reverse=True)
return [name for _, name in scored[:max_suggestions]]
@instrument
def compute_direct_route(origin: str, vacation spot: str, road_class: str = "arterial", traffic_multiplier: float = DEFAULT_TRAFFIC_MULTIPLIER) -> Dict[str, Any]:
origin = _normalize_site_name(origin)
vacation spot = _normalize_site_name(vacation spot)
rc = (road_class or "arterial").strip().decrease()
if rc not in SPEED_PROFILES:
increase ValueError(f"Unknown road_class '{road_class}'. Use certainly one of: {sorted(SPEED_PROFILES.keys())}")
pace = SPEED_PROFILES[rc]
return compute_route_metrics([origin, destination], speed_kmph=pace, traffic_multiplier=float(traffic_multiplier))
@instrument
def optimize_route(origin: str, vacation spot: str, allowed_waypoints: Elective[List[str]] = None, max_stops: int = 2, road_class: str = "arterial", traffic_multiplier: float = DEFAULT_TRAFFIC_MULTIPLIER, goal: str = "eta", top_k: int = 3) -> Dict[str, Any]:
origin = _normalize_site_name(origin)
vacation spot = _normalize_site_name(vacation spot)
rc = (road_class or "arterial").strip().decrease()
if rc not in SPEED_PROFILES:
increase ValueError(f"Unknown road_class '{road_class}'. Use certainly one of: {sorted(SPEED_PROFILES.keys())}")
pace = SPEED_PROFILES[rc]
allowed_waypoints = allowed_waypoints or []
allowed_waypoints = [_normalize_site_name(w) for w in allowed_waypoints]
return find_best_route(origin, vacation spot, allowed_waypoints, int(max_stops), float(pace), float(traffic_multiplier), str(goal), int(top_k))We expose the routing and discovery logic as callable instruments for the agent. We enable the agent to checklist websites, examine web site particulars, resolve ambiguous names, and compute each direct and optimized routes. This instrument layer ensures that the agent all the time causes by calling verified capabilities relatively than hallucinating outcomes.
class RouteLeg(BaseModel):
from_site: str
to_site: str
distance_km: float
eta_minutes: float
class RoutePlan(BaseModel):
route: Record[str]
distance_km: float
eta_minutes: float
legs: Record[RouteLeg]
goal: str
class RouteDecision(BaseModel):
chosen: RoutePlan
alternate options: Record[RoutePlan] = []
assumptions: Dict[str, Any] = {}
notes: str = ""
audit: Record[str] = []
llm = ChatOpenAI(mannequin="gpt-4o-mini", temperature=0.2)
SYSTEM_PROMPT = (
"You're the Route Optimizer Agent for a logistics dispatch heart.n"
"You MUST use instruments for any distance/ETA calculation.n"
"Return ONLY the structured RouteDecision."
)
route_agent = create_agent(
mannequin=llm,
instruments=[list_sites, get_site_details, suggest_site, compute_direct_route, optimize_route],
system_prompt=SYSTEM_PROMPT,
response_format=RouteDecision,
)
def get_route_decision(origin: str, vacation spot: str, road_class: str = "arterial", traffic_multiplier: float = DEFAULT_TRAFFIC_MULTIPLIER, allowed_waypoints: Elective[List[str]] = None, max_stops: int = 2, goal: str = "eta", top_k: int = 3) -> RouteDecision:
user_msg = {
"function": "consumer",
"content material": (
f"Optimize the route from {origin} to {vacation spot}.n"
f"road_class={road_class}, traffic_multiplier={traffic_multiplier}n"
f"goal={goal}, top_k={top_k}n"
f"allowed_waypoints={allowed_waypoints}, max_stops={max_stops}n"
"Return the structured RouteDecision solely."
),
}
consequence = route_agent.invoke({"messages": [user_msg]})
return consequence["structured_response"]
decision1 = get_route_decision("Yard_Main", "Rig_B", road_class="arterial", traffic_multiplier=1.12)
print(decision1.model_dump())
decision2 = get_route_decision("Rig_C", "Rig_B", road_class="freeway", traffic_multiplier=1.08, allowed_waypoints=["Depot_1", "Depot_2", "Yard_Main"], max_stops=2, goal="eta", top_k=3)
print(decision2.model_dump())We outline strict Pydantic schemas to implement structured, machine-readable outputs from the agent. We initialize the language mannequin and create the agent with a transparent system immediate and response format. We then display learn how to invoke the agent and procure dependable route choices prepared for actual logistics workflows.
In conclusion, we have now applied a strong, extensible route optimization agent that selects the perfect path between websites whereas clearly explaining its assumptions and alternate options. We demonstrated how combining deterministic routing logic with a tool-calling LLM produces dependable, auditable choices appropriate for actual logistics operations. This basis permits us to simply prolong the system with stay site visitors information, fleet constraints, or cost-based targets, making the agent a sensible element in a bigger dispatch or fleet-management platform.
Try the Full Codes right here. Additionally, be at liberty to observe us on Twitter and don’t overlook to affix our 100k+ ML SubReddit and Subscribe to our Publication. Wait! are you on telegram? now you’ll be able to be part of us on telegram as properly.
