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 discover ways to create a Streamlit-based interface that permits a human operator (Tier 2) to view stay buyer messages and reply instantly inside the similar session, bridging the hole between automation and human experience. Try the FULL CODES here.
Organising the dependencies
Be sure you have a sound OpenAI API key earlier than beginning. When 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:
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 conduct, 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 guide mode, we’ll transfer on to growing 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 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(information=["Claim #123 - Pending", "Claim #456 - Approved"])
@p.instrument
async def file_claim(context: p.ToolContext, claim_details: str) -> p.ToolResult:
return p.ToolResult(information=f"New declare filed: {claim_details}")
@p.instrument
async def get_policy_details(context: p.ToolContext) -> p.ToolResult:
return p.ToolResult(information={
"policy_number": "POL-7788",
"protection": "Covers unintentional 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 instrument represents an asynchronous operate that retrieves an inventory of open insurance coverage claims, permitting the agent to supply customers with up-to-date details about pending or permitted 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 consumer.
Lastly, the get_policy_details instrument offers important coverage info, such because the coverage quantity and protection limits, enabling the agent to reply precisely to questions on insurance coverage protection. Try the FULL CODES 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 not adequately assist the shopper.
"""
print(f"🚨 Initiating human handoff: {motive}")
# Setting session to guide mode stops computerized AI responses
return p.ToolResult(
information=f"Human handoff initiated as a result of: {motive}",
management={
"mode": "guide" # Change session to guide mode
}
)The initiate_human_handoff instrument allows 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 guide 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 stage 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 keep accuracy and model alignment by giving the agent clear, predefined solutions for frequent domain-specific queries. Try the FULL CODES 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 offers particulars")
s2 = await s1.goal.transition_to(chat_state="Verify 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 data is obtainable",
)
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 by way of 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 circulate.
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 here.
Defining the Primary Runner
async def primary():
async with p.Server() as server:
agent = await server.create_agent(
identify="Insurance coverage Assist 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 difficulty however would not specify if it is a declare or coverage"
)
await status_obs.disambiguate([claim_journey, policy_journey])
# World 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 Assist Agent with Human Handoff is prepared! Open the Parlant UI to talk.")
if __name__ == "__main__":
asyncio.run(primary())
Working the Agent
It will begin the Parlant agent domestically on http://localhost:8800 , the place it should 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 part of and handle stay conversations utilizing the Parlant session ID. Try the FULL CODES here.
Human Handoff (handoff.py)
Importing Libraries
import asyncio
import streamlit as st
from datetime import datetime
from parlant.shopper import AsyncParlantClientSetting Up the Parlant Consumer
As soon as the AI agent script is operating, Parlant will host its server domestically (normally at http://localhost:8800).
Right here, we connect with that operating occasion by creating an asynchronous shopper. Try the FULL CODES here.
shopper = AsyncParlantClient(base_url="http://localhost:8800")While 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 information throughout consumer interactions — reminiscent of storing acquired messages and monitoring the newest occasion offset to fetch new ones effectively. Try the FULL CODES 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 operate controls how messages seem within the Streamlit interface — differentiating between prospects, AI, and human brokers for readability. Try the FULL CODES 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 operate retrieves new messages (occasions) from Parlant for the given session.
Every occasion represents a message within the dialog — whether or not despatched by the shopper, AI, or human operator. Try the FULL CODES here.
async def fetch_events(session_id):
attempt:
occasions = await shopper.classes.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 capabilities 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 here.
async def send_human_message(session_id: str, message: str, operator_name: str = "Tier-2 Operator"):
occasion = await shopper.classes.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 shopper.classes.create_event(
session_id=session_id,
variety="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 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 here. Be at liberty to take a look at our GitHub Page for Tutorials, Codes and Notebooks. Additionally, be happy to observe 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.
I’m a Civil Engineering Graduate (2022) from Jamia Millia Islamia, New Delhi, and I’ve a eager curiosity in Information Science, particularly Neural Networks and their utility in numerous areas.