Category: Uncategorized

In this tutorial, we implement an agentic AI pattern using LangGraph that treats reasoning and action as a transactional workflow rather than a single-shot decision. We model a two-phase commit system in which an agent stages reversible changes, validates strict invariants, pauses for human approval via graph interrupts, and commits or rolls back only then. With this, we demonstrate how agentic systems can be designed with safety, auditability, and controllability at their core, moving beyond reactive chat agents toward structured, governance-aware AI workflows that run reliably in Google Colab using OpenAI models. Check out the Full Codes here.

Copy CodeCopiedUse a different Browser!pip -q install -U langgraph langchain-openai

import os, json, uuid, copy, math, re, operator
from typing import Any, Dict, List, Optional
from typing_extensions import TypedDict, Annotated

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, AnyMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

def _set_env_openai():
if os.environ.get(“OPENAI_API_KEY”):
return
try:
from google.colab import userdata
k = userdata.get(“OPENAI_API_KEY”)
if k:
os.environ[“OPENAI_API_KEY”] = k
return
except Exception:
pass
import getpass
os.environ[“OPENAI_API_KEY”] = getpass.getpass(“Enter OPENAI_API_KEY: “)

_set_env_openai()

MODEL = os.environ.get(“OPENAI_MODEL”, “gpt-4o-mini”)
llm = ChatOpenAI(model=MODEL, temperature=0)

We set up the execution environment by installing LangGraph and initializing the OpenAI model. We securely load the API key and configure a deterministic LLM, ensuring that all downstream agent behavior remains reproducible and controlled. Check out the Full Codes here.

Copy CodeCopiedUse a different BrowserSAMPLE_LEDGER = [
{“txn_id”: “T001”, “name”: “Asha”, “email”: “ASHA@Example.com”, “amount”: “1,250.50”, “date”: “12/01/2025”, “note”: “Membership renewal”},
{“txn_id”: “T002”, “name”: “Ravi”, “email”: “ravi@example.com”, “amount”: “-500”, “date”: “2025-12-02”, “note”: “Chargeback?”},
{“txn_id”: “T003”, “name”: “Sara”, “email”: “sara@example.com”, “amount”: “700”, “date”: “02-12-2025”, “note”: “Late fee waived”},
{“txn_id”: “T003”, “name”: “Sara”, “email”: “sara@example.com”, “amount”: “700”, “date”: “02-12-2025”, “note”: “Duplicate row”},
{“txn_id”: “T004”, “name”: “Lee”, “email”: “lee@example.com”, “amount”: “NaN”, “date”: “2025/12/03”, “note”: “Bad amount”},
]

ALLOWED_OPS = {“replace”, “remove”, “add”}

def _parse_amount(x):
if isinstance(x, (int, float)):
return float(x)
if isinstance(x, str):
try:
return float(x.replace(“,”, “”))
except:
return None
return None

def _iso_date(d):
if not isinstance(d, str):
return None
d = d.replace(“/”, “-“)
p = d.split(“-“)
if len(p) == 3 and len(p[0]) == 4:
return d
if len(p) == 3 and len(p[2]) == 4:
return f”{p[2]}-{p[1]}-{p[0]}”
return None

def profile_ledger(rows):
seen, anomalies = {}, []
for i, r in enumerate(rows):
if _parse_amount(r.get(“amount”)) is None:
anomalies.append(i)
if r.get(“txn_id”) in seen:
anomalies.append(i)
seen[r.get(“txn_id”)] = i
return {“rows”: len(rows), “anomalies”: anomalies}

def apply_patch(rows, patch):
out = copy.deepcopy(rows)
for op in sorted([p for p in patch if p[“op”] == “remove”], key=lambda x: x[“idx”], reverse=True):
out.pop(op[“idx”])
for op in patch:
if op[“op”] in {“add”, “replace”}:
out[op[“idx”]][op[“field”]] = op[“value”]
return out

def validate(rows):
issues = []
for i, r in enumerate(rows):
if _parse_amount(r.get(“amount”)) is None:
issues.append(i)
if _iso_date(r.get(“date”)) is None:
issues.append(i)
return {“ok”: len(issues) == 0, “issues”: issues}

We define the core ledger abstraction along with the patching, normalization, and validation logic. We treat data transformations as reversible operations, allowing the agent to reason about changes safely before committing them. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserclass TxnState(TypedDict):
messages: Annotated[List[AnyMessage], add_messages]
raw_rows: List[Dict[str, Any]]
sandbox_rows: List[Dict[str, Any]]
patch: List[Dict[str, Any]]
validation: Dict[str, Any]
approved: Optional[bool]

def node_profile(state):
p = profile_ledger(state[“raw_rows”])
return {“messages”: [AIMessage(content=json.dumps(p))]}

def node_patch(state):
sys = SystemMessage(content=”Return a JSON patch list fixing amounts, dates, emails, duplicates”)
usr = HumanMessage(content=json.dumps(state[“raw_rows”]))
r = llm.invoke([sys, usr])
patch = json.loads(re.search(r”[.*]”, r.content, re.S).group())
return {“patch”: patch, “messages”: [AIMessage(content=json.dumps(patch))]}

def node_apply(state):
return {“sandbox_rows”: apply_patch(state[“raw_rows”], state[“patch”])}

def node_validate(state):
v = validate(state[“sandbox_rows”])
return {“validation”: v, “messages”: [AIMessage(content=json.dumps(v))]}

def node_approve(state):
decision = interrupt({“validation”: state[“validation”]})
return {“approved”: decision == “approve”}

def node_commit(state):
return {“messages”: [AIMessage(content=”COMMITTED”)]}

def node_rollback(state):
return {“messages”: [AIMessage(content=”ROLLED BACK”)]}

We model the agent’s internal state and define each node in the LangGraph workflow. We express agent behavior as discrete, inspectable steps that transform state while preserving message history. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserbuilder = StateGraph(TxnState)

builder.add_node(“profile”, node_profile)
builder.add_node(“patch”, node_patch)
builder.add_node(“apply”, node_apply)
builder.add_node(“validate”, node_validate)
builder.add_node(“approve”, node_approve)
builder.add_node(“commit”, node_commit)
builder.add_node(“rollback”, node_rollback)

builder.add_edge(START, “profile”)
builder.add_edge(“profile”, “patch”)
builder.add_edge(“patch”, “apply”)
builder.add_edge(“apply”, “validate”)

builder.add_conditional_edges(
“validate”,
lambda s: “approve” if s[“validation”][“ok”] else “rollback”,
{“approve”: “approve”, “rollback”: “rollback”}
)

builder.add_conditional_edges(
“approve”,
lambda s: “commit” if s[“approved”] else “rollback”,
{“commit”: “commit”, “rollback”: “rollback”}
)

builder.add_edge(“commit”, END)
builder.add_edge(“rollback”, END)

app = builder.compile(checkpointer=InMemorySaver())

We construct the LangGraph state machine and explicitly encode the control flow between profiling, patching, validation, approval, and finalization. We use conditional edges to enforce governance rules rather than rely on implicit model decisions. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserdef run():
state = {
“messages”: [],
“raw_rows”: SAMPLE_LEDGER,
“sandbox_rows”: [],
“patch”: [],
“validation”: {},
“approved”: None,
}

cfg = {“configurable”: {“thread_id”: “txn-demo”}}
out = app.invoke(state, config=cfg)

if “__interrupt__” in out:
print(json.dumps(out[“__interrupt__”], indent=2))
decision = input(“approve / reject: “).strip()
out = app.invoke(Command(resume=decision), config=cfg)

print(out[“messages”][-1].content)

run()

We run the transactional agent and handle human-in-the-loop approval through graph interrupts. We resume execution deterministically, demonstrating how agentic workflows can pause, accept external input, and safely conclude with either a commit or rollback.

In conclusion, we showed how LangGraph enables us to build agents that reason over states, enforce validation gates, and collaborate with humans at precisely defined control points. We treated the agent not as an oracle, but as a transaction coordinator that can stage, inspect, and reverse its own actions while maintaining a full audit trail. This approach highlights how agentic AI can be applied to real-world systems that require trust, compliance, and recoverability, and it provides a practical foundation for building production-grade autonomous workflows that remain safe, transparent, and human-supervised.

Check out the Full Codes here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post How to Design Transactional Agentic AI Systems with LangGraph Using Two-Phase Commit, Human Interrupts, and Safe Rollbacks appeared first on MarkTechPost.

Cloudflare has open sourced tokio-quiche, an asynchronous QUIC and HTTP/3 Rust library that wraps its battle tested quiche implementation with the Tokio runtime. The library has been refined inside production systems such as Apple iCloud Private Relay, next generation Oxy based proxies and WARP’s MASQUE client, where it handles millions of HTTP/3 requests per second with low latency and high throughput. tokio-quiche targets Rust teams that want QUIC and HTTP/3 without writing their own UDP and event loop integration code.

From quiche to tokio-quiche

quiche is Cloudflare’s open source QUIC and HTTP/3 implementation written in Rust and designed as a low level, sans-io library. It implements the QUIC transport state machine, including connection establishment, flow control and stream multiplexing, while making no assumptions about how applications perform IO. To use quiche directly, integrators must open UDP sockets, send and receive datagrams, manage timers and feed all packet data into quiche in the correct order. This design gives flexibility, but it makes integration error prone and time consuming.

tokio-quiche packages this integration work into a reusable crate. It combines the sans-io QUIC or HTTP/3 implementation from quiche with the Tokio async runtime, and exposes an API that already manages UDP sockets, packet routing and calls into the quiche state machine.

Actor based architecture on Tokio

Internally, tokio-quiche uses an actor model on top of Tokio. Actors are small tasks with local state that communicate through message passing over channels, which aligns well with sans-io protocol implementations that own internal state and operate on message like buffers.

The primary actor is the IO loop actor, which moves packets between quiche and the UDP socket. One of the key message types is an Incoming struct that describes received UDP packets. Async integration follows a fixed pattern, the IO loop awaits new messages, translates them into inputs for quiche, advances the QUIC state machine, then translates outputs into outbound packets that are written back to the socket.

For each UDP socket, tokio-quiche spawns two important tasks. InboundPacketRouter owns the receiving half of the socket and routes inbound datagrams by destination connection ID to per connection channels. IoWorker is the per connection IO loop and drives a single quiche Connection, interleaving calls to quiche with calls to application specific logic implemented through ApplicationOverQuic. This design encapsulates connection state inside each actor and keeps QUIC processing isolated from higher level protocol code.

ApplicationOverQuic and H3Driver

QUIC is a transport protocol and can carry multiple application protocols. HTTP/3, DNS over QUIC and Media over QUIC are examples covered by IETF specifications. To avoid coupling tokio-quiche to a single protocol, Cloudflare team exposes an ApplicationOverQuic trait. The trait abstracts over quiche methods and the underlying IO, and presents higher level events and hooks to the application that implements the protocol. For example, the HTTP/3 debug and test client h3i uses a non HTTP/3 implementation of ApplicationOverQuic.

On top of this trait, tokio-quiche ships a dedicated HTTP/3 focused implementation named H3Driver. H3Driver connects quiche’s HTTP/3 module to the IO loop actor and converts raw HTTP/3 events into higher level events with asynchronous body streams that are convenient for application code. H3Driver is generic and exposes ServerH3Driver and ClientH3Driver variants that add server side and client side behavior on top of the core driver. These components provide the building blocks for HTTP/3 servers and clients that share implementation patterns with Cloudflare’s internal infrastructure.

Production usage and roadmap

tokio-quiche has been used for several years inside Cloudflare before its public release. It powers Proxy B in Apple iCloud Private Relay, Oxy based HTTP/3 servers and the WARP MASQUE client, as well as the async version of h3i. In the WARP client, MASQUE based tunnels built on tokio-quiche replace earlier WireGuard based tunnels with QUIC based tunnels. These systems run at Cloudflare edge scale and demonstrate that the integration can sustain millions of HTTP/3 requests per second in production.

Cloudflare positions tokio-quiche as a foundation rather than a complete HTTP/3 framework. The library exposes low level protocol capabilities and example client and server event loops, and leaves room for higher level projects to implement opinionated HTTP servers, DNS over QUIC clients, MASQUE based VPNs and other QUIC applications on top. By releasing the crate, Cloudflare aims to lower the barrier for Rust teams to adopt QUIC, HTTP/3 and MASQUE, and to align external integrations with the same transport stack used in its edge services.

Key Takeaways

tokio-quiche = quiche + Tokio: tokio-quiche is an async Rust library that integrates Cloudflare’s sans-io QUIC and HTTP/3 implementation, quiche, with the Tokio runtime, so developers do not need to hand write UDP and event loop plumbing.

Actor based architecture for QUIC connections: The library uses an actor model on Tokio, with an InboundPacketRouter that routes UDP datagrams by connection ID and an IoWorker that drives a single quiche Connection per task, keeping transport state isolated and composable.

ApplicationOverQuic abstraction: Protocol logic is separated through the ApplicationOverQuic trait, which abstracts over quiche and I O details so different QUIC based protocols such as HTTP/3, DNS over QUIC or custom protocols can be implemented on top of the same transport core.

HTTP/3 via H3Driver, ServerH3Driver and ClientH3Driver: tokio-quiche ships H3Driver plus ServerH3Driver and ClientH3Driver variants that bridge quiche’s HTTP/3 module to async Rust code, exposing HTTP/3 streams and bodies in a way that fits typical Tokio based services.

Check out the Technical details. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post How Cloudflare’s tokio-quiche Makes QUIC and HTTP/3 a First Class Citizen in Rust Backends appeared first on MarkTechPost.

In this tutorial, we demonstrate how we simulate a privacy-preserving fraud detection system using Federated Learning without relying on heavyweight frameworks or complex infrastructure. We build a clean, CPU-friendly setup that mimics ten independent banks, each training a local fraud-detection model on its own highly imbalanced transaction data. We coordinate these local updates through a simple FedAvg aggregation loop, allowing us to improve a global model while ensuring that no raw transaction data ever leaves a client. Alongside this, we integrate OpenAI to support post-training analysis and risk-oriented reporting, demonstrating how federated learning outputs can be translated into decision-ready insights. Check out the Full Codes here.

Copy CodeCopiedUse a different Browser!pip -q install torch scikit-learn numpy openai

import time, random, json, os, getpass
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import roc_auc_score, average_precision_score, accuracy_score
from openai import OpenAI

SEED = 7
random.seed(SEED); np.random.seed(SEED); torch.manual_seed(SEED)

DEVICE = torch.device(“cpu”)
print(“Device:”, DEVICE)

We set up the execution environment and import all required libraries for data generation, modeling, evaluation, and reporting. We also fix random seeds and the device configuration to ensure our federated simulation remains deterministic and reproducible on CPU. Check out the Full Codes here.

Copy CodeCopiedUse a different BrowserX, y = make_classification(
n_samples=60000,
n_features=30,
n_informative=18,
n_redundant=8,
weights=[0.985, 0.015],
class_sep=1.5,
flip_y=0.01,
random_state=SEED
)

X = X.astype(np.float32)
y = y.astype(np.int64)

X_train_full, X_test, y_train_full, y_test = train_test_split(
X, y, test_size=0.2, stratify=y, random_state=SEED
)

server_scaler = StandardScaler()
X_train_full_s = server_scaler.fit_transform(X_train_full).astype(np.float32)
X_test_s = server_scaler.transform(X_test).astype(np.float32)

test_loader = DataLoader(
TensorDataset(torch.from_numpy(X_test_s), torch.from_numpy(y_test)),
batch_size=1024,
shuffle=False
)

We generate a highly imbalanced, credit-card-like fraud dataset & split it into training & test sets. We standardize the server-side data and prepare a global test loader that allows us to consistently evaluate the aggregated model after each federated round. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserdef dirichlet_partition(y, n_clients=10, alpha=0.35):
classes = np.unique(y)
idx_by_class = [np.where(y == c)[0] for c in classes]
client_idxs = [[] for _ in range(n_clients)]
for idxs in idx_by_class:
np.random.shuffle(idxs)
props = np.random.dirichlet(alpha * np.ones(n_clients))
cuts = (np.cumsum(props) * len(idxs)).astype(int)
prev = 0
for cid, cut in enumerate(cuts):
client_idxs[cid].extend(idxs[prev:cut].tolist())
prev = cut
return [np.array(ci, dtype=np.int64) for ci in client_idxs]

NUM_CLIENTS = 10
client_idxs = dirichlet_partition(y_train_full, NUM_CLIENTS, 0.35)

def make_client_split(X, y, idxs):
Xi, yi = X[idxs], y[idxs]
if len(np.unique(yi)) < 2:
other = np.where(y == (1 – yi[0]))[0]
add = np.random.choice(other, size=min(10, len(other)), replace=False)
Xi = np.concatenate([Xi, X[add]])
yi = np.concatenate([yi, y[add]])
return train_test_split(Xi, yi, test_size=0.15, stratify=yi, random_state=SEED)

client_data = [make_client_split(X_train_full, y_train_full, client_idxs[c]) for c in range(NUM_CLIENTS)]

def make_client_loaders(Xtr, ytr, Xva, yva):
sc = StandardScaler()
Xtr_s = sc.fit_transform(Xtr).astype(np.float32)
Xva_s = sc.transform(Xva).astype(np.float32)
tr = DataLoader(TensorDataset(torch.from_numpy(Xtr_s), torch.from_numpy(ytr)), batch_size=512, shuffle=True)
va = DataLoader(TensorDataset(torch.from_numpy(Xva_s), torch.from_numpy(yva)), batch_size=512)
return tr, va

client_loaders = [make_client_loaders(*cd) for cd in client_data]

We simulate realistic non-IID behavior by partitioning the training data across ten clients using a Dirichlet distribution. We then create independent client-level train and validation loaders, ensuring that each simulated bank operates on its own locally scaled data. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserclass FraudNet(nn.Module):
def __init__(self, in_dim):
super().__init__()
self.net = nn.Sequential(
nn.Linear(in_dim, 64),
nn.ReLU(),
nn.Dropout(0.1),
nn.Linear(64, 32),
nn.ReLU(),
nn.Dropout(0.1),
nn.Linear(32, 1)
)
def forward(self, x):
return self.net(x).squeeze(-1)

def get_weights(model):
return [p.detach().cpu().numpy() for p in model.state_dict().values()]

def set_weights(model, weights):
keys = list(model.state_dict().keys())
model.load_state_dict({k: torch.tensor(w) for k, w in zip(keys, weights)}, strict=True)

@torch.no_grad()
def evaluate(model, loader):
model.eval()
bce = nn.BCEWithLogitsLoss()
ys, ps, losses = [], [], []
for xb, yb in loader:
logits = model(xb)
losses.append(bce(logits, yb.float()).item())
ys.append(yb.numpy())
ps.append(torch.sigmoid(logits).numpy())
y_true = np.concatenate(ys)
y_prob = np.concatenate(ps)
return {
“loss”: float(np.mean(losses)),
“auc”: roc_auc_score(y_true, y_prob),
“ap”: average_precision_score(y_true, y_prob),
“acc”: accuracy_score(y_true, (y_prob >= 0.5).astype(int))
}

def train_local(model, loader, lr):
opt = torch.optim.Adam(model.parameters(), lr=lr)
bce = nn.BCEWithLogitsLoss()
model.train()
for xb, yb in loader:
opt.zero_grad()
loss = bce(model(xb), yb.float())
loss.backward()
opt.step()

We define the neural network used for fraud detection along with utility functions for training, evaluation, and weight exchange. We implement lightweight local optimization and metric computation to keep client-side updates efficient and easy to reason about. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserdef fedavg(weights, sizes):
total = sum(sizes)
return [
sum(w[i] * (s / total) for w, s in zip(weights, sizes))
for i in range(len(weights[0]))
]

ROUNDS = 10
LR = 5e-4

global_model = FraudNet(X_train_full.shape[1])
global_weights = get_weights(global_model)

for r in range(1, ROUNDS + 1):
client_weights, client_sizes = [], []
for cid in range(NUM_CLIENTS):
local = FraudNet(X_train_full.shape[1])
set_weights(local, global_weights)
train_local(local, client_loaders[cid][0], LR)
client_weights.append(get_weights(local))
client_sizes.append(len(client_loaders[cid][0].dataset))
global_weights = fedavg(client_weights, client_sizes)
set_weights(global_model, global_weights)
metrics = evaluate(global_model, test_loader)
print(f”Round {r}: {metrics}”)

We orchestrate the federated learning process by iteratively training local client models and aggregating their parameters using FedAvg. We evaluate the global model after each round to monitor convergence and understand how collective learning improves fraud detection performance. Check out the Full Codes here.

Copy CodeCopiedUse a different BrowserOPENAI_API_KEY = getpass.getpass(“Enter OPENAI_API_KEY (input hidden): “).strip()

if OPENAI_API_KEY:
os.environ[“OPENAI_API_KEY”] = OPENAI_API_KEY
client = OpenAI()

summary = {
“rounds”: ROUNDS,
“num_clients”: NUM_CLIENTS,
“final_metrics”: metrics,
“client_sizes”: [len(client_loaders[c][0].dataset) for c in range(NUM_CLIENTS)],
“client_fraud_rates”: [float(client_data[c][1].mean()) for c in range(NUM_CLIENTS)]
}

prompt = (
“Write a concise internal fraud-risk report.n”
“Include executive summary, metric interpretation, risks, and next steps.nn”
+ json.dumps(summary, indent=2)
)

resp = client.responses.create(model=”gpt-5.2″, input=prompt)
print(resp.output_text)

We transform the technical results into a concise analytical report using an external language model. We securely accept the API key via keyboard input and generate decision-oriented insights that summarize performance, risks, and recommended next steps.

In conclusion, we showed how to implement federated learning from first principles in a Colab notebook while remaining stable, interpretable, and realistic. We observed how extreme data heterogeneity across clients influences convergence and why careful aggregation and evaluation are critical in fraud-detection settings. We also extended the workflow by generating an automated risk-team report, demonstrating how analytical results can be translated into decision-ready insights. At last, we presented a practical blueprint for experimenting with federated fraud models that emphasizes privacy awareness, simplicity, and real-world relevance.

Check out the Full Codes here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post A Coding Implementation of an OpenAI-Assisted Privacy-Preserving Federated Fraud Detection System from Scratch Using Lightweight PyTorch Simulations appeared first on MarkTechPost.

LLMRouter is an open source routing library from the U Lab at the University of Illinois Urbana Champaign that treats model selection as a first class system problem. It sits between applications and a pool of LLMs and chooses a model for each query based on task complexity, quality targets, and cost, all exposed through a unified Python API and CLI. The project ships with more than 16 routing models, a data generation pipeline over 11 benchmarks, and a plugin system for custom routers.

Router families and supported models

LLMRouter organizes routing algorithms into four families, Single-Round Routers, Multi-Round Routers, Personalized Routers, and Agentic Routers. Single round routers include knnrouter, svmrouter, mlprouter, mfrouter, elorouter, routerdc, automix, hybrid_llm, graphrouter, causallm_router, and the baselines smallest_llm and largest_llm. These models implement strategies such as k nearest neighbors, support vector machines, multilayer perceptrons, matrix factorization, Elo rating, dual contrastive learning, automatic model mixing, and graph based routing.

Multi round routing is exposed through router_r1, a pre trained instance of Router R1 integrated into LLMRouter. Router R1 formulates multi LLM routing and aggregation as a sequential decision process where the router itself is an LLM that alternates between internal reasoning steps and external model calls. It is trained with reinforcement learning using a rule based reward that balances format, outcome, and cost. In LLMRouter, router_r1 is available as an extra installation target with pinned dependencies tested on vllm==0.6.3 and torch==2.4.0.

Personalized routing is handled by gmtrouter, described as a graph based personalized router with user preference learning. GMTRouter represents multi turn user LLM interactions as a heterogeneous graph over users, queries, responses, and models. It runs a message passing architecture over this graph to infer user specific routing preferences from few shot interaction data, and experiments show accuracy and AUC gains over non personalized baselines.

Agentic routers in LLMRouter extend routing to multi step reasoning workflows. knnmultiroundrouter uses k nearest neighbor reasoning over multi turn traces and is intended for complex tasks. llmmultiroundrouter exposes an LLM based agentic router that performs multi step routing without its own training loop. These agentic routers share the same configuration and data formats as the other router families and can be swapped through a single CLI flag.

Data generation pipeline for routing datasets

LLMRouter ships with a full data generation pipeline that turns standard benchmarks and LLM outputs into routing datasets. The pipeline supports 11 benchmarks, Natural QA, Trivia QA, MMLU, GPQA, MBPP, HumanEval, GSM8K, CommonsenseQA, MATH, OpenBookQA, and ARC Challenge. It runs in three explicit stages. First, data_generation.py extracts queries and ground truth labels and creates train and test JSONL splits. Second, generate_llm_embeddings.py builds embeddings for candidate LLMs from metadata. Third, api_calling_evaluation.py calls LLM APIs, evaluates responses, and fuses scores with embeddings into routing records. (GitHub)

The pipeline outputs query files, LLM embedding JSON, query embedding tensors, and routing data JSONL files. A routing entry includes fields such as task_name, query, ground_truth, metric, model_name, response, performance, embedding_id, and token_num. Configuration is handled entirely through YAML, so engineers point the scripts to new datasets and candidate model lists without modifying code.

Chat interface and plugin system

For interactive use, llmrouter chat launches a Gradio based chat frontend over any router and configuration. The server can bind to a custom host and port and can expose a public sharing link. Query modes control how routing sees context. current_only uses only the latest user message, full_context concatenates the dialogue history, and retrieval augments the query with the top k similar historical queries. The UI visualizes model choices in real time and is driven by the same router configuration used for batch inference.

LLMRouter also provides a plugin system for custom routers. New routers live under custom_routers, subclass MetaRouter, and implement route_single and route_batch. Configuration files under that directory define data paths, hyperparameters, and optional default API endpoints. Plugin discovery scans the project custom_routers folder, a ~/.llmrouter/plugins directory, and any extra paths in the LLMROUTER_PLUGINS environment variable. Example custom routers include randomrouter, which selects a model at random, and thresholdrouter, which is a trainable router that estimates query difficulty.

Key Takeaways

Routing as a first class abstraction: LLMRouter is an open source routing layer from UIUC that sits between applications and heterogeneous LLM pools and centralizes model selection as a cost and quality aware prediction task rather than ad hoc scripts.

Four router families covering 16 plus algorithms: The library standardizes more than 16 routers into four families, single round, multi round, personalized, and agentic, including knnrouter, graphrouter, routerdc, router_r1, and gmtrouter, all exposed through a unified config and CLI.

Multi round RL routing via Router R1: router_r1 integrates the Router R1 framework, where an LLM router interleaves internal “think” steps with external “route” calls and is trained with a rule based reward that combines format, outcome, and cost to optimize performance cost trade offs.

Graph based personalization with GMTRouter: gmtrouter models users, queries, responses and LLMs as nodes in a heterogeneous graph and uses message passing to learn user specific routing preferences from few shot histories, achieving up to around 21% accuracy gains and substantial AUC improvements over strong baselines.

End to end pipeline and extensibility: LLMRouter provides a benchmark driven data pipeline, CLI for training and inference, a Gradio chat UI, centralized API key handling, and a plugin system based on MetaRouter that allows teams to register custom routers while reusing the same routing datasets and infrastructure.

Check out the GitHub Repo and Technical details. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post Meet LLMRouter: An Intelligent Routing System designed to Optimize LLM Inference by Dynamically Selecting the most Suitable Model for Each Query appeared first on MarkTechPost.

In this tutorial, we demonstrate how to design a contract-first agentic decision system using PydanticAI, treating structured schemas as non-negotiable governance contracts rather than optional output formats. We show how we define a strict decision model that encodes policy compliance, risk assessment, confidence calibration, and actionable next steps directly into the agent’s output schema. By combining Pydantic validators with PydanticAI’s retry and self-correction mechanisms, we ensure that the agent cannot produce logically inconsistent or non-compliant decisions. Throughout the workflow, we focus on building an enterprise-grade decision agent that reasons under constraints, making it suitable for real-world risk, compliance, and governance scenarios rather than toy prompt-based demos. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser!pip -q install -U pydantic-ai pydantic openai nest_asyncio

import os
import time
import asyncio
import getpass
from dataclasses import dataclass
from typing import List, Literal

import nest_asyncio
nest_asyncio.apply()

from pydantic import BaseModel, Field, field_validator
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

OPENAI_API_KEY = os.getenv(“OPENAI_API_KEY”)
if not OPENAI_API_KEY:
try:
from google.colab import userdata
OPENAI_API_KEY = userdata.get(“OPENAI_API_KEY”)
except Exception:
OPENAI_API_KEY = None
if not OPENAI_API_KEY:
OPENAI_API_KEY = getpass.getpass(“Enter OPENAI_API_KEY: “).strip()

We set up the execution environment by installing the required libraries and configuring asynchronous execution for Google Colab. We securely load the OpenAI API key and ensure the runtime is ready to handle async agent calls. This establishes a stable foundation for running the contract-first agent without environment-related issues. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserclass RiskItem(BaseModel):
risk: str = Field(…, min_length=8)
severity: Literal[“low”, “medium”, “high”]
mitigation: str = Field(…, min_length=12)

class DecisionOutput(BaseModel):
decision: Literal[“approve”, “approve_with_conditions”, “reject”]
confidence: float = Field(…, ge=0.0, le=1.0)
rationale: str = Field(…, min_length=80)
identified_risks: List[RiskItem] = Field(…, min_length=2)
compliance_passed: bool
conditions: List[str] = Field(default_factory=list)
next_steps: List[str] = Field(…, min_length=3)
timestamp_unix: int = Field(default_factory=lambda: int(time.time()))

@field_validator(“confidence”)
@classmethod
def confidence_vs_risk(cls, v, info):
risks = info.data.get(“identified_risks”) or []
if any(r.severity == “high” for r in risks) and v > 0.70:
raise ValueError(“confidence too high given high-severity risks”)
return v

@field_validator(“decision”)
@classmethod
def reject_if_non_compliant(cls, v, info):
if info.data.get(“compliance_passed”) is False and v != “reject”:
raise ValueError(“non-compliant decisions must be reject”)
return v

@field_validator(“conditions”)
@classmethod
def conditions_required_for_conditional_approval(cls, v, info):
d = info.data.get(“decision”)
if d == “approve_with_conditions” and (not v or len(v) < 2):
raise ValueError(“approve_with_conditions requires at least 2 conditions”)
if d == “approve” and v:
raise ValueError(“approve must not include conditions”)
return v

We define the core decision contract using strict Pydantic models that precisely describe a valid decision. We encode logical constraints such as confidence–risk alignment, compliance-driven rejection, and conditional approvals directly into the schema. This ensures that any agent output must satisfy business logic, not just syntactic structure. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser@dataclass
class DecisionContext:
company_policy: str
risk_threshold: float = 0.6

model = OpenAIChatModel(
“gpt-5″,
provider=OpenAIProvider(api_key=OPENAI_API_KEY),
)

agent = Agent(
model=model,
deps_type=DecisionContext,
output_type=DecisionOutput,
system_prompt=”””
You are a corporate decision analysis agent.
You must evaluate risk, compliance, and uncertainty.
All outputs must strictly satisfy the DecisionOutput schema.
“””
)

We inject enterprise context through a typed dependency object and initialize the OpenAI-backed PydanticAI agent. We configure the agent to produce only structured decision outputs that conform to the predefined contract. This step formalizes the separation between business context and model reasoning. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser@agent.output_validator
def ensure_risk_quality(result: DecisionOutput) -> DecisionOutput:
if len(result.identified_risks) < 2:
raise ValueError(“minimum two risks required”)
if not any(r.severity in (“medium”, “high”) for r in result.identified_risks):
raise ValueError(“at least one medium or high risk required”)
return result

@agent.output_validator
def enforce_policy_controls(result: DecisionOutput) -> DecisionOutput:
policy = CURRENT_DEPS.company_policy.lower()
text = (
result.rationale
+ ” “.join(result.next_steps)
+ ” “.join(result.conditions)
).lower()
if result.compliance_passed:
if not any(k in text for k in [“encryption”, “audit”, “logging”, “access control”, “key management”]):
raise ValueError(“missing concrete security controls”)
return result

We add output validators that act as governance checkpoints after the model generates a response. We force the agent to identify meaningful risks and to explicitly reference concrete security controls when claiming compliance. If these constraints are violated, we trigger automatic retries to enforce self-correction. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserasync def run_decision():
global CURRENT_DEPS
CURRENT_DEPS = DecisionContext(
company_policy=(
“No deployment of systems handling personal data or transaction metadata ”
“without encryption, audit logging, and least-privilege access control.”
)
)

prompt = “””
Decision request:
Deploy an AI-powered customer analytics dashboard using a third-party cloud vendor.
The system processes user behavior and transaction metadata.
Audit logging is not implemented and customer-managed keys are uncertain.
“””

result = await agent.run(prompt, deps=CURRENT_DEPS)
return result.output

decision = asyncio.run(run_decision())

from pprint import pprint
pprint(decision.model_dump())

We run the agent on a realistic decision request and capture the validated structured output. We demonstrate how the agent evaluates risk, policy compliance, and confidence before producing a final decision. This completes the end-to-end contract-first decision workflow in a production-style setup.

In conclusion, we demonstrate how to move from free-form LLM outputs to governed, reliable decision systems using PydanticAI. We show that by enforcing hard contracts at the schema level, we can automatically align decisions with policy requirements, risk severity, and confidence realism without manual prompt tuning. This approach allows us to build agents that fail safely, self-correct when constraints are violated, and produce auditable, structured outputs that downstream systems can trust. Ultimately, we demonstrate that contract-first agent design enables us to deploy agentic AI as a dependable decision layer within production and enterprise environments.

Check out the FULL CODES here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post How to Build Contract-First Agentic Decision Systems with PydanticAI for Risk-Aware, Policy-Compliant Enterprise AI appeared first on MarkTechPost.

Businesses face a growing challenge: customers need answers fast, but support teams are overwhelmed. Support documentation like product manuals and knowledge base articles typically require users to search through hundreds of pages, and support agents often run 20–30 customer queries per day to locate specific information.
This post demonstrates how to solve this challenge by building an AI-powered website assistant using Amazon Bedrock and Amazon Bedrock Knowledge Bases. This solution is designed to benefit both internal teams and external customers, and can offer the following benefits:

Instant, relevant answers for customers, alleviating the need to search through documentation
A powerful knowledge retrieval system for support agents, reducing resolution time
Round-the-clock automated support

Solution overview
The solution uses Retrieval-Augmented Generation (RAG) to retrieve relevant information from a knowledge base and return it to the user based on their access. It consists of the following key components:

Amazon Bedrock Knowledge Bases – Content from the company’s website is crawled and stored in the knowledge base. Documents from an Amazon Simple Storage Service (Amazon S3) bucket, including manuals and troubleshooting guides, are also indexed and stored in the knowledge base. With Amazon Bedrock Knowledge Bases, you can configure multiple data sources and use the filter configurations to differentiate between internal and external information. This helps protect internal data through advanced security controls.
Amazon Bedrock managed LLMs – A large language model (LLM) from Amazon Bedrock generates AI-powered responses to user questions.
Scalable serverless architecture – The solution uses Amazon Elastic Container Service (Amazon ECS) to host the UI, and an AWS Lambda function to handle the user requests.
Automated CI/CD deployment – The solution uses the AWS Cloud Development Kit (AWS CDK) to handle continuous integration and delivery (CI/CD) deployment.

The following diagram illustrates the architecture of this solution.

The workflow consists of the following steps:

Amazon Bedrock Knowledge Bases processes documents uploaded to Amazon S3 by chunking them and generating embeddings. Additionally, the Amazon Bedrock web crawler accesses selected websites to extract and ingest their contents.
The web application runs as an ECS application. Internal and external users use browsers to access the application through Elastic Load Balancing (ELB). Users log in to the application using their login credentials registered in an Amazon Cognito user pool.
When a user submits a question, the application invokes a Lambda function, which uses the Amazon Bedrock APIs to retrieve the relevant information from the knowledge base. It also supplies the relevant data source IDs to Amazon Bedrock based on user type (external or internal) so the knowledge base retrieves only the information available to that user type.
The Lambda function then invokes the Amazon Nova Lite LLM to generate responses. The LLM augments the information from the knowledge base to generate a response to the user query, which is returned from the Lambda function and displayed to the user.

In the following sections, we demonstrate how to crawl and configure the external website as a knowledge base, and also upload internal documentation.
Prerequisites
You must have the following in place to deploy the solution in this post:

An AWS account.
Model access in Amazon Bedrock for Amazon Titan and Amazon Nova Lite. Use the same AWS Region for model access as the Region where you deploy the solution.
An S3 bucket in the same Region to store internal data.

Create knowledge base and ingest website data
The first step is to build a knowledge base to ingest data from a website and operational documents from an S3 bucket. Complete the following steps to create your knowledge base:

On the Amazon Bedrock console, choose Knowledge Bases under Builder tools in the navigation pane.
On the Create dropdown menu, choose Knowledge Base with vector store.

For Knowledge Base name, enter a name.
For Choose a data source, select Web Crawler.
Choose Next.

For Data source name, enter a name for your data source.
For Source URLs, enter the target website HTML page to crawl. For example, we use https://docs.aws.amazon.com/AmazonS3/latest/userguide/GetStartedWithS3.html.
For Website domain range, select Default as the crawling scope. You can also configure it to host only domains or subdomains if you want to restrict the crawling to a specific domain or subdomain.
For URL regex filter, you can configure the URL patterns to include or exclude specific URLs. For this example, we leave this setting blank.

For Chunking strategy, you can configure the content parsing options to customize the data chunking strategy. For this example, we leave it as Default chunking.
Choose Next.

Choose the Amazon Titan Text Embeddings V2 model, then choose Apply.

For Vector store type, select Amazon OpenSearch Serverless, then choose Next.

Review the configurations and choose Create Knowledge Base.

You have now created a knowledge base with the data source configured as the website link you provided.

On the knowledge base details page, select your new data source and choose Sync to crawl the website and ingest the data.

Configure Amazon S3 data source
Complete the following steps to configure documents from your S3 bucket as an internal data source:

On the knowledge base details page, choose Add in the Data source section.

Specify the data source as Amazon S3.
Choose your S3 bucket.
Leave the parsing strategy as the default setting.
Choose Next.
Review the configurations and choose Add data source.
In the Data source section of the knowledge base details page, select your new data source and choose Sync to index the data from the documents in the S3 bucket.

Upload internal document
For this example, we upload a document in the new S3 bucket data source. The following screenshot shows an example of our document.

Complete the following steps to upload the document:

On the Amazon S3 console, choose Buckets in the navigation pane.
Select the bucket you created and choose Upload to upload the document.

On the Amazon Bedrock console, go to the knowledge base you created.
Choose the internal data source you created and choose Sync to sync the uploaded document with the vector store.

Note the knowledge base ID and the data source IDs for the external and internal data sources. You use this information in the next step when deploying the solution infrastructure.
Deploy solution infrastructure
To deploy the solution infrastructure using the AWS CDK, complete the following steps:

Download the code from code repository.
Go to the iac directory inside the downloaded project:

cd ./customer-support-ai/iac

Open the parameters.json file and update the knowledge base and data source IDs with the values captured in the previous section:

“external_source_id”: “Set this to value from Amazon Bedrock Knowledge Base datasource”,
“internal_source_id”: “Set this to value from Amazon Bedrock Knowledge Base datasource”,
“knowledge_base_id”: “Set this to value from Amazon Bedrock Knowledge Base”,

Follow the deployment instructions defined in the customer-support-ai/README.md file to set up the solution infrastructure.

When the deployment is complete, you can find the Application Load Balancer (ALB) URL and demo user details in the script execution output.

You can also open the Amazon EC2 console and choose Load Balancers in the navigation pane to view the ALB.

On the ALB details page, copy the DNS name. You can use it to access the UI to try out the solution.

Submit questions
Let’s explore an example of Amazon S3 service support. This solution supports different classes of users to help resolve their queries while using Amazon Bedrock Knowledge Bases to manage specific data sources (such as website content, documentation, and support tickets) with built-in filtering controls that separate internal operational documents from publicly accessible information. For example, internal users can access both company-specific operational guides and public documentation, whereas external users are limited to publicly available content only.
Open the DNS URL in the browser. Enter the external user credentials and choose Login.

After you’re successfully authenticated, you will be redirected to the home page.

Choose Support AI Assistant in the navigation pane to ask questions related to Amazon S3. The assistant can provide relevant responses based on the information available in the Getting started with Amazon S3 guide. However, if an external user asks a question that is related to information available only for internal users, the AI assistant will not provide the internal information to user and will respond only with information available for external users.

Log out and log in again as an internal user, and ask the same queries. The internal user can access the relevant information available in the internal documents.

Clean up
If you decide to stop using this solution, complete the following steps to remove its associated resources:

Go to the iac directory inside the project code and run the following command from terminal:

To run a cleanup script, use the following command:

cd iac
./cleanup.sh

To perform this operation manually, use the following command:

cd iac
cdk destroy –all

On the Amazon Bedrock console, choose Knowledge Bases under Builder tools in the navigation pane.
Choose the knowledge base you created, then choose Delete.
Enter delete and choose Delete to confirm.
On the OpenSearch Service console, choose Collections under Serverless in the navigation pane.
Choose the collection created during infrastructure provisioning, then choose Delete.
Enter confirm and choose Delete to confirm.

Conclusion
This post demonstrated how to create an AI-powered website assistant to retrieve information quickly by constructing a knowledge base through web crawling and uploading documents. You can use the same approach to develop other generative AI prototypes and applications.
If you’re interested in the fundamentals of generative AI and how to work with FMs, including advanced prompting techniques, check out the hands-on course Generative AI with LLMs. This on-demand, 3-week course is for data scientists and engineers who want to learn how to build generative AI applications with LLMs. It’s the good foundation to start building with Amazon Bedrock. Sign up to learn more about Amazon Bedrock.

About the authors
Shashank Jain is a Cloud Application Architect at Amazon Web Services (AWS), specializing in generative AI solutions, cloud-native application architecture, and sustainability. He works with customers to design and implement secure, scalable AI-powered applications using serverless technologies, modern DevSecOps practices, Infrastructure as Code, and event-driven architectures that deliver measurable business value.
Jeff Li is a Senior Cloud Application Architect with the Professional Services team at AWS. He is passionate about diving deep with customers to create solutions and modernize applications that support business innovations. In his spare time, he enjoys playing tennis, listening to music, and reading.
Ranjith Kurumbaru Kandiyil is a Data and AI/ML Architect at Amazon Web Services (AWS) based in Toronto. He specializes in collaborating with customers to architect and implement cutting-edge AI/ML solutions. His current focus lies in leveraging state-of-the-art artificial intelligence technologies to solve complex business challenges.

In this tutorial, we build an end-to-end, production-style agentic workflow using GraphBit that demonstrates how graph-structured execution, tool calling, and optional LLM-driven agents can coexist in a single system. We start by initializing and inspecting the GraphBit runtime, then define a realistic customer-support ticket domain with typed data structures and deterministic, offline-executable tools. We show how these tools can be composed into a reliable, rule-based pipeline for classification, routing, and response drafting, and then elevate that same logic into a validated GraphBit workflow in which agent nodes orchestrate tool usage via a directed graph. Throughout the tutorial, we keep the system running in offline mode while enabling seamless promotion to online execution by simply providing an LLM configuration, illustrating how GraphBit supports the gradual adoption of agentic intelligence without sacrificing reproducibility or operational control. Check out the Full Codes here.

Copy CodeCopiedUse a different Browser!pip -q install graphbit rich pydantic numpy

import os
import time
import json
import random
from dataclasses import dataclass
from typing import Dict, Any, List, Optional
import numpy as np
from rich import print as rprint
from rich.panel import Panel
from rich.table import Table

We begin by installing all required dependencies and importing the core Python, numerical, and visualization libraries needed for the tutorial. We set up the runtime environment so the notebook remains self-contained and reproducible on Google Colab. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserfrom graphbit import init, shutdown, configure_runtime, get_system_info, health_check, version
from graphbit import Workflow, Node, Executor, LlmConfig
from graphbit import tool, ToolExecutor, ExecutorConfig
from graphbit import get_tool_registry, clear_tools

configure_runtime(worker_threads=4, max_blocking_threads=8, thread_stack_size_mb=2)
init(log_level=”warn”, enable_tracing=False, debug=False)

info = get_system_info()
health = health_check()

sys_table = Table(title=”System Info / Health”)
sys_table.add_column(“Key”, style=”bold”)
sys_table.add_column(“Value”)
for k in [“version”, “python_binding_version”, “cpu_count”, “runtime_worker_threads”, “runtime_initialized”, “build_target”, “build_profile”]:
sys_table.add_row(k, str(info.get(k)))
sys_table.add_row(“graphbit_version()”, str(version()))
sys_table.add_row(“overall_healthy”, str(health.get(“overall_healthy”)))
rprint(sys_table)

We initialize the GraphBit runtime and explicitly configure its execution parameters to control threading and resource usage. We then query system metadata and perform a health check to verify that the runtime is correctly initialized. Check out the Full Codes here.

Copy CodeCopiedUse a different Browser@dataclass
class Ticket:
ticket_id: str
user_id: str
text: str
created_at: float

def make_tickets(n: int = 10) -> List[Ticket]:
seeds = [
“My card payment failed twice, what should I do?”,
“I want to cancel my subscription immediately.”,
“Your app crashes when I open the dashboard.”,
“Please update my email address on the account.”,
“Refund not received after 7 days.”,
“My delivery is delayed and tracking is stuck.”,
“I suspect fraudulent activity on my account.”,
“How can I change my billing cycle date?”,
“The website is very slow and times out.”,
“I forgot my password and cannot login.”,
“Chargeback process details please.”,
“Need invoice for last month’s payment.”
]
random.shuffle(seeds)
out = []
for i in range(n):
out.append(
Ticket(
ticket_id=f”T-{1000+i}”,
user_id=f”U-{random.randint(100,999)}”,
text=seeds[i % len(seeds)],
created_at=time.time() – random.randint(0, 7 * 24 * 3600),
)
)
return out

tickets = make_tickets(10)
rprint(Panel.fit(“n”.join([f”- {t.ticket_id}: {t.text}” for t in tickets]), title=”Sample Tickets”))

We define a strongly typed data model for support tickets and generate a synthetic dataset that simulates realistic customer issues. We construct tickets with timestamps and identifiers to mirror production inputs. This dataset serves as the shared input across both offline and agent-driven pipelines. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserclear_tools()

@tool(_description=”Classify a support ticket into a coarse category.”)
def classify_ticket(text: str) -> Dict[str, Any]:
t = text.lower()
if “fraud” in t or “fraudulent” in t:
return {“category”: “fraud”, “priority”: “p0”}
if “cancel” in t:
return {“category”: “cancellation”, “priority”: “p1”}
if “refund” in t or “chargeback” in t:
return {“category”: “refunds”, “priority”: “p1”}
if “password” in t or “login” in t:
return {“category”: “account_access”, “priority”: “p2”}
if “crash” in t or “slow” in t or “timeout” in t:
return {“category”: “bug”, “priority”: “p2”}
if “payment” in t or “billing” in t or “invoice” in t:
return {“category”: “billing”, “priority”: “p2”}
if “delivery” in t or “tracking” in t:
return {“category”: “delivery”, “priority”: “p3”}
return {“category”: “general”, “priority”: “p3″}

@tool(_description=”Route a ticket to a queue (returns queue id and SLA hours).”)
def route_ticket(category: str, priority: str) -> Dict[str, Any]:
queue_map = {
“fraud”: (“risk_ops”, 2),
“cancellation”: (“retention”, 8),
“refunds”: (“payments_ops”, 12),
“account_access”: (“identity”, 12),
“bug”: (“engineering_support”, 24),
“billing”: (“billing_support”, 24),
“delivery”: (“logistics_support”, 48),
“general”: (“support_general”, 48),
}
q, sla = queue_map.get(category, (“support_general”, 48))
if priority == “p0”:
sla = min(sla, 2)
elif priority == “p1”:
sla = min(sla, 8)
return {“queue”: q, “sla_hours”: sla}

@tool(_description=”Generate a playbook response based on category + priority.”)
def draft_response(category: str, priority: str, ticket_text: str) -> Dict[str, Any]:
templates = {
“fraud”: “We’ve temporarily secured your account. Please confirm last 3 transactions and reset credentials.”,
“cancellation”: “We can help cancel your subscription. Please confirm your plan and the effective date you want.”,
“refunds”: “We’re checking the refund status. Please share the order/payment reference and date.”,
“account_access”: “Let’s get you back in. Please use the password reset link; if blocked, we’ll verify identity.”,
“bug”: “Thanks for reporting. Please share device/browser + a screenshot; we’ll attempt reproduction.”,
“billing”: “We can help with billing. Please confirm the last 4 digits and the invoice period you need.”,
“delivery”: “We’re checking shipment status. Please share your tracking ID and delivery address PIN/ZIP.”,
“general”: “Thanks for reaching out.”
}
base = templates.get(category, templates[“general”])
tone = “urgent” if priority == “p0” else (“fast” if priority == “p1” else “standard”)
return {
“tone”: tone,
“message”: f”{base}nnContext we received: ‘{ticket_text}'”,
“next_steps”: [“request_missing_info”, “log_case”, “route_to_queue”]
}

registry = get_tool_registry()
tools_list = registry.list_tools() if hasattr(registry, “list_tools”) else []
rprint(Panel.fit(f”Registered tools: {tools_list}”, title=”Tool Registry”))

We register deterministic business tools for ticket classification, routing, and response drafting using GraphBit’s tool interface. We encode domain logic directly into these tools so they can be executed without any LLM dependency. This establishes a reliable, testable foundation for later agent orchestration. Check out the Full Codes here.

Copy CodeCopiedUse a different Browsertool_exec_cfg = ExecutorConfig(
max_execution_time_ms=10_000,
max_tool_calls=50,
continue_on_error=False,
store_results=True,
enable_logging=False
)
tool_executor = ToolExecutor(config=tool_exec_cfg) if “config” in ToolExecutor.__init__.__code__.co_varnames else ToolExecutor()

def offline_triage(ticket: Ticket) -> Dict[str, Any]:
c = classify_ticket(ticket.text)
rt = route_ticket(c[“category”], c[“priority”])
dr = draft_response(c[“category”], c[“priority”], ticket.text)
return {
“ticket_id”: ticket.ticket_id,
“user_id”: ticket.user_id,
“category”: c[“category”],
“priority”: c[“priority”],
“queue”: rt[“queue”],
“sla_hours”: rt[“sla_hours”],
“draft”: dr[“message”],
“tone”: dr[“tone”],
“steps”: [
(“classify_ticket”, c),
(“route_ticket”, rt),
(“draft_response”, dr),
]
}

offline_results = [offline_triage(t) for t in tickets]

res_table = Table(title=”Offline Pipeline Results”)
res_table.add_column(“Ticket”, style=”bold”)
res_table.add_column(“Category”)
res_table.add_column(“Priority”)
res_table.add_column(“Queue”)
res_table.add_column(“SLA (h)”)
for r in offline_results:
res_table.add_row(r[“ticket_id”], r[“category”], r[“priority”], r[“queue”], str(r[“sla_hours”]))
rprint(res_table)

prio_counts: Dict[str, int] = {}
sla_vals: List[int] = []
for r in offline_results:
prio_counts[r[“priority”]] = prio_counts.get(r[“priority”], 0) + 1
sla_vals.append(int(r[“sla_hours”]))

metrics = {
“offline_mode”: True,
“tickets”: len(offline_results),
“priority_distribution”: prio_counts,
“sla_mean”: float(np.mean(sla_vals)) if sla_vals else None,
“sla_p95″: float(np.percentile(sla_vals, 95)) if sla_vals else None,
}

rprint(Panel.fit(json.dumps(metrics, indent=2), title=”Offline Metrics”))

We compose the registered tools into an offline execution pipeline and apply it across all tickets to produce structured triage results. We aggregate outputs into tables and compute priority and SLA metrics to evaluate system behavior. It demonstrates how GraphBit-based logic can be validated deterministically before introducing agents. Check out the Full Codes here.

Copy CodeCopiedUse a different BrowserSYSTEM_POLICY = “You are a reliable support ops agent. Return STRICT JSON only.”

workflow = Workflow(“Ticket Triage Workflow (GraphBit)”)

summarizer = Node.agent(
name=”Summarizer”,
agent_id=”summarizer”,
system_prompt=SYSTEM_POLICY,
prompt=”Summarize this ticket in 1-2 lines. Return JSON: {“summary”:”…”}nTicket: {input}”,
temperature=0.2,
max_tokens=200
)

router_agent = Node.agent(
name=”RouterAgent”,
agent_id=”router”,
system_prompt=SYSTEM_POLICY,
prompt=(
“You MUST use tools.n”
“Call classify_ticket(text), route_ticket(category, priority), draft_response(category, priority, ticket_text).n”
“Return JSON with fields: category, priority, queue, sla_hours, message.n”
“Ticket: {input}”
),
tools=[classify_ticket, route_ticket, draft_response],
temperature=0.1,
max_tokens=700
)

formatter = Node.agent(
name=”FinalFormatter”,
agent_id=”final_formatter”,
system_prompt=SYSTEM_POLICY,
prompt=(
“Validate the JSON and output STRICT JSON only:n”
“{“ticket_id”:”…”,”category”:”…”,”priority”:”…”,”queue”:”…”,”sla_hours”:0,”customer_message”:”…”}n”
“Input: {input}”
),
temperature=0.0,
max_tokens=500
)

sid = workflow.add_node(summarizer)
rid = workflow.add_node(router_agent)
fid = workflow.add_node(formatter)

workflow.connect(sid, rid)
workflow.connect(rid, fid)
workflow.validate()

rprint(Panel.fit(“Workflow validated: Summarizer -> RouterAgent -> FinalFormatter”, title=”Workflow Graph”))

We construct a directed GraphBit workflow composed of multiple agent nodes with clearly defined responsibilities and strict JSON contracts. We connect these nodes into a validated execution graph that mirrors the earlier offline logic at an agent level. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserdef pick_llm_config() -> Optional[Any]:
if os.getenv(“OPENAI_API_KEY”):
return LlmConfig.openai(os.getenv(“OPENAI_API_KEY”), “gpt-4o-mini”)
if os.getenv(“ANTHROPIC_API_KEY”):
return LlmConfig.anthropic(os.getenv(“ANTHROPIC_API_KEY”), “claude-sonnet-4-20250514”)
if os.getenv(“DEEPSEEK_API_KEY”):
return LlmConfig.deepseek(os.getenv(“DEEPSEEK_API_KEY”), “deepseek-chat”)
if os.getenv(“MISTRALAI_API_KEY”):
return LlmConfig.mistralai(os.getenv(“MISTRALAI_API_KEY”), “mistral-large-latest”)
return None

def run_agent_flow_once(ticket_text: str) -> Dict[str, Any]:
llm_cfg = pick_llm_config()
if llm_cfg is None:
return {
“mode”: “offline”,
“note”: “Set OPENAI_API_KEY / ANTHROPIC_API_KEY / DEEPSEEK_API_KEY / MISTRALAI_API_KEY to enable execution.”,
“input”: ticket_text
}
executor = Executor(llm_cfg, lightweight_mode=True, timeout_seconds=90, debug=False) if “lightweight_mode” in Executor.__init__.__code__.co_varnames else Executor(llm_cfg)
if hasattr(executor, “configure”):
executor.configure(timeout_seconds=90, max_retries=2, enable_metrics=True, debug=False)
wf = Workflow(“Single Ticket Run”)
s = Node.agent(
name=”Summarizer”,
agent_id=”summarizer”,
system_prompt=SYSTEM_POLICY,
prompt=f”Summarize this ticket in 1-2 lines. Return JSON: {{“summary”:”…”}}nTicket: {ticket_text}”,
temperature=0.2,
max_tokens=200
)
r = Node.agent(
name=”RouterAgent”,
agent_id=”router”,
system_prompt=SYSTEM_POLICY,
prompt=(
“You MUST use tools.n”
“Call classify_ticket(text), route_ticket(category, priority), draft_response(category, priority, ticket_text).n”
“Return JSON with fields: category, priority, queue, sla_hours, message.n”
f”Ticket: {ticket_text}”
),
tools=[classify_ticket, route_ticket, draft_response],
temperature=0.1,
max_tokens=700
)
f = Node.agent(
name=”FinalFormatter”,
agent_id=”final_formatter”,
system_prompt=SYSTEM_POLICY,
prompt=(
“Validate the JSON and output STRICT JSON only:n”
“{“ticket_id”:”…”,”category”:”…”,”priority”:”…”,”queue”:”…”,”sla_hours”:0,”customer_message”:”…”}n”
“Input: {input}”
),
temperature=0.0,
max_tokens=500
)
sid = wf.add_node(s)
rid = wf.add_node(r)
fid = wf.add_node(f)
wf.connect(sid, rid)
wf.connect(rid, fid)
wf.validate()
t0 = time.time()
result = executor.execute(wf)
dt_ms = int((time.time() – t0) * 1000)
out = {“mode”: “online”, “execution_time_ms”: dt_ms, “success”: bool(result.is_success()) if hasattr(result, “is_success”) else None}
if hasattr(result, “get_all_variables”):
out[“variables”] = result.get_all_variables()
else:
out[“raw”] = str(result)[:3000]
return out

sample = tickets[0]
agent_run = run_agent_flow_once(sample.text)
rprint(Panel.fit(json.dumps(agent_run, indent=2)[:3000], title=”Agent Workflow Run”))

rprint(Panel.fit(“Done”, title=”Complete”))

We add optional LLM configuration and execution logic that enables the same workflow to run autonomously when a provider key is available. We execute the workflow on a single ticket and capture execution status and outputs. This final step illustrates how the system seamlessly transitions from offline determinism to fully agentic execution.

In conclusion, we implemented a complete GraphBit workflow spanning runtime configuration, tool registration, offline deterministic execution, metric aggregation, and optional agent-based orchestration with external LLM providers. We demonstrated how the same business logic can be executed both manually via tools and automatically via agent nodes connected in a validated graph, highlighting GraphBit’s strength as an execution substrate rather than just an LLM wrapper. We showed that complex agentic systems can be designed to fail gracefully, run without external dependencies, and still scale to fully autonomous workflows when LLMs are enabled.

Check out the Full Codes here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post How to Build Production-Grade Agentic Workflows with GraphBit Using Deterministic Tools, Validated Execution Graphs, and Optional LLM Orchestration appeared first on MarkTechPost.

Google has released FunctionGemma, a specialized version of the Gemma 3 270M model that is trained specifically for function calling and designed to run as an edge agent that maps natural language to executable API actions.

But, What is FunctionGemma?

FunctionGemma is a 270M parameter text only transformer based on Gemma 3 270M. It keeps the same architecture as Gemma 3 and is released as an open model under the Gemma license, but the training objective and chat format are dedicated to function calling rather than free form dialogue.

The model is intended to be fine tuned for specific function calling tasks. It is not positioned as a general chat assistant. The primary design goal is to translate user instructions and tool definitions into structured function calls, then optionally summarize tool responses for the user.

From an interface perspective, FunctionGemma is presented as a standard causal language model. Inputs and outputs are text sequences, with an input context of 32K tokens and an output budget of up to 32K tokens per request, shared with the input length.

Architecture and training data

The model uses the Gemma 3 transformer architecture and the same 270M parameter scale as Gemma 3 270M. The training and runtime stack reuse the research and infrastructure used for Gemini, including JAX and ML Pathways on large TPU clusters.

FunctionGemma uses Gemma’s 256K vocabulary, which is optimized for JSON structures and multilingual text. This improves token efficiency for function schemas and tool responses and reduces sequence length for edge deployments where latency and memory are tight.

The model is trained on 6T tokens, with a knowledge cutoff in August 2024. The dataset focuses on two main categories:

public tool and API definitions

tool use interactions that include prompts, function calls, function responses and natural language follow up messages that summarize outputs or request clarification

This training signal teaches both syntax, which function to call and how to format arguments, and intent, when to call a function and when to ask for more information.

Conversation format and control tokens

FunctionGemma does not use a free form chat format. It expects a strict conversation template that separates roles and tool related regions. Conversation turns are wrapped with <start_of_turn>role … <end_of_turn> where roles are typically developer, user or model.

Within those turns, FunctionGemma relies on a fixed set of control token pairs

<start_function_declaration> and <end_function_declaration> for tool definitions

<start_function_call> and <end_function_call> for the model’s tool calls

<start_function_response> and <end_function_response> for serialized tool outputs

These markers let the model distinguish natural language text from function schemas and from execution results. The Hugging Face apply_chat_template API and the official Gemma templates generate this structure automatically for messages and tool lists.

Fine tuning and Mobile Actions performance

Out of the box, FunctionGemma is already trained for generic tool use. However, the official Mobile Actions guide and the model card emphasize that small models reach production level reliability only after task specific fine tuning.

The Mobile Actions demo uses a dataset where each example exposes a small set of tools for Android system operations, for example create a contact, set a calendar event, control the flashlight and map viewing. FunctionGemma learns to map utterances such as ‘Create a calendar event for lunch tomorrow’ or ‘Turn on the flashlight’ to those tools with structured arguments.

On the Mobile Actions evaluation, the base FunctionGemma model reaches 58 percent accuracy on a held out test set. After fine tuning with the public cookbook recipe, accuracy increases to 85 percent.

Edge agents and reference demos

The main deployment target for FunctionGemma is edge agents that run locally on phones, laptops and small accelerators such as NVIDIA Jetson Nano. The small parameter count, 0.3B, and support for quantization allow inference with low memory and low latency on consumer hardware.

Google ships several reference experiences through the Google AI Edge Gallery

Mobile Actions shows a fully offline assistant style agent for device control using FunctionGemma fine tuned on the Mobile Actions dataset and deployed on device.

Tiny Garden is a voice controlled game where the model decomposes commands such as “Plant sunflowers in the top row and water them” into domain specific functions like plant_seed and water_plots with explicit grid coordinates.

FunctionGemma Physics Playground runs entirely in the browser using Transformers.js and lets users solve physics puzzles via natural language instructions that the model converts into simulation actions.

These demos validate that a 270M parameter function caller can support multi step logic on device without server calls, given appropriate fine tuning and tool interfaces.

Key Takeaways

FunctionGemma is a 270M parameter, text only variant of Gemma 3 that is trained specifically for function calling, not for open ended chat, and is released as an open model under the Gemma terms of use.

The model keeps the Gemma 3 transformer architecture and 256k token vocabulary, supports 32k tokens per request shared between input and output, and is trained on 6T tokens.

FunctionGemma uses a strict chat template with <start_of_turn>role … <end_of_turn> and dedicated control tokens for function declarations, function calls and function responses, which is required for reliable tool use in production systems.

On the Mobile Actions benchmark, accuracy improves from 58 percent for the base model to 85 percent after task specific fine tuning, showing that small function callers need domain data more than prompt engineering.

The 270M scale and quantization support let FunctionGemma run on phones, laptops and Jetson class devices, and the model is already integrated into ecosystems such as Hugging Face, Vertex AI, LM Studio and edge demos like Mobile Actions, Tiny Garden and the Physics Playground.

Check out the Technical details and Model on HF. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post From Gemma 3 270M to FunctionGemma, How Google AI Built a Compact Function Calling Specialist for Edge Workloads appeared first on MarkTechPost.

In this tutorial, we build an advanced, fully autonomous logistics simulation in which multiple smart delivery trucks operate within a dynamic city-wide road network. We design the system so that each truck behaves as an agent capable of bidding on delivery orders, planning optimal routes, managing battery levels, seeking charging stations, and maximizing profit through self-interested decision-making. Through each code snippet, we explore how agentic behaviors emerge from simple rules, how competition shapes order allocation, and how a graph-based world enables realistic movement, routing, and resource constraints. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserimport networkx as nx
import matplotlib.pyplot as plt
import random
import time
from IPython.display import clear_output
from dataclasses import dataclass, field
from typing import List, Dict, Optional

NUM_NODES = 30
CONNECTION_RADIUS = 0.25
NUM_AGENTS = 5
STARTING_BALANCE = 1000
FUEL_PRICE = 2.0
PAYOUT_MULTIPLIER = 5.0
BATTERY_CAPACITY = 100
CRITICAL_BATTERY = 25

@dataclass
class Order:
id: str
target_node: int
weight_kg: int
payout: float
status: str = “pending”

class AgenticTruck:
def __init__(self, agent_id, start_node, graph, capacity=100):
self.id = agent_id
self.current_node = start_node
self.graph = graph
self.battery = BATTERY_CAPACITY
self.balance = STARTING_BALANCE
self.capacity = capacity
self.state = “IDLE”
self.path: List[int] = []
self.current_order: Optional[Order] = None
self.target_node: int = start_node

We set up all the core building blocks of the simulation, including imports, global parameters, and the basic data structures. We also define the AgenticTruck class and initialize key attributes, including position, battery, balance, and operating state. We lay the foundation for all agent behaviors to evolve. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser def get_path_cost(self, start, end):
try:
length = nx.shortest_path_length(self.graph, start, end, weight=’weight’)
path = nx.shortest_path(self.graph, start, end, weight=’weight’)
return length, path
except nx.NetworkXNoPath:
return float(‘inf’), []

def find_nearest_charger(self):
chargers = [n for n, attr in self.graph.nodes(data=True) if attr.get(‘type’) == ‘charger’]
best_charger = None
min_dist = float(‘inf’)
best_path = []
for charger in chargers:
dist, path = self.get_path_cost(self.current_node, charger)
if dist < min_dist:
min_dist = dist
best_charger = charger
best_path = path
return best_charger, best_path

def calculate_bid(self, order):
if order.weight_kg > self.capacity:
return float(‘inf’)
if self.state != “IDLE” or self.battery < CRITICAL_BATTERY:
return float(‘inf’)
dist_to_target, _ = self.get_path_cost(self.current_node, order.target_node)
fuel_cost = dist_to_target * FUEL_PRICE
expected_profit = order.payout – fuel_cost
if expected_profit < 10:
return float(‘inf’)
return dist_to_target

def assign_order(self, order):
self.current_order = order
self.state = “MOVING”
self.target_node = order.target_node
_, self.path = self.get_path_cost(self.current_node, self.target_node)
if self.path: self.path.pop(0)

def go_charge(self):
charger_node, path = self.find_nearest_charger()
if charger_node is not None:
self.state = “TO_CHARGER”
self.target_node = charger_node
self.path = path
if self.path: self.path.pop(0)

We implement advanced decision-making logic for the trucks. We calculate shortest paths, identify nearby charging stations, and evaluate whether an order is profitable and feasible. We also prepare the truck to accept assignments or proactively seek charging when needed. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser def step(self):
if self.state == “IDLE” and self.battery < CRITICAL_BATTERY:
self.go_charge()

if self.state == “CHARGING”:
self.battery += 10
self.balance -= 5
if self.battery >= 100:
self.battery = 100
self.state = “IDLE”
return

if self.path:
next_node = self.path[0]
edge_data = self.graph.get_edge_data(self.current_node, next_node)
distance = edge_data[‘weight’]
self.current_node = next_node
self.path.pop(0)
self.battery -= (distance * 2)
self.balance -= (distance * FUEL_PRICE)

if not self.path:
if self.state == “MOVING”:
self.balance += self.current_order.payout
self.current_order.status = “completed”
self.current_order = None
self.state = “IDLE”
elif self.state == “TO_CHARGER”:
self.state = “CHARGING”

We manage the step-by-step actions of each truck as the simulation runs. We handle battery recharging, financial impacts of movement, fuel consumption, and order completion. We ensure that agents transition smoothly between states, such as moving, charging, and idling. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserclass Simulation:
def __init__(self):
self.setup_graph()
self.setup_agents()
self.orders = []
self.order_count = 0

def setup_graph(self):
self.G = nx.random_geometric_graph(NUM_NODES, CONNECTION_RADIUS)
for (u, v) in self.G.edges():
self.G.edges[u, v][‘weight’] = random.uniform(1.0, 3.0)
for i in self.G.nodes():
r = random.random()
if r < 0.15:
self.G.nodes[i][‘type’] = ‘charger’
self.G.nodes[i][‘color’] = ‘red’
else:
self.G.nodes[i][‘type’] = ‘house’
self.G.nodes[i][‘color’] = ‘#A0CBE2’

def setup_agents(self):
self.agents = []
for i in range(NUM_AGENTS):
start_node = random.randint(0, NUM_NODES-1)
cap = random.choice([50, 100, 200])
self.agents.append(AgenticTruck(i, start_node, self.G, capacity=cap))

def generate_order(self):
target = random.randint(0, NUM_NODES-1)
weight = random.randint(10, 120)
payout = random.randint(50, 200)
order = Order(id=f”ORD-{self.order_count}”, target_node=target, weight_kg=weight, payout=payout)
self.orders.append(order)
self.order_count += 1
return order

def run_market(self):
for order in self.orders:
if order.status == “pending”:
bids = {agent: agent.calculate_bid(order) for agent in self.agents}
valid_bids = {k: v for k, v in bids.items() if v != float(‘inf’)}
if valid_bids:
winner = min(valid_bids, key=valid_bids.get)
winner.assign_order(order)
order.status = “assigned”

We create the simulated world and orchestrate agent interactions. We generate the graph-based city, spawn trucks with varying capacities, and produce new delivery orders. We also implement a simple market where agents bid for tasks based on profitability and distance. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser def step(self):
if random.random() < 0.3:
self.generate_order()
self.run_market()
for agent in self.agents:
agent.step()

def visualize(self, step_num):
clear_output(wait=True)
plt.figure(figsize=(10, 8))
pos = nx.get_node_attributes(self.G, ‘pos’)
node_colors = [self.G.nodes[n][‘color’] for n in self.G.nodes()]
nx.draw(self.G, pos, node_color=node_colors, with_labels=True, node_size=300, edge_color=’gray’, alpha=0.6)

for agent in self.agents:
x, y = pos[agent.current_node]
jitter_x = x + random.uniform(-0.02, 0.02)
jitter_y = y + random.uniform(-0.02, 0.02)
color = ‘green’ if agent.state == “IDLE” else (‘orange’ if agent.state == “MOVING” else ‘red’)
plt.plot(jitter_x, jitter_y, marker=’s’, markersize=12, color=color, markeredgecolor=’black’)
plt.text(jitter_x, jitter_y+0.03, f”A{agent.id}n${int(agent.balance)}n{int(agent.battery)}%”,
fontsize=8, ha=’center’, fontweight=’bold’, bbox=dict(facecolor=’white’, alpha=0.7, pad=1))

for order in self.orders:
if order.status in [“assigned”, “pending”]:
ox, oy = pos[order.target_node]
plt.plot(ox, oy, marker=’*’, markersize=15, color=’gold’, markeredgecolor=’black’)

plt.title(f”Graph-Based Logistics Swarm | Step: {step_num}nRed Nodes = Chargers | Gold Stars = Orders”, fontsize=14)
plt.show()

print(“Initializing Advanced Simulation…”)
sim = Simulation()

for t in range(60):
sim.step()
sim.visualize(t)
time.sleep(0.5)

print(“Simulation Finished.”)

We step through the full simulation loop and visualize the logistics swarm in real time. We update agent states, draw the network, display active orders, and animate each truck’s movement. By running this loop, we observe the emergent coordination and competition that define our multi-agent logistics ecosystem.

In conclusion, we saw how the individual components, graph generation, autonomous routing, battery management, auctions, and visualization, come together to form a living, evolving system of agentic trucks. We watch as agents negotiate workloads, compete for profitable opportunities, and respond to environmental pressures such as distance, fuel costs, and charging needs. By running the simulation, we observe emergent dynamics that mirror real-world fleet behavior, providing a powerful sandbox for experimenting with logistics intelligence.

Check out the FULL CODES here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post A Coding Guide to Build an Autonomous Multi-Agent Logistics System with Route Planning, Dynamic Auctions, and Real-Time Visualization Using Graph-Based Simulation appeared first on MarkTechPost.