Tutorial: Bikin AI Agent yang Baca PDF dan Kasih Summary

pip install PyPDF2 pdfplumber langchain langchain-openai langchain-chroma openai tiktoken chromadb fastapi uvicorn

import PyPDF2
from typing import List

def extract_text_from_pdf(pdf_path: str) -> str:
    text = ""
    with open(pdf_path, 'rb') as file:
        reader = PyPDF2.PdfReader(file)
        print(f"Total halaman: {len(reader.pages)}")
        for page in reader.pages:
            text += page.extract_text() + "\n"
    return text

import pdfplumber
from typing import List

def extract_text_with_pdfplumber(pdf_path: str) -> str:
    """Extract text lebih akurat, termasuk tabel."""
    full_text = []
    with pdfplumber.open(pdf_path) as pdf:
        for i, page in enumerate(pdf.pages):
            # Extract text biasa
            text = page.extract_text()
            if text:
                full_text.append(f"--- Halaman {i+1} ---\n{text}")

            # Bonus: extract tabel juga
            tables = page.extract_tables()
            for table in tables:
                table_text = "\n".join(
                    [" | ".join(str(cell or "") for cell in row) for row in table]
                )
                full_text.append(f"[Tabel]:\n{table_text}")

    return "\n\n".join(full_text)

def chunk_text(text: str, chunk_size: int = 3000, overlap: int = 200) -> List[str]:
    """Chunking dengan overlap untuk menjaga konteks antar bagian."""
    words = text.split()
    chunks = []
    current_chunk = []
    current_size = 0

    for word in words:
        current_chunk.append(word)
        current_size += len(word) + 1

        if current_size >= chunk_size:
            chunks.append(' '.join(current_chunk))
            # Overlap: simpen beberapa kata terakhir buat konteks
            current_chunk = current_chunk[-overlap//5:] if overlap > 0 else []
            current_size = sum(len(w) + 1 for w in current_chunk)

    if current_chunk:
        chunks.append(' '.join(current_chunk))

    print(f"Total chunks: {len(chunks)}, rata-rata {chunk_size} kata per chunk")
    return chunks

import chromadb
from chromadb.utils import embedding_functions
import os

def setup_vector_db(persist_dir: str = "./chroma_db"):
    """Setup ChromaDB client dengan OpenAI embeddings."""
    client = chromadb.PersistentClient(path=persist_dir)

    # Gunakan OpenAI embeddings (atau ganti ke local model biar hemat API)
    openai_ef = embedding_functions.OpenAIEmbeddingFunction(
        api_key=os.getenv("OPENAI_API_KEY"),
        model_name="text-embedding-3-small"
    )

    # Buat atau ambil collection yang sudah ada
    collection = client.get_or_create_collection(
        name="pdf_documents",
        embedding_function=openai_ef,
        metadata={"hnsw:space": "cosine"}
    )

    return client, collection

def store_chunks_in_db(collection, chunks: List[str], pdf_name: str):
    """Simpen chunks ke vector database."""
    ids = [f"{pdf_name}_chunk_{i}" for i in range(len(chunks))]

    # Batch insert untuk efisiensi (ChromaDB max 5461 per batch)
    batch_size = 500
    for i in range(0, len(chunks), batch_size):
        batch_chunks = chunks[i:i+batch_size]
        batch_ids = ids[i:i+batch_size]
        collection.add(
            documents=batch_chunks,
            ids=batch_ids,
            metadatas=[{"source": pdf_name}] * len(batch_chunks)
        )

    print(f"Stored {len(chunks)} chunks dari '{pdf_name}' ke database")

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser

def search_similar_chunks(collection, query: str, n_results: int = 5) -> List[str]:
    """Cari chunks yang paling relevan dengan query."""
    results = collection.query(
        query_texts=[query],
        n_results=n_results
    )
    return results["documents"][0]

def build_rag_chain(collection):
    """Build RAG chain menggunakan LangChain."""
    llm = ChatOpenAI(
        model="gpt-4o",
        temperature=0.2,
        max_tokens=2000
    )

    def retrieve_context(query: str) -> str:
        chunks = search_similar_chunks(collection, query)
        return "\n\n---\n\n".join(chunks)

    prompt = ChatPromptTemplate.from_template("""
    Kamu adalah assistant ahli yang membantu menganalisis dokumen PDF.

    Gunakan context berikut untuk menjawab pertanyaan user.
    Jika jawaban tidak ada di context, katakan "Maaf, informasi ini
    tidak ditemukan di dokumen."
    Selalu cantumkan referensi dari mana informasi diambil.

    Context:
    {context}

    Pertanyaan: {question}

    Jawaban (dalam Bahasa Indonesia yang jelas dan terstruktur):
    """)

    def rag_query(question: str) -> str:
        context = retrieve_context(question)
        chain = prompt | llm | StrOutputParser()
        return chain.invoke({"context": context, "question": question})

    return rag_query

from openai import OpenAI
import os

client = OpenAI()

def summarize_chunk(chunk: str, custom_prompt: str = None) -> str:
    """Summarize satu chunk menggunakan OpenAI."""
    if not custom_prompt:
        custom_prompt = (
            "Kamu adalah expert summarizer. Ringkas teks berikut "
            "dalam Bahasa Indonesia. Fokus pada: poin-poin utama, "
            "argumen/klaim, data penting, dan kesimpulan. "
            "Gunakan format bullet point yang rapi dan mudah dibaca. "
            "Jangan menambahkan informasi yang tidak ada di teks asli."
        )

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": custom_prompt},
            {"role": "user", "content": chunk}
        ],
        temperature=0.2
    )

    return response.choices[0].message.content

from pathlib import Path

def process_pdf(pdf_path: str, mode: str = "summarize") -> str:
    """
    Main agent function.
    mode: 'summarize' untuk summary, 'index' untuk simpen ke vector DB,
          'ask' untuk tanya ke PDF.
    """
    pdf_name = Path(pdf_path).stem
    print(f"📄 Processing: {pdf_path}")

    text = extract_text_with_pdfplumber(pdf_path)

    if not text.strip():
        return "❌ Error: Tidak bisa extract text dari PDF."

    chunks = chunk_text(text)

    if mode == "index":
        _, collection = setup_vector_db()
        store_chunks_in_db(collection, chunks, pdf_name)
        return f"✅ '{pdf_name}' berhasil di-index! {len(chunks)} chunks tersimpan."

    elif mode == "ask":
        _, collection = setup_vector_db()
        question = input("Masukkan pertanyaan kamu: ")
        rag_chain = build_rag_chain(collection)
        return rag_chain(question)

    else:  # summarize
        summaries = []
        for i, chunk in enumerate(chunks):
            print(f"📝 Summarizing chunk {i+1}/{len(chunks)}...")
            summary = summarize_chunk(chunk)
            summaries.append(summary)

        combined = "\n\n".join(summaries)

        if len(summaries) > 1:
            print("🔄 Generating final combined summary...")
            final_prompt = (
                "Gabungkan semua ringkasan berikut jadi satu summary "
                "komprehensif. Format: heading untuk topik utama, "
                "bullet point untuk detail. Bahasa: Indonesia. "
                "Panjang: 500-800 kata"
            )
            final = summarize_chunk(combined, final_prompt)
        else:
            final = summaries[0]

        return final


# === Cara Pakai ===
if __name__ == "__main__":
    import sys

    pdf_file = sys.argv[1] if len(sys.argv) > 1 else "sample.pdf"
    mode = sys.argv[2] if len(sys.argv) > 2 else "summarize"

    result = process_pdf(pdf_file, mode)
    print("\n" + "=" * 60)
    print(result)
    print("=" * 60)

# ❌ Bad
"Summarize this text"

# ✅ Good
"Kamu adalah analis riset senior. Ringkas dokumen akademik berikut
dengan fokus pada metodologi, hasil utama, dan limitasi studi."

prompt = """Analisis dokumen ini dan jawab dalam format:

## Ringkasan Eksekutif
[2-3 kalimat]

## Poin-Poin Utama
- [bullet point]
- [bullet point]

## Data/Kutipan Penting
[angka, statistik, atau quote relevan]

## Kesimpulan
[1 paragraf]
"""

# Tambahkan instruksi edge case handling
"""
Jika informasi tidak ditemukan di dokumen, jawab:
"Informasi ini tidak tersedia di dokumen yang diberikan."
Jangan pernah mengarang atau hallucinate jawaban.
"""

# app.py - FastAPI server
from fastapi import FastAPI, UploadFile, File
from fastapi.responses import JSONResponse
import tempfile
import os

app = FastAPI(title="PDF AI Agent API")

@app.post("/summarize")
async def summarize_endpoint(file: UploadFile = File(...)):
    """Upload PDF dan dapatkan summary."""
    with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp:
        content = await file.read()
        tmp.write(content)
        tmp_path = tmp.name

    try:
        result = process_pdf(tmp_path, mode="summarize")
        return {"status": "success", "summary": result}
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})
    finally:
        os.unlink(tmp_path)

@app.post("/ask")
async def ask_endpoint(file: UploadFile = File(...), question: str = ""):
    """Upload PDF dan tanya sesuatu."""
    with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp:
        content = await file.read()
        tmp.write(content)
        tmp_path = tmp.name

    try:
        process_pdf(tmp_path, mode="index")
        _, collection = setup_vector_db()
        rag_chain = build_rag_chain(collection)
        answer = rag_chain(question)
        return {"status": "success", "answer": answer}
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})
    finally:
        os.unlink(tmp_path)

# Jalankan: uvicorn app:app --host 0.0.0.0 --port 8000

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
EXPOSE 8000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

# Deploy dengan Docker Compose (production-ready)
docker compose up -d --build