AI agentic workflows: Tutorial & Best Practices

import os
import sys
import io
from typing import Dict, Any, TypedDict
import json
import os
from datetime import datetime
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from PIL import Image
import fitz  # PyMuPDF library for PDF processing

# -----------------------------------------------------------------------------
# State Definition: Tracks document information throughout processing
# -----------------------------------------------------------------------------
class WorkflowState(TypedDict):
    """Represents the state of a document as it's processed through the workflow.
   
    Attributes:
        file_path: Path to the document being processed
        file_type: Type of file (pdf, image, docx)
        is_text_readable: Whether the document can be processed with direct text extraction
        result: The final output after document processing
    """
    file_path: str          # Path to input document
    file_type: str          # Type of document (pdf, image, docx)
    is_text_readable: bool  # Whether text can be directly extracted
    result: Dict[str, Any]  # Processing results from specialized agents

def extract_text_from_pdf(file_path, is_text_readable):
    """Extract text from a PDF file using the appropriate method based on PDF type.
   
    This function uses two different approaches:
    1. For text-readable PDFs: Direct text extraction with PyPDF2
    2. For scanned PDFs: OCR using GPT-4o Vision capabilities
   
    Args:
        file_path: Path to the PDF file
        is_text_readable: Whether the PDF contains directly extractable text
       
    Returns:
        Extracted text content as a string
    """
    # APPROACH 1: Text-readable PDF - Use standard text extraction
    if is_text_readable:
        import PyPDF2
        text = ""
        with open(file_path, 'rb') as file:
            pdf_reader = PyPDF2.PdfReader(file)
            # Extract text from each page and combine
            for page in pdf_reader.pages:
                text += page.extract_text()
        return text
   
    # APPROACH 2: Scanned PDF - Use vision-based OCR with GPT-4o
    else:
        try:
            print("Processing scanned PDF with GPT-4o Vision...")
            # Step 1: Convert the first page to an image
            doc = fitz.open(file_path)
            pix = doc[0].get_pixmap()  # Render page to image
            img_data = pix.tobytes("png")  # Convert to PNG bytes
            image = Image.open(io.BytesIO(img_data))  # Create PIL Image
           
            # Step 2: Use GPT-4o's vision capabilities to extract text
            return extract_text_with_gpt4o_vision(image)
        except Exception as e:
            print(f"Error extracting text from scanned PDF: {e}")
            return f"Error processing scanned PDF: {str(e)}"


# -----------------------------------------------------------------------------
# DOCX Processing: Extract text from Word documents
# -----------------------------------------------------------------------------
def extract_text_from_docx(file_path):
    """Extract text from Microsoft Word DOCX files.
   
    Uses the python-docx library to extract text from all paragraphs
    in the document and combines them with line breaks.
   
    Args:
        file_path: Path to the DOCX file
       
    Returns:
        Extracted text content as a string
    """
    import docx
    doc = docx.Document(file_path)
    # Extract text from all paragraphs and join with newlines
    return "\n".join([para.text for para in doc.paragraphs])

def extract_text_with_gpt4o_vision(image):
    """Extract text from an image using GPT-4o vision capabilities.
   
    Args:
        image: PIL.Image object
       
    Returns:
        str: Extracted text from the image
    """
    try:
        # Initialize the GPT-4o model with vision capabilities
        llm = ChatOpenAI(model="gpt-4o")
       
        # Convert PIL image to base64 encoded string
        buffered = io.BytesIO()
        image.save(buffered, format="PNG")
        img_str = base64.b64encode(buffered.getvalue()).decode()
       
        # Prepare the message with the image content using the correct format
        from langchain_core.messages import HumanMessage
       
        message = HumanMessage(
            content=[
                {
                    "type": "text",
                    "text": "Extract all text visible in this image. Output only the text content, nothing else."
                },
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/png;base64,{img_str}"}
                }
            ]
        )
       
        # Get response from GPT-4o
        response = llm.invoke([message])
       
        # Extract the text content from the response
        extracted_text = response.content
       
        return extracted_text
    except Exception as e:
        print(f"Error in GPT-4o vision text extraction: {e}")
        return f"Error in GPT-4o vision text extraction: {str(e)}"

# -----------------------------------------------------------------------------
# Image Processing: Extract text from image files
# -----------------------------------------------------------------------------

def check_file_type(file_path):
    """
    Determine the type of file and whether it's text-readable or requires OCR.
   
    Returns:
        tuple: (file_type, is_text_readable)
    """
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"File not found: {file_path}")
   
    # Initialize mimetypes
    mimetypes.init()
   
    # Get file extension and determine MIME type
    file_ext = os.path.splitext(file_path)[1].lower()
    mime_type, _ = mimetypes.guess_type(file_path)
   
    # Handle cases where mimetypes doesn't recognize the file
    if not mime_type:
        if file_ext == '.pdf':
            mime_type = 'application/pdf'
        elif file_ext == '.docx':
            mime_type = 'application/vnd.openxmlformats-officedocument.wordprocessingml.document'
        elif file_ext in ['.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.gif']:
            mime_type = f'image/{file_ext[1:]}'
   
    print(f"Detected MIME type: {mime_type}")
   
    # Image files
    if mime_type and mime_type.startswith('image/'):
        return "image", False
   
    # PDF files
    elif mime_type == 'application/pdf' or file_ext == '.pdf':
        # Check if PDF is text-readable or scanned
        return check_pdf_readability(file_path)
   
    # DOCX files
    elif mime_type == 'application/vnd.openxmlformats-officedocument.wordprocessingml.document' or file_ext == '.docx':
        return "docx", True
   
    else:
        raise ValueError(f"Unsupported file type: {mime_type or file_ext}")



def extract_text_from_image(file_path):
    """Extract text from image files using GPT-4o's advanced vision capabilities.
   
    This function leverages GPT-4o's multimodal abilities to perform OCR on images,
    which provides more accurate text extraction than traditional OCR tools.  
   
    Args:
        file_path: Path to the image file
       
    Returns:
        Extracted text content as a string
    """
    try:
        # Load the image using PIL
        image = Image.open(file_path)
       
        # Use GPT-4o vision model to extract text
        return extract_text_with_gpt4o_vision(image)
    except Exception as e:
        print(f"Error extracting text from image: {e}")
        return f"Error processing image: {str(e)}"

class EntityExtractorAgent:
    """Agent specialized for extracting structured information from scanned documents/images."""

    def __init__(self, llm):
        """Initialize the agent with a language model.
       
        Args:
            llm: Language model for entity extraction
        """
        self.llm = llm
       
        # Define the prompt that guides the model to extract specific entity types
        self.prompt = ChatPromptTemplate.from_template("""
        You are an expert entity extraction agent.
        Your task is to extract key entities from the following text that was obtained from a scanned document or image.
       
        Please extract the following entities:
        1. Names of people
        2. Organizations
        3. Dates
        4. Locations
        5. Key numerical values (amounts, percentages, etc.)
       
        Format your response as a JSON object with these categories as keys.
       
        Text content:
        {text_content}
        """)
   def process(self, text_content):
        """Process extracted text to identify and categorize entities.
       
        Args:
            text_content: The text extracted from the document
           
        Returns:
            Structured dictionary of extracted entities by category
        """
       
        print("Entity Extractor Agent: Processing text content...")
        # Invoke the processing chain with the extracted text
        result = self.chain.invoke({"text_content": text_content})
       
       
        return result

class DocumentSummarizerAgent:
    """Agent specialized for creating structured summaries of text-readable documents."""
   

    def __init__(self, llm):
        """Initialize the agent with a language model.
       
        Args:
            llm: Language model for document summarization
        """
        self.llm = llm
       
        # Define the prompt that guides the model to create specific summary components
        self.prompt = ChatPromptTemplate.from_template("""
        You are an expert document summarization agent.
        Your task is to create a comprehensive summary of the following document.
       
        Please provide:
        1. A concise executive summary (2-3 sentences)
        2. Key points (bullet points)
        3. Main topics covered
        4. Important conclusions or recommendations
       
        Format your response as a JSON object with these categories as keys.
       
        Document content:
        {text_content}
        """)
       
        # Configure JSON output parser to ensure structured responses
        self.output_parser = JsonOutputParser()
       
        # Create the LangChain processing pipeline: prompt → LLM → JSON parser
        self.chain = self.prompt | self.llm | self.output_parser


   def process(self, text_content):
        """Process document text to create a structured summary.
       
        Args:
            text_content: The text extracted from the document
           
        Returns:
            Structured dictionary with summary components
        """
       
        print("Document Summarizer Agent: Processing text content...")
        # Invoke the processing chain with the extracted text
        result = self.chain.invoke({"text_content": text_content})
       
       
        return result

# Router function
def route_document(file_path):
    """Route the document to the appropriate agent based on file type"""
    print(f"\nRouting document: {file_path}")
   
    # Initialize state
    state = WorkflowState(
        file_path=file_path,
        file_type="",
        is_text_readable=False,
        result={}
    )
   
    # Check file type
    file_type, is_text_readable = check_file_type(file_path)
    state["file_type"] = file_type
    state["is_text_readable"] = is_text_readable
   
    print(f"Router Agent: File type detected: {file_type}, Text readable: {is_text_readable}")
   
    # Initialize LLM
    llm = ChatOpenAI(model="gpt-4o")
   
    # Extract text based on file type
    if file_type == "pdf":
        text_content = extract_text_from_pdf(file_path, is_text_readable)
    elif file_type == "image":
        text_content = extract_text_from_image(file_path)
    elif file_type == "docx":
        text_content = extract_text_from_docx(file_path)
    else:
        return {"error": f"Unsupported file type: {file_type}"}
   
    # Preview the extracted text
    print("\nExtracted text preview:")
    print(text_content[:500] + "..." if len(text_content) > 500 else text_content)
   
    # Use OpenAI LLM to decide which agent to route to
    print("\nAsking Router LLM to determine the best agent for processing...")
   
    # Create a router prompt for the LLM to decide on the appropriate agent
    router_prompt = ChatPromptTemplate.from_template(
        """
        You are a document routing agent that needs to decide which specialized agent
        should process a document based on its content and format.
       
        Document Type: {file_type}
        Is Text Readable: {is_text_readable}
       
        Document Content Preview:
        {text_preview}
       
        Based on this information, determine which agent should process this document:
        - Entity Extractor Agent: Specialized in extracting structured entities from scanned documents,
          images, or non-text-readable documents. Good for processing forms, invoices, receipts, etc.
        - Document Summarizer Agent: Specialized in summarizing and extracting key information from
          text-readable documents like reports, articles, or regular PDFs with extractable text.
       
        Output ONLY 'entity_extractor' or 'document_summarizer' based on your decision.
        """
    )
   
    # Execute the router LLM call
    text_preview = text_content[:1000] + "..." if len(text_content) > 1000 else text_content
    router_input = {
        "file_type": file_type,
        "is_text_readable": is_text_readable,
        "text_preview": text_preview
    }
   
    router_response = llm.invoke(router_prompt.format_messages(**router_input))
    routing_decision = router_response.content.strip().lower()
   
    # Apply the LLM's routing decision
    if "entity_extractor" in routing_decision:
        print("\nLLM Routing Decision: Entity Extractor Agent")
        entity_extractor = EntityExtractorAgent(llm)
       
        result = entity_extractor.process(text_content)
       
        state["result"] = {
            "source": "entity_extractor",
            "extracted_entities": result
        }
    elif "document_summarizer" in routing_decision:
        print("\nLLM Routing Decision: Document Summarizer Agent")
        document_summarizer = DocumentSummarizerAgent(llm)
       
        result = document_summarizer.process(text_content)
       
        state["result"] = {
            "source": "document_summarizer",
            "document_summary": result
        }
    else:
        # Fallback in case the LLM doesn't return a clear decision
        print(f"\nUnclear routing decision: '{routing_decision}', using default routing logic")
        if file_type == "image" or (file_type == "pdf" and not is_text_readable):
            print("Defaulting to Entity Extractor Agent...")
            entity_extractor = EntityExtractorAgent(llm)
           
            result = entity_extractor.process(text_content)
           
            state["result"] = {
                "source": "entity_extractor",
                "extracted_entities": result
            }
        else:
            print("Defaulting to Document Summarizer Agent...")
            document_summarizer = DocumentSummarizerAgent(llm)
           
            result = document_summarizer.process(text_content)
           
            state["result"] = {
                "source": "document_summarizer",
                "document_summary": result
            }
   
    return state
    

def save_output_to_file(result):
    """Save processing results to organized output files.
   
    This function creates a structured output directory with:
    1. Agent-specific results based on which agent processed the document
    2. Complete results with all workflow data
   
    Args:
        result: The final state dictionary with processing results
       
    Returns:
        Path to the output directory
    """

   
    # Create output directory with timestamp
    output_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "outputs")
    os.makedirs(output_dir, exist_ok=True)
   
    # Generate a timestamp for unique filenames
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
   
    # STEP 1: Save agent-specific results based on which agent processed the document
   
    # For documents processed by the Entity Extractor
    if result["result"].get("source") == "entity_extractor":
        output_file = os.path.join(output_dir, f"entity_extraction_results_{timestamp}.json")
        with open(output_file, "w") as f:
            json.dump(result["result"]["extracted_entities"], f, indent=4)
        print(f"Entity extraction results saved to: {output_file}")
   
    # For documents processed by the Document Summarizer
    elif result["result"].get("source") == "document_summarizer":
        output_file = os.path.join(output_dir, f"document_summary_results_{timestamp}.json")
        with open(output_file, "w") as f:
            json.dump(result["result"]["document_summary"], f, indent=4)
        print(f"Document summary results saved to: {output_file}")
   
    # STEP 2: Save the complete workflow results
    full_results_file = os.path.join(output_dir, f"full_results_{timestamp}.json")
    with open(full_results_file, "w") as f:
        # Convert any non-serializable objects to strings (like Path objects)
        json_result = {k: str(v) if not isinstance(v, (dict, list, str, int, float, bool, type(None))) else v
                      for k, v in result.items()}
        json.dump(json_result, f, indent=4)
    print(f"Complete results saved to: {full_results_file}")
   
    return output_dir

# -----------------------------------------------------------------------------
# Main Application: Entry point for document processing
# -----------------------------------------------------------------------------

def main():
    """Main function to run the document processing workflow.
   
    This function:  
    1. Gets the document path from the command line or uses a default
    2. Processes the document through the router workflow
    3. Saves all outputs to organized files
    """    
   
    # STEP 1: Get the document to process (from command line or default)
    if len(sys.argv) > 1:
        file_path = sys.argv[1]
    else:
        # Default to a sample document if none provided
        file_path = "C:\\Users\\sample_invoice_scanned.pdf"
        print(f"Using default file path: {file_path}")
   
    # STEP 2: Process the document
    # Process the document using our routing system
    result = route_document(file_path)
   
    # STEP 3: Display the processing results
    print("\n--- Processing Result ---")
    print(result)  # Show the final state with processing results
   
    # STEP 4: Save all outputs to organized files
    output_dir = save_output_to_file(result)
   


# Entry point for running the application
if __name__ == "__main__":
    try:
        main()
    finally:
        # Ensure logs are saved even if there's an error
        # logger.save_logs()
           pass

{
    "Names of people": [
        "Jack Reilly",
        "Jim Daley"
    ],
    "Organizations": [
        "Shook, Hardy & Bacon L.L.P.",
        "Federal Express",
        "Legal Edge"
    ],
    "Dates": [
        "January 31, 1997",
        "February 5"
    ],
    "Locations": [
        "One Kansas City Place",
        "1200 Main Street",
        "Kansas City, Missouri 64105-2118"
    ],
    "Key numerical values": [
        {
            "telephone": "(816) 474-6550",
            "facsimile": "(816) 421-5547",
            "telecopy no.": "212/545-3297",
            "pages transmitted": 3,
            "client matter no.": "LORI:45048",
            "operator extension": 200,
            "other number": 83772145
        }
    ]
}