The Python Utility Stack for LLM Pipelines

In practice, most of the code around LLM does not generate tokens, but prepares the data for the model call and parses the result afterwards. This is where prompt templates, schema validation, string cleanup, JSON parsing, file/path logic, retries, batching and tracing live.

TYPICAL NON-MODEL LAYER

input text
  ↓
normalize / strip / parse
  ↓
render prompt template
  ↓
call model
  ↓
parse JSON / validate schema
  ↓
log/retry/cache/save

Often it is this layer that determines the reliability of the system more than the choice of model

Almost all “helper libraries” around LLM solve one of four problems: text rendering, structure checking, data transformation and orchestration runtime. If you keep this in mind, it is easier to choose a tool without the cult of libraries.

1.render:
   collect string/prompt/message

2.validate:
   make sure the structure is correct

3. transform:
   clean, parse, serialize, cut

4. orchestrate:
   batch, cache, run async, retrace

Not every task requires Jinja or Pydantic. Sometimes `f-string`, `dict`, `json.loads` and a couple of string methods are enough. A library is justified when it provides repeatability, safety, or readability better than bare Python.

LLM pipelines almost always stand between a non-ideal input and a non-ideal output. The user text is dirty, the retrieval context is noisy, the model output is unstable. Therefore, service Python is needed as a layer of normalization and insurance on both sides.

dirty input
→ make predictable
→ give to the model
→ get fragile output
→ make it safe and structured

For short prompts, `f-string` is usually sufficient. If reuse and a little more explicit naming is needed, `.format()` is sometimes suitable. When loops, conditions, blocks and external patterns appear on disk, the Jinja zone begins.

In Jinja, the central object is `Environment`. It stores loader, filters, globals, undefined behavior and template cache. For applications, this means that it is better to initialize the template engine once at the start of the service, rather than creating a new `Template(...)` for each request.

from jinja2 import Environment, PackageLoader, StrictUndefined

env = Environment(
    loader=PackageLoader("app"),
    undefined=StrictUndefined,
)

template = env.get_template("answer_prompt.j2")
prompt = template.render(query=query, context=context)

By default, the template may silently swallow the missing variable and render empty. For HTML this is already unpleasant, for prompt it is even worse: you will silently lose part of the instruction or context. Therefore, for LLM templates it is often wise to use `StrictUndefined` and fail early.

As soon as the prompt says “insert all retrieved chunks”, “if there is history - show it”, “if debug mode - add trace”, string concatenation begins to fall apart. Jinja makes these branches explicit and readable.

{% if history %}
Conversation history:
{% for msg in history %}
- {{ msg }}
{% endfor %}
{% endif %}

Question:
{{ query }}

{% for item in context_items %}
[{{ item.label }}]
{{ item.text }}
{% endfor %}

If the template is small, lives close to the code, and has no logic, `f-string` is usually simpler. Jinja makes sense when templates become a separate resource, require reuse, or are used by different teams and modes.

if prompt_fits_on_screen and has_no_loops:
    use f-string
else:
    use Jinja

`typing` describes the expected form of the data. `dataclass` conveniently creates lightweight Python objects. `Pydantic` adds runtime validation, coercion, serialization and JSON schema generation. In LLM pipelines, most often `typing` specifies a contract, and `Pydantic` makes it executable on dirty external input/output.

The main value of `BaseModel` is that it takes imperfect data and tries to force it into the expected form. This is especially useful for model output, webhook payloads, retrieval traces and structured response contracts.

class Answer(BaseModel):
    answer: str
    citations: list[str]
    confidence: float

obj = Answer.model_validate(payload)

In LLM pipelines, these three methods cover the lion's share of needs. `model_validate()` - when you already have a dict. `model_validate_json()` - when model output came as a JSON string. `model_dump()` - when you need to send the structure further to the log, API or prompt.

Answer.model_validate({"answer": "...", "citations": []})
Answer.model_validate_json(raw_json_string)
answer_obj.model_dump()

In real code, it is often not enough to say “the field must be a string.” You also need to check that `citations` are not empty, that `score` is in the range, that the two fields are consistent, that the Markdown is cleaned, and that the JSON does not contain forbidden keys. This is where `field_validator` and `model_validator` come in.

class Answer(BaseModel):
    answer: str
    citations: list[str]

    @field_validator("answer")
    @classmethod
    def clean_answer(cls, value: str) -> str:
        return value.strip()

    @model_validator(mode="after")
    def ensure_citations(self):
        if not self.citations:
            raise ValueError("citations required")
        return self

If the data is already trusted and you just want a clean lightweight object for internal calculations, `dataclass` is often better. It's simpler, lighter, and faster for domain objects that don't require runtime validation on every input.

Even without Pydantic, typing helps LLM code a lot. `Protocol` allows you to describe adapter interfaces, `Literal` fixes valid modes, `TypedDict` is convenient for intermediate JSON-like structures where you don’t want full-fledged objects yet.

class Embedder(Protocol):
    def embed_query(self, text: str) -> list[float]: ...

Mode = Literal["chat", "rag", "judge"]

A lot of LLM code comes down to neatly assembling and disassembling strings. `strip()` removes excess noise at the edges, `split()` cuts along delimiters, `splitlines()` is useful for line-based parsing responses, `join()` puts text back together without ugly concatenation.

text = raw_text.strip()
lines = text.splitlines()
lines = [line.strip() for line in lines if line.strip()]
normalized = "\n".join(lines)

When you need to split a string once into a before/delimiter/after, `partition()` gives a stable triple result and doesn't make you wonder if the delimiter was found. For LLM pipelines, this is useful when parsing headers, source labels, Markdown front matter, and prefixes like `Answer:`.

prefix, sep, rest = raw.partition("Answer:")

if sep:
    answer = rest.strip()
else:
    answer = raw.strip()

If you need to remove a specific prefix or suffix, it is better to use special methods rather than `replace()`. Otherwise, you can unexpectedly delete the desired fragment somewhere inside the line. This is especially noticeable when working with marks like `json\n...`, `assistant:` and markdown fences.

raw = raw.removeprefix("```json").removesuffix("```")
raw = raw.strip()

When prompt is written in triple quotes inside Python code, there is almost always extra indentation. `textwrap.dedent()` makes the template visually beautiful in code and at the same time removes the overall leading whitespace before sending it to the model.

prompt = dedent("""
    You are a helpful assistant.
    Return valid JSON.
    Question: {query}
""").strip()

Regular expressions in LLM code are useful for limited, formal things: extracting fenced JSON, replacing extra spaces, finding citation labels, normalizing bullet markers. They are bad for complex semantic parsing, but great for dirty mechanics.

json_block = re.search(r"```json\s*(.*?)\s*```", raw, re.S)
citations = re.findall(r"\[SOURCE\s+\d+\]", raw)
clean = re.sub(r"\n{3,}", "\n\n", raw)

`replace()` is great for banal normalization: replace `\r\n` with `\n`, remove non-breaking spaces, bring Unicode quotes to the same style. But as soon as the replacement has context and conditions, it is better to move on to regex or explicit parsing.

text = text.replace("\r\n", "\n")
text = text.replace("\u00a0", " ")
text = text.replace("—", "-")

Many small preparatory steps can be expressed in one readable line: clear the list of chunks, discard empty ones, add labels and assemble the final context block. This makes prompt-building shorter and more transparent.

context_block = "\n\n".join(
    f"[SOURCE {i}] {item.text.strip()}"
    for i, item in enumerate(context_items, start=1)
    if item.text.strip()
)

Even if the model promises to “return JSON,” the code still needs an explicit parse step. `json.loads()` converts string into Python structures, `json.dumps()` - vice versa. It's useful to remember that parsing untrusted JSON can be expensive, meaning input size and parsing errors need to be controlled.

try:
    data = json.loads(raw_model_output)
except json.JSONDecodeError:
    return fallback_or_retry()

validated = ResponseModel.model_validate(data)

For prompt files, evaluation datasets, cached traces, and local documents, `Path` is usually more convenient than manually working with path strings. Plus it is easier to combine with glob search and cross-platform logic.

from pathlib import Path

prompt_dir = Path("prompts")
template_path = prompt_dir / "judge.j2"
text = template_path.read_text(encoding="utf-8")

There is a surprising amount of CSV around LLM pipelines: offline evaluation, manual markup, log export, FAQ import, list of golden queries. The standard `csv` module can already do quite a lot and does not require dragging pandas if the task is small.

with open("eval.csv", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    rows = list(reader)

As soon as trace, chunks, answers or prompts are written to a file or log, you already live in a world of format conventions. Therefore, it is useful to decide in advance whether you have line-delimited JSON, a regular JSON array, CSV or plain text logs, rather than writing “whatever happens.”

Not every intermediate structure in a pipeline requires a full model. When you're working with half-baked dicts from retrieval payloads, API responses, or trace blocks, `get()` and `setdefault()` make the code smoother and reduce the noise from checking for keys.

source = payload.get("source_url")
meta = payload.setdefault("metadata", {})
meta.setdefault("tenant_id", tenant_id)

LLM pipelines constantly group something and count: hits by `doc_id`, errors by type, citation coverage, usage stats, retry reasons. For this, `defaultdict` and `Counter` often produce more readable code than manual `if key not in dict` and increments.

by_doc = defaultdict(list)
for hit in hits:
    by_doc[hit.doc_id].append(hit)

error_counts = Counter(trace.reason for trace in traces)

When you need to process documents into embedding batches, lazily merge multiple data streams, or take only the first N elements without extra copies, `itertools` come in very handy. In LLM code, this is especially common in ingestion and bulk evaluation.

for batch in batched(chunks, 32):
    vectors = embedder.embed_documents([c.text for c in batch])

all_queries = chain(user_queries, golden_queries)
preview = list(islice(all_queries, 10))

A cache based on pure helper functions sometimes gives unexpected benefits: tokenization of repeating pieces, reading templates from disk, normalizing directories, calculating small derived settings. But the cache is poorly suited for functions with side effects, async generators, and things that depend on rapidly changing external state.

@lru_cache(maxsize=128)
def load_prompt(name: str) -> str:
    return (Path("prompts") / name).read_text(encoding="utf-8")

When the same helper is called with almost the same arguments, `partial()` helps to capture some of the parameters and make the code shorter. This can be convenient for render helpers, logger wrappers and small adapters over SDK calls.

render_judge = partial(render_prompt, mode="judge")
render_rag = partial(render_prompt, mode="rag")

Many LLM systems are network-based in nature: external LLM API, vector DB, reranker API, object storage, telemetry backend. Therefore, `asyncio` and `async/await` are often useful not for “AI magic”, but for normal control of parallel I/O. Especially in FastAPI and batch-evaluation pipelines.

results = await asyncio.gather(
    retrieve(query1),
    retrieve(query2),
    retrieve(query3),
)

usually useful in conjunction with semaphore/rate limiting

If you have 1000 documents for embeddings or 500 eval queries, a naive piece-by-piece loop will often be too slow and expensive. A small utility-layer should be able to accurately transfer data to batches, log the batch size and not lose connection between inputs and outputs.

for batch in batched(items, BATCH_SIZE):
    outputs = call_api(batch)
    for item, output in zip(batch, outputs):
        save_result(item, output)

Most application pipelines do not need zoo from 20 libraries. Quite often, one reasonable set is enough: `Jinja` for complex templates, `Pydantic` for contracts, `json`, `re`, `pathlib`, `textwrap`, `collections`, `itertools` and `asyncio` as needed.

Many problems are not in the absence of a library, but in its incorrect use. Unseparated prompt lines, silently missing variables in the template, manual parsing of JSON via `split("{")`, regex instead of normal validation and giant utils.py - all this is then costly.

If in doubt, it is useful to run the problem through a short decision tree. Usually this is enough to choose a tool without unnecessary complication.

MINI TREE

Do you just need to insert 2-3 variables?
  → f-string

Need loops/if/templates on disk?
  → Jinja

Need to accept or return external data?
  → Pydantic

Need to quickly clean up a line?
  → str methods/textwrap/re

Need to batch/cache/group?
  → itertools/functools/collections

Around LLM pipelines, it is not the most “AI magic” code that wins, but the most predictable service layer. A good utility stack makes prompts explicit, structures validated, strings clean, data serializable, and runtime observable. And this often provides more benefits than another model change.