Construir una interfaz de transferencia humana para un agente de seguros con IA utilizando Parlant y Strewlit

OPENAI_API_KEY=your_api_key_here

pip install parlant dotenv streamlit

import asyncio
import os
from datetime import datetime
from dotenv import load_dotenv
import parlant.sdk as p

load_dotenv()

@p.tool
async def get_open_claims(context: p.ToolContext) -> p.ToolResult:
    return p.ToolResult(data=("Claim #123 - Pending", "Claim #456 - Approved"))

@p.tool
async def file_claim(context: p.ToolContext, claim_details: str) -> p.ToolResult:
    return p.ToolResult(data=f"New claim filed: {claim_details}")

@p.tool
async def get_policy_details(context: p.ToolContext) -> p.ToolResult:
    return p.ToolResult(data={
        "policy_number": "POL-7788",
        "coverage": "Covers accidental damage and theft up to $50,000"
    })

@p.tool
async def initiate_human_handoff(context: p.ToolContext, reason: str) -> p.ToolResult:
    """
    Initiate handoff to a human agent when the AI cannot adequately help the customer.
    """
    print(f"🚨 Initiating human handoff: {reason}")
    # Setting session to manual mode stops automatic AI responses
    return p.ToolResult(
        data=f"Human handoff initiated because: {reason}",
        control={
            "mode": "manual"  # Switch session to manual mode
        }
    )

async def add_domain_glossary(agent: p.Agent):
    await agent.create_term(
        name="Customer Service Number",
        description="You can reach us at +1-555-INSURE",
    )
    await agent.create_term(
        name="Operating Hours",
        description="We are available Mon-Fri, 9AM-6PM",
    )

# ---------------------------
# Claim Journey
# ---------------------------

async def create_claim_journey(agent: p.Agent) -> p.Journey:
    journey = await agent.create_journey(
        title="File an Insurance Claim",
        description="Helps customers report and submit a new claim.",
        conditions=("The customer wants to file a claim"),
    )

    s0 = await journey.initial_state.transition_to(chat_state="Ask for accident details")
    s1 = await s0.target.transition_to(tool_state=file_claim, condition="Customer provides details")
    s2 = await s1.target.transition_to(chat_state="Confirm claim was submitted", condition="Claim successfully created")
    await s2.target.transition_to(state=p.END_JOURNEY, condition="Customer confirms submission")

    return journey

# ---------------------------
# Policy Journey
# ---------------------------

async def create_policy_journey(agent: p.Agent) -> p.Journey:
    journey = await agent.create_journey(
        title="Explain Policy Coverage",
        description="Retrieves and explains customer's insurance coverage.",
        conditions=("The customer asks about their policy"),
    )

    s0 = await journey.initial_state.transition_to(tool_state=get_policy_details)
    await s0.target.transition_to(
        chat_state="Explain the policy coverage clearly",
        condition="Policy info is available",
    )

    await agent.create_guideline(
        condition="Customer presses for judicial interpretation of coverage",
        action="Politely explain that judicial advice cannot be provided",
    )
    return journey

async def main():
    async with p.Server() as server:
        agent = await server.create_agent(
            name="Insurance Support Agent",
            description=(
                "Friendly Tier-1 AI assistant that helps with claims and policy questions. "
                "Escalates complex or unresolved issues to human agents (Tier-2)."
            ),
        )

        # Add shared terms & definitions
        await add_domain_glossary(agent)

        # Journeys
        claim_journey = await create_claim_journey(agent)
        policy_journey = await create_policy_journey(agent)

        # Disambiguation rule
        status_obs = await agent.create_observation(
            "Customer mentions an issue but doesn't specify if it's a claim or policy"
        )
        await status_obs.disambiguate((claim_journey, policy_journey))

        # Integral Guidelines
        await agent.create_guideline(
            condition="Customer asks about unrelated topics",
            action="Kindly redirect them to insurance-related support only",
        )

        # Human Handoff Guideline
        await agent.create_guideline(
            condition="Customer requests human assistance or AI is uncertain about the next step",
            action="Initiate human handoff and notify Tier-2 support.",
            tools=(initiate_human_handoff),
        )

        print("✅ Insurance Support Agent with Human Handoff is ready! Open the Parlant UI to chat.")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import streamlit as st
from datetime import datetime
from parlant.client import AsyncParlantClient

client = AsyncParlantClient(base_url="http://localhost:8800")

if "events" not in st.session_state:
    st.session_state.events = ()
if "last_offset" not in st.session_state:
    st.session_state.last_offset = 0

def render_message(message, source, participant_name, timestamp):
    if source == "customer":
        st.markdown(f"**🧍‍♂️ Customer ({timestamp}):** {message}")
    elif source == "ai_agent":
        st.markdown(f"**🤖 AI ({timestamp}):** {message}")
    elif source == "human_agent":
        st.markdown(f"**🙋 {participant_name} ({timestamp}):** {message}")
    elif source == "human_agent_on_behalf_of_ai_agent":
        st.markdown(f"**👤 (Human as AI) ({timestamp}):** {message}")

async def fetch_events(session_id):
    try:
        events = await client.sessions.list_events(
            session_id=session_id,
            kinds="message",
            min_offset=st.session_state.last_offset,
            wait_for_data=5
        )
        for event in events:
            message = event.data.get("message")
            source = event.source
            participant_name = event.data.get("participant", {}).get("display_name", "Unknown")
            timestamp = getattr(event, "created", None) or event.data.get("created", "Unknown Time")
            event_id = getattr(event, "id", "Unknown ID")

            st.session_state.events.append(
                (message, source, participant_name, timestamp, event_id)
            )
            st.session_state.last_offset = max(st.session_state.last_offset, event.offset + 1)

    except Exception as e:
        st.error(f"Error fetching events: {e}")

async def send_human_message(session_id: str, message: str, operator_name: str = "Tier-2 Operator"):
    event = await client.sessions.create_event(
        session_id=session_id,
        kind="message",
        source="human_agent",
        message=message,
        participant={
            "id": "operator-001",
            "display_name": operator_name
        }
    )
    return event


async def send_message_as_ai(session_id: str, message: str):
    event = await client.sessions.create_event(
        session_id=session_id,
        kind="message",
        source="human_agent_on_behalf_of_ai_agent",
        message=message
    )
    return event

st.title("💼 Human Handoff Assistant")

session_id = st.text_input("Enter Parlant Session ID:")

if session_id:
    st.subheader("Chat History")
    if st.button("Refresh Messages"):
        asyncio.run(fetch_events(session_id))

    for msg, source, participant_name, timestamp, event_id in st.session_state.events:
        render_message(msg, source, participant_name, timestamp)

    st.subheader("Send a Message")
    operator_msg = st.text_input("Type your message:")

    if st.button("Send as Human"):
        if operator_msg.strip():
            asyncio.run(send_human_message(session_id, operator_msg))
            st.success("Message sent as human agent ✅")
            asyncio.run(fetch_events(session_id))

    if st.button("Send as AI"):
        if operator_msg.strip():
            asyncio.run(send_message_as_ai(session_id, operator_msg))
            st.success("Message sent as AI ✅")
            asyncio.run(fetch_events(session_id))