Human handoff is a key part of customer support automation—it ensures that when AI reaches its limits, a talented human can seamlessly take over. On this tutorial, we’ll implement a human handoff system for an AI-powered insurance coverage agent utilizing Parlant. You’ll learn to create a Streamlit-based interface that enables a human operator (Tier 2) to view reside buyer messages and reply instantly throughout the similar session, bridging the hole between automation and human experience. Try the FULL CODES right here.
Organising the dependencies
Be sure you have a legitimate OpenAI API key earlier than beginning. When you’ve generated it out of your OpenAI dashboard, create a .env file in your challenge’s root listing and retailer the important thing securely there like this:
OPENAI_API_KEY=your_api_key_hereThis retains your credentials protected and prevents them from being hardcoded into your codebase.
pip set up parlant dotenv streamlitInsurance coverage 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. It will type the core logic that powers our insurance coverage assistant in Parlant. As soon as the agent is prepared and able to escalating to handbook mode, we’ll transfer on to creating the Streamlit-based human handoff interface, the place human operators can view ongoing classes, learn buyer messages, and reply in actual time — making a seamless collaboration between AI automation and human experience. Try the FULL CODES right here.
Loading the required libraries
import asyncio
import os
from datetime import datetime
from dotenv import load_dotenv
import parlant.sdk as p
load_dotenv()Defining the Agent’s Instruments
@p.instrument
async def get_open_claims(context: p.ToolContext) -> p.ToolResult:
return p.ToolResult(knowledge=["Claim #123 - Pending", "Claim #456 - Approved"])
@p.instrument
async def file_claim(context: p.ToolContext, claim_details: str) -> p.ToolResult:
return p.ToolResult(knowledge=f"New declare filed: {claim_details}")
@p.instrument
async def get_policy_details(context: p.ToolContext) -> p.ToolResult:
return p.ToolResult(knowledge={
"policy_number": "POL-7788",
"protection": "Covers unintentional injury 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 instrument represents an asynchronous perform that retrieves a listing of open insurance coverage claims, permitting the agent to offer customers with up-to-date details about pending or accepted claims.
- The file_claim instrument accepts declare particulars as enter and simulates the method of submitting a brand new insurance coverage declare, returning a affirmation message to the person.
Lastly, the get_policy_details instrument supplies important coverage data, such because the coverage quantity and protection limits, enabling the agent to reply precisely to questions on insurance coverage protection. Try the FULL CODES right here.
@p.instrument
async def initiate_human_handoff(context: p.ToolContext, motive: str) -> p.ToolResult:
"""
Provoke handoff to a human agent when the AI can't adequately assist the client.
"""
print(f"🚨 Initiating human handoff: {motive}")
# Setting session to handbook mode stops automated AI responses
return p.ToolResult(
knowledge=f"Human handoff initiated as a result of: {motive}",
management={
"mode": "handbook" # Swap session to handbook mode
}
)The initiate_human_handoff instrument permits the AI agent to gracefully switch a dialog to a human operator when it detects that the difficulty requires human intervention. By switching the session to handbook mode, it pauses all automated responses, making certain the human agent can take full management. This instrument helps keep a clean transition between AI and human help, making certain complicated 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 persistently. It helps keep accuracy and model alignment by giving the agent clear, predefined solutions for widespread domain-specific queries. Try the FULL CODES right here.
async def add_domain_glossary(agent: p.Agent):
await agent.create_term(
identify="Buyer Service Quantity",
description="You may attain us at +1-555-INSURE",
)
await agent.create_term(
identify="Working Hours",
description="We can be found Mon-Fri, 9AM-6PM",
)Defining the Journeys
# ---------------------------
# Declare Journey
# ---------------------------
async def create_claim_journey(agent: p.Agent) -> p.Journey:
journey = await agent.create_journey(
title="File an Insurance coverage Declare",
description="Helps prospects report and submit a brand new declare.",
situations=["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="Buyer supplies particulars")
s2 = await s1.goal.transition_to(chat_state="Affirm declare was submitted", situation="Declare efficiently created")
await s2.goal.transition_to(state=p.END_JOURNEY, situation="Buyer confirms submission")
return journey
# ---------------------------
# Coverage Journey
# ---------------------------
async def create_policy_journey(agent: p.Agent) -> p.Journey:
journey = await agent.create_journey(
title="Clarify Coverage Protection",
description="Retrieves and explains buyer's insurance coverage protection.",
situations=["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="Clarify the coverage protection clearly",
situation="Coverage information is accessible",
)
await agent.create_guideline(
situation="Buyer presses for authorized interpretation of protection",
motion="Politely clarify that authorized recommendation can't be supplied",
)
return journeyThe Declare Journey guides prospects via the method of submitting a brand new insurance coverage declare. It collects accident particulars, triggers the declare submitting instrument, confirms profitable submission, after which ends the journey—automating your entire declare initiation stream.
The Coverage Journey helps prospects perceive their insurance coverage protection by retrieving coverage particulars and explaining them clearly. It additionally features a guideline to make sure the AI avoids giving authorized interpretations, sustaining compliance and professionalism. Try the FULL CODES right here.
Defining the Primary Runner
async def most important():
async with p.Server() as server:
agent = await server.create_agent(
identify="Insurance coverage Help Agent",
description=(
"Pleasant Tier-1 AI assistant that helps with claims and coverage questions. "
"Escalates complicated 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(
"Buyer mentions a problem however would not specify if it is a declare or coverage"
)
await status_obs.disambiguate([claim_journey, policy_journey])
# International Tips
await agent.create_guideline(
situation="Buyer asks about unrelated subjects",
motion="Kindly redirect them to insurance-related help solely",
)
# Human Handoff Guideline
await agent.create_guideline(
situation="Buyer requests human help or AI is unsure concerning the subsequent step",
motion="Provoke human handoff and notify Tier-2 help.",
instruments=[initiate_human_handoff],
)
print("✅ Insurance coverage Help Agent with Human Handoff is prepared! Open the Parlant UI to talk.")
if __name__ == "__main__":
asyncio.run(most important())
Working the Agent
It will begin the Parlant agent regionally on http://localhost:8800 , the place it can deal with all dialog logic and session administration.
Within the subsequent step, we’ll join this operating agent to our Streamlit-based Human Handoff interface, permitting a human operator to seamlessly be a part of and handle reside conversations utilizing the Parlant session ID. Try the FULL CODES right here.
Human Handoff (handoff.py)
Importing Libraries
import asyncio
import streamlit as st
from datetime import datetime
from parlant.consumer import AsyncParlantClientSetting Up the Parlant Shopper
As soon as the AI agent script is operating, Parlant will host its server regionally (normally at http://localhost:8800).
Right here, we hook up with that operating occasion by creating an asynchronous consumer. Try the FULL CODES right here.
consumer = AsyncParlantClient(base_url="http://localhost:8800")Whenever 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 Administration
Streamlit’s session_state is used to persist knowledge throughout person interactions — akin to storing obtained messages and monitoring the most recent occasion offset to fetch new ones effectively. Try the FULL CODES right here.
if "occasions" not in st.session_state:
st.session_state.occasions = []
if "last_offset" not in st.session_state:
st.session_state.last_offset = 0Message Rendering Perform
This perform controls how messages seem within the Streamlit interface — differentiating between prospects, AI, and human brokers for readability. Try the FULL CODES right here.
def render_message(message, supply, participant_name, timestamp):
if supply == "buyer":
st.markdown(f"**🧍♂️ Buyer [{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 Occasions from Parlant
This asynchronous perform retrieves new messages (occasions) from Parlant for the given session.
Every occasion represents a message within the dialog — whether or not despatched by the client, AI, or human operator. Try the FULL CODES right here.
async def fetch_events(session_id):
strive:
occasions = await consumer.classes.list_events(
session_id=session_id,
varieties="message",
min_offset=st.session_state.last_offset,
wait_for_data=5
)
for occasion in occasions:
message = occasion.knowledge.get("message")
supply = occasion.supply
participant_name = occasion.knowledge.get("participant", {}).get("display_name", "Unknown")
timestamp = getattr(occasion, "created", None) or occasion.knowledge.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”)
- One other as if despatched by the AI, however manually triggered by a human (supply=”human_agent_on_behalf_of_ai_agent”)
- Try the FULL CODES right here.
async def send_human_message(session_id: str, message: str, operator_name: str = "Tier-2 Operator"):
occasion = await consumer.classes.create_event(
session_id=session_id,
sort="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.classes.create_event(
session_id=session_id,
sort="message",
supply="human_agent_on_behalf_of_ai_agent",
message=message
)
return occasion
Streamlit Interface
Lastly, we construct a easy, interactive Streamlit UI:
- Enter a session ID (from the Parlant UI)
- View chat historical past
- Ship messages as both Human or AI
- Refresh to drag new messages
- Try the FULL CODES right here.
st.title("💼 Human Handoff Assistant")
session_id = st.text_input("Enter Parlant Session ID:")
if session_id:
st.subheader("Chat Historical past")
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("Ship a Message")
operator_msg = st.text_input("Kind your message:")
if st.button("Ship 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("Ship 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))
Try the FULL CODES right here. Be happy to take a look at our GitHub Web page for Tutorials, Codes and Notebooks. Additionally, be at liberty to comply with us on Twitter and don’t overlook to affix our 100k+ ML SubReddit and Subscribe to our Publication. Wait! are you on telegram? now you’ll be able to be a part of us on telegram as effectively.
I’m a Civil Engineering Graduate (2022) from Jamia Millia Islamia, New Delhi, and I’ve a eager curiosity in Knowledge Science, particularly Neural Networks and their utility in numerous areas.
