CrewAI real-world example: legal contract review

CrewAI works well for legal contract review because the task naturally decomposes into specialized roles: a parser that handles messy PDFs, an extractor that pulls structured clauses, a classifier that tags risk categories, and a reviewer that surfaces issues for human sign-off. This tutorial walks through a complete, runnable pipeline you can adapt to your own contract types and risk taxonomy. We’ll use a local LLM via Ollama for data privacy, but the same structure works with any OpenAI-compatible endpoint.

Step 1: Set up the environment and dependencies

Create a fresh virtual environment and install the minimal set. We’ll use pdfplumber for text extraction, pydantic for structured outputs, and crewai with langchain-community for the Ollama integration.

python -m venv .venv && source .venv/bin/activate
pip install crewai==0.28.8 pdfplumber pydantic pydantic-settings langchain-community ollama

Verify Ollama is running and pull a capable model. llama3.1:8b balances speed and reasoning for clause classification; swap in llama3.1:70b if you have VRAM.

ollama serve &
ollama pull llama3.1:8b

Create a config.yaml to keep model and path settings out of code:

# config.yaml
llm:
  model: "ollama/llama3.1:8b"
  base_url: "http://localhost:11434"
  temperature: 0.1

paths:
  contracts_dir: "./contracts"
  output_dir: "./output"

risk_taxonomy:
  - "payment_terms"
  - "termination"
  - "liability_cap"
  - "indemnification"
  - "ip_ownership"
  - "confidentiality"
  - "data_privacy"
  - "force_majeure"
  - "governing_law"
  - "assignment"

Load it with a tiny settings class:

# settings.py
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict
import yaml

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
    
    llm_model: str = "ollama/llama3.1:8b"
    llm_base_url: str = "http://localhost:11434"
    llm_temperature: float = 0.1
    contracts_dir: Path = Path("./contracts")
    output_dir: Path = Path("./output")
    risk_taxonomy: list[str] = []

    @classmethod
    def from_yaml(cls, path: Path) -> "Settings":
        with open(path) as f:
            data = yaml.safe_load(f)
        # Flatten nested yaml into flat fields
        flat = {
            "llm_model": data["llm"]["model"],
            "llm_base_url": data["llm"]["base_url"],
            "llm_temperature": data["llm"]["temperature"],
            "contracts_dir": Path(data["paths"]["contracts_dir"]),
            "output_dir": Path(data["paths"]["output_dir"]),
            "risk_taxonomy": data["risk_taxonomy"],
        }
        return cls(**flat)

settings = Settings.from_yaml(Path("config.yaml"))

Step 2: Define the data models for structured outputs

Pydantic models give you validation and make the crew’s hand-offs explicit. Each agent returns a typed object the next agent can consume without string parsing.

# models.py
from pydantic import BaseModel, Field
from typing import Literal
from enum import Enum

class RiskCategory(str, Enum):
    PAYMENT_TERMS = "payment_terms"
    TERMINATION = "termination"
    LIABILITY_CAP = "liability_cap"
    INDEMNIFICATION = "indemnification"
    IP_OWNERSHIP = "ip_ownership"
    CONFIDENTIALITY = "confidentiality"
    DATA_PRIVACY = "data_privacy"
    FORCE_MAJEURE = "force_majeure"
    GOVERNING_LAW = "governing_law"
    ASSIGNMENT = "assignment"

class Severity(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class Clause(BaseModel):
    clause_id: str
    heading: str | None
    text: str
    page_start: int
    page_end: int

class ClassifiedClause(Clause):
    category: RiskCategory
    severity: Severity
    rationale: str = Field(description="Why this severity and category")
    redline_suggestion: str | None = Field(default=None, description="Suggested rewrite if problematic")

class ContractAnalysis(BaseModel):
    contract_id: str
    filename: str
    total_clauses: int
    classified_clauses: list[ClassifiedClause]
    summary: str
    high_risk_count: int
    critical_risk_count: int

Step 3: Build the PDF parser agent

This agent owns one job: turn a PDF into a clean list of Clause objects. It handles multi-column layouts, headers/footers, and page boundaries. We keep the LLM out of this step — deterministic extraction is faster and cheaper.

# agents/parser.py
import pdfplumber
import re
from pathlib import Path
from typing import Iterator
from models import Clause
from settings import settings

class ContractParser:
    def __init__(self, contracts_dir: Path = None):
        self.contracts_dir = contracts_dir or settings.contracts_dir

    def extract_clauses(self, pdf_path: Path) -> list[Clause]:
        clauses = []
        clause_counter = 0
        
        with pdfplumber.open(pdf_path) as pdf:
            full_text = ""
            page_map = []  # (page_num, char_offset)
            
            for i, page in enumerate(pdf.pages):
                text = page.extract_text() or ""
                page_map.append((i + 1, len(full_text)))
                full_text += f"\n--- PAGE {i+1} ---\n" + text

        # Split on common clause delimiters: numbered sections, ALL CAPS headings
        # This regex catches "1.1", "Section 3", "ARTICLE V", "Confidentiality", etc.
        pattern = re.compile(
            r'(?:\n|^)'
            r'(?:'
            r'(?:\d+(?:\.\d+)+)\s+'           # 1.1, 2.3.4
            r'|(?:Section|Article|ARTICLE|SECTION)\s+\d+[A-Z]?'  # Section 3, Article V
            r'|(?:[A-Z][A-Z\s]{3,}:)'        # CONFIDENTIALITY:
            r')'
            r'(.+?)(?=\n(?:\d+(?:\.\d+)+|Section|Article|[A-Z]{4,}:)|\Z)',
            re.DOTALL | re.MULTILINE
        )

        for match in pattern.finditer(full_text):
            clause_text = match.group(0).strip()
            if len(clause_text) < 50:  # skip noise
                continue
            
            # Determine page range
            start_offset = match.start()
            end_offset = match.end()
            page_start = self._find_page(page_map, start_offset)
            page_end = self._find_page(page_map, end_offset)
            
            # Extract heading from first line
            first_line = clause_text.split('\n')[0][:120]
            
            clauses.append(Clause(
                clause_id=f"cl_{clause_counter:04d}",
                heading=first_line if len(first_line) > 3 else None,
                text=clause_text,
                page_start=page_start,
                page_end=page_end
            ))
            clause_counter += 1

        # Fallback: if regex found nothing, chunk by page
        if not clauses:
            for i, page in enumerate(pdf.pages):
                text = page.extract_text() or ""
                if len(text) > 100:
                    clauses.append(Clause(
                        clause_id=f"cl_{clause_counter:04d}",
                        heading=f"Page {i+1}",
                        text=text,
                        page_start=i+1,
                        page_end=i+1
                    ))
                    clause_counter += 1

        return clauses

    def _find_page(self, page_map: list[tuple[int, int]], offset: int) -> int:
        for i in range(len(page_map) - 1):
            if page_map[i][1] <= offset < page_map[i+1][1]:
                return page_map[i][0]
        return page_map[-1][0] if page_map else 1

    def process_all(self) -> dict[str, list[Clause]]:
        results = {}
        for pdf_file in self.contracts_dir.glob("*.pdf"):
            print(f"Parsing {pdf_file.name}...")
            results[pdf_file.stem] = self.extract_clauses(pdf_file)
        return results

Verify it works:

mkdir -p contracts
# Drop a sample contract PDF in contracts/
python -c "
from agents.parser import ContractParser
from settings import settings
parser = ContractParser()
results = parser.process_all()
for name, clauses in results.items():
    print(f'{name}: {len(clauses)} clauses')
    for c in clauses[:2]:
        print(f'  {c.clause_id} p{c.page_start}-{c.page_end}: {c.heading}')
"

You should see clause counts and first few headings per contract.

Step 4: Create the classifier agent with structured prompting

This agent receives Clause objects and returns ClassifiedClause with category, severity, and rationale. We use a single prompt that enforces the taxonomy and outputs JSON matching our Pydantic model.

# agents/classifier.py
from crewai import Agent, Task, Crew, LLM
from pydantic import BaseModel
from typing import List
from models import Clause, ClassifiedClause, RiskCategory, Severity
from settings import settings
import json

# Configure the LLM for CrewAI
llm = LLM(
    model=settings.llm_model,
    base_url=settings.llm_base_url,
    temperature=settings.llm_temperature,
)

# Taxonomy description injected into prompt
TAXONOMY_DESC = "\n".join([
    f"- {cat.value}: {cat.value.replace('_', ' ').title()}"
    for cat in RiskCategory
])

CLASSIFIER_PROMPT = f"""You are a senior contract attorney classifying clauses by risk category and severity.

Risk taxonomy (use EXACT category names):
{TAXONOMY_DESC}

Severity definitions:
- LOW: Standard language, minimal negotiation needed
- MEDIUM: Non-standard terms, review recommended
- HIGH: Unfavorable terms, negotiation required
- CRITICAL: Deal-breaker risk, immediate escalation

For each clause, output JSON matching this schema:
{{
  "clause_id": "string",
  "category": "exact_taxonomy_value",
  "severity": "low|medium|high|critical",
  "rationale": "specific reason referencing clause language",
  "redline_suggestion": "optional rewrite if severity >= medium"
}}

Return ONLY a JSON array of classifications, one per input clause. No markdown, no commentary.
"""

classifier_agent = Agent(
    role="Contract Risk Classifier",
    goal="Classify each clause by risk category and severity with precise rationale",
    backstory="You have 15 years reviewing SaaS, MSA, and NDA agreements for enterprise legal teams.",
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

def classify_clauses(clauses: List[Clause]) -> List[ClassifiedClause]:
    # Batch clauses to stay within context window (8 clauses ~ 4k tokens)
    batch_size = 8
    all_results = []
    
    for i in range(0, len(clauses), batch_size):
        batch = clauses[i:i+batch_size]
        
        # Build compact input
        input_json = [
            {
                "clause_id": c.clause_id,
                "heading": c.heading,
                "text": c.text[:3000],  # truncate long clauses
                "page_start": c.page_start,
                "page_end": c.page_end,
            }
            for c in batch
        ]
        
        task = Task(
            description=f"Classify these {len(batch)} contract clauses:\n{json.dumps(input_json, indent=2)}",
            expected_output="JSON array of classifications matching the schema",
            agent=classifier_agent,
        )
        
        crew = Crew(agents=[classifier_agent], tasks=[task], verbose=True)
        result = crew.kickoff()
        
        # Parse JSON from CrewAI result (string)
        try:
            classifications = json.loads(str(result).strip())
        except json.JSONDecodeError as e:
            print(f"JSON parse error on batch {i//batch_size}: {e}")
            print(f"Raw output: {result}")
            continue
        
        # Map back to Pydantic models
        for cls_data in classifications:
            # Find original clause for metadata
            orig = next(c for c in batch if c.clause_id == cls_data["clause_id"])
            all_results.append(ClassifiedClause(
                **orig.model_dump(),
                category=RiskCategory(cls_data["category"]),
                severity=Severity(cls_data["severity"]),
                rationale=cls_data["rationale"],
                redline_suggestion=cls_data.get("redline_suggestion"),
            ))
    
    return all_results

Test the classifier on one contract:

python -c "
from agents.parser import ContractParser
from agents.classifier import classify_clauses
from settings import settings

parser = ContractParser()
clauses = parser.extract_clauses(list(settings.contracts_dir.glob('*.pdf'))[0])
classified = classify_clauses(clauses)
for c in classified:
    print(f'{c.clause_id} [{c.severity.value}] {c.category.value}: {c.rationale[:80]}...')
"

Step 5: Build the reviewer agent for human-readable reports

The reviewer synthesizes classified clauses into an executive summary, flags critical items, and produces a markdown report your legal team can drop into a ticket or Notion page.

# agents/reviewer.py
from crewai import Agent, Task, Crew
from models import ClassifiedClause, ContractAnalysis, Severity
from settings import settings
import json
from datetime import datetime

llm = LLM(
    model=settings.llm_model,
    base_url=settings.llm_base_url,
    temperature=0.2,  # slightly higher for synthesis
)

reviewer_agent = Agent(
    role="Senior Legal Reviewer",
    goal="Produce a concise risk summary and actionable report for legal counsel",
    backstory="You summarize contract risk for GCs and CFOs. You write crisp, executive-ready memos.",
    llm=llm,
    verbose=True,
)

REVIEWER_PROMPT = """You are writing a contract risk memo for the General Counsel.

Input: JSON array of classified clauses with category, severity, rationale, and optional redline.

Produce a markdown report with:
1. **Executive Summary** (3-4 sentences: overall risk posture, top concerns)
2. **Critical & High Risk Findings** (table: Clause ID | Category | Severity | Key Issue | Suggested Action)
3. **Medium Risk Items** (bullet list grouped by category)
4. **Low Risk / Standard** (one line: count only)
5. **Negotiation Priorities** (ranked 1-3 with specific clause references)

Be specific. Reference clause IDs. No fluff.
"""

def generate_report(contract_id: str, filename: str, classified: list[ClassifiedClause]) -> ContractAnalysis:
    # Prepare compact input for LLM
    input_data = [
        {
            "clause_id": c.clause_id,
            "category": c.category.value,
            "severity": c.severity.value,
            "heading": c.heading,
            "rationale": c.rationale,
            "redline_suggestion": c.redline_suggestion,
        }
        for c in classified
    ]
    
    task = Task(
        description=f"Contract: {filename}\nClassified clauses:\n{json.dumps(input_data, indent=2)}",
        expected_output="Markdown risk memo per the template",
        agent=reviewer_agent,
    )
    
    crew = Crew(agents=[reviewer_agent], tasks=[task], verbose=True)
    markdown_report = str(crew.kickoff())
    
    # Compute summary stats
    high_risk = sum(1 for c in classified if c.severity == Severity.HIGH)
    critical_risk = sum(1 for c in classified if c.severity == Severity.CRITICAL)
    
    # Extract a one-paragraph summary from the report (first paragraph after executive summary)
    summary_lines = []
    in_summary = False
    for line in markdown_report.split('\n'):
        if line.startswith('## Executive Summary') or line.startswith('# Executive Summary'):
            in_summary = True
            continue
        if in_summary and line.startswith('##'):
            break
        if in_summary and line.strip():
            summary_lines.append(line.strip())
    
    analysis = ContractAnalysis(
        contract_id=contract_id,
        filename=filename,
        total_clauses=len(classified),
        classified_clauses=classified,
        summary=" ".join(summary_lines)[:500] if summary_lines else "See full report",
        high_risk_count=high_risk,
        critical_risk_count=critical_risk,
    )
    
    # Save markdown report
    output_path = settings.output_dir / f"{contract_id}_review.md"
    settings.output_dir.mkdir(parents=True, exist_ok=True)
    output_path.write_text(markdown_report)
    
    # Save structured JSON for downstream systems
    json_path = settings.output_dir / f"{contract_id}_analysis.json"
    json_path.write_text(analysis.model_dump_json(indent=2))
    
    return analysis

Step 6: Wire the pipeline end-to-end

A single orchestration script chains parser → classifier → reviewer for every PDF in the input directory. It also produces a portfolio-level summary CSV for tracking across contracts.

# run_pipeline.py
from pathlib import Path
import csv
from agents.parser import ContractParser
from agents.classifier import classify_clauses
from agents.reviewer import generate_report
from settings import settings

def main():
    parser = ContractParser()
    all_analyses = []
    
    pdf_files = list(settings.contracts_dir.glob("*.pdf"))
    if not pdf_files:
        print(f"No PDFs found in {settings.contracts_dir}")
        return
    
    for pdf_path in pdf_files:
        contract_id = pdf_path.stem
        print(f"\n{'='*60}")
        print(f"Processing: {contract_id}")
        print(f"{'='*60}")
        
        # Step 1: Parse
        clauses = parser.extract_clauses(pdf_path)
        print(f"  Extracted {len(clauses)} clauses")
        
        # Step 2: Classify
        classified = classify_clauses(clauses)
        print(f"  Classified {len(classified)} clauses")
        
        # Step 3: Review & Report
        analysis = generate_report(contract_id, pdf_path.name, classified)
        all_analyses.append(analysis)
        print(f"  Report saved to {settings.output_dir}/{contract_id}_review.md")
        print(f"  Critical: {analysis.critical_risk_count}, High: {analysis.high_risk_count}")
    
    # Portfolio summary CSV
    csv_path = settings.output_dir / "portfolio_summary.csv"
    with open(csv_path, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow([
            "contract_id", "filename", "total_clauses",
            "critical", "high", "medium", "low",
            "top_categories"
        ])
        for a in all_analyses:
            sev_counts = {}
            cat_counts = {}
            for c in a.classified_clauses:
                sev_counts[c.severity.value] = sev_counts.get(c.severity.value, 0) + 1
                cat_counts[c.category.value] = cat_counts.get(c.category.value, 0) + 1
            top_cats = sorted(cat_counts.items(), key=lambda x: -x[1])[:3]
            writer.writerow([
                a.contract_id, a.filename, a.total_clauses,
                sev_counts.get("critical", 0),
                sev_counts.get("high", 0),
                sev_counts.get("medium", 0),
                sev_counts.get("low", 0),
                "; ".join(f"{c}({n})" for c, n in top_cats)
            ])
    
    print(f"\nPortfolio summary: {csv_path}")
    print("Done.")

if __name__ == "__main__":
    main()

Run it:

python run_pipeline.py

Step 7: Verify success and iterate

Open the generated markdown report for one contract. You should see:

1. Executive Summary — a 3-4 sentence risk posture
2. Critical & High Risk Findings — a table with clause IDs you can trace back to the PDF
3. Negotiation Priorities — ranked, specific, actionable

Check the JSON output (output/<contract>_analysis.json) validates against your Pydantic model:

python -c "
from models import ContractAnalysis
import json
with open('output/your_contract_analysis.json') as f:
    data = json.load(f)
analysis = ContractAnalysis(**data)
print(f'Valid: {analysis.contract_id}')
print(f'Clauses: {analysis.total_clauses}')
print(f'Critical: {analysis.critical_risk_count}')
print(f'High: {analysis.high_risk_count}')
for c in analysis.classified_clauses[:3]:
    print(f'  {c.clause_id} [{c.severity.value}] {c.category.value}')
"

Open output/portfolio_summary.csv in Excel or csvlens to compare risk profiles across your contract corpus.

Common failure modes and fixes

Step 8: Extend for production use

Three changes make this pipeline production-ready:

1. Swap Ollama for a hosted gateway — if you route through an OpenAI-compatible endpoint that supports automatic fallback and per-token metering, you get provider diversity without code changes. Update config.yaml:

llm:
  model: "openai/gpt-4o-mini"
  base_url: "https://api.n4n.ai/v1"  # or your gateway
  temperature: 0.1

2. Add caching — wrap the classifier with a disk cache keyed by clause hash. Identical clauses across contracts (standard NDAs, boilerplate) skip the LLM entirely.

# agents/classifier.py (add at top)
import hashlib
import diskcache as dc

cache = dc.Cache("./cache/classifier")

def clause_hash(clause: Clause) -> str:
    return hashlib.sha256(clause.text.encode()).hexdigest()[:16]

# In classify_clauses, before calling crew:
cached = cache.get(clause_hash(c))
if cached:
    all_results.append(ClassifiedClause(**cached))
    continue
# ... after classification ...
cache.set(clause_hash(c), cls_data)

3. Human-in-the-loop queue — export critical/high findings to a review tool (Linear, Jira, Notion) with clause text, page reference, and suggested redline. The JSON output already has everything needed.

# agents/escalation.py
import requests
from models import ClassifiedClause, Severity

def escalate_to_linear(classified: list[ClassifiedClause], contract_id: str, api_key: str, team_id: str):
    critical_high = [c for c in classified if c.severity in (Severity.CRITICAL, Severity.HIGH)]
    for c in critical_high:
        payload = {
            "title": f"[{c.severity.value.upper()}] {contract_id}: {c.category.value} - {c.clause_id}",
            "description": f"**Clause:** {c.clause_id} (pp. {c.page_start}-{c.page_end})\n"
                          f"**Category:** {c.category.value}\n"
                          f"**Issue:** {c.rationale}\n"
                          f"**Text:**\n```\n{c.text[:1500]}\n```\n"
                          f"**Suggested Redline:** {c.redline_suggestion or 'TBD'}",
            "teamId": team_id,
            "labels": ["legal-review", c.category.value, c.severity.value],
        }
        requests.post(
            "https://api.linear.app/graphql",
            json={"query": "mutation IssueCreate($input: IssueCreateInput!) { issueCreate(input: $input) { success } }", "variables": {"input": payload}},
            headers={"Authorization": api_key, "Content-Type": "application/json"},
        )

This pipeline gives you a repeatable, auditable contract review process. The parser is deterministic, the classifier is constrained by a fixed taxonomy, and the reviewer produces artifacts your legal team can act on immediately. Swap the LLM backend, extend the taxonomy, or hook the escalation path — the structure holds.