Building a Data-Analysis Assistant
From Individual Calls to a Reusable Component
Every previous lesson in this unit demonstrated one request at a time: upload a file, ask a question, extract a result. A real application needs these pieces wired together behind a stable interface — something a web endpoint, a CLI tool, or a chat UI can call repeatedly across a user's session without re-deriving the upload, container-reuse, and extraction logic on every call site. This lesson builds that component: a small DataAnalysisAssistant class that owns a session's uploaded files, its container, and the conversation chain, and exposes a simple ask() method.
Designing the Interface First
Before writing the implementation, it helps to decide what the class should look like from the outside, since that shapes every decision inside it:
assistant = DataAnalysisAssistant()
assistant.upload_dataset("quarterly_sales.csv")
answer = assistant.ask("What were total sales by region?")
answer2 = assistant.ask("Now break that down by month for the top region.")
charts = assistant.get_generated_files()
This interface hides three things a caller should not have to think about on every use: which files are currently attached, which container and response ID the conversation is currently chained to, and how to extract generated files from the raw response. Each ask() call should feel like a single, stateless-looking function call from the outside, while internally maintaining the state needed for the follow-up question to correctly reuse the loaded data (the pattern from Lesson 2 and Lesson 4).
Implementation
from openai import OpenAI
class DataAnalysisAssistant:
"""Wraps code interpreter to provide a simple, stateful interface for
multi-turn data analysis over one or more uploaded datasets."""
def __init__(self, client: OpenAI | None = None, model: str = "gpt-5.6-terra"):
self.client = client or OpenAI()
self.model = model
self._file_ids: list[str] = []
self._previous_response_id: str | None = None
self._last_response = None
def upload_dataset(self, path: str) -> str:
uploaded = self.client.files.create(file=open(path, "rb"), purpose="assistants")
self._file_ids.append(uploaded.id)
return uploaded.id
def ask(self, question: str) -> str:
tool_config = {"type": "code_interpreter", "container": {"type": "auto"}}
if self._file_ids and self._previous_response_id is None:
tool_config["container"]["file_ids"] = self._file_ids
kwargs = {
"model": self.model,
"tools": [tool_config],
"input": question,
}
if self._previous_response_id:
kwargs["previous_response_id"] = self._previous_response_id
response = self.client.responses.create(**kwargs)
self._previous_response_id = response.id
self._last_response = response
return response.output_text
def get_generated_files(self) -> list[dict]:
if self._last_response is None:
return []
return extract_generated_files(self._last_response, self.client)
def cleanup(self) -> None:
for file_id in self._file_ids:
self.client.files.delete(file_id)
self._file_ids = []
Walking Through the Design Decisions
File IDs are only attached on the first call. The condition if self._file_ids and self._previous_response_id is None deliberately attaches file_ids to the container only when there is no prior response to chain from — that is, on the very first question. Every subsequent ask() call omits file_ids because the files are already loaded into the container from the first call, and previous_response_id carries that container's state forward, exactly as demonstrated in Lesson 2. Attaching the same file IDs again on every call would be redundant at best and, depending on platform behavior, could unnecessarily trigger a fresh load of the file rather than reusing the already-parsed dataframe.
previous_response_id is only added to kwargs when it exists. This lets ask() work correctly on both the very first call (no prior response yet) and every call after, using the same method body — a common and useful pattern for wrapping a conversational API where the first turn and later turns need slightly different arguments but should share one code path.
get_generated_files() reuses the extract_generated_files helper from Lesson 6 rather than reimplementing extraction logic inside the class. This is deliberate: extraction is a pure function of a response object, and keeping it as a standalone function (rather than a method tightly coupled to this class) means it can be tested and reused independently — which is exactly what the test below does.
cleanup() is a separate, explicit method rather than automatic. Deleting uploaded files the moment an ask() call finishes would break the entire point of the class, since follow-up questions need those files to still exist for container recreation scenarios. Cleanup belongs at the end of a session's lifetime, called explicitly by whatever code owns the assistant's lifecycle — a request handler's finally block, a context manager, or an explicit "end session" action in a UI.
Using It as a Context Manager
For code that wants a guarantee that cleanup always happens, wrapping the class with context-manager support is a natural extension:
class ManagedDataAnalysisAssistant(DataAnalysisAssistant):
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
self.cleanup()
return False
with ManagedDataAnalysisAssistant() as assistant:
assistant.upload_dataset("quarterly_sales.csv")
print(assistant.ask("What is the total revenue across all regions?"))
__exit__ returning False means any exception raised inside the with block is not suppressed — cleanup happens, but the caller still finds out something went wrong, rather than the error being silently swallowed. This is the correct default for cleanup logic: clean up resources unconditionally, but never hide a real failure from the caller by accident.
Testing the Assistant Without Live API Calls
Following this course's dependency-injection testing pattern, the assistant is tested by injecting a fake client rather than calling the real API:
class FakeFilesResource:
def __init__(self):
self.created = []
self.deleted = []
def create(self, file, purpose):
file_id = f"file_{len(self.created)}"
self.created.append(file_id)
class FakeFile:
id = file_id
return FakeFile()
def delete(self, file_id):
self.deleted.append(file_id)
class FakeResponsesResource:
def __init__(self):
self.calls = []
def create(self, **kwargs):
self.calls.append(kwargs)
class FakeResponse:
id = f"resp_{len(self.calls)}"
output_text = f"answer to: {kwargs['input']}"
output = []
return FakeResponse()
class FakeClient:
def __init__(self):
self.files = FakeFilesResource()
self.responses = FakeResponsesResource()
def test_assistant_attaches_files_only_on_first_call():
fake_client = FakeClient()
assistant = DataAnalysisAssistant(client=fake_client)
assistant.upload_dataset("dummy_path.csv")
first_answer = assistant.ask("first question")
second_answer = assistant.ask("second question")
assert first_answer == "answer to: first question"
assert second_answer == "answer to: second question"
first_call, second_call = fake_client.responses.calls
assert "file_ids" in first_call["tools"][0]["container"]
assert "file_ids" not in second_call["tools"][0]["container"]
assert "previous_response_id" not in first_call
assert second_call["previous_response_id"] == "resp_1"
print("PASS: file_ids attached only on first call, chaining used afterward")
def test_cleanup_deletes_all_uploaded_files():
fake_client = FakeClient()
assistant = DataAnalysisAssistant(client=fake_client)
assistant.upload_dataset("a.csv")
assistant.upload_dataset("b.csv")
assistant.cleanup()
assert fake_client.files.deleted == ["file_0", "file_1"]
assert assistant._file_ids == []
print("PASS: cleanup deletes every uploaded file and clears local state")
test_assistant_attaches_files_only_on_first_call()
test_cleanup_deletes_all_uploaded_files()
FakeFilesResource and FakeResponsesResource record every call they receive instead of hitting a real API, which lets the tests assert on exactly how DataAnalysisAssistant uses its client — specifically, that file_ids appears only on the first responses.create call and that previous_response_id correctly chains the second call to the first response's ID. This verifies the class's internal logic precisely, without needing network access, an API key, or any nondeterminism from an actual model response — the fake output_text is deterministic and directly reflects the input it was given, making the assertions exact rather than approximate.
Note that upload_dataset("dummy_path.csv") works against the fake client without the file needing to actually exist on disk, because FakeFilesResource.create never calls open() itself — a small but important detail: the real DataAnalysisAssistant.upload_dataset does call open(path, "rb") before handing it to self.client.files.create, so a fully faithful unit test would need either a real temporary file or a slightly different injection point. In practice, most teams solve this by injecting an already-open file-like object or by using a temporary file created within the test itself, keeping the fake client focused purely on faking the network boundary.
Common Mistakes
Building a new DataAnalysisAssistant instance per question instead of per session. This defeats the entire purpose of the class — a fresh instance has no previous_response_id and no memory of uploaded files, so every "follow-up" question would actually re-upload files and start a brand-new, unrelated conversation.
Forgetting to call cleanup() at the end of a session, leaking uploaded files exactly as described in Lesson 3. Wrapping session-ending code paths (including error paths) in a finally block or a context manager, as shown above, prevents this.
Testing this class only through the real API during development. This makes tests slow, costly, and flaky (subject to model nondeterminism and network conditions). The fake-client pattern above should be the default way this logic is verified during day-to-day development, with real API calls reserved for periodic integration testing.
Best Practices
Keep the assistant class thin and delegate reusable logic (like file extraction) to standalone functions, so those functions can be tested and reused independently of the class that happens to call them.
Expose an explicit cleanup() method and pair it with context-manager support so that both explicit and automatic resource management styles are available to callers, matching how they structure the rest of their application.
Design the fake objects used in tests to record what they were called with, not just to return canned data. Asserting on the actual arguments passed to responses.create (as the tests above do) verifies the class's real behavior — correct container and chaining logic — rather than merely confirming it doesn't crash.