Spaces:
Sleeping
Sleeping
File size: 5,949 Bytes
6f113af 842ea04 6f113af 32c6457 6f113af 32c6457 6f113af 842ea04 6f113af ab28991 842ea04 6f113af 842ea04 6f113af |
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 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 |
import io
import re
import os
import docx
from PyPDF2 import PdfReader
import numpy as np
from PIL import Image
from doctr.models import ocr_predictor
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from typing import List, Literal
from agents import (
Agent,
AsyncOpenAI,
OpenAIChatCompletionsModel,
AgentOutputSchema,
AgentOutputSchemaBase,
enable_verbose_stdout_logging,
set_tracing_disabled
)
enable_verbose_stdout_logging()
set_tracing_disabled(True)
load_dotenv()
model = ocr_predictor(pretrained=True)
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()
API = os.getenv("GEM_API_KEY")
class AiInsights(BaseModel):
overallAssessment: str
keyHighlights: List[dict[str, str]]
dietaryRecommendations: List[str]
lifestyleAdvice: List[str]
precautions: List[str]
risks: List[str]
actions: List[dict[str, str]]
tips: List[str]
class KeyFinding(BaseModel):
test: str
value: int
unit: str
range: str
shortExplaination: str
status: Literal["Red", "Yellow", "Green"]
class AnalysisResult(BaseModel):
fileName: str
reportType: str
summary: str
keyFindings: List[KeyFinding]
aiInsights: AiInsights
client = AsyncOpenAI(
api_key = API,
base_url = "https://generativelanguage.googleapis.com/v1beta/openai/",
)
agent_model = OpenAIChatCompletionsModel(
model = "gemini-2.0-flash",
openai_client = client,
)
def format_json(result):
analyzer_results = analyzer.analyze(text=result, language='en')
anonymized_text = anonymizer.anonymize(text=result, analyzer_results=analyzer_results)
result_text = anonymized_text.text
pattern = r'(<PERSON>\s+[\w\s\-]+)'
hospital_pattern = r'(?i)\b(?:[A-Z][a-zA-Z]+(?:\s+|,|&)?){1,6}(hospital|lab|clinic|diagnostic|medical|centre|pathology)\b'
result_text = re.sub(r'[,.()\'"-]', ' ', result_text).strip()
result_text = re.sub(pattern, r'<NAME>', result_text)
result_text = re.sub(hospital_pattern, r'<HOSPITAL>', result_text,)
print(result_text)
return result_text
def extract_text(content ,pdf ,doc) -> str:
if pdf:
reader = PdfReader(io.BytesIO(content))
text = ''
for page in reader.pages:
text += page.extract_text() + '\n'
print(text)
return text.strip()
elif doc:
doc = docx.Document(io.BytesIO(content))
text = ''
for para in doc.paragraphs:
text += para.text + '\n'
print(text)
return text.strip()
else:
image = Image.open(io.BytesIO(content)).convert("RGB")
npImg = np.ascontiguousarray(np.array(image, dtype='uint8'))
ORCresult = model([npImg])
clean_jason = format_json(ORCresult.render())
print(clean_jason)
return clean_jason
Report_Agent = Agent(
name = "Report_Analysis_Agent",
instructions = """You are a Medical Report Analysis Agent.
Your role is to analyze uploaded medical test reports and generate clear, accurate health advice in structured JSON format.
Your Main Task:
1. Analyze the extracted medical text carefully.
2. Identify each test name, its result (user value), and the normal reference range.
3. Assign a flag to each test based on the result:
- Red: Critical or abnormal
- Yellow: Slightly out of range or borderline
- Green: Normal or safe
4. Provide a clear summary of the findings.
5. Offer relevant AI-driven health tips, highlight potential risks, and suggest dietary and lifestyle improvements.
6. Structure the output in the specified JSON format.
Response format: {'type': 'json_schema', 'json_schema': {'name': 'final_output', 'strict': False, 'schema': {'$defs': {'AiInsights': {'properties': {'overallAssessment': {'title': 'Overallassessment', 'type': 'string'}, 'keyHighlights': {'items': {'additionalProperties': {'type': 'string'}, 'type': 'object'}, 'title': 'Keyhighlights', 'type': 'array'}, 'dietaryRecommendations': {'items': {'type': 'string'}, 'title': 'Dietaryrecommendations', 'type': 'array'}, 'lifestyleAdvice': {'items': {'type': 'string'}, 'title': 'Lifestyleadvice', 'type': 'array'}, 'precautions': {'items': {'type': 'string'}, 'title': 'Precautions', 'type': 'array'}, 'risks': {'items': {'type': 'string'}, 'title': 'Risks', 'type': 'array'}, 'actions': {'items': {'additionalProperties': {'type': 'string'}, 'type': 'object'}, 'title': 'Actions', 'type': 'array'}, 'tips': {'items': {'type': 'string'}, 'title': 'Tips', 'type': 'array'}}, 'required': ['overallAssessment', 'keyHighlights', 'dietaryRecommendations', 'lifestyleAdvice', 'precautions', 'risks', 'actions', 'tips'], 'title': 'AiInsights', 'type': 'object'}, 'KeyFinding': {'properties': {'test': {'title': 'Test', 'type': 'string'}, 'value': {'title': 'Value', 'type': 'integer'}, 'unit': {'title': 'Unit', 'type': 'string'}, 'range': {'title': 'Range', 'type': 'string'}, 'status': {'enum': ['Red', 'Yellow', 'Green'], 'title': 'Status', 'type': 'string'}}, 'required': ['test', 'value', 'unit', 'range', 'status'], 'title': 'KeyFinding', 'type': 'object'}}, 'properties': {'fileName': {'title': 'Filename', 'type': 'string'}, 'reportType': {'title': 'Reporttype', 'type': 'string'}, 'summary': {'title': 'Summary', 'type': 'string'}, 'keyFindings': {'items': {'$ref': '#/$defs/KeyFinding'}, 'title': 'Keyfindings', 'type': 'array'}, 'aiInsights': {'$ref': '#/$defs/AiInsights'}},
'required': ['fileName', 'reportType', 'summary', 'keyFindings', 'aiInsights'], 'title': 'AnalysisResult', 'type': 'object'}}}
""",
model = agent_model,
output_type= AgentOutputSchema(AnalysisResult, strict_json_schema=False)
)
|