Preview
Open Original
This is an end-to-end project on building a multi-agent insurance support system using Agentic AI [LangGraph and OpenAI API]. [Code Included]
35 min read2 hours ago
–
Press enter or click to view image in full size
Multi Agent System Architecture [Image by Author]
Analogy
I recently called my insurance company’s customer care number to know the premium amount for my policy. They said: “Let me transfer your call to the billing department.”
Then one executive from the billing department resolved my query.
A few days later, I called them again and asked about the status of my claim. They said: “Let me transfer your call to the claims department.”
Then one executive from the claims department resolved my query.
So basically, for each category of queries, they …
This is an end-to-end project on building a multi-agent insurance support system using Agentic AI [LangGraph and OpenAI API]. [Code Included]
35 min read2 hours ago
–
Press enter or click to view image in full size
Multi Agent System Architecture [Image by Author]
Analogy
I recently called my insurance company’s customer care number to know the premium amount for my policy. They said: “Let me transfer your call to the billing department.”
Then one executive from the billing department resolved my query.
A few days later, I called them again and asked about the status of my claim. They said: “Let me transfer your call to the claims department.”
Then one executive from the claims department resolved my query.
So basically, for each category of queries, they have expert executives who specialize in resolving those specific issues.
That’s exactly how a multi-agent system works — different agents are designed for different kinds of tasks, and a central system routes the query to the right expert.
In short: Each agent does what it’s best at, and together they deliver a smarter, faster solution.
Press enter or click to view image in full size
Insurance Support System (Image by Author : Generated by Gemini)
Problem Statement:
Traditional customer support in the insurance industry is slow, fragmented, and inefficient. Customers often face:
- Long wait times and multiple handoffs
- Repetitive data requests (“Can you confirm your policy number again?”)
- Generic chatbot responses that lack context or accuracy
- Difficulty accessing personalized policy, billing, or claim information in real-time
Even with automation, most systems rely on single-intent chatbots that cannot handle the complex, multi-domain nature of insurance inquiries. This results in frustrated customers, overloaded support teams, and missed efficiency opportunities.
Solution Approach:
We built an AI-powered multi-agent insurance support system that mimics a team of human specialists using LangGraph orchestration and RAG (Retrieval-Augmented Generation).
A Supervisor Agent analyzes user intent and routes each query to specialized agents for policy, billing, claims, or general help. These agents access real data via SQLite and ChromaDB, ensuring factual, context-aware responses.
By managing conversation state, routing logic, and data retrieval intelligently, the system delivers accurate, personalized, and scalable customer support — while seamlessly escalating complex cases to humans when needed.
In this article, I’ll walk you through building a sophisticated multi-agent insurance support system that intelligently routes customer queries to specialized agents, retrieves information from databases, and leverages RAG (Retrieval-Augmented Generation) for accurate responses.
Refer to this Github Repo for complete code.
What We Are Building
Our system features:
1. Supervisor Agent — The orchestrator that analyzes user intent and routes to appropriate specialists
2. Policy Agent — Handles policy details, coverage, and auto insurance specifics
3. Billing Agent— Manages billing inquiries, payment history, and invoice information
4. Claims Agent — Processes claim status and filing assistance
5. General Help Agent — Answers FAQs using RAG with a vector database
6. Human Escalation Agent — Gracefully hands off complex cases to human representatives
The system uses **LangGraph** for workflow orchestration, **ChromaDB** for semantic search, **SQLite** for structured data, and **OpenAI GPT** for natural language understanding.
System Architecture
The workflow functions as a state machine, where:
- Each agent is a node in the graph.
- The supervisor acts as the central router, making intelligent decisions based on the conversation.
- Specialist agents (Policy, Billing, etc.) perform their tasks and return control to the supervisor.
- The conversation ends when the query is resolved or successfully escalated to a human.
Press enter or click to view image in full size
Insurance Support System: Multi Agent Architecture (Image by Author)
Technical Stack
- **Orchestration: **
LangGraph - LLM: OpenAI
gpt-5-mini - Vector Database:
ChromaDB - Relational Database:
SQLite - Data Processing:
Pandas,NumPy - Logging: Python
logging - **Prompt Monitoring: **
**Arize Phoenix**
Let’s get Started:
Press enter or click to view image in full size
Multi Agent System (Insurance Support System): [Image by Author]
Importing the required libraries:
We import the required libraries and load the Open AI API key and phoenix_endpoint.
#importing the required librariesimport pandas as pdimport numpy as npimport jsonimport refrom typing import List, Dict, Any, Tuplefrom openai import OpenAIimport timefrom dotenv import load_dotenvimport openaiimport osfrom datasets import load_datasetimport chromadbfrom datetime import datetime, timedeltaimport randomimport sqlite3from opentelemetry.trace.status import Status, StatusCodefrom opentelemetry.trace import get_current_spanfrom phoenix.otel import register# ✅ Load environment variablesload_dotenv()openai_api_key = os.getenv("OPEN_AI_KEY")phoenix_endpoint = os.getenv("PHOENIX_COLLECTOR_ENDPOINT")
Data Infrastructure
A. Synthetic Data on SQLite DB
We now create synthetic, interrelated datasets for a mock insurance company — including customers, policies, billing, payments, and claims — using pandas, numpy, and Python’s datetime and random modules.
import pandas as pdimport numpy as npfrom datetime import datetime, timedeltaimport randomdef generate_sample_data(random_state=42): """Generate enriched sample data for all tables with 50 first/last names""" random.seed(random_state) np.random.seed(random_state) # Define 50 first and last names first_names = [ "John", "Jane", "Robert", "Maria", "David", "Lisa", "Michael", "Sarah", "James", "Emily", "William", "Emma", "Joseph", "Olivia", "Charles", "Ava", "Thomas", "Isabella", "Daniel", "Mia", "Matthew", "Sophia", "Anthony", "Charlotte", "Christopher", "Amelia", "Andrew", "Harper", "Joshua", "Evelyn", "Ryan", "Abigail", "Brandon", "Ella", "Justin", "Scarlett", "Tyler", "Grace", "Alexander", "Chloe", "Kevin", "Victoria", "Jason", "Lily", "Brian", "Hannah", "Eric", "Aria", "Kyle", "Zoey" ] last_names = [ "Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia", "Miller", "Davis", "Rodriguez", "Martinez", "Hernandez", "Lopez", "Gonzalez", "Wilson", "Anderson", "Thomas", "Taylor", "Moore", "Jackson", "Martin", "Lee", "Perez", "Thompson", "White", "Harris", "Sanchez", "Clark", "Ramirez", "Lewis", "Robinson", "Walker", "Young", "Allen", "King", "Wright", "Scott", "Torres", "Nguyen", "Hill", "Flores", "Green", "Adams", "Nelson", "Baker", "Hall", "Rivera", "Campbell", "Mitchell", "Carter", "Roberts" ] # Generate customers (1000 random name combinations) customers = pd.DataFrame({ 'customer_id': [f'CUST{str(i).zfill(5)}' for i in range(1, 1001)], 'first_name': [random.choice(first_names) for _ in range(1000)], 'last_name': [random.choice(last_names) for _ in range(1000)], 'email': [f'user{i}@example.com' for i in range(1, 1001)], 'phone': [f'555-{str(random.randint(100,999)).zfill(3)}-{str(random.randint(1000,9999)).zfill(4)}' for _ in range(1000)], 'date_of_birth': [datetime(1980, 1, 1) + timedelta(days=random.randint(0, 10000)) for _ in range(1000)], 'state': [random.choice(['CA', 'NY', 'TX', 'FL', 'IL', 'PA', 'OH', 'GA']) for _ in range(1000)] }) # Policies policies = pd.DataFrame({ 'policy_number': [f'POL{str(i).zfill(6)}' for i in range(1, 1501)], 'customer_id': [f'CUST{str(random.randint(1, 1000)).zfill(5)}' for _ in range(1500)], 'policy_type': [random.choice(['auto', 'home', 'life']) for _ in range(1500)], 'start_date': [datetime(2023, 1, 1) + timedelta(days=random.randint(0, 365)) for _ in range(1500)], 'premium_amount': [round(random.uniform(50, 500), 2) for _ in range(1500)], 'billing_frequency': [random.choice(['monthly', 'quarterly', 'annual']) for _ in range(1500)], 'status': [random.choice(['active', 'active', 'active', 'cancelled']) for _ in range(1500)] }) # Auto Policy Details (subset) auto_policies = policies[policies['policy_type'] == 'auto'].copy() auto_policy_details = pd.DataFrame({ 'policy_number': auto_policies['policy_number'], 'vehicle_vin': [f'VIN{random.randint(10000000000000000, 99999999999999999)}' for _ in range(len(auto_policies))], 'vehicle_make': [random.choice(['Toyota', 'Honda', 'Ford', 'Chevrolet', 'Nissan']) for _ in range(len(auto_policies))], 'vehicle_model': [random.choice(['Camry', 'Civic', 'F-150', 'Malibu', 'Altima']) for _ in range(len(auto_policies))], 'vehicle_year': [random.randint(2015, 2023) for _ in range(len(auto_policies))], 'liability_limit': [random.choice([50000, 100000, 300000]) for _ in range(len(auto_policies))], 'collision_deductible': [random.choice([250, 500, 1000]) for _ in range(len(auto_policies))], 'comprehensive_deductible': [random.choice([250, 500, 1000]) for _ in range(len(auto_policies))], 'uninsured_motorist': [random.choice([0, 1]) for _ in range(len(auto_policies))], 'rental_car_coverage': [random.choice([0, 1]) for _ in range(len(auto_policies))] }) # Billing billing = pd.DataFrame({ 'bill_id': [f'BILL{str(i).zfill(6)}' for i in range(1, 5001)], 'policy_number': [random.choice(policies['policy_number']) for _ in range(5000)], 'billing_date': [datetime(2024, 1, 1) + timedelta(days=random.randint(0, 90)) for _ in range(5000)], 'due_date': [datetime(2024, 1, 15) + timedelta(days=random.randint(0, 90)) for _ in range(5000)], 'amount_due': [round(random.uniform(100, 1000), 2) for _ in range(5000)], 'status': [random.choice(['paid', 'pending', 'overdue']) for _ in range(5000)] }) # Payments payments = pd.DataFrame({ 'payment_id': [f'PAY{str(i).zfill(6)}' for i in range(1, 4001)], 'bill_id': [random.choice(billing['bill_id']) for _ in range(4000)], 'payment_date': [datetime(2024, 1, 1) + timedelta(days=random.randint(0, 90)) for _ in range(4000)], 'amount': [round(random.uniform(50, 500), 2) for _ in range(4000)], 'payment_method': [random.choice(['credit_card', 'debit_card', 'bank_transfer']) for _ in range(4000)], 'transaction_id': [f'TXN{random.randint(100000,999999)}' for _ in range(4000)], 'status': [random.choice(['completed', 'pending', 'failed']) for _ in range(4000)] }) # Claims claims = pd.DataFrame({ 'claim_id': [f'CLM{str(i).zfill(6)}' for i in range(1, 301)], 'policy_number': [random.choice(policies['policy_number']) for _ in range(300)], 'claim_date': [datetime(2024, 1, 1) + timedelta(days=random.randint(0, 90)) for _ in range(300)], 'incident_type': [random.choice(['collision', 'theft', 'property_damage', 'medical', 'liability']) for _ in range(300)], 'estimated_loss': [round(random.uniform(500, 20000), 2) for _ in range(300)], 'status': [random.choice(['submitted', 'under_review', 'approved', 'paid', 'denied']) for _ in range(300)] }) return { 'customers': customers, 'policies': policies, 'auto_policy_details': auto_policy_details, 'billing': billing, 'payments': payments, 'claims': claims, }sample_data = generate_sample_data()
Hence we geneate total 6 tables as below:
A1: Customers Table
Contains personal and contact details for all insurance customers.
print("Customers Data Shape:", sample_data["customers"].shape)sample_data["customers"].head()
Press enter or click to view image in full size
A2: Billing Table
Tracks all bills issued for insurance policies, including due dates and payment status.
print("billing Data Shape:", sample_data["billing"].shape)sample_data["billing"].head()
Press enter or click to view image in full size
A3: Payment Data
Records payment transactions made toward bills, including method and transaction status.
print("payments Data Shape:", sample_data["payments"].shape)sample_data["payments"].head()
Press enter or click to view image in full size
A4: Claims Data
Logs insurance claims filed under policies with details of incidents and claim status.
print("claims Data Shape:", sample_data["claims"].shape)sample_data["claims"].head()
Press enter or click to view image in full size
A5: Policy Data
Stores information about each customer’s insurance policies, including type, start date, and status.
print("policies Data Shape:", sample_data["policies"].shape)sample_data["policies"].head()
Press enter or click to view image in full size
A6: Auto Policy Data
Holds vehicle-specific details for auto insurance policies.
print("auto policies Data Shape:", sample_data["auto_policy_details"].shape)sample_data["auto_policy_details"].head()
Press enter or click to view image in full size
A7: Uploading the Data to the SQLite DB:
We first create SQLite3 insurance_support.db and define the function to create the tables.
import sqlite3import pandas as pdimport numpy as npfrom datetime import datetime, timedeltaimport randomimport jsondef connect_db(db_path='insurance_support.db'): """Connect to SQLite database""" return sqlite3.connect(db_path)def drop_and_create_tables(conn): """Drop existing tables and recreate the schema""" cursor = conn.cursor() # Drop tables if exist cursor.executescript(""" DROP TABLE IF EXISTS claims; DROP TABLE IF EXISTS payments; DROP TABLE IF EXISTS billing; DROP TABLE IF EXISTS auto_policy_details; DROP TABLE IF EXISTS policies; DROP TABLE IF EXISTS customers; """) # Create tables cursor.executescript(""" CREATE TABLE customers ( customer_id VARCHAR(20) PRIMARY KEY, first_name VARCHAR(50), last_name VARCHAR(50), email VARCHAR(100), phone VARCHAR(20), date_of_birth DATE, state VARCHAR(20) ); CREATE TABLE policies ( policy_number VARCHAR(20) PRIMARY KEY, customer_id VARCHAR(20), policy_type VARCHAR(50), start_date DATE, premium_amount DECIMAL(10,2), billing_frequency VARCHAR(20), status VARCHAR(20), FOREIGN KEY (customer_id) REFERENCES customers(customer_id) ); CREATE TABLE auto_policy_details ( policy_number VARCHAR(20) PRIMARY KEY, vehicle_vin VARCHAR(50), vehicle_make VARCHAR(50), vehicle_model VARCHAR(50), vehicle_year INTEGER, liability_limit DECIMAL(10,2), collision_deductible DECIMAL(10,2), comprehensive_deductible DECIMAL(10,2), uninsured_motorist BOOLEAN, rental_car_coverage BOOLEAN, FOREIGN KEY (policy_number) REFERENCES policies(policy_number) ); CREATE TABLE billing ( bill_id VARCHAR(20) PRIMARY KEY, policy_number VARCHAR(20), billing_date DATE, due_date DATE, amount_due DECIMAL(10,2), status VARCHAR(20), FOREIGN KEY (policy_number) REFERENCES policies(policy_number) ); CREATE TABLE payments ( payment_id VARCHAR(20) PRIMARY KEY, bill_id VARCHAR(20), payment_date DATE, amount DECIMAL(10,2), payment_method VARCHAR(50), transaction_id VARCHAR(100), status VARCHAR(20), FOREIGN KEY (bill_id) REFERENCES billing(bill_id) ); CREATE TABLE claims ( claim_id VARCHAR(20) PRIMARY KEY, policy_number VARCHAR(20), claim_date DATE, incident_type VARCHAR(100), estimated_loss DECIMAL(10,2), status VARCHAR(20), FOREIGN KEY (policy_number) REFERENCES policies(policy_number) ); """) conn.commit()def insert_data(conn, data): """Insert all DataFrames into SQLite""" for table, df in data.items(): df.to_sql(table, conn, if_exists='append', index=False) conn.commit()def setup_insurance_database(data): """Main function to create and populate the insurance database""" conn = connect_db() drop_and_create_tables(conn) insert_data(conn, data) conn.close() print("✅ Database created successfully with enriched synthetic data!")setup_insurance_database(sample_data)
✅ Database created successfully with enriched synthetic data!
Test the SQL connection:
# test the SQL connectionconn = connect_db()query = "SELECT * FROM customers LIMIT 5;"df_sql = pd.read_sql_query(query, conn)conn.close()df_sql
Press enter or click to view image in full size
B. FAQ Data
B1: Load the Data
To prepare FAQ (Frequently Asked Question), we first load the InsuranceQA-v2 dataset from Hugging Face. This is a question-answering dataset in the insurance domain (Data Source)
We first load the InsuranceQA-v2 dataset (from Hugging Face datasets library) and then merges all splits into one pandas DataFrame.
Further we combine each question and answer into a single text field combined, and display the first five rows.
from datasets import load_datasetimport pandas as pd# Load the dataset from Hugging Faceds = load_dataset("deccan-ai/insuranceQA-v2")# Combine all splits into a single DataFramedf = pd.concat([split.to_pandas() for split in ds.values()], ignore_index=True)df["combined"] = "Question: " + df["input"] + " \n Answer: " + df["output"]# Inspectprint(df.shape)df.head()
Press enter or click to view image in full size
First Five Rows of the InsuranceQA-v2 dataset
B2: Setup the Vector Store : ChromaDB
We first setup chromadb store using PersistentClient and then create a collection named insurance_FAQ_collection
import chromadb# Setting up the Chromadbchroma_client = chromadb.PersistentClient(path="./chroma_db")# Add a collectioncollection = chroma_client.get_or_create_collection(name="insurance_FAQ_collection")
Next we add the data to the collection
Here we sample 500 question and answers from the entire dataset for the test purpose.
Please note that the embeddings will be default embedding from sentence transformers (all-MiniLM-L6-v2).
# Collection 1 for insurance Q&A Datasetfrom tqdm import tqdmdf = df.sample(200, random_state=42).reset_index(drop=True) # For testing, use a smaller subset# Add data to collection# here the chroma db will use default embeddings (sentence transformers)# Split into batches of <= 5000batch_size = 100for i in tqdm(range(0, len(df), batch_size)): batch_df = df.iloc[i:i+batch_size] collection.add( documents=batch_df["combined"].tolist(), metadatas=[{"question": q, "answer": a} for q, a in zip(batch_df["input"], batch_df["output"])], ids=batch_df.index.astype(str).tolist() )
Testing the Vector Store:
## Testing the retrievalquery = "What does life insurance cover?"collection = chroma_client.get_collection(name="insurance_FAQ_collection")results = collection.query( query_texts=[query], n_results=3,)for i, m in enumerate(results["metadatas"][0]): print(f"Result {i+1}:") print("Distance:", results["distances"][0][i]) print("Q:", m["question"]) print("A:", m["answer"]) print("-" * 50)
Result 1:Distance: 0.3581928610801697Q: What Does Life Insurance Typically Cover? A: Life insurance is intended to provide some monetary assistance in the event that the insured passes away while covered . Life insurance is typically meant as a way for the family of the insured to be financially capable of moving on after the loss . The funds from the policy are intended to help provide for those things that could have possibly been provided -LRB- i.e. pay off debt , pay for college , etc. . . -RRB- by the insured if the insured had not passed . This is a very limited explanation of life insurance , contact a local agent to discuss your wants and needs with this type of coverage . --------------------------------------------------Result 2:Distance: 0.41367289423942566Q: What Does Life Insurance Typically Cover? A: Life insurance is designed to pay a determined benefit amount upon the death of the insured . Any cause of death is typically covered except suicide within the first two years of the policy . If it can be proved that fraud was committed in obtaining the policy then only the amount of premiums paid will be returned to the beneficiary . --------------------------------------------------Result 3:Distance: 0.44703125953674316Q: What Does Life Insurance Not Cover? A: The only thing that all life insurance policies do not cover is death due to suicide in the first two years of the policy . Some policies will exclude death caused during the commission of a crime , or by acts of war or terrorism . Others could have a stipulation , such as death caused by parachuting , but that will be an exception stated explicitly in your policy for your particular situation . Life insurance pays for death due to old age , anything health related , and accident . This covers the vast majority of causes of death . --------------------------------------------------
Read my other article to know more about ChromaDB: Practical Guide to Using ChromaDB for RAG and Semantic Search
Setting up Prompt Monitoring
In any GenAI application, when there are multiple prompts, it becomes essential to monitor the prompts.
We use open source arize-phoenix library (OpenTelemetry backend) for tracing and debugging to setup the monitoring system.
Set up the phoenix_enpoint in the .env file as below:
PHOENIX_COLLECTOR_ENDPOINT=http://127.0.0.1:6006/v1/traces
Then run below command in the terminal:
phoenix serve
By default, it runs at:
http://localhost:6006
Then we define below trace_provider and trace_agent to enable agent level tracing.
from functools import wrapsimport timefrom opentelemetry.trace.status import Status, StatusCodefrom opentelemetry.trace import get_current_spanfrom phoenix.otel import registerfrom dotenv import load_dotenvload_dotenv()phoenix_endpoint = os.getenv("PHOENIX_COLLECTOR_ENDPOINT")tracer_provider = register( project_name="multi-agent-system", endpoint=phoenix_endpoint, auto_instrument=True)tracer = tracer_provider.get_tracer(__name__)# --- Decorator ---def trace_agent(func): """ Decorator to wrap multi-agent functions in a Phoenix span with metadata. """ @wraps(func) def wrapper(*args, **kwargs): state = args[0] if args else {} agent_name = func.__name__ with tracer.start_as_current_span(agent_name) as span: # --- Add standard metadata --- span.set_attribute("agent.name", agent_name) span.set_attribute("agent.type", agent_name.replace("_node", "")) span.set_attribute("user.id", state.get("customer_id", "unknown")) span.set_attribute("policy.number", state.get("policy_number", "unknown")) span.set_attribute("claim.id", state.get("claim_id", "unknown")) span.set_attribute("task", state.get("task", "none")) span.set_attribute("timestamp", state.get("timestamp", "n/a")) start_time = time.time() try: result = func(*args, **kwargs) # --- Record execution metadata --- duration = time.time() - start_time span.set_attribute("execution.duration_sec", duration) if isinstance(result, dict): span.set_attribute("result.keys", list(result.keys())) span.set_status(Status(StatusCode.OK)) return result except Exception as e: span.record_exception(e) span.set_status(Status(StatusCode.ERROR, str(e))) raise return wrapper
Setting up the OpenAI Client
We define Open AI client which requires openai_api_key. Ensure that you have defined the api key in a .env file in the repo.
Please note, here we are using gpt-5-mini model. You can use the model of your choice.
We define a function where LLM itself can decide to call the tool if available. There are two cases:
Case 1: No tool defined:
In this case LLM simply generates the output for given input (prompt)
Input → LLM → Output
Case 2: Tool is defined:
In this case tool schema is also passed to the LLM alogn with the prompt.
- if LLM decides to call the tool, then that tool is executed locally and then output of the tool call is again fed to the LLM for final answer generation.
- if LLM decides not to call the tool, then the first output is the final output.
# Case 2.1Input → LLM → tool-call→ tool-execution → LLM → output# Case 2.2Input → LLM → no-tool-call→ output
All above cases are explained using below chart also:
Press enter or click to view image in full size
LLM with and without Tool call [Image by Author]
from dotenv import load_dotenvfrom openai import OpenAIimport jsonfrom typing import List, Dict, Any, Optional# ✅ Load environment variablesload_dotenv()openai_api_key = os.getenv("OPEN_AI_KEY")client = OpenAI(api_key=openai_api_key)def run_llm( prompt: str, tools: Optional[List[Dict]] = None, tool_functions: Optional[Dict[str, Any]] = None, model: str = "gpt-5-mini",) -> str: """ Run an LLM request that optionally supports tools. Args: prompt (str): The system or user prompt to send. tools (list[dict], optional): Tool schema list for model function calling. tool_functions (dict[str, callable], optional): Mapping of tool names to Python functions. model (str): Model name to use (default: gpt-5-mini). Returns: str: Final LLM response text. """ # Step 1: Initial LLM call response = client.chat.completions.create( model=model, messages=[{"role": "system", "content": prompt}], tools=tools if tools else None, tool_choice="auto" if tools else None ) message = response.choices[0].message # Step 2: If no tools or no tool calls, return simple model response if not getattr(message, "tool_calls", None): return message.content # Step 3: Handle tool calls dynamically if not tool_functions: return message.content + "\n\n⚠️ No tool functions provided to execute tool calls." tool_messages = [] for tool_call in message.tool_calls: func_name = tool_call.function.name args = json.loads(tool_call.function.arguments or "{}") tool_fn = tool_functions.get(func_name) try: result = tool_fn(**args) if tool_fn else {"error": f"Tool '{func_name}' not implemented."} except Exception as e: result = {"error": str(e)} tool_messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result) }) # Step 4: Second pass — send tool outputs back to the model followup_messages = [ {"role": "system", "content": prompt}, { "role": "assistant", "content": message.content, "tool_calls": [ { "id": tc.id, "type": tc.type, "function": { "name": tc.function.name, "arguments": tc.function.arguments, }, } for tc in message.tool_calls ], }, *tool_messages, ] final = client.chat.completions.create(model=model, messages=followup_messages) return final.choices[0].message.content
Case 1: Calling LLM without tool
prompt = "Explain the theory of relativity in simple terms in 30 words."response = run_llm(prompt)print(response)
Relativity links space and time into spacetime. Special relativity: light speed constant, motion changes measured time and length. General relativity: mass-energy curves spacetime, producing gravity, orbital motion, and observable effects.
Case 2.1: Calling LLM with tool
def add_numbers(a: int, b: int): return {"result": a + b}tools = [{ "type": "function", "function": { "name": "add_numbers", "description": "Add two numbers together", "parameters": { "type": "object", "properties": { "a": {"type": "integer"}, "b": {"type": "integer"} }, "required": ["a", "b"] }, }}]prompt = "Please add 3 and 5 using the available tool."response = run_llm(prompt, tools=tools, tool_functions={"add_numbers": add_numbers})print(response)
3 + 5 = 8. I used the tool to compute this.
Case 2.2: Calling LLM with tool but asking a non-tool call question
prompt = "Explain the theory of relativity in simple terms in 30 words."response = run_llm(prompt, tools=tools, tool_functions={"add_numbers": add_numbers})print(response)
Relativity explains how space, time, motion, and gravity connect: observers moving differently measure distances, durations, and mass differently; light speed is constant, and gravity curves spacetime, affecting clocks and paths.
Defining the Tools for the Agents
We now define the tools that can be used by the different agents.
**ask_user()**– Prompts the user for input and logs their response.**get_policy_details()**– Retrieves detailed policy information (and customer info) by policy number.**get_claim_status()**– Fetches claim details and current status using a claim ID or policy number.**get_billing_info()**– Retrieves current billing information such as balance, due date, and premium.**get_payment_history()**– Returns the most recent payment records for a given policy.**get_auto_policy_details()**– Fetches auto policy–specific details like vehicle info and deductibles
# =====================================================# Define the TOOL functions (these run locally)# =====================================================import loggingfrom datetime import datetime# Set up logginglogging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('insurance_agent.log'), logging.StreamHandler() ])logger = logging.getLogger(__name__)def ask_user(question: str, missing_info: str = ""): """Ask the user for input and return the response.""" logger.info(f"🗣️ Asking user for input: {question}") if missing_info: print(f"---USER INPUT REQUIRED---\nMissing information: {missing_info}") else: print(f"---USER INPUT REQUIRED---") answer = input(f"{question}: ") logger.info(f"✅ User provided response: {answer}") return {"context": answer, "source": "User Input"}def get_policy_details(policy_number: str) -> Dict[str, Any]: """Fetch a customer's policy details by policy number""" logger.info(f"🔍 Fetching policy details for: {policy_number}") conn = sqlite3.connect('insurance_support.db') cursor = conn.cursor() cursor.execute(""" SELECT p.*, c.first_name, c.last_name FROM policies p JOIN customers c ON p.customer_id = c.customer_id WHERE p.policy_number = ? """, (policy_number,)) result = cursor.fetchone() conn.close() if result: logger.info(f"✅ Policy found: {policy_number}") columns = [desc[0] for desc in cursor.description] return dict(zip(columns, result)) logger.warning(f"❌ Policy not found: {policy_number}") return {"error": "Policy not found"}def get_claim_status(claim_id: str = None, policy_number: str = None) -> Dict[str, Any]: """Get claim status and details""" logger.info(f"🔍 Fetching claim status - Claim ID: {claim_id}, Policy: {policy_number}") conn = sqlite3.connect('insurance_support.db') cursor = conn.cursor() if claim_id: cursor.execute(""" SELECT c.*, p.policy_type FROM claims c JOIN policies p ON c.policy_number = p.policy_number WHERE c.claim_id = ? """, (claim_id,)) elif policy_number: cursor.execute(""" SELECT c.*, p.policy_type FROM claims c JOIN policies p ON c.policy_number = p.policy_number WHERE c.policy_number = ? ORDER BY c.claim_date DESC LIMIT 3 """, (policy_number,)) result = cursor.fetchall() conn.close() if result: logger.info(f"✅ Found {len(result)} claim(s)") columns = [desc[0] for desc in cursor.description] return [dict(zip(columns, row)) for row in result] logger.warning("❌ No claims found") return {"error": "Claim not found"}def get_billing_info(policy_number: str = None, customer_id: str = None) -> Dict[str, Any]: """Get billing information including current balance and due dates""" logger.info(f"🔍 Fetching billing info - Policy: {policy_number}, Customer: {customer_id}") conn = sqlite3.connect('insurance_support.db') cursor = conn.cursor() if policy_number: cursor.execute(""" SELECT b.*, p.premium_amount, p.billing_frequency FROM billing b JOIN policies p ON b.policy_number = p.policy_number WHERE b.policy_number = ? AND b.status = 'pending' ORDER BY b.due_date DESC LIMIT 1 """, (policy_number,)) elif customer_id: cursor.execute(""" SELECT b.*, p.premium_amount, p.billing_frequency FROM billing b JOIN policies p ON b.policy_number = p.policy_number WHERE p.customer_id = ? AND b.status = 'pending' ORDER BY b.due_date DESC LIMIT 1 """, (customer_id,)) result = cursor.fetchone() conn.close() if result: logger.info("✅ Billing info found") columns = [desc[0] for desc in cursor.description] return dict(zip(columns, result)) logger.warning("❌ Billing info not found") return {"error": "Billing information not found"}def get_payment_history(policy_number: str) -> List[Dict[str, Any]]: """Get payment history for a policy""" logger.info(f"🔍 Fetching payment history for policy: {policy_number}") conn = sqlite3.connect('insurance_support.db') cursor = conn.cursor() cursor.execute(""" SELECT p.payment_date, p.amount, p.status, p.payment_method FROM payments p JOIN billing b ON p.bill_id = b.bill_id WHERE b.policy_number = ? ORDER BY p.payment_date DESC LIMIT 10 """, (policy_number,)) results = cursor.fetchall() conn.close() if results: logger.info(f"✅ Found {len(results)} payment records") columns = [desc[0] for desc in cursor.description] return [dict(zip(columns, row)) for row in results] logger.warning("❌ No payment history found") return []def get_auto_policy_details(policy_number: str) -> Dict[str, Any]: """Get auto-specific policy details including vehicle info and deductibles""" logger.info(f"🔍 Fetching auto policy details for: {policy_number}") conn = sqlite3.connect('insurance_support.db') cursor = conn.cursor() cursor.execute(""" SELECT apd.*, p.policy_type, p.premium_amount FROM auto_policy_details apd JOIN policies p ON apd.policy_number = p.policy_number WHERE apd.policy_number = ? """, (policy_number,)) result = cursor.fetchone() conn.close() if result: logger.info("✅ Auto policy details found") columns = [desc[0] for desc in cursor.description] return dict(zip(columns, result)) logger.warning("❌ Auto policy details not found") return {"error": "Auto policy details not found"}
Defining the Prompts:
Let’s now setup the most crucial part of the agents: Prompts. We define specific prompts for each agent.
SUPERVISOR_PROMPT = """You are the SUPERVISOR AGENT managing a team of insurance support specialists.Your role:1. Go throught the conversatio history and understand the current requirement.2. Understand the user's intent and context.3. Evaluate available information and decide if clarification is needed.4. Route to the appropriate specialist agent.5. End conversation when the task is complete.AVAILABLE INFORMATION:- Conversation History: {conversation_history}CRITICAL RULES:- If policy number is already available, DO NOT ask for it again- If customer ID is already available, DO NOT ask for it again - Only use ask_user tool if ESSENTIAL information is missing. Keep the clarification questions minimal (within 15 words) and specific.- Route directly to appropriate agent if you have sufficient information- Check the conversation history carefully - policy numbers or customer IDs mentioned earlier in the conversation should be considered available- If the user just provided information in response to your clarification question, that information is NOW available and should not be asked for againSpecialist agents:- policy_agent → policy details, coverage, endorsements- billing_agent → billing, payments, premium questions- claims_agent → claim filing, tracking, settlements- human_escalation_agent → for complex cases- general_help_agent → for general questionsCLARIFICICATION QUESTION GUIDELINES:1. Keep questions concise (<=15 words)2. Ask only for ESSENTIAL missing info (policy number, customer ID, claim ID)EVALUATION INSTRUCTIONS:- Review the conversation history thoroughly.- Agents answers are also part of the conversation history.- If agents ask for more information, use ask_user tool to get it from the user.- Evaluate the anwer of the agent carefully to see if the user's question is fully answered.- If user's question is fully answered, route to 'end'.DECISION GUIDELINES:1. Policy/coverage questions → policy_agent2. Billing/payment questions → billing_agent 3. Claims questions → claims_agent4. General questions (example: In general, what does life insurance cover?) → general_help_agent5. Complete + answered → endTASK GENERATION GUIDELINES:1. If routing to a specialist, summarize the user's main request.2. Keep the policy number, customer ID, claim ID (if applicable and available) in Task also.Respond in JSON:{{ "next_agent": "<agent_name or 'end'>", "task": "<concise task description>", "justification": "<why this decision>"}}Only use ask_user tool if absolutely necessary."""POLICY_AGENT_PROMPT = """You are a **Policy Specialist Agent** for an insurance company.Assigned Task:{task}Responsibilities:1. Policy details, coverage, and deductibles2. Vehicle info and auto policy specifics3. Endorsements and policy updatesTools:- get_policy_details- get_auto_policy_detailsContext:- Policy Number: {policy_number}- Customer ID: {customer_id}- Conversation History: {conversation_history}Instructions:- Use tools to retrieve information as needed.- Ask politely for missing details.- Keep responses professional and clear."""BILLING_AGENT_PROMPT = """You are a **Billing Specialist Agent**.Assigned Task:{task}Responsibilities:1. Billing statements, payments, and invoices2. Premiums, due dates, and payment historyInstructions:- Use tools to retrieve billing and payment information.- Ask politely for any missing details.- Just answer the questions that are asked. Don't provide extra information.- If you think the question is answered, don't ask for more information. Just retrun with the specific answer.Tools:- get_billing_info- get_payment_historyContext:- Conversation History: {conversation_history}"""CLAIMS_AGENT_PROMPT = """You are a **Claims Specialist Agent**.Assigned Task:{task}Responsibilities:1. Retrieve or update claim status2. Help file new claims3. Explain claim process and settlementsTools:- get_claim_statusContext:- Policy Number: {policy_number}- Claim ID: {claim_id}- Conversation History: {conversation_history}"""GENERAL_HELP_PROMPT = """You are a **General Help Agent** for insurance customers.Assigned Task:{task}Goal:Answer FAQs and explain insurance topics in simple, clear, and accurate language.Context:- Conversation History: {conversation_history}Retrieved FAQs from the knowledge base:{faq_context}Instructions:1. Review the retrieved FAQs carefully before answering.2. If one or more FAQs directly answer the question, use them to construct your response.3. If the FAQs are related but not exact, summarize the most relevant information.4. If no relevant FAQs are found, politely inform the user and provide general guidance.5. Keep responses clear, concise, and written for a non-technical audience.6. Do not fabricate details beyond what’s supported by the FAQs or obvious domain knowledge.7. End by offering further help (e.g., “Would you like to know more about this topic?”).Now provide the best possible answer for the user’s question."""HUMAN_ESCALATION_PROMPT = """You are handling a **Customer Escalation**.Assigned Task:{task}Conversation History: {conversation_history}Respond empathetically, acknowledge the request for a human, and confirm that a human representative will join shortly.Don't attempt to answer any questions or provide information yourself.Don't ask any further questions. Just acknowledge the escalation request."""FINAL_ANSWER_PROMPT = """ The user asked: "{user_query}" The specialist agent provided this detailed response: {specialist_response} Your task: Create a FINAL, CLEAN response that: 1. Directly answers the user's original question in a friendly tone 2. Includes only the most relevant information (remove technical details) 3. Is concise and easy to understand 4. Ends with a polite closing Important: Do NOT include any internal instructions, tool calls, or technical details. Just provide the final answer that the user should see. Final response: """
These prompts will be used in the next section where we define the agents.
Setting up the Agents
Let’s now setup the agents themselves.
1. Supervisor Agent
This is the central agent which coordinates with user and all the other agents. The Supervisor agent:
- Analyzes conversation history to identify the user’s current request, previously provided details (like policy or customer ID), and any missing information.
- Determines the next step by routing the query to the appropriate specialist agent — such as
policy_agent,billing_agent, orclaims_agent— based on the detected intent. - Manages clarification flow by invoking the
ask_usertool to request only essential missing details in a concise, targeted way. - Tracks iteration progress to prevent infinite loops and escalates to a human support agent if the issue remains unresolved after multiple supervisor iterations.
- Maintains and updates conversation history with all assistant–user exchanges, ensuring full context is preserved for downstream agents.
@trace_agent # used to trace the prompt level monitoringdef supervisor_agent(state): print("---SUPERVISOR AGENT---") # Increment iteration counter n_iter = state.get("n_iteration", 0) + 1 state["n_iteration"] = n_iter print(f"🔢 Supervisor iteration: {n_iter}") # Force end if iteration limit reached # Escalate to human support if iteration limit reached if n_iter >= 3: print("⚠️ Maximum supervisor iterations reached — escalating to human agent") updated_history = ( state.get("conversation_history", "") + "\nAssistant: It seems this issue requires human review. Escalating to a human support specialist." ) return { "escalate_to_human": True, "conversation_history": updated_history, "next_agent": "human_escalation_agent", "n_iteration": n_iter } # Check if we're coming from a clarification if state.get("needs_clarification", False): user_clarification = state.get("user_clarification", "") print(f"🔄 Processing user clarification: {user_clarification}") # Update conversation history with the clarification exchange clarification_question = state.get("clarification_question", "") updated_conversation = state.get("conversation_history", "") + f"\nAssistant: {clarification_question}\nUser: {user_clarification}" # Update state to clear clarification flags and update history updated_state = state.copy() updated_state["needs_clarification"] = False updated_state["conversation_history"] = updated_conversation # Clear clarification fields if "clarification_question" in updated_state: del updated_state["clarification_question"] if "user_clarification" in updated_state: del updated_state["user_clarification"] return updated_state user_query = state["user_input"] conversation_history = state.get("conversation_history", "") print(f"User Query: {user_query}") print(f"Conversation History: {conversation_history}") # Include the ENTIRE conversation history in the prompt full_context = f"Full Conversation:\n{conversation_history}" prompt = SUPERVISOR_PROMPT.format( conversation_history=full_context, # Use full context instead of just history ) tools = [ { "type": "function", "function": { "name": "ask_user", "description": "Ask the user for clarification or additional information when their query is unclear or missing important details. ONLY use this if essential information like policy number or customer ID is missing.", "parameters": { "type": "object", "properties": { "question": { "type": "string", "description": "The specific question to ask the user for clarification" }, "missing_info": { "type": "string", "description": "What specific information is missing or needs clarification" } }, "required": ["question", "missing_info"] } } } ] print("🤖 Calling LLM for supervisor decision...") response = client.chat.completions.create( model="gpt-5-mini", messages=[{"role": "system", "content": prompt}], tools=tools, tool_choice="auto" ) message = response.choices[0].message # Check if supervisor wants to ask user for clarification if getattr(message, "tool_calls", None): print("🛠️ Supervisor requesting user clarification") for tool_call in message.tool_calls: if tool_call.function.name == "ask_user": args = json.loads(tool_call.function.arguments) question = args.get("question", "Can you please provide more details?") missing_info = args.get("missing_info", "additional information") print(f"❓ Asking user: {question}") user_response_data = ask_user(question, missing_info) user_response = user_response_data["context"] print(f"✅ User response: {user_response}") # Update conversation history with the question updated_history = conversation_history + f"\nAssistant: {question}" updated_history = updated_history + f"\nUser: {user_response}" return { "needs_clarification": True, "clarification_question": question, "user_clarification": user_response, "conversation_history": updated_history } # If no tool calls, proceed with normal supervisor decision message_content = message.content try: parsed = json.loads(message_content) print("✅ Supervisor output parsed successfully") except json.JSONDecodeError: print("❌ Supervisor output invalid JSON, using fallback") parsed = {} next_agent = parsed.get("next_agent", "general_help_agent") task = parsed.get("task", "Assist the user with their query.") justification = parsed.get("justification", "") print(f"---SUPERVISOR DECISION: {next_agent}---") print(f"Task: {task}") print(f"Reason: {justification}") # Update conversation history with the current exchange updated_conversation = conversation_history + f"\nAssistant: Routing to {next_agent} for: {task}" print(f"➡️ Routing to: {next_agent}") return { "next_agent": next_agent, "task": task, "justification": justification, "conversation_history": updated_conversation, "n_iter