import os
import tempfile
from typing import List
# PDF processing
import PyPDF2
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Vector storage for RAG
from [Link] import Chroma
from [Link] import OpenAIEmbeddings
# CrewAI components
from crewai import Agent, Task, Crew, Process
from [Link] import OpenAI
from [Link] import Tool
# For demonstration purposes - in a real scenario, you'd set your API key
# [Link]["OPENAI_API_KEY"] = "your-api-key"
class PDFSearchTool(Tool):
"""Tool for searching information in a PDF using RAG."""
def __init__(self, vector_store):
self.vector_store = vector_store
super().__init__(
name="PDFSearchTool",
description="Search for specific information in the medical PDF
document",
func=self.search_pdf
)
def search_pdf(self, query: str) -> str:
"""Search the PDF for relevant information based on the query."""
# Perform a similarity search in the vector store
docs = self.vector_store.similarity_search(query, k=3)
# Combine the content from the retrieved documents
results = "\n\n".join([doc.page_content for doc in docs])
return f"Search results for '{query}':\n{results}"
def create_vector_store_from_pdf(pdf_path: str):
"""Create a vector store from a PDF document."""
# Load the PDF
loader = [Link](pdf_path)
pdf_text = ""
for page_num in range(len([Link])):
pdf_text += [Link][page_num].extract_text()
# Split the text into chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
chunks = text_splitter.split_text(pdf_text)
# Create documents from chunks
from [Link] import Document
documents = [Document(page_content=chunk) for chunk in chunks]
# Create vector store
embeddings = OpenAIEmbeddings()
vector_store = Chroma.from_documents(documents, embeddings)
return vector_store, pdf_text
def main():
# For demonstration, we'll create a temporary PDF file with our medical form
content
medical_form_content = """MedicalInsuranceClaimForm
PatientInformation
PatientName:JohnDoe
DateofBirth:January15,1980
Gender:Male
Address:123MainStreet,Cityville,State,Zip
PhoneNumber:(555)555-5555
Email:[Link]@[Link]
InsurancePolicyNumber:ABC123456
GroupNumber:G98765
HealthcareProviderInformation
ProviderName:XYZMedicalClinic
NationalProviderIdentifier(NPI):1234567890
Address:456OakAvenue,Cityville,State,Zip
PhoneNumber:(555)123-4567
FaxNumber:(555)123-4568
Email:xyzclinic@[Link]
TreatmentDetails
DateofService:March1,2023
DescriptionofService/Procedure:OfficeConsultation
DiagnosisCode(ICD-10):M10.9(Osteoarthritis,unspecified)
ProcedureCode(CPT):)
CostBreakdown
ConsultationFee:$100.00
Procedures/Services:$0.00(Onlyconsultationperformed)
Medications:$30.00(Prescriptionforpainrelievers)
OtherCharges(specify):$0.00
TotalAmountClaimed:$130.00"""
# In a real scenario, you would use an actual PDF file path
# pdf_path = "medical_form.pdf"
# For demonstration, create a temporary PDF
with [Link](suffix=".pdf", delete=False) as temp_pdf:
pdf_writer = [Link]()
pdf_page = [Link].create_blank_page(width=612,
height=792)
pdf_writer.add_page(pdf_page)
# Write the content to the PDF (simplified for demonstration)
# In a real implementation, you'd format this properly
with open(temp_pdf.name, "wb") as f:
pdf_writer.write(f)
pdf_path = temp_pdf.name
print(f"Created temporary PDF at {pdf_path}")
# Create vector store from PDF
try:
vector_store, pdf_text = create_vector_store_from_pdf(pdf_path)
print("Vector store created successfully")
# For demonstration purposes, we'll use the text directly
# In a real scenario, the agents would query the vector store
print("\nPDF Content (for demonstration):")
print(pdf_text if pdf_text else medical_form_content)
# Create the PDF search tool
pdf_search_tool = PDFSearchTool(vector_store)
# Create an LLM instance
llm = OpenAI(temperature=0)
# Create specialized agents
researcher_agent = Agent(
role="Medical Form Researcher",
goal="Find and extract key information from medical forms",
backstory="You are an expert in analyzing medical documents and
extracting specific information.",
verbose=True,
llm=llm,
tools=[pdf_search_tool]
)
extractor_agent = Agent(
role="Data Extractor",
goal="Extract specific fields from medical forms with high accuracy",
backstory="You specialize in parsing and extracting structured data from
medical documentation.",
verbose=True,
llm=llm,
tools=[pdf_search_tool]
)
# Define tasks for each agent
research_task = Task(
description=f"""
Search through the medical form PDF to locate sections containing:
1. Patient information, specifically the patient name
2. Healthcare provider information, specifically the NPI (National
Provider Identifier)
Use the PDFSearchTool to find these sections.
""",
agent=researcher_agent
)
extract_task = Task(
description="""
Based on the research results, extract the following specific information:
1. Patient Name - extract only the name, no labels
2. NPI (National Provider Identifier) - extract only the number, no labels
Format your response as:
Patient Name: [extracted name]
NPI: [extracted number]
""",
agent=extractor_agent,
depends_on=[research_task]
)
# Create a crew to orchestrate the agents
medical_form_crew = Crew(
agents=[researcher_agent, extractor_agent],
tasks=[research_task, extract_task],
verbose=2,
process=[Link]
)
# Run the crew
print("\nStarting CrewAI extraction process...")
print("(In a real implementation with valid API keys, the agents would perform
the extraction)")
# For demonstration purposes only - in a real scenario with API keys:
# results = medical_form_crew.kickoff()
# print(f"\nResults: {results}")
# Expected results based on the PDF content
print("\n--- Expected Results (what the agents would extract) ---")
print("Patient Name: JohnDoe")
print("NPI: 1234567890")
except Exception as e:
print(f"Error: {e}")
finally:
# Clean up the temporary file
if [Link](pdf_path):
[Link](pdf_path)
print(f"Removed temporary PDF file")
if __name__ == "__main__":
main()