|

Building a Human Handoff Interface for AI-Powered Insurance Agent Using Parlant and Streamlit

ByRicardo
🚨

Human handoff is a key element of customer support automation—it ensures that when AI reaches its limits, a expert human can seamlessly take over. In this tutorial, we’ll implement a human handoff system for an AI-powered insurance coverage agent utilizing Parlant. You’ll discover ways to create a Streamlit-based interface that permits a human operator (Tier 2) to view stay buyer messages and reply straight inside the identical session, bridging the hole between automation and human experience. Check out the FULL CODES here.

Setting up the dependencies

Make positive you may have a legitimate OpenAI API key earlier than beginning. Once you’ve generated it out of your OpenAI dashboard, create a .env file in your mission’s root listing and retailer the important thing securely there like this:

Copy Code
OPENAI_API_KEY=your_api_key_here

This retains your credentials secure and prevents them from being hardcoded into your codebase.

Copy Code
pip set up parlant dotenv streamlit

Insurance Agent (agent.py) 

We’ll begin by constructing the agent script, which defines the AI’s habits, dialog journeys, glossary, and the human handoff mechanism. This will type the core logic that powers our insurance coverage assistant in Parlant. Once the agent is prepared and able to escalating to guide mode, we’ll transfer on to creating the Streamlit-based human handoff interface, the place human operators can view ongoing periods, learn buyer messages, and reply in actual time — creating a seamless collaboration between AI automation and human experience. Check out the FULL CODES here.

Loading the required libraries

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

load_dotenv()

Defining the Agent’s Tools

Copy Code
@p.software
async def get_open_claims(context: p.ToolContext) -> p.ToolResult:
    return p.ToolResult(information=["Claim #123 - Pending", "Claim #456 - Approved"])

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

@p.software
async def get_policy_details(context: p.ToolContext) -> p.ToolResult:
    return p.ToolResult(information={
        "policy_number": "POL-7788",
        "protection": "Covers unintended harm and theft as much as $50,000"
    })

The code block introduces three instruments that simulate interactions an insurance coverage assistant may want. 

  • The get_open_claims software represents an asynchronous perform that retrieves a checklist of open insurance coverage claims, permitting the agent to offer customers with up-to-date details about pending or authorised claims. 
  • The file_claim software accepts declare particulars as enter and simulates the method of submitting a new insurance coverage declare, returning a affirmation message to the consumer. 

Finally, the get_policy_details software supplies important coverage info, such because the coverage quantity and protection limits, enabling the agent to reply precisely to questions on insurance coverage protection. Check out the FULL CODES here.

Copy Code
@p.software
async def initiate_human_handoff(context: p.ToolContext, purpose: str) -> p.ToolResult:
    """
    Initiate handoff to a human agent when the AI can not adequately assist the client.
    """
    print(f"🚨 Initiating human handoff: {purpose}")
    # Setting session to guide mode stops automated AI responses
    return p.ToolResult(
        information=f"Human handoff initiated as a result of: {purpose}",
        management={
            "mode": "guide"  # Switch session to guide mode
        }
    )

The initiate_human_handoff software permits the AI agent to gracefully switch a dialog to a human operator when it detects that the problem requires human intervention. By switching the session to guide mode, it pauses all automated responses, guaranteeing the human agent can take full management. This software helps preserve a easy transition between AI and human help, guaranteeing advanced or delicate buyer queries are dealt with with the suitable degree of experience.

Defining the Glossary

A glossary defines key phrases and phrases that the AI agent ought to acknowledge and reply to constantly. It helps preserve accuracy and model alignment by giving the agent clear, predefined solutions for frequent domain-specific queries. Check out the FULL CODES here.

Copy Code
async def add_domain_glossary(agent: p.Agent):
    await agent.create_term(
        identify="Customer Service Number",
        description="You can attain us at +1-555-INSURE",
    )
    await agent.create_term(
        identify="Operating Hours",
        description="We can be found Mon-Fri, 9AM-6PM",
    )

Defining the Journeys

Copy Code
# ---------------------------
# Claim Journey
# ---------------------------

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

    s0 = await journey.initial_state.transition_to(chat_state="Ask for accident particulars")
    s1 = await s0.goal.transition_to(tool_state=file_claim, situation="Customer supplies particulars")
    s2 = await s1.goal.transition_to(chat_state="Confirm declare was submitted", situation="Claim efficiently created")
    await s2.goal.transition_to(state=p.END_JOURNEY, situation="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 buyer's insurance coverage protection.",
        circumstances=["The customer asks about their policy"],
    )

    s0 = await journey.initial_state.transition_to(tool_state=get_policy_details)
    await s0.goal.transition_to(
        chat_state="Explain the coverage protection clearly",
        situation="Policy data is accessible",
    )

    await agent.create_guideline(
        situation="Customer presses for authorized interpretation of protection",
        motion="Politely clarify that authorized recommendation can't be supplied",
    )
    return journey

The Claim Journey guides clients by the method of submitting a new insurance coverage declare. It collects accident particulars, triggers the declare submitting software, confirms profitable submission, and then ends the journey—automating all the declare initiation move.

The Policy Journey helps clients perceive their insurance coverage protection by retrieving coverage particulars and explaining them clearly. It additionally contains a guideline to make sure the AI avoids giving authorized interpretations, sustaining compliance and professionalism. Check out the FULL CODES here.

Defining the Main Runner

Copy Code
async def important():
    async with p.Server() as server:
        agent = await server.create_agent(
            identify="Insurance Support Agent",
            description=(
                "Friendly Tier-1 AI assistant that helps with claims and coverage questions. "
                "Escalates advanced or unresolved points to human brokers (Tier-2)."
            ),
        )

        # Add shared phrases & 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 a problem however does not specify if it is a declare or coverage"
        )
        await status_obs.disambiguate([claim_journey, policy_journey])

        # Global Guidelines
        await agent.create_guideline(
            situation="Customer asks about unrelated matters",
            motion="Kindly redirect them to insurance-related help solely",
        )

        # Human Handoff Guideline
        await agent.create_guideline(
            situation="Customer requests human help or AI is unsure in regards to the subsequent step",
            motion="Initiate human handoff and notify Tier-2 help.",
            instruments=[initiate_human_handoff],
        )

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

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

Running the Agent

Copy Code
python agent.py

This will begin the Parlant agent domestically on http://localhost:8800 , the place it is going to deal with all dialog logic and session administration.

In the following step, we’ll join this operating agent to our Streamlit-based Human Handoff interface, permitting a human operator to seamlessly be part of and handle stay conversations utilizing the Parlant session ID. Check out the FULL CODES here.

Human Handoff (handoff.py) 

Importing Libraries

Copy Code
import asyncio
import streamlit as st
from datetime import datetime
from parlant.consumer import AsyncParlantClient

Setting Up the Parlant Client

Once the AI agent script is operating, Parlant will host its server domestically (often at http://localhost:8800).

Here, we hook up with that operating occasion by creating an asynchronous consumer. Check out the FULL CODES here.

Copy Code
consumer = AsyncParlantClient(base_url="http://localhost:8800")

When you run the agent and get a session ID, we’ll use that ID on this UI to attach and handle that particular dialog.

Session State Management

Streamlit’s session_state is used to persist information throughout consumer interactions — corresponding to storing obtained messages and monitoring the newest occasion offset to fetch new ones effectively. Check out the FULL CODES here.

Copy Code
if "occasions" not in st.session_state:
    st.session_state.occasions = []
if "last_offset" not in st.session_state:
    st.session_state.last_offset = 0

Message Rendering Function

This perform controls how messages seem within the Streamlit interface — differentiating between clients, AI, and human brokers for readability. Check out the FULL CODES here.

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

Fetching Events from Parlant

This asynchronous perform retrieves new messages (occasions) from Parlant for the given session.

Each occasion represents a message within the dialog — whether or not despatched by the client, AI, or human operator. Check out the FULL CODES here.

Copy Code
async def fetch_events(session_id):
    strive:
        occasions = await consumer.periods.list_events(
            session_id=session_id,
            sorts="message",
            min_offset=st.session_state.last_offset,
            wait_for_data=5
        )
        for occasion in occasions:
            message = occasion.information.get("message")
            supply = occasion.supply
            participant_name = occasion.information.get("participant", {}).get("display_name", "Unknown")
            timestamp = getattr(occasion, "created", None) or occasion.information.get("created", "Unknown Time")
            event_id = getattr(occasion, "id", "Unknown ID")

            st.session_state.occasions.append(
                (message, supply, participant_name, timestamp, event_id)
            )
            st.session_state.last_offset = max(st.session_state.last_offset, occasion.offset + 1)

    besides Exception as e:
        st.error(f"Error fetching occasions: {e}")

Sending Messages as Human or AI

Two helper features are outlined to ship messages:

  • One as a human operator (supply=”human_agent”)
  • Another as if despatched by the AI, however manually triggered by a human (supply=”human_agent_on_behalf_of_ai_agent”)
  • Check out the FULL CODES here.
Copy Code

async def send_human_message(session_id: str, message: str, operator_name: str = "Tier-2 Operator"):
    occasion = await consumer.periods.create_event(
        session_id=session_id,
        variety="message",
        supply="human_agent",
        message=message,
        participant={
            "id": "operator-001",
            "display_name": operator_name
        }
    )
    return occasion


async def send_message_as_ai(session_id: str, message: str):
    occasion = await consumer.periods.create_event(
        session_id=session_id,
        variety="message",
        supply="human_agent_on_behalf_of_ai_agent",
        message=message
    )
    return occasion

Streamlit Interface

Finally, we construct a easy, interactive Streamlit UI:

  • Enter a session ID (from the Parlant UI)
  • View chat historical past
  • Send messages as both Human or AI
  • Refresh to drag new messages
  • Check out the FULL CODES here.
Copy Code
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, supply, participant_name, timestamp, event_id in st.session_state.occasions:
        render_message(msg, supply, 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 despatched 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 despatched as AI ✅")
            asyncio.run(fetch_events(session_id))

Check out the FULL CODES here. Feel free to take a look at our GitHub Page for Tutorials, Codes and Notebooks. Also, be at liberty to comply with us on Twitter and don’t overlook to hitch our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.

The put up Building a Human Handoff Interface for AI-Powered Insurance Agent Using Parlant and Streamlit appeared first on MarkTechPost.

Similar Posts