Conversational AI Patterns.

from anthropic import Anthropic
client = Anthropic()

def build_system_prompt(case_facts: dict) -> str:
    return f"""You are a conversational support agent.

CASE_FACTS (immutable; re-read every turn; never paraphrased):
- customer_id: {case_facts['customer_id']}
- decision_made: {case_facts.get('decision_made', 'none')}
- contact_preference: {case_facts.get('contact_preference', 'unset')}
- escalation_requested: {case_facts.get('escalation_requested', False)}
- policy_cap: ${case_facts.get('cap', 500)}

Constraints:
- Branch on stop_reason. Never on response text.
- If escalation_requested is True: route to human queue, no negotiation.
- If a clarifying question was already answered, do not re-ask (state below)."""

def build_session_block(state: dict) -> str:
    return f"""SESSION_STATE (updated post-each-turn):
- clarifications_answered: {state.get('clarifications_answered', [])}
- last_tool: {state.get('last_tool')} → {state.get('last_tool_result_summary')}
- escalation_requested: {state.get('escalation_requested', False)}
- contact_preference: {state.get('contact_preference', 'email')}
"""

# After every tool call, update + persist
state['clarifications_answered'].append('order_id')
state['last_tool'] = 'lookup_order'
state['last_tool_result_summary'] = 'order_status=delivered'
session_store.save(case_facts['customer_id'], state)

# .claude/hooks/clarification_gate.py
import sys, json

def main():
    payload = json.loads(sys.stdin.read())
    tool_name = payload["tool_name"]
    case_facts = payload.get("case_facts", {})
    session = payload.get("session_state", {})

    # Account-modifying tools require verified identity
    if tool_name in ("process_refund", "update_account", "escalate_to_human"):
        if not case_facts.get("verified_id"):
            print("verified_id missing. Call get_customer first", file=sys.stderr)
            sys.exit(2)

    # Honor explicit escalation request. No further tool calls
    if session.get("escalation_requested") and tool_name != "escalate_to_human":
        print("user requested human; route to escalation queue, no other tools", file=sys.stderr)
        sys.exit(2)

    sys.exit(0)

if __name__ == "__main__":
    main()

def run_conversation_turn(user_msg: str, case_facts: dict, state: dict, max_iter: int = 12):
    messages = load_history(case_facts['customer_id']) + [{"role": "user", "content": user_msg}]
    system = build_system_prompt(case_facts) + "\n\n" + build_session_block(state)

    for _ in range(max_iter):
        resp = client.messages.create(
            model="claude-sonnet-4.5",
            max_tokens=2048,
            system=system,
            tools=tools,
            messages=messages,
        )
        if resp.stop_reason == "end_turn":
            persist_history(case_facts['customer_id'], messages, resp)
            return extract_text(resp)
        if resp.stop_reason == "tool_use":
            tool_uses = [b for b in resp.content if b.type == "tool_use"]
            results = [execute_tool(t, case_facts, state) for t in tool_uses]
            messages.append({"role": "assistant", "content": resp.content})
            messages.append({"role": "user", "content": results})
            update_state(state, tool_uses, results)
            continue
        if resp.stop_reason == "max_tokens":
            persist_partial(case_facts, state, resp)
            return {"status": "partial_escalate", "text": extract_text(resp)}
    return {"status": "iteration_cap"}

def compress_history(messages: list, threshold: int = 15) -> list:
    if len(messages) <= threshold:
        return messages

    # Preserve the original user message + the most recent exchange.
    first = messages[0]
    last = messages[-1]
    middle = messages[1:-1]

    # Summarize: extract decision points only (not full transcript)
    decisions = extract_decisions(middle)  # e.g. ["user_asked_for_refund", "agent_verified_customer", "policy_allowed_$50"]
    summary = "CONVERSATION SUMMARY (turns 2-" + str(len(messages) - 1) + "):\n" + "\n".join(f"- {d}" for d in decisions)

    return [first, {"role": "user", "content": summary}, last]

# Usage in the turn loop
messages = compress_history(messages, threshold=15)

# Detection lives in the agent's prompt; latching lives in state
EXPLICIT_HUMAN_PHRASES = [
    "speak to a human",
    "talk to a person",
    "i want a human",
    "transfer me",
    "give me a human",
]

def detect_explicit_handoff(user_msg: str) -> bool:
    msg = user_msg.lower()
    return any(phrase in msg for phrase in EXPLICIT_HUMAN_PHRASES)

def handle_user_turn(user_msg: str, case_facts: dict, state: dict):
    if detect_explicit_handoff(user_msg):
        state['escalation_requested'] = True
        # The next agent loop will see this in session_state and the hook
        # will block any tool except escalate_to_human.
    return run_conversation_turn(user_msg, case_facts, state)

# Sentiment is intentionally NOT consulted here.

# Split system into stable (cached) + dynamic (fresh) blocks
resp = client.messages.create(
    model="claude-sonnet-4.5",
    max_tokens=2048,
    system=[
        {
            "type": "text",
            "text": STABLE_SYSTEM_PREAMBLE,  # role + constraints, never changes
            "cache_control": {"type": "ephemeral"},
        },
        {
            "type": "text",
            "text": build_case_facts_block(case_facts) + build_session_block(state),  # changes per turn
        },
    ],
    tools=tools,  # tools array also auto-cached when stable
    messages=messages,
)
# Inspect resp.usage.cache_creation_input_tokens / cache_read_input_tokens to verify hit rate.

def audit_conversation(case_facts: dict, state: dict, agent_path: list, elapsed_ms: int, csat: int | None):
    db.audit.insert({
        "ts": datetime.utcnow(),
        "customer_id": case_facts["customer_id"],
        "turn_count": len(agent_path),
        "tool_calls": [c["name"] for c in agent_path if c.get("type") == "tool_use"],
        "stop_reasons": [c.get("stop_reason") for c in agent_path],
        "escalation_reason": state.get("escalation_reason"),
        "compression_fired_at_turn": state.get("compression_at_turn"),
        "elapsed_ms": elapsed_ms,
        "csat": csat,
    })