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 inside assist notes, demonstrating how Griptape allows managed, auditable, and production-ready AI workflows with out counting on retrieval or exterior data 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:
go
if not os.environ.get("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = getpass("Enter OPENAI_API_KEY: ")We arrange the execution setting 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, Optionally available
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):
attempt:
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):
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 = "safety"
elif any(ok in t for ok 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, Optionally available(Literal("channel"), default="net"): str})})
def priority_and_sla(self, params: dict):
attempt:
cat = params["values"]["category"].decrease()
t = params["values"]["text"].decrease()
channel = params["values"].get("channel", "net")
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):
attempt:
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 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": "electronic 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 handle to 21 King St, Montreal.", "channel": "electronic 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": "net", "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 knowledge. We use this dataset to persistently 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.masses(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) Inner notes
3) Escalation resolution
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 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 primarily based 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 accumulate the structured outcomes. We print escalation payloads and agent-generated Markdown outputs to confirm correctness and readability. We use this ultimate 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 are able to scale AI-assisted operations safely, combine them into present assist programs, and keep strict management over habits, outputs, and repair ensures utilizing Griptape’s core abstractions.
Take a look at the Full Codes here. Additionally, be happy to observe us on Twitter and don’t neglect to affix our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
