Customer Support Concession Approval¶
Support organizations gate costly concessions behind a tier-based escalation chain: the more a concession costs and the more precedent it sets, the more approvals it takes.
This example runs four support tickets of rising risk through a
StateGraph. A ticket analyst summarizes what the customer is asking
for, an impact analyst weighs the cost of the concession and its
precedent risk against the customer's standing and the downside of a
denial (churn, escalation, public complaint), and a risk-tier router
sends the case to automatic approval or to up to three human gates —
support manager, billing lead, support director — each a separate
interrupt().
The terminal node, the decision recorder (the
scribe-decision-recorder agent in the source), emits a typed
ConcessionDecision Pydantic model, ready to file into a concessions
ledger without parsing. A large refund or contract make-good commits
real money and sets a precedent other customers will cite, which is why
the impact assessment sits in front of the gates.
What this notebook covers:
- Risk-tier router is a plain conditional edge — no domain-specific language (DSL), no policy file.
- Each gate is its own node — easy to add a tier, easy to re-order, easy to swap a human gate for an automated rule.
output_schema=ConcessionDecisionkeeps the terminal artifact typed.
The drawing is the graph the example builds: the risk score arrives with
the ticket, a score under 25 is approved automatically, 25–49 stops at
the support manager, 50–74 adds the billing lead, 75 and above adds the
support director, and a no at any gate goes straight to the decision
recorder with status denied:
Key lines¶
One factory builds the three human gates. The interrupt() payload
carries the tier, the question and the yes/no options — all the run loop
needs to print the gate — and _record_decision turns the answer into
state: a yes appends the approver's title to approvals, a no sets
status to denied, which is what the gate functions check before
climbing a tier:
def _approval_node(tier: str, title: str):
"""Build one human-approval gate.
The three tiers differ only in label, but each stays its own graph node
so a reviewer can come back to exactly that gate later.
"""
async def approve(state: dict[str, Any]) -> dict[str, Any]:
decision = interrupt(
{
"type": "approval",
"tier": tier,
"question": (
f"{title} approval needed at risk score "
f"{state['risk_score']:.0f}/100 "
f"({state['customer']} — {state['concession']}). Approve?"
),
"options": ["yes", "no"],
}
)
return _record_decision(state, title, decision)
approve.__name__ = f"approve_{tier}"
return approve
The wiring is ordinary graph edges. risk_tier_router is a plain
conditional edge off the impact assessment: a score under 25 goes to
auto_approve, every other tier enters at the manager gate, and
gate_after_manager and gate_after_billing climb to the next tier only
while the score requires it and no gate has said no:
def build_review_graph() -> StateGraph:
g = StateGraph(name="concession-approval-review")
g.add_node("summarize_ticket", summarize_ticket)
g.add_node("assess_impact", assess_impact)
g.add_node("auto_approve", auto_approve)
g.add_node("approve_manager", approve_manager)
g.add_node("approve_billing", approve_billing)
g.add_node("approve_director", approve_director)
g.add_node("record_decision", record_decision)
g.add_edge(START, "summarize_ticket")
g.add_edge("summarize_ticket", "assess_impact")
g.add_conditional_edges(
"assess_impact",
risk_tier_router,
targets={
"auto": "auto_approve",
"manager": "approve_manager",
"billing": "approve_manager",
"director": "approve_manager",
},
)
g.add_edge("auto_approve", "record_decision")
g.add_conditional_edges(
"approve_manager",
gate_after_manager,
targets={
"approve_billing": "approve_billing",
"record_decision": "record_decision",
},
)
g.add_conditional_edges(
"approve_billing",
gate_after_billing,
targets={
"approve_director": "approve_director",
"record_decision": "record_decision",
},
)
g.add_edge("approve_director", "record_decision")
g.add_edge("record_decision", END)
return g
The run loop answers each gate from a scripted list. execute() returns
with result.interrupt set at every gate and the reviewer's answer goes
back inside Command(resume=...). The graph has no checkpointer, so the
state travels back in the same Command: the returned final_state
carries a __resume_node__ marker that execute() pops to find the gate
to re-run, and the loop keeps the model object in that state so every
later node can still build its agent:
async def _drive(graph: StateGraph, initial: dict[str, Any], answers: list[str]) -> Any:
"""Run the graph, auto-resuming interrupts with the supplied answers."""
result = await graph.execute(initial)
answer_idx = 0
while result.interrupt:
answer = answers[answer_idx] if answer_idx < len(answers) else "yes"
answer_idx += 1
payload = result.interrupt.interrupt.payload
print(f" ⏸ [{payload.get('tier', '?')}] {payload.get('question')}")
print(f" ▶ Reviewer responds: {answer!r}")
# The graph has no checkpointer, so the state handed back at the
# interrupt (with its __resume_node__ marker) goes forward again in
# the Command, the model object kept alongside the reviewer's answer.
result = await graph.execute(
Command(
resume=answer,
update={**result.final_state, "__model__": initial["__model__"]},
)
)
return result
Run it¶
Model choice comes from examples/config.py: OCI Generative AI when
~/.oci/config has a usable profile, otherwise the bundled mock model, so the
example also runs offline with no credentials. Set LOCUS_MODEL_PROVIDER
(oci, openai, anthropic or mock) to force one.
The decision recorder needs a model that can honor the ConcessionDecision
JSON schema. The mock model returns plain text, so under it output_schema
yields no parsed ConcessionDecision and the record_decision node raises.
execute() records the failed node and returns success=False rather than
propagating the error, and main() does not check result.success, so the
script prints the gates and the reviewer answers, reports the final decision
as (missing) and still exits with status 0:
Pin a model that supports structured output; the direct openai form needs
OPENAI_API_KEY in the environment:
export LOCUS_MODEL_PROVIDER=openai
export LOCUS_MODEL_ID=gpt-4o
python examples/notebook_78_concession_approval.py
See also¶
- Notebook 15 — Human in the loop —
interrupt(payload),Command(update=..., resume=...)andinterrupt_before - Concepts: Interrupts
- Concepts: Structured output —
output_schemaandAgentResult.parsed
Source¶
Full source — examples/notebook_78_concession_approval.py
# Copyright (c) 2025, 2026 Oracle and/or its affiliates.
# Licensed under the Universal Permissive License v1.0 as shown at
# https://oss.oracle.com/licenses/upl/
"""Notebook 78: Customer-support concession approval with risk-tiered gates.
Before a support agent can grant a costly concession — a refund, a
service credit, a goodwill gesture, a contract-level make-good — the
request has to clear a tier-based escalation chain. The more it costs
and the more precedent it sets, the more approvals it takes: a blanket
"yes" to every angry customer drains margin and trains customers to
escalate, so larger concessions climb the ladder::
Ticket intake (case history on file)
│
▼
Ticket analyst (summarises what the customer is asking for and why)
│
▼
Impact analyst (concession cost + precedent risk + churn downside)
│
▼
Risk-tier router ── score < 25 ──> auto-approve (small credit, low cost)
── 25–49 ──> support-manager approval (interrupt)
── 50–74 ──> manager + billing approval (two interrupts)
── >= 75 ──> manager + billing + director approval (three interrupts)
│
▼
Decision recorder (emits structured ConcessionDecision)
Each approval gate is a separate interrupt() so a reviewer can come back
to it later. The terminal node is SCRIBE, the support org's case
recorder: it emits a typed ConcessionDecision Pydantic model that files
into the concessions ledger without parsing. A large concession spends
real money and sets a precedent other customers will cite, so the impact
step is where you weigh the customer's standing and lifetime value
against the cost of the make-good and the downside of a denial.
- Risk-tier router is a plain conditional edge — no DSL, no policy file.
- Each gate is its own node — easy to add a tier, easy to re-order,
easy to swap a human gate for an automated rule.
- output_schema=ConcessionDecision keeps SCRIBE's terminal artifact typed.
Run it:
# Default: OCI GenAI if ~/.oci/config has a usable profile, else the bundled mock model
python examples/notebook_78_concession_approval.py
# Offline / no credentials:
LOCUS_MODEL_PROVIDER=mock python examples/notebook_78_concession_approval.py
# Pin a strong-enough model for the structured ConcessionDecision schema:
LOCUS_MODEL_PROVIDER=openai LOCUS_MODEL_ID=gpt-4o python examples/notebook_78_concession_approval.py
"""
from __future__ import annotations
import asyncio
from typing import Any
from config import get_model
from pydantic import BaseModel, Field
from locus.agent import Agent, AgentConfig
from locus.core import Command, interrupt
from locus.core.events import TerminateEvent
from locus.multiagent.graph import END, START, StateGraph
# Data shape for the terminal artifact.
class ConcessionDecision(BaseModel):
"""Final structured artifact filed into the customer-concessions ledger."""
request_id: str
customer: str
concession: str
risk_score: float
ticket_summary: str
impact_assessment: str
approvals: list[str] = Field(description="ordered list of approver titles")
approved_at: str
status: str = Field(description="approved | denied")
# Specialist prompts.
PROMPTS = {
"ticket": (
"You are a customer-support triage analyst. Given a support ticket "
"excerpt, write a two-sentence summary of what the customer is asking "
"for and the reason they give."
),
"impact": (
"You are a customer-support concession assessor. Given a customer and "
"the remedy they are requesting, write a one-paragraph assessment "
"covering: the cost of the concession to the business, the precedent "
"risk (will granting it set an expectation other customers will cite), "
"the customer's standing and lifetime value, and the downside if the "
"concession is denied (churn, escalation, public complaint)."
),
}
def _make_agent(role: str, model: Any) -> Agent:
return Agent(
config=AgentConfig(
agent_id=f"concession-{role}",
model=model,
system_prompt=PROMPTS[role],
max_iterations=2,
max_tokens=300,
)
)
async def _run(agent: Agent, prompt: str) -> str:
final = ""
async for event in agent.run(prompt):
if isinstance(event, TerminateEvent):
final = event.final_message or ""
return final.strip()
# Graph nodes.
async def summarize_ticket(state: dict[str, Any]) -> dict[str, Any]:
agent = _make_agent("ticket", state["__model__"])
text = await _run(
agent,
f"Customer: {state['customer']}\nTicket excerpt: {state['ticket']}",
)
return {"ticket_summary": text}
async def assess_impact(state: dict[str, Any]) -> dict[str, Any]:
agent = _make_agent("impact", state["__model__"])
text = await _run(
agent, f"Customer: {state['customer']}\nRequested remedy: {state['concession']}"
)
return {"impact_assessment": text}
def risk_tier_router(state: dict[str, Any]) -> str:
"""Route by risk score; each tier picks up the prior tier's approvals."""
score = float(state.get("risk_score", 0.0))
if score < 25:
return "auto"
if score < 50:
return "manager"
if score < 75:
return "billing"
return "director"
def _record_decision(state: dict[str, Any], role: str, decision: str) -> dict[str, Any]:
approvals: list[str] = list(state.get("approvals", []))
if decision == "yes":
approvals.append(role)
return {"approvals": approvals, "status": "pending"}
return {"approvals": approvals, "status": "denied"}
def _approval_node(tier: str, title: str):
"""Build one human-approval gate.
The three tiers differ only in label, but each stays its own graph node
so a reviewer can come back to exactly that gate later.
"""
async def approve(state: dict[str, Any]) -> dict[str, Any]:
decision = interrupt(
{
"type": "approval",
"tier": tier,
"question": (
f"{title} approval needed at risk score "
f"{state['risk_score']:.0f}/100 "
f"({state['customer']} — {state['concession']}). Approve?"
),
"options": ["yes", "no"],
}
)
return _record_decision(state, title, decision)
approve.__name__ = f"approve_{tier}"
return approve
approve_manager = _approval_node("manager", "Support Manager")
approve_billing = _approval_node("billing", "Billing Lead")
approve_director = _approval_node("director", "Support Director")
def gate_after_manager(state: dict[str, Any]) -> str:
"""A denial at any tier jumps straight to the decision node (status=denied)."""
if state.get("status") == "denied":
return "record_decision"
score = float(state.get("risk_score", 0.0))
if score >= 50:
return "approve_billing"
return "record_decision"
def gate_after_billing(state: dict[str, Any]) -> str:
if state.get("status") == "denied":
return "record_decision"
score = float(state.get("risk_score", 0.0))
if score >= 75:
return "approve_director"
return "record_decision"
async def auto_approve(state: dict[str, Any]) -> dict[str, Any]:
return {"approvals": ["AUTO (low cost tier)"], "status": "pending"}
async def record_decision(state: dict[str, Any]) -> dict[str, Any]:
"""SCRIBE writes the record via Agent.output_schema=ConcessionDecision.
Routing through an Agent with output_schema means the artifact is a
typed Pydantic instance — the workflow can POST it directly to the
concessions ledger without parsing.
"""
final_status = "approved" if state.get("status") != "denied" else "denied"
agent = Agent(
config=AgentConfig(
agent_id="scribe-decision-recorder",
model=state["__model__"],
system_prompt=(
"You are a customer-support case officer producing a "
"ConcessionDecision. Use the supplied fields verbatim. Don't "
"invent customers or scores."
),
output_schema=ConcessionDecision,
max_iterations=2,
max_tokens=300,
)
)
prompt = (
f"Request: {state.get('request_id')}\n"
f"Customer: {state['customer']}\n"
f"Concession: {state['concession']}\n"
f"Risk score: {float(state['risk_score'])}\n"
f"Status: {final_status}\n"
f"Approvals: {state.get('approvals', [])}\n"
f"Ticket summary: {state.get('ticket_summary', '')[:200]}\n"
f"Impact assessment: {state.get('impact_assessment', '')[:200]}\n\n"
"Emit the ConcessionDecision."
)
last_exc: BaseException | None = None
result = None
for attempt in range(3):
try:
result = await agent.arun(prompt)
break
except Exception as exc: # noqa: BLE001 — retry transient provider flakiness
last_exc = exc
await asyncio.sleep(0.5 * (attempt + 1))
if result is None:
raise RuntimeError(
f"Decision recorder failed after 3 attempts. Last error: {last_exc!r}"
) from last_exc
decision = result.parsed
if decision is None:
raise RuntimeError(
"Decision recorder returned no parsed ConcessionDecision. The "
"configured model could not honor the JSON schema. Set "
"LOCUS_MODEL_PROVIDER / LOCUS_MODEL_ID to a provider and model that "
f"support structured output. Raw output: {result.message!r}"
)
return {"concession_decision": decision}
# Build the concession-review graph.
def build_review_graph() -> StateGraph:
g = StateGraph(name="concession-approval-review")
g.add_node("summarize_ticket", summarize_ticket)
g.add_node("assess_impact", assess_impact)
g.add_node("auto_approve", auto_approve)
g.add_node("approve_manager", approve_manager)
g.add_node("approve_billing", approve_billing)
g.add_node("approve_director", approve_director)
g.add_node("record_decision", record_decision)
g.add_edge(START, "summarize_ticket")
g.add_edge("summarize_ticket", "assess_impact")
g.add_conditional_edges(
"assess_impact",
risk_tier_router,
targets={
"auto": "auto_approve",
"manager": "approve_manager",
"billing": "approve_manager",
"director": "approve_manager",
},
)
g.add_edge("auto_approve", "record_decision")
g.add_conditional_edges(
"approve_manager",
gate_after_manager,
targets={
"approve_billing": "approve_billing",
"record_decision": "record_decision",
},
)
g.add_conditional_edges(
"approve_billing",
gate_after_billing,
targets={
"approve_director": "approve_director",
"record_decision": "record_decision",
},
)
g.add_edge("approve_director", "record_decision")
g.add_edge("record_decision", END)
return g
# Run four tickets of rising risk through the graph, answering each
# approval interrupt from a scripted list, and print the typed decision.
def _print_decision(d: ConcessionDecision | None) -> None:
print("\nConcession decision:")
print("-" * 60)
if d is None:
print("(missing)")
return
print(f" Request: {d.request_id}")
print(f" Status: {d.status}")
print(f" Customer: {d.customer}")
print(f" Concession: {d.concession}")
print(f" Risk score: {d.risk_score:.0f}/100")
print(" Approvals: " + (" → ".join(d.approvals) if d.approvals else "(none)"))
print(f" Ticket: {d.ticket_summary[:120]}")
async def _drive(graph: StateGraph, initial: dict[str, Any], answers: list[str]) -> Any:
"""Run the graph, auto-resuming interrupts with the supplied answers."""
result = await graph.execute(initial)
answer_idx = 0
while result.interrupt:
answer = answers[answer_idx] if answer_idx < len(answers) else "yes"
answer_idx += 1
payload = result.interrupt.interrupt.payload
print(f" ⏸ [{payload.get('tier', '?')}] {payload.get('question')}")
print(f" ▶ Reviewer responds: {answer!r}")
# The graph has no checkpointer, so the state handed back at the
# interrupt (with its __resume_node__ marker) goes forward again in
# the Command, the model object kept alongside the reviewer's answer.
result = await graph.execute(
Command(
resume=answer,
update={**result.final_state, "__model__": initial["__model__"]},
)
)
return result
async def main() -> None:
print("Notebook 78: Customer-support concession approval with risk-tiered gates")
print("=" * 60)
model = get_model()
graph = build_review_graph()
# (request, customer, concession, risk score, ticket excerpt, reviewer answers)
# CS-1003 is denied at the billing tier to show a "no" skipping straight
# to the decision node with status=denied.
scenarios = [
(
"CS-1001",
"Dana R. (Starter plan)",
"$15 account credit for a late shipping notification",
12.0,
"Polite first-time complaint; order arrived two days late; asks for a small credit.",
[],
),
(
"CS-1002",
"Marco T. (Pro plan)",
"one-month subscription refund after a billing double-charge",
38.0,
"Charged twice this cycle; wants the duplicate month refunded; otherwise happy customer.",
["yes"],
),
(
"CS-1003",
"Acme Studio (Team plan, 40 seats)",
"full annual refund plus a goodwill credit after repeated outages",
62.0,
"Three outages this quarter; threatening to switch; asking to unwind the annual contract.",
["yes", "no"],
),
(
"CS-1004",
"Globex Corp (Enterprise, $480k ARR)",
"SLA-breach service credit and a contract make-good after a data-export failure",
88.0,
"Missed an SLA-bound export during their fiscal close; legal CC'd; demanding a contract concession.",
["yes", "yes", "yes"],
),
]
for req_id, customer, concession, score, ticket, answers in scenarios:
print(f"\n--- {req_id}: risk {score:.0f}/100 — {customer} — {concession} ---")
initial = {
"request_id": req_id,
"customer": customer,
"concession": concession,
"risk_score": score,
"ticket": ticket,
# The model rides along in graph state so every node can build its
# agent from it (see _drive for how it travels across a resume).
"__model__": model,
}
result = await _drive(graph, initial, answers)
_print_decision(result.final_state.get("concession_decision"))
if __name__ == "__main__":
asyncio.run(main())