tl;dr
This guide walks you step-by-step through building a state-of-the-art tabular review app from scratch in minutes.
The end result, shown below, will offer more advanced functionality than Harvey and Legora’s own tabular review tools at a fraction of the cost while achieving greater accuracy, lower latency, and guaranteeing zero generative hallucinations.
Every classification, extraction, and annotation will be grounded directly in the text of documents. Backlinks and panels will enable navigation across and between document structures and structured entities.
This will all be achieved without using a single generative model, which would be slow, expensive, and hallucination-prone. Instead, we’ll be relying on Isaacus’ cutting-edge specialized legal enrichment, embedding, and extraction models.
This guide is open source and free to use, adapt, extend, and even commercialize. You can follow along by either replicating the code here or cloning the corresponding Isaacus Cookbook for this guide.
Introduction
For decades, tables have been an integral part of legal review workflows, enabling lawyers to compare key terms across many documents at once.
With the advent of advanced language models, it has now become possible to automate one of the most time-consuming elements of tabular-based legal analysis: transforming unstructured documents into structured labels and entities.
Legal tech companies such as Harvey, Legora, and Clio have started releasing their own AI-powered tabular review apps to do exactly that.
Those apps often end up relying on generative models, which, although convenient in their flexibility, are slow, computationally inefficient, and prone to hallucination
No matter how advanced a generative model is, there is always a possibility that it may generate an extraction that is not contained within the text it is meant to be extracting from.
Fortunately, there is a better way of doing tabular review.
Instead of bootstrapping generative models for a task they were never trained for, you can use specialized enrichment, extraction, and classification models that directly annotate documents at the token level, offering statistically informative confidence scores for every annotation.
That is the approach we take in this guide. As you’ll see, using models that are fit for purpose ends up yielding a final product that is superior in every dimension to its generative-based competitors, from accuracy and efficiency all the way to the richness and immersiveness of its interface.
Setup
To get started, ensure you have an Isaacus account and valid API key. You can obtain an API key by following the first step of our API quickstart guide.
After obtaining an API key, set your ISAACUS_API_KEY environment variable to your key. You can do that by creating a .env file in the same directory as your code and then loading it with load_dotenv().
Let’s now install and import our dependencies and set up our environment variables and clients.
# Install dependencies.
%pip install -q fastapi uvicorn isaacus qdrant-client dotenv httpx
# Load dependencies.
import os
import json
import uuid
import threading
from typing import Any
import httpx
import uvicorn
from dotenv import load_dotenv
from fastapi import Body, FastAPI, HTTPException, BackgroundTasks
from isaacus import Isaacus
from qdrant_client import models, QdrantClient
from fastapi.responses import HTMLResponse, PlainTextResponse
from isaacus.types.ilgs.v1.document import Document
# Initialize an Isaacus API client.
load_dotenv() # Load environment variables from `.env` files if any are present.
isaacus_client = Isaacus(api_key=os.getenv("ISAACUS_API_KEY"))
# Initialize a FastAPI app.
app = FastAPI(title="Isaacus Tabular Review Demo Server")
app_state = {"docs": {}, "doc_order": []}
app_lock = threading.RLock()
# Initialize our Qdrant in-memory vector store.
vector_db = QdrantClient(":memory:", force_disable_check_same_thread=True)
Enrichment
Normally, building something like a knowledge graph would require hours, if not days, designing a schema that fits a specific use case.
Kanon 2 Enricher gives us a much faster way to build knowledge graphs. In a matter of milliseconds, the model can identify the key entities and relationships in a document and organize them into the Isaacus Legal Graph Schema (ILGS), a knowledge graph schema designed to capture most of the core features present in legal documents.
To begin using Kanon 2 Enricher, we’ll define a small wrapper function around it that batches input texts and returns an ILGS document for each one.
def batched(items: list[Any], size: int) -> list[list[Any]]:
"""Batch a list of items into smaller lists of a specified size."""
return [items[i : i + size] for i in range(0, len(items), size)]
def enrich(texts: list[str]) -> list[Any]:
"""Enrich texts using Kanon 2 Enricher and return the results."""
results: list[Any] = []
for batch in batched(texts, 8):
response = isaacus_client.enrichments.create(
model="kanon-2-enricher",
texts=batch,
overflow_strategy="auto",
)
results.extend([result.document for result in response.results])
return results
Now, let’s pass a small sample passage into the enricher and inspect the knowledge graph it produces.
This example is designed to highlight some of the core structures Kanon 2 Enricher can identify, including people, organizations, roles, locations, dates, and document segments. We’ll use it as a bite-sized window into the richer network of relationships and hierarchies that Kanon 2 Enricher can infer from legal text.
# Print the result of enriching an example document. example_text = """\ Sample Document Here is a list of some of the features Kanon 2 Enricher can automatically extract and organize into a knowledge graph: 1. Persons: John Doe wrote this document. 2. Relationships: John Doe is the director of Acme Corporation. 3. Locations: Acme Corporation is located in San Francisco. 4. Dates: This contract was signed on January 1, 2020. 5. Segments: All of these list items will be labeled as list items in the graph. 6. Much, much more! This is a confidentiality clause. This sentence will make more sense in Step 2 when we design a custom classification pipeline to label spans of the text based on a query. This Agreement shall be governed by the laws of the State of California. This sentence will make more sense in Step 3 when we create a pipeline to define custom relationships between entities in the graph based on a query.""" ilgs_document = enrich([example_text])[0] print(json.dumps(ilgs_document.to_dict(), indent=2))
{
"text": "Sample Document\n\nHere is a list of some of the features Kanon 2 Enricher can automatically extract and organize into a knowledge graph:\n1. Persons: John Doe wrote this document.\n2. Relationships: John Doe is the director of Acme Corporation.\n3. Locations: Acme Corporation is located in San Francisco.\n4. Dates: This contract was signed on January 1, 2020.\n5. Segments: All of these list items will be labeled as list items in the graph.\n6. Much, much more!\n\nThis is a confidentiality clause. This sentence will make more sense in Step 2 when we design a custom classification pipeline to label spans of the text based on a query.\n\nThis Agreement shall be governed by the laws of the State of California. This sentence will make more sense in Step 3 when we will create a pipeline to define custom relationships between entities in the graph based on a query.",
"title": {
"start": 0,
"end": 15
},
"subtitle": null,
"type": "other",
"jurisdiction": "US-CA",
"segments": [
{
"id": "seg:0",
"kind": "container",
"type": null,
"category": "main",
"type_name": null,
"code": null,
"title": null,
"parent": null,
"children": [
"seg:1",
"seg:2",
"seg:3",
"seg:4",
"seg:5",
"seg:6",
"seg:7"
],
"level": 0,
"span": {
"start": 17,
"end": 457
}
},
{
"id": "seg:1",
"kind": "unit",
"type": null,
"category": "main",
"type_name": null,
"code": null,
"title": null,
"parent": "seg:0",
"children": [],
"level": 1,
"span": {
"start": 17,
"end": 135
}
},
{
"id": "seg:2",
"kind": "item",
"type": null,
"category": "main",
"type_name": null,
"code": {
"start": 136,
"end": 138
},
"title": null,
"parent": "seg:0",
"children": [],
"level": 1,
"span": {
"start": 136,
"end": 177
}
},
{
"id": "seg:3",
"kind": "item",
"type": null,
"category": "main",
"type_name": null,
"code": {
"start": 178,
"end": 180
},
"title": null,
"parent": "seg:0",
"children": [],
"level": 1,
"span": {
"start": 178,
"end": 241
}
},
{
"id": "seg:4",
"kind": "item",
"type": null,
"category": "main",
"type_name": null,
"code": {
"start": 242,
"end": 244
},
"title": null,
"parent": "seg:0",
"children": [],
"level": 1,
"span": {
"start": 242,
"end": 301
}
},
{
"id": "seg:5",
"kind": "item",
"type": null,
"category": "main",
"type_name": null,
"code": {
"start": 302,
"end": 304
},
"title": null,
"parent": "seg:0",
"children": [],
"level": 1,
"span": {
"start": 302,
"end": 356
}
},
{
"id": "seg:6",
"kind": "item",
"type": null,
"category": "main",
"type_name": null,
"code": {
"start": 357,
"end": 359
},
"title": null,
"parent": "seg:0",
"children": [],
"level": 1,
"span": {
"start": 357,
"end": 437
}
},
{
"id": "seg:7",
"kind": "item",
"type": null,
"category": "main",
"type_name": null,
"code": {
"start": 438,
"end": 440
},
"title": null,
"parent": "seg:0",
"children": [],
"level": 1,
"span": {
"start": 438,
"end": 457
}
},
{
"id": "seg:8",
"kind": "unit",
"type": null,
"category": "main",
"type_name": null,
"code": null,
"title": null,
"parent": null,
"children": [],
"level": 0,
"span": {
"start": 459,
"end": 630
}
},
{
"id": "seg:9",
"kind": "unit",
"type": null,
"category": "main",
"type_name": null,
"code": null,
"title": null,
"parent": null,
"children": [],
"level": 0,
"span": {
"start": 632,
"end": 859
}
}
],
"crossreferences": [],
"locations": [
{
"id": "loc:0",
"name": {
"start": 287,
"end": 300
},
"type": "city",
"parent": "loc:1",
"children": [],
"mentions": [
{
"start": 287,
"end": 300
}
]
},
{
"id": "loc:1",
"name": {
"start": 693,
"end": 703
},
"type": "state",
"parent": null,
"children": [
"loc:0"
],
"mentions": [
{
"start": 693,
"end": 703
}
]
}
],
"persons": [
{
"id": "per:0",
"name": {
"start": 196,
"end": 204
},
"type": "natural",
"role": "director",
"parent": "per:1",
"children": [],
"residence": null,
"mentions": [
{
"start": 148,
"end": 156
},
{
"start": 196,
"end": 204
}
]
},
{
"id": "per:1",
"name": {
"start": 256,
"end": 272
},
"type": "corporate",
"role": "other",
"parent": null,
"children": [
"per:0"
],
"residence": "loc:0",
"mentions": [
{
"start": 224,
"end": 240
},
{
"start": 256,
"end": 272
}
]
},
{
"id": "per:2",
"name": {
"start": 684,
"end": 703
},
"type": "politic",
"role": "governing_jurisdiction",
"parent": null,
"children": [],
"residence": "loc:1",
"mentions": [
{
"start": 684,
"end": 703
}
]
}
],
"emails": [],
"websites": [],
"phone_numbers": [],
"id_numbers": [],
"terms": [],
"external_documents": [],
"quotes": [],
"dates": [
{
"value": "2020-01-01",
"type": "signature",
"person": null,
"mentions": [
{
"start": 340,
"end": 355
}
]
}
],
"headings": [
{
"start": 0,
"end": 15
}
],
"junk": [],
"version": "ilgs@1"
}
For such a small passage, the output is already quite rich! Although the raw ILGS document is not especially easy to read, we can already see that Kanon 2 Enricher has identified multiple entity types and structured them in useful ways. In this example, it captures a natural person (John Doe), a corporate person (Acme Corporation), and a governing jurisdiction (the State of California), along with their roles and relationships. It also identifies two locations, classifies them by type, extracts a date labeled as a signature, and breaks the text into several document segments.
Because the raw output is span-based, it is a little hard to read as a human. To make it easier to inspect, we can decode some of those spans back into text and verify that the knowledge graph is accurate.
# Render some of the key extractions from the document in a more human-readable format.
person_names = [p.name.decode(ilgs_document.text) for p in ilgs_document.persons]
location_names = [l.name.decode(ilgs_document.text) for l in ilgs_document.locations]
segment_kinds = [s.kind for s in ilgs_document.segments]
john = ilgs_document.persons[0]
acme = ilgs_document.persons[1]
sf = ilgs_document.locations[0]
cali = ilgs_document.locations[1]
date = ilgs_document.dates[0]
print(
f"""\
Title: {ilgs_document.title.decode(ilgs_document.text)}
Entities:
- {person_names[0]} (id: {john.id})
- {person_names[1]} (id: {acme.id}; residence: {acme.residence})
Locations:
- {location_names[0]} (id: {sf.id})
- {location_names[1]} (id: {cali.id})
Dates:
- {date.value} (type: {date.type})
Relationships:
- {person_names[0]} is the {john.role} of {john.parent}
- {location_names[0]} is the child of {sf.parent}""".strip()
)
Title: Sample Document Entities: - John Doe (id: per:0) - Acme Corporation (id: per:1; residence: loc:0) Locations: - San Francisco (id: loc:0) - California (id: loc:1) Dates: - 2020-01-01 (type: signature) Relationships: - John Doe is the director of per:1 - San Francisco is the child of loc:1
Span-level classification
With our enrichment function done, we can now turn our attention to adding support for custom labeling, which is a core part of the AI-powered tabular review experience. To do this, we’ll use another Isaacus model, Kanon 2 Embedder, the world’s best legal embedding model.
Although embedding models are commonly used for retrieval, they can also perform classification tasks by encoding spans of text and comparing them against a query framed as a label or attribute. With the right filtering logic, this gives us a flexible and reliable way to add custom span-level labels to a document and extend the knowledge graph with features tailored to a specific use case.
To get started, we’ll define two small helper functions. The first batches a list into smaller chunks. The second sends those chunks to Kanon 2 Embedder and returns vectors for either document spans or queries. We’ll build the actual classification logic on top of these helpers in the next step.
def embed(texts: list[str], task: str) -> list[list[float]]:
vectors: list[list[float]] = []
for batch in batched(texts, 128):
response = isaacus_client.embeddings.create(
model="kanon-2-embedder",
texts=batch,
task=task,
)
vectors.extend([list(item.embedding) for item in response.embeddings])
return vectors
If we called the function above directly, it would just return embeddings. To make those embeddings useful, we need to store them in an index that we can query later.
We also do not want to embed the full document as a single block. As with RAG, classification works best over meaningful spans of text. Those spans might be sentences, clauses, phrases, or larger multi-sentence sections, depending on the task.
Choosing the level of span granularity by hand can be difficult. One advantage of Kanon 2 Enricher is that it already breaks the document into meaningful, hierarchical segments. That gives us a natural set of embedding units while preserving the structure of the original document.
To store those embeddings, we’ll use Qdrant, a vector database that supports vector search with custom metadata. We can use the unique IDs from the ILGS document as metadata on each vector, which makes it easy to link every result back to the original span in the knowledge graph.
The code below sets up the document record, collects the ILGS segments we want to embed, and builds a Qdrant index for each document. That gives us a clean foundation for span-level retrieval and classification later on.
def create_record(file_name: str, ilgs_document: Document) -> dict[str, Any]:
"""Create a new document record in the app state."""
file_id = str(uuid.uuid4())
record = {
"file_id": file_id,
"file_name": file_name,
"ilgs_document": ilgs_document,
"columns": {},
"index_status": "pending",
}
with app_lock:
app_state["docs"][file_id] = record
app_state["doc_order"].append(file_id)
return record
def get_record(file_id: str) -> dict[str, Any]:
"""Retrieve a stored document record or raise a 404 error if it does not exist."""
with app_lock:
record = app_state["docs"].get(file_id)
if record is None:
raise HTTPException(status_code=404, detail=f"Document not found: {file_id}.")
return record
def collect_segments(doc: Document) -> list[tuple[str, str]]:
"""Collect all non-empty segment spans from a document."""
segments: list[tuple[str, str]] = []
for segment in doc.segments:
text = segment.span.decode(doc.text).strip()
if text:
segments.append((segment.id, text))
return segments
def build_index(file_id: str) -> None:
"""Build a Qdrant index for a document using ILGS segments as embedding spans."""
record = get_record(file_id)
doc = record["ilgs_document"]
segments = collect_segments(doc)
if not segments:
return
texts = [text for _, text in segments]
vectors = embed(texts, task="retrieval/document")
collection_name = f"doc_{file_id.replace('-', '_')}"
points = [
models.PointStruct(
id=i,
vector=vector,
payload={
"segment_id": segment_id,
"text": text,
},
)
for i, ((segment_id, text), vector) in enumerate(zip(segments, vectors))
]
with app_lock:
if vector_db.collection_exists(collection_name):
vector_db.delete_collection(collection_name)
vector_db.create_collection(
collection_name=collection_name,
vectors_config=models.VectorParams(
size=len(vectors[0]),
distance=models.Distance.COSINE,
),
)
vector_db.upsert(collection_name=collection_name, points=points)
record["collection_name"] = collection_name
record["index_status"] = "ready"
With the vector database logic in place, we can now turn to querying the index for classification.
We start with two small helper functions: spans_overlap(), which checks whether two ILGS spans overlap, and map_segments(), which maps segment IDs back to their corresponding segment objects. These will make it easier to work with span-based results later on.
We then define ensure_index(), which makes sure a document’s vector index has been built and updates the record with progress flags along the way. This is important because our code is designed to sit inside a live asynchronous application, where we will need to keep track of whether a document is still being processed or is ready to query.
Finally, we define delete_record(), which removes a document from the application state and cleans up its associated Qdrant collection when the document is deleted.
def spans_overlap(a: Any, b: Any) -> bool:
"""Determine whether two ILGS spans overlap."""
return a and b and a.start < b.end and a.end > b.start
def map_segments(doc: Document) -> dict[str, Any]:
"""Map segment IDs back to segment objects."""
return {segment.id: segment for segment in doc.segments}
def ensure_index(file_id: str) -> dict[str, Any]:
"""Ensure the vector index for a document exists, building it if needed."""
record = get_record(file_id)
if record["index_status"] == "ready":
return record
with app_lock:
record["index_status"] = "building"
build_index(file_id)
with app_lock:
record["index_status"] = "ready"
return record
def delete_record(file_id: str) -> dict[str, Any]:
"""Delete a document record and its associated Qdrant collection."""
record = get_record(file_id)
collection_name = f"doc_{file_id.replace('-', '_')}"
with app_lock:
if vector_db.collection_exists(collection_name):
vector_db.delete_collection(collection_name)
app_state["docs"].pop(file_id, None)
app_state["doc_order"] = [
item for item in app_state["doc_order"] if item != file_id
]
return record
Now, we can write the main querying logic for span-level classification.
In a standard semantic search pipeline, we would query the vector database and return the most relevant segments. Here, though, we want to treat the query more like a classification prompt, so we also need a way to filter out segments that are not similar enough to be meaningful matches.
To do that, we apply a similarity threshold to the scores returned by Qdrant and keep only the segments above that cutoff. Kanon 2 Embedder was not explicitly trained around a fixed classification threshold, but in practice a cosine similarity of 0.4 works well as a starting point. The right threshold will still depend on the task and the kind of labels you are trying to surface.
The code below applies that threshold, resolves the matching vectors back to ILGS spans, and then deduplicates overlapping results. If both a parent span and one of its children are returned, we keep only the parent span, since it usually carries more useful context for downstream review.
SIMILARITY_THRESHOLD = 0.4
def search_spans(
file_id: str,
query: str,
threshold: float = SIMILARITY_THRESHOLD,
) -> list[dict[str, Any]]:
"""Run semantic search over Qdrant hits, then resolve them back to ILGS spans."""
record = ensure_index(file_id)
collection_name = f"doc_{file_id.replace('-', '_')}"
if not vector_db.collection_exists(collection_name):
return []
query_vector = embed([query], "retrieval/query")[0]
with app_lock:
response = vector_db.query_points(
collection_name=collection_name,
query=query_vector,
score_threshold=threshold,
)
doc = record["ilgs_document"]
segments = map_segments(doc)
hits: list[tuple[Any, float, str | None]] = []
for point in response.points:
payload = point.payload or {}
segment_id = payload.get("segment_id")
if not segment_id or segment_id not in segments:
continue
hits.append(
(
segments[segment_id],
float(point.score),
payload.get("text"),
)
)
# Prefer larger spans first so parent spans win over overlapping children.
hits.sort(
key=lambda item: (
-(item[0].span.end - item[0].span.start),
-item[1],
)
)
deduped_hits: list[tuple[Any, float, str | None]] = []
for segment, score, text in hits:
if any(spans_overlap(segment.span, existing[0].span) for existing in deduped_hits):
continue
deduped_hits.append((segment, score, text))
# Return final results ordered by score.
deduped_hits.sort(key=lambda item: item[1], reverse=True)
return [
{
"score": score,
"text": text or segment.span.decode(doc.text).strip(),
"metadata": {
"segment_id": segment.id,
"start": segment.span.start,
"end": segment.span.end,
},
}
for segment, score, text in deduped_hits
]
With the helper functions in place, we can now try out span-level classification on our example text.
In this case, we’ll look for segments related to confidentiality obligations. We can phrase that as a natural-language query, embed it as a vector, and use our span search logic to return the most relevant segments from the document’s Qdrant index.
The code below creates a document record, builds its vector index, runs the query, and prints the matching spans along with their scores and character offsets.
# Create a document record, build an index for it, and run a sample query against the index to retrieve relevant spans from the document.
example_record = create_record(
file_name="example.txt",
ilgs_document=ilgs_document,
)
build_index(example_record["file_id"])
example_record["index_status"] = "ready"
query = "What are the confidentiality obligations outlined in this agreement?"
results = search_spans(example_record["file_id"], query)
for result in results:
print(f"Score: {result['score']:.3f}")
print(f"Character offsets: {result['metadata']['start']}:{result['metadata']['end']}")
print(result["text"])
print()
Score: 0.482 Character offsets: 459:630 This is a confidentiality clause. This sentence will make more sense in Step 2 when we design a custom classification pipeline to label spans of the text based on a query.
Sure enough, our classification logic has surfaced the passage that best answers our query!
With traditional chunking and retrieval techniques, we would not be able to classify a span at this level of granularity. Indeed, in many legal AI applications the same query would return an entire chunk, likely with the irrelevant governing law clause included in the context. When context is contaminated like this, it not only makes for a worse reviewing experience, but it also poisons the context window of generative models in RAG pipelines, leading to a higher incidence of hallucinations.
We’ll delete this test record so it does not show up in the app later. Before doing that, though, it is useful to inspect the record format directly, since this is the same structure the server will keep in memory and eventually send back to the client.
To keep the output readable, we’ll remove the full ILGS document before printing the record.
deleted_record = delete_record(example_record["file_id"])
record_preview = {
**deleted_record,
"ilgs_document": "ILGS document data removed for brevity.",
}
print(
"Here is what a stored record looks like:\n"
+ json.dumps(record_preview, indent=2, default=str)
)
Here is what a stored record looks like:
{
"file_id": "29cca176-a399-437b-b175-cc9f7eab0662",
"file_name": "example.txt",
"ilgs_document": "ILGS document data removed for brevity.",
"columns": {},
"index_status": "ready",
"collection_name": "doc_29cca176_a399_437b_b175_cc9f7eab0662"
}
Entity linking and relationship extraction
At this point, we can already reproduce much of the functionality offered by Harvey and Legora, and then some, all without relying on a generative model. We can go further, however, by allowing our users to define custom relationships on top of our knowledge graph at run time.
To do that, we’ll use Kanon Answer Extractor. Given a user query, it returns answer spans from the source document. We can then cross-reference those spans against the ILGS document and check which entities are mentioned within them. Any matching entities can be linked back to the query, effectively creating new relationships on the fly without having to define them in advance.
def link_entities(entities: list[Any], answer: Any, entity_ids: list[str], seen: set[str]) -> None:
"""If an entity mention overlaps with an extracted answer span, save its ID so it can be linked back to the original ILGS document later."""
for entity in entities:
spans = list(entity.mentions)
if any(spans_overlap(span, answer) for span in spans) and entity.id not in seen:
seen.add(entity.id)
entity_ids.append(entity.id)
def extract_entities(doc: Document, query: str) -> list[str]:
"""Run QA extraction for a query, then link any overlapping ILGS entities back to the extracted answer spans."""
response = isaacus_client.extractions.qa.create(
model="kanon-answer-extractor",
query=query,
texts=[doc.text],
ignore_inextractability=False,
)
entity_ids: list[str] = []
seen: set[str] = set()
for answer in response.extractions[0].answers:
link_entities(doc.persons, answer, entity_ids, seen)
link_entities(doc.locations, answer, entity_ids, seen)
link_entities(doc.terms, answer, entity_ids, seen)
return entity_ids
Now, let’s try this code out on our example document.
The extraction pipeline should identify the span that best answers the query, then link any entities mentioned in that span back to the original ILGS document.
In this example, we want to identify the governing law of the agreement. Here, the State of California appears as both a political entity and a linked location. Because they are already linked via the residence attribute, this single query is actually identifying multiple links at once, highlighting the power of approaching document review as a knowledge graph problem.
extracted_entity_ids = extract_entities(
ilgs_document,
"What is the governing jurisdiction of this agreement?",
)
entity_lookup = {
entity.id: entity
for entity in (
ilgs_document.persons + ilgs_document.locations + ilgs_document.terms
)
}
entity = entity_lookup[extracted_entity_ids[0]]
entity_location = entity_lookup[extracted_entity_ids[1]]
entity_location_name = entity_location.name.decode(ilgs_document.text)
entity_location_type = entity_location.type
entity_name = entity.name.decode(ilgs_document.text)
print("Governing jurisdiction")
print("---------------------------")
print(f"Name: {entity_name}")
print(f"Type: {entity.type}")
print(f"ID: {entity.id}")
print(f"Residence: {entity_location_name} ({entity_location_type.title()})")
Governing jurisdiction --------------------------- Name: State of California Type: politic ID: per:2 Residence: California (State)
We’ve now linked the governing jurisdiction to the entity of the State of California, which is itself linked to the location of California. If we wanted to extend this entity with more labels and relationships, all we’d need to do is send more queries, and the network would automatically expand as everything is linked using the unique IDs first extracted by the enrichment model.
To create a single endpoint that the server can use, we will wrap our querying logic into a simple helper. For now, we’ll go with two modes of querying: span-level classification-style columns and entity extraction for relationship-style columns.
def classify_document(file_id: str, query: str, col_type: str) -> dict[str, Any]:
"""Run the appropriate classification or entity linking pipeline based on the column type supplied."""
record = ensure_index(file_id)
if col_type == "span":
return {"col_data": search_spans(file_id, query)}
if col_type == "entity":
return {"col_data": extract_entities(record["ilgs_document"], query)}
raise ValueError(f"Unsupported column type: {col_type}")
Server
With entity linking complete, we now have all the core intelligence capabilities we need to power our tabular review app.
- Create a knowledge graph from each document using Kanon 2 Enricher.
- Extend that graph with custom span-level classification using Kanon 2 Embedder and Qdrant.
- Add entity linking and relationship extraction using Kanon Answer Extractor.
All that remains is to expose this logic through a live server that can receive requests from the client, run the appropriate enrichment and querying logic, and return the results in a format the UI can display.
To get started, we’ll define two small helpers. The first packages a stored document record into the response shape expected by the client. The second triggers index building in the background when a document is uploaded, so the document is ready to query by the time the user interacts with it.
def package_record(record: dict[str, Any]) -> dict[str, Any]:
"""Convert a stored record into the response shape expected by the client."""
return {
"doc_id": record["file_id"],
"file_name": record["file_name"],
"document": record["ilgs_document"].model_dump(mode="json"),
"columns": record["columns"],
"index_status": record["index_status"],
}
def build_index_in_background(file_id: str) -> None:
"""Trigger index building for a document if it is not already ready."""
if app_state["docs"][file_id]["index_status"] != "ready":
ensure_index(file_id)
Now let’s define the server logic for handling the initial request that enriches uploaded documents. To keep this guide simple, we’ve baked OCR into the client for our server.
# Fetch the HTML for our viewer client from GitHub.
client_html = httpx.get(
"https://raw.githubusercontent.com/isaacus-dev/cookbooks/main/cookbooks/tabular-review/viewer.html"
).text
# Serve the client HTML file at the root endpoint.
@app.get("/", response_class=HTMLResponse)
def index():
return HTMLResponse(content=client_html)
# Receive documents from the client, convert them to ILGS, create records, trigger background indexing, and return the packaged records to the UI.
@app.post("/review")
def review(body: dict[str, Any], background_tasks: BackgroundTasks):
lines: list[str] = []
for batch in batched(body["documents"], 128):
texts = [str(doc["text"]) for doc in batch]
ilgs_documents = enrich(texts)
for input_doc, ilgs_document in zip(batch, ilgs_documents):
record = create_record(
file_name=str(input_doc["file_name"]),
ilgs_document=ilgs_document,
)
lines.append(json.dumps(package_record(record), ensure_ascii=False))
background_tasks.add_task(build_index_in_background, record["file_id"])
return PlainTextResponse(
"\n".join(lines),
media_type="application/x-ndjson",
headers={"Cache-Control": "no-store", "X-Tabular-Review-Version": "Isaacus Tabular Review Demo Server"},
)
We then define the endpoints for custom column queries and document deletion.
For column queries, the server receives a list of document IDs, a column type, and a natural-language query. It then runs the appropriate retrieval or extraction logic for each document and returns the results to the client for display in the UI.
For document deletion, the server removes the document from the application state and deletes its corresponding vector index from Qdrant to keep everything clean and in sync between the server and the client.
# Receive a custom column query, run the appropriate retrieval or extraction logic for each document, and return the results to the client.
@app.post("/column-query")
def column_query(body: dict[str, Any] = Body(default_factory=dict)) -> dict[str, Any]:
doc_ids = list(body["doc_ids"])
query = str(body["query"]).strip()
col_type = str(body["col_type"]).strip()
column_id = str(uuid.uuid4())
results: list[dict[str, Any]] = []
for file_id in doc_ids:
col_data = classify_document(file_id, query, col_type)["col_data"]
get_record(file_id)["columns"][column_id] = {
"column_id": column_id,
"query": query,
"col_type": col_type,
"col_data": col_data,
}
results.append({"doc_id": file_id, "col_data": col_data})
if len(results) == 1:
return {"column_id": column_id, "col_data": results[0]["col_data"]}
return {"column_id": column_id, "results": results}
# Delete a document and its associated vector index.
@app.delete("/docs/{doc_id}")
def delete_doc(doc_id: str) -> dict[str, Any]:
record = delete_record(doc_id)
return {
"deleted": True,
"doc_id": doc_id,
"file_name": record["file_name"],
}
Conclusion
We’ve now reached the final step of this guide. At this point, we have everything we need to run our tabular review app: a pipeline for building and extending a knowledge graph and a server that can process client requests and return results in a format the client can use.
All that remains is to start the server and connect it to the client.
As you’ll see, our application can do things Harvey and Legora cannot. Rather than treating tabular review as a sequence of extracted answers inside a table, we turn documents into human-navigable knowledge graphs, empowering users to inspect source texts, move through linked entities and sections, and verify how every annotation connects back to the document itself.
To use the app, simply run the code below and then open your server at http://127.0.0.1:8000. From there, you can upload files, add new columns using natural language queries, and explore the resulting knowledge graph through both entity panels and an interactive ILGS document viewer.
We encourage you to build on top of this foundation and adapt the experience to fit your own workflow and use case.
def run_server():
uvicorn.run(app, host="127.0.0.1", port=8000)
threading.Thread(target=run_server, daemon=True).start()
print("Server started at http://127.0.0.1:8000")
If you enjoyed this guide and would like to stay up to date on our latest releases, the best way to get notified is to follow us on LinkedIn, Twitter/X, or Reddit. To access the source code behind this project, check out the corresponding Isaacus Cookbook on our GitHub.