Ollama Contract Analysis: Complete Setup Guide for Legal Document Review Automation

# Download and install Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Verify installation
ollama version

# Install Llama 2 13B model (recommended for legal work)
ollama pull llama2:13b

# Alternative: Install Mistral 7B for faster processing
ollama pull mistral:7b

# Create virtual environment
python -m venv legal-ai-env

# Activate environment (Linux/Mac)
source legal-ai-env/bin/activate

# Activate environment (Windows)
legal-ai-env\Scripts\activate

# Install required packages
pip install requests python-docx PyPDF2 pandas ollama-python

import ollama
import json
import os
from pathlib import Path
import PyPDF2
from docx import Document

class ContractAnalyzer:
    def __init__(self, model_name="llama2:13b"):
        """Initialize the contract analyzer with specified model."""
        self.model = model_name
        self.client = ollama.Client()
        
    def extract_text_from_pdf(self, file_path):
        """Extract text content from PDF files."""
        with open(file_path, 'rb') as file:
            reader = PyPDF2.PdfReader(file)
            text = ""
            for page in reader.pages:
                text += page.extract_text()
        return text
    
    def extract_text_from_docx(self, file_path):
        """Extract text content from Word documents."""
        doc = Document(file_path)
        text = ""
        for paragraph in doc.paragraphs:
            text += paragraph.text + "\n"
        return text
    
    def analyze_contract(self, contract_text):
        """Perform comprehensive contract analysis."""
        
        # Define analysis prompt for legal documents
        prompt = f"""
        Analyze this legal contract and provide a structured analysis:

        CONTRACT TEXT:
        {contract_text[:4000]}  # Limit text for token constraints

        Please provide analysis in this JSON format:
        {{
            "contract_type": "Type of contract (e.g., Service Agreement, NDA)",
            "parties": ["List of contracting parties"],
            "key_terms": ["Important terms and conditions"],
            "obligations": {{
                "party_1": ["Obligations of first party"],
                "party_2": ["Obligations of second party"]
            }},
            "important_dates": ["Key dates and deadlines"],
            "termination_clauses": ["Termination conditions"],
            "risk_factors": ["Potential legal risks identified"],
            "governing_law": "Applicable jurisdiction",
            "summary": "Brief summary of contract purpose and key points"
        }}

        Focus on accuracy and completeness in your analysis.
        """
        
        try:
            # Send analysis request to Ollama
            response = self.client.chat(
                model=self.model,
                messages=[{
                    'role': 'user',
                    'content': prompt
                }]
            )
            
            return response['message']['content']
            
        except Exception as e:
            return f"Analysis error: {str(e)}"

    def process_document(self, file_path):
        """Process a single document and return analysis."""
        file_extension = Path(file_path).suffix.lower()
        
        # Extract text based on file type
        if file_extension == '.pdf':
            text = self.extract_text_from_pdf(file_path)
        elif file_extension == '.docx':
            text = self.extract_text_from_docx(file_path)
        elif file_extension == '.txt':
            with open(file_path, 'r', encoding='utf-8') as file:
                text = file.read()
        else:
            return "Unsupported file format"
        
        # Perform analysis
        analysis = self.analyze_contract(text)
        
        return {
            'file_name': Path(file_path).name,
            'file_path': file_path,
            'analysis': analysis,
            'word_count': len(text.split())
        }

# Usage example
if __name__ == "__main__":
    analyzer = ContractAnalyzer()
    
    # Process a single contract
    result = analyzer.process_document("contracts/sample_contract.pdf")
    
    # Save results
    with open("output/analysis_result.json", "w") as f:
        json.dump(result, f, indent=2)
    
    print("Contract analysis complete!")

class LegalPrompts:
    
    @staticmethod
    def nda_analysis():
        """Specialized prompt for NDA analysis."""
        return """
        Analyze this Non-Disclosure Agreement (NDA) and identify:
        
        1. CONFIDENTIAL INFORMATION DEFINITION
        2. PERMITTED DISCLOSURES
        3. DURATION OF CONFIDENTIALITY
        4. RETURN/DESTRUCTION OBLIGATIONS
        5. REMEDIES FOR BREACH
        6. JURISDICTION AND GOVERNING LAW
        
        Rate the NDA's protection level (1-10) and explain your reasoning.
        Highlight any unusual or concerning clauses.
        """
    
    @staticmethod
    def service_agreement():
        """Specialized prompt for service agreement analysis."""
        return """
        Analyze this Service Agreement focusing on:
        
        1. SCOPE OF SERVICES
        2. PAYMENT TERMS AND SCHEDULE
        3. PERFORMANCE STANDARDS
        4. INTELLECTUAL PROPERTY RIGHTS
        5. LIABILITY LIMITATIONS
        6. TERMINATION CONDITIONS
        
        Identify potential risks for both service provider and client.
        Flag any ambiguous service definitions.
        """
    
    @staticmethod
    def employment_contract():
        """Specialized prompt for employment contract analysis."""
        return """
        Review this Employment Contract for:
        
        1. COMPENSATION STRUCTURE
        2. BENEFITS AND PERQUISITES
        3. NON-COMPETE CLAUSES
        4. TERMINATION PROCEDURES
        5. INTELLECTUAL PROPERTY ASSIGNMENTS
        6. CONFIDENTIALITY OBLIGATIONS
        
        Assess fairness of terms and identify any red flags.
        Check compliance with local employment laws.
        """

import os
import json
from datetime import datetime
from contract_analyzer import ContractAnalyzer

class BatchProcessor:
    def __init__(self, input_folder="contracts", output_folder="output"):
        self.input_folder = input_folder
        self.output_folder = output_folder
        self.analyzer = ContractAnalyzer()
        
        # Create output folder if it doesn't exist
        os.makedirs(output_folder, exist_ok=True)
    
    def process_all_contracts(self):
        """Process all contracts in the input folder."""
        results = []
        supported_formats = ['.pdf', '.docx', '.txt']
        
        # Get all supported files
        files = [f for f in os.listdir(self.input_folder) 
                if any(f.lower().endswith(ext) for ext in supported_formats)]
        
        print(f"Found {len(files)} contracts to process...")
        
        for i, filename in enumerate(files, 1):
            print(f"Processing {i}/{len(files)}: {filename}")
            
            file_path = os.path.join(self.input_folder, filename)
            result = self.analyzer.process_document(file_path)
            results.append(result)
            
            # Save individual result
            output_file = f"{os.path.splitext(filename)[0]}_analysis.json"
            output_path = os.path.join(self.output_folder, output_file)
            
            with open(output_path, 'w') as f:
                json.dump(result, f, indent=2)
        
        # Save batch summary
        batch_summary = {
            'processing_date': datetime.now().isoformat(),
            'total_contracts': len(results),
            'successful_analyses': len([r for r in results if 'error' not in r.get('analysis', '')]),
            'results': results
        }
        
        summary_path = os.path.join(self.output_folder, 'batch_summary.json')
        with open(summary_path, 'w') as f:
            json.dump(batch_summary, f, indent=2)
        
        print(f"Batch processing complete! Results saved to {self.output_folder}")
        return batch_summary

# Usage example
if __name__ == "__main__":
    processor = BatchProcessor()
    summary = processor.process_all_contracts()
    
    print(f"Processed {summary['total_contracts']} contracts")
    print(f"Successful analyses: {summary['successful_analyses']}")

import re
import json
from typing import Dict, List

class RiskDetector:
    def __init__(self):
        self.risk_patterns = {
            'high_risk': [
                r'unlimited liability',
                r'personal guarantee',
                r'liquidated damages',
                r'automatic renewal',
                r'exclusive dealing'
            ],
            'medium_risk': [
                r'force majeure',
                r'indemnification',
                r'intellectual property',
                r'non-compete',
                r'confidentiality'
            ],
            'compliance_risk': [
                r'gdpr',
                r'data protection',
                r'privacy policy',
                r'regulatory compliance',
                r'audit rights'
            ]
        }
    
    def detect_risks(self, contract_text: str) -> Dict:
        """Detect potential risks in contract text."""
        risks_found = {
            'high_risk': [],
            'medium_risk': [],
            'compliance_risk': [],
            'risk_score': 0
        }
        
        contract_lower = contract_text.lower()
        
        for risk_level, patterns in self.risk_patterns.items():
            for pattern in patterns:
                if re.search(pattern, contract_lower):
                    risks_found[risk_level].append(pattern)
        
        # Calculate risk score
        risk_score = (len(risks_found['high_risk']) * 3 + 
                     len(risks_found['medium_risk']) * 2 + 
                     len(risks_found['compliance_risk']) * 1)
        
        risks_found['risk_score'] = risk_score
        risks_found['risk_level'] = self.categorize_risk_level(risk_score)
        
        return risks_found
    
    def categorize_risk_level(self, score: int) -> str:
        """Categorize overall risk level based on score."""
        if score >= 8:
            return "HIGH"
        elif score >= 4:
            return "MEDIUM"
        else:
            return "LOW"
    
    def generate_risk_report(self, risks: Dict) -> str:
        """Generate human-readable risk report."""
        report = f"RISK ASSESSMENT REPORT\n"
        report += f"Overall Risk Level: {risks['risk_level']}\n"
        report += f"Risk Score: {risks['risk_score']}/10\n\n"
        
        for risk_type, items in risks.items():
            if isinstance(items, list) and items:
                report += f"{risk_type.upper().replace('_', ' ')}:\n"
                for item in items:
                    report += f"  - {item}\n"
                report += "\n"
        
        return report

from flask import Flask, render_template, request, jsonify, send_file
import os
import json
from werkzeug.utils import secure_filename
from contract_analyzer import ContractAnalyzer
from risk_detector import RiskDetector

app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = 'uploads'
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB max file size

# Initialize components
analyzer = ContractAnalyzer()
risk_detector = RiskDetector()

@app.route('/')
def index():
    """Main interface for contract upload."""
    return '''
    <!DOCTYPE html>
    <html>
    <head>
        <title>Ollama Contract Analyzer</title>
        <style>
            body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }
            .upload-area { border: 2px dashed #ccc; padding: 40px; text-align: center; margin: 20px 0; }
            .results { background: #f5f5f5; padding: 20px; margin: 20px 0; border-radius: 5px; }
            button { background: #007cba; color: white; padding: 10px 20px; border: none; border-radius: 5px; cursor: pointer; }
            button:hover { background: #005a85; }
        </style>
    </head>
    <body>
        <h1>Ollama Contract Analysis System</h1>
        <div class="upload-area">
            <form action="/analyze" method="post" enctype="multipart/form-data">
                <input type="file" name="contract" accept=".pdf,.docx,.txt" required>
                <br><br>
                <button type="submit">Analyze Contract</button>
            </form>
        </div>
        <div id="results"></div>
    </body>
    </html>
    '''

@app.route('/analyze', methods=['POST'])
def analyze_contract():
    """Analyze uploaded contract."""
    if 'contract' not in request.files:
        return jsonify({'error': 'No file uploaded'})
    
    file = request.files['contract']
    if file.filename == '':
        return jsonify({'error': 'No file selected'})
    
    # Save uploaded file
    filename = secure_filename(file.filename)
    filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
    file.save(filepath)
    
    try:
        # Analyze contract
        result = analyzer.process_document(filepath)
        
        # Detect risks
        if result.get('analysis'):
            risks = risk_detector.detect_risks(result['analysis'])
            result['risk_assessment'] = risks
        
        # Clean up uploaded file
        os.remove(filepath)
        
        return jsonify(result)
        
    except Exception as e:
        return jsonify({'error': f'Analysis failed: {str(e)}'})

if __name__ == '__main__':
    os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
    app.run(debug=True, port=5000)

# Optimize memory usage for large documents
def process_large_contract(self, file_path, chunk_size=2000):
    """Process large contracts in chunks to manage memory."""
    text = self.extract_text_from_file(file_path)
    
    # Split into manageable chunks
    words = text.split()
    chunks = [' '.join(words[i:i+chunk_size]) 
              for i in range(0, len(words), chunk_size)]
    
    analyses = []
    for chunk in chunks:
        analysis = self.analyze_contract(chunk)
        analyses.append(analysis)
    
    return self.merge_analyses(analyses)

from concurrent.futures import ThreadPoolExecutor
import threading

class ParallelContractProcessor:
    def __init__(self, max_workers=4):
        self.max_workers = max_workers
        self.analyzer = ContractAnalyzer()
    
    def process_contracts_parallel(self, file_paths):
        """Process multiple contracts simultaneously."""
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [executor.submit(self.analyzer.process_document, path) 
                      for path in file_paths]
            
            results = []
            for future in futures:
                try:
                    result = future.result(timeout=300)  # 5-minute timeout
                    results.append(result)
                except Exception as e:
                    results.append({'error': str(e)})
            
            return results

# Switch to faster model
ollama pull mistral:7b

# Or use quantized version
ollama pull llama2:7b-q4_0

def handle_large_document(self, text, max_tokens=3000):
    """Handle documents exceeding token limits."""
    if len(text.split()) > max_tokens:
        # Split document into sections
        sections = self.split_by_sections(text)
        
        # Analyze each section
        section_analyses = []
        for section in sections:
            analysis = self.analyze_contract(section)
            section_analyses.append(analysis)
        
        # Merge results
        return self.merge_section_analyses(section_analyses)
    
    return self.analyze_contract(text)

def validate_analysis(self, analysis_result):
    """Validate analysis results for completeness."""
    required_fields = ['contract_type', 'parties', 'key_terms']
    
    for field in required_fields:
        if field not in analysis_result:
            # Re-analyze with more specific prompt
            return self.re_analyze_with_focus(field)
    
    return analysis_result

class DMSIntegration:
    def __init__(self, dms_api_key):
        self.api_key = dms_api_key
        self.analyzer = ContractAnalyzer()
    
    def sync_with_dms(self, document_id):
        """Sync analysis results with document management system."""
        # Fetch document from DMS
        document = self.fetch_from_dms(document_id)
        
        # Analyze contract
        analysis = self.analyzer.process_document(document['file_path'])
        
        # Update DMS with analysis results
        self.update_dms_metadata(document_id, analysis)
        
        return analysis

import imaplib
import email
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

class EmailIntegration:
    def __init__(self, email_config):
        self.config = email_config
        self.analyzer = ContractAnalyzer()
    
    def process_email_attachments(self):
        """Monitor email for contract attachments and analyze automatically."""
        mail = imaplib.IMAP4_SSL(self.config['server'])
        mail.login(self.config['username'], self.config['password'])
        mail.select('inbox')
        
        # Search for emails with attachments
        status, messages = mail.search(None, 'UNSEEN')
        
        for msg_id in messages[0].split():
            # Process email and attachments
            email_body = mail.fetch(msg_id, '(RFC822)')
            # Extract and analyze attachments
            # Send analysis results back via email

from cryptography.fernet import Fernet
import os

class SecureContractProcessor:
    def __init__(self):
        self.key = self.load_or_generate_key()
        self.cipher = Fernet(self.key)
        self.analyzer = ContractAnalyzer()
    
    def load_or_generate_key(self):
        """Load existing encryption key or generate new one."""
        key_file = 'encryption.key'
        if os.path.exists(key_file):
            with open(key_file, 'rb') as f:
                return f.read()
        else:
            key = Fernet.generate_key()
            with open(key_file, 'wb') as f:
                f.write(key)
            return key
    
    def encrypt_document(self, file_path):
        """Encrypt document before processing."""
        with open(file_path, 'rb') as f:
            data = f.read()
        
        encrypted_data = self.cipher.encrypt(data)
        
        encrypted_path = f"{file_path}.encrypted"
        with open(encrypted_path, 'wb') as f:
            f.write(encrypted_data)
        
        return encrypted_path
    
    def decrypt_and_process(self, encrypted_path):
        """Decrypt and process document securely."""
        with open(encrypted_path, 'rb') as f:
            encrypted_data = f.read()
        
        decrypted_data = self.cipher.decrypt(encrypted_data)
        
        # Create temporary file for processing
        temp_path = f"temp_{os.path.basename(encrypted_path)}"
        with open(temp_path, 'wb') as f:
            f.write(decrypted_data)
        
        try:
            # Process document
            result = self.analyzer.process_document(temp_path)
            return result
        finally:
            # Clean up temporary file
            if os.path.exists(temp_path):
                os.remove(temp_path)

import hashlib
import jwt
from datetime import datetime, timedelta

class AccessControl:
    def __init__(self, secret_key):
        self.secret_key = secret_key
        self.authorized_users = {}
    
    def authenticate_user(self, username, password):
        """Authenticate user for system access."""
        password_hash = hashlib.sha256(password.encode()).hexdigest()
        
        if username in self.authorized_users:
            if self.authorized_users[username] == password_hash:
                # Generate JWT token
                token = jwt.encode({
                    'user': username,
                    'exp': datetime.utcnow() + timedelta(hours=8)
                }, self.secret_key, algorithm='HS256')
                
                return {'success': True, 'token': token}
        
        return {'success': False, 'message': 'Invalid credentials'}
    
    def verify_token(self, token):
        """Verify JWT token for API access."""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
            return {'valid': True, 'user': payload['user']}
        except jwt.ExpiredSignatureError:
            return {'valid': False, 'message': 'Token expired'}
        except jwt.InvalidTokenError:
            return {'valid': False, 'message': 'Invalid token'}

import sqlite3
import json
from datetime import datetime

class AnalyticsTracker:
    def __init__(self, db_path="analytics.db"):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """Initialize analytics database."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS contract_analyses (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp DATETIME,
                file_name TEXT,
                file_size INTEGER,
                processing_time REAL,
                contract_type TEXT,
                risk_level TEXT,
                user_id TEXT
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def log_analysis(self, analysis_data):
        """Log contract analysis for tracking."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO contract_analyses 
            (timestamp, file_name, file_size, processing_time, contract_type, risk_level, user_id)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (
            datetime.now(),
            analysis_data.get('file_name'),
            analysis_data.get('file_size', 0),
            analysis_data.get('processing_time', 0),
            analysis_data.get('contract_type'),
            analysis_data.get('risk_level'),
            analysis_data.get('user_id')
        ))
        
        conn.commit()
        conn.close()
    
    def generate_usage_report(self):
        """Generate system usage analytics."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Get basic statistics
        cursor.execute('''
            SELECT 
                COUNT(*) as total_analyses,
                AVG(processing_time) as avg_processing_time,
                AVG(file_size) as avg_file_size
            FROM contract_analyses
        ''')
        
        stats = cursor.fetchone()
        
        # Get contract type distribution
        cursor.execute('''
            SELECT contract_type, COUNT(*) as count
            FROM contract_analyses
            GROUP BY contract_type
            ORDER BY count DESC
        ''')
        
        contract_types = cursor.fetchall()
        
        conn.close()
        
        return {
            'total_analyses': stats[0],
            'average_processing_time': stats[1],
            'average_file_size': stats[2],
            'contract_type_distribution': dict(contract_types)
        }

class LegalPromptOptimizer:
    @staticmethod
    def create_context_aware_prompt(contract_type, analysis_focus):
        """Create optimized prompts based on contract type and focus area."""
        
        base_prompt = f"""
        You are a legal AI assistant specializing in {contract_type} analysis.
        Your task is to provide accurate, comprehensive analysis focusing on {analysis_focus}.
        
        Guidelines:
        - Be precise and use legal terminology appropriately
        - Identify potential risks and flag ambiguous language
        - Provide specific references to contract sections when possible
        - Consider jurisdiction-specific legal requirements
        """
        
        type_specific_additions = {
            'employment': """
            Pay special attention to:
            - Compensation structure and benefits
            - Non-compete and non-disclosure clauses
            - Termination procedures and severance
            - Intellectual property assignments
            """,
            'service_agreement': """
            Focus on:
            - Scope of services and deliverables
            - Performance standards and metrics
            - Payment terms and conditions
            - Liability limitations and indemnification
            """,
            'nda': """
            Examine:
            - Definition of confidential information
            - Permitted uses and disclosures
            - Duration of confidentiality obligations
            - Remedies for breach
            """
        }
        
        return base_prompt + type_specific_additions.get(contract_type.lower(), "")

class QualityAssurance:
    def __init__(self):
        self.validation_rules = {
            'completeness': self.check_completeness,
            'consistency': self.check_consistency,
            'accuracy': self.check_accuracy
        }
    
    def validate_analysis(self, analysis_result, original_text):
        """Comprehensive validation of analysis results."""
        validation_results = {}
        
        for rule_name, rule_function in self.validation_rules.items():
            validation_results[rule_name] = rule_function(analysis_result, original_text)
        
        overall_score = sum(validation_results.values()) / len(validation_results)
        
        return {
            'validation_score': overall_score,
            'individual_scores': validation_results,
            'passed': overall_score >= 0.8
        }
    
    def check_completeness(self, analysis, original_text):
        """Check if analysis covers all major contract elements."""
        required_elements = ['parties', 'key_terms', 'obligations']
        found_elements = [elem for elem in required_elements if elem in analysis]
        
        return len(found_elements) / len(required_elements)
    
    def check_consistency(self, analysis, original_text):
        """Verify analysis consistency with source document."""
        # Implement consistency checks
        return 0.9  # Placeholder score
    
    def check_accuracy(self, analysis, original_text):
        """Validate accuracy of extracted information."""
        # Implement accuracy validation
        return 0.85  # Placeholder score

# Dockerfile
FROM python:3.9-slim

# Install system dependencies
RUN apt-get update && apt-get install -y \
    curl \
    && rm -rf /var/lib/apt/lists/*

# Install Ollama
RUN curl -fsSL https://ollama.ai/install.sh | sh

# Set working directory
WORKDIR /app

# Copy requirements and install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Expose port
EXPOSE 5000

# Start services
CMD ["python", "web_interface.py"]

# production_config.py
import os

class ProductionConfig:
    # Database configuration
    DATABASE_URL = os.environ.get('DATABASE_URL', 'sqlite:///legal_ai.db')
    
    # Security settings
    SECRET_KEY = os.environ.get('SECRET_KEY', 'your-secret-key-here')
    ENCRYPT_DOCUMENTS = True
    
    # Performance settings
    MAX_CONCURRENT_ANALYSES = 10
    CACHE_ANALYSIS_RESULTS = True
    CACHE_TIMEOUT = 3600  # 1 hour
    
    # Model configuration
    OLLAMA_MODEL = os.environ.get('OLLAMA_MODEL', 'llama2:13b')
    MODEL_TEMPERATURE = 0.1  # Low temperature for consistent legal analysis
    
    # File handling
    MAX_FILE_SIZE = 50 * 1024 * 1024  # 50MB
    ALLOWED_EXTENSIONS = {'.pdf', '.docx', '.txt'}
    
    # Logging
    LOG_LEVEL = 'INFO'
    LOG_FILE = 'legal_ai.log'