Llama Cloud Services: Revolutionizing Document Processing với GenAI#

1
# Advanced parsing với custom instructions
2
parser = LlamaParse(
3
    api_key="YOUR_API_KEY",
4
    result_type="markdown",  # markdown, text, json
5
    parsing_instruction="""
6
    Extract all financial data with high precision.
7
    Preserve table structures and numerical relationships.
8
    Identify key performance indicators and metrics.
9
    """
10
)
11

12
# Batch processing với metadata
13
documents = parser.load_data([
14
    "./financial_report_2024.pdf",
15
    "./quarterly_earnings.xlsx",
16
    "./market_analysis.docx"
17
])

1
# Financial report analysis
2
financial_parser = LlamaParse(
3
    api_key=API_KEY,
4
    parsing_instruction="""
5
    Focus on:
6
    - Revenue and profit margins
7
    - Cash flow statements
8
    - Balance sheet items
9
    - Key financial ratios
10
    - Risk factors and footnotes
11
    """
12
)
13

14
reports = financial_parser.load_data("./annual_report_2024.pdf")

1
# Contract và legal document parsing
2
legal_parser = LlamaParse(
3
    api_key=API_KEY,
4
    parsing_instruction="""
5
    Extract:
6
    - Key terms and conditions
7
    - Important dates and deadlines
8
    - Party obligations and responsibilities
9
    - Penalty clauses and conditions
10
    - Signature requirements
11
    """
12
)
13

14
contracts = legal_parser.load_data("./service_agreement.pdf")

1
# Academic và research document parsing
2
research_parser = LlamaParse(
3
    api_key=API_KEY,
4
    parsing_instruction="""
5
    Structure the content to highlight:
6
    - Abstract and key findings
7
    - Methodology and experimental setup
8
    - Results and statistical data
9
    - Citations and references
10
    - Tables and figures with captions
11
    """
12
)
13

14
papers = research_parser.load_data("./research_paper.pdf")

1
from llama_cloud_services import LlamaExtract
2

3
# Define extraction schema
4
extraction_schema = {
5
    "company_info": {
6
        "name": "string",
7
        "industry": "string",
8
        "headquarters": "string",
9
        "founded_year": "integer"
10
    },
11
    "financial_metrics": {
12
        "revenue": "number",
13
        "profit_margin": "number",
14
        "employees": "integer",
15
        "market_cap": "number"
16
    },
17
    "key_executives": [
18
        {
19
            "name": "string",
20
            "position": "string",
21
            "tenure": "string"
22
        }
23
    ]
24
}
25

26
extractor = LlamaExtract(api_key=API_KEY)
27
result = extractor.extract(
28
    file_path="./company_profile.pdf",
29
    schema=extraction_schema
30
)

1
# Invoice data extraction
2
invoice_schema = {
3
    "invoice_details": {
4
        "invoice_number": "string",
5
        "date": "date",
6
        "due_date": "date",
7
        "total_amount": "number",
8
        "currency": "string"
9
    },
10
    "vendor_info": {
11
        "name": "string",
12
        "address": "string",
13
        "tax_id": "string"
14
    },
15
    "line_items": [
16
        {
17
            "description": "string",
18
            "quantity": "number",
19
            "unit_price": "number",
20
            "total": "number"
21
        }
22
    ]
23
}
24

25
invoice_extractor = LlamaExtract(api_key=API_KEY)
26
extracted_data = invoice_extractor.extract(
27
    file_path="./invoice_batch/",
28
    schema=invoice_schema,
29
    batch_processing=True
30
)

1
# Resume processing for HR
2
resume_schema = {
3
    "personal_info": {
4
        "name": "string",
5
        "email": "string",
6
        "phone": "string",
7
        "location": "string"
8
    },
9
    "experience": [
10
        {
11
            "company": "string",
12
            "position": "string",
13
            "duration": "string",
14
            "responsibilities": ["string"]
15
        }
16
    ],
17
    "education": [
18
        {
19
            "institution": "string",
20
            "degree": "string",
21
            "graduation_year": "integer"
22
        }
23
    ],
24
    "skills": ["string"],
25
    "certifications": ["string"]
26
}
27

28
hr_extractor = LlamaExtract(api_key=API_KEY)
29
candidate_data = hr_extractor.extract(
30
    file_path="./resumes/",
31
    schema=resume_schema
32
)

1
from llama_cloud_services import LlamaCloudIndex
2

3
# Create và configure index
4
index = LlamaCloudIndex(
5
    "knowledge_base_v1",
6
    project_name="enterprise_docs",
7
    api_key=API_KEY,
8
    embedding_model="text-embedding-3-large",
9
    chunk_size=1024,
10
    chunk_overlap=128
11
)
12

13
# Batch document ingestion
14
documents = [
15
    "./company_policies/",
16
    "./technical_docs/",
17
    "./training_materials/",
18
    "./compliance_docs/"
19
]
20

21
index.insert_files(documents)

1
# Semantic search với context
2
search_results = index.query(
3
    "What are our data privacy policies for international customers?",
4
    similarity_top_k=10,
5
    response_mode="tree_summarize",
6
    streaming=True
7
)
8

9
# Multi-modal search
10
complex_query = index.query(
11
    query="Revenue trends in Q4 2024",
12
    filters={"document_type": "financial", "year": 2024},
13
    include_metadata=True,
14
    return_sources=True
15
)

1
# Dockerfile for Llama Cloud Services application
2
FROM python:3.11-slim
3

4
WORKDIR /app
5

6
# Install dependencies
7
COPY requirements.txt .
8
RUN pip install -r requirements.txt
9

10
# Copy application code
11
COPY . .
12

13
# Environment variables
14
ENV LLAMA_CLOUD_API_KEY=""
15
ENV ENVIRONMENT="production"
16
ENV LOG_LEVEL="INFO"
17

18
# Health check endpoint
19
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
20
    CMD curl -f http://localhost:8000/health || exit 1
21

22
# Run application
23
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

1
apiVersion: apps/v1
2
kind: Deployment
3
metadata:
4
  name: llama-cloud-app
5
spec:
6
  replicas: 3
7
  selector:
8
    matchLabels:
9
      app: llama-cloud-app
10
  template:
11
    metadata:
12
      labels:
13
        app: llama-cloud-app
14
    spec:
15
      containers:
16
      - name: app
17
        image: llama-cloud-app:latest
18
        ports:
19
        - containerPort: 8000
20
        env:
21
        - name: LLAMA_CLOUD_API_KEY
22
          valueFrom:
23
            secretKeyRef:
24
              name: llama-cloud-secret
25
              key: api-key
26
        resources:
27
          requests:
28
            memory: "512Mi"
29
            cpu: "250m"
30
          limits:
31
            memory: "1Gi"
32
            cpu: "500m"

1
# Monitoring setup với Prometheus
2
from prometheus_client import Counter, Histogram, start_http_server
3
import time
4

5
# Metrics definition
6
REQUEST_COUNT = Counter('llama_cloud_requests_total', 'Total requests', ['service', 'status'])
7
REQUEST_DURATION = Histogram('llama_cloud_request_duration_seconds', 'Request duration')
8

9
class MonitoredLlamaCloudClient:
10
    def __init__(self, api_key):
11
        self.parser = LlamaParse(api_key=api_key)
12
        self.extractor = LlamaExtract(api_key=api_key)
13
        self.index = LlamaCloudIndex("monitored_index", api_key=api_key)
14

15
    @REQUEST_DURATION.time()
16
    def parse_document(self, file_path):
17
        try:
18
            result = self.parser.load_data(file_path)
19
            REQUEST_COUNT.labels(service='parse', status='success').inc()
20
            return result
21
        except Exception as e:
22
            REQUEST_COUNT.labels(service='parse', status='error').inc()
23
            raise e
24

25
# Start metrics server
26
start_http_server(8080)

1
class OptimizedDocumentProcessor:
2
    def __init__(self, api_key, batch_size=10):
3
        self.client = LlamaParse(api_key=api_key)
4
        self.batch_size = batch_size
5
        self.cache = {}
6

7
    def process_documents_efficiently(self, document_paths):
8
        # Group similar documents
9
        grouped_docs = self.group_by_type(document_paths)
10

11
        results = []
12
        for doc_type, docs in grouped_docs.items():
13
            # Process in optimized batches
14
            for batch in self.create_batches(docs, self.batch_size):
15
                batch_results = self.process_batch(batch, doc_type)
16
                results.extend(batch_results)
17

18
        return results
19

20
    def process_batch(self, batch, doc_type):
21
        # Use type-specific optimization
22
        parsing_instruction = self.get_optimized_instruction(doc_type)
23

24
        return self.client.load_data(
25
            batch,
26
            parsing_instruction=parsing_instruction
27
        )

1
# Secure API key handling
2
import os
3
from cryptography.fernet import Fernet
4

5
class SecureLlamaCloudClient:
6
    def __init__(self):
7
        # Never hardcode API keys
8
        self.api_key = self.get_secure_api_key()
9

10
    def get_secure_api_key(self):
11
        # Use environment variables
12
        encrypted_key = os.getenv('LLAMA_CLOUD_API_KEY_ENCRYPTED')
13
        encryption_key = os.getenv('ENCRYPTION_KEY')
14

15
        if encrypted_key and encryption_key:
16
            f = Fernet(encryption_key.encode())
17
            return f.decrypt(encrypted_key.encode()).decode()
18

19
        # Fallback to environment variable
20
        return os.getenv('LLAMA_CLOUD_API_KEY')

1
# Enterprise financial document processing
2
class FinancialDocumentProcessor:
3
    def __init__(self):
4
        self.parser = LlamaParse(
5
            api_key=API_KEY,
6
            parsing_instruction="""
7
            Extract financial data with regulatory compliance focus:
8
            - SEC filing requirements
9
            - Risk disclosures
10
            - Financial statements
11
            - Audit information
12
            """
13
        )
14

15
        self.extractor = LlamaExtract(api_key=API_KEY)
16
        self.index = LlamaCloudIndex("financial_compliance")
17

18
    def process_regulatory_filings(self, filing_paths):
19
        # Compliance-focused processing
20
        results = []
21
        for filing in filing_paths:
22
            parsed = self.parser.load_data(filing)
23

24
            # Extract key regulatory data
25
            extracted = self.extractor.extract(
26
                content=parsed.text,
27
                schema=self.get_regulatory_schema()
28
            )
29

30
            # Index for compliance searches
31
            self.index.insert(parsed, metadata=extracted)
32
            results.append(extracted)
33

34
        return results

1
flowchart TD
2
    A[Medical Documents] --> B[HIPAA-Compliant Processing]
3
    B --> C[LlamaParse Medical]
4
    C --> D[PHI Detection & Masking]
5
    D --> E[Clinical Data Extraction]
6
    E --> F[Medical Knowledge Index]
7

8
    F --> G[Clinical Decision Support]
9
    F --> H[Research Analytics]
10
    F --> I[Patient Care Optimization]
11

12
    subgraph "Security Layer"
13
        J[Encryption at Rest]
14
        K[Access Controls]
15
        L[Audit Logging]
16
    end
17

18
    B --> J
19
    C --> K
20
    E --> L
21

22
    style B fill:#ffebee
23
    style D fill:#fff3e0
24
    style F fill:#e8f5e8

1
# Future multimodal capabilities
2
class NextGenProcessor:
3
    def __init__(self):
4
        self.multimodal_parser = LlamaParse(
5
            version="2.0",
6
            capabilities=[
7
                "video_analysis",
8
                "audio_transcription",
9
                "3d_document_understanding",
10
                "real_time_processing"
11
            ]
12
        )
13

14
    def process_multimedia_content(self, content):
15
        # Advanced multimodal processing
16
        return self.multimodal_parser.analyze(
17
            content,
18
            modalities=["text", "image", "video", "audio"],
19
            cross_modal_reasoning=True
20
        )

1
import logging
2
from tenacity import retry, stop_after_attempt, wait_exponential
3

4
class RobustLlamaCloudClient:
5
    def __init__(self, api_key):
6
        self.client = LlamaParse(api_key=api_key)
7
        self.logger = logging.getLogger(__name__)
8

9
    @retry(
10
        stop=stop_after_attempt(3),
11
        wait=wait_exponential(multiplier=1, min=4, max=10)
12
    )
13
    def parse_with_retry(self, file_path):
14
        try:
15
            return self.client.load_data(file_path)
16
        except Exception as e:
17
            self.logger.error(f"Parse failed for {file_path}: {e}")
18
            raise

1
import asyncio
2
from concurrent.futures import ThreadPoolExecutor
3

4
class AsyncLlamaProcessor:
5
    def __init__(self, api_key, max_workers=5):
6
        self.client = LlamaParse(api_key=api_key)
7
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
8

9
    async def process_documents_async(self, document_paths):
10
        loop = asyncio.get_event_loop()
11
        tasks = []
12

13
        for path in document_paths:
14
            task = loop.run_in_executor(
15
                self.executor,
16
                self.client.load_data,
17
                path
18
            )
19
            tasks.append(task)
20

21
        results = await asyncio.gather(*tasks)
22
        return results

1
# .env file
2
LLAMA_CLOUD_API_KEY=your_api_key_here
3
LLAMA_CLOUD_BASE_URL=https://api.cloud.llamaindex.ai
4
ENVIRONMENT=development
5

6
# config.py
7
import os
8
from dotenv import load_dotenv
9

10
load_dotenv()
11

12
class Config:
13
    API_KEY = os.getenv('LLAMA_CLOUD_API_KEY')
14
    BASE_URL = os.getenv('LLAMA_CLOUD_BASE_URL')
15
    ENVIRONMENT = os.getenv('ENVIRONMENT', 'development')

1
from llama_cloud_services import LlamaParse, LlamaExtract, LlamaCloudIndex
2
from config import Config
3

4
def main():
5
    # Initialize services
6
    parser = LlamaParse(api_key=Config.API_KEY)
7
    extractor = LlamaExtract(api_key=Config.API_KEY)
8
    index = LlamaCloudIndex(
9
        "getting_started_index",
10
        project_name="demo",
11
        api_key=Config.API_KEY
12
    )
13

14
    # Process a sample document
15
    documents = parser.load_data("./sample_document.pdf")
16

17
    # Extract structured data
18
    extracted = extractor.extract(
19
        content=documents[0].text,
20
        schema={
21
            "title": "string",
22
            "summary": "string",
23
            "key_points": ["string"]
24
        }
25
    )
26

27
    # Index for search
28
    index.insert(documents)
29

30
    # Query the index
31
    response = index.query("What are the main topics covered?")
32

33
    print("Extraction Results:", extracted)
34
    print("Query Response:", response)
35

36
if __name__ == "__main__":
37
    main()