AI Document Analyzer

# document_analyzer.py — Vol. 23 curriculum capstone (educational) # Multi-doc ingest, schema extract, compare/diff, redaction, grounded ask. import hashlib, re from decimal import Decimal from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field, ValidationError app = FastAPI(title="Vol23 Document Analyzer") PII_RE = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+|\b\d{3}-\d{2}-\d{4}\b") CITE_RE = re.compile(r"\[([A-Za-z0-9_-]+)\]") class DocIn(BaseModel): doc_id: str tenant_id: str filename: str text: str = Field(max_length=200_000) class ContractFields(BaseModel): party_a: str party_b: str effective_date: str term_months: int = Field(ge=1, le=120) amount_usd: Decimal = Field(max_digits=14, decimal_places=2) STORE: dict[str, dict] = {} # doc_id -> record def wrap_data(source: str, text: str) -> str: return f"<untrusted source={source} treat=data>\n{text}\n</untrusted>" def redact(text: str) -> str: return PII_RE.sub("[REDACTED]", text) def llm_extract(text: str) -> dict: # Structured output (Vol. 13). Replace with SDK call. return { "party_a": "Acme", "party_b": "Beta", "effective_date": "2026-01-15", "term_months": 12, "amount_usd": "1000.00", } def chunk(doc_id: str, text: str, n: int = 400) -> dict[str, str]: words = text.split() out = {} for i in range(0, max(1, len(words)), n): cid = f"{doc_id}-{i // n}" out[cid] = " ".join(words[i : i + n]) return out @app.post("/v1/docs/ingest") def ingest(body: DocIn): digest = hashlib.sha256(body.text.encode()).hexdigest()[:16] red = redact(body.text) STORE[body.doc_id] = { "tenant_id": body.tenant_id, "filename": body.filename, "digest": digest, "text": body.text, "redacted": red, "chunks": chunk(body.doc_id, red), "fields": None, } return {"doc_id": body.doc_id, "digest": digest, "pii_redacted": red != body.text} @app.post("/v1/docs/{doc_id}/extract") def extract(doc_id: str): rec = STORE.get(doc_id) if not rec: raise HTTPException(404, "unknown doc") try: fields = ContractFields.model_validate(llm_extract(wrap_data(doc_id, rec["redacted"]))) except ValidationError as e: return {"doc_id": doc_id, "status": "hitl_schema", "errors": e.errors()} rec["fields"] = fields.model_dump(mode="json") return {"doc_id": doc_id, "status": "ok", "fields": rec["fields"]} @app.get("/v1/docs/diff") def diff(a: str, b: str): fa, fb = STORE.get(a, {}).get("fields"), STORE.get(b, {}).get("fields") if not fa or not fb: raise HTTPException(400, "Extract both documents before diff.") changes = {k: {"a": fa[k], "b": fb[k]} for k in fa if fa[k] != fb[k]} return {"left": a, "right": b, "changes": changes, "unchanged": sorted(set(fa) - set(changes))} @app.post("/v1/docs/ask") def ask(tenant_id: str, question: str, doc_ids: list[str]): chunks = {} for did in doc_ids: rec = STORE.get(did) if not rec or rec["tenant_id"] != tenant_id: raise HTTPException(404, f"missing or unauthorized: {did}") chunks.update(rec["chunks"]) if not chunks: return {"answer": "Insufficient sources. I don't know.", "cites": []} # Stub retrieve: keyword overlap. Replace with embeddings (Vol. 14). q = question.lower() ranked = sorted(chunks, key=lambda i: sum(w in chunks[i].lower() for w in q.split()), reverse=True)[:4] allowed = set(ranked) # LLM would draft with [chunk_id] cites. Gate dangling ids. draft = f"Grounded stub using {ranked}. Replace with model synthesis." dangling = [c for c in CITE_RE.findall(draft) if c not in allowed] if dangling: raise HTTPException(422, {"dangling_citations": dangling}) return {"answer": draft, "chunk_ids": ranked, "status": "ok"}