On this tutorial, we construct a sophisticated Griptape-based buyer assist automation system that mixes deterministic tooling with agentic reasoning to course of real-world assist tickets end-to-end. We design customized instruments to sanitize delicate info, categorize points, assign priorities with clear SLA targets, and generate structured escalation payloads, all earlier than involving the language mannequin. We then use a Griptape Agent to synthesize these instrument outputs into skilled buyer replies and inside assist notes, demonstrating how Griptape permits managed, auditable, and production-ready AI workflows with out counting on retrieval or exterior information bases.
!pip -q set up "griptape[all]" wealthy schema pandas
import os, re, json
from getpass import getpass
attempt:
from google.colab import userdata
os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY")
besides Exception:
move
if not os.environ.get("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = getpass("Enter OPENAI_API_KEY: ")We arrange the execution surroundings by putting in all required Griptape dependencies and supporting libraries. We securely load the OpenAI API key utilizing Colab secrets and techniques or a runtime immediate to maintain credentials out of the code. We make sure the pocket book is prepared for agent execution earlier than any logic is outlined.
tool_code = r'''
import re, json
from schema import Schema, Literal, Elective
from griptape.instruments import BaseTool
from griptape.utils.decorators import exercise
from griptape.artifacts import TextArtifact, ErrorArtifact
def _redact(textual content: str) -> str:
textual content = re.sub(r"[w.-]+@[w.-]+.w+", "[REDACTED_EMAIL]", textual content)
textual content = re.sub(r"+?d[d-s()]{7,}d", "[REDACTED_PHONE]", textual content)
textual content = re.sub(r"b(d{4}[s-]?){3}d{4}b", "[REDACTED_CARD]", textual content)
return textual content
class TicketOpsTool(BaseTool):
@exercise(config={"description": "Redact PII", "schema": Schema({Literal("text"): str})})
def redact_pii(self, params: dict):
attempt:
return TextArtifact(_redact(params["values"]["text"]))
besides Exception as e:
return ErrorArtifact(str(e))
@exercise(config={"description": "Categorize ticket", "schema": Schema({Literal("text"): str})})
def categorize(self, params: dict):
attempt:
t = params["values"]["text"].decrease()
if any(ok in t for ok in ["charged", "refund", "invoice", "billing", "payment"]):
cat = "billing"
elif any(ok in t for ok in ["crash", "error", "bug", "export", "0x"]):
cat = "bug"
elif any(ok in t for ok in ["locked", "password", "login attempts", "unauthorized", "security"]):
cat = "security"
elif any(ok in t for ok in ["account", "profile", "access"]):
cat = "account"
else:
cat = "other"
return TextArtifact(cat)
besides Exception as e:
return ErrorArtifact(str(e))
@exercise(config={"description": "Priority and SLA", "schema": Schema({Literal("category"): str, Literal("text"): str, Elective(Literal("channel"), default="web"): str})})
def priority_and_sla(self, params: dict):
attempt:
cat = params["values"]["category"].decrease()
t = params["values"]["text"].decrease()
channel = params["values"].get("channel", "web")
if cat == "security" or "urgent" in t or "asap" in t:
p, sla = 1, "15 minutes"
elif cat in ["billing", "account"]:
p, sla = 2, "2 hours"
elif cat == "bug":
p, sla = 3, "1 business day"
else:
p, sla = 4, "3 business days"
if channel == "chat" and p > 1:
p = max(2, p - 1)
return TextArtifact(json.dumps({"priority": p, "sla_target": sla}))
besides Exception as e:
return ErrorArtifact(str(e))
@exercise(config={"description": "Escalation payload", "schema": Schema({Literal("ticket_id"): str, Literal("customer"): str, Literal("category"): str, Literal("priority"): int, Literal("sanitized_text"): str})})
def build_escalation_json(self, params: dict):
attempt:
v = params["values"]
payload = {
"summary": f"[{v['category'].upper()}][P{v['priority']}] Ticket {v['ticket_id']} - {v['customer']}",
"labels": [v["category"], f"p{v['priority']}"],
"description": v["sanitized_text"],
"customer": v["customer"],
"source_ticket": v["ticket_id"]
}
return TextArtifact(json.dumps(payload, indent=2))
besides Exception as e:
return ErrorArtifact(str(e))
'''
with open("/content/ticket_tools.py", "w", encoding="utf-8") as f:
f.write(tool_code)
import importlib, sys
sys.path.append("/content")
ticket_tools = importlib.import_module("ticket_tools")
TicketOpsTool = ticket_tools.TicketOpsTool
instrument = TicketOpsTool()We implement the core operational logic by defining a customized Griptape instrument inside a standalone Python module. We encode deterministic guidelines for PII redaction, ticket categorization, precedence scoring, SLA project, and the technology of escalation payloads. We then import and instantiate this instrument so it may be safely inspected and utilized by Griptape.
TICKETS = [
{"ticket_id": "TCK-1001", "customer": "Leila", "text": "I was charged twice on my card ending 4432. Please refund ASAP. email: [email protected]", "channel": "email", "created_at": "2026-02-01T10:14:00Z"},
{"ticket_id": "TCK-1002", "customer": "Rohan", "text": "App crashes every time I try to export. Screenshot shows error code 0x7f. My phone: +1 514-555-0188", "channel": "chat", "created_at": "2026-02-01T10:20:00Z"},
{"ticket_id": "TCK-1003", "customer": "Mina", "text": "Need invoice for January. Also update billing address to 21 King St, Montreal.", "channel": "email", "created_at": "2026-02-01T10:33:00Z"},
{"ticket_id": "TCK-1004", "customer": "Sam", "text": "My account got locked after password reset. I’m seeing login attempts I don't recognize. Please help urgently.", "channel": "web", "created_at": "2026-02-01T10:45:00Z"}
]
We create a practical stream of buyer assist tickets that acts as our enter workload. We construction every ticket with metadata resembling channel, timestamp, and free-form textual content to replicate actual operational information. We use this dataset to constantly check and show the complete pipeline.
from griptape.buildings import Agent
from griptape.drivers.immediate.openai import OpenAiChatPromptDriver
prompt_driver = OpenAiChatPromptDriver(mannequin="gpt-4.1")
agent = Agent(prompt_driver=prompt_driver, instruments=[tool])
def run_ticket(ticket: dict) -> dict:
sanitized = instrument.redact_pii({"values": {"text": ticket["text"]}}).to_text()
class = instrument.categorize({"values": {"text": sanitized}}).to_text().strip()
pr_sla = json.masses(instrument.priority_and_sla({"values": {"category": class, "text": sanitized, "channel": ticket["channel"]}}).to_text())
escalation = instrument.build_escalation_json({"values": {"ticket_id": ticket["ticket_id"], "customer": ticket["customer"], "category": class, "priority": int(pr_sla["priority"]), "sanitized_text": sanitized}}).to_text()
immediate = f"""
You're a senior assist lead. Produce:
1) A customer-facing reply
2) Inside notes
3) Escalation determination
Ticket:
- id: {ticket['ticket_id']}
- buyer: {ticket['customer']}
- channel: {ticket['channel']}
- class: {class}
- precedence: {pr_sla['priority']}
- SLA goal: {pr_sla['sla_target']}
- sanitized_text: {sanitized}
Output in Markdown.
"""
out = agent.run(immediate).to_text()
return {"ticket_id": ticket["ticket_id"], "category": class, "priority": pr_sla["priority"], "sla_target": pr_sla["sla_target"], "escalation_payload_json": escalation, "agent_output_markdown": out}We initialize a Griptape Agent with the customized instrument and a immediate driver to allow managed reasoning. We outline a deterministic processing operate that chains instrument calls earlier than invoking the agent, guaranteeing all delicate dealing with and classification are accomplished first. We then ask the agent to generate buyer responses and inside notes based mostly solely on instrument outputs.
outcomes = [run_ticket(t) for t in TICKETS]
for r in outcomes:
print("n" + "=" * 88)
print(f"{r['ticket_id']} | category={r['category']} | P{r['priority']} | SLA={r['sla_target']}")
print(r["escalation_payload_json"])
print(r["agent_output_markdown"])We execute the pipeline throughout all tickets and acquire the structured outcomes. We print escalation payloads and agent-generated Markdown outputs to confirm correctness and readability. We use this remaining step to validate that the workflow runs end-to-end with out hidden dependencies or retrieval logic.
In conclusion, we demonstrated how Griptape can be utilized to orchestrate advanced operational workflows through which logic, coverage, and AI reasoning coexist cleanly. We relied on deterministic instruments for classification, danger dealing with, and escalation, utilizing the agent solely the place natural-language judgment is required to maintain the system dependable and explainable. This sample illustrates how we are able to scale AI-assisted operations safely, combine them into present assist methods, and preserve strict management over conduct, outputs, and repair ensures utilizing Griptape’s core abstractions.
Take a look at the Full Codes right here. Additionally, be happy to comply with us on Twitter and don’t neglect to affix our 100k+ ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you may be a part of us on telegram as properly.
