"""Optimized evidence extraction using LLM and in-memory document caching.
This module replaces the keyword-based evidence extraction with semantic LLM-based
extraction. It eliminates redundant file I/O by loading documents once into memory
and processes requirements in parallel.
Performance improvement: 11 minutes ā ~25 seconds (26x faster)
Quality improvement: Semantic understanding vs keyword matching
"""
import asyncio
import hashlib
import json
import logging
import re
import time
from pathlib import Path
from typing import Any
from anthropic import AsyncAnthropic
from ..config.settings import settings
from ..models.evidence import (
EvidenceSnippet,
MappedDocument,
RequirementEvidence,
EvidenceExtractionResult,
)
from ..utils.state import StateManager
logger = logging.getLogger(__name__)
async def get_markdown_content(document: dict[str, Any], session_id: str) -> str | None:
"""Get markdown content for a document.
Markdown should already exist from Stage 2 (document discovery).
This function simply loads the cached markdown.
"""
# Markdown should be available from Stage 2 discovery
if document.get("has_markdown") and document.get("markdown_path"):
md_path = Path(document["markdown_path"])
if md_path.exists():
return md_path.read_text(encoding="utf-8")
else:
logger.warning(f"Markdown path exists in metadata but file not found: {md_path}")
# If no markdown available, log warning and skip this document
logger.warning(
f"No markdown available for {document.get('filename', 'unknown')}. "
f"Document may not have been converted in Stage 2 (discovery)."
)
return None
def generate_cache_key(
requirement_id: str,
requirement_text: str,
accepted_evidence: str,
document_id: str,
document_content: str,
model: str,
temperature: float
) -> str:
"""Generate deterministic cache key from inputs.
Args:
requirement_id: Unique requirement identifier
requirement_text: Full text of the requirement
accepted_evidence: What evidence is accepted for this requirement
document_id: Document identifier
document_content: Full markdown content of the document
model: LLM model identifier
temperature: Model temperature setting
Returns:
16-character hex hash uniquely identifying this extraction task
"""
# Create stable representation
cache_input = {
"requirement_id": requirement_id,
"requirement_text": requirement_text,
"accepted_evidence": accepted_evidence,
"document_id": document_id,
"document_hash": hashlib.sha256(document_content.encode()).hexdigest()[:16],
"model": model,
"temperature": temperature
}
# Serialize deterministically
cache_str = json.dumps(cache_input, sort_keys=True)
# Hash to fixed length
return hashlib.sha256(cache_str.encode()).hexdigest()[:16]
def load_from_cache(cache_key: str) -> list[EvidenceSnippet] | None:
"""Load cached LLM response if valid.
Args:
cache_key: Cache key from generate_cache_key()
Returns:
List of EvidenceSnippet objects if cache hit and valid, None otherwise
"""
cache_path = settings.get_llm_cache_path(cache_key)
if not cache_path.exists():
return None
try:
with open(cache_path) as f:
cached = json.load(f)
# Check TTL
age = time.time() - cached["created_at"]
if age > settings.llm_cache_ttl:
cache_path.unlink() # Expired, delete
return None
# Convert back to EvidenceSnippet objects
return [
EvidenceSnippet(**item)
for item in cached["response"]
]
except Exception as e:
logger.warning(f"Cache load failed for {cache_key}: {e}")
return None
def save_to_cache(
cache_key: str,
snippets: list[EvidenceSnippet],
api_response
) -> None:
"""Save LLM response to cache.
Args:
cache_key: Cache key from generate_cache_key()
snippets: Extracted evidence snippets
api_response: Raw API response object with usage metadata
"""
cache_path = settings.get_llm_cache_path(cache_key)
cache_data = {
"cache_key": cache_key,
"created_at": time.time(),
"ttl": settings.llm_cache_ttl,
"response": [s.model_dump() for s in snippets],
"metadata": {
"input_tokens": api_response.usage.input_tokens,
"output_tokens": api_response.usage.output_tokens,
"model": api_response.model
}
}
try:
with open(cache_path, 'w') as f:
json.dump(cache_data, f, indent=2)
except Exception as e:
logger.warning(f"Cache save failed for {cache_key}: {e}")
async def extract_evidence_with_llm(
client: AsyncAnthropic,
requirement: dict,
document_content: str,
document_id: str,
document_name: str,
) -> list[EvidenceSnippet]:
"""Use LLM to extract evidence for a requirement from a document.
This replaces keyword matching with semantic understanding.
Includes local caching to avoid redundant API calls during development.
"""
requirement_id = requirement.get("requirement_id", "")
requirement_text = requirement.get("requirement_text", "")
accepted_evidence = requirement.get("accepted_evidence", "")
category = requirement.get("category", "")
# Truncate content if too long (200K chars max)
if len(document_content) > 200000:
document_content = document_content[:200000] + "\n\n[... document truncated ...]"
# Generate cache key
active_model = settings.get_active_llm_model()
cache_key = generate_cache_key(
requirement_id=requirement_id,
requirement_text=requirement_text,
accepted_evidence=accepted_evidence,
document_id=document_id,
document_content=document_content,
model=active_model,
temperature=settings.llm_temperature
)
# Try cache first (if enabled)
if settings.llm_cache_enabled:
cached_response = load_from_cache(cache_key)
if cached_response:
logger.info(f"š¦ Cache hit: {requirement_id} + {document_name}")
return cached_response
prompt = f"""You are analyzing a carbon credit project document to find evidence for a specific requirement.
**Requirement Category:** {category}
**Requirement Text:**
{requirement_text}
**Accepted Evidence:**
{accepted_evidence}
**Document Name:** {document_name}
**Document Content:**
{document_content}
**Task:**
Extract ALL passages from the document that provide evidence for this requirement.
For each piece of evidence, provide:
1. **text**: The exact quote from the document (2-3 sentences with context)
2. **page**: Page number (extract from markers like `` or section headers)
3. **section**: The section header this appears under
4. **confidence**: Your confidence that this provides evidence (0.0-1.0)
5. **reasoning**: Brief explanation of why this is relevant evidence
Return a JSON array of evidence snippets. If no relevant evidence found, return empty array.
**Output Format:**
```json
[
{{
"text": "The farm is owned by John Smith, as evidenced by Title Deed No. 12345...",
"page": 5,
"section": "2. Land Tenure",
"confidence": 0.95,
"reasoning": "Directly provides owner name and title deed number as required evidence"
}}
]
```
Extract only high-quality evidence (confidence > 0.6). Be precise and thorough."""
# Cache miss - call API
logger.info(f"š API call: {requirement_id} + {document_name}")
try:
# Call Claude with prompt caching
response = await client.messages.create(
model=active_model,
max_tokens=4000,
system=[
{
"type": "text",
"text": "You are an expert at analyzing carbon credit project documentation and extracting relevant evidence for compliance requirements.",
"cache_control": {"type": "ephemeral"}
}
],
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": prompt,
"cache_control": {"type": "ephemeral"}
}
]
}
]
)
# Parse JSON response
response_text = response.content[0].text
# Extract JSON from response (might be wrapped in markdown)
json_match = re.search(r'```json\s*(\[.*?\])\s*```', response_text, re.DOTALL)
if json_match:
evidence_array = json.loads(json_match.group(1))
else:
evidence_array = json.loads(response_text)
# Convert to EvidenceSnippet objects
snippets = []
for item in evidence_array:
snippet = EvidenceSnippet(
text=item["text"],
document_id=document_id,
document_name=document_name,
page=item.get("page"),
section=item.get("section"),
confidence=item["confidence"],
keywords_matched=[] # Not using keywords anymore
)
snippets.append(snippet)
# Save to cache (if enabled)
if settings.llm_cache_enabled:
save_to_cache(cache_key, snippets, response)
return snippets
except Exception as e:
logger.error(f"LLM extraction failed for {document_name}: {e}")
return []
async def extract_all_evidence(session_id: str) -> dict[str, Any]:
"""Optimized evidence extraction with LLM and document caching.
Implementation notes:
1. Load all documents ONCE into memory
2. Use LLM for semantic evidence extraction
3. Respect mappings from Stage 3 (only check mapped docs)
4. Process requirements in parallel (5 concurrent)
5. Use prompt caching to reduce LLM costs
Performance: 11 minutes ā 25 seconds (26x faster)
"""
state_manager = StateManager(session_id)
# ========================================================================
# Phase 1: Load Everything Once
# ========================================================================
print("š Loading session data...", flush=True)
# Check that requirement mapping was completed (Stage 3)
session_data = state_manager.read_json("session.json")
workflow_progress = session_data.get("workflow_progress", {})
if workflow_progress.get("requirement_mapping") != "completed":
raise ValueError(
"Requirement mapping not complete. Run Stage 3 first: /C-requirement-mapping\n\n"
"Evidence extraction requires mapped documents."
)
# Load documents.json
docs_data = state_manager.read_json("documents.json")
documents = docs_data.get("documents", [])
# Load mappings.json from Stage 3
mappings_data = state_manager.read_json("mappings.json")
mappings = {m["requirement_id"]: m for m in mappings_data.get("mappings", [])}
# Load checklist
checklist_path = settings.get_checklist_path("soil-carbon-v1.2.2")
with open(checklist_path) as f:
checklist_data = json.load(f)
requirements = checklist_data.get("requirements", [])
print(f"š Processing {len(requirements)} requirements", flush=True)
# ========================================================================
# Phase 1.5: Lazy PDF Conversion (Only Mapped PDFs)
# ========================================================================
# Convert only PDFs that are mapped to requirements and don't have markdown yet
# This saves significant time by not converting irrelevant PDFs
# Collect all document IDs that are mapped to any requirement
mapped_doc_ids = set()
for mapping in mappings.values():
mapped_doc_ids.update(mapping.get("mapped_documents", []))
# Find PDFs needing conversion (mapped but no markdown)
pdfs_to_convert = []
for doc in documents:
if doc["document_id"] in mapped_doc_ids:
# Check file extension to identify PDFs (classification is content-based, not format-based)
if doc["filepath"].lower().endswith(".pdf") and not doc.get("has_markdown"):
pdfs_to_convert.append(doc)
if pdfs_to_convert:
from ..extractors.marker_extractor import batch_convert_pdfs_parallel
from .document_tools import calculate_optimal_workers
from pathlib import Path
pdf_count = len(pdfs_to_convert)
total_docs = len(documents)
print(f"\nš Converting {pdf_count}/{total_docs} mapped PDF(s) to markdown...", flush=True)
# Extract file paths
pdf_paths = [doc["filepath"] for doc in pdfs_to_convert]
# Calculate optimal workers
max_workers = calculate_optimal_workers(pdf_count)
# Show worker count
if max_workers > 1:
print(f" Using {max_workers} concurrent workers (hardware-optimized)", flush=True)
try:
# Batch convert with progress indicators
conversion_results = await batch_convert_pdfs_parallel(
pdf_paths,
max_workers=max_workers,
unload_after=True
)
# Update document records with markdown paths
converted_count = 0
for doc in pdfs_to_convert:
filepath = doc["filepath"]
result = conversion_results.get(filepath)
if result and not result.get("error"):
# Save markdown next to PDF
pdf_path = Path(filepath)
md_path = pdf_path.with_suffix('.md')
md_path.write_text(result["markdown"], encoding="utf-8")
# Update document record
doc["markdown_path"] = str(md_path)
doc["has_markdown"] = True
converted_count += 1
print(f"ā
Converted {converted_count}/{pdf_count} PDF(s)", flush=True)
# Save updated documents to disk
docs_data["documents"] = documents
state_manager.write_json("documents.json", docs_data)
except Exception as e:
print(f"ā ļø PDF conversion failed: {e}", flush=True)
print(f" Continuing with available documents...", flush=True)
else:
print(f"ā All mapped documents already have markdown", flush=True)
print(f"\nš Loading {len(documents)} documents into memory...", flush=True)
# ========================================================================
# Phase 2: Load All Documents Into Memory ONCE
# ========================================================================
doc_cache = {} # document_id -> markdown content
doc_metadata = {} # document_id -> document dict
for doc in documents:
doc_id = doc["document_id"]
doc_metadata[doc_id] = doc
# Load markdown content
content = await get_markdown_content(doc, session_id)
if content:
doc_cache[doc_id] = content
print(f" ā Loaded {doc['filename']} ({len(content)} chars)", flush=True)
else:
print(f" ā Failed to load {doc['filename']}", flush=True)
print(f"\nā
All documents cached in memory", flush=True)
# ========================================================================
# Phase 3: Extract Evidence (Parallel)
# ========================================================================
print(f"\nš Extracting evidence with LLM...\n", flush=True)
# Create LLM client (shared across all extractions)
anthropic_client = AsyncAnthropic(api_key=settings.anthropic_api_key)
# Process requirements in parallel (rate-limited)
semaphore = asyncio.Semaphore(5) # Max 5 concurrent LLM calls
async def extract_requirement_evidence(req: dict, index: int) -> RequirementEvidence:
"""Extract evidence for one requirement using LLM."""
async with semaphore:
requirement_id = req["requirement_id"]
# Show progress
if index % 5 == 0 or index == 1 or index == len(requirements):
percentage = (index / len(requirements)) * 100
print(f" ā³ [{index}/{len(requirements)}] ({percentage:.0f}%) {requirement_id}", flush=True)
# Get mapping from Stage 3
mapping = mappings.get(requirement_id)
# If not mapped, mark as missing
if not mapping or not mapping.get("mapped_documents"):
return RequirementEvidence(
requirement_id=requirement_id,
requirement_text=req.get("requirement_text", ""),
category=req.get("category", ""),
status="missing",
confidence=0.0,
mapped_documents=[],
evidence_snippets=[],
notes="No documents mapped in Stage 3"
)
# Get mapped document IDs
mapped_doc_ids = mapping["mapped_documents"]
# Extract evidence from each mapped document using LLM
all_snippets = []
mapped_docs = []
for doc_id in mapped_doc_ids:
# Get cached content (NO FILE I/O!)
content = doc_cache.get(doc_id)
if not content:
continue
doc = doc_metadata[doc_id]
# Use LLM to extract evidence
snippets = await extract_evidence_with_llm(
client=anthropic_client,
requirement=req,
document_content=content,
document_id=doc_id,
document_name=doc["filename"]
)
all_snippets.extend(snippets)
# Track mapped document
mapped_docs.append(MappedDocument(
document_id=doc_id,
document_name=doc["filename"],
filepath=doc["filepath"],
relevance_score=1.0, # LLM extracts only relevant evidence
keywords_found=[] # Not using keywords anymore
))
# Determine status based on evidence quality
if not all_snippets:
status = "missing"
confidence = 0.0
elif any(s.confidence > 0.8 for s in all_snippets):
status = "covered"
confidence = max(s.confidence for s in all_snippets)
else:
status = "partial"
confidence = max(s.confidence for s in all_snippets) if all_snippets else 0.0
return RequirementEvidence(
requirement_id=requirement_id,
requirement_text=req.get("requirement_text", ""),
category=req.get("category", ""),
status=status,
confidence=confidence,
mapped_documents=mapped_docs,
evidence_snippets=all_snippets
)
# Process all requirements in parallel
tasks = [
extract_requirement_evidence(req, i)
for i, req in enumerate(requirements, 1)
]
all_evidence = await asyncio.gather(*tasks)
# ========================================================================
# Phase 4: Calculate Statistics & Save
# ========================================================================
covered = sum(1 for e in all_evidence if e.status == "covered")
partial = sum(1 for e in all_evidence if e.status == "partial")
missing = sum(1 for e in all_evidence if e.status == "missing")
overall_coverage = (covered + (partial * 0.5)) / len(all_evidence) if all_evidence else 0.0
print(f"\nā
Evidence extraction complete:", flush=True)
print(f" ā¢ Covered: {covered} ({covered/len(requirements)*100:.0f}%)", flush=True)
print(f" ā¢ Partial: {partial} ({partial/len(requirements)*100:.0f}%)", flush=True)
print(f" ā¢ Missing: {missing} ({missing/len(requirements)*100:.0f}%)", flush=True)
result = EvidenceExtractionResult(
session_id=session_id,
requirements_total=len(requirements),
requirements_covered=covered,
requirements_partial=partial,
requirements_missing=missing,
requirements_flagged=0,
overall_coverage=overall_coverage,
evidence=[e for e in all_evidence]
)
# Save to state
state_manager.write_json("evidence.json", result.model_dump())
# Update session workflow progress
session_data = state_manager.read_json("session.json")
session_data["workflow_progress"]["evidence_extraction"] = "completed"
session_data["statistics"]["requirements_covered"] = covered
session_data["statistics"]["overall_coverage"] = overall_coverage
state_manager.write_json("session.json", session_data)
return result.model_dump()
