Have you ever felt like building with AI agents is a bit like hiring a brilliant but incredibly chaotic intern? They can do amazing things, but you’d never let them push code to production without some serious supervision. They’re unpredictable, sometimes make stuff up, and debugging their "thought process" can feel like a nightmare.
That’s the big challenge we’re all facing right now. How do we get the incredible power of Large Language Models (LLMs) into our real-world applications without losing control, reliability, and our sanity?
What if I told you there’s a way to build systems that have a solid, predictable foundation but can be "upgraded" with AI intelligence when you’re ready? A way to create workflows that can run completely offline, based on simple rules, but can also be handed over to a powerful LLM to orchestrate on the fly.
That’s exactly what we’re going to build today using a cool framework called GraphBit. We're going to create a customer support ticket system that starts as a 100% deterministic, rule-based pipeline and then, by just flipping a switch, transforms into a smart, agentic workflow. Let’s get into it.
First Things First: Getting Our Workspace Ready
Before we can build anything, we need to set up our environment. Think of this as laying out all our tools on the workbench. We'll install GraphBit and a few other helpful libraries.
# Full code available in the original source
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
With everything installed, the next step is to actually start the GraphBit runtime. This is like turning on the power in our workshop. We’re configuring things like how many threads it can use, making sure it’s healthy, and just generally getting it ready for action.
# Full code available in the original source
from 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 and initialize the runtime
configure_runtime(worker_threads=4, max_blocking_threads=8, thread_stack_size_mb=2)
init(log_level="warn", enable_tracing=False, debug=False)
# Let's check if everything is working
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)
If you run this, you should see a nice table confirming that the system is up and running. Perfect. Our workshop is open for business.
Let's Create Some Realistic Data to Work With
We can't build a support ticket system without, well, support tickets. We’re not going to connect to a real database just yet. Instead, let's create some realistic-looking sample data. This gives us a consistent set of inputs to test our system against, both in its simple "offline" mode and its "online" agent mode.
We’ll define a simple structure for a Ticket and a helper function to generate a bunch of them.
# Full code available in the original source
@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"))
Now we have a list of tickets that look like real customer problems. This is the raw material our workflow will process.
Building Our Reliable "Worker Bots": The Deterministic Tools
Here’s where we build the foundation of our reliable system. We're going to create a few Python functions that perform specific tasks: classifying a ticket, routing it to the right team, and drafting a basic response.
The key word here is deterministic. These functions are just plain old code. They don't use an LLM. They have no randomness. If you give them the same input, they will produce the exact same output, every single time. Think of them as specialized, reliable robots on an assembly line.
GraphBit makes it easy to turn any function into a "tool" that an agent can use later. You just add the @tool decorator.
# Full code available in the original source
clear_tools()
@tool(_description="Classify a support ticket into a coarse category.")
def classify_ticket(text: str) -> Dict[str, Any]:
# ... (simple if/else logic based on keywords)
t = text.lower()
if "fraud" in t or "fraudulent" in t:
return {"category": "fraud", "priority": "p0"}
# ... more rules
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]:
# ... (simple dictionary lookup logic)
queue_map = { "fraud": ("risk_ops", 2), "cancellation": ("retention", 8), #... more mappings
}
q, sla = queue_map.get(category, ("support_general", 48))
# ... priority adjustments
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]:
# ... (simple template-based logic)
templates = { "fraud": "We’ve temporarily secured your account...", #... more templates
}
base = templates.get(category, templates["general"])
# ... tone adjustments
return { "tone": tone, "message": f"{base}\n\nContext 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 now have three reliable, testable tools: classify_ticket, route_ticket, and draft_response. They form the backbone of our entire operation.
The "Offline" Test Run: A Predictable, Rule-Based Pipeline
Before we even think about bringing in an LLM, let's prove that our basic logic works. We can chain our new tools together in a simple Python function to create a fully "offline" pipeline. This means no AI, no API calls, just our own code running.
This step is so important. It gives us a baseline. We can run our tickets through this pipeline and see exactly what happens. We can write tests for it. We can trust it.
# Full code available in the original source
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]
# --- Code to display results in a table and calculate metrics ---
res_table = Table(title="Offline Pipeline Results")
# ... (add columns and rows)
rprint(res_table)
# ... (code to calculate and print metrics like priority distribution and SLA)
metrics = { "offline_mode": True, "tickets": len(offline_results), #... more metrics
}
rprint(Panel.fit(json.dumps(metrics, indent=2), title="Offline Metrics"))
When you run this, you'll get a clean table of categorized, prioritized, and routed tickets, plus some useful metrics. It’s fast, free to run, and 100% predictable. This is our safety net.
Leveling Up: Designing the Agentic Workflow Graph
Okay, our reliable, rule-based system is working perfectly. Now, let's design the "smart" version. We’re going to define a workflow using GraphBit's Workflow and Node structure.
Think of it like designing the blueprint for a smart factory floor. Each Node is a workstation, and we connect them to define the flow of work. Some workstations might be simple robots, while others are highly skilled human workers (our LLM agents).
We'll create a three-step workflow:
- Summarizer: A simple agent to create a one-line summary.
- RouterAgent: The core agent. This one’s job is to use the deterministic tools we built earlier to process the ticket.
- FinalFormatter: A cleanup agent that ensures the final output is in perfect JSON format.
# Full code available in the original source
SYSTEM_POLICY = "You are a reliable support ops agent. Return STRICT JSON only."
workflow = Workflow("Ticket Triage Workflow (GraphBit)")
# Node 1: Summarizer
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,
)
# Node 2: The main agent that uses our tools
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], # <-- We give it our tools!
temperature=0.1,
)
# Node 3: Final Formatter
formatter = Node.agent(
name="FinalFormatter",
# ... (configuration for final JSON validation)
)
# Connect the nodes to form a graph: Summarizer -> Router -> Formatter
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’ve now defined the structure of our intelligent workflow. We've told the RouterAgent that it has access to our reliable tools. But notice, we haven't actually run anything yet. We've just drawn the blueprint.
Flipping the Switch: Going "Online" with an LLM
This is the magic moment. We have our deterministic tools and our agentic workflow blueprint. Now, we'll write the code that can execute this workflow using a real LLM.
The beauty of this setup is that it only runs the "online" version if you provide an API key (for OpenAI, Anthropic, etc.). If no key is found, it can gracefully do nothing. This is how you build systems that can be progressively enhanced.
Let’s write a function that picks an LLM configuration if an API key is present and then executes our graph for a single ticket.
# Full code available in the original source
def pick_llm_config() -> Optional[Any]:
if os.getenv("OPENAI_API_KEY"):
return LlmConfig.openai(os.getenv("OPENAI_API_KEY"), "gpt-4o-mini")
# ... (similar checks for Anthropic, DeepSeek, Mistral)
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 an LLM API_KEY to enable execution.",
"input": ticket_text
}
# The executor is what runs the workflow with the LLM config
executor = Executor(llm_cfg, lightweight_mode=True, timeout_seconds=90, debug=False)
# We redefine the workflow here for a single run
wf = Workflow("Single Ticket Run")
# ... (the same node definitions as before, but with the ticket_text baked in)
s = Node.agent(...)
r = Node.agent(...)
f = Node.agent(...)
# ... (add and connect nodes)
wf.validate()
t0 = time.time()
result = executor.execute(wf) # <-- This is where the magic happens!
dt_ms = int((time.time() - t0) * 1000)
out = {"mode": "online", "execution_time_ms": dt_ms, "success": bool(result.is_success())}
if hasattr(result, "get_all_variables"):
out["variables"] = result.get_all_variables()
else:
out["raw"] = str(result)[:3000]
return out
# Let's try it on one ticket
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"))
If you set an environment variable like OPENAI_API_KEY, this code will execute the full agentic graph. The LLM will receive the ticket, the RouterAgent will intelligently decide to call our classify_ticket, route_ticket, and draft_response tools with the right arguments, and the final result will be assembled and formatted.
And there you have it. We’ve built a system that bridges the gap between predictable, rule-based logic and flexible, AI-driven orchestration. We started with a foundation we could trust and test, and then layered the intelligence on top. This isn't just a cool demo; it’s a practical pattern for building production-grade AI systems that are reliable, debuggable, and ready for the real world.




