LLM Pipeline Patterns

Before sending a request to LLM or a search engine, reformulate it in a different way. This compensates for inaccurate user formulations and aligns the semantics to the index vector space.

→ user: "why does my back hurt"
→ LLM rewrite: “causes of low back pain, risk factors”
→ search by reformulated query
→ results are combined with the original context

Breaking a long document into chunks for indexing or transferring into context. Strategies: fixed-size, recursive semantic, sentence-based, paragraph-based, sliding window with overlap.

overlap = 20% // do not lose context at the junction of chunks
chunk size = 512 tokens // accuracy/recall trade-off
metadata = {source, page, section} // for filtering

Maximum filling of the context window with relevant information from different sources: retrieved chunks + metadata + examples + history. The goal is to convey everything the model needs in one call.

[system_persona] [task_description]
[retrieved_chunks × N]
[few_shot_examples × K]
[conversation_history]
[user_query]

Instead of searching by raw query, ask LLM to generate a hypothetical answer, and then search by its embedding. This works because the answer embedding is semantically closer to the actual documents than the question embedding.

query: "how does the transformer work?"
↓ LLM(query) → hypothetical answer ~200 words
↓ embed(hypothetical_answer)
↓vector_search(embedding)
↓ real relevant documents

Explicitly instructing the model to “think out loud” before answering. The model generates intermediate steps of reasoning, which reduces errors on complex tasks - especially mathematics, logic, and multi-step inferences.

Prompt: "Think step by step before answering"
→ the model reveals reasoning in its response
→ the final answer follows from the steps derived

Options:
  zero-shot CoT: add "Let's think step by step"
  few-shot CoT: show examples with reasoning
  invisible CoT: ... tags (hidden)

Passing input→output examples directly to the prompt. Dynamic version - examples are selected from the repository through semantic search, relevant to the current request. This is in-context learning without fine-tuning.

query → embed(query)
       → knn_search(examples_store, k=3)
       → selected examples in prompt
       → LLM(system + examples + query)

LLM generates or optimizes a prompt for another LLM call. It is used in automatic optimization of prompts (APE, DSPy-style), where the model itself offers improvements for the task.

task + examples of failures
→ Meta-LLM: "write the best prompt for X"
→ generated prompt
→ Worker-LLM(generated_prompt)
→ evaluation → iteration

Assign a role to system prompt to control style, tone, level of expertise, and output format. Activates the desired distribution of tokens in the model’s weights - it “switches the register”.

System: "You are a senior data engineer with 10 years of experience
in industrial analytics. Answer technically
no water, use specific numbers."
→ the model activates technical vocabulary and style

Interleaved reasoning + action. The model alternates between "Thought: [reasoning]" and "Action: [calling a tool]" and "Observation: [result]". This allows the agent to adapt to the results of the tools on the fly.

Thought: you need to check the current exchange rate
Action: search("USD RUB exchange rate today")
Observation: 1 USD = 91.3 RUB (03/09/2025)
Thought: now I can calculate the amount
Action: calc(1500 * 91.3)
Observation: 136950
Answer: 1500 USD = 136,950 RUB

One incoming request launches N parallel LLM calls with different prompts, configurations, or subtasks. All calls are independent - there are no dependencies between them. The main goal is either to cover different aspects of the problem, or to collect a variety of answers for subsequent selection of the best one.

                   ┌─── LLM(prompt_A) ──→ result_A
query ──→ router ──┼─── LLM(prompt_B) ──→ result_B
                   └─── LLM(prompt_C) ──→ result_C
                                 ↓↓↓ fan-in ↓↓↓
                         aggregator(A, B, C) → final_output

Assembling the results from N parallel calls into one final response. Strategies: concatenation, map-reduce summarization, voting, LLM synthesis.

concat: just concatenate
vote: by majority vote
reduce: LLM(summarize([A,B,C]))
score+pick: select the best by metric
merge: LLM(merge([A,B,C]))

Two-stage search/generation: in the first stage, many candidates are quickly received (recall); in the second, an expensive model re-evaluates and rearranges them in descending order of relevance (precision). Key insight: fast retriever (BM25, vector search) can find, but ranks poorly. A slow reranker (cross-encoder, LLM) can accurately compare query-document pairs because it sees both at once - this is a fundamentally different type of attention.

Why is reranker more accurate than vector search?

bi-encoder: embed(query) embed(doc) ← independent, fast, ~100ms
cross-encoder: LLM([query; doc]) → score ← together, slowly, accurately

Pipeline:
query → fast_retriever (top-100 docs)
      → reranker (top-5 out of 100)
      → LLM (generation according to top-5)

Run the same prompt N times with temperature > 0, get different reasoning paths, take the answer by majority vote. Statistically more reliable than a single greedy call.

runs = [LLM(q, temp=0.7) for _ in range(5)]
answers = [extract_answer(r) for r in runs]
final = Counter(answers).most_common(1)[0]

For a document longer than the context window: split into chunks, summarize each in parallel (map), then combine the summaries into the final one (reduce). Can be done recursively.

doc → [chunk1, chunk2, ..., chunkN]
map: [summarize(c) for c in chunks] // in parallel
reduce: summarize(join(summaries)) // final

Router determines the type of request and routes it to a specialized prompt/agent. Each “expert” is an LLM with a narrow system imperative. This is a prompt-level analogue of an architectural MoE.

query → Router-LLM: {type: "code"|"math"|"write"}
                ↓
  "code"  → Code-Expert  (system: senior engineer)
  "math"  → Math-Expert  (system: mathematician + CoT)
  "write" → Write-Expert (system: copywriter)

The output of one LLM call becomes the input of the next. Each step transforms the data: extract → analyze → generate → format. Allows you to break a complex task into manageable subtasks.

raw_text
 → [Step 1] extract_facts(text)      → facts_json
 → [Step 2] analyze_sentiment(facts)  → sentiment
 → [Step 3] generate_report(facts, sentiment)

LLM or deterministic code decides on the next step. If condition A is branch 1, if B is branch 2. Allows you to build non-linear pipelines with if-else logic.

query → classifier → {intent}
  intent == "complaint"  → escalation_flow
  intent == "question"   → faq_flow
  intent == "purchase"   → sales_flow

Split into two independent LLM calls: Planner creates a list of steps, Executor executes each step in turn. Planner sees the entire task strategically, Executor is focused on one step - this reduces errors.

Planner(task) → ["step1", "step2", "step3"]
                                      ↓
for step in plan:
    result = Executor(step, context=prev_results)
    context.append(result)

LLM first breaks down a problem into subtasks from simple to complex. Then solves them sequentially, using simple answers to solve complex ones. An analogue of dynamic programming for LLM.

Q: "How many minutes are in 3.5 days?"
→ Sub-Q1: “How many hours are there in a day?”        → 24
→ Sub-Q2: “How many minutes are there in an hour?”       → 60
→ Sub-Q3: “How many minutes are there in 3.5 * 24h?” → 5040

The model generates a draft, criticizes it, improves it - and so on for N iterations. This simulates the editing process. Stopping based on quality condition or number of iterations.

draft = LLM(task)
for i in range(max_iter):
    critique = LLM(f"Find the flaws: {draft}")
    if "no comments" in critique: break
    draft = LLM(f"Improve taking into account: {critique}\n{draft}")
return draft

Force JSON/structured output via Pydantic, Instructor, function calling or JSON mode. Scheme error → immediate retry with a description of the error in the prompt.

class Output(BaseModel):
    sentiment: Literal["pos","neg","neu"]
    score: float = Field(ge=0, le=1)

result = instructor.chat(model, Output, prompt)
// if parsing error occurs - auto-retry with traceback

Separate LLM call to evaluate the quality of the main output. Judge receives the original question + answer and gives a score based on the criteria. Works as an automated reviewer.

System: "You are a strict evaluator of the quality of answers"
User: f"""
Question: {question}
Answer: {answer}
Rate: accuracy (0-5), recall (0-5),
        hallucinations (yes/no)
Reply JSON.
"""

If there is an error, do not just repeat the call, but pass on to the next attempt a description of what went wrong. LLM sees his mistake and takes it into account. Exponential backoff for rate limits.

for attempt in range(max_retries):
    result = LLM(prompt + error_context)
    ok, error = validate(result)
    if ok: return result
    error_context += f"\n{attempt} failed: {error}"
    sleep(2 ** attempt) // exponential backoff
raise MaxRetriesError

Checking that every factual statement in the answer is supported by sources. Either through the NLI model (entailment), or through a separate LLM call, or through a citation search.

claims = extract_claims(llm_answer)
for claim in claims:
    score = nli_model(premise=source_docs,
                      hypothesis=claim)
    // entail / neutral / contradict
    if score == "contradict": flag(claim)

If the main model fails, automatic transition to the backup model occurs. Hierarchy: first the expensive model, in case of timeout/error - cheaper, then deterministic fallback.

try: return cloud_opus(prompt)
except:
  try: return gpt4o(prompt)
  except:
    try: return cloud_haiku(prompt)
    except: return "The service is temporarily unavailable"

Security layer before and after the LLM call. Input guardrails: toxicity, PII detection, jailbreak detection. Output guardrails: checking for admissibility, filtering personal data, subject restrictions.

user_input
 → [input_guard] → moderation, PII-masking
 → LLM(safe_input)
 → [output_guard] → toxic filter, PII-unmask
 → user

A combination of keyword search (BM25/TF-IDF) and semantic (vector) search. BM25 is precise for exact terms and abbreviations, vector for meaning. RRF (Reciprocal Rank Fusion) combines ranks.

score(doc) = Σ 1 / (k + rank_i(doc))
// k=60 — smoothing constant
// summarized across all search engines
bm25_results + vector_results → RRF → unified_top_k

LLM generates N rephrases of the original query, each searched in the index, the results deduplicated and merged. Covers different semantic aspects of one question.

query → LLM → [q1, q2, q3, q4] // N options
[search(q) for q in queries] // in parallel
→ deduplicate → top-k → LLM

After the initial response, the model determines what knowledge is missing, formulates a follow-up query and performs an additional search. The cycle continues until the context is complete. This is a RAG with an internal feedback loop.

context = []
while not sufficient(context):
    gaps = LLM(f"What is missing for the answer? {context}")
    new_docs = search(gaps)
    context.extend(new_docs)
return LLM(query, context)

Index small chunks for precise search, but pass their parent (large) chunks into the context. Search accuracy + context completeness. Different granularities for different tasks.

Index: small_chunks (128 tokens) → embeddings
Storage: parent_chunks (512 tokens)

query → search(small_chunks) → top-k small
      → fetch parent(small_chunk) → into LLM context

When the dialogue history exceeds the context window, compress old messages into a compact summary, leaving new ones complete. Rolling summary is updated with each new message.

if len(history) > threshold:
    old = history[:-10]
    summary = LLM(f"Summary: {old}")
    history = [summary_msg(summary)] + history[-10:]

Extracting and accumulating information about entities (people, companies, products) from the dialogue into a separate store. The next time an entity is mentioned, its profile is pulled into the context.

entity_store = {}
msg → LLM extract → {Ivan: “CTO”, “loves Python”}
entity_store["Ivan"].update(...)

// with a new message:
context += entity_store.get(detected_entities)

Past interactions are saved as embeddings. With a new request, search for semantically similar episodes from the past and add them to the context. Long-term memory without storing a full log.

after each dialogue:
  store(embed(summary), metadata)

on a new request:
  memories = search(embed(query), top_k=3)
  context = format_memories(memories) + query

Allocation of a special zone in the prompt for intermediate calculations, notes and agent state. The model writes to the scratchpad during reasoning and reads from it - this is external “working memory”.

[SCRATCHPAD]
Facts: X=42, Y=18
Previous step: found document #3
Current goal: check date
[/SCRATCHPAD]

[TASK] next step...

LLM decides when and which tool to call, generates call parameters, receives the result and continues reasoning. Tools: search, calculator, database, API, browser, interpreter code.

LLM → tool_call: {name: "search", args: {q: "..."}}
System → tool_result: "..."
LLM → tool_call: {name: "calculator", args: {...}}
System → tool_result: 42
LLM → final_answer: "Answer: 42"

After performing an action, the agent reflects: what happened, what went wrong, what needs to be corrected. This is a supervised self-improvement within one inference loop without updating the weights.

action = Agent.act(state)
result = env.step(action)
reflection = Agent.reflect(action, result)
// "I used the wrong tool because..."
next_action = Agent.act(state, reflection)

Several LLM agents with different roles interact through a common message bus or hierarchically. Orchestrator delegates tasks to specialized subagents and collects the results. Allows you to solve problems that require parallel expertise in different domains.

Orchestrator
   ├── delegate("market analysis") ──→ Research Agent → report
   ├── delegate("financial model") ──→ Finance Agent → spreadsheet
   └── delegate("presentation") ──→ Writer Agent → slides
                                    ↓ fan-in ↓
                          Orchestrator(report, spreadsheet, slides) → final

A dedicated critical agent looks at the output of the main agent and issues specific comments based on the specified criteria (checklist). The main agent fixes it. The critic can be stronger than the main one - asymmetry is useful.

draft = Worker(task)
critique = Critic(draft, criteria=[
  "accuracy of facts", "structure", "style"
])
final = Worker(task, critique=critique)

CoT extension: instead of one chain, there is a tree of reasoning options. At each step, N continuations are generated, they are evaluated, and beam search selects the best. Expensive, but powerful for complex planning.

thoughts = [initial_thought]
for depth in range(max_depth):
    candidates = []
    for t in thoughts:
        candidates += generate_k_continuations(t, k=3)
    thoughts = evaluate_and_prune(candidates, keep=beam_w)

Forced limitation of token space during generation at the logits level. Outlines/LMQL/Guidance masks invalid tokens at every moment of time - 100% guarantee of valid JSON/regex without retry.

schema = {"type": "object", "properties": {...}}
// at each generation step:
valid_tokens = grammar.get_valid_next_tokens(current_state)
logits[~valid_tokens] = -inf // masking
next_token = sample(softmax(logits))

After receiving the LLM output, a chain of deterministic transformations: strip markdown, extract JSON, normalize, deduplicate, sort, format. The less logic you put on LLM, the more predictable the system.

raw = llm_response
→ strip_think_tags(raw) // remove <think>
→ extract_json(raw) // regex/parser
→ validate_schema(json) // pydantic
→ normalize_values(json) // types, case
→ enrich_from_db(json) // join with data

Processing tokens as they are generated - without waiting for a complete response. Allows you to terminate generation early (early stopping), show progress to the user, and process the beginning of the response in parallel.

for token in llm.stream(prompt):
    buffer += token
    if detect_complete_sentence(buffer):
        yield process_sentence(buffer)
        buffer = ""
    if should_stop_early(buffer):
        llm.cancel()

Compress long context before sending to expensive model. The small model evaluates the “importance” of each token and removes the unimportant ones. 4-10x compression with minimal quality loss.

small_lm.score_tokens(long_context)
→ perplexity per token
→ drop low-perplexity tokens (predictable/noisy)
→ compressed_context (10-25% of the original)
→ expensive_llm(compressed_context)

Caching responses not by exact string match, but by semantic proximity of the request. If cosine distance < threshold - return cache. Reduces costs on repeated requests in different formulations.

q_emb = embed(query)
cached = search_cache(q_emb, threshold=0.95)
if cached: return cached.response  // hit

response = LLM(query)              // miss
cache.store(q_emb, response)
return response

The query complexity classifier directs simple queries to the cheap, fast model, and complex queries to the expensive one. 70-80% of queries in an average product are simple. RouteLLM, LLM Router.

complexity = classifier(query)  // fast, cheap
if complexity < 0.4:
    return haiku(query)          // $0.25/M tokens
elif complexity < 0.8:
    return sonnet(query)         // $3/M tokens
else:
    return opus(query)           // $15/M tokens

The small draft model generates K tokens quickly, the large model verifies them in one forward pass. Accepted tokens are free, rejected tokens are regenerated. 2-3x speedup with the same accuracy.

draft_tokens = small_model.generate(k=5) // fast
accepted = large_model.verify(draft_tokens) // 1 forward
// accept matches, regenerate the first non-match

Anthropic and OpenAI support caching of KV cache prefixes. If system prompt + context are the same between requests, they are not recalculated. Up to 90% savings on prefill tokens for long system prompts.

Structure the prompt: static → up, dynamic → down

[system + docs + examples] ← cached
────────────────────────────
[user query] ← changes every time

→ Anthropic: cache_control: {"type": "ephemeral"}