Exception Handling Anti-Patterns

def load_prompt(path):
    try:
        return Path(path).read_text()
    except:
        return DEFAULT_PROMPT

def load_prompt(path: str) -> str:
    try:
        return Path(path).read_text(encoding="utf-8")
    except FileNotFoundError:
        logger.warning("prompt %s not found; using default", path)
        return DEFAULT_PROMPT

def run_eval(suite):
    try:
        return suite.run()
    except BaseException as err:
        return {"status": "failed", "error": str(err)}   # Ctrl+C now reports "failed" and keeps going

def run_eval(suite):
    try:
        return suite.run()
    except Exception as err:
        logger.exception("eval suite %s crashed", suite.name)
        return {"status": "failed", "error": type(err).__name__}

def rerank(query, docs):
    try:
        return cross_encoder.rank(query, docs)
    except Exception:
        return docs

def rerank(query: str, docs: list) -> list:
    try:
        return cross_encoder.rank(query, docs)
    except (ProviderError, TimeoutError):
        logger.warning("reranker unavailable; keeping vector order", exc_info=True)
        metrics.increment("rerank.fallback")
        return docs

try:
    resp = await http.post(url, json=payload)
    resp.raise_for_status()
except httpx.HTTPStatusError as err:
    raise ProviderError("embedding request failed")

except httpx.HTTPStatusError as err:
    raise ProviderError(f"embedding request failed with {err.response.status_code}",
                        provider="tei") from err

if len(tokens) > limit:
    raise Exception("too long")

if len(tokens) > limit:
    raise ContextLengthError(f"prompt has {len(tokens)} tokens, limit {limit}",
                             provider=provider, tokens=len(tokens), limit=limit)

try:
    return await client.chat(messages)
except openai.APIError:
    raise ProviderError("openai failed") from None
except openai.RateLimitError as err:          # never runs: RateLimitError is an APIError
    raise RateLimitError("429") from err

try:
    return await client.chat(messages)
except openai.RateLimitError as err:
    raise RateLimitError("rate limited", provider="openai") from err
except openai.APIError as err:
    raise ProviderError("openai failed", provider="openai") from err

def save_checkpoint(state, path):
    try:
        torch.save(state, path)
    finally:
        return path      # disk-full error disappears; caller thinks the checkpoint exists

def save_checkpoint(state: dict, path: str) -> str:
    tmp = path + ".tmp"
    try:
        torch.save(state, tmp)
        os.replace(tmp, path)     # atomic
    finally:
        if os.path.exists(tmp):
            os.remove(tmp)
    return path

async def stream(prompt):
    try:
        async for tok in llm.stream(prompt):
            yield tok
    except asyncio.CancelledError:
        logger.info("client disconnected")

async def stream(prompt: str):
    try:
        async for tok in llm.stream(prompt):
            yield tok
    except asyncio.CancelledError:
        logger.info("client disconnected")
        raise
    finally:
        await llm.aclose()

def chunk(text: str, size: int, overlap: int) -> list:
    assert size > overlap

def chunk(text: str, size: int, overlap: int) -> list:
    if overlap >= size:
        raise ValueError(f"overlap ({overlap}) must be smaller than size ({size})")

except VectorStoreError as e:
    logger.error(e)
    logger.error(f"upsert failed: {e}")

except VectorStoreError:
    logger.exception("upsert failed", extra={"collection": coll, "batch": i})

try:
    cfg = load_config()
    client = make_client(cfg)
    docs = load_docs(cfg["corpus"])
    chunks = [c for d in docs for c in chunker.split(d)]
    vectors = embed(chunks)
    store.upsert(vectors)
except KeyError:
    raise ConfigurationError("missing key")

cfg = load_config()
try:
    corpus = cfg["corpus"]
except KeyError:
    raise ConfigurationError("config has no 'corpus' entry") from None
client = make_client(cfg)
chunks = [c for d in load_docs(corpus) for c in chunker.split(d)]
store.upsert(embed(chunks))

try:
    return tokenizer.encode(text)
except ValueError:
    raise                                  # delete; or add context:

except ValueError as err:
    err.add_note(f"text starts with {text[:30]!r}")
    raise

except json.JSONDecodeError as e:
    raise ValueError("bad tool arguments: " + str(e))

except json.JSONDecodeError as err:
    raise ToolArgumentsError("tool arguments are not valid JSON", tool_name=name) from err

try:
    import tiktoken
except ModuleNotFoundError as err:
    if err.name != "tiktoken":
        raise                       # a dependency OF tiktoken is missing: a real error
    tiktoken = None


def count_tokens(text: str, encoding: str = "cl100k_base") -> int:
    if tiktoken is None:
        raise ConfigurationError("token counting needs `pip install tiktoken`")
    return len(tiktoken.get_encoding(encoding).encode(text))

failures = []
for batch in loader:
    try:
        loss = step(model, batch)
    except torch.cuda.OutOfMemoryError as err:
        failures.append(repr(err))     # not `err`: its traceback references frames holding CUDA tensors
        del batch
        torch.cuda.empty_cache()