A team builds a pipeline to summarize 5,000 support documents with an LLM. They test it on 20 files, and it works. They start the full run on Friday evening and go home.
On Monday morning, they open the terminal. The script stopped at document 3,412.
The LLM provider had returned a 429 Too Many Requests for a few seconds on Friday night. The code didn't expect it, so Python raised an error, and the whole script stopped. Worse, the summaries were being collected in a list and written to a file at the end, so the 3,411 finished summaries were gone too.
Nothing was wrong with the model. Nothing was wrong with the prompt. The problem was missing error handling and careless file handling, in a few lines of very ordinary Python.
This is the second post in our AI & LLM series at CreoWis. In the first post, we looked at how to work with large data without running out of memory. This time, we'll look at two things that decide whether your AI app survives the real world: how it handles errors and how it handles files.
Why AI Engineers Should Care About This
Because AI code talks to a lot of things you don't control:
LLM APIs that rate-limit you, time out, or go down
Files from clients that are missing, empty, or in a strange encoding
Vector databases and other services over the network
Users who send input you never imagined
In a notebook, an error is just a red message. You fix it and run the cell again. In production, an unhandled error stops your app, loses work, and sometimes nobody notices until a client complains.
There's one more reason. A lot of code today is written with AI coding assistants, and they often leave out error handling. If you can't spot a missing try/except or a missing timeout, you can't catch that gap in review.
What an Exception Actually Is
Let's start small.
score = 85
total = 0
print(score / total)
print("Hello") # never runs
ZeroDivisionError: division by zero
That error message has two parts:
ZeroDivisionError is the exception type. It's a class name.
division by zero is the exception message.
Different mistakes raise different types. 85 / "0" raises a TypeError. Opening a missing file raises a .
FileNotFoundError
And notice that print("Hello") never ran. Without handling, any exception stops the whole script right there.
try / except: Surviving the Error
try:
print(score / total)
except ZeroDivisionError:
print("Can't divide by zero.")
print("Hello") # still runs
The code that might fail goes in try. If it raises an error, Python jumps to except. If it doesn't, the except block is skipped. Either way, the script keeps going.
Catch specific exceptions, and order them correctly
You can have several except blocks:
try:
result = score / total
except ZeroDivisionError:
result = 0
except TypeError:
result = None
except Exception:
result = None # catch-all for anything else
Python checks the except blocks from top to bottom and uses the first one that matches.
Exception is the parent of almost all built-in errors. So if you put except Exception first, it catches everything, and your specific handlers never run. The rule is simple: most specific first, most general last.
Avoid a bare except:
You'll often see this:
try:
do_something()
except:
pass
A bare except: catches everything, including KeyboardInterrupt (when you press Ctrl+C) and SystemExit. That can make a script impossible to stop. And pass hides the error completely, so you'll never know what went wrong.
If you really need a catch-all, use except Exception:, and at least log the error.
Getting the actual message with as
try:
print(85 / 0)
except ZeroDivisionError as error:
print(f"Something went wrong: {error}")
# Something went wrong: division by zero
error here is the exception object. You can name it anything. Printing it gives you the real message, which is what you want in your logs.
You can also catch several types in one line:
except (ValueError, TypeError) as error:
...
else and finally: The Full Structure
There are two more optional blocks:
try:
result = do_something() # risky code
except SomeError as error:
handle(error) # runs only if try raised SomeError
else:
use(result) # runs only if try raised NOTHING
finally:
cleanup() # ALWAYS runs
Block
When it runs
Typical use
try
Always attempted
The code that might fail
except
Only when a matching exception happens
Recover, use a fallback, log, or re-raise
else
Only when no exception happens
Code that depends on success
finally
Always, even after a return
Cleanup: closing files, connections
A common mix-up: else does not run when there's an error. That's what except is for. else is the "everything went fine" block.
One small warning: don't put a return inside finally. It overrides any return value, and it even swallows exceptions from the other blocks.
A quick tip: write the logic first
People often ask, "How do I know which exception to catch?"
You usually don't, at first. Write the code withouttry/except, run it with real and bad inputs, and see which errors actually show up. Then add handlers for those. If you wrap everything in try/except from the start, you'll hide the very errors you need to see.
Custom Exceptions: Giving Your Errors a Name
Python's built-in exceptions cover Python problems. But many failures in your app aren't Python problems. They're business rules.
For example: "every site must have at least 2 inspections a day". Python doesn't know about that rule. You do.
class TooFewInspections(Exception):
pass
def validate(inspection_count):
if inspection_count < 2:
raise TooFewInspections("At least 2 inspections are required per day.")
try:
validate(1)
except TooFewInspections as error:
print(error)
# At least 2 inspections are required per day.
Two things to notice:
A custom exception is just a class that inherits from Exception. An empty body (pass) is enough.
raise means "throw this error now".
So who creates exceptions? ZeroDivisionError was written by Python's developers. HTTPError was written by the requests library's developers. Your custom exceptions are written by you, for the failures that only exist in your app.
Why not just print() a message instead? Because printing only shows text in a terminal that nobody may be watching. Raising an exception lets the calling code decide what to do: retry, show a friendly message in the UI, log it, or send an alert.
The Real Use Case: A Robust LLM API Call
Now let's go back to our Friday-night pipeline. Here's how most people first write an LLM call:
It works in the demo. But it has three hidden problems:
No timeout. If the provider hangs, your program can wait forever.
No status check. A 429 or 500 response still gets passed to .json() as if it were a normal answer.
No error handling. Any network problem crashes the whole run.
Let's fix it step by step.
Step 1: Define your own error types
class LLMProviderError(Exception):
"""Base error for anything that goes wrong with the LLM provider."""
class RateLimitError(LLMProviderError):
pass
class ProviderUnavailableError(LLMProviderError):
pass
class BadRequestError(LLMProviderError):
pass
This is a small hierarchy. Code that only cares "did the LLM call fail?" can catch LLMProviderError. Code that wants to react differently to a rate limit can catch RateLimitError on its own.
Step 2: Make the call safe
import requests
def call_llm_provider(endpoint, api_key, prompt):
headers = {
"Authorization": f"Bearer {api_key}", # who we are
"Content-Type": "application/json", # what we're sending
}
payload = {"prompt": prompt}
try:
response = requests.post(
endpoint,
headers=headers,
json=payload,
timeout=(3, 30), # 3 s to connect, 30 s to wait for data
)
response.raise_for_status() # raises HTTPError for 4xx and 5xx
return response.json()
except requests.exceptions.Timeout as error:
raise ProviderUnavailableError("The LLM provider took too long to respond.") from error
except requests.exceptions.ConnectionError as error:
raise ProviderUnavailableError("Could not connect to the LLM provider.") from error
except requests.exceptions.HTTPError as error:
status_code = error.response.status_code
if status_code == 429:
raise RateLimitError("The LLM provider rate limit was reached.") from error
if status_code >= 500:
raise ProviderUnavailableError("The LLM provider returned a server error.") from error
raise BadRequestError(f"The LLM provider rejected the request ({status_code}).") from error
Let's walk through the important pieces.
timeout=(3, 30) means: wait at most 3 seconds to connect, and at most 30 seconds between pieces of data from the server. It's not "33 seconds total". A slow, streaming response can take longer overall. But without any timeout, one dead server can freeze your whole pipeline.
raise_for_status() does nothing for a successful (2xx) response. For a 4xx or 5xx response, it raises an HTTPError. One line replaces a manual if response.status_code ... check.
Different status codes mean different things:
429 means you're sending too many requests. This is very common on free API tiers. Waiting and trying again usually works.
5xx means the provider has a problem. Trying again later may work.
Other 4xx (like 401 for a wrong API key, or 400 for a bad request) means your request is wrong. Trying again won't help.
raise ... from error keeps the original error attached as the cause. When you look at the traceback, you'll see both your clear message and the original HTTP error underneath. That's very useful when debugging at 2 a.m.
Step 3: Retry only the errors that deserve it
Now that our errors have names, retrying is easy:
import time
def call_with_retry(endpoint, api_key, prompt, retries=3):
for attempt in range(retries):
try:
return call_llm_provider(endpoint, api_key, prompt)
except (RateLimitError, ProviderUnavailableError):
if attempt == retries - 1:
raise # out of attempts, let the caller handle it
time.sleep(2 ** attempt) # wait 1 s, then 2 s, then 4 s...
Notice what's not caught here: BadRequestError. If your API key is wrong, retrying three times just wastes time. That error goes straight back to the caller.
Waiting a little longer after each failure is called exponential backoff. It gives a busy provider room to recover instead of hitting it again immediately.
Working With Files: The Problem With open() and close()
Now let's look at the second half of the Friday-night problem: files.
The basic way to read a file looks like this:
f = open("report.txt", "r")
content = f.read()
f.close()
open() is a built-in function. It takes a path and a mode, and gives you a file object. While the file is open, it holds a resource from the operating system, so you must close it.
But what if an error happens between open() and close()?
f = open("report.txt", "r")
content = f.read()
result = process(content) # this raises an error...
f.close() # ...so this never runs
The file stays open. Do that inside a loop over thousands of documents, and you'll eventually hit a "too many open files" error. And as we saw in the first post, you can't rely on garbage collection to clean things up at the right moment.
You could use finally to close it. But Python has a cleaner tool.
The Context Manager: with
with open("report.txt", "r", encoding="utf-8") as f:
content = f.read()
print(content)
# the block has ended, so the file is already closed
When the with block ends, the file is closed automatically, even if an exception happens inside the block. You don't have to remember anything.
Under the hood, with calls a special method called __enter__() at the start and __exit__() at the end. The file's __exit__() is where it closes itself.
You'll see this pattern all over AI code, not just for files. Database connections, HTTP sessions, async clients, and locks all use with. We already used one in the first post:
with requests.get(url, stream=True, timeout=30) as response:
...
Same idea: when the block ends, the connection is cleaned up.
File Modes: Choose Carefully
The mode you pass to open() decides what you can do with the file and where the "cursor" starts.
Mode
Meaning
Cursor starts at
Wipes the file?
Creates the file if missing?
r
Read only (default)
Beginning
No
No, raises FileNotFoundError
w
Write only
Beginning
Yes, immediately
Yes
a
Append
End
No
Yes
r+
Read and write
Beginning
No
No
w+
Write and read
Beginning
Yes
Yes
a+
Append and read
End
No
Yes
There are two more worth knowing: x creates a new file and fails if it already exists, and b (as in rb or wb) opens the file in binary mode, for things like images and PDFs.
Be careful with w
This is the one that hurts people:
with open("results.txt", "w") as f:
pass # we wrote nothing...
The file is now empty. Mode w erases the file the moment it's opened, whether you write anything or not.
So if a script opens your results file with w by mistake, hours of work are gone. For adding results, use a. For changing existing content, read it first (more on that below).
a+ has a small surprise
In a+, the cursor starts at the end of the file. So this returns an empty string:
with open("log.txt", "a+", encoding="utf-8") as f:
print(f.read()) # ''
You need f.seek(0) to move back to the start before reading. And even after seek(), writes in append mode still go to the end. Append can only add, never modify.
Reading and Writing Methods
Method
What it does
f.read()
Reads the whole file as one string
f.readline()
Reads one line per call
f.readlines()
Reads all lines into a list
for line in f:
Reads line by line, lazily
f.write(text)
Writes a string
f.writelines(list)
Writes a list of strings
f.seek(pos)
Moves the cursor (f.seek(0) goes to the start)
f.tell()
Tells you where the cursor is
If you read the first post, for line in f: should look familiar. A file object is an iterator, so it hands you one line at a time, just like a generator. For a large log file or CSV, this is the memory-friendly choice. read() and readlines() load everything at once.
One small gotcha: writelines() does not add newlines for you.
f.writelines(["line 1", "line 2"]) # line 1line 2
f.writelines(["line 1\n", "line 2\n"]) # what you actually wanted
The "Off by Two" Bug: Always Set the Encoding
Here's a bug that looks like magic until you understand it.
Someone wanted to replace a report ID inside a text file. The plan was clever: find the position of the ID, jump there with seek(), and overwrite it.
with open("inspection.txt", "r+") as f:
content = f.read()
start = content.find("REP-")
f.seek(start)
f.write("REP-1111-9999")
The new ID landed two characters off. The "fix" was to add start += 2 and move on.
That works on one machine, with one file. It breaks on the next one. Here's what's really going on.
find()andseek()don't count the same thing.
content.find() counts characters in the decoded text.
In text mode, seek() works with positions in the file, which behave like bytes.
Those two numbers drift apart whenever one character takes more than one byte:
A bullet like • is one character, but three bytes in UTF-8.
On Windows, a line break is saved as \r\n (two bytes), but Python reads it as \n (one character).
So every bullet or line break before your target shifts the position. Hard-coding +2 just patches the number for one specific file.
There's a second, related problem. If you don't pass an encoding, Python uses your system's default. On many Windows machines that's cp1252, not UTF-8. A UTF-8 bullet then shows up as garbage like •. This garbled text has a name: mojibake. If you see it in your output, an encoding mismatch is almost always the reason.
The fix has two parts.
1. Always pass the encoding:
with open("inspection.txt", "r", encoding="utf-8") as f:
content = f.read()
This matters a lot in AI work, because your text will include bullets, accented names, emojis, and non-English scripts.
2. Don't do position math in text mode. The Python docs only promise that seek() works in text mode with 0 or a value you got from tell(). If you really need exact byte positions, open the file in binary mode and search bytes:
with open("inspection.txt", "rb") as f:
data = f.read()
start = data.find(b"REP-") # this is a real byte position
But for most cases, there's a better way altogether.
The Safe Way to Update a File
Instead of editing in place, read the whole thing, change it in memory, and write the result to a temporary file. Then swap the temporary file in.
import os
def replace_in_file(path, old, new):
with open(path, "r", encoding="utf-8") as f:
content = f.read()
updated = content.replace(old, new)
temp_path = path + ".tmp"
with open(temp_path, "w", encoding="utf-8") as f:
f.write(updated)
os.replace(temp_path, path) # swaps the files in one step
replace_in_file("inspection.txt", "REP-2026-8842", "REP-1111-9999")
Why the temporary file? If your script crashes halfway through writing, the original file is still intact. os.replace() swaps the new file in as a single step, so you never end up with a half-written file.
Yes, we're using w here, but only on a fresh temporary file we just created. That's exactly the kind of place where w is safe.
This works well for normal-sized files. For huge files, combine it with what we learned in the first post: read line by line, write each changed line to the temporary file, then swap.
Putting It All Together
Let's rebuild the Friday-night pipeline with everything from this post.
import json
def summarize_documents(paths, endpoint, api_key):
failed = 0
with (
open("summaries.jsonl", "a", encoding="utf-8") as results,
open("failures.log", "a", encoding="utf-8") as failures,
): # this form needs Python 3.10+
for path in paths:
try:
with open(path, "r", encoding="utf-8") as f:
text = f.read()
summary = call_with_retry(endpoint, api_key, f"Summarize:\n{text}")
except FileNotFoundError:
failures.write(f"{path}\tfile not found\n")
failed += 1
except UnicodeDecodeError:
failures.write(f"{path}\tnot valid UTF-8\n")
failed += 1
except LLMProviderError as error:
failures.write(f"{path}\t{error}\n")
failed += 1
else:
results.write(json.dumps({"path": path, "summary": summary}) + "\n")
results.flush() # push it to disk now, not later
return failed
Here's what changed compared to the original version:
One bad document doesn't stop the run. Each document has its own try/except, so a failure is logged, and the loop moves on.
Rate limits and outages are retried with backoff. A wrong API key is not.
Results are saved as we go, one line per document, in append mode. If the script stops at document 3,412, the first 3,411 summaries are already on disk.
Every file is closed properly, thanks to with.
Encoding is explicit, so the same code behaves the same on your laptop and on the client's server.
Failures are written down, so on Monday morning you know exactly which documents to rerun.
Saving one JSON object per line (the .jsonl format) is a common pattern in AI pipelines. It's easy to append to, and you can read it back line by line without loading everything into memory.
Quick Summary
An exception has a type and a message. Without handling, any exception stops your script.
Use try/except, and catch specific exceptions first, with Exception last. Avoid a bare except:.
else runs only on success. finally always runs.
Write the logic first, see which errors really happen, then handle those.
Create custom exceptions for business rules and integration failures, and use raise ... from error to keep the original cause.
Every LLM API call needs a timeout, raise_for_status(), and handling that's different for 429, 5xx, and other 4xx errors.
Retry only the errors that can succeed on retry.
Use with open(...) so files always close, and always passencoding="utf-8".
Be careful with mode w. It erases the file the moment it opens.
To update a file safely: read, change in memory, write to a temp file, then os.replace().
None of this is glamorous. But it's the difference between a demo that works on Friday and a pipeline that's still running on Monday.
What's Next in This Series
Our pipeline can now survive rate limits, timeouts, and bad files. But it still has one big weakness: it's slow. It sends one document to the LLM, waits for the answer, and only then moves on to the next one. With 5,000 documents, most of that time is spent just waiting on the network.
In the next post, we'll look at how to fix that. We'll cover the difference between concurrency and parallelism, when to use threads, processes, or async Python, and how to run many LLM calls at the same time without hitting rate limits or buying a bigger server.
If you're building an AI product and running into problems like these, whether it's a pipeline that crashes halfway through, API calls that fail without warning, or a prototype that fell apart on real data, let's talk. This is exactly the kind of work we do at CreoWis.
We at CreoWis believe in sharing knowledge publicly to help the developer community grow. Let’s collaborate, ideate, and craft passion to deliver awe-inspiring product experiences to the world.
This article is crafted by Syket Bhattachergee, a passionate developer at CreoWis. You can reach out to him on X/Twitter, LinkedIn, and follow his work on the GitHub.
