Contract review with LangChain: extracting clauses

Legal teams drown in PDFs, but engineers can automate the first pass. This tutorial builds a contract review LangChain clause extraction pipeline that ingests a sample agreement, splits it into sections, and classifies each clause with an LLM.

Prerequisites

• Python 3.10 or newer
• pip install langchain langchain-openai langchain-community pypdf
• An OpenAI-compatible API key. If you want a single endpoint that fronts 240+ models with automatic fallback when a provider is rate-limited, n4n.ai exposes an OpenAI-compatible API you can drop into ChatOpenAI.
• A sample contract PDF. Grab any NDA or use a two-page text saved as PDF.

export OPENAI_API_KEY="sk-..."

Load the contract PDF

PyPDFLoader returns one Document per page. Keep the page metadata; you will need it for traceability.

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("sample_contract.pdf")
pages = loader.load()
print(f"Loaded {len(pages)} pages")

Expected output:

Loaded 2 pages

Split into clause-sized chunks

Page-level splitting is too coarse. Most clauses sit under headings like “1. Confidentiality” or “Article 4 – Termination”. A RecursiveCharacterTextSplitter with legal-specific separators beats a fixed window.

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    separators=["\n\n", "\n", "Section", "Article", "Clause"],
    chunk_size=800,
    chunk_overlap=50,
)
clauses = splitter.split_documents(pages)
print(f"Split into {len(clauses)} candidate chunks")

Expected output:

Split into 14 candidate chunks

Overlap preserves cross-references (“see Section 2”). Drop overlap to zero only if your clauses are strictly non-contiguous.

Define the extraction schema

Unstructured text out of an LLM is useless in a pipeline. Pydantic forces a contract between your code and the model.

from pydantic import BaseModel, Field

class Clause(BaseModel):
    clause_type: str = Field(description="e.g. Confidentiality, Indemnification, Termination")
    summary: str = Field(description="One sentence plain-English summary")
    parties: list[str] = Field(description="Entities bound by the clause")
    risk_flag: bool = Field(description="True if clause is one-sided or unusual")

risk_flag is a heuristic. Do not treat it as legal advice; treat it as a triage signal for human review.

Build the extraction chain

LangChain Expression Language (LCEL) composes the prompt, model, and parser into a runnable. Use temperature=0 for deterministic extraction.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a contract review assistant. Extract the clause from the text."),
    ("human", "{text}")
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = JsonOutputParser(pydantic_object=Clause)

chain = prompt | llm | parser

If you route through n4n.ai, set base_url="https://api.n4n.ai/v1" and keep the same ChatOpenAI interface; the gateway forwards cache-control hints so repeated runs on the same PDF hit provider caches.

Run extraction over chunks

Iterate, skip noise, catch parsing errors so one bad chunk does not kill the batch.

import json

results = []
for i, chunk in enumerate(clauses):
    text = chunk.page_content.strip()
    if len(text) < 50:
        continue
    try:
        data = chain.invoke({"text": text})
        data["chunk_id"] = i
        results.append(data)
    except Exception as e:
        print(f"Chunk {i} failed: {e}")

print(f"Extracted {len(results)} clauses")

Expected output:

Extracted 9 clauses

Inspect a sample clause

Print the first result to verify the schema holds.

print(json.dumps(results[0], indent=2))

Example output:

{
  "clause_type": "Confidentiality",
  "summary": "Receiving party must not disclose confidential info for 3 years.",
  "parties": ["Acme Corp", "Beta LLC"],
  "risk_flag": false,
  "chunk_id": 2
}

Filter and rank risks

Legal wants the outliers first. Sort by risk_flag, then clause type.

flagged = [r for r in results if r["risk_flag"]]
print(f"{len(flagged)} clauses need human review")

If risk_flag is true, route to a senior attorney. The pipeline already cut first-pass review from hours to minutes.

Export to JSON

Write the full set for downstream ingest into a search index or case management system.

with open("extracted_clauses.json", "w") as f:
    json.dump(results, f, indent=2)

Handling tables and signatures

Standard splitters mangle tabular clauses (payment schedules, equity tables). Use PdfPlumberLoader to extract tables and feed each row as a separate chunk.

from langchain_community.document_loaders import PdfPlumberLoader

loader = PdfPlumberLoader("sample_contract.pdf")
pages = loader.load()
# tables accessible via pages[i].metadata["tables"]

Attach table text to the chunk before extraction. The same Clause schema works; just prompt the model to expect row-based input by appending “Input may be a table row” to the system message.

Tune the prompt for legal specificity

Generic prompts miss nuance. Add few-shot examples for your contract type:

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a contract review assistant. Extract the clause. "
               "Flag non-standard indemnification caps as risk."),
    ("human", "Example: 'Supplier indemnifies Buyer for all claims.' -> risk_flag true"),
    ("human", "{text}")
])

Test on a held-out contract before trusting risk_flag in production.

Evaluate extraction quality

Build a gold set of 20 clauses labeled by a paralegal. Compute precision against clause_type:

def precision(preds, golds):
    correct = sum(p == g for p, g in zip(preds, golds))
    return correct / len(golds)

# preds/golds are lists of clause_type strings aligned by chunk
print(f"Type precision: {precision(pred_types, gold_types):.2f}")

Aim for >0.9 before human-in-the-loop review. Below that, tighten the prompt or upgrade the model.

Production notes

Batch calls with chain.batch() to respect rate limits. Set max_concurrency=5 to avoid 429s.

batch_results = chain.batch(
    [{"text": c.page_content} for c in clauses if len(c.page_content) > 50],
    config={"max_concurrency": 5}
)

If a provider degrades, a gateway with automatic fallback saves the job—n4n.ai switches models mid-batch without code changes. Cache the raw PDF parse; re-running extraction on the same bytes should hit provider prompt caches when you forward cache_control headers.

That is a complete contract review LangChain clause extraction loop. Swap the LLM, tune the schema, and point it at your document store.