Application Shape

This lesson builds a complete, small end-to-end application: a command-line tool that ingests one or more PDFs into a vector store and answers questions about them with citations. It draws together file upload (Lesson 3), request construction (Lesson 4), and citation handling — while noting where this differs from the input_file approach from Unit 7.

Unit 7 showed sending a PDF's contents directly in a request via input_file, which works well for a single document read once. This lesson's approach — indexing PDFs into a vector store — is the right choice instead when you have multiple documents, want to ask many questions across a session without re-sending the full document every time, or need the document to persist across separate requests and even separate users.

Step 1: Ingesting the PDFs

from openai import OpenAI

client = OpenAI()


def create_knowledge_base(pdf_paths, store_name):
    store = client.vector_stores.create(name=store_name)

    for path in pdf_paths:
        with open(path, "rb") as f:
            result = client.vector_stores.files.upload_and_poll(
                vector_store_id=store.id,
                file=f,
            )
        status_label = "OK" if result.status == "completed" else result.status
        print(f"{path}: {status_label}")

    return store.id


store_id = create_knowledge_base(
    pdf_paths=["contract-2026.pdf", "amendment-a.pdf", "amendment-b.pdf"],
    store_name="contract-review-session",
)
print("Vector store ready:", store_id)

This function creates one fresh vector store for the session and uploads each given PDF into it, printing a simple status line per file so ingestion problems are visible immediately rather than discovered later. Creating a dedicated store per "session" or per "case" (as opposed to reusing one giant shared store) is a reasonable pattern here specifically because this application's documents — a contract and its amendments — form a self-contained unit that doesn't need to be searched alongside unrelated material; it also makes cleanup trivial once the review is done.

Step 2: Asking Questions With Citations

def ask_question(store_id, question):
    response = client.responses.create(
        model="gpt-5.6-terra",
        instructions=(
            "You are a document analysis assistant. Answer strictly using "
            "information found in the provided documents via file search. "
            "If the documents do not contain enough information to answer "
            "confidently, say so explicitly instead of guessing."
        ),
        input=question,
        tools=[{"type": "file_search", "vector_store_ids": [store_id]}],
    )

    citations = []
    for item in response.output:
        if item.type == "message":
            for block in item.content:
                for annotation in getattr(block, "annotations", []):
                    filename = getattr(annotation, "filename", "unknown file")
                    citations.append(filename)

    return response.output_text, citations


answer, sources = ask_question(store_id, "What is the termination notice period?")
print("Answer:", answer)
print("Sources:", sorted(set(sources)))

Note: The exact item.type values and annotation attribute names (filename, and any others available) are specific to the current API version — verify these against official documentation before relying on the exact field names in production.

ask_question sends the user's question along with strict grounding instructions, then walks the structured response.output to pull out every citation attached to the answer text. Deduplicating with sorted(set(citations)) produces a clean list of which source documents actually contributed to the answer — useful both for building a "Sources" section in a UI and for the kind of evidence-checking covered in Lesson 8. Separating the "get the answer" and "extract the citations" concerns into one function that returns both, rather than just printing the text, makes this logic reusable in a web backend, a CLI, or a test, none of which is possible if the function only prints.

Step 3: A Minimal Interactive Loop

def run_qa_session(store_id):
    print("Ask questions about the loaded documents. Type 'quit' to exit.")
    while True:
        question = input("\n> ").strip()
        if question.lower() in {"quit", "exit"}:
            break
        if not question:
            continue

        answer, sources = ask_question(store_id, question)
        print(answer)
        if sources:
            print("Sources:", ", ".join(sorted(set(sources))))
        else:
            print("Sources: none identified")


if __name__ == "__main__":
    run_qa_session(store_id)

This wraps ask_question in a simple read-input, call, print-output loop — the smallest useful interface for testing a document Q&A system interactively during development. The if __name__ == "__main__": guard is standard Python practice: it ensures run_qa_session only executes when this file is run directly as a script, not when it's imported as a module elsewhere (for example, by the test functions shown next, or by a future web framework wrapping this same logic).

Handling Multi-Document Questions

A realistic scenario for this application is a question that spans the base contract and one of its amendments — for example, "does Amendment A change the termination notice period from the original contract?" Because both documents live in the same vector store, a single file_search call can retrieve relevant chunks from both, and the model is responsible for reconciling them:

answer, sources = ask_question(
    store_id,
    "Does Amendment A change the termination notice period defined in the original contract?",
)
print(answer)
print("Sources:", sorted(set(sources)))

This is functionally identical code to the single-document case — the same ask_question function — which is precisely the benefit of indexing all related documents into one vector store rather than writing separate logic per document. The model can retrieve and reason across chunks from both the contract and the amendment in a single request, citing whichever document actually contains the relevant clause. If the amendment doesn't address termination notice at all, the well-grounded instructions from Step 2 should produce an answer noting the original contract's terms apply and that the amendment is silent on that point — which is exactly the kind of "confirm absence of evidence" behavior Lesson 8 examines in more depth.

Building in Session Cleanup

Because this application creates a dedicated vector store per session, it should also support tearing that store down when the session ends, both to avoid unbounded storage growth and, in shared or multi-tenant applications, to avoid leaving sensitive document content indexed indefinitely:

def cleanup_knowledge_base(store_id):
    client.vector_stores.delete(store_id)
    print(f"Deleted vector store {store_id}")

Note: The exact deletion method and whether deleting a vector store also deletes the underlying uploaded files (versus only removing them from the store) are version-specific behaviors — confirm current semantics against official documentation, especially before relying on deletion for data-retention or compliance requirements.

Calling this function at the end of a review session (or on a scheduled cleanup job for sessions older than some retention window) keeps storage usage proportional to active work rather than growing indefinitely. In a real production system handling sensitive documents like contracts, this kind of explicit lifecycle management is not optional — it's a data-retention requirement, and it's much easier to get right if you build it into the application from the start rather than retrofitting it after documents have already accumulated for months.

Testing the Application Logic

The parts of this application worth unit testing are the pure logic — citation extraction and deduplication — not the live API call itself:

class FakeAnnotation:
    def __init__(self, filename):
        self.filename = filename


class FakeContentBlock:
    def __init__(self, annotations):
        self.annotations = annotations


class FakeMessageItem:
    def __init__(self, annotations_per_block):
        self.type = "message"
        self.content = [FakeContentBlock(a) for a in annotations_per_block]


class FakeResponse:
    def __init__(self, output_text, output):
        self.output_text = output_text
        self.output = output


def extract_citations(response):
    citations = []
    for item in response.output:
        if item.type == "message":
            for block in item.content:
                for annotation in getattr(block, "annotations", []):
                    citations.append(getattr(annotation, "filename", "unknown file"))
    return citations


def test_extract_citations_deduplicates_across_blocks():
    fake_response = FakeResponse(
        output_text="The notice period is 30 days per the amendment.",
        output=[
            FakeMessageItem([
                [FakeAnnotation("contract-2026.pdf")],
                [FakeAnnotation("amendment-a.pdf"), FakeAnnotation("amendment-a.pdf")],
            ]),
        ],
    )

    citations = extract_citations(fake_response)

    assert sorted(set(citations)) == ["amendment-a.pdf", "contract-2026.pdf"]
    print("PASS: extract_citations pulls filenames from every content block")


test_extract_citations_deduplicates_across_blocks()

This test builds a small hierarchy of fake objects mirroring the real response shape closely enough to exercise extract_citations — a standalone version of the citation logic from ask_question — without any network call. It confirms citations are correctly gathered across multiple content blocks and that duplicate filenames collapse to one entry when deduplicated. Structuring citation extraction as its own testable function, separate from the API call that produces the response, is what makes this kind of fast, reliable test possible.

Common Mistakes

Sending the full PDF text as part of input on every question instead of relying on file_search, which reintroduces the exact context-window and cost problems RAG exists to solve — once documents are indexed in a vector store, let retrieval pull only the relevant chunks per question.

Never surfacing citations to the end user, which makes it impossible for anyone to verify an answer against the source document — always extract and display which files (and ideally which sections) contributed to an answer, especially for documents like contracts where accuracy has real consequences.

Leaving session-scoped vector stores around indefinitely, accumulating storage costs and retaining potentially sensitive documents longer than necessary — build cleanup into the application's lifecycle from the start.

Best Practices

Group genuinely related documents (a contract and its amendments) into one vector store so cross-document questions can be answered in a single retrieval call, rather than forcing the model to reason without seeing related documents.

Always pair a document Q&A assistant with explicit grounding instructions and citation extraction, since the entire value of this kind of application over a general-purpose chat model is that its answers are traceable back to specific source text.

Treat session or case-scoped vector stores as ephemeral resources with an explicit lifecycle — create them deliberately, and delete them deliberately once their purpose is served, rather than letting them accumulate as an afterthought.