Seamless Human Handoff for an AI Insurance Agent with Parlant and Streamlit
Why human handoff matters
Automated customer support often handles the majority of routine requests, but some cases demand human judgment. Human handoff ensures that when the AI reaches its limits, a human operator can take over the conversation without losing context. This guide shows how to implement a human handoff system for an AI-powered insurance agent using Parlant, and how to build a Streamlit interface for Tier-2 operators to monitor and join live sessions.
Setup and dependencies
Make sure you have a valid OpenAI API key and store it in a .env file at your project’s root. Keep credentials out of code and version control.
OPENAI_API_KEY=your_api_key_here
Install the required Python packages:
pip install parlant dotenv streamlit
Agent implementation (agent.py)
The agent script defines the AI behavior, tools, conversation journeys, glossary, and the human handoff mechanism. Once the agent can escalate to manual mode, you can build a Streamlit UI to let human operators view and respond in active sessions.
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 Tools
These tools simulate actions the insurance assistant may perform: listing open claims, filing a claim, and returning policy details.
@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"
})
Initiating the human handoff
When the AI detects it can’t handle an issue, it can call a tool to switch the session to manual mode so a human can take control.
@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
}
)
Glossary and domain terms
Defining shared terms helps the agent respond consistently to common domain questions.
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",
)
Conversation journeys
Journeys encode conversational flows like filing a claim or explaining policy coverage, including when to call tools or escalate.
# ---------------------------
# 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 legal interpretation of coverage",
action="Politely explain that legal advice cannot be provided",
)
return journey
Main runner and server
This starts a local Parlant server, registers the agent, glossary, journeys, and guidelines, and defines the human handoff trigger.
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])
# Global 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())
Running the agent
Run:
python agent.py
Parlant will serve the agent locally (by default at http://localhost:8800) and manage sessions and events. Once the agent is running, you can connect with a UI to perform human handoffs.
Human handoff UI (handoff.py)
This Streamlit app connects to the running Parlant instance and lets a Tier-2 operator monitor sessions, view messages, and send replies either as a human or on behalf of the AI.
Importing libraries
import asyncio
import streamlit as st
from datetime import datetime
from parlant.client import AsyncParlantClient
Setting up the Parlant client
client = AsyncParlantClient(base_url="http://localhost:8800")
Session state management
Use Streamlit’s session_state to store events and the last offset so you can fetch new events incrementally.
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
Message rendering
A small helper formats messages differently depending on who sent them.
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}")
Fetching events
This async function polls Parlant for new message events and appends them to session_state.
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}")
Sending messages
Two helpers let the operator send messages either as a human operator or as if sent by the AI.
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
Streamlit interface
The UI allows entering a Parlant session ID, refreshing message history, and sending messages.
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))
This setup provides a straightforward way for Tier-2 operators to monitor live Parlant sessions and either take control or assist by sending messages on behalf of the AI. Combine this with robust logging, access controls, and operator training to support sensitive or complex insurance interactions.