What "Production" Adds Beyond a Working Prototype

Every previous lesson in this unit built pieces that work correctly in isolation: creating stores, uploading documents, filtering by metadata, detecting missing evidence, combining tools. A production assistant is the same pieces assembled with attention to concerns that don't show up in a quick script — configuration management, error handling that doesn't crash the whole request, observability into what happened on every call, and a clear boundary between the reusable core logic and whatever interface (web API, chat UI, Slack bot) sits on top of it. This lesson assembles those pieces into a single cohesive module.

Designing the Core Assistant Class

from dataclasses import dataclass, field
from openai import OpenAI


FALLBACK_PHRASE = "I don't have enough information in the knowledge base to answer that."

GROUNDED_INSTRUCTIONS = (
    "You are a knowledge base assistant. Answer only using information "
    "retrieved via file search from the provided documents. If the "
    "retrieved documents do not contain enough information to answer "
    f"confidently, respond with exactly: \"{FALLBACK_PHRASE}\" "
    "Cite the source document for every factual claim you make."
)


@dataclass
class AssistantConfig:
    model: str
    vector_store_ids: list
    max_retries: int = 2


@dataclass
class AssistantAnswer:
    text: str
    sources: list = field(default_factory=list)
    evidence_found: bool = True
    error: str = None


class KnowledgeBaseAssistant:
    def __init__(self, client, config: AssistantConfig):
        self.client = client
        self.config = config

    def ask(self, question, extra_filter=None):
        tool_config = {
            "type": "file_search",
            "vector_store_ids": self.config.vector_store_ids,
        }
        if extra_filter is not None:
            tool_config["filters"] = extra_filter

        last_error = None
        for attempt in range(self.config.max_retries + 1):
            try:
                response = self.client.responses.create(
                    model=self.config.model,
                    instructions=GROUNDED_INSTRUCTIONS,
                    input=question,
                    tools=[tool_config],
                )
                return self._parse_response(response)
            except Exception as exc:
                last_error = str(exc)

        return AssistantAnswer(text="", sources=[], evidence_found=False, error=last_error)

    def _parse_response(self, response):
        sources = []
        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")
                        sources.append(filename)

        evidence_found = FALLBACK_PHRASE not in response.output_text

        return AssistantAnswer(
            text=response.output_text,
            sources=sorted(set(sources)),
            evidence_found=evidence_found,
        )

Note: The exact exception types raised by the SDK on network or API errors, and the precise item.type and annotation field names used in _parse_response, are version-specific — confirm both against current OpenAI documentation before relying on this exact structure in production.

This class brings together several lessons at once. AssistantConfig centralizes everything that varies between deployments — which model, which vector stores, how many retries — as data rather than scattered literals throughout the code, which is what lets you run, say, a staging configuration pointed at a test vector store and a production configuration pointed at the real one, using the exact same KnowledgeBaseAssistant code. AssistantAnswer gives every call a consistent, structured return value instead of a bare string, carrying the answer text, extracted sources (from Lesson 6), whether evidence was actually found (from Lesson 8), and an error field for when things go wrong — this uniform shape is what makes the assistant safe to call from a web handler, since the caller never has to guess what shape of value came back.

The retry loop in ask handles a category of failure this unit hasn't discussed yet but that matters in any live system: transient network or API errors that have nothing to do with retrieval quality. Retrying a small, bounded number of times before giving up, and returning a structured error rather than letting an exception propagate uncaught, is what separates code you'd run in a script from code you'd put behind a real endpoint that other systems depend on.

Wrapping the Assistant for a Web Endpoint

def handle_question_request(assistant, user_id, question, get_user_tier_fn):
    if not question or not question.strip():
        return {"error": "question must not be empty"}, 400

    tier = get_user_tier_fn(user_id)
    filter_for_tier = {"type": "eq", "key": "product_tier", "value": tier}

    answer = assistant.ask(question.strip(), extra_filter=filter_for_tier)

    if answer.error is not None:
        return {"error": "internal error, please try again"}, 502

    return {
        "answer": answer.text,
        "sources": answer.sources,
        "evidence_found": answer.evidence_found,
    }, 200

This function represents the boundary between the reusable KnowledgeBaseAssistant core and a specific delivery mechanism — here modeled as a plain function returning a body and status code, the same shape you'd adapt to whatever web framework you actually use. Several production concerns are visible here: input validation happens before any API call is made (an empty question shouldn't cost you a request); the metadata filter (Lesson 5) is derived from get_user_tier_fn(user_id) — a trusted, server-side lookup — never from anything the client directly supplied, which is the access-boundary discipline Lesson 5 emphasized; and an internal error is translated into a generic client-facing message and a 502 status rather than leaking exception details or stack traces to the caller.

Observability: Logging Every Call's Outcome

A production assistant should produce a structured log entry for every question it answers, capturing enough detail to debug quality problems after the fact without having to reproduce them live:

import json
import time


def log_interaction(user_id, question, answer, duration_seconds):
    log_entry = {
        "timestamp": time.time(),
        "user_id": user_id,
        "question": question,
        "evidence_found": answer.evidence_found,
        "sources": answer.sources,
        "error": answer.error,
        "duration_seconds": round(duration_seconds, 3),
    }
    print(json.dumps(log_entry))


def ask_and_log(assistant, user_id, question, extra_filter=None):
    start = time.time()
    answer = assistant.ask(question, extra_filter=extra_filter)
    duration = time.time() - start
    log_interaction(user_id, question, answer, duration)
    return answer

log_interaction writes a single structured JSON line per interaction (in a real deployment this would go to a logging system rather than print, but the structure is what matters). Recording evidence_found and sources on every call, not just the ones that failed, is what turns Lesson 8's failure-detection logic into an ongoing metric: aggregating this log over a week tells you what fraction of real questions hit evidence_found: False, which questions those were, and therefore exactly what content is missing from the knowledge base — direct, data-driven input into the document preparation work from Lesson 7, rather than guesswork about what to add next.

Testing the Assembled Assistant

The value of separating KnowledgeBaseAssistant from any real network call is that its logic — response parsing, error handling, retry counting — can be fully tested with a fake client, following the same dependency-injection pattern used throughout this unit:

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


class FakeContentBlock:
    def __init__(self, filenames):
        self.annotations = [FakeAnnotation(f) for f in filenames]


class FakeMessageItem:
    def __init__(self, filenames):
        self.type = "message"
        self.content = [FakeContentBlock(filenames)]


class FakeResponse:
    def __init__(self, output_text, filenames):
        self.output_text = output_text
        self.output = [FakeMessageItem(filenames)]


class FakeResponsesAPI:
    def __init__(self, response_or_exception):
        self.response_or_exception = response_or_exception

    def create(self, **kwargs):
        if isinstance(self.response_or_exception, Exception):
            raise self.response_or_exception
        return self.response_or_exception


class FakeClient:
    def __init__(self, response_or_exception):
        self.responses = FakeResponsesAPI(response_or_exception)


def test_ask_returns_sources_on_success():
    fake_response = FakeResponse("The notice period is 30 days.", ["policy.pdf"])
    client = FakeClient(fake_response)
    assistant = KnowledgeBaseAssistant(client, AssistantConfig(model="gpt-5.6-terra", vector_store_ids=["vs_1"]))

    answer = assistant.ask("What is the notice period?")

    assert answer.evidence_found is True
    assert answer.sources == ["policy.pdf"]
    assert answer.error is None
    print("PASS: successful response is parsed with sources and no error")


def test_ask_detects_fallback_phrase():
    fake_response = FakeResponse(FALLBACK_PHRASE, [])
    client = FakeClient(fake_response)
    assistant = KnowledgeBaseAssistant(client, AssistantConfig(model="gpt-5.6-terra", vector_store_ids=["vs_1"]))

    answer = assistant.ask("What is our policy on time travel reimbursement?")

    assert answer.evidence_found is False
    print("PASS: fallback phrase is correctly detected as missing evidence")


def test_ask_returns_error_after_exhausting_retries():
    client = FakeClient(RuntimeError("simulated network failure"))
    assistant = KnowledgeBaseAssistant(
        client, AssistantConfig(model="gpt-5.6-terra", vector_store_ids=["vs_1"], max_retries=1)
    )

    answer = assistant.ask("What is the notice period?")

    assert answer.error == "simulated network failure"
    assert answer.evidence_found is False
    print("PASS: repeated failures return a structured error instead of raising")


test_ask_returns_sources_on_success()
test_ask_detects_fallback_phrase()
test_ask_returns_error_after_exhausting_retries()

These three tests exercise the assistant's three key behaviors without ever calling the real OpenAI API: a normal successful answer with citations, correct detection of the fallback phrase indicating missing evidence, and graceful handling of an API that always fails, confirming the retry loop gives up after the configured number of attempts and returns a structured error rather than propagating an exception. FakeResponsesAPI.create simulates both the successful and failing case depending on what it's constructed with, letting each test set up exactly the scenario it needs. This test suite is what gives you confidence to refactor or extend KnowledgeBaseAssistant later — adding a new tool, changing the retry strategy — without manually re-verifying every behavior against the live API each time.

Assembling the Pieces

A complete production setup, pulling this lesson's pieces together with a config value for which vector stores to use, looks like this:

client = OpenAI()

config = AssistantConfig(
    model="gpt-5.6-terra",
    vector_store_ids=["vs_product_docs_55c1"],
    max_retries=2,
)

assistant = KnowledgeBaseAssistant(client, config)


def get_user_tier(user_id):
    # In a real application, this looks up the authenticated user's plan
    # from your own user database — never trust a client-supplied value here.
    return "pro"


result, status_code = handle_question_request(
    assistant=assistant,
    user_id="user_4471",
    question="Does the Pro plan include single sign-on?",
    get_user_tier_fn=get_user_tier,
)

print(status_code, result)

This final wiring shows the full shape of a production knowledge-base assistant: a configuration object, a core assistant class with retry and structured-error handling, a request-handling boundary that enforces access control and input validation, and — in the pieces just before this one — logging and tests around the whole thing. Every individual technique here traces back to an earlier lesson in this unit; what makes it "production" is that these techniques are combined deliberately and consistently, rather than any single new trick.

Common Mistakes

Letting exceptions from the API client propagate uncaught into a web handler, which turns a transient network blip into a full request failure (or worse, an unhandled server error) instead of a clean, retried, or gracefully degraded response.

Deriving access-control filter values from client-supplied input rather than a trusted server-side lookup, reopening the exact security gap discussed in Lesson 5 the moment metadata filtering is wired into a real endpoint.

Building the assistant's core logic so tightly coupled to a specific web framework or interface that it can't be tested without spinning up a server — keep the core logic (like KnowledgeBaseAssistant) framework-agnostic and testable in isolation, with the framework-specific code only in a thin wrapper.

Best Practices

Centralize configuration in one place (a dataclass, environment variables, or a config file) rather than scattering model names, store IDs, and retry counts as literals throughout the codebase, so promoting a change from staging to production is a configuration change, not a code change.

Return a consistent, structured result type from every core operation, including an explicit error field and evidence-found flag, so calling code never has to guess what shape of value it received or whether an answer was actually grounded.

Log evidence status and sources for every interaction, not just failures, and review that log regularly — it is the most reliable source of truth for what your knowledge base is missing and where retrieval quality needs attention.