Приклад AI‑агента підтримки (теорія + production-практика)

examples/
└── support-agent/
    └── python/
        ├── main.py
        ├── llm.py
        ├── gateway.py
        ├── policy.py
        ├── tools.py
        ├── requirements.txt
        └── README.md

cd examples/support-agent/python
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

export OPENAI_API_KEY="sk-..."
# optional
# export OPENAI_MODEL="gpt-4.1-mini"
# export OPENAI_TIMEOUT_SECONDS="60"

python main.py

HIGH_RISK_CATEGORIES = {"security", "billing_refund", "legal", "outage"}


def classify_ticket_risk(ticket: dict[str, Any]) -> tuple[str, str]:
    text = f"{ticket.get('subject', '')} {ticket.get('body', '')}".lower()

    if any(word in text for word in ["refund", "charged", "chargeback"]):
        return "billing_refund", "contains_refund_or_chargeback"
    if any(word in text for word in ["hacked", "2fa", "compromised", "password reset"]):
        return "security", "contains_security_signal"
    if any(word in text for word in ["gdpr", "legal", "lawyer", "contract"]):
        return "legal", "contains_legal_signal"
    if any(word in text for word in ["outage", "down", "incident", "500"]):
        return "outage", "contains_incident_signal"

    return "general", "no_high_risk_signals"

class ToolGateway:
    def __init__(
        self,
        *,
        allow: set[str],
        registry: dict[str, Callable[..., dict[str, Any]]],
        budget: Budget,
    ):
        self.allow = set(allow)
        self.registry = registry
        self.budget = budget
        self.tool_calls = 0
        self.started = time.monotonic()
        self.seen_signatures: dict[str, int] = {}

    def call(self, name: str, args: dict[str, Any]) -> dict[str, Any]:
        if time.monotonic() - self.started > self.budget.max_seconds:
            raise StopRun("max_seconds")

        self.tool_calls += 1
        if self.tool_calls > self.budget.max_tool_calls:
            raise StopRun("max_tool_calls")

        if name not in self.allow:
            raise StopRun(f"tool_denied:{name}")

        signature = f"{name}:{args_hash(args)}"
        hits = self.seen_signatures.get(signature, 0) + 1
        self.seen_signatures[signature] = hits
        if hits > 2:
            raise StopRun("loop_detected")

        tool = self.registry.get(name)
        if tool is None:
            raise StopRun(f"tool_missing:{name}")

        return tool(**args)

def run_support_agent(ticket_id: str) -> dict[str, Any]:
    gateway = ToolGateway(allow=ALLOWED_TOOLS, registry=TOOL_REGISTRY, budget=BUDGET)

    ticket_payload = gateway.call("tickets_get", {"ticket_id": ticket_id})
    ticket = ticket_payload["ticket"]

    risk_category, risk_reason = classify_ticket_risk(ticket)
    customer = gateway.call("customers_get", {"customer_id": ticket["customer_id"]})["customer"]

    if should_force_manual_review(risk_category=risk_category, customer=customer):
        handoff = build_handoff_note(ticket=ticket, risk_category=risk_category, risk_reason=risk_reason)
        gateway.call("tickets_add_internal_note", {"ticket_id": ticket_id, "note": handoff})
        artifact = gateway.call("artifacts_put", {"ticket_id": ticket_id, "kind": "handoff", "payload": handoff})
        gateway.call("audit_emit", {"event_type": "support.handoff.created", "details": {"ticket_id": ticket_id, "artifact_id": artifact["artifact_id"]}})
        return {
            "status": "success",
            "outcome": "handoff",
            "requires_human_approval": True,
            "send_allowed": False,
            "risk_category": risk_category,
        }

    kb = gateway.call("kb_search", {"query": ticket["subject"], "k": 3})
    policy = gateway.call("policy_search", {"query": "refund policy sla", "k": 3})

    safe_customer = redact_customer(customer)
    draft = generate_support_draft(ticket=ticket, customer=safe_customer, kb_matches=kb["matches"], policy_matches=policy["matches"])

    violations = []
    violations.extend(validate_no_commitments(draft["customer_reply"]))
    violations.extend(validate_citations(claims=draft["claims"], citations=draft["citations"]))

    if violations:
        return {"status": "blocked", "stop_reason": "unsafe_draft", "violations": violations}

    artifact = gateway.call("artifacts_put", {"ticket_id": ticket_id, "kind": "draft", "payload": draft})
    gateway.call("audit_emit", {"event_type": "support.draft.created", "details": {"ticket_id": ticket_id, "artifact_id": artifact["artifact_id"]}})

    return {
        "status": "success",
        "outcome": "draft_ready",
        "requires_human_approval": True,
        "send_allowed": False,
        "artifact_id": artifact["artifact_id"],
    }

REQUIRED_CITATION_KINDS = {"refund", "credit", "sla", "timeline"}


def validate_citations(
    *,
    claims: list[dict[str, Any]],
    citations: list[dict[str, Any]],
) -> list[str]:
    errors: list[str] = []
    cited_ids = {
        str(item.get("id", "")).strip()
        for item in citations
        if isinstance(item, dict) and str(item.get("id", "")).strip()
    }

    for claim in claims:
        kind = str(claim.get("kind", "")).strip().lower()
        citation_id = str(claim.get("citation_id", "")).strip()
        if kind in REQUIRED_CITATION_KINDS and not citation_id:
            errors.append(f"missing_citation:{kind}")
            continue
        if citation_id and citation_id not in cited_ids:
            errors.append(f"unknown_citation:{citation_id}")

    return errors

SYSTEM_PROMPT = """
You are a support drafting assistant.
Return only JSON with keys:
- customer_reply: string
- internal_note: string
- claims: [{"kind": string, "text": string, "citation_id": string}]
- citations: [{"id": string, "title": string}]

Rules:
- Do not promise refunds or credits.
- If policy claim is made, include citation_id.
- Keep customer_reply concise and professional.
""".strip()