Spaces:
Sleeping
Sleeping
File size: 3,109 Bytes
4cccee3 59d8519 4cccee3 532a092 4cccee3 532a092 4cccee3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 |
# main.py
import os
import json
import uuid
from fastapi import FastAPI, UploadFile, File, Form
from fastapi.responses import JSONResponse
from dotenv import load_dotenv
from utils.loader import extract_text_from_pdf
from utils.evaluator import evaluate
from utils.parser import parse_query_with_gemini
import google.generativeai as genai
from fastapi.responses import RedirectResponse
# Load environment variables
load_dotenv()
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
print("Loaded Gemini API Key:", os.getenv("GEMINI_API_KEY"))
app = FastAPI()
# Ensure data directory exists
os.makedirs("data/documents", exist_ok=True)
# @app.get("/")
# def root():
# return {"message": "LLM Claims API is up and running!"}
@app.get("/")
def redirect_to_docs():
return RedirectResponse(url="/docs")
@app.post("/evaluate")
async def evaluate_query(query: str = Form(...), file: UploadFile = File(...)):
# Save uploaded file
file_id = str(uuid.uuid4())
file_path = f"data/documents/{file_id}.pdf"
with open(file_path, "wb") as f:
f.write(await file.read())
try:
# Extract and parse
policy_text = extract_text_from_pdf(file_path)
parsed_query = await parse_query_with_gemini(query) \
if callable(getattr(parse_query_with_gemini, "__await__", None)) else parse_query_with_gemini(query)
gemini_response = await query_gemini(policy_text, query)
rule_decision = evaluate(parsed_query, gemini_response.get("matched_clause", ""))
final_result = {
**gemini_response,
"parsed_query": parsed_query,
"rule_based_decision": rule_decision,
}
except Exception as e:
final_result = {
"error": str(e)
}
finally:
if os.path.exists(file_path):
os.remove(file_path)
return JSONResponse(content=final_result)
async def query_gemini(policy_text: str, query_text: str):
model = genai.GenerativeModel("models/gemini-1.5-flash-latest")
prompt = f"""
You are an insurance claim evaluator. Based on the policy document and query, respond in JSON with:
1. decision: 'approved' or 'rejected'
2. justification: brief explanation
3. amount: estimated payout
4. matched_clause: snippet of the policy that supports the decision
5. similarity_score: float between 0 and 1
Policy:
{policy_text}
Query:
{query_text}
"""
try:
response = model.generate_content(prompt)
content = response.text.strip()
# Clean markdown-style code formatting
if content.startswith("```json") or content.startswith("```"):
content = content.replace("```json", "").replace("```", "").strip()
return json.loads(content)
except Exception as e:
return {
"decision": "rejected",
"justification": f"Gemini Error: {str(e)}",
"amount": "₹0",
"matched_clause": "",
"similarity_score": 0.0
}
|