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 data, 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 inner assist notes, demonstrating how Griptape allows 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
strive:
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 atmosphere 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, Non-compulsory
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("textual content"): str})})
def redact_pii(self, params: dict):
strive:
return TextArtifact(_redact(params["values"]["text"]))
besides Exception as e:
return ErrorArtifact(str(e))
@exercise(config={"description": "Categorize ticket", "schema": Schema({Literal("textual content"): str})})
def categorize(self, params: dict):
strive:
t = params["values"]["text"].decrease()
if any(okay in t for okay in ["charged", "refund", "invoice", "billing", "payment"]):
cat = "billing"
elif any(okay in t for okay in ["crash", "error", "bug", "export", "0x"]):
cat = "bug"
elif any(okay in t for okay in ["locked", "password", "login attempts", "unauthorized", "security"]):
cat = "safety"
elif any(okay in t for okay in ["account", "profile", "access"]):
cat = "account"
else:
cat = "different"
return TextArtifact(cat)
besides Exception as e:
return ErrorArtifact(str(e))
@exercise(config={"description": "Precedence and SLA", "schema": Schema({Literal("class"): str, Literal("textual content"): str, Non-compulsory(Literal("channel"), default="internet"): str})})
def priority_and_sla(self, params: dict):
strive:
cat = params["values"]["category"].decrease()
t = params["values"]["text"].decrease()
channel = params["values"].get("channel", "internet")
if cat == "safety" or "pressing" in t or "asap" in t:
p, sla = 1, "quarter-hour"
elif cat in ["billing", "account"]:
p, sla = 2, "2 hours"
elif cat == "bug":
p, sla = 3, "1 enterprise day"
else:
p, sla = 4, "3 enterprise days"
if channel == "chat" and p > 1:
p = max(2, p - 1)
return TextArtifact(json.dumps({"precedence": p, "sla_target": sla}))
besides Exception as e:
return ErrorArtifact(str(e))
@exercise(config={"description": "Escalation payload", "schema": Schema({Literal("ticket_id"): str, Literal("buyer"): str, Literal("class"): str, Literal("precedence"): int, Literal("sanitized_text"): str})})
def build_escalation_json(self, params: dict):
strive:
v = params["values"]
payload = {
"abstract": f"[{v['category'].higher()}][P{v['priority']}] Ticket {v['ticket_id']} - {v['customer']}",
"labels": [v["category"], f"p{v['priority']}"],
"description": v["sanitized_text"],
"buyer": 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 material/ticket_tools.py", "w", encoding="utf-8") as f:
f.write(tool_code)
import importlib, sys
sys.path.append("/content material")
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 task, and the era 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": "e mail", "created_at": "2026-02-01T10:14:00Z"},
{"ticket_id": "TCK-1002", "buyer": "Rohan", "textual content": "App crashes each time I attempt to export. Screenshot reveals error code 0x7f. My telephone: +1 514-555-0188", "channel": "chat", "created_at": "2026-02-01T10:20:00Z"},
{"ticket_id": "TCK-1003", "buyer": "Mina", "textual content": "Want bill for January. Additionally replace billing tackle to 21 King St, Montreal.", "channel": "e mail", "created_at": "2026-02-01T10:33:00Z"},
{"ticket_id": "TCK-1004", "buyer": "Sam", "textual content": "My account acquired locked after password reset. I’m seeing login makes an attempt I do not acknowledge. Please assist urgently.", "channel": "internet", "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 akin to channel, timestamp, and free-form textual content to mirror actual operational information. We use this dataset to constantly check and display the total pipeline.
from griptape.constructions 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": {"textual content": ticket["text"]}}).to_text()
class = instrument.categorize({"values": {"textual content": sanitized}}).to_text().strip()
pr_sla = json.hundreds(instrument.priority_and_sla({"values": {"class": class, "textual content": sanitized, "channel": ticket["channel"]}}).to_text())
escalation = instrument.build_escalation_json({"values": {"ticket_id": ticket["ticket_id"], "buyer": ticket["customer"], "class": class, "precedence": 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 choice
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"], "class": class, "precedence": 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 perform that chains instrument calls earlier than invoking the agent, making certain all delicate dealing with and classification are accomplished first. We then ask the agent to generate buyer responses and inner 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']} | class={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 gather the structured outcomes. We print escalation payloads and agent-generated Markdown outputs to confirm correctness and readability. We use this closing 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 complicated operational workflows through which logic, coverage, and AI reasoning coexist cleanly. We relied on deterministic instruments for classification, threat 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 will scale AI-assisted operations safely, combine them into present assist programs, and preserve strict management over habits, outputs, and repair ensures utilizing Griptape’s core abstractions.
Try the Full Codes right here. Additionally, be at liberty to observe us on Twitter and don’t overlook to hitch our 100k+ ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you possibly can be part of us on telegram as nicely.
Elevate your perspective with NextTech Information, the place innovation meets perception.
Uncover the newest breakthroughs, get unique updates, and join with a worldwide community of future-focused thinkers.
Unlock tomorrow’s tendencies immediately: learn extra, subscribe to our publication, and turn into a part of the NextTech neighborhood at NextTech-news.com
