ts/virgil-ai
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor estimation service to use structured agent

Browse source
This commit is contained in:
ipu 2025-07-28 17:07:48 +03:00
parent 47cc1541ed
commit 96421b264c
7 changed files with 316 additions and 361 deletions
Download patch file Download diff file Expand all files Collapse all files

50
README.md
View file

@ -5,7 +5,7 @@ A FastAPI-based insurance AI assistant that integrates with talestorm-ai for cha
## Features
- **Insurance Chat**: AI-powered chat interface for insurance-related questions
- **Insurance Estimation**: Comprehensive underwriting and pricing estimation
- **Insurance Estimation**: AI-powered estimation using TALESTORM API for comprehensive underwriting and pricing
- **Integration**: Seamless integration with talestorm-ai API
- **Modern API**: Built with FastAPI for high performance and automatic documentation
@ -43,7 +43,8 @@ OPENAI_API_KEY=your-openai-api-key-here
# lolly-ai Configuration
PUBLIC_LOLLY_PORT=7310
TALESTORM_API_KEY=your-talestorm-api-key-here
TALESTORM_AGENT_ID=your-talestorm-agent-id-here
TALESTORM_AGENT_ID=your-talestorm-chat-agent-id-here
TALESTORM_ESTIMATION_AGENT_ID=your-talestorm-estimation-agent-id-here
INSURANCE_API_BASE_URL=https://apilolly.cyberpug.ru
# Server Configuration
@ -90,6 +91,51 @@ To remove volumes as well:
docker compose down -v
```
## Agent Configuration
The system uses two different TALESTORM agents for different purposes:
- **TALESTORM_AGENT_ID**: Used for general insurance chat functionality
- **TALESTORM_ESTIMATION_AGENT_ID**: Used specifically for insurance estimation with structured output schema
This separation allows for specialized agents optimized for their respective tasks.
## Estimation Service
The estimation service has been refactored to use the TALESTORM API instead of local underwriting logic. This provides:
- **AI-Powered Analysis**: Uses advanced AI models for risk assessment and pricing
- **Comprehensive Evaluation**: Analyzes all health factors, medications, and conditions
- **Structured Output**: Returns detailed tier assignments and pricing for each applicant
- **Flexible Response**: Handles both accepted and rejected applications with detailed reasoning
### Estimation Request Format
The service accepts insurance application data and returns structured estimation results:
```json
{
"status": "accepted",
"details": {
"dtq": false,
"reason": "All applicants have manageable conditions...",
"tier": 3,
"total_price": 815.75
},
"results": [
{
"name": "Sarah Johnson",
"applicant_type": "primary",
"age": 39,
"bmi": 23.38,
"tier": 3,
"rx_spend": 40.0,
"message": "Type 2 diabetes well-controlled..."
}
]
}
```
## API Endpoints
All API endpoints are versioned under `/api/v1/`:

172
src/api/v1/router.py
View file

@ -4,7 +4,7 @@ import httpx
from . import models
from ...services.chat_service import chat_service
from ...services.estimation_service import run_underwriting
from ...services.estimation_service import estimation_service
from ...config import settings
router = APIRouter()
@ -26,6 +26,7 @@ async def insurance_chat(request: models.InsuranceChatRequest):
)
except Exception as e:
raise e
raise HTTPException(status_code=500, detail=f"Error processing chat request: {str(e)}")
@router.post("/estimation", response_model=models.EstimationResponse)
@ -39,104 +40,8 @@ async def estimate(request: models.EstimationRequest):
detail="Missing required applicants or plans"
)
# Convert request to the format expected by run_underwriting
applicants_dict = []
for applicant in request.applicants:
applicants_dict.append({
"applicant": applicant.applicant,
"firstName": applicant.firstName,
"lastName": applicant.lastName,
"midName": applicant.midName,
"phone": applicant.phone,
"gender": applicant.gender,
"dob": applicant.dob.strftime("%d/%m/%Y"),
"nicotine": applicant.nicotine,
"weight": applicant.weight,
"heightFt": applicant.heightFt,
"heightIn": applicant.heightIn
})
phq_dict = {
"treatment": request.phq.treatment,
"invalid": request.phq.invalid,
"pregnancy": request.phq.pregnancy,
"effectiveDate": request.phq.effectiveDate.strftime("%d/%m/%Y"),
"disclaimer": request.phq.disclaimer,
"signature": request.phq.signature,
"medications": [
{
"applicant": med.applicant,
"name": med.name,
"rxcui": med.rxcui,
"dosage": med.dosage,
"frequency": med.frequency,
"description": med.description
} for med in request.phq.medications
],
"issues": [
{
"key": issue.key,
"details": [
{
"key": detail.key,
"description": detail.description
} for detail in issue.details
]
} for issue in request.phq.issues
],
"conditions": [
{
"key": condition.key,
"description": condition.description
} for condition in request.phq.conditions
]
}
plans_dict = [
{
"id": plan.id,
"coverage": plan.coverage,
"tier": plan.tier
} for plan in request.plans
]
# Step 1: Run estimation
underwriting_result = run_underwriting(applicants_dict, phq_dict, plans_dict)
# Step 2: Check if DTQ → reject application
if underwriting_result["combined"].get("dtq"):
# For DTQ cases, call external reject API and return rejected status
if request.uid:
reject_response = await reject_application(request.uid)
return models.EstimationResponse(
status="rejected",
details=models.EstimationDetails(
dtq=True,
reason="Declined due to high-risk conditions (DTQ triggered).",
tier=int(underwriting_result["combined"]["tier"]),
total_price=underwriting_result["combined"]["total_price"]
),
results=[
models.EstimationResult(
name=result["name"],
applicant_type=result["applicant_type"],
age=result["age"] or 0,
bmi=result["bmi"] or 0.0,
tier=int(result["tier"]),
rx_spend=result["rx_spend"],
message=result["message"]
) for result in underwriting_result["results"]
]
)
# Step 3: Else → assign tier and submit to external API
final_tier = underwriting_result["combined"]["tier"]
plans = request.plans.copy()
if plans:
plans[0].tier = f"tier_{str(final_tier).replace('.', '_')}"
# Assemble external payload
submission_payload = {
# Convert request to the format expected by the estimation service
request_data = {
"applicants": [
{
"applicant": applicant.applicant,
@ -157,7 +62,7 @@ async def estimate(request: models.EstimationRequest):
"id": plan.id,
"coverage": plan.coverage,
"tier": plan.tier
} for plan in plans
} for plan in request.plans
],
"phq": {
"treatment": request.phq.treatment,
@ -204,27 +109,42 @@ async def estimate(request: models.EstimationRequest):
}
}
submit_response = await submit_application(submission_payload)
# Call the TALESTORM API for estimation
estimation_result = await estimation_service.estimate_insurance(request_data)
# Handle the response from TALESTORM API
if estimation_result.get("status") == "error":
raise HTTPException(
status_code=500,
detail=estimation_result.get("details", {}).get("reason", "Estimation service error")
)
# Extract details and results from the TALESTORM response
details = estimation_result.get("details", {})
results = estimation_result.get("results", [])
# Convert results to the expected format
estimation_results = []
for result in results:
estimation_results.append(models.EstimationResult(
name=result.get("name", "Unknown"),
applicant_type=result.get("applicant_type", "Unknown"),
age=result.get("age", 0),
bmi=result.get("bmi", 0.0),
tier=result.get("tier", 4),
rx_spend=result.get("rx_spend", 0.0),
message=result.get("message", "")
))
return models.EstimationResponse(
status="accepted",
status=estimation_result.get("status", "accepted"),
details=models.EstimationDetails(
dtq=False,
reason=underwriting_result["combined"]["message"],
tier=int(underwriting_result["combined"]["tier"]),
total_price=underwriting_result["combined"]["total_price"]
dtq=details.get("dtq", False),
reason=details.get("reason", ""),
tier=details.get("tier", 4),
total_price=details.get("total_price", 0.0)
),
results=[
models.EstimationResult(
name=result["name"],
applicant_type=result["applicant_type"],
age=result["age"] or 0,
bmi=result["bmi"] or 0.0,
tier=int(result["tier"]),
rx_spend=result["rx_spend"],
message=result["message"]
) for result in underwriting_result["results"]
]
results=estimation_results
)
except HTTPException:
@ -236,23 +156,7 @@ async def estimate(request: models.EstimationRequest):
)
async def reject_application(uid: str) -> Dict[str, Any]:
"""Reject application via external API"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{settings.INSURANCE_API_BASE_URL}/applications/reject",
json={"applicationId": uid}
)
return response.json() if response.status_code == 200 else {"error": "Failed to reject application"}
async def submit_application(application_payload: Dict[str, Any]) -> Dict[str, Any]:
"""Submit application via external API"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{settings.INSURANCE_API_BASE_URL}/applications/submit",
json=application_payload
)
return response.json() if response.status_code == 200 else {"error": "Failed to submit application"}

1
src/config.py
View file

@ -12,6 +12,7 @@ class Settings(BaseSettings):
TALESTORM_API_BASE_URL: str = ""
TALESTORM_API_KEY: str
TALESTORM_AGENT_ID: str
TALESTORM_ESTIMATION_AGENT_ID: str

2
src/examples/test_agent_id.py
View file

@ -17,7 +17,7 @@ async def test_agent_id_usage():
# Test session creation
print("\nCreating session with configured agent...")
session_id = await session_service.create_session()
session_id = await session_service.create_session(agent_id=settings.TALESTORM_AGENT_ID)
if session_id:
print(f"✅ Session created: {session_id}")

10
src/services/chat_service.py
View file

@ -18,6 +18,7 @@ class ChatService:
headers["X-API-Key"] = self.api_key
return httpx.AsyncClient(
base_url=self.base_url,
headers=headers
)
@ -25,7 +26,7 @@ class ChatService:
"""Send a message to talestorm-ai and get response"""
async with await self.get_client() as client:
try:
response = await client.post(self.base_url + "/chat/", json={
response = await client.post("/chat/", json={
"chat_session_id": session_id,
"user_message": message
})
@ -47,7 +48,7 @@ class ChatService:
"""Get chat history for a session and format it properly"""
async with await self.get_client() as client:
try:
response = await client.get(self.base_url + "/chat/", params={"chat_session_id": session_id})
response = await client.get("/chat", params={"chat_session_id": session_id})
if response.status_code == 200:
messages = response.json()
history = []
@ -101,10 +102,10 @@ class ChatService:
"""Process an insurance chat request"""
try:
if not session_id:
session_id = await session_service.create_session()
session_id = await session_service.create_session(agent_id=settings.TALESTORM_AGENT_ID)
elif not await session_service.validate_session(session_id):
session_id = await session_service.create_session()
session_id = await session_service.create_session(agent_id=settings.TALESTORM_AGENT_ID)
chat_response = await self.send_message(session_id, message)
history = await self.get_chat_history(session_id)
@ -119,6 +120,7 @@ class ChatService:
}
except Exception as e:
raise e
return {
"session_id": session_id or "fallback-session",
"answer": f"I'm sorry, I'm experiencing technical difficulties. Please try again later. Error: {str(e)}",

425
src/services/estimation_service.py
View file

@ -1,225 +1,228 @@
from datetime import datetime
import httpx
import json
from typing import List, Dict, Any, Optional
from typing import Dict, Any, List, Optional
from ..config import settings
from .session_service import session_service
# ---------------- Constants ---------------- #
class EstimationService:
"""Service for handling insurance estimation via TALESTORM API"""
DECLINABLE_CONDITIONS = {
"aids", "organ_transplant", "multiple_sclerosis", "alzheimer_disease",
"heart_attack", "congestive_heart_failure", "bipolar", "seizures"
def __init__(self):
self.base_url = settings.TALESTORM_API_BASE_URL
self.api_key = settings.TALESTORM_API_KEY
self.agent_id = settings.TALESTORM_ESTIMATION_AGENT_ID
async def get_client(self) -> httpx.AsyncClient:
"""Get HTTP client for TALESTORM API"""
headers = {
"accept": "application/json",
"Content-Type": "application/json"
}
if self.api_key:
headers["X-API-Key"] = self.api_key
DECLINABLE_ISSUE_KEYS = {"heart_attack", "congestive_heart_failure", "bipolar", "seizures"}
HIGH_RISK_FLAGS = {"50_units", "treat_diabetes"}
PRICING_TABLE = {
1: [122, 243, 219, 365],
1.5: [127, 253, 228, 379],
2: [133, 265, 239, 397],
2.5: [139, 277, 249, 415],
3: [145, 290, 261, 435],
3.5: [151, 302, 272, 452],
4: [158, 315, 283, 472],
4.5: [164, 327, 295, 491],
5: [172, 343, 309, 514],
5.5: [182, 364, 327, 545],
6: [196, 390, 351, 585],
6.5: [209, 417, 376, 626],
7: [222, 444, 400, 666]
}
# Simplified drug price map (in a real implementation, you'd load this from a file)
DRUG_PRICE_MAP = {
"metformin": 15.0,
"insulin": 45.0,
"aspirin": 8.0,
"atorvastatin": 25.0,
"lisinopril": 12.0,
"amlodipine": 18.0,
"omeprazole": 22.0,
"simvastatin": 20.0,
"hydrochlorothiazide": 10.0,
"losartan": 30.0
}
# ---------------- Helper Functions ---------------- #
def calculate_age(dob_str: str, effective_date_str: str) -> Optional[int]:
"""Calculate age from date of birth and effective date"""
try:
dob = datetime.strptime(dob_str, "%d/%m/%Y")
ref = datetime.strptime(effective_date_str, "%d/%m/%Y")
return (ref - dob).days // 365
except (ValueError, TypeError):
return None
def calculate_bmi(weight: float, ft: int, inch: int) -> Optional[float]:
"""Calculate BMI from weight and height"""
try:
inches = ft * 12 + inch
return (weight / (inches ** 2)) * 703
except (ValueError, TypeError, ZeroDivisionError):
return None
def check_declinable(phq: Dict[str, Any]) -> bool:
"""Check if application should be declined based on PHQ data"""
if phq.get("treatment") or phq.get("invalid"):
return True
for cond in phq.get("conditions", []):
if cond.get("key") in DECLINABLE_CONDITIONS:
return True
for issue in phq.get("issues", []):
for detail in issue.get("details", []):
if detail.get("key") in DECLINABLE_ISSUE_KEYS:
return True
return False
def base_tier(age: Optional[int]) -> float:
"""Calculate base tier based on age"""
if age is None:
return 4.0
if age <= 30:
return 1.0
elif age <= 40:
return 2.0
elif age <= 50:
return 3.0
elif age <= 60:
return 3.5
else:
return 4.0
def adjust_tier(tier: float, bmi: Optional[float], meds: List[Dict[str, Any]],
issues: List[Dict[str, Any]], applicant_index: int) -> float:
"""Adjust tier based on health factors"""
if bmi is not None and (bmi < 16 or bmi > 40):
tier += 1.0
personal_meds = [m for m in meds if m.get("applicant") == applicant_index]
if len(personal_meds) >= 3:
tier += 0.5
for issue in issues:
for detail in issue.get("details", []):
if detail.get("key") in HIGH_RISK_FLAGS:
tier += 0.5
return round(min(tier, 7.0) * 2) / 2
def calculate_rx_spend(medications: List[Dict[str, Any]], applicant_index: int) -> float:
"""Calculate total prescription spend for an applicant"""
total = 0.0
for med in medications:
if med.get("applicant") != applicant_index:
continue
drug = med.get("name", "").lower().strip()
if drug in DRUG_PRICE_MAP:
total += DRUG_PRICE_MAP[drug]
return round(total, 2)
def get_rx_price(tier: float, coverage_type: int) -> float:
"""Get prescription price based on tier and coverage type"""
tier = round(tier * 2) / 2
coverage_index = {
1: 0, 2: 1, 3: 2, 4: 3
}.get(coverage_type, 0)
return PRICING_TABLE.get(tier, PRICING_TABLE[5])[coverage_index]
# ---------------- Main Underwriting Logic ---------------- #
def run_underwriting(applicants: List[Dict[str, Any]], phq: Dict[str, Any],
plans: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Run comprehensive underwriting analysis"""
results = []
max_tier = 0
dtq_found = False
ages = []
# Calculate ages for all applicants
for applicant in applicants:
age = calculate_age(applicant.get("dob"), phq.get("effectiveDate"))
if age is not None:
ages.append(age)
oldest_age = max(ages) if ages else None
base = base_tier(oldest_age) if oldest_age else None
coverage = plans[0]["coverage"] if plans else 1
coverage_index = {1: 0, 2: 1, 3: 2, 4: 3}.get(coverage, 0)
for idx, applicant in enumerate(applicants):
try:
age = calculate_age(applicant.get("dob"), phq.get("effectiveDate"))
bmi = calculate_bmi(
applicant.get("weight"),
applicant.get("heightFt"),
applicant.get("heightIn")
return httpx.AsyncClient(
base_url=self.base_url,
headers=headers
)
name = applicant.get("firstName", "Unknown")
applicant_type = {
1: "Primary",
2: "Spouse",
3: "Dependent"
}.get(applicant.get("applicant"), "Unknown")
is_dtq = check_declinable(phq)
rx_spend = calculate_rx_spend(phq.get("medications", []), idx)
def _format_estimation_request(self, request_data: Dict[str, Any]) -> str:
"""Format the estimation request as a natural language message for the AI"""
applicants = request_data.get("applicants", [])
plans = request_data.get("plans", [])
phq = request_data.get("phq", {})
income = request_data.get("income", 0)
address = request_data.get("address", {})
if is_dtq:
tier = 5.0
dtq_found = True
message = "Declined due to high-risk conditions (DTQ triggered)."
elif age is None or bmi is None or base is None:
tier = 4.5
message = "Assigned fallback Tier 4.5 due to missing age or BMI."
else:
tier = base
tier = adjust_tier(tier, bmi, phq.get("medications", []),
phq.get("issues", []), idx)
# Build the request message
message_parts = ["Please provide an insurance estimation for the following application. Analyze all health factors, medications, conditions, and risk factors to determine appropriate tier assignments and pricing:"]
tier_price = PRICING_TABLE.get(tier, PRICING_TABLE[5])[coverage_index]
max_price = PRICING_TABLE[7][coverage_index]
# Add applicants information
message_parts.append("\nApplicants:")
for i, applicant in enumerate(applicants):
applicant_type = "primary" if i == 0 else "dependent" if i > 0 else "unknown"
message_parts.append(f"- {applicant.get('firstName', '')} {applicant.get('lastName', '')} ({applicant_type})")
message_parts.append(f" Age: {applicant.get('dob', '')}")
message_parts.append(f" Gender: {applicant.get('gender', '')}")
message_parts.append(f" Weight: {applicant.get('weight', '')} lbs")
message_parts.append(f" Height: {applicant.get('heightFt', '')}'{applicant.get('heightIn', '')}\"")
message_parts.append(f" Nicotine use: {'Yes' if applicant.get('nicotine') else 'No'}")
if rx_spend > max_price:
tier = 5.0
message = f"Declined due to high Rx spend (${rx_spend} > ${max_price})"
elif rx_spend > tier_price:
for t in sorted(PRICING_TABLE.keys()):
if PRICING_TABLE[t][coverage_index] >= rx_spend:
tier = t
break
message = f"Rx spend ${rx_spend} increased tier to {tier}."
else:
message = f"Tier {tier} assigned with Rx spend ${rx_spend} within allowed limits."
# Add plan information
if plans:
plan = plans[0]
message_parts.append(f"\nPlan: Coverage type {plan.get('coverage', '')}")
# Add PHQ information
if phq:
message_parts.append("\nHealth Information:")
if phq.get("medications"):
message_parts.append("Medications:")
for med in phq["medications"]:
message_parts.append(f"- {med.get('name', '')} ({med.get('dosage', '')} {med.get('frequency', '')})")
if phq.get("issues"):
message_parts.append("Health Issues:")
for issue in phq["issues"]:
message_parts.append(f"- {issue.get('key', '')}")
for detail in issue.get("details", []):
message_parts.append(f" * {detail.get('description', '')}")
if phq.get("conditions"):
message_parts.append("Conditions:")
for condition in phq["conditions"]:
message_parts.append(f"- {condition.get('description', '')}")
# Add income and address
message_parts.append(f"\nIncome: ${income:,.2f}")
if address:
addr_parts = []
if address.get("address1"):
addr_parts.append(address["address1"])
if address.get("address2"):
addr_parts.append(address["address2"])
if address.get("city") and address.get("state"):
addr_parts.append(f"{address['city']}, {address['state']}")
if address.get("zipcode"):
addr_parts.append(address["zipcode"])
if addr_parts:
message_parts.append(f"Address: {', '.join(addr_parts)}")
message_parts.append("\nPlease provide a detailed estimation including tier assignment, pricing, and individual applicant analysis.")
message_parts.append("\nConsider the following factors in your analysis:")
message_parts.append("- Age and BMI of each applicant")
message_parts.append("- Medical conditions and their severity")
message_parts.append("- Prescription medications and their costs")
message_parts.append("- Risk factors (nicotine use, etc.)")
message_parts.append("- Family coverage type and plan details")
message_parts.append("\nRespond with a JSON object that follows this exact schema:")
message_parts.append("{")
message_parts.append(' "status": "accepted" or "rejected",')
message_parts.append(' "details": {')
message_parts.append(' "dtq": boolean,')
message_parts.append(' "reason": "string",')
message_parts.append(' "tier": integer,')
message_parts.append(' "total_price": float')
message_parts.append(' },')
message_parts.append(' "results": [')
message_parts.append(' {')
message_parts.append(' "name": "string",')
message_parts.append(' "applicant_type": "string",')
message_parts.append(' "age": integer,')
message_parts.append(' "bmi": float,')
message_parts.append(' "tier": integer,')
message_parts.append(' "rx_spend": float,')
message_parts.append(' "message": "string"')
message_parts.append(' }')
message_parts.append(' ]')
message_parts.append('}')
return "\n".join(message_parts)
async def estimate_insurance(self, request_data: Dict[str, Any]) -> Dict[str, Any]:
"""Send estimation request to TALESTORM API and parse response"""
try:
# Create or get existing session with estimation agent
session_id = await session_service.create_session(agent_id=self.agent_id)
# Format the request as a natural language message
estimation_message = self._format_estimation_request(request_data)
# Send request to TALESTORM API
async with await self.get_client() as client:
response = await client.post(
"/chat/",
json={
"chat_session_id": session_id,
"user_message": estimation_message
}
)
if response.status_code != 200:
raise Exception(f"TALESTORM API error: {response.status_code} {response.text}")
# Parse the response
response_data = response.json()
message_content = response_data.get("message", "")
# Try to parse the JSON response from the message
try:
# The response should contain a JSON string within the message
# First, try to extract JSON from the message if it's wrapped in code blocks
if "```json" in message_content:
start = message_content.find("```json") + 7
end = message_content.find("```", start)
if end != -1:
message_content = message_content[start:end].strip()
elif "```" in message_content:
# Handle generic code blocks
start = message_content.find("```") + 3
end = message_content.find("```", start)
if end != -1:
message_content = message_content[start:end].strip()
# Clean up any leading/trailing whitespace and newlines
message_content = message_content.strip()
parsed_response = json.loads(message_content)
# Validate that the response has the expected structure
if not isinstance(parsed_response, dict):
raise ValueError("Response is not a JSON object")
required_fields = ["status", "details", "results"]
for field in required_fields:
if field not in parsed_response:
raise ValueError(f"Missing required field: {field}")
# Validate details structure
details = parsed_response.get("details", {})
required_details = ["dtq", "reason", "tier", "total_price"]
for field in required_details:
if field not in details:
raise ValueError(f"Missing required detail field: {field}")
# Validate results structure
results = parsed_response.get("results", [])
if not isinstance(results, list):
raise ValueError("Results must be an array")
for result in results:
required_result_fields = ["name", "applicant_type", "age", "bmi", "tier", "rx_spend", "message"]
for field in required_result_fields:
if field not in result:
raise ValueError(f"Missing required result field: {field}")
return parsed_response
except (json.JSONDecodeError, ValueError) as e:
# If the response is not valid JSON or doesn't match schema, create a fallback response
print(f"Failed to parse JSON response: {e}")
print(f"Message content: {message_content}")
return {
"status": "error",
"details": {
"dtq": False,
"reason": f"Unable to parse AI response: {str(e)}",
"tier": 4,
"total_price": 0.0
},
"results": []
}
except Exception as e:
tier = 4.5
rx_spend = 0.0
message = f"Fallback Tier 4.5 due to system error: {str(e)}"
max_tier = max(max_tier, tier)
results.append({
"name": name,
"applicant_type": applicant_type,
"age": age,
"bmi": round(bmi, 2) if bmi else None,
"tier": tier,
"rx_spend": rx_spend,
"message": message
})
total_price = get_rx_price(max_tier, coverage)
# Return error response
return {
"results": results,
"combined": {
"tier": max_tier,
"total_price": total_price,
"dtq": dtq_found,
"message": f"Final assigned tier is {max_tier} based on age, health, Rx cost, or fallback logic."
}
"status": "error",
"details": {
"dtq": False,
"reason": f"Estimation service error: {str(e)}",
"tier": 4,
"total_price": 0.0
},
"results": []
}
# Global estimation service instance
estimation_service = EstimationService()

11
src/services/session_service.py
View file

@ -26,7 +26,7 @@ class SessionService:
"""List available agents from talestorm-ai"""
async with await self.get_client() as client:
try:
response = await client.get("/agents/")
response = await client.get("/agents")
if response.status_code == 200:
return response.json()
else:
@ -57,12 +57,11 @@ class SessionService:
"""Create a new chat session in talestorm-ai"""
async with await self.get_client() as client:
if not agent_id:
if self.agent_id:
agent_id = self.agent_id
else:
agent_id = settings.TALESTORM_AGENT_ID
# Use the default agent ID for backward compatibility
agent_id = self.agent_id or settings.TALESTORM_AGENT_ID
response = await client.post("/sessions/", params={"agent_id": agent_id})
print(response.request.url)
session_data = response.json()
return str(session_data["id"])
@ -82,7 +81,7 @@ class SessionService:
"""List all sessions from talestorm-ai"""
async with await self.get_client() as client:
try:
response = await client.get("/sessions/")
response = await client.get("/sessions")
if response.status_code == 200:
return response.json()
return []