What Are AI Agents for Legal Intake and Research
AI agents for legal intake and research are autonomous software systems that use large language models (LLMs), retrieval-augmented generation (RAG), and workflow orchestration to handle two of the most time-intensive tasks in legal practice: onboarding new clients and conducting case law research. Unlike simple chatbots or document templates, these agents can reason over unstructured data, make decisions about classification and routing, extract structured information from messy documents, and perform multi-step research across legal databases — all while maintaining context, citing sources, and adhering to ethical guardrails.
A legal intake agent ingests prospective client communications (emails, form submissions, uploaded documents), classifies the matter type, extracts key parties, deadlines, and jurisdictional facts, flags conflicts, and routes the matter to the appropriate attorney or practice group. A legal research agent takes a natural language query — such as "find recent Ninth Circuit rulings on fair use in AI training datasets" — and performs semantic search across case law databases, retrieves relevant opinions, summarizes holdings, and generates a formatted research memo with pinpoint citations.
Together, these agents form the backbone of an AI-augmented law practice: they reduce administrative overhead by 60–80%, eliminate misfiled intake, and produce research deliverables in minutes rather than days. Critically, they do not replace attorney judgment — they amplify it by handling the mechanical grunt work so lawyers can focus on strategy, advocacy, and client counseling.
The Core Components
Every effective legal AI agent shares a common architectural skeleton:
- LLM Backbone: A capable language model (GPT-4o, Claude 3.5 Sonnet, or a fine-tuned open model) that handles reasoning, summarization, and structured output generation.
- Retrieval Engine: A vector database (Pinecone, Weaviate, or pgvector) paired with an embedding model to enable semantic search over case law, statutes, and firm knowledge bases.
- Tool Belt: Function-calling integrations that let the agent search legal APIs (CourtListener, Caselaw Access Project, Westlaw, LexisNexis), query internal DMS systems, send emails, and update CRM records.
- Orchestration Layer: A directed graph or state machine (LangGraph, Haystack pipelines, or custom workflows) that sequences operations: classify → extract → conflict-check → route, or query → retrieve → rerank → synthesize → cite.
- Guardrails & Audit: Policy filters for PII redaction, ethical walls, unauthorized practice of law detection, and full audit logging for every agent action.
Why It Matters for Law Firms
The economics are compelling. AmLaw 100 firms report that associates spend 30–40% of their time on intake triage and research — tasks that are necessary but do not require the full exercise of legal judgment. An AI agent handling intake can process 500+ matters per day with consistent accuracy, flagging high-value or emergency cases for immediate human attention. Research agents compress 6-hour manual research sessions into 3-minute query-to-memo workflows, while providing better coverage of obscure precedents that a rushed associate might miss.
Beyond efficiency, agents improve consistency: every intake is classified using the same rubric, every research memo follows the firm's citation style guide, and every step is logged for malpractice audit trails. For boutique firms and solo practitioners, these agents level the playing field, providing research depth and intake rigor that rivals large firm infrastructure at a fraction of the cost.
Building an AI Intake Agent
Architecture Overview
The intake agent follows a pipeline pattern. Raw input arrives via email or web form → the agent classifies the matter type → extracts structured entities → checks for conflicts → generates a standardized intake summary → routes to the responsible attorney and logs everything to the practice management system. Each stage is a discrete node in an orchestration graph, with conditional branching for edge cases (e.g., "is this a mass tort inquiry? Route to the mass tort group and flag for rapid response").
Setting Up the Environment
We will build a production-ready intake agent using Python, LangChain for the LLM interface, Pydantic for structured output validation, and a lightweight vector store for conflict checking. Install the dependencies:
# requirements.txt
langchain>=0.3.0
langchain-openai>=0.2.0
langgraph>=0.2.0
pydantic>=2.0.0
chromadb>=0.5.0
sqlalchemy>=2.0.0
python-dotenv>=1.0.0
fastapi>=0.115.0
uvicorn>=0.30.0pip install -r requirements.txtSet up your environment variables for API keys and database connections:
# .env
OPENAI_API_KEY=sk-your-key-here
DATABASE_URL=postgresql://user:pass@localhost:5432/legal_intake
FIRM_NAME=Smith & Associates LLPBuilding the Intake Pipeline
We start by defining our structured data models. Pydantic gives us runtime validation and forces the LLM to conform to a strict schema — essential for legal workflows where a malformed extraction could mean a missed statute of limitations.
# models/intake_schema.py
from pydantic import BaseModel, Field, field_validator
from enum import Enum
from typing import Optional, List
from datetime import date
class MatterType(str, Enum):
PERSONAL_INJURY = "personal_injury"
FAMILY_LAW = "family_law"
CORPORATE = "corporate"
REAL_ESTATE = "real_estate"
EMPLOYMENT = "employment"
IP = "intellectual_property"
CRIMINAL_DEFENSE = "criminal_defense"
IMMIGRATION = "immigration"
TAX = "tax"
OTHER = "other"
class UrgencyLevel(str, Enum):
CRITICAL = "critical" # immediate attention required
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
class ExtractedEntity(BaseModel):
"""A key entity extracted from the intake communication."""
entity_type: str # person, organization, date, monetary_amount, case_reference
value: str
confidence: float = Field(ge=0.0, le=1.0)
context_snippet: str
class IntakeSummary(BaseModel):
"""The fully structured output of the intake agent."""
matter_type: MatterType
matter_subtype: Optional[str] = None
urgency: UrgencyLevel
primary_client_name: str
opposing_party_names: List[str] = Field(default_factory=list)
jurisdiction: Optional[str] = None
estimated_claim_value: Optional[float] = None
critical_deadlines: List[date] = Field(default_factory=list)
key_facts: List[str] = Field(min_length=1, max_length=10)
extracted_entities: List[ExtractedEntity] = Field(default_factory=list)
conflict_check_required: bool = True
recommended_attorney: Optional[str] = None
recommended_practice_group: Optional[str] = None
summary_paragraph: str = Field(min_length=50, max_length=500)
@field_validator('summary_paragraph')
@classmethod
def no_privileged_phrases(cls, v):
forbidden = ['attorney-client privilege', 'work product', 'confidential per']
for phrase in forbidden:
if phrase.lower() in v.lower():
# This is metadata, not the privilege itself
pass
return vNext, we build the classification and extraction chain. We use LangChain's structured output parser with a carefully engineered prompt that includes firm-specific rubrics and jurisdictional triggers.
# chains/intake_extraction.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from models.intake_schema import IntakeSummary, MatterType, UrgencyLevel
import json
class IntakeExtractionChain:
"""Extracts structured intake data from unstructured client communications."""
def __init__(self, model_name: str = "gpt-4o-mini", temperature: float = 0.1):
self.llm = ChatOpenAI(
model=model_name,
temperature=temperature,
max_tokens=2000
)
self.parser = PydanticOutputParser(pydantic_object=IntakeSummary)
def build_prompt(self) -> ChatPromptTemplate:
system_template = """You are an expert legal intake specialist at {firm_name}.
Your task is to analyze prospective client communications and produce a structured intake summary.
CLASSIFICATION RULES:
- PERSONAL_INJURY: auto accidents, slip-and-fall, medical malpractice, product liability
- FAMILY_LAW: divorce, custody, support, adoption, domestic violence
- CORPORATE: M&A, contract disputes, shareholder issues, business formation
- REAL_ESTATE: property disputes, landlord-tenant, zoning, eminent domain
- EMPLOYMENT: wrongful termination, discrimination, wage disputes, non-compete
- IP: patents, trademarks, copyrights, trade secrets, fair use
- CRIMINAL_DEFENSE: any criminal charge representation
- IMMIGRATION: visas, green cards, asylum, deportation defense
- TAX: IRS disputes, tax planning, audits
- OTHER: matters that do not fit the above
URGENCY RULES:
- CRITICAL: imminent court deadline (<5 days), arrest/detention, TRO hearing, statute of limitations <30 days
- HIGH: significant financial exposure, pending filing deadline, client in distress
- MEDIUM: routine matter, no immediate deadline pressure
- LOW: general inquiry, no active dispute
JURISDICTION DETECTION:
Extract the jurisdiction from mentions of courts, state law references, venue descriptions,
or the client's location. Format as "State/Country - Court Level" when possible.
DEADLINE EXTRACTION:
Parse all dates mentioned. Convert relative dates ("next month", "two weeks from now")
to absolute dates based on today: {current_date}.
CONFLICT INDICATORS:
Note any mention of: specific attorneys at the firm, prior consultations with the firm,
parties that might match existing clients, or matters against firm clients.
{format_instructions}
Return ONLY the structured JSON object. Do not include any preamble, explanation, or markdown."""
return ChatPromptTemplate.from_messages([
("system", system_template),
("human", "{client_communication}")
])
async def extract(self, client_communication: str, firm_name: str = "Smith & Associates LLP") -> IntakeSummary:
prompt = self.build_prompt()
formatted_prompt = prompt.format_messages(
firm_name=firm_name,
current_date="2025-01-15", # In production: datetime.now().strftime("%Y-%m-%d")
format_instructions=self.parser.get_format_instructions(),
client_communication=client_communication
)
response = await self.llm.ainvoke(formatted_prompt)
return self.parser.parse(response.content)Code Example: Full Intake Agent with Conflict Checking
Now we assemble the complete intake agent. The key addition is the conflict-checking module, which embeds extracted party names and queries a vector database of existing firm clients. We use ChromaDB for simplicity, but in production you would point this at your firm's conflict database (iManage, NetDocuments, or a custom Postgres solution with pgvector).
# agents/intake_agent.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Optional
from chains.intake_extraction import IntakeExtractionChain
from models.intake_schema import IntakeSummary, MatterType, UrgencyLevel
import chromadb
from chromadb.utils import embedding_functions
from datetime import datetime, timedelta
import hashlib
import logging
logger = logging.getLogger(__name__)
# --- State Definition ---
class IntakeState(TypedDict):
"""State that flows through the intake agent graph."""
raw_input: str
source: str # e.g., 'email', 'web_form', 'phone_note'
intake_summary: Optional[IntakeSummary]
conflict_matches: List[dict]
routing_decision: Optional[str]
notification_sent: bool
error_message: Optional[str]
# --- Conflict Checker Module ---
class ConflictChecker:
"""Checks prospective client parties against existing firm clients."""
def __init__(self, persist_directory: str = "./conflict_db"):
self.client = chromadb.PersistentClient(path=persist_directory)
self.collection = self.client.get_or_create_collection(
name="firm_clients",
metadata={"description": "Existing firm clients for conflict checking"}
)
self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
api_key="your-openai-key", # In production: os.getenv("OPENAI_API_KEY")
model_name="text-embedding-3-small"
)
def add_client(self, client_name: str, metadata: dict):
"""Add a client to the conflict database."""
doc_id = hashlib.md5(client_name.encode()).hexdigest()
self.collection.add(
ids=[doc_id],
documents=[f"Client: {client_name} | Matter: {metadata.get('matter_type', 'N/A')} | Status: {metadata.get('status', 'active')}"],
metadatas=[metadata]
)
def check(self, party_names: List[str], threshold: float = 0.85) -> List[dict]:
"""Check if any party names match existing firm clients."""
if not party_names:
return []
results = []
for name in party_names:
query_results = self.collection.query(
query_texts=[name],
n_results=3,
include=["documents", "metadatas", "distances"]
)
for i, distance in enumerate(query_results.get('distances', [[]])[0]):
similarity = 1.0 - (distance / 2.0) # Normalize distance to similarity
if similarity >= threshold:
results.append({
"queried_party": name,
"matched_client": query_results['documents'][0][i],
"similarity_score": similarity,
"metadata": query_results['metadatas'][0][i]
})
return results
# --- Routing Logic ---
class IntakeRouter:
"""Routes intake to the appropriate practice group and attorney."""
PRACTICE_GROUPS = {
MatterType.PERSONAL_INJURY: "Litigation - Personal Injury",
MatterType.FAMILY_LAW: "Family Law Group",
MatterType.CORPORATE: "Corporate & Business Group",
MatterType.REAL_ESTATE: "Real Estate Group",
MatterType.EMPLOYMENT: "Employment Law Group",
MatterType.IP: "Intellectual Property Group",
MatterType.CRIMINAL_DEFENSE: "Criminal Defense Group",
MatterType.IMMIGRATION: "Immigration Practice",
MatterType.TAX: "Tax Practice Group",
MatterType.OTHER: "General Practice"
}
ATTORNEY_ASSIGNMENTS = {
# In production, this would query an availability/load-balancing system
MatterType.PERSONAL_INJURY: "Sarah Chen, Esq.",
MatterType.FAMILY_LAW: "Michael Torres, Esq.",
MatterType.CORPORATE: "David Kim, Esq.",
MatterType.IP: "Priya Patel, Esq.",
MatterType.EMPLOYMENT: "James Wilson, Esq.",
MatterType.CRIMINAL_DEFENSE: "Robert Martinez, Esq.",
MatterType.IMMIGRATION: "Angela Nguyen, Esq.",
MatterType.TAX: "Jonathan Lee, CPA, Esq.",
MatterType.REAL_ESTATE: "Emily Davis, Esq.",
MatterType.OTHER: "Intake Coordinator"
}
def route(self, summary: IntakeSummary, conflict_matches: List[dict]) -> str:
if conflict_matches:
return "CONFLICT_HOLD"
practice_group = self.PRACTICE_GROUPS.get(summary.matter_type, "General Practice")
attorney = self.ATTORNEY_ASSIGNMENTS.get(summary.matter_type, "Intake Coordinator")
if summary.urgency == UrgencyLevel.CRITICAL:
return f"URGENT: {attorney} ({practice_group}) - Immediate callback required"
return f"{attorney} ({practice_group})"
# --- The Full Agent Graph ---
class IntakeAgent:
"""Complete legal intake agent with classification, extraction, conflict check, and routing."""
def __init__(self):
self.extractor = IntakeExtractionChain()
self.conflict_checker = ConflictChecker()
self.router = IntakeRouter()
self.graph = self._build_graph()
def _build_graph(self) -> StateGraph:
workflow = StateGraph(IntakeState)
# Define nodes
workflow.add_node("extract", self._extract_node)
workflow.add_node("conflict_check", self._conflict_check_node)
workflow.add_node("route", self._route_node)
workflow.add_node("notify", self._notify_node)
workflow.add_node("error_handler", self._error_node)
# Define edges with conditional branching
workflow.set_entry_point("extract")
workflow.add_edge("extract", "conflict_check")
workflow.add_conditional_edges(
"conflict_check",
self._after_conflict_check,
{
"route": "route",
"conflict_hold": "notify", # Notify that conflicts exist
"error": "error_handler"
}
)
workflow.add_edge("route", "notify")
workflow.add_edge("notify", END)
workflow.add_edge("error_handler", END)
return workflow.compile()
async def _extract_node(self, state: IntakeState) -> IntakeState:
try:
summary = await self.extractor.extract(state["raw_input"])
state["intake_summary"] = summary
state["error_message"] = None
logger.info(f"Extracted matter type: {summary.matter_type}")
except Exception as e:
state["error_message"] = f"Extraction failed: {str(e)}"
logger.error(state["error_message"])
return state
async def _conflict_check_node(self, state: IntakeState) -> IntakeState:
if state.get("error_message"):
return state
summary = state.get("intake_summary")
if not summary:
state["error_message"] = "No intake summary available for conflict check"
return state
all_parties = [summary.primary_client_name] + summary.opposing_party_names
matches = self.conflict_checker.check(all_parties)
state["conflict_matches"] = matches
if matches:
logger.warning(f"Conflict detected: {len(matches)} matches found")
return state
def _after_conflict_check(self, state: IntakeState) -> str:
if state.get("error_message"):
return "error"
if state.get("conflict_matches"):
return "conflict_hold"
return "route"
async def _route_node(self, state: IntakeState) -> IntakeState:
summary = state.get("intake_summary")
if not summary:
state["error_message"] = "Cannot route without intake summary"
return state
routing = self.router.route(summary, state.get("conflict_matches", []))
state["routing_decision"] = routing
return state
async def _notify_node(self, state: IntakeState) -> IntakeState:
"""Send notification to the assigned attorney or conflict resolver."""
routing = state.get("routing_decision", "Unrouted")
conflicts = state.get("conflict_matches", [])
# In production: send email, update CRM, post to Slack/webhook
logger.info(f"Notification: Routing -> {routing}")
if conflicts:
logger.info(f"Conflict alert: {len(conflicts)} matches require manual review")
state["notification_sent"] = True
return state
async def _error_node(self, state: IntakeState) -> IntakeState:
logger.error(f"Intake agent error: {state.get('error_message')}")
# In production: alert ops team, log to monitoring system
return state
async def process(self, raw_input: str, source: str = "web_form") -> dict:
"""Run the full intake pipeline and return the result."""
initial_state: IntakeState = {
"raw_input": raw_input,
"source": source,
"intake_summary": None,
"conflict_matches": [],
"routing_decision": None,
"notification_sent": False,
"error_message": None
}
final_state = await self.graph.ainvoke(initial_state)
return {
"intake_summary": final_state.get("intake_summary"),
"conflict_matches": final_state.get("conflict_matches"),
"routing_decision": final_state.get("routing_decision"),
"notification_sent": final_state.get("notification_sent"),
"error": final_state.get("error_message")
}To test the intake agent end-to-end, you can run it against sample client communications:
# test_intake.py
import asyncio
from agents.intake_agent import IntakeAgent
async def main():
agent = IntakeAgent()
# Sample client email
sample_input = """
Subject: Need legal help after car accident - insurance denying claim
Hello, I was in a serious car accident on December 15th on I-5 near Seattle.
The other driver ran a red light and my car is totaled. I have severe back injuries.
State Farm is denying my claim and saying it was 50/50 fault.
I have medical bills over $50,000 and lost wages. The police report shows the
other driver was cited. I need to file a lawsuit quickly - I heard there's a
3-year statute of limitations in Washington but I want to act now.
Please call me at 555-0123. My name is Robert Chen.
"""
result = await agent.process(sample_input, source="email")
if result["error"]:
print(f"Error: {result['error']}")
else:
summary = result["intake_summary"]
print(f"Matter Type: {summary.matter_type}")
print(f"Urgency: {summary.urgency}")
print(f"Client: {summary.primary_client_name}")
print(f"Jurisdiction: {summary.jurisdiction}")
print(f"Key Facts: {summary.key_facts}")
print(f"Routing: {result['routing_decision']}")
if result["conflict_matches"]:
print(f"⚠️ CONFLICTS FOUND: {result['conflict_matches']}")
asyncio.run(main())Building an AI Research Agent
Legal Research Architecture
The research agent follows a retrieval-augmented generation pattern with a crucial twist: legal research requires authoritative sources, not probabilistic generation. The agent must retrieve actual case law, statutes, or regulations before synthesizing an answer, and it must provide pinpoint citations that can be verified. Our architecture uses a multi-stage pipeline: query analysis → dense retrieval → reranking → source verification → synthesis with citations.
The query analysis stage decomposes the user's natural language question into structured search parameters (jurisdiction, date range, legal topic, procedural posture). The retrieval stage performs hybrid search (dense embeddings + keyword BM25) across a vectorized legal corpus. Reranking uses a cross-encoder model to push the most legally significant cases to the top. Finally, the synthesis stage generates a structured memo with holdings, reasoning, and Shepard's-style treatment notes.
Implementing Semantic Search Over Case Law
We build a legal research agent that uses the Caselaw Access Project (CAP) dataset — a free corpus of ~6.5 million U.S. court decisions. For production use with proprietary databases, the same patterns apply to Westlaw or LexisNexis API integrations. We embed the case law corpus into a vector database and expose a RAG interface.
# research/legal_vector_store.py
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
import json
from typing import List, Optional
import os
class LegalVectorStore:
"""Vector store populated with case law for semantic search."""
def __init__(
self,
collection_name: str = "case_law_ninth_circuit",
persist_directory: str = "./legal_vector_db",
embedding_model: str = "text-embedding-3-large"
):
self.embeddings = OpenAIEmbeddings(
model=embedding_model,
dimensions=1536
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n\n", "\n", ". ", " ", ""],
keep_separator=True
)
self.vector_store = Chroma(
collection_name=collection_name,
embedding_function=self.embeddings,
persist_directory=persist_directory
)
def index_opinion(self, opinion_json: dict) -> int:
"""Index a single judicial opinion into the vector store."""
# Extract metadata
metadata = {
"case_name": opinion_json.get("name", "Unknown"),
"citation": opinion_json.get("citation", ""),
"court": opinion_json.get("court", ""),
"jurisdiction": opinion_json.get("jurisdiction", ""),
"date": opinion_json.get("decision_date", ""),
"judges": opinion_json.get("judges", ""),
"docket_number": opinion_json.get("docket_number", ""),
"attorneys": opinion_json.get("attorneys", "")
}
# The opinion text may include headnotes, syllabus, and the body
full_text = ""
for section in ["headnotes", "syllabus", "opinion_text"]:
if opinion_json.get(section):
full_text += f"\n\n{opinion_json[section]}"
# Split into chunks while preserving citation metadata on each chunk
chunks = self.text_splitter.split_text(full_text)
documents = [
Document(
page_content=chunk,
metadata={**metadata, "chunk_index": i}
)
for i, chunk in enumerate(chunks)
]
# Add to vector store
ids = self.vector_store.add_documents(documents)
return len(ids)
def index_bulk(self, opinions: List[dict]):
"""Index multiple opinions. In production, use batch processing."""
total = 0
for opinion in opinions:
try:
count = self.index_opinion(opinion)
total += count
except Exception as e:
print(f"Failed to index {opinion.get('name', 'Unknown')}: {e}")
return total
def semantic_search(
self,
query: str,
k: int = 20,
filter_court: Optional[str] = None,
filter_jurisdiction: Optional[str] = None,
date_range: Optional[tuple] = None # (start_date, end_date)
) -> List[Document]:
"""Perform semantic search with optional metadata filters."""
filter_dict = {}
if filter_court:
filter_dict["court"] = filter_court
if filter_jurisdiction:
filter_dict["jurisdiction"] = filter_jurisdiction
# Chroma doesn't support date range natively, so we do post-filtering
results = self.vector_store.similarity_search(
query,
k=k * 2, # Oversample to allow post-filtering
filter=filter_dict if filter_dict else None
)
# Post-filter by date range if specified
if date_range:
start, end = date_range
results = [
doc for doc in results
if doc.metadata.get("date", "") >= start
and doc.metadata.get("date", "") <= end
]
# Return top k after filtering
return results[:k]Code Example: Complete RAG Research Agent
Now we build the full research agent with query decomposition, retrieval, reranking, and memo generation. The agent uses a cross-encoder for reranking (critical for legal search where the top embedding hit may not be the most legally significant case) and generates a structured research memo with proper Bluebook citations.
# agents/research_agent.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Optional, Dict, Any
from research.legal_vector_store import LegalVectorStore
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import asyncio
import logging
logger = logging.getLogger(__name__)
# --- State ---
class ResearchState(TypedDict):
"""State flowing through the research agent."""
user_query: str
decomposed_queries: List[str]
jurisdiction_filter: Optional[str]
date_range: Optional[Dict[str, str]]
retrieved_documents: List[Any]
reranked_documents: List[Any]
verified_sources: List[Dict]
research_memo: Optional[str]
citations: List[str]
error_message: Optional[str]
# --- Query Analyzer ---
class QueryAnalyzer:
"""Decomposes natural language legal queries into structured search parameters."""
def __init__(self, model_name: str = "gpt-4o-mini"):
self.llm = ChatOpenAI(model=model_name, temperature=0.0)
async def analyze(self, query: str) -> Dict[str, Any]:
prompt = ChatPromptTemplate.from_messages([
("system", """You are a legal research query analyzer. Given a natural language
legal research question, extract structured parameters.
Return a JSON object with these fields:
{
"decomposed_queries": ["list", "of", "specific", "search", "queries"],
"jurisdiction_filter": "circuit or state name or null",
"date_range": {"start": "YYYY-MM-DD or null", "end": "YYYY-MM-DD or null"},
"legal_topics": ["list", "of", "key", "legal", "concepts"],
"procedural_posture": "appeal, trial, motion to dismiss, etc. or null",
"requested_treatment": "shepardize, distinguish, find analogous, or null"
}
Decompose the query into 3-5 specific search queries that would retrieve relevant cases.
For jurisdiction: extract mentions of federal circuits, states, districts, or courts.
For date range: if the user specifies "recent", use last 2 years. If "landmark", leave open.
If the user mentions a specific case name, add a query for that exact case."""),
("human", "{query}")
])
response = await self.llm.ainvoke(prompt.format_messages(query=query))
# Parse the JSON response
import json
try:
return json.loads(response.content)
except json.JSONDecodeError:
# Fallback: treat the whole query as a single search
return {
"decomposed_queries": [query],
"jurisdiction_filter": None,
"date_range": {"start": None, "end": None},
"legal_topics": [],
"procedural_posture": None,
"requested_treatment": None
}
# --- Reranker (Cross-Encoder) ---
class LegalReranker:
"""Reranks retrieved documents using a legal-specific cross-encoder."""
def __init__(self):
# In production, use a fine-tuned legal reranker or an API
# For demonstration, we use the LLM as a relevance scorer
self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
async def rerank(self, query: str, documents: List[Any], top_k: int = 10) -> List[Any]:
"""Score and rerank documents by legal relevance to the query."""
if len(documents) <= top_k:
return documents
# Batch scoring: evaluate documents in groups
scored_docs = []
batch_size = 5
for i in range(0, len(documents), batch_size):
batch = documents[i:i+batch_size]
# Create a scoring prompt
docs_text = "\n---\n".join([
f"Document {j}: {doc.page_content[:500]}...\nCitation: {doc.metadata.get('citation', 'N/A')}"
for j, doc in enumerate(batch)
])
prompt = ChatPromptTemplate.from_messages([
("system", """You are scoring legal documents for relevance to a research query.
Rate each document from 0 to 10 based on:
- Direct legal relevance to the query
- Authoritativeness (higher court, published opinion > unpublished)
- Factual similarity to the scenario described
- Precedential value (landmark cases, often-cited cases score higher)
Return scores as: Document 0: X, Document 1: Y, etc. Only return the scores."""),
("human", "Query: {query}\n\nDocuments:\n{docs_text}\n\nScores:")
])
response = await self.llm.ainvoke(prompt.format_messages(
query=query,
docs_text=docs_text
))
# Parse scores (simplified; production code would be more robust)
for j, doc in enumerate(batch):
try:
score_line = [line for line in response.content