โพรโทคอลการส่งต่องานให้มนุษย์

import openai, json

client = openai.OpenAI(api_key='YOUR_OPENAI_KEY')

def should_handoff(message: str, history: list[dict]) -> dict:
    context = '\n'.join(f"{m['role']}: {m['content']}" for m in history[-4:])
    prompt = (
        f'Conversation context:\n{context}\n'
        f'Latest message: "{message}"\n'
        f'Should this be handed to a human agent? Reasons: '
        f'angry_customer, explicit_human_request, complex_issue, legal_risk, other.\n'
        f'JSON: {{"handoff": true/false, "reason": "..."}}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    return json.loads(resp.choices[0].message.content)

def generate_handoff_summary(history: list[dict]) -> str:
    transcript = '\n'.join(
        f"{m['role'].upper()}: {m['content']}" for m in history
    )
    prompt = (
        f'Summarize this support conversation for a human agent.\n'
        f'Include: customer issue, what was tried, current status, and urgency level.\n'
        f'Be brief (3-5 sentences).\n\n'
        f'TRANSCRIPT:\n{transcript}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}]
    )
    return resp.choices[0].message.content

summary = generate_handoff_summary([
    {'role': 'user', 'content': 'My account was charged twice for January'},
    {'role': 'assistant', 'content': 'I can look into that for you...'}
])
print(summary)

import requests

ZENDESK_DOMAIN = 'yourcompany.zendesk.com'
ZENDESK_TOKEN = 'YOUR_ZENDESK_API_TOKEN'
ZENDESK_EMAIL = 'agent@yourcompany.com'

def create_zendesk_ticket(customer_email: str, subject: str,
                          summary: str, transcript: str) -> str:
    payload = {
        'ticket': {
            'subject': subject,
            'comment': {'body': f'AI Agent Summary:\n{summary}\n\nFull Transcript:\n{transcript}'},
            'requester': {'email': customer_email},
            'tags': ['ai_handoff'],
            'priority': 'high'
        }
    }
    resp = requests.post(
        f'https://{ZENDESK_DOMAIN}/api/v2/tickets.json',
        json=payload,
        auth=(f'{ZENDESK_EMAIL}/token', ZENDESK_TOKEN)
    )
    resp.raise_for_status()
    return str(resp.json()['ticket']['id'])

import requests

INTERCOM_TOKEN = 'YOUR_INTERCOM_ACCESS_TOKEN'

def handoff_to_intercom_team(conversation_id: str, team_id: str,
                             note: str) -> bool:
    headers = {
        'Authorization': f'Bearer {INTERCOM_TOKEN}',
        'Content-Type': 'application/json'
    }
    # Add a note with the AI summary
    requests.post(
        f'https://api.intercom.io/conversations/{conversation_id}/parts',
        headers=headers,
        json={'type': 'note', 'body': note}
    )
    # Assign to human team
    resp = requests.put(
        f'https://api.intercom.io/conversations/{conversation_id}/parts',
        headers=headers,
        json={'type': 'assignment', 'assignee_id': team_id,
              'message_type': 'assignment'}
    )
    return resp.status_code == 200

def generate_handoff_message(reason: str, wait_minutes: int = 5) -> str:
    messages = {
        'explicit_human_request':
            f'Of course! I am connecting you with a human agent right now. '
            f'Estimated wait: {wait_minutes} minutes. Your conversation history '
            f'has been shared so you will not need to repeat anything.',
        'angry_customer':
            f'I completely understand your frustration. Let me get a senior '
            f'team member on the line immediately. Wait: ~{wait_minutes} min.',
        'complex_issue':
            f'This situation needs specialist attention. I am escalating now '
            f'and sharing all the context we have discussed. Wait: ~{wait_minutes} min.',
        'legal_risk':
            f'This matter requires our compliance team. Connecting you now.'
    }
    return messages.get(reason, f'Connecting you with a human agent. ~{wait_minutes} min wait.')

if __name__ == '__main__':
    print(generate_handoff_message('angry_customer', wait_minutes=3))
    print(generate_handoff_message('explicit_human_request'))

import requests

SLACK_WEBHOOK = 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'

def notify_on_call(ticket_id: str, summary: str, priority: str):
    payload = {
        'text': f'*New AI Handoff* [{priority.upper()}]',
        'attachments': [{
            'color': '#ff0000' if priority == 'high' else '#ffcc00',
            'fields': [
                {'title': 'Ticket ID', 'value': ticket_id, 'short': True},
                {'title': 'Summary', 'value': summary[:500]}
            ]
        }]
    }
    requests.post(SLACK_WEBHOOK, json=payload)

def agent_post_handoff_response(message: str, handoff_complete: bool) -> str:
    if not handoff_complete:
        return 'Connecting you now...'

    # Still answer simple factual questions
    simple_keywords = ['status', 'where', 'when', 'policy', 'link']
    if any(kw in message.lower() for kw in simple_keywords):
        return 'I can help with that while you wait for the agent.'

    # Defer everything else
    return (
        'Your case has been assigned to a specialist. '
        'They will respond shortly. I will step back to avoid confusion.'
    )

if __name__ == '__main__':
    print(agent_post_handoff_response('Where is my order?', handoff_complete=True))
    print(agent_post_handoff_response('I want a refund now', handoff_complete=True))

from collections import Counter
import json

handoff_log = []  # In production: a database table

def record_handoff(session_id: str, reason: str, turn_number: int):
    handoff_log.append({
        'session_id': session_id,
        'reason': reason,
        'turns_before_handoff': turn_number
    })

def handoff_analytics() -> dict:
    reasons = Counter(h['reason'] for h in handoff_log)
    avg_turns = sum(h['turns_before_handoff'] for h in handoff_log) / max(len(handoff_log), 1)
    return {
        'total_handoffs': len(handoff_log),
        'reason_breakdown': dict(reasons),
        'avg_turns_before_handoff': round(avg_turns, 1)
    }

if __name__ == '__main__':
    record_handoff('s1', 'angry_customer', 4)
    record_handoff('s2', 'complex_issue', 7)
    record_handoff('s3', 'angry_customer', 2)
    stats = handoff_analytics()
    print(f"Total handoffs: {stats['total_handoffs']}")
    print(f"Reasons: {stats['reason_breakdown']}")
    print(f"Avg turns before handoff: {stats['avg_turns_before_handoff']}")

def execute_handoff(session: dict, reason: str) -> str:
    # 1. Generate summary
    summary = generate_handoff_summary(session['history'])
    transcript = '\n'.join(
        f"{m['role']}: {m['content']}" for m in session['history']
    )
    # 2. Create ticket
    ticket_id = create_zendesk_ticket(
        session['customer_email'],
        f'AI Handoff: {reason}',
        summary,
        transcript
    )
    # 3. Notify on-call team
    notify_on_call(ticket_id, summary, priority='high')
    # 4. Record metrics
    record_handoff(session['id'], reason, len(session['history']))
    # 5. Return customer-facing message
    wait = 5  # fetch from queue depth in production
    return generate_handoff_message(reason, wait)