Legal Document Automation: Ollama Contract Generation and Review Tutorial

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

# Verify installation
ollama --version

# Pull a legal-focused model
ollama pull llama2:13b

import ollama

# Initialize client with legal model
client = ollama.Client()

# Test model response
response = client.generate(
    model='llama2:13b',
    prompt="Draft a simple NDA clause for technology consulting.",
    stream=False
)

print(response['response'])

class ContractGenerator:
    def __init__(self, model_name='llama2:13b'):
        self.client = ollama.Client()
        self.model = model_name
        
    def generate_nda(self, party1, party2, purpose, duration="2 years"):
        """Generate Non-Disclosure Agreement"""
        prompt = f"""
        Draft a comprehensive NDA between {party1} and {party2}.
        Purpose: {purpose}
        Duration: {duration}
        
        Include these sections:
        1. Definition of confidential information
        2. Obligations of receiving party
        3. Exceptions to confidentiality
        4. Term and termination
        5. Remedies for breach
        
        Use professional legal language. Make it enforceable.
        """
        
        response = self.client.generate(
            model=self.model,
            prompt=prompt,
            stream=False
        )
        
        return response['response']
    
    def generate_service_agreement(self, provider, client, services, payment_terms):
        """Generate Service Agreement"""
        prompt = f"""
        Create a service agreement between {provider} (service provider) 
        and {client} (client).
        
        Services: {services}
        Payment Terms: {payment_terms}
        
        Include:
        - Scope of work
        - Payment structure
        - Intellectual property rights
        - Limitation of liability
        - Termination clauses
        
        Format as a professional legal contract.
        """
        
        return self.client.generate(
            model=self.model,
            prompt=prompt,
            stream=False
        )['response']

import re
from datetime import datetime, timedelta

class DynamicContractBuilder:
    def __init__(self):
        self.templates = {}
        self.generator = ContractGenerator()
    
    def create_template(self, contract_type, base_prompt):
        """Store contract template with placeholders"""
        self.templates[contract_type] = base_prompt
    
    def fill_template(self, contract_type, **kwargs):
        """Replace placeholders with actual values"""
        if contract_type not in self.templates:
            raise ValueError(f"Template {contract_type} not found")
        
        template = self.templates[contract_type]
        
        # Add automatic date calculations
        kwargs['current_date'] = datetime.now().strftime("%B %d, %Y")
        kwargs['expiration_date'] = (datetime.now() + timedelta(days=365)).strftime("%B %d, %Y")
        
        # Replace placeholders
        for key, value in kwargs.items():
            placeholder = f"{{{key}}}"
            template = template.replace(placeholder, str(value))
        
        return template
    
    def generate_from_template(self, contract_type, **kwargs):
        """Generate contract using template and dynamic fields"""
        filled_template = self.fill_template(contract_type, **kwargs)
        
        response = self.generator.client.generate(
            model=self.generator.model,
            prompt=filled_template,
            stream=False
        )
        
        return response['response']

# Example usage
builder = DynamicContractBuilder()

# Create employment agreement template
employment_template = """
Draft an employment agreement for {employee_name} as {position} 
at {company_name}.

Start date: {start_date}
Salary: {salary}
Location: {work_location}

Include standard employment terms, confidentiality, and termination clauses.
"""

builder.create_template('employment', employment_template)

# Generate specific contract
contract = builder.generate_from_template(
    'employment',
    employee_name="John Smith",
    position="Senior Developer",
    company_name="Tech Innovations LLC",
    start_date="March 1, 2025",
    salary="$120,000 annually",
    work_location="Remote"
)

print(contract)

import json
from typing import List, Dict

class DocumentReviewer:
    def __init__(self, model_name='llama2:13b'):
        self.client = ollama.Client()
        self.model = model_name
        
    def analyze_contract_risks(self, contract_text: str) -> Dict:
        """Identify potential risks in contract language"""
        
        risk_prompt = f"""
        Analyze this contract for potential legal risks and issues:
        
        {contract_text}
        
        Identify:
        1. Unclear or ambiguous language
        2. Missing essential clauses
        3. Unfavorable terms
        4. Potential compliance issues
        5. Recommended improvements
        
        Format your response as JSON with categories and specific findings.
        """
        
        response = self.client.generate(
            model=self.model,
            prompt=risk_prompt,
            stream=False
        )
        
        # Parse AI response into structured format
        try:
            analysis = json.loads(response['response'])
        except:
            # Fallback if JSON parsing fails
            analysis = {"raw_analysis": response['response']}
        
        return analysis
    
    def check_compliance(self, document_text: str, regulations: List[str]) -> Dict:
        """Check document compliance with specific regulations"""
        
        regulations_list = "\n".join([f"- {reg}" for reg in regulations])
        
        compliance_prompt = f"""
        Review this document for compliance with these regulations:
        {regulations_list}
        
        Document:
        {document_text}
        
        For each regulation, indicate:
        - Compliance status (Compliant/Non-compliant/Unclear)
        - Specific issues found
        - Recommended actions
        
        Provide detailed explanations for any compliance concerns.
        """
        
        response = self.client.generate(
            model=self.model,
            prompt=compliance_prompt,
            stream=False
        )
        
        return {"compliance_review": response['response']}
    
    def suggest_improvements(self, contract_text: str) -> List[str]:
        """Generate improvement suggestions for contract language"""
        
        improvement_prompt = f"""
        Review this contract and suggest specific improvements:
        
        {contract_text}
        
        Focus on:
        1. Clearer language
        2. Better protection for all parties
        3. Standard industry practices
        4. Risk mitigation
        
        Provide numbered suggestions with explanations.
        """
        
        response = self.client.generate(
            model=self.model,
            prompt=improvement_prompt,
            stream=False
        )
        
        # Extract numbered suggestions
        suggestions = []
        lines = response['response'].split('\n')
        
        for line in lines:
            if re.match(r'^\d+\.', line.strip()):
                suggestions.append(line.strip())
        
        return suggestions

class ClauseAnalyzer:
    def __init__(self):
        self.reviewer = DocumentReviewer()
        
    def extract_clauses(self, contract_text: str) -> Dict[str, str]:
        """Extract and categorize contract clauses"""
        
        extraction_prompt = f"""
        Break down this contract into its main clauses:
        
        {contract_text}
        
        Identify and extract:
        - Payment terms
        - Termination clauses
        - Intellectual property rights
        - Liability limitations
        - Confidentiality provisions
        - Dispute resolution
        - Other significant clauses
        
        Format as JSON with clause names as keys and text as values.
        """
        
        response = self.reviewer.client.generate(
            model=self.reviewer.model,
            prompt=extraction_prompt,
            stream=False
        )
        
        try:
            clauses = json.loads(response['response'])
        except:
            # Simple text parsing fallback
            clauses = {"full_text": contract_text}
        
        return clauses
    
    def analyze_clause(self, clause_name: str, clause_text: str) -> Dict:
        """Analyze individual clause for issues"""
        
        analysis_prompt = f"""
        Analyze this {clause_name} clause:
        
        {clause_text}
        
        Evaluate:
        1. Clarity and enforceability
        2. Fairness to all parties
        3. Industry standard compliance
        4. Potential risks
        5. Improvement recommendations
        
        Provide specific, actionable feedback.
        """
        
        response = self.reviewer.client.generate(
            model=self.reviewer.model,
            prompt=analysis_prompt,
            stream=False
        )
        
        return {
            "clause": clause_name,
            "analysis": response['response']
        }
    
    def comprehensive_review(self, contract_text: str) -> Dict:
        """Perform complete clause-by-clause analysis"""
        
        clauses = self.extract_clauses(contract_text)
        analysis_results = {}
        
        for clause_name, clause_text in clauses.items():
            if clause_text:  # Only analyze non-empty clauses
                analysis_results[clause_name] = self.analyze_clause(
                    clause_name, clause_text
                )
        
        return analysis_results

from flask import Flask, request, jsonify
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)

# Initialize AI components
contract_generator = ContractGenerator()
document_reviewer = DocumentReviewer()
clause_analyzer = ClauseAnalyzer()

@app.route('/generate/nda', methods=['POST'])
def generate_nda():
    """Generate NDA via API"""
    try:
        data = request.json
        
        contract = contract_generator.generate_nda(
            party1=data['party1'],
            party2=data['party2'],
            purpose=data['purpose'],
            duration=data.get('duration', '2 years')
        )
        
        return jsonify({
            'success': True,
            'contract': contract,
            'type': 'nda'
        })
        
    except Exception as e:
        logging.error(f"NDA generation error: {str(e)}")
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/review/document', methods=['POST'])
def review_document():
    """Automated document review via API"""
    try:
        data = request.json
        document_text = data['document']
        
        # Perform comprehensive analysis
        risk_analysis = document_reviewer.analyze_contract_risks(document_text)
        improvements = document_reviewer.suggest_improvements(document_text)
        
        # Optional compliance check
        regulations = data.get('regulations', [])
        compliance_check = None
        if regulations:
            compliance_check = document_reviewer.check_compliance(
                document_text, regulations
            )
        
        return jsonify({
            'success': True,
            'risk_analysis': risk_analysis,
            'improvements': improvements,
            'compliance_check': compliance_check
        })
        
    except Exception as e:
        logging.error(f"Document review error: {str(e)}")
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/analyze/clauses', methods=['POST'])
def analyze_clauses():
    """Clause-by-clause analysis via API"""
    try:
        data = request.json
        contract_text = data['contract']
        
        analysis = clause_analyzer.comprehensive_review(contract_text)
        
        return jsonify({
            'success': True,
            'clause_analysis': analysis
        })
        
    except Exception as e:
        logging.error(f"Clause analysis error: {str(e)}")
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)

import requests
from typing import Optional

class DocumentManager:
    def __init__(self, api_base_url: str):
        self.api_url = api_base_url
        self.generator = ContractGenerator()
        self.reviewer = DocumentReviewer()
    
    def upload_generated_contract(self, contract_text: str, 
                                 filename: str, 
                                 client_id: str) -> bool:
        """Upload AI-generated contract to document management system"""
        
        payload = {
            'filename': filename,
            'content': contract_text,
            'client_id': client_id,
            'generated_by': 'ollama_ai',
            'type': 'contract'
        }
        
        try:
            response = requests.post(
                f"{self.api_url}/documents/upload",
                json=payload,
                timeout=30
            )
            return response.status_code == 200
        except:
            return False
    
    def retrieve_and_review(self, document_id: str) -> Dict:
        """Retrieve document and perform AI review"""
        
        try:
            # Fetch document
            response = requests.get(
                f"{self.api_url}/documents/{document_id}",
                timeout=30
            )
            
            if response.status_code != 200:
                return {"error": "Document not found"}
            
            document_data = response.json()
            document_text = document_data['content']
            
            # Perform AI review
            analysis = self.reviewer.analyze_contract_risks(document_text)
            improvements = self.reviewer.suggest_improvements(document_text)
            
            # Update document with review results
            review_payload = {
                'document_id': document_id,
                'ai_review': {
                    'analysis': analysis,
                    'improvements': improvements,
                    'reviewed_at': datetime.now().isoformat()
                }
            }
            
            requests.patch(
                f"{self.api_url}/documents/{document_id}/review",
                json=review_payload,
                timeout=30
            )
            
            return {
                'document': document_data,
                'review': {
                    'analysis': analysis,
                    'improvements': improvements
                }
            }
            
        except Exception as e:
            return {"error": str(e)}

class LegalDomainSpecializer:
    def __init__(self):
        self.specialized_prompts = {
            'real_estate': self._real_estate_context(),
            'employment': self._employment_context(),
            'corporate': self._corporate_context(),
            'intellectual_property': self._ip_context()
        }
    
    def _real_estate_context(self) -> str:
        return """
        You are a real estate law specialist. Focus on:
        - Property transfer requirements
        - Zoning compliance
        - Environmental disclosures
        - Title insurance considerations
        - Escrow procedures
        """
    
    def _employment_context(self) -> str:
        return """
        You specialize in employment law. Consider:
        - Labor law compliance
        - Equal opportunity requirements
        - Wage and hour regulations
        - Worker classification rules
        - Benefits administration
        """
    
    def _corporate_context(self) -> str:
        return """
        You are a corporate law expert. Address:
        - Corporate governance
        - Securities regulations
        - Merger and acquisition requirements
        - Shareholder rights
        - Board responsibilities
        """
    
    def _ip_context(self) -> str:
        return """
        You specialize in intellectual property law. Focus on:
        - Patent requirements
        - Trademark protection
        - Copyright considerations
        - Trade secret safeguards
        - Licensing agreements
        """
    
    def generate_specialized_contract(self, domain: str, 
                                   contract_type: str, 
                                   **kwargs) -> str:
        """Generate contract with domain-specific expertise"""
        
        if domain not in self.specialized_prompts:
            raise ValueError(f"Domain {domain} not supported")
        
        context = self.specialized_prompts[domain]
        
        prompt = f"""
        {context}
        
        Draft a {contract_type} contract with these parameters:
        {json.dumps(kwargs, indent=2)}
        
        Ensure compliance with {domain} law requirements.
        Include all necessary protective clauses.
        Use appropriate legal terminology for this domain.
        """
        
        client = ollama.Client()
        response = client.generate(
            model='llama2:13b',
            prompt=prompt,
            stream=False
        )
        
        return response['response']

class DocumentValidator:
    def __init__(self):
        self.client = ollama.Client()
        self.required_sections = {
            'nda': ['confidential_information', 'obligations', 'term', 'remedies'],
            'service_agreement': ['scope', 'payment', 'ip_rights', 'termination'],
            'employment': ['duties', 'compensation', 'confidentiality', 'termination']
        }
    
    def validate_completeness(self, contract_text: str, 
                            contract_type: str) -> Dict:
        """Check if contract includes required sections"""
        
        if contract_type not in self.required_sections:
            return {"error": f"Unknown contract type: {contract_type}"}
        
        required = self.required_sections[contract_type]
        
        validation_prompt = f"""
        Check if this {contract_type} contract includes these required sections:
        {', '.join(required)}
        
        Contract:
        {contract_text}
        
        For each required section, indicate:
        - Present: Yes/No
        - Quality: Good/Adequate/Poor
        - Missing elements if any
        
        Format as JSON.
        """
        
        response = self.client.generate(
            model='llama2:13b',
            prompt=validation_prompt,
            stream=False
        )
        
        try:
            validation_results = json.loads(response['response'])
        except:
            validation_results = {"raw_analysis": response['response']}
        
        return validation_results
    
    def check_legal_language(self, contract_text: str) -> Dict:
        """Validate legal language and terminology"""
        
        language_prompt = f"""
        Review this contract for proper legal language:
        
        {contract_text}
        
        Check for:
        1. Appropriate legal terminology
        2. Clear and unambiguous language
        3. Proper sentence structure
        4. Consistent terminology throughout
        5. Professional tone
        
        Identify any language issues and suggest improvements.
        """
        
        response = self.client.generate(
            model='llama2:13b',
            prompt=language_prompt,
            stream=False
        )
        
        return {"language_review": response['response']}
    
    def comprehensive_validation(self, contract_text: str, 
                               contract_type: str) -> Dict:
        """Perform complete document validation"""
        
        results = {}
        
        # Check completeness
        results['completeness'] = self.validate_completeness(
            contract_text, contract_type
        )
        
        # Check language quality
        results['language'] = self.check_legal_language(contract_text)
        
        # Generate overall score
        results['validation_summary'] = self._calculate_score(results)
        
        return results
    
    def _calculate_score(self, validation_results: Dict) -> Dict:
        """Calculate overall document quality score"""
        
        scoring_prompt = f"""
        Based on these validation results, provide an overall quality score:
        
        {json.dumps(validation_results, indent=2)}
        
        Provide:
        - Overall score (1-10)
        - Key strengths
        - Critical issues
        - Ready for use (Yes/No)
        - Required improvements
        """
        
        response = self.client.generate(
            model='llama2:13b',
            prompt=scoring_prompt,
            stream=False
        )
        
        return {"summary": response['response']}

import hashlib
import pickle
from datetime import datetime, timedelta
from typing import Optional

class DocumentCache:
    def __init__(self, cache_duration_hours: int = 24):
        self.cache = {}
        self.cache_duration = timedelta(hours=cache_duration_hours)
    
    def _generate_key(self, prompt: str, model: str) -> str:
        """Generate cache key from prompt and model"""
        content = f"{prompt}:{model}"
        return hashlib.sha256(content.encode()).hexdigest()
    
    def get(self, prompt: str, model: str) -> Optional[str]:
        """Retrieve cached response if available and fresh"""
        key = self._generate_key(prompt, model)
        
        if key in self.cache:
            cached_data = self.cache[key]
            
            # Check if cache is still fresh
            if datetime.now() - cached_data['timestamp'] < self.cache_duration:
                return cached_data['response']
            else:
                # Remove stale cache
                del self.cache[key]
        
        return None
    
    def set(self, prompt: str, model: str, response: str):
        """Cache response with timestamp"""
        key = self._generate_key(prompt, model)
        
        self.cache[key] = {
            'response': response,
            'timestamp': datetime.now()
        }
    
    def clear_expired(self):
        """Remove expired cache entries"""
        current_time = datetime.now()
        expired_keys = []
        
        for key, data in self.cache.items():
            if current_time - data['timestamp'] >= self.cache_duration:
                expired_keys.append(key)
        
        for key in expired_keys:
            del self.cache[key]

class OptimizedContractGenerator(ContractGenerator):
    def __init__(self, model_name='llama2:13b'):
        super().__init__(model_name)
        self.cache = DocumentCache()
    
    def generate_with_cache(self, prompt: str) -> str:
        """Generate response with caching support"""
        
        # Check cache first
        cached_response = self.cache.get(prompt, self.model)
        if cached_response:
            return cached_response
        
        # Generate new response
        response = self.client.generate(
            model=self.model,
            prompt=prompt,
            stream=False
        )
        
        result = response['response']
        
        # Cache the result
        self.cache.set(prompt, self.model, result)
        
        return result
    
    def batch_generate(self, prompts: List[str]) -> List[str]:
        """Generate multiple contracts efficiently"""
        results = []
        
        for prompt in prompts:
            result = self.generate_with_cache(prompt)
            results.append(result)
        
        return results

import time
from collections import defaultdict
from datetime import datetime

class LegalAIAnalytics:
    def __init__(self):
        self.metrics = defaultdict(list)
        self.start_time = datetime.now()
    
    def track_generation(self, contract_type: str, 
                        generation_time: float, 
                        success: bool):
        """Track contract generation metrics"""
        
        self.metrics['generations'].append({
            'type': contract_type,
            'time': generation_time,
            'success': success,
            'timestamp': datetime.now()
        })
    
    def track_review(self, document_size: int, 
                    review_time: float, 
                    issues_found: int):
        """Track document review metrics"""
        
        self.metrics['reviews'].append({
            'size': document_size,
            'time': review_time,
            'issues': issues_found,
            'timestamp': datetime.now()
        })
    
    def get_performance_stats(self) -> Dict:
        """Generate performance statistics"""
        
        generation_times = [
            m['time'] for m in self.metrics['generations'] 
            if m['success']
        ]
        
        review_times = [m['time'] for m in self.metrics['reviews']]
        
        stats = {
            'total_generations': len(self.metrics['generations']),
            'successful_generations': len(generation_times),
            'avg_generation_time': sum(generation_times) / len(generation_times) if generation_times else 0,
            'total_reviews': len(self.metrics['reviews']),
            'avg_review_time': sum(review_times) / len(review_times) if review_times else 0,
            'uptime_hours': (datetime.now() - self.start_time).total_seconds() / 3600
        }
        
        return stats
    
    def export_metrics(self) -> Dict:
        """Export all collected metrics"""
        return dict(self.metrics)

# Usage example with analytics
analytics = LegalAIAnalytics()

def timed_contract_generation(generator, contract_type, **kwargs):
    """Generate contract with timing and analytics"""
    start_time = time.time()
    success = False
    
    try:
        if contract_type == 'nda':
            result = generator.generate_nda(**kwargs)
        elif contract_type == 'service':
            result = generator.generate_service_agreement(**kwargs)
        else:
            raise ValueError(f"Unknown contract type: {contract_type}")
        
        success = True
        return result
        
    except Exception as e:
        result = f"Error: {str(e)}"
        return result
        
    finally:
        end_time = time.time()
        generation_time = end_time - start_time
        
        analytics.track_generation(
            contract_type, 
            generation_time, 
            success
        )

# Dockerfile for Legal Document Automation
FROM python:3.11-slim

WORKDIR /app

# 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

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

# Copy application code
COPY . .

# Expose port
EXPOSE 5000

# Start script
COPY start.sh .
RUN chmod +x start.sh

CMD ["./start.sh"]

#!/bin/bash
# start.sh - Production startup script

# Start Ollama service
ollama serve &

# Wait for Ollama to be ready
sleep 10

# Pull required models
ollama pull llama2:13b
ollama pull mistral:7b

# Start the application
python app.py

import os
import jwt
from functools import wraps
from flask import request, jsonify

class SecurityManager:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
    
    def require_auth(self, f):
        """Decorator for API authentication"""
        @wraps(f)
        def decorated_function(*args, **kwargs):
            token = request.headers.get('Authorization')
            
            if not token:
                return jsonify({'error': 'Missing authentication token'}), 401
            
            try:
                # Remove 'Bearer ' prefix
                token = token.replace('Bearer ', '')
                payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
                request.user = payload
                
            except jwt.InvalidTokenError:
                return jsonify({'error': 'Invalid authentication token'}), 401
            
            return f(*args, **kwargs)
        
        return decorated_function
    
    def log_access(self, user_id: str, action: str, document_type: str):
        """Log access for audit trail"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'user_id': user_id,
            'action': action,
            'document_type': document_type,
            'ip_address': request.remote_addr
        }
        
        # In production, use proper logging system
        with open('audit.log', 'a') as f:
            f.write(f"{json.dumps(log_entry)}\n")

# Apply security to API endpoints
security = SecurityManager(os.environ.get('JWT_SECRET', 'your-secret-key'))

@app.route('/secure/generate/nda', methods=['POST'])
@security.require_auth
def secure_generate_nda():
    """Secure NDA generation endpoint"""
    user_id = request.user.get('user_id')
    
    # Log access
    security.log_access(user_id, 'generate_nda', 'nda')
    
    # Generate contract (existing logic)
    data = request.json
    contract = contract_generator.generate_nda(**data)
    
    return jsonify({
        'success': True,
        'contract': contract,
        'generated_by': user_id
    })