Spaces:
Sleeping
Sleeping
File size: 3,363 Bytes
1fee1c2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | # COST: Gemini gemini-2.5-flash called ONCE per document on first summarization.
# Result is written to a JSON sidecar file (<filename>.summary.json) beside
# the source document. Every subsequent call reads the sidecar and returns
# immediately — zero additional tokens spent.
# Sidecar is invalidated only by manual deletion.
import json
import os
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from dotenv import load_dotenv
load_dotenv()
from ingestion import parse_document
SUMMARY_SUFFIX = ".summary.json"
MAX_SUMMARY_CHARS = 12_000 # truncation guard before sending to LLM
def _cache_path(file_path: Path) -> Path:
return file_path.with_name(file_path.name + SUMMARY_SUFFIX)
def summarize_document(file_path: str) -> str:
"""Return a concise summary of a document, using a local cache when available.
On the first call the function extracts the full text from the document,
sends it to the OpenAI Chat Completions API (gpt-4o-mini), and writes the
resulting summary to a JSON sidecar file next to the original document
(e.g. 'report.pdf.summary.json'). Every subsequent call for the same file
reads the sidecar and returns immediately without making another LLM call,
keeping costs low and responses fast.
The cache is invalidated only by deleting the sidecar file manually.
Args:
file_path: Absolute or relative path to the .pdf or .docx document.
Returns:
A plain-text summary paragraph. If the file cannot be found or parsed,
an error message is returned instead of raising so the agent can handle
it gracefully.
"""
path = Path(file_path)
if not path.exists():
return f"Error: file not found at '{file_path}'."
cache = _cache_path(path)
if cache.exists():
try:
data = json.loads(cache.read_text(encoding="utf-8"))
return data["summary"]
except (json.JSONDecodeError, KeyError):
cache.unlink(missing_ok=True) # corrupt cache — regenerate
try:
pages = parse_document(path)
except ValueError as exc:
return f"Error: {exc}"
full_text = "\n\n".join(p["text"] for p in pages)
truncated = full_text[:MAX_SUMMARY_CHARS]
if len(full_text) > MAX_SUMMARY_CHARS:
truncated += "\n\n[...document truncated for summarisation...]"
from openai import OpenAI
client = OpenAI(
api_key=os.environ["GEMINI_API_KEY"],
base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
)
response = client.chat.completions.create(
model="gemini-2.5-flash",
messages=[
{
"role": "system",
"content": (
"You are a document summarisation assistant. "
"Write a clear, concise summary (3-5 paragraphs) of the document "
"the user provides. Focus on key topics, findings, and conclusions."
),
},
{"role": "user", "content": f"Summarise this document:\n\n{truncated}"},
],
)
summary = response.choices[0].message.content or ""
cache.write_text(
json.dumps({"source_file": path.name, "summary": summary}, ensure_ascii=False, indent=2),
encoding="utf-8",
)
return summary
|