Every AI agent already runs in a loop. Almost nobody designs that loop on purpose — and that’s the difference between an agent that quietly gets the job done and one that burns through your budget overnight, convinced it’s making progress. That gap has a name now: loop engineering.

The loop itself is simple. On a single turn an agent reasons about what to do, takes an action (calls a tool, runs a search), observes the result, and reasons again — the reason → act → observe cycle that powers tools like Claude Code and Cursor. For a while the skill that mattered most was prompt engineering: wording that one request as well as possible. But once models could act, search, and read their own results, the bigger question became what decides the agent’s next move — and when it should stop. Loop engineering is the choice to design that on purpose instead of leaving it to chance: deciding the goal, the checks, and the stopping point, then building a small system that defines the goal once and lets the agent iterate until the work is done. The shift is from telling the agent what to do at each step to building the system that does the telling.

That’s powerful, but it’s where the real work starts. A poorly designed loop wastes tokens, runs forever, or fools itself into thinking it’s making progress; a well-designed one is efficient, stops at the right moment, and produces reliable output. The difference isn’t a cleverer prompt — it’s a better loop. This article is about when that’s worth doing, when it isn’t, and how to build one.

A quick note on tools. The examples below use LangGraph, because it makes the structure of a loop easy to see. But loop engineering isn’t tied to any one framework — the same ideas apply whether we build with LangGraph, a different agent framework, or plain code we write ourselves. The framework changes; the concepts stay the same.

Let’s dive in.

Table of Contents

1. The Anatomy of an Agentic Loop — the five parts every loop is made of
2. When to Use Loop Engineering — the problem shapes that genuinely need a loop
3. Common Types of Loops — the handful of shapes most loops take
4. How to Build a Loop — a step-by-step LangGraph walkthrough
5. Failure Modes & Guardrails — what breaks, and how to prevent each
6. When NOT to Use Loop Engineering — where a loop adds cost and risk for no benefit
7. Conclusion — a quick decision checklist
8. References

1. The Anatomy of an Agentic Loop

Before deciding when to use a loop, it helps to know what one is made of. Put simply, almost every agentic loop has five parts.

• A trigger. Something kicks the loop off — a human saying “go,” a scheduled job, a new support ticket, a pull request opening. Until there’s a trigger, nothing happens.
• A goal we can check. The loop needs to know what “done” looks like, ideally in a way a machine can verify. “Make the app better” is a recipe for an infinite loop; “make all the tests pass” or “produce an answer where every claim is backed by a cited source” gives the loop a real finish line.
• Tools. The loop only works if the agent can actually do things and see the results — run code and read the error, search the web and read what came back. Without real feedback from the environment, the agent is just guessing in a vacuum.
• State (memory). The current picture of the work: what’s been tried, what’s been found, how many times the loop has run. Managing this well is the hardest part of loop design — let the state balloon and the model loses the signal in the noise.
• Stop rules. The conditions that end the loop — and there are three ways out, not two: success (“goal met”), surrender (“tried five times, giving up”), and hand-off, where the loop hits something risky or ambiguous and escalates to a human instead of guessing.

Keep these five in mind. Most of the “when,” “when not,” and “how” below comes straight back to them.

2. When to Use Loop Engineering

A loop earns its keep when a single pass through the model isn’t enough — when the task genuinely needs the agent to try, check, and try again.

When to use:

• The work can be verified. This is the big one. If we can check the result automatically — tests pass or fail, an answer either cites real sources or doesn’t, a JSON output is valid or malformed — a loop can use that check to drive itself toward “correct.” Verification is the steering wheel; without it, looping just spends money.
• The first attempt is often wrong but fixable. Writing code, drafting and refining research, cleaning messy data — these improve with iteration. The agent produces something, sees what’s wrong, and patches it.
• The number of steps isn’t known in advance. Some problems take two steps; the same kind of problem might take seven the next time. A fixed pipeline can’t handle that variability, but a loop that runs until the goal is met can.
• The task runs unattended. If we want an agent to handle tickets overnight, or keep retrying a flaky integration until it succeeds, we need a system that decides its own next move rather than waiting for us.

Sample Use Case: A research agent that verifies its own findings. It drafts an answer, checks whether each claim is actually supported by sources, and if not, goes back and researches the weak spots — looping until the answer holds up or it hits a limit. This fits all four signals above, and it’s the example we’ll build in Section 4.

3. Common Types of Loops

Most agentic loops are variations on a few shapes. Knowing them helps you spot which one a problem wants before wiring anything together.

Generate-and-verify (reflection). One step produces something, a second step checks it, and the loop repeats until it passes. Steering is pass/fail — the check says “good enough” or “here’s what’s wrong.” Best when correctness is checkable: code that must pass tests, research whose claims must hold up. This is the loop we’ll build below.

   +-----------+      +-----------+
-->| Generate  |----->|  Verify   |----- pass -----> [ done ]
   +-----------+      +-----+-----+
        ^                   |
        +------- fail ------+

Pick it when you have an objective check for “correct” and want the agent to keep fixing until it passes. Example: an agent that writes a SQL query, runs it, and loops until the query executes and returns sensible rows.

Evaluator-optimizer (generate-and-score). Instead of a yes/no check, a scorer rates each attempt and the loop pushes toward the highest score rather than merely “correct.” Use it when there’s no clean pass/fail but you can still measure better — tuning a prompt, refining ad copy, ranking results.

   +-----------+      +-----------+
-->| Generate  |----->|  Score    |--- good enough --> [ done ]
   +-----------+      +-----+-----+
        ^                   |
        +-- improve --------+
            (use the score)

Pick it when quality is a spectrum rather than pass/fail, but you can put a number on it. Example: drafting a product tagline where a scorer rates clarity and punch, and the loop keeps refining the strongest candidate.

Plan-execute-replan. A planner lays out the steps, executors carry them out, and the loop circles back to revise the plan when a step turns up something it didn’t expect. Here the loop wraps the plan, not a single output — a fit for open-ended, multi-step work like research or a coding project.

   +-----------+      +-----------+
-->|   Plan    |----->|  Execute  |-- all steps done --> [ done ]
   +-----------+      +-----+-----+
        ^                   |
        +-- replan ---------+
            (hit a surprise)

Pick it when the task is multi-step and you can’t know every step up front. Example: “migrate this service to a new framework” — plan the steps, start executing, and replan the moment a step uncovers an unexpected dependency.

Scheduled / monitoring loop. Rather than iterating to a goal in one run, it wakes on a cadence or a trigger to watch and react — triage incoming tickets, sweep for failing builds, retry a flaky job. The loop runs across time instead of within a single task.

   +-----------+     +-----------+     +-----------+
-->|   Wait    |---->|   Check   |---->|    Act    |---+
   | (timer /  |     +-----------+     +-----------+   |
   |  trigger) |                                       |
   +-----------+ <-------------------------------------+
        (no fixed end — keeps watching over time)

Pick it when there’s no single finish line — you want ongoing watching and reacting. Example: a job that checks an inbox every few minutes and triages new tickets, or a nightly sweep that flags stale pull requests.

These aren’t mutually exclusive — real systems often nest them (a scheduled loop that, on each wake, runs a generate-and-verify pass inside it). We’ll take the most common shape, generate-and-verify, and build it end to end next.

4. How to Build a Loop

What we’re building. A research agent that answers a question and refuses to trust its own first draft. Its job: take a question, gather information and write an answer, then fact-check every claim in that answer against real sources. Any claim that isn’t supported gets sent back for another round of research. The agent keeps doing this until the whole answer holds up — or until it hits a retry limit and hands back the best version it has, flagging what’s still shaky. In short, its role is not just to answer, but to answer and then prove the answer to itself before returning it.

That self-checking behavior is exactly the generate-and-verify loop from Section 3, so it’s a clean first build.

LangGraph is a good fit for loop engineering because it models an agent as a graph — boxes (nodes) that do work, connected by arrows (edges) that decide what runs next. Crucially, those arrows can point backward, which is what makes a loop possible. (A plain linear chain can’t loop; a graph can.)

Our research agent has three jobs, so it has three nodes. Notice that the first two split the work into a maker and a checker — a doer and a critic kept separate, one of the most common loop patterns:

                    +------------------+
  START ----------> |     Research     | <-------------+
                    | (draft / revise) |               |
                    +--------+---------+               |
                             |                         |
                             v                         | needs work
                    +------------------+               | (+ feedback)
                    |      Verify      |               |
                    |  (check claims)  |               |
                    +--------+---------+               |
                             |                         |
                             v                         |
                       +-----------+   needs work      |
                       |  Decide   |-------------------+
                       | (router)  |
                       +-----+-----+
                             | verified / out of tries
                             v
                          [ END ]

1. Research — gather information and draft an answer.
2. Verify — critique the draft: check whether each claim in it is actually supported.
3. Decide — a small router that reads the verdict and chooses what happens next: finish, or loop back and research the weak spots.

Setup. This example runs on Python 3.12. Install the three packages it needs:

pip install langgraph==1.2.6 langchain-openai==1.3.3 pydantic==2.13.4

• langgraph — the graph/loop engine (StateGraph, START, END).
• langchain-openai — the OpenAI chat model wrapper (ChatOpenAI).
• pydantic — defines the structured verdict the verifier returns.

We’ll also need an OpenAI API key in your environment before running it:

export OPENAI_API_KEY="your-key-here"

python

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# A model client on OpenAI's Responses API, with the built-in web-search tool
# bound on — so the model can search the web itself, no extra package needed.
# "gpt-5" is a reasoning model; swap it for any current model you have access to.
llm = ChatOpenAI(model="gpt-5", use_responses_api=True)
searcher = llm.bind_tools([{"type": "web_search"}])

MAX_ITERATIONS = 3         # a stop rule, decided up front

# STATE — the shared memory passed between every node
class ResearchState(TypedDict):
    question: str          # what we're answering
    draft: str             # the current best answer
    verdict: str           # "verified" or "needs_work"
    feedback: str          # what the verifier said to fix
    iterations: int        # how many times we've looped

# The verifier returns a typed result, so the router gets a clean
# "verified" / "needs_work" to branch on instead of parsing prose.
class Verdict(BaseModel):
    status: str = Field(description='"verified" or "needs_work"')
    feedback: str = Field(description="claims lacking support, if any")

# NODE 1 — the maker: search the web, then draft (or revise) the answer.
def research(state: ResearchState) -> dict:
    prompt = "Search the web, then answer the question. Back every claim with a source.\n"
    if state.get("feedback"):
        prompt += f"A reviewer flagged these gaps — fix them:\n{state['feedback']}\n"
    prompt += f"\nQuestion: {state['question']}"
    draft = searcher.invoke(prompt).text
    return {"draft": draft, "iterations": state["iterations"] + 1}

# NODE 2 — the checker: search for evidence, then critique the draft.
def verify(state: ResearchState) -> dict:
    evidence = searcher.invoke(
        f"Search the web for evidence to fact-check claims about: {state['question']}"
    ).text
    checker = llm.with_structured_output(Verdict)
    result = checker.invoke(
        "You are a fact-checker. Using the evidence below, reply 'verified' "
        "only if every claim in the answer is supported; otherwise 'needs_work' "
        "and list the unsupported claims as feedback.\n\n"
        f"Evidence:\n{evidence}\n\nAnswer:\n{state['draft']}"
    )
    return {"verdict": result.status, "feedback": result.feedback}

# ROUTER — the heart of the loop. Reads the verdict, picks the next step.
def decide(state: ResearchState) -> Literal["research", "__end__"]:
    if state["verdict"] == "verified":
        return "__end__"                      # success: goal met
    if state["iterations"] >= MAX_ITERATIONS:
        return "__end__"                      # surrender: out of tries
    return "research"                         # loop back and fix the gaps

# WIRE IT UP
graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("verify", verify)

graph.add_edge(START, "research")            # trigger
graph.add_edge("research", "verify")         # always verify a fresh draft
graph.add_conditional_edges("verify", decide, {
    "research": "research",                   # the backward arrow = the loop
    "__end__": END,
})

agent = graph.compile()

result = agent.invoke({
    "question": "What were the main causes of the 2008 financial crisis?",
    "draft": "", "verdict": "", "feedback": "", "iterations": 0,
})
print(result["draft"])

Two things worth noting. First, both nodes reach the live web through OpenAI’s built-in web_search tool (bound on with bind_tools) — the research node searches before drafting so its claims rest on real results, and the verify node searches again to fact-check the draft against fresh evidence rather than the model's memory. That second search is what gives the loop something objective to steer by. Second, the verify node uses with_structured_output, which forces the model to return a typed verdict and feedback rather than prose — that's what lets the router branch reliably instead of parsing free text.

Notice how cleanly the five parts of a loop map onto the code:

• Trigger: invoke(...) starts it with a question.
• Goal we can check: the verify node decides "verified" vs "needs_work".
• Tools: both nodes use OpenAI’s built-in web search for real evidence; research drafts from it, verify fact-checks against it.
• State: the ResearchState dictionary carries the draft, the feedback, and the iteration count from node to node.
• Stop rules: the decide router ends the loop on either success or hitting MAX_ITERATIONS.

That last point is the whole game. The router is where loop engineering actually lives. A single line — “stop if verified, stop if we’ve tried three times, otherwise go back” — is the difference between an agent that reliably improves its answer and one that runs up our bill forever.

Our example has two exits — verified, or out of tries — but production loops usually add a third: hand off to a human. When the agent hits something risky or ambiguous, the smartest move isn’t to guess or to quit silently; it’s to stop and escalate with everything it has gathered. In the router that’s just one more branch: if confidence is low or the action is sensitive, return an "escalate" route that ends the loop and surfaces the draft, the open questions, and what's been tried. Knowing when not to decide is part of good loop design.

5. Failure Modes & Guardrails

Once a loop can run on its own, a handful of predictable failures show up. Each has a standard defense.

• The infinite loop. The agent never decides it’s done and runs until something crashes or our budget does. Guard: always include a hard cap — a maximum number of iterations, like MAX_ITERATIONS above. Every loop needs an escape hatch that doesn't depend on the agent's own judgment.
• Token burn. Each pass re-sends a growing pile of context, so cost creeps up quietly. Guard: keep state lean. Pass forward only what the next step needs — the current draft and the specific feedback — not the entire history of everything that’s been tried.
• Hallucinated progress. The agent reports success without achieving it (“I’ve fixed the issue!” — it hasn’t). Guard: never let the agent grade its own homework on the thing that matters. Use a separate verification step with an objective check — tests that really run, sources that really exist — so “done” means done, not “the model said so.”
• Spinning on the same error. The loop retries the identical failing action and gets the identical failure. Guard: feed the previous failure into the next attempt so the agent changes its approach instead of repeating it — exactly why our research node receives the verifier's feedback.
• Silent, unaccountable actions. As loops act over longer stretches of time, we lose track of what happened and why. Guard: log every step — what the agent saw, what it decided, what it did. Observability stops being optional the moment a loop touches anything that matters, and tools like LangSmith exist specifically to trace these runs.
• Going fully autonomous on day one. A loop that takes real actions unsupervised from the start is one we haven’t earned the right to trust yet. Guard: roll it out in phases — start in report-only mode (it tells us what it would do), then let it make assisted changes we approve, then let it run unattended. Each phase has to earn the next.

6. When NOT to Use Loop Engineering

Loops are powerful, which makes them easy to over-apply. Reach for something simpler in these cases.

When NOT to use:

• One pass is enough. Summarizing a document, translating text, answering a factual question, classifying an email — these are single-shot tasks. Wrapping them in a loop adds cost, latency, and new failure modes for zero benefit. If a straight line of steps does the job, use a straight line.
• We can’t tell when it’s done. If there’s no way to check the result — the goal is subjective, fuzzy, or only a human can judge it — a loop has nothing to steer by. It will either stop arbitrarily or spin forever. Without a verifiable goal, we don’t have a loop; we have a slot machine.
• Mistakes are expensive or irreversible. A loop that sends emails, moves money, or deletes data on each iteration can do real damage fast. Keep a human in the approval path, or don’t loop on the consequential action at all.
• Cost and latency matter more than autonomy. Every iteration is more model calls, more tokens, more time. For a high-volume, low-margin, or latency-sensitive feature, a predictable single call often beats a smart-but-expensive loop.

Rule of thumb: if we can’t write down how the loop knows it succeeded and when it must give up, we’re not ready to build one yet.

Conclusion

Loop engineering is a genuine shift in where the leverage lives. The most valuable skill is no longer writing the perfect prompt; it’s designing the system that tells the agent what to do — the trigger, the goal, the tools, the state, and the stop rules. But a loop is a tool, not a default.

Before building one, run through five quick questions:

1. Can we verify the result automatically? If not, don’t loop — we have nothing to steer with.
2. Does the task actually improve with iteration? If one pass is enough, keep it simple.
3. What exactly does “done” mean? Write the success condition down before writing any code.
4. When must it give up? Set a hard limit so the loop can’t run forever.
5. Are the actions safe to repeat? If each iteration could cause real harm, add a human checkpoint or don’t loop on that step.

If we can answer all five, we’re ready to build a loop that’s reliable rather than just impressive. Start small — a single research-and-verify cycle like the one above — get the stop rules right, and grow from there.

One thought to close on: a loop doesn’t just speed the work up, it multiplies whatever judgment goes into it — repeating good calls and bad ones alike, at machine speed. So the goal isn’t to wire one up and walk away. The discipline is to stay close enough to keep understanding what the loop does, even after it runs on its own — to remain its engineer rather than just the person who starts it.

References

• Loop Engineering (Cobus Greyling, Substack) — https://cobusgreyling.substack.com/p/loop-engineering
• Loop Engineering (Addy Osmani) — https://addyosmani.com/blog/loop-engineering/
• loop-engineering — patterns & reference (GitHub) — https://github.com/cobusgreyling/loop-engineering
• LangGraph documentation — https://langchain-ai.github.io/langgraph/
• LangGraph (overview) — https://www.langchain.com/langgraph
• Building Effective Agents (Anthropic) — https://www.anthropic.com/research/building-effective-agents

In this agentic world we’re living in, it’s tempting to believe that one powerful AI agent can do it all — research, write, code, review, and deploy. And honestly, for simple tasks, it can.

But before we go further, let’s get on the same page about what we actually mean by an “agent.”

An agent isn’t just an LLM responding to a prompt. It’s an LLM with autonomy — it can reason about a goal, decide what steps to take, use tools (APIs, databases, search engines, code interpreters), observe the results, and adjust its approach. Think of it as the difference between asking someone a question and hiring someone to get a job done. A chatbot answers. An agent acts.

Now here’s where it gets interesting.

The moment we ask a single agent to juggle ten things at once — pull data from five sources, cross-check facts, draft a report, get it reviewed, and fix what’s wrong — the cracks show. It loses context. It hallucinates under pressure. It forgets what it said three steps ago.

Sound familiar? It should. We solved this exact problem in software engineering decades ago. We stopped building monoliths. We broke systems into microservices — small, focused, independently deployable. Each service did one thing well and talked to others through clean interfaces.

Multi-agent AI is that same evolution, and we’re right in the middle of it.

Instead of asking one LLM to be the architect, builder, tester, and project manager all at once, we give each role to a specialized agent. One plans. One executes. One critiques. One decides. They collaborate, challenge each other, and together deliver what no single agent could.

The real question for us isn’t whether to use multiple agents — it’s which pattern fits our problem and when.

That’s exactly what we’ll cover in this article. We’ll walk through the most widely adopted multi-agent design patterns — each with a visual diagram, a real-world use case, clear guidance on when to use it, and which frameworks support it out of the box.

Whether we’re building with LangGraph, CrewAI, AutoGen, Google ADK, or the Anthropic Agent SDK — by the end, we’ll know exactly which pattern to reach for and why.

Let’s dive in.

What We’ll Cover

Core Patterns

1. Sequential / Pipeline — Agents in a straight line, one after another
2. Orchestrator-Worker — A manager delegates to specialists
3. Parallel / Fan-Out → Fan-In — Independent tasks run concurrently
4. Reflection / Self-Critique — Generate, review, improve, repeat
5. Router / Dispatch — Classify and route to the right agent
6. Planning + Execution — Separate thinking from doing
7. Handoff — Transfer control when out of scope
8. Evaluator-Optimizer Loop — Generate, score, improve

Bonus

• Emerging Patterns — Debate, Group Chat, Mixture of Agents

Putting It All Together

• Quick Reference: Pick Your Pattern
• Combining Patterns
• Final Thoughts

1. Sequential / Pipeline

+-----------+     +-----------+     +-----------+     +-----------+
|  Agent A  | --> |  Agent B  | --> |  Agent C  | --> |  Agent D  |
| (Extract) |     |(Transform)|     | (Validate)|     |(Summarize)|
+-----------+     +-----------+     +-----------+     +-----------+

The simplest pattern we can start with. We line up our agents like an assembly line — each one does its job and passes the result to the next.

When to use:

• Our task has clear, ordered stages
• Each stage has a distinct responsibility
• We need auditability and traceability at every step

When NOT to use:

• Stages are independent and don’t need sequential ordering
• We need speed — this pattern is only as fast as the slowest agent

Sample Use Case: A content publishing pipeline — Agent A extracts key points from raw research, Agent B transforms them into a blog draft, Agent C validates facts and grammar, Agent D generates the final summary and SEO metadata.

Framework Support:

• LangGraph — Native linear graph edges
• CrewAI — Process.sequential
• AutoGen — initiate_chats with sequential carryover
• Google ADK — SequentialAgent (native workflow agent)
• Anthropic Agent SDK — No dedicated sequential API; Claude naturally chains steps through its reasoning loop

2. Orchestrator-Worker (Hierarchical)

                          +----------------+
                          |  Orchestrator  |
                          |   (Manager)    |
                          +-------+--------+
                                  |
                     +------------+------------+
                     |            |            |
                     v            v            v
               +----------+  +----------+  +----------+
               | Worker A |  | Worker B |  | Worker C |
               |(Frontend)|  | (Backend)|  |(Database)|
               +----+-----+  +----+-----+  +----+-----+
                    |              |              |
                    +--------------+--------------+
                                   |
                                   v
                          +----------------+
                          |  Orchestrator  |
                          |  (Aggregate)   |
                          +----------------+

This is probably the most widely used pattern in production. We have a smart manager agent that understands the big picture, dynamically decides what sub-tasks to create, delegates to specialists, monitors progress, and stitches the results together. The key word here is dynamic — the orchestrator reasons about the task and may change its plan mid-execution based on what workers return.

How is this different from Parallel Fan-Out? In Fan-Out, we know all the sub-tasks upfront and just run them simultaneously. Here, the orchestrator figures out the sub-tasks at runtime, may run some in sequence and others in parallel, and can reassign or retry if something fails. Think of it as a project manager vs a mail room — the project manager thinks and adapts, the mail room just distributes.

When to use:

• Sub-tasks aren’t known upfront and need dynamic decomposition
• Workers may have dependencies on each other’s output
• We need centralized coordination, error handling, and adaptive replanning

When NOT to use:

• The task is simple enough for a single agent
• All sub-tasks are independent and known upfront — just use Parallel Fan-Out instead

Sample Use Case: An e-commerce order processing system — a customer places an order, and the orchestrator agent takes over. It delegates to an Inventory agent (check stock availability), a Payment agent (process the payment), and a Shipping agent (calculate delivery options and schedule dispatch). But here’s the key: if the Inventory agent reports “out of stock,” the orchestrator adapts — it asks a Recommendation agent to suggest alternatives, updates the customer, and only proceeds to Payment once the customer confirms. A simple fan-out can’t handle this kind of dynamic decision-making.

Framework Support:

• LangGraph — Supervisor pattern with sub-graphs
• CrewAI — Process.hierarchical with manager_llm
• AutoGen — GroupChat with GroupChatManager
• Google ADK — LlmAgent with sub_agents and transfer_to_agent
• Anthropic Agent SDK — Parent agent spawns subagents via Agent tool (core pattern)

3. Parallel / Fan-Out → Fan-In

                          +-----------+
                          | Splitter  |
                          +-----+-----+
                                |
                  +------+------+------+---------+
                  |      |             |         |
                  v      v             v         v
              +-------+  +-------+  +-------+  +-------+
              |Agent 1|  |Agent 2|  |Agent 3|  |Agent 4|
              | (Web) |  | (DB)  |  | (API) |  |(Docs) |
              +---+---+  +---+---+  +---+---+  +---+---+
                  |          |           |          |
                  +----------+-----------+----------+
                                 |
                                 v
                          +------------+
                          | Aggregator |
                          +------------+

This is the speed pattern. When we have independent sub-tasks that don’t depend on each other, why run them one by one? We fan out to multiple agents running concurrently, then fan in to merge the results. Unlike the Orchestrator-Worker, there’s no smart manager here — we know all the sub-tasks upfront, fire them all at once, and a simple aggregator combines the results.

How is this different from Orchestrator-Worker? The splitter doesn’t think — it just distributes pre-defined tasks. There’s no dynamic decision-making, no replanning, no dependency management between workers. It’s pure parallelism for speed. If any worker’s output should change what other workers do, we need the Orchestrator pattern instead.

When to use:

• All sub-tasks are known upfront and independent of each other
• Latency is critical — we want results as fast as the slowest agent
• We’re gathering data from multiple sources that don’t affect each other

When NOT to use:

• Tasks have dependencies on each other — use Orchestrator-Worker instead
• We need dynamic task creation based on intermediate results

Sample Use Case: A competitive analysis tool — we need data from four independent sources. Agent 1 scrapes competitor websites, Agent 2 pulls financial data from APIs, Agent 3 gathers social media sentiment, Agent 4 searches news articles. None of them need each other’s output. They all run at the same time, and the aggregator combines everything into a single report. Total time = slowest agent, not the sum of all four.

Framework Support:

• LangGraph — Parallel branches with fan-in node, Send API for map-reduce
• CrewAI — async_execution=True on tasks
• AutoGen — a_initiate_chats with concurrent execution
• Google ADK — ParallelAgent (native workflow agent, runs sub-agents concurrently)
• Anthropic Agent SDK — Multiple Agent tool calls in a single message execute concurrently

4. Reflection / Self-Critique

                      +-----------+
                      | Generator |<-----------+
                      +-----+-----+            |
                            |                  |
                            |  Output          | Feedback
                            v                  | (if not good enough)
                      +-----------+            |
                      | Reviewer  |------------+
                      +-----+-----+
                            |
                            |  Approved (meets quality threshold)
                            v
                      [Final Output]

We all write better with a good editor. This pattern gives our AI a built-in editor — one agent generates, another reviews, and we iterate until the output meets our quality bar.

When to use:

• Code generation (write → test → fix cycles)
• Content creation where first drafts aren’t good enough
• Any task where we can define clear quality criteria

When NOT to use:

• We need real-time responses — iteration adds latency
• The task is simple enough that the first pass is usually correct
• We don’t have clear criteria for “good enough”

Sample Use Case: An automated code review system — the Generator agent writes a function, the Reviewer agent runs tests, checks for edge cases, security issues, and style. If anything fails, it sends feedback. The Generator rewrites. We repeat until all tests pass and the review score exceeds our threshold.

Framework Support:

• LangGraph — Cycles with conditional stop (well-documented pattern)
• CrewAI — guardrail for LLM-based validation, max_iter for retries
• AutoGen — register_nested_chats for inner critic conversations
• Google ADK — LoopAgent with max_iterations and escalate=True to exit
• Anthropic Agent SDK — No dedicated reflection API; Claude naturally self-evaluates and retries through its reasoning loop

5. Router / Dispatch

                              +--------------+
                User -------> |    Router    |
                Query         | (Classifier) |
                              +------+-------+
                                     |
                        +------------+------------+
                        |            |            |
                        v            v            v
                  +----------+  +----------+  +----------+
                  | Billing  |  |Technical |  |  Sales   |
                  |  Agent   |  |  Agent   |  |  Agent   |
                  +----------+  +----------+  +----------+

Not every query needs a specialist. But every query needs the right specialist. We use a lightweight router agent to classify the input once and send it to the best-fit agent. That’s it — the router’s job is done.

How is this different from Orchestrator-Worker? The router is a traffic cop, not a project manager. It makes one decision (which agent should handle this?) and steps aside. It doesn’t break the task into sub-tasks, doesn’t wait for results, doesn’t coordinate between agents, and doesn’t aggregate anything. The Orchestrator does all of that. Use a Router when the task goes to one specialist. Use an Orchestrator when the task needs to be split across multiple specialists.

When to use:

• We receive diverse input types that each need a single specialist
• We want to optimize cost (route simple queries to cheaper/smaller models)
• Customer support, helpdesks, and multi-domain assistants

When NOT to use:

• The task needs to be split across multiple agents — use Orchestrator-Worker instead
• All queries need the same type of processing
• We only have one specialist — just call it directly

Sample Use Case: A SaaS customer support system — the Router classifies incoming tickets as billing, technical, account, or feature-request. A billing query goes entirely to the Billing agent. A technical issue goes entirely to the Technical agent. The router doesn’t coordinate between them — each specialist handles the full request on its own.

Framework Support:

• LangGraph — Conditional edges with routing functions (native)
• CrewAI — @router decorator in Flows, ConditionalTask
• AutoGen — Custom speaker_selection_method in GroupChat
• Google ADK — LlmAgent with sub_agents and routing instructions (no dedicated RouterAgent)
• Anthropic Agent SDK — LLM-driven dispatch via Agent tool descriptions

6. Planning + Execution

                     +-----------+
                +--->|  Planner  |
                |    +-----+-----+
                |          |  Step 1, Step 2, Step 3...
                |          v
                |    +-----------+     +-----------+     +-----------+
                |    | Executor  | --> | Executor  | --> | Executor  |
                |    | (Step 1)  |     | (Step 2)  |     | (Step 3)  |
                |    +-----------+     +-----------+     +-----+-----+
                |                                              |
                |    Replan if needed                          |
                +----------------------------------------------+

We separate the thinking from the doing. A planner agent creates the full roadmap upfront, then executor agents carry out each step. The planner steps back and only re-engages if something fails and we need to replan.

How is this different from Orchestrator-Worker? The Orchestrator is a micromanager — it stays involved at every step, makes decisions between tasks, and coordinates in real-time. Plan + Execute is more like an architect and builders — the architect draws the full blueprint upfront, hands it to builders, and only comes back if the foundation cracks. The planning and execution are clearly separated phases, not interleaved.

When to use:

• Complex, multi-step goals where we benefit from thinking before acting
• We need the ability to replan when intermediate results change the approach
• Research tasks, coding projects, data analysis workflows

When NOT to use:

• The workflow is fixed and well-known — just use a Pipeline
• We need real-time decision-making at every step — use Orchestrator-Worker instead
• The task is too simple to justify a separate planning step

Sample Use Case: An automated research assistant — the Planner agent breaks down “write a market analysis report on EV batteries” into: (1) identify top 5 companies, (2) gather financial data, (3) analyze patents, (4) compare market share, (5) draft report. The planner hands this off and steps back. Executor agents handle each step sequentially. If Step 2 reveals a major company we missed, the Planner re-engages, revises the plan, and executors continue with the updated steps.

Framework Support:

• LangGraph — Plan-and-execute pattern (well-documented with tutorials)
• CrewAI — Task context dependencies + Flows for planning
• AutoGen — Planner + executor via two-agent or GroupChat (manual setup)
• Google ADK — Compose with SequentialAgent + LoopAgent for replanning
• Anthropic Agent SDK — No dedicated planning API; Claude naturally plans and executes through its reasoning loop

7. Handoff

    +----------+  "I can't handle this"  +------------+
    |  Agent A |------------------------>|  Agent B   |
    | (General)|   + context transfer    |(Specialist)|
    +----------+                         +------+-----+
                                                |
                           "Needs escalation"   |
                                                v
                                           +----------+
                                           |  Agent C |
                                           |  (Human) |
                                           +----------+

Sometimes an agent knows it’s out of its depth. Instead of hallucinating an answer, it hands off the conversation — with full context — to a more capable agent or a human.

When to use:

• Customer service with escalation tiers
• Multi-domain assistants where scope changes mid-conversation
• Human-in-the-loop workflows where some cases need manual review

When NOT to use:

• We can handle everything with a single agent
• We prefer routing upfront rather than mid-conversation handoffs

Sample Use Case: A healthcare triage bot — Agent A handles general wellness questions. When it detects symptoms that need medical advice, it hands off to Agent B (a medically-trained agent with access to clinical guidelines). If Agent B determines the case is urgent, it escalates to Agent C — a human doctor — with the full conversation history attached.

Framework Support:

• LangGraph — Conditional edges with full state transfer between nodes
• CrewAI — allow_delegation=True for autonomous agent delegation
• AutoGen — register_hand_off in Swarm pattern (v0.4+)
• Google ADK — transfer_to_agent with session state context
• Anthropic Agent SDK — Subagents return to parent (no direct peer handoff)

8. Evaluator-Optimizer Loop

            +-------------+     Candidates     +-------------+
            |  Generator  |------------------->|  Evaluator  |
            | (N outputs) |                    |  (Score)    |
            +------+------+                    +------+------+
                   ^                                  |
                   |        Feedback + Best           |
                   +----------------------------------+
                          Iterate until optimal

We generate multiple candidates, score them, and use the feedback to generate better ones. It’s evolution in action — each iteration gets us closer to optimal.

When to use:

• We can define a clear scoring function or evaluation criteria
• Prompt optimization, query refinement, search ranking
• Code optimization where we can measure performance

When NOT to use:

• We don’t have clear evaluation criteria
• We need a single quick answer, not an optimized one
• The generation cost is too high to produce multiple candidates

Sample Use Case: An ad copy optimizer — the Generator creates 10 variations of ad copy. The Evaluator agent scores each on readability, brand alignment, emotional appeal, and CTA strength. The top 3 go back to the Generator with feedback. We repeat for 3 rounds and pick the winner.

Framework Support:

• LangGraph — Cycles with scoring nodes and conditional exit
• CrewAI — Task output validation functions + guardrail + max_iter
• AutoGen — register_nested_chats for evaluation before reply
• Google ADK — LoopAgent with evaluator LlmAgent (custom composition)
• Anthropic Agent SDK — No dedicated scoring API; implement scoring as a custom tool and Claude iterates through its reasoning loop

Bonus: Emerging Patterns

These patterns are powerful in specific scenarios but less commonly used in production today. They’re worth knowing as our multi-agent systems mature.

Debate / Adversarial:

                  +------------+           +------------+
                  |  Proposer  | <-------> |   Critic   |
                  |   (For)    |  Debate   |  (Against) |
                  +------+-----+  Rounds   +------+-----+
                         |                        |
                         +------------+-----------+
                                      |
                                      v
                               +------------+
                               |   Judge    |
                               | (Decision) |
                               +------------+

Sometimes the best way to find the truth is to argue about it. We set up agents on opposing sides and let them challenge each other. A judge agent listens to both and makes the final call. Best for high-stakes decisions, fact verification, and red-teaming.

Sample Use Case: An investment analysis system — a Bull agent argues for, a Bear agent argues against. A Judge agent weighs both sides and generates a balanced recommendation.

Framework Support:

• AutoGen — Two-agent initiate_chat (built-in, natural fit)
• LangGraph — Cyclic graphs with conditional edges
• Google ADK — LoopAgent with two LlmAgent sub-agents

Multi-Agent Group Chat:

              +----------+    +----------+    +----------+
              | Product  |    | Engineer |    | Designer |
              | Manager  |    |          |    |          |
              +----+-----+    +----+-----+    +----+-----+
                   |               |               |
                   +---------------+---------------+
                                   |
                                   v
                        +--------------------+
                        |  Shared Context    |
                        |  (Conversation)    |
                        +--------------------+

Multiple agents sit in a shared conversation, each contributing from their expertise. Best for brainstorming and simulating cross-functional team discussions. Expensive — every agent reads every message.

Sample Use Case: Designing a new API — PM, Engineer, Security, and QA agents iterate in shared conversation until we have a complete spec.

Framework Support:

• AutoGen — GroupChat + GroupChatManager (native, best support)
• CrewAI — Crew shares task outputs between agents (not true group chat)
• Anthropic Agent SDK — Not supported (subagents run in isolated contexts)

Mixture of Agents (MoA):

      Layer 1 (Generate)      Layer 2 (Refine)      Layer 3 (Final)

          +--------+
          | LLM A  |--+
          +--------+  |          +--------+
          +--------+  +--------->| LLM D  |--+
          | LLM B  |--+          +--------+  |       +------------+
          +--------+  |          +--------+  +------>| Aggregator |
          +--------+  +--------->| LLM E  |--+       +------------+
          | LLM C  |--+          +--------+
          +--------+

We use multiple models to generate diverse responses, then refine and aggregate. Different models have different strengths — together, they’re better than any one alone. Best for maximum accuracy when cost and latency aren’t constraints.

Sample Use Case: A legal contract review — Claude analyzes clause risks, GPT-4 checks regulatory compliance, Gemini cross-references case law. An aggregator synthesizes findings into a unified risk report.

Framework Support:

• LangGraph — Parallel nodes with different model configs per node
• CrewAI — Each agent configurable with different llm parameter
• AutoGen — Each agent in GroupChat can use different llm_config
• Google ADK — Model-agnostic; each LlmAgent can use Gemini, Claude, GPT via LiteLLM
• Anthropic Agent SDK — Claude-only (no multi-provider support)

Quick Reference: Pick Your Pattern

Core Patterns:

Emerging Patterns:

Combining Patterns: How Production Systems Actually Work

In the real world, we rarely use a single pattern in isolation. Most production systems combine two or three patterns together. Here are some common combinations we see:

Router + Reflection We route incoming queries to specialists, and each specialist uses a reflection loop to ensure quality before responding.

Orchestrator + Parallel + Reflection The orchestrator decomposes a task, fans out to workers in parallel, each worker reflects on its output, and the orchestrator aggregates the results.

Plan + Execute + Debate The planner creates a strategy, executors carry it out, and at critical decision points, a debate between agents validates the approach before we proceed.

Router + Handoff + Human-in-the-Loop The router classifies the query, specialist agents handle it, and if confidence is low, we hand off to a human — with full context preserved.

Final Thoughts

Multi-agent design patterns aren’t just academic concepts — they’re practical tools we can reach for when a single agent isn’t enough. The key is to start simple and add complexity only when we need it.

Here’s a simple decision framework we can follow:

1. Can one agent handle it? → Don’t use multi-agent. Keep it simple.
2. Do we need specialization? → Orchestrator-Worker or Router.
3. Do we need speed? → Parallel Fan-Out.
4. Do we need accuracy? → Reflection, Debate, or Mixture of Agents.
5. Do we need adaptability? → Plan + Execute.
6. Do we need graceful escalation? → Handoff.

The agentic era is just getting started, and these patterns will evolve as we build more sophisticated systems. But the fundamentals — decomposition, specialization, coordination, and feedback — these are timeless engineering principles wearing a new hat.

Now let’s go build something.

References

Framework Documentation

• LangGraph — https://langchain-ai.github.io/langgraph/
• CrewAI — https://docs.crewai.com/
• AutoGen — https://microsoft.github.io/autogen/
• Google ADK — https://google.github.io/adk-docs/
• Anthropic Agent SDK — https://platform.claude.com/docs/en/agent-sdk/overview

Pattern Deep Dives

• LangGraph Multi-Agent Patterns — https://langchain-ai.github.io/langgraph/concepts/multi_agent/
• Google ADK Multi-Agent Systems — https://google.github.io/adk-docs/agents/multi-agents/
• Google Developers Blog: Multi-Agent Patterns in ADK — https://developers.googleblog.com/en/developers-guide-to-multi-agent-patterns-in-adk/
• AutoGen Conversation Patterns — https://microsoft.github.io/autogen/0.2/docs/tutorial/conversation-patterns/
• CrewAI Processes (Sequential & Hierarchical) — https://docs.crewai.com/concepts/processes
• Anthropic: Building Effective Agents — https://www.anthropic.com/research/building-effective-agents

Research Papers & Articles

• “Mixture-of-Agents Enhances Large Language Model Capabilities” (2024) — https://arxiv.org/abs/2406.04692
• “Plan-and-Solve Prompting” — https://arxiv.org/abs/2305.04091
• “Reflexion: Language Agents with Verbal Reinforcement Learning” — https://arxiv.org/abs/2303.11366
• “CAMEL: Communicative Agents for Mind Exploration” — https://arxiv.org/abs/2303.17760

Workflow Agent Types (Google ADK)

• SequentialAgent — https://google.github.io/adk-docs/agents/workflow-agents/sequential-agents/
• ParallelAgent — https://google.github.io/adk-docs/agents/workflow-agents/parallel-agents/
• LoopAgent — https://google.github.io/adk-docs/agents/workflow-agents/loop-agents/

Drawing inspiration from NotebookLM and OpenAI’s Ghibli-style art, this guide explores how to create a captivating podcast on any topic, infused with the charm and aesthetics of the Ghibli style.

Let us dive into creating engaging and interactive content for a solo podcast using Generative AI. Once we select an interesting topic for which we would like to create a podcast, we can leverage various AI tools to generate compelling podcast content for our topic.

The following steps will demonstrate how to do this:

1. Choose a Podcast Topic
2. Set Up the Notebook
3. Gather Content
4. Data Cleaning and Chunking for LLMs
5. Generate the Podcast Script
6. Add Voice to the Podcast Script
7. Generate an Image for the podcast
8. Bring the Podcast Together

1. Choose a Podcast Topic

The first step in creating a podcast is selecting a topic and gathering content to serve as its foundation. In this case, we’ll use the Wikipedia data to generate the text corpus. However, there are various other ways to source content.

For this tutorial, as a sports enthusiast, I have chosen the IPL (Indian Premier League) as a topic for my podcast, as it is currently ongoing and a major event in my country. However, you can choose any topic that interests you.

2. Set Up the Notebook

In this tutorial:

• Code:
  To code, we’ll be using a Google Colab notebook, but you can use any Jupyter notebook environment.
• Content:
  To generate the podcast content, we will be using the OpenAI GPT-4o mini model. To use this model, we will need an API key. If you’ve already signed up with OpenAI, you can generate a new key from the profile section or use any existing key.

If you haven't signed in yet, you can use the URL below to sign in to OpenAI and then get the key from the profile section.

https://platform.openai.com/docs/overview

• Voiceover:
  To generate a voiceover for the podcast script, we’ll use ElevenLabs with the Multilingual V2 model. Like OpenAI, ElevenLabs requires an API key to access its models. If you’re already signed up, you can generate a new key from the profile section or use an existing one.

To set up the notebook, perform the following steps:

Step 2.1: Open a new notebook and write a brief description of the activity or the topic of your podcast.

Step 2.2: Once done, connect to the runtime.
Note: In Google Colab, a “runtime” is an environment where your code runs.

Step 2.3: Run the following code blocks in the notebook to store the API keys of Open API and Eleven Labs, which we will use later.

from getpass import getpass

api_key = getpass('Enter your OPENAI_API_KEY')

eleven_labs_api_key = getpass('Enter your ELEVEN_LABS_API_KEY')

3. Gather Content

Wikipedia is a good starting point for our text corpus as it is a free online encyclopedia and covers an extensive range of subjects such as history, science, current events, celebrities, and much more. Using the Wikipedia library, we can get all the details of a particular Wiki page based on the title or the Wiki ID.

Note: Wikipedia is a useful tool for general information, but it should not be relied upon as the sole or primary source for in-depth, high-stakes research.

To proceed with this tutorial, we need to install several libraries in the Colab notebook.

Step 3.1: Install LangChain and required libraries

Install all the required libraries using the following command.

!pip install wikipedia tiktoken langchain_community langchain_experimental langchain_openai langgraph elevenlabs pydub --quiet

Step 3.2: Load Wikipedia Document Using LangChain

Once the libraries are successfully installed, to gather content about the IPL (Indian Premier League) from Wikipedia, we will use the LangChain WikipediaLoader. LangChain provides a streamlined way to load, process, and query data from various sources, including Wikipedia.

To load Wikipedia documents using LangChain, you can use the WikipediaLoader from the LangChain library in the following way.

from langchain_community.document_loaders import WikipediaLoader

docs = WikipediaLoader(query="Indian_Premier_League", load_max_docs=3).load()

To read more about the Langchain WikipediaLoader, check the following link.

Wikipedia | 🦜️🔗 LangChain

We can’t provide all the downloaded data to the LLM at the same time due to context length limitations. Therefore, we need to split the data into smaller, logical chunks that can be processed within these constraints. Each LLM model has its own context length, which determines how much text it can handle at a time. So, how do we address this challenge? Let’s explore the solution in the next step.

4. Data Cleaning and Chunking for LLMs

Before training any machine learning model, data cleaning is essential. Similarly, before providing data as input to an LLM, we must perform both data cleaning and chunking.

What is Data Chunking?

Data chunking is a strategy to work around the context window limitation of Large Language Models (LLMs), ensuring that longer texts can be effectively processed. It involves:

• Splitting text into sections that fit within the model’s context window.
• Maintaining inter-chunk dependencies to preserve narrative flow.

Using LangChain’s SemanticChunker

To efficiently split the Wikipedia-derived text corpus, we will use LangChain’s Semantic Text Splitter. This tool ensures that the text is broken down in a way that preserves meaning and coherence. You may need a language model or embeddings to help with semantic splitting. In this case, we will use OpenAI embeddings, but you can use other embeddings as well.

Below is the reference code used to split the documents using SemanticChunker.

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings

text_splitter = SemanticChunker(
    OpenAIEmbeddings(api_key=api_key), breakpoint_threshold_type="gradient"
)
split_docs = text_splitter.split_documents(docs)
len(split_docs)

5. Generate the Podcast Script

To create a compelling podcast script, we need to identify key points from the content that would be engaging for the audience. Once we’ve gathered these interesting discussion points, we should structure them in a format suitable for voiceover narration.

Here, we’ll use LangGraph’s MapReduce pattern to summarize the content in a structured format that can be effectively used for our podcast.

The following code creates the map and reduce chains separately using ChatPromptTemplate, which will later be used in LangGraph.

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = api_key

llm = init_chat_model("gpt-4o-mini", model_provider="openai",temperature = 0.7)

# Map
map_template = """
You are a Sports enthusiast who is doing a research for a podcast. Your task is to extract relevant information from the Result delimited by triple quotes.
Please identify 3 interesting questions and answers which can be used for a podcast discussion.
The identified discussions should be returned in the following format.
- Highlight 1 from the text
- Highlight 2 from the text
- Highlight 3 from the text
Result: {docs}"""

# Reduce
reduce_template ="""
You are a writer creating the script for the another episode of a podcast "Sport 101" hosted by \"Tom\".
Make the podcast casual, engaging and informative
Extract relevant information for the solo podcast from the Result delimited by triple quotes.
Use \"Tom\" as the person speaking and sharing insights about the topic with the audience.
Use the below format for the solo podcast.
1. Introduction about the topic and welcome everyone for another episode of the podcast "Sport 101".
2. Introduce the speaker in brief.
3. Then start the podcast.
4. Start the podcast with some casual discussion like what he is doing right now at this moment.
5. End the podcast with thank you speech to everyone.
6. Do not use the word \"conversation\" in the response.
7. Do not use the word \"Introduction\" in the response.
8. Do not use the word \"Podcast Script\" in response.
Result: "{doc_summaries}"
"""

map_prompt = ChatPromptTemplate([("human", map_template)])
reduce_prompt = ChatPromptTemplate([("human", reduce_template)])

map_chain = map_prompt | llm | StrOutputParser()
reduce_chain = reduce_prompt | llm | StrOutputParser()

Here, the map chain is used to extract interesting facts from the content, while the reduce chain helps generate the podcast script in the desired format for our IPL podcast.

Below is a sample LangGraph implementation using the map and reduce chains created above. The graph includes a node for generating summaries, which is applied to a list of input documents. This output then flows into a second node that generates the final summary.

import operator
from typing import Annotated, List, TypedDict
from langgraph.constants import Send
from langgraph.graph import END, START, StateGraph

# This will be the overall state of the main graph.
# It will contain the input document contents, corresponding
# summaries, and a final summary.
class OverallState(TypedDict):
    # Notice here we use the operator.add
    # This is because we want combine all the summaries we generate
    # from individual nodes back into one list - this is essentially
    # the "reduce" part
    contents: List[str]
    summaries: Annotated[list, operator.add]
    final_summary: str


# This will be the state of the node that we will "map" all
# documents to in order to generate summaries
class SummaryState(TypedDict):
    content: str


# Here we generate a summary, given a document
async def generate_summary(state: SummaryState):
    response = await map_chain.ainvoke(state["content"])
    return {"summaries": [response]}


# Here we define the logic to map out over the documents
# We will use this an edge in the graph
def map_summaries(state: OverallState):
    # We will return a list of `Send` objects
    # Each `Send` object consists of the name of a node in the graph
    # as well as the state to send to that node
    return [
        Send("generate_summary", {"content": content}) for content in state["contents"]
    ]

# Here we will generate the final summary
async def generate_final_summary(state: OverallState):
    response = await reduce_chain.ainvoke(state["summaries"])
    return {"final_summary": response}


# Construct the graph: here we put everything together to construct our graph
graph = StateGraph(OverallState)
graph.add_node("generate_summary", generate_summary)
graph.add_node("generate_final_summary", generate_final_summary)
graph.add_conditional_edges(START, map_summaries, ["generate_summary"])
graph.add_edge("generate_summary", "generate_final_summary")
graph.add_edge("generate_final_summary", END)
app = graph.compile()

Let’s now call the Graph to get the desired output.

# Call the graph:
async for step in app.astream({"contents": [doc.page_content for doc in split_docs]}):
    print(step)

Our podcast script is now ready! Next, we need to add a voiceover. Let’s see how we can do that using ElevenLabs.

6. Add Voice to the Podcast Script

First, let’s select a voice for our podcast. We can find a list of available voices in the voice section on the ElevenLabs website. Here, we’ll use the ‘Sports Guy’ voice, but feel free to choose any voice that suits your preference.

Note: While ElevenLabs offers some free credits, they are only sufficient for short scripts. For longer scripts, additional credits must be purchased to generate the voiceover.

We will use the helper function below to convert the text to voice using ElevenLabs.

from elevenlabs.client import ElevenLabs

client = ElevenLabs(
  api_key=eleven_labs_api_key,
)

def createPodcast(podcastScript, speakerChoice):
  genPodcast = []
  podcastLines = podcastScript.split('\n\n')
  for line in podcastLines:
    print(line)
    genVoice = client.generate(text=line, voice=speakerChoice, model="eleven_multilingual_v2")
    # Change: Iterate through the generator and write bytes to a bytearray
    audio_bytes = bytearray()
    for chunk in genVoice:
        audio_bytes.extend(chunk)
    genPodcast.append(audio_bytes)  # Append bytearray to genPodcast
  return genPodcast

speakerName1 = "Tom"
speakerChoice1 = "Sports Guy - Excited and fast, giving you that play by play."
genPodcast = createPodcast(podcastScript, speakerChoice1)

Once we execute the script above, an audio file will be generated. We can then manually download it for later use.

Now our podcast is ready with a voiceover. Next, we need to generate an image in Ghibli style for our podcast.

7. Generate an Image for the podcast

This step is simple — choose any podcast image as a reference and use ChatGPT to transform it into a Ghibli-style illustration.

Here is my Ghibli-style podcast image generated using ChatGPT.

Sharing the prompt that we used for the above image transformation.

Convert the image into and anime style using immersive realism similar at 99.99% to Studio Ghiblis style but not the same but similar so it’s not infrigment

8. Bring the Podcast Together

This is the most exciting part of our podcast creation! We have our podcast audio script ready, along with a Ghibli-style podcast image. Now, let’s use Hedra to generate the podcast video.

Hedra Studio is a content creation platform that allows users to generate high-quality videos, images, and audio. It enables video creation using an image and an audio script.

Note: Although Hedra provides some free credits, they might not be sufficient to generate a full podcast video. If we need to create the entire video, we may have to purchase additional credits from Hedra.

The prompt that we used in Hedra: A sports enthusiast speaking on a solo podcast.

Hedra takes some time to generate the video. Once done, we can download and share our podcast video.

Below is the video which got generated using the above prompt.

If you’d like to access the complete notebook, please refer to the repository provided below.

Machine-Learning-with-Python/GenerativeAI/Ghibli_style_podcast.ipynb at main · sumanentc/Machine-Learning-with-Python

That’s all, folks! Looking forward to seeing the creative tasks you accomplish with these LLM advancements.

References

• Free Text To Speech Online with Lifelike AI Voices | ElevenLabs
• Streaming text to speech | ElevenLabs Documentation
• LangGraph: Agent Orchestration Framework for Reliable AI Agents
• How to split text based on semantic similarity | 🦜️🔗 LangChain
• Migrating from MapReduceDocumentsChain | 🦜️🔗 LangChain
• Hedra

The field of natural language processing has been revolutionized by large language models (LLMs), which showcase advanced capabilities and sophisticated solutions. Trained on extensive text datasets, these models excel in tasks like text generation, translation, summarization, and question-answering. Despite their power, LLMs may not always align with specific tasks or domains.

In this tutorial, we will explore how fine-tuning LLMs can significantly improve model performance, reduce training costs, and enable more accurate and context-specific results.

What is LLM Fine-tuning?

Fine-tuning LLM involves the additional training of a pre-existing model, which has previously acquired patterns and features from an extensive dataset, using a smaller, domain-specific dataset. In the context of “LLM Fine-Tuning,” LLM denotes a “Large Language Model,” such as the GPT series by OpenAI. This approach holds significance as training a large language model from the ground up is highly resource-intensive in terms of both computational power and time. Utilizing the existing knowledge embedded in the pre-trained model allows for achieving high performance on specific tasks with substantially reduced data and computational requirements.

Below are some of the key steps involved in LLM Fine-tuning:

1. Select a pre-trained model: For LLM Fine-tuning first step is to carefully select a base pre-trained model that aligns with our desired architecture and functionalities. Pre-trained models are generic purpose models that have been trained on a large corpus of unlabeled data.
2. Gather relevant Dataset: Then we need to gather a dataset that is relevant to our task. The dataset should be labeled or structured in a way that the model can learn from it.
3. Preprocess Dataset: Once the dataset is ready, we need to do some preprocessing for fine-tuning by cleaning it, splitting it into training, validation, and test sets, and ensuring it’s compatible with the model on which we want to fine-tune.
4. Fine-tuning: After selecting a pre-trained model we need to fine tune it on our preprocessed relevant dataset which is more specific to the task at hand. The dataset which we will select might be related to a particular domain or application, allowing the model to adapt and specialize for that context.
5. Task-specific adaptation: During fine-tuning, the model’s parameters are adjusted based on the new dataset, helping it better understand and generate content relevant to the specific task. This process retains the general language knowledge gained during pre-training while tailoring the model to the nuances of the target domain.

Fine-tuning LLMs is commonly used in natural language processing tasks such as sentiment analysis, named entity recognition, summarization, translation, or any other application where understanding context and generating coherent language is crucial. It helps leverage the knowledge encoded in pre-trained models for more specialized and domain-specific tasks.

Fine-tuning methods

Fine-tuning a Large Language Model (LLM) involves a supervised learning process. In this method, a dataset comprising labeled examples is utilized to adjust the model’s weights, enhancing its proficiency in specific tasks. Now, let’s delve into some noteworthy techniques employed in the fine-tuning process.

1. Full Fine Tuning (Instruction fine-tuning): Instruction fine-tuning is a strategy to enhance a model’s performance across various tasks by training it on examples that guide its responses to queries. The choice of the dataset is crucial and tailored to the specific task, such as summarization or translation. This approach, known as full fine-tuning, updates all model weights, creating a new version with improved capabilities. However, it demands sufficient memory and computational resources, similar to pre-training, to handle the storage and processing of gradients, optimizers, and other components during training.
2. Parameter Efficient Fine-Tuning (PEFT) is a form of instruction fine-tuning that is much more efficient than full fine-tuning. Training a language model, especially for full LLM fine-tuning, demands significant computational resources. Memory allocation is not only required for storing the model but also for essential parameters during training, presenting a challenge for simple hardware. PEFT addresses this by updating only a subset of parameters, effectively “freezing” the rest. This reduces the number of trainable parameters, making memory requirements more manageable and preventing catastrophic forgetting. Unlike full fine-tuning, PEFT maintains the original LLM weights, avoiding the loss of previously learned information. This approach proves beneficial for handling storage issues when fine-tuning for multiple tasks. There are various ways of achieving Parameter efficient fine-tuning. Low-Rank Adaptation LoRA & QLoRA are the most widely used and effective.

What is LoRa?

LoRA is an improved finetuning method where instead of finetuning all the weights that constitute the weight matrix of the pre-trained large language model, two smaller matrices that approximate this larger matrix are fine-tuned. These matrices constitute the LoRA adapter. This fine-tuned adapter is then loaded into the pre-trained model and used for inference.

After LoRA fine-tuning for a specific task or use case, the outcome is an unchanged original LLM and the emergence of a considerably smaller “LoRA adapter,” often representing a single-digit percentage of the original LLM size (in MBs rather than GBs).

During inference, the LoRA adapter must be combined with its original LLM. The advantage lies in the ability of many LoRA adapters to reuse the original LLM, thereby reducing overall memory requirements when handling multiple tasks and use cases.

What is Quantized LoRA (QLoRA)?

QLoRA represents a more memory-efficient iteration of LoRA. QLoRA takes LoRA a step further by also quantizing the weights of the LoRA adapters (smaller matrices) to lower precision (e.g., 4-bit instead of 8-bit). This further reduces the memory footprint and storage requirements. In QLoRA, the pre-trained model is loaded into GPU memory with quantized 4-bit weights, in contrast to the 8-bit used in LoRA. Despite this reduction in bit precision, QLoRA maintains a comparable level of effectiveness to LoRA.

In this tutorial, we will use Parameter-efficient fine-tuning with QLoRA.

Now let’s explore how we can fine-tune LLM on a custom dataset using QLoRA on a single GPU.

1. Setting up the NoteBook
2. Install required libraries
3. Loading dataset
4. Create Bitsandbytes configuration
5. Loading the Pre-Trained model
6. Tokenization
7. Test the Model with Zero Shot Inferencing
8. Pre-processing dataset
9. Preparing the model for QLoRA
10. Setup PEFT for Fine-Tuning
11. Train PEFT Adapter
12. Evaluate the Model Qualitatively (Human Evaluation)
13. Evaluate the Model Quantitatively (with ROUGE Metric)

1. Setting up the NoteBook.

While we will utilize a Kaggle notebook for this demonstration, feel free to use any Jupyter notebook environment. Kaggle offers a generous allowance of 30 hours of free GPU usage per week, which is ample for our experimentation. To begin, let’s open a new notebook, establish some headings, and then proceed to connect to the runtime.

Here, we will select the GPU P100 as the ACCELERATOR. Feel free to try other GPU options available in Kaggle or any other environment.

In this tutorial, we will be using HuggingFace libraries to download and train the model. To download models from HuggingFace, we will need an Access Token. If you’ve already signed up with HuggingFace, you can generate a new Access Token from the settings section or use any existing Access Token.

2. Install required libraries

Now, let’s install the necessary libraries for this experiment.

!pip install -q -U bitsandbytes transformers peft accelerate datasets scipy einops evaluate trl rouge_score

Let’s understand the importance of some of these libraries.

• Bitsandbytes: An excellent package that provides a lightweight wrapper around custom CUDA functions that make LLMs go faster — optimizers, matrix multiplication, and quantization. In this tutorial, we’ll be using this library to load our model as efficiently as possible.
• transformers: A library by Hugging Face (🤗) that provides pre-trained models and training utilities for various natural language processing tasks.
• peft: A library by Hugging Face (🤗) that enables parameter-efficient fine-tuning.
• accelerate: Accelerate abstracts exactly and only the boilerplate code related to multi-GPUs/TPU/fp16 and leave the rest of your code unchanged.
• datasets: Another library by Hugging Face (🤗) that provides easy access to a wide range of datasets.
• einops: A library that simplifies tensor operations.

Loading the required libraries

from datasets import load_dataset
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    BitsAndBytesConfig,
    HfArgumentParser,
    AutoTokenizer,
    TrainingArguments,
    Trainer,
    GenerationConfig
)
from tqdm import tqdm
from trl import SFTTrainer
import torch
import time
import pandas as pd
import numpy as np
from huggingface_hub import interpreter_login

interpreter_login()

For this tutorial we are not going to track our training metrics, so let’s disable Weights and Biases. The W&B Platform constitutes a fundamental collection of robust components for monitoring, visualizing data and models, and conveying the results. To deactivate Weights and Biases during the fine-tuning process, set the below environment property.

import os
# disable Weights and Biases
os.environ['WANDB_DISABLED']="true"

If you have an account with Weights and Biases, feel free to enable it and experiment with it.

3. Loading dataset

Numerous datasets are available for fine-tuning the model. In this instance, we will utilize the DialogSum DataSet from HuggingFace for the fine-tuning process. DialogSum is an extensive dialogue summarization dataset, featuring 13,460 dialogues along with manually labeled summaries and topics.

There is no specific reason for selecting this dataset. Feel free to try this experiment with any custom dataset.

Let’s execute the below code to load the above dataset from HuggingFace.

huggingface_dataset_name = "neil-code/dialogsum-test"
dataset = load_dataset(huggingface_dataset_name)

Once the dataset is loaded, we can take a look at it to understand what it contains:

It contains the below fields.

• dialogue: text of the dialogue.
• summary: human-written summary of the dialogue.
• topic: human written topic/one-liner of the dialogue.
• id: unique file id of an example.

4. Create Bitsandbytes configuration

To load the model, we need a configuration class that specifies how we want the quantization to be performed. We’ll be using BitsAndBytesConfig to load our model in 4-bit format. This will reduce memory consumption considerably, at a cost of some accuracy.

compute_dtype = getattr(torch, "float16")
bnb_config = BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_quant_type='nf4',
        bnb_4bit_compute_dtype=compute_dtype,
        bnb_4bit_use_double_quant=False,
    )

5. Loading the Pre-Trained model

Microsoft recently open-sourced the Phi-2, a Small Language Model(SLM) with 2.7 billion parameters. Here, we will use Phi-2 for the fine-tuning process. This language model exhibits remarkable reasoning and language understanding capabilities, achieving state-of-the-art performance among base language models.

Let’s now load Phi-2 using 4-bit quantization from HuggingFace.

model_name='microsoft/phi-2'
device_map = {"": 0}
original_model = AutoModelForCausalLM.from_pretrained(model_name, 
                                                      device_map=device_map,
                                                      quantization_config=bnb_config,
                                                      trust_remote_code=True,
                                                      use_auth_token=True)

The model is loaded in 4-bit using the `BitsAndBytesConfig` from the bitsandbytes library. This is a part of the QLoRA process, which involves quantizing the pre-trained weights of the model to 4-bit and keeping them fixed during fine-tuning.

6. Tokenization

Now, let’s configure the tokenizer, incorporating left-padding to optimize memory usage during training.

tokenizer = AutoTokenizer.from_pretrained(model_name,trust_remote_code=True,padding_side="left",add_eos_token=True,add_bos_token=True,use_fast=False)
tokenizer.pad_token = tokenizer.eos_token

7. Test the Model with Zero Shot Inferencing

We will evaluate the base model that we loaded above using a few sample inputs.

%%time
from transformers import set_seed
seed = 42
set_seed(seed)

index = 10

prompt = dataset['test'][index]['dialogue']
summary = dataset['test'][index]['summary']

formatted_prompt = f"Instruct: Summarize the following conversation.\n{prompt}\nOutput:\n"
res = gen(original_model,formatted_prompt,100,)
#print(res[0])
output = res[0].split('Output:\n')[1]

dash_line = '-'.join('' for x in range(100))
print(dash_line)
print(f'INPUT PROMPT:\n{formatted_prompt}')
print(dash_line)
print(f'BASELINE HUMAN SUMMARY:\n{summary}\n')
print(dash_line)
print(f'MODEL GENERATION - ZERO SHOT:\n{output}')

From the observation above, it’s evident that the model faces challenges in summarizing the dialogue compared to the baseline summary. However, it manages to extract essential information from the text, suggesting the potential for fine-tuning the model for the specific task at hand.

8. Pre-processing dataset

The dataset cannot be directly employed for fine-tuning. It is essential to format the prompt in a way that the model can comprehend. Referring to the HuggingFace model documentation, it is evident that a prompt needs to be generated using dialogue and summary in the specified format below.

We’ll create some helper functions to format our input dataset, ensuring its suitability for the fine-tuning process. Here, we need to convert the dialog-summary (prompt-response) pairs into explicit instructions for the LLM.

def create_prompt_formats(sample):
    """
    Format various fields of the sample ('instruction','output')
    Then concatenate them using two newline characters 
    :param sample: Sample dictionnary
    """
    INTRO_BLURB = "Below is an instruction that describes a task. Write a response that appropriately completes the request."
    INSTRUCTION_KEY = "### Instruct: Summarize the below conversation."
    RESPONSE_KEY = "### Output:"
    END_KEY = "### End"
    
    blurb = f"\n{INTRO_BLURB}"
    instruction = f"{INSTRUCTION_KEY}"
    input_context = f"{sample['dialogue']}" if sample["dialogue"] else None
    response = f"{RESPONSE_KEY}\n{sample['summary']}"
    end = f"{END_KEY}"
    
    parts = [part for part in [blurb, instruction, input_context, response, end] if part]

    formatted_prompt = "\n\n".join(parts)
    sample["text"] = formatted_prompt

    return sample

The above function can be used to convert our input into prompt format.

Now, we will use our model tokenizer to process these prompts into tokenized ones.

Our aim here is to generate input sequences with consistent lengths, which is beneficial for fine-tuning the language model by optimizing efficiency and minimizing computational overhead. It is essential to ensure that these sequences do not surpass the model’s maximum token limit.

from functools import partial

# SOURCE https://github.com/databrickslabs/dolly/blob/master/training/trainer.py
def get_max_length(model):
    conf = model.config
    max_length = None
    for length_setting in ["n_positions", "max_position_embeddings", "seq_length"]:
        max_length = getattr(model.config, length_setting, None)
        if max_length:
            print(f"Found max lenth: {max_length}")
            break
    if not max_length:
        max_length = 1024
        print(f"Using default max length: {max_length}")
    return max_length


def preprocess_batch(batch, tokenizer, max_length):
    """
    Tokenizing a batch
    """
    return tokenizer(
        batch["text"],
        max_length=max_length,
        truncation=True,
    )

# SOURCE https://github.com/databrickslabs/dolly/blob/master/training/trainer.py
def preprocess_dataset(tokenizer: AutoTokenizer, max_length: int,seed, dataset):
    """Format & tokenize it so it is ready for training
    :param tokenizer (AutoTokenizer): Model Tokenizer
    :param max_length (int): Maximum number of tokens to emit from tokenizer
    """
    
    # Add prompt to each sample
    print("Preprocessing dataset...")
    dataset = dataset.map(create_prompt_formats)#, batched=True)
    
    # Apply preprocessing to each batch of the dataset & and remove 'instruction', 'context', 'response', 'category' fields
    _preprocessing_function = partial(preprocess_batch, max_length=max_length, tokenizer=tokenizer)
    dataset = dataset.map(
        _preprocessing_function,
        batched=True,
        remove_columns=['id', 'topic', 'dialogue', 'summary'],
    )

    # Filter out samples that have input_ids exceeding max_length
    dataset = dataset.filter(lambda sample: len(sample["input_ids"]) < max_length)
    
    # Shuffle dataset
    dataset = dataset.shuffle(seed=seed)

    return dataset

By utilizing these functions, our dataset will be prepared for the fine-tuning process!

## Pre-process dataset
max_length = get_max_length(original_model)
print(max_length)

train_dataset = preprocess_dataset(tokenizer, max_length,seed, dataset['train'])
eval_dataset = preprocess_dataset(tokenizer, max_length,seed, dataset['validation'])

9. Preparing the model for QLoRA

# 2 - Using the prepare_model_for_kbit_training method from PEFT
# Preparing the Model for QLoRA
original_model = prepare_model_for_kbit_training(original_model)

Here, the model is prepared for QLoRA training using the `prepare_model_for_kbit_training()` function. This function initializes the model for QLoRA by setting up the necessary configurations.

10. Setup PEFT for Fine-Tuning

Let us now define the LoRA config for Fine-tuning the base model.

from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training

config = LoraConfig(
    r=32, #Rank
    lora_alpha=32,
    target_modules=[
        'q_proj',
        'k_proj',
        'v_proj',
        'dense'
    ],
    bias="none",
    lora_dropout=0.05,  # Conventional
    task_type="CAUSAL_LM",
)

# 1 - Enabling gradient checkpointing to reduce memory usage during fine-tuning
original_model.gradient_checkpointing_enable()

peft_model = get_peft_model(original_model, config)

Note the rank (r) hyper-parameter, which defines the rank/dimension of the adapter to be trained. r is the rank of the low-rank matrix used in the adapters, which thus controls the number of parameters trained. A higher rank will allow for more expressivity, but there is a compute tradeoff.

alpha here is the scaling factor for the learned weights. The weight matrix is scaled by alpha/r, and thus a higher value for alpha assigns more weight to the LoRA activations.

Once everything is set up and the PEFT is prepared, we can use the print_trainable_parameters() helper function to see how many trainable parameters are in the model.

print(print_number_of_trainable_model_parameters(peft_model))

11. Train PEFT Adapter

Define training arguments and create Trainer instance.

output_dir = f'./peft-dialogue-summary-training-{str(int(time.time()))}'
import transformers

peft_training_args = TrainingArguments(
    output_dir = output_dir,
    warmup_steps=1,
    per_device_train_batch_size=1,
    gradient_accumulation_steps=4,
    max_steps=1000,
    learning_rate=2e-4,
    optim="paged_adamw_8bit",
    logging_steps=25,
    logging_dir="./logs",
    save_strategy="steps",
    save_steps=25,
    evaluation_strategy="steps",
    eval_steps=25,
    do_eval=True,
    gradient_checkpointing=True,
    report_to="none",
    overwrite_output_dir = 'True',
    group_by_length=True,
)

peft_model.config.use_cache = False

peft_trainer = transformers.Trainer(
    model=peft_model,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
    args=peft_training_args,
    data_collator=transformers.DataCollatorForLanguageModeling(tokenizer, mlm=False),
)

Here, we have used 1000 training steps. It seems to be good enough for our custom dataset. We need to try out different numbers before finalizing with training steps. Also, the hyperparameters used above might vary depending on the dataset/model we are trying to fine-tune. This is just to show the capability of fine-tuning.

Let’s start the training now. Training the model will take some time depending upon the hyperparameters used in TrainingArguments.

peft_trainer.train()

Once the model is trained successfully, we can use it for inference. Let’s now prepare the inference model by adding an adapter to the original Phi-2 model. Here, we are setting is_trainable=False because the plan is only to perform inference with this PEFT model.

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM

base_model_id = "microsoft/phi-2"
base_model = AutoModelForCausalLM.from_pretrained(base_model_id, 
                                                      device_map='auto',
                                                      quantization_config=bnb_config,
                                                      trust_remote_code=True,
                                                      use_auth_token=True)

eval_tokenizer = AutoTokenizer.from_pretrained(base_model_id, add_bos_token=True, trust_remote_code=True, use_fast=False)
eval_tokenizer.pad_token = eval_tokenizer.eos_token

from peft import PeftModel

ft_model = PeftModel.from_pretrained(base_model, "/kaggle/working/peft-dialogue-summary-training-1705417060/checkpoint-1000",torch_dtype=torch.float16,is_trainable=False)

Fine-tuning is often an iterative process. Based on the validation and test sets results, we may need to make further adjustments to the model’s architecture, hyperparameters, or training data to improve its performance. Let’s now see how to evaluate the results of Fine-tuned LLM.

12. Evaluate the Model Qualitatively (Human Evaluation)

Now, let’s perform inference using the same input but with the PEFT model, as we did previously in step 7 with the original model.

%%time
from transformers import set_seed
set_seed(seed)

index = 5
dialogue = dataset['test'][index]['dialogue']
summary = dataset['test'][index]['summary']

prompt = f"Instruct: Summarize the following conversation.\n{dialogue}\nOutput:\n"

peft_model_res = gen(ft_model,prompt,100,)
peft_model_output = peft_model_res[0].split('Output:\n')[1]
#print(peft_model_output)
prefix, success, result = peft_model_output.partition('###')

dash_line = '-'.join('' for x in range(100))
print(dash_line)
print(f'INPUT PROMPT:\n{prompt}')
print(dash_line)
print(f'BASELINE HUMAN SUMMARY:\n{summary}\n')
print(dash_line)
print(f'PEFT MODEL:\n{prefix}')

13. Evaluate the Model Quantitatively (with ROUGE Metric)

ROUGE, or Recall-Oriented Understudy for Gisting Evaluation, is a set of metrics and a software package used for evaluating automatic summarization and machine translation software in natural language processing. The metrics compare an automatically produced summary or translation against a reference or a set of references (human-produced) summary or translation.

Let’s now use the ROUGE metric to quantify the validity of summarizations produced by models. It compares summarizations to a “baseline” summary which is usually created by a human. While it’s not a perfect metric, it does indicate the overall increase in summarization effectiveness that we have accomplished by fine-tuning.

To demonstrate the capability of ROUGE Metric Evaluation we will use some sample inputs to evaluate.

original_model = AutoModelForCausalLM.from_pretrained(base_model_id, 
                                                      device_map='auto',
                                                      quantization_config=bnb_config,
                                                      trust_remote_code=True,
                                                      use_auth_token=True)

import pandas as pd

dialogues = dataset['test'][0:10]['dialogue']
human_baseline_summaries = dataset['test'][0:10]['summary']

original_model_summaries = []
instruct_model_summaries = []
peft_model_summaries = []

for idx, dialogue in enumerate(dialogues):
    human_baseline_text_output = human_baseline_summaries[idx]
    prompt = f"Instruct: Summarize the following conversation.\n{dialogue}\nOutput:\n"
    
    original_model_res = gen(original_model,prompt,100,)
    original_model_text_output = original_model_res[0].split('Output:\n')[1]
    
    peft_model_res = gen(ft_model,prompt,100,)
    peft_model_output = peft_model_res[0].split('Output:\n')[1]
    print(peft_model_output)
    peft_model_text_output, success, result = peft_model_output.partition('###')

    original_model_summaries.append(original_model_text_output)
    peft_model_summaries.append(peft_model_text_output)

zipped_summaries = list(zip(human_baseline_summaries, original_model_summaries, peft_model_summaries))
 
df = pd.DataFrame(zipped_summaries, columns = ['human_baseline_summaries', 'original_model_summaries', 'peft_model_summaries'])
df

import evaluate

rouge = evaluate.load('rouge')

original_model_results = rouge.compute(
    predictions=original_model_summaries,
    references=human_baseline_summaries[0:len(original_model_summaries)],
    use_aggregator=True,
    use_stemmer=True,
)

peft_model_results = rouge.compute(
    predictions=peft_model_summaries,
    references=human_baseline_summaries[0:len(peft_model_summaries)],
    use_aggregator=True,
    use_stemmer=True,
)

print('ORIGINAL MODEL:')
print(original_model_results)
print('PEFT MODEL:')
print(peft_model_results)

print("Absolute percentage improvement of PEFT MODEL over ORIGINAL MODEL")

improvement = (np.array(list(peft_model_results.values())) - np.array(list(original_model_results.values())))
for key, value in zip(peft_model_results.keys(), improvement):
    print(f'{key}: {value*100:.2f}%')

As we can see in the above results, there is a significant improvement in the PEFT model as compared to the original model denoted in terms of percentage.

If you’d like to access the complete notebook, please refer to the repository below.

FineTune Phi-2 on Custom DataSet

Conclusion

Fine-tuning Large Language Models (LLMs) has become essential for enterprises seeking to optimize their operational processes. While the initial training of LLMs imparts a broad language understanding, the fine-tuning process refines these models into specialized tools capable of handling specific topics and providing more accurate results. Tailoring LLMs for distinct tasks, industries, or datasets extends the capabilities of these models, ensuring their relevance and value in a dynamic digital landscape. Looking ahead, ongoing exploration and innovation in LLMs, coupled with refined fine-tuning methodologies, are poised to advance the development of smarter, more efficient, and contextually aware AI systems.

References

• microsoft/phi-2 · Hugging Face
• Fine-tuning large language models (LLMs) in 2024 | SuperAnnotate
• microsoft/phi-2 · How to fine-tune this? + Training code
• Phi-2: The surprising power of small language models
• While fine-tuning a decoder only LLM like LLaMA on chat dataset, what kind of padding should one use?
• ROUGE - a Hugging Face Space by evaluate-metric
• GitHub - TimDettmers/bitsandbytes: Accessible large language models via k-bit quantization for PyTorch.

Fine Tune Large Language Model (LLM) on a Custom Dataset with QLoRA was originally published in The Startup on Medium, where people are continuing the conversation by highlighting and responding to this story.

You must have heard about podcasts. They’re like really cool radio shows that people can listen to on their phones or computers. Lots of people love listening to them because they tell interesting stories, share facts, and make people laugh or learn new things.

In this article, we will see how to use Gen AI to create our own podcast content in a really amazing way. It’s like adding magical powers to the stories and shows we make!

Gen AI (Generative AI):

“Gen AI” is a broader term that refers to a class of AI systems designed to generate content autonomously. This content can include text, images, music, videos, and more. These systems leverage generative algorithms and models to create content that often simulates human creativity and style. Gen AI encompasses a wide range of AI models, including language models like LLMs, as well as models specifically designed for generating other types of content like images or music.

LLM (Large Language Model):

LLMs are a specific subset of Generative AI models that are focused on generating human-like text. These models are trained on massive amounts of text data to understand the patterns, structure, and context of language. LLMs like GPT-3 (Generative Pre-trained Transformer 3) are capable of generating coherent and contextually relevant text based on the input they receive. They can perform tasks such as answering questions, generating stories, providing explanations, and more by predicting the next words or phrases in a given text sequence.

In summary, “Gen AI” is a broader category that includes various AI systems for generating different types of content, while “LLM” specifically refers to large language models that generate text-based content.

Now, let’s explore how we can create the content for an interactive and engaging podcast using AI. You can choose any interesting topic from the internet for your podcast. The upcoming steps will demonstrate how to utilize various AI tools to generate the podcast’s content:

1. Setting Up the Notebook
2. Choose a Podcast Topic
3. Gather Content
4. Preprocessing and Data Chunking
5. Data Summarization
6. Generate the Podcast Script
7. Add Voice to the Podcast Script
8. Generate Podcast Intro Music
9. Bring the Podcast Together

1. Setting Up the Notebook

We’ll be using a Google Colab notebook for this, but you can use any Jupyter notebook environment.

First, let’s open a new notebook and create some headings. After that, we’ll connect to the runtime.

For this tutorial, we will be using the OpenAI GPT-3.5 model. To use this model, we will need an API key. If you’ve already signed up with OpenAI, you can generate a new key from the profile section or use any existing key.

If you have not signed in yet then you can use the below URL to sign-in to OpenAI and then get the key from the profile section.

OpenAI Platform

Copy and save the API key in a secure location. We’ll need to use this key later on.

2. Choose a Podcast Topic

The initial step in creating a podcast is to choose a topic and gather content that will serve as the backbone of our podcast. In this instance, we’ll utilize Wikipedia data to generate the text corpus. However, there are other methods available for generating content for our podcast as well. In our case, we’ll be creating a podcast centered around Lionel Messi, the legendary footballer. Given my fondness for Messi, I’m inclined toward this topic, but you’re free to select any other topic of your preference.

3. Gather Content

Wikipedia is a great source for gathering information about the history, current events, celebrities and so much more! It’s a good starting point for our text corpus. Using the Wikipedia library, we can get all the details of a certain Wiki page based on the title or the Wiki ID.

To proceed with our tutorial, we’ll need to install several libraries in the Colab notebook. Let’s begin by installing the Wikipedia library using the command below.

!pip install wikipedia

Once the library is installed let’s verify if we can extract the content of any page from Wikipedia.

import wikipedia
print (wikipedia.summary("Wikipedia"))

As we can see below the library is successfully installed and we are able to extract content from Wikipedia.

Moving forward, let’s collect some information about Lionel Messi from Wikipedia. As previously mentioned, feel free to choose a different topic if you prefer. When we access Lionel Messi’s Wikipedia page, you’ll notice that the Wiki ID appears at the end of the URI. We’ll utilize this ID to fetch the necessary data.

input = wikipedia.page("Lionel_Messi", auto_suggest=False)
wiki_input = input.content
print (len(wiki_input))

As shown above, the text corpus is substantial, encompassing 131,764 characters. However, we can’t utilize this data as is. The limitation arises from the context length of Large Language Models (LLMs), which determines the extent of text they can comprehend at once. Each LLM model has its own context length, and they process text within this constraint. So, how can we address this challenge? Let’s explore the solution in the next step.

4. Preprocessing and Data Chunking

Before training any machine learning model, it’s important to perform data cleaning. Likewise, before providing data as input to an LLM, we should perform both data cleaning and chunking.

Data chunking is a strategy to work around the Large Language Models (LLMs) context window limitation and ensure that longer texts can be effectively processed by the model. It involves dividing the text into sections that are within the model’s context window size, processing these sections sequentially, and handling any inter-chunk dependencies to maintain the flow of the narrative.

What are tokens in the context of LLM?

In the context of Large Language Models (LLMs) like GPT-3, tokens refer to the individual units that text is broken down into. A token can be as short as a single character or as long as a word. For example:

• The word “apple” is one token.
• The phrase “New York” is two tokens, as it consists of two words.
• The sentence “I love ice cream” is four tokens: [“I”, “love”, “ice”, “cream”].

Large Language Models have limitations on the maximum number of tokens they can process in a single input due to computational constraints. This limitation requires users to be mindful of input length, especially when dealing with lengthy texts.

Here, we will use tiktoken library from OpenAI to overcome the token limitation.

We will use the below command to install this library.

!pip install tiktoken

Let’s check the total number of tokens present in our text corpus.

import tiktoken
enc = tiktoken.encoding_for_model("gpt-3.5-turbo")
print ("Number of tokens in corpus ", len(enc.encode(wiki_input)))

As we can see above, the number of tokens is large for our text corpus. So, we will need to do some processing to divide the corpus into batches before sending it to OpenAI. In this tutorial, we’ll be using the GPT-3.5-turbo model, which has a maximum token length of 4k. This limit is much lower than the token length of our text corpus.

Merely splitting the text corpus based on token length isn’t sufficient, as doing so might lead to the loss of sentence meanings when we send the chunk to ChatGPT. Therefore, our approach involves initially segmenting the text corpus into coherent sections and subsequently evaluating the token length by combining these subsections.

To split our text corpus we will use the library called NLTK. The sent_tokenize function from the NLTK library can help us to tokenize the text into logical sentences.

We’ll create some helper functions to divide our text corpus into smaller chunks making it suitable for processing by the GPT-3.5 model within a single request. These chunks will be grouped into batches, with each chunk limited to a maximum of 2,500 tokens.

import nltk
nltk.download('punkt')
from nltk.tokenize import sent_tokenize
# Return a sentence-tokenized copy of text, using NLTK’s recommended sentence tokenizer (currently PunktSentenceTokenizer for the specified language).
def split_text (input_text):
  split_texts = sent_tokenize(input_text)
  return split_texts

def create_chunks(split_sents, max_token_len=2500):
  current_token_len = 0
  input_chunks = []
  current_chunk = ""
  for sents in split_sents:
    sent_token_len = len(enc.encode(sents))
    if (current_token_len + sent_token_len) > max_token_len:
      input_chunks.append(current_chunk)
      current_chunk = ""
      current_token_len = 0
    current_chunk = current_chunk + sents
    current_token_len = current_token_len + sent_token_len
  if current_chunk != "":
    input_chunks.append(current_chunk)
  return input_chunks

Let’s now use the above function to split our text corpus.

split_sents = split_text(wiki_input)
input_chunks = create_chunks(split_sents, max_token_len=2500)
print(f'Batch size: {len(input_chunks)}')
# Batch size: 13

Now, our total batch size is 13, with approximately 2500 tokens in each batch.

5. Data Summarization

We’ll now construct a prompt for the LLM, using the input text corpus, to generate a summary. This helps us pinpoint the crucial themes and standout points of the selected topic. This summary will serve as a foundation for creating our podcast script. Remember, a simple summary might be too brief and not provide adequate context for an in-depth podcast discussion. Hence, we need to ensure that the summary contains the most important highlights of the topic that we want to be a part of the podcast conversation.

Moving forward, let’s install the Openai package and ensure that it works seamlessly with our API key.

!pip install openai

import openai
from getpass import getpass

openai.api_key = getpass('Enter your OPENAI_API_KEY')

# we can confirm that the API key works by listing all the OpenAI models
# we will be using the gpt-3.5-turbo version for this project
models = openai.Model.list()
for model in models["data"]:
  print (model["root"])

Let’s proceed to instruct ChatGPT to craft a summary from our text corpus. We’ve reached the stage where we’ve segmented our source text into chunks, and now it’s time to collaborate with ChatGPT to generate the podcast’s summary.

Remember, we’re not seeking a typical summary here — our goal is to create a summary that holds enough content for a captivating podcast discussion. Additionally, it’s important that the information is presented in a structured manner, facilitating easy parsing and reuse for the next step.

Let’s generate a sample prompt for the topic we’ve chosen. We’ll attempt to create a general prompt that can be adapted for various topics. Don’t hesitate to experiment with different prompts. The prompt provided below is meant to offer some guidance in summarizing the text corpus for our podcast creation.

topic = "Sports"

instructPrompt = f"""
You are a {topic} enthusiast who is doing a research for a podcast. Your task is to extract relevant information from the Result delimited by triple quotes. 
Please identify 2 interesting questions and answers which can be used for a podcast discussion.
The identified discussions should be returned in the following format.
- Highlight 1 from the text
- Highlight 2 from the text
"""

requestMessages = []
for text in input_chunks:
  requestMessage = instructPrompt + f"Result: ```{text}```"
  requestMessages.append(requestMessage)

Next, we’ll proceed to execute these prompts to summarize our text corpus and gather the results.

chatOutputs = []
for request in requestMessages:
  chatOutput = openai.ChatCompletion.create(model="gpt-3.5-turbo",
                                            messages=[{"role": "system", "content": "You are a helpful assistant."},
                                                      {"role": "user", "content": request}
                                                      ]
                                            )
  chatOutputs.append(chatOutput)

podcastFacts = ""
for chats in chatOutputs:
  podcastFacts = podcastFacts + chats.choices[0].message.content

We got the below summary.

The output above should present a list of significant facts, themes, or highlights from our chosen topic. These are the aspects we’d like to address in the podcast conversation.

6. Generate the Podcast Script

We now possess all the essential facts and highlights that we intend to incorporate into our podcast episode. With this foundation, let’s proceed to create a podcast script using these highlighted points.

For this purpose, let’s formulate a new prompt that will guide ChatGPT in generating the podcast script using the aforementioned highlights.

While large-language models exhibit impressive capabilities even with zero-shot prompting, they can encounter challenges with more complex tasks. In such cases, we need to provide thinking time to LLM models. It might be helpful to provide clear instructions with defined steps. It’s a similar way to how we offer stepwise instructions to a child.

Below, you’ll find the prompt utilized to generate the podcast script. In this prompt, we’ve designated ‘Tom’ as the main host of the podcast and ‘Jerry’ as the co-host.

podcast_name = "Sport 101"
podcastPrompt = f"""
You are a writer creating the script for the another episode of a podcast {podcast_name} hosted by \"Tom\" and \"Jerry\".
Use \"Tom\" as the person asking questions and \"Jerry\" as the person providing interesting insights to those questions.
Always specify speaker name as  \"Tom\" or \"Jerry\" to identify who is speaking.
Make the convesation casual and interesting.
Extract relevant information for the podcast conversation from the Result delimited by triple quotes.
Use the below format for the podcast conversation.
1. Introduction about the topic and welcome everyone for another episode of the podcast {podcast_name}.
2. Tom is the main host.
2. Introduce both the speakers in brief.
3. Then start the conversation.
4. Start the conversation with some casual discussion like what they are doing right now at this moment.
5. End the conversation with thank you speech to everyone.
6. Do not use the word \"conversation\" response.

"""

requestMessage = podcastPrompt + f"Result: ```{podcastFacts}```"

Let’s now use this prompt to generate the podcast script using the below command.

finalOutput = openai.ChatCompletion.create(model="gpt-3.5-turbo",
                                            messages=[{"role": "system", "content": "You are a helpful assistant."},
                                                      {"role": "user", "content": requestMessage}
                                                      ],
                                           temperature = 0.7
                                            )

Below is the sample podcast script that we received as a response from ChatGPT.

7. Add Voice to the Podcast Script

Our podcast script is now prepared, and all that’s left is to have human voices read this script. To accomplish this, we will employ ElevenLabs for voice generation.

ElevenLabs leverages advanced AI technology to produce top-notch spoken audio in various voices, styles, and languages.

Initially, we’ll need to sign up for ElevenLabs and generate the API key. Upon registration, we’re granted complimentary credits, which are more than sufficient for generating the audio required for our podcast.

To sign up, we can use the URL provided below.

Free AI Voice Generator & Voice Agents Platform | ElevenLabs

Once we have completed the sign-up process, we need to navigate to the profile settings to access the API Key. Make sure to keep this API key secure, as it will be required for future steps.

We can use the below command to install the dependency first.

!pip install elevenlabs

Let’s set the API Key for ElevenLabs next.

from elevenlabs import set_api_key

set_api_key(getpass('Enter your ELEVEN_LABS_API_KEY  '))

In ElevenLabs, we will find a variety of speakers listed in their demo dropdown. They offer a playground feature where we can select different speakers to listen to their voices. Simply choose a speaker and click the play button to hear their voice. For our interactive podcast, we’ll need to select two speakers.

After finalizing our choice of speakers, make a note of their names. We’ll need this information for the next steps.

To facilitate voice generation, we’ll create a helper method. This method assumes that each line of the input text is spoken by a distinct person, and the person’s name is included within the text. It’s important to ensure that the output of our podcast script adheres to this style. If needed, feel free to modify the method accordingly.

from elevenlabs import generate

def createPodcast(podcastScript, speakerName1, speakerChoice1, speakerName2, speakerChoice2):
  genPodcast = []
  podcastLines = podcastScript.split('\n\n')
  podcastLineNumber = 0
  for line in podcastLines:
    if podcastLineNumber % 2 == 0:
      speakerChoice = speakerChoice1
      line = line.replace(speakerName1+":", '')
    else:
      speakerChoice = speakerChoice2
      line = line.replace(speakerName2+":", '')
    genVoice = generate(text=line, voice=speakerChoice, model="eleven_monolingual_v1")
    genPodcast.append(genVoice)
    podcastLineNumber += 1
  return genPodcast

speakerName1 = "Tom"
speakerChoice1 = "Adam"
speakerName2 = "Jerry"
speakerChoice2 = "Domi"
genPodcast = createPodcast(podcastScript, speakerName1, speakerChoice1, speakerName2, speakerChoice2)

When creating the podcast script, we designated ‘Tom’ as the main host and ‘Jerry’ as the cohost. During the voice generation process, we’ll be using the ‘Adam’ voice for ‘Tom’ and the ‘Domi’ voice for ‘Jerry’.

We can use the below method to save our podcast script.

with open("/content/sample_data/genPodcast.mpeg", "wb") as f:
  for pod in genPodcast:
    f.write(pod)

Finally, our podcast audio file is generated and saved as ‘genPodcast.mpeg’.

8. Generate Podcast Intro Music

To give our podcast a professional touch, we can incorporate an introductory music segment. To generate this introduction music, we’ll utilize the MusicGen model, available on HuggingFace.

MusicGen is a text-to-music model with the ability to create high-quality music samples based on text descriptions or audio prompts.

Here’s a sample prompt that we can use to generate the introduction music. Don’t hesitate to experiment with different prompts.

Create a musical composition that evokes a sense of tranquility and relaxation. Incorporate gentle melodies, soft harmonies, and gradual transitions to craft a soothing and smoothening music piece.

Once the audio is generated, download it and import it to the colab notebook.

9. Bring the Podcast Together

At this point, we possess both the podcast audio file and the introduction music file. The next step involves merging these two files to create the complete, final podcast.

To combine these audio files we will use pydub package. It is versatile and allows us to load audio from various formats — MP3 and WAV files in our case and combine them in ways that we like.

We can use the below command to install this dependency.

!pip install pydub

With the dependency now installed, let’s proceed to merge the audio files using the pydub library. You can utilize the following code to combine the audio files and create the final podcast.

from pydub import AudioSegment

# Import your respective saved Audio files
music_intro = AudioSegment.from_file("/content/sample_data/intro_music.mp4","mp4")
podcast_content = AudioSegment.from_file("/content/sample_data/genPodcast.mpeg")

final_version = music_intro.append(podcast_content, crossfade=1500)
#Save your final audio file
final_version.export("/content/finalPodcast.mp3", format="mp3")

Our podcast is now ready. An optional step we can take is to add an image to our final podcast audio script.

Here is our podcast episode which we generated using Gen AI.

If you’d like to access the complete notebook, please refer to the repository provided below.

Machine-Learning-with-Python/GenerativeAI/Generate-Podcast at main · sumanentc/Machine-Learning-with-Python

References

• podcast.ai
• Lionel Messi - Wikipedia
• GitHub - openai/tiktoken: tiktoken is a fast BPE tokeniser for use with OpenAI's models.
• OpenAI Platform
• Few-Shot Prompting | Prompt Engineering Guide
• Introduction - ElevenLabs
• ElevenLabs API Documentation
• MusicGen - a Hugging Face Space by facebook
• GitHub - jiaaro/pydub: Manipulate audio with a simple and easy high level interface
• NLTK

Harnessing GenAI to Craft Podcast Episodes was originally published in The Startup on Medium, where people are continuing the conversation by highlighting and responding to this story.

Autoscaling (sometimes spelled as auto scaling or auto-scaling) is the process of automatically increasing or decreasing the usage of computational resources required for a cloud workload based on the application's need. Scalability is one of the most important aspects to consider for modern container based application deployments. Autoscaling has become an integral component in almost all the cloud platforms, microservices aka containers/pods are not an exemption to this. Microservices in fact are known for flexible and decoupled design, which can best fit in for auto-scaling as they are much easier to create on-demand than virtual machines.

The primary benefit of autoscaling, when configured and managed properly, is that the workload gets exactly the cloud computational resources it requires at any given point in time. We pay only for the server resources we need at that point in time.

Since Kubernetes has become the de facto standard for container orchestration and has been adopted very well across all industries, we need to think about how we can scale these applications deployed on Kubernetes on demand and also to scale it down when not in use. Since most of the cloud providers support Cluster Autoscaler, cluster autoscaler along with container autoscaling enabled becomes a powerful combination to scale any applications with minimal cloud cost.

Cluster Autoscaler

To adjust to changing application demands, such as between the workday and evening or on a weekend, clusters often need a way to automatically scale. Most of the Cloud providers like Azure, AWS, and GCP support the given Kubernetes Autoscalers. Cluster AutoScaler is nothing but the capability provided by Cloud providers to add/remove computing resources (machines -> VM) at runtime. In general Kubernetes clusters scale in and scale out in one of two ways:

• The Cluster Autoscaler watches for pods that can’t be scheduled on nodes because of resource constraints. The cluster then automatically increases the number of nodes. In case there are no pods running on any nodes then it will remove that node automatically.
• The Horizontal Pod Autoscaler uses the Metrics Server in a Kubernetes cluster to monitor the resource demand of pods. If an application needs more resources, the number of pods is automatically increased to meet the demand.

Here, we will not demonstrate the use of the Cluster Autoscaler feature for Node AutoScaling. But in real-world scenarios, we can enable Cluster AutoScaler along with Pod AutoScaling to save on cloud costs. Using KEDA, we can remove the pods when not required, and then the Cluster Autoscaler will remove the node on which no pods are scheduled. This will help us to reduce cloud costs.

What is KEDA?

KEDA is a Kubernetes-based Event Driven Autoscaler. With KEDA, we can drive the scaling of any container in Kubernetes based on the number of events needing to be processed. KEDA handles the triggers to respond to events that occur in other services and scales workloads as needed.

KEDA is a single-purpose and lightweight component that can be added to any Kubernetes cluster. KEDA works alongside standard Kubernetes components like the Horizontal Pod Autoscaler and can extend functionality without overwriting or duplication. With KEDA we can explicitly map the apps we want to scale, with other apps continuing to function. This makes KEDA a flexible and safe option to run alongside any Kubernetes applications or frameworks.

Scaler

KEDA has a wide range of scalers that can detect if a deployment should be activated or deactivated, and feed custom metrics for a specific event source.

ScaledObject

ScaledObject is deployed as a Kubernetes CRD (Custom Resource Definition) which is used to define how KEDA should scale our application and what triggers to use.

In this tutorial, we will see how we can leverage KEDA to autoscale Microservices deployed as containers or pods in the Kubernetes cluster, based on the below events.

1. External event or trigger.
2. HTTP event

Although the example that we will use here is very trivial, it is just for pure demonstration purposes only. Here, we will explore some of the scalers supported by KEDA.

We will follow the below steps to illustrate, how we can leverage KEDA for auto-scaling pods deployed in the Kubernetes cluster.

1. Installing KEDA
2. Deploying sample application
3. Deploying KEDA Event Scaler
4. Testing Auto-Scaling with Event Scaler
5. Deploying HTTP Scaler
6. Testing Auto-Scaling with HTTP Scaler

Prerequisites

• Kubernetes cluster: We will need a running Kubernetes cluster. In case Kubernetes Cluster is not available then we can follow the instructions to Set Up a Kubernetes Cluster.
• Docker Hub Account: An account at Docker Hub for storing Docker images that we will create during this tutorial. Refer to the Docker Hub page for details about creating a new account.
• kubectl: Refer to Install and Set Up kubectl page for details about installing kubectl.
• Helm: Here we will use Helm (Helm installation guide) to deploy KEDA.

Let’s begin.

1. Installing KEDA

Below are the various options which can be used to install KEDA on Kubernetes Cluster.

• Helm charts
• Operator Hub
• YAML declarations

Here, we will use Helm Chart to deploy KEDA.

1.1 Add Keda Core Repo to Helm: First, we will add the kedacore repository to our Helm instance by executing the below command.

helm repo add kedacore https://kedacore.github.io/charts

1.2 Update Helm repo: Then, let’s fetch the information on the new kedacore repo that we added above:

helm repo update

1.3 Install keda Helm chart: Finally, we need to create a namespace for KEDA and install kedacore using Helm chart:

kubectl create namespace keda
helm install keda kedacore/keda --namespace keda

Once the installation is successful, let’s verify whether the keda pods are up and running by executing the below command.

kubectl get deploy,crd -n keda

As we can see above in the image the keda pods are up and running.

2. Deploying Sample Application

It’s time to deploy a demo application. Here, we will use the official latest nginx image at Kubernetes. To deploy the sample application, we will execute the below commands. It creates a ReplicaSet to bring up nginx Pods.

2.1 Creating a Deployment

Deployment provides declarative updates for Pods and ReplicaSets. In Kubernetes, a deployment is a method of launching a pod with containerized applications and ensuring that the necessary number of replicas is always running on the cluster. Let’s create a file nginx-deployment.yaml with the below content for our sample application. The file contains the deployment details for our sample application.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: nginx
  labels:
    app: nginx
spec:
  replicas: 1
  selector:
    matchLabels:
      app: nginx
  template:
    metadata:
      labels:
        app: nginx
    spec:
      containers:
      - name: nginx
        image: nginx:1.14.2
        ports:
        - containerPort: 80 

Now we will create the Deployment using the above file.

kubectl apply -f nginx-deployment.yaml

Let’s check if the deployment was created successfully by executing the below command.

kubectl get deployments

As we can see above in the image our deployment is deployed successfully with one ReplicaSet.

Let’s also check the status of the pod using the below command. Its status should be Running.

kubectl get pods

We can see that 1 instance of our pod is running that we have specified in our deployment configuration.

2.4 Exposing the pods

Now we will be exposing our pods so that they can be accessed on the browser. A NodePort service is the most primitive way to get external traffic directly to our service. NodePort, as the name implies, opens a specific port on all the Nodes (the VMs), and any traffic that is sent to this port is forwarded to the service. We will use the below command to create the NodePort service.

kubectl expose deployment nginx --port=80 --type=NodePort

Once the above command runs successfully, we should be able to list out the services.

kubectl get svc

Now we should be able to access the sample application page using the below command:

curl http://localhost:32511

As we can see in the above image we are getting a valid response from our deployed application.

We now have our Kubernetes Cluster up and running with our sample application deployed. We have also deployed KEDA related configurations. Let’s now see how we can scale our application using KEDA.

3. Deploying KEDA Event Scaler

KEDA integrates with multiple Scalers (event sources) and uses Custom Resources (CRDs) to define the required/desired scaling behavior and parameters. Deployments and StatefulSets are the most common way to scale workloads with KEDA. It allows us to define the Kubernetes Deployment or StatefulSet that we want KEDA to scale based on a scale trigger. KEDA will monitor that service and based on the events that occur it will automatically scale our resource out/in accordingly. Behind the scenes, KEDA monitors the event source and feeds that data to Kubernetes and the HPA (Horizontal Pod Autoscaler) to drive the rapid scale of a resource.

3.1 Scaler

KEDA uses a Scaler to detect if a deployment should be activated or deactivated (scaling) which in turn is fed into a specific event source. Here we will use Cron event scaler to demonstrate Auto Scaling. We will scale out our application for some duration and then later scale it back to normal count. Cron scaler is especially useful when there is a known pattern in the application usage like it’s used during weekdays and not used during weekends.

3.2 ScaledObject

ScaledObject is deployed as a Kubernetes CRD (Custom Resource Definition) which defines the relationship between an event source to a specific workload (i.e. Deployment, StatefulSet) for scaling.

Let’s create a file with the name scaledobjects.yaml with the below content. The file contains the event details that KEDA will monitor to scale our sample application.

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: nginx-deployment
  namespace: default
spec:
  scaleTargetRef:
    apiVersion:    apps/v1        # Optional. Default: apps/v1
    kind:          Deployment     # Optional. Default: Deployment
    name:          nginx      # Mandatory. Must be in the same namespace as the ScaledObject
  pollingInterval:  5                    # Optional. Default: 5 seconds
  cooldownPeriod:   300                  # Optional. Default: 300 seconds
  minReplicaCount:  0                   # Optional. Default: 0
  maxReplicaCount:  5                    # Optional. Default: 100
  fallback:                                        # Optional. Section to specify fallback options
    failureThreshold: 3                            # Mandatory if fallback section is included
    replicas: 1                                    # Mandatory if fallback section is included
  advanced:                                        # Optional. Section to specify advanced options
    restoreToOriginalReplicaCount: true     # Optional. Default: false
    horizontalPodAutoscalerConfig:          # Optional. Section to specify HPA related options
      name: keda-hpa-nginx       # Optional. Default: keda-hpa-{scaled-object-name}
      behavior:                             # Optional. Use to modify HPA's scaling behavior
        scaleDown:
          stabilizationWindowSeconds: 600
          policies:
            - type: Percent
              value: 100
              periodSeconds: 15
  triggers:
  - type: cron
    metadata:
      # Required
      timezone: Asia/Kolkata  # The acceptable values would be a value from the IANA Time Zone Database.
      start: 30 * * * *       # Every hour on the 30th minute
      end: 45 * * * *         # Every hour on the 45th minute
      desiredReplicas: "5"

Here we have used Cron scaler as a trigger event. The above ScaledObject Custom Resource is used to define how KEDA will scale our application and what the triggers are. The minimum replica count is set to 0 and the maximum replica count is 5.

Let’s create the ScaledObject using the below command.

kubectl apply -f scaledobjects.yaml

As we can see, it’s all pretty straightforward. The most important thing is the triggers definition, which will configure what kind of event our application will be scaled in response to.

Cron trigger contains the below parameters.

• timezone - One of the acceptable values from the IANA Time Zone Database. The list of timezones can be found here.
• start - Cron expression indicating the start of the Cron schedule.
• end - Cron expression indicating the end of the Cron schedule.
• desiredReplicas - Number of replicas to which the resource has to be scaled between the start and end of the Cron schedule.

Once the ScaledObject is created, the KEDA controller automatically syncs the configuration and starts watching the deployment nginx created above. KEDA seamlessly creates a HPA (Horizontal Pod Autoscaler) object with the required configuration and scales out the replicas based on the trigger-rule provided through ScaledObject (in this case it is Cron expression).

4. Testing Auto-Scaling with Event Scaler

Once the ScaledObject is created for an application with a relevant trigger, KEDA will start monitoring our application. We can check the status of ScaledObjects using the below command.

kubectl get scaledobjects

As we can see above in the image our ScaledObject is deployed successfully with a minimum replica count of 0 and a maximum replica count of 5 which we have specified in our ScaledObject definition. Once the event is triggered then KEDA will scale out our application automatically. Let’s check the KEDA logs to monitor the events.

We can use the below command to list the KEDA pods.

kubectl get pods -n keda

We can then monitor the logs of keda-operator pod for events.

kubectl logs -f keda-operator-6c99649b58-r5qtl -n keda --tail=100

The above image shows the scale-out event was triggered. Once the scale-out event is triggered, our application will be scaled to run 5 instances and once the scale-in event is triggered it will be scaled back to 0.

Let’s check the count of our application instances after the scale-out event is triggered:

The below image displays our application deployment status after the scale-in event is triggered:

5. Deploying HTTP Scaler

We just saw above, how we can scale our application based on events. Now let’s say we have some application where we cannot generate events, it supports only HTTP traffic. Then how do we scale those applications?

Let’s see that now.

Currently, there are 2 ways by which we can scale our application based on HTTP Traffic using KEDA.

1. Using Prometheus scaler to create scale rules based on metrics around HTTP events.
2. Using KEDA HTTP Add-on.

Here we will explore KEDA HTTP Add-on.

5.1 Install HTTP add-on

The KEDA HTTP Add-on allows Kubernetes users to automatically scale their application up and down (including to/from zero) based on incoming HTTP traffic.

KEDA doesn’t come with an HTTP scaler by default, so we will have to install it separately. We need to execute the below command to install HTTP Add-on on our Kubernetes Cluster.

helm install http-add-on kedacore/keda-add-ons-http --namespace keda

5.2 Defining an Autoscaling Strategy for HTTP Traffic

Now we have KEDA installed along with HTTP Add-on. To keep things simple we will delete the above deployed ScaledObject. We don’t have to do this in production since we will be using different scaling strategies for different applications. Since we are using the same application for demonstration purposes, so we are deleting the above deployed ScaledObject.

To delete the ScaledObject deployed earlier :

Get the deployed ScaledObject name using the below command.

kubectl get scaledobject

Then delete the ScaledObject using the below command.

kubectl delete scaledobject nginx-deployment

Let’s now start defining the new ScaledObject for HTTP Trigger. The KEDA HTTP add-on exposes a CRD where we can describe how our application should be scaled. Let’s create a new file with the name “scaledobjects-http.yaml” with the below contents.

kind: HTTPScaledObject
apiVersion: http.keda.sh/v1alpha1
metadata:
    name: nginx-http-scaledobject
spec:
    host: myhost.com
    targetPendingRequests: 1
    scaleTargetRef:
        deployment: nginx
        service: nginx
        port: 80
    replicas:
        min: 0
        max: 3

The above code defines a ScaledObject of type HTTPScaledObject. It will listen to the requests which come from myhost.com. The minimum replica count is set to 0 and the maximum replica count is 3.

Let’s now deploy the above ScaledObject using the below command.

kubectl apply -f scaledobjects-http.yaml

When we deploy the above ScaledObject, our application will be scaled down to 0. This is because we have set the minimum replica count to 0.

So now, our application is not running. This is to simulate the scenario when our application is not in use. So what happens when someone makes a request to our application? Let’s see that next.

6. Testing Auto-Scaling with HTTP Scaler

A Kubernetes Service called keda-add-ons-http-interceptor-proxy was created automatically when we installed the Helm chart for HTTP-add-on.

For autoscaling to work appropriately, the HTTP traffic must route through the above service first. We can use kubectl port-forward to test this quickly in our local setup.

kubectl port-forward svc/keda-add-ons-http-interceptor-proxy -n keda 32511:8080

Here, connections made to local port 32511 are forwarded to port 8080 of the keda-add-ons-http-interceptor-proxy service that is running in Kubernetes Cluster.

Let’s now make some requests to our application pretending the request comes from myhost.com.

curl localhost:32511 -H 'Host: myhost.com'

If we inspect the pods, we will notice that the deployment was scaled to a single replica. So when we routed the traffic to the KEDA’s service, the interceptor keeps track of the number of pending HTTP requests that haven’t had a reply yet.

The KEDA scaler periodically checks the size of the queue of the interceptor and stores the metrics. The KEDA controller monitors the metrics and increases or decreases the number of replicas as needed. In this case, a single request was pending so the KEDA controller scaled the deployment to a single replica.

As mentioned above, the Service that the add-on creates will be inaccessible over the network from outside of the Kubernetes cluster.

While we can access it via the kubectl port-forward command above, it is not the recommended way of using it in a production setting. Instead, we should use an ingress controller to route traffic to the Keda interceptor service.

Conclusion

In this article, we learned about KEDA and its concepts. We also saw it in action through scaling a sample NGINX application in our Kubernetes cluster. As we can see if we combine Kubernetes pod auto-scaling along with Cluster Autoscaler, then not only can we scale our application on demand but also with minimal cloud cost.

References

KEDA

About cluster autoscaling | Google Kubernetes Engine (GKE) | Google Cloud

• Overview
• Use the cluster autoscaler in Azure Kubernetes Service (AKS) - Azure Kubernetes Service
• KEDA | KEDA Concepts
• KEDA | Deploying KEDA
• Deployments
• KEDA | Announcing KEDA HTTP Add-on v0.1.0
• KEDA | Prometheus
• http-add-on/walkthrough.md at main · kedacore/http-add-on

Auto Scaling Microservices with Kubernetes Event-Driven Autoscaler (KEDA) was originally published in Crux Intelligence on Medium, where people are continuing the conversation by highlighting and responding to this story.

In this world of Microservices, we want to build and ship things faster. When we ship our code, we want to make sure we ship well tested quality code. Hence, testing our code becomes very important. Since testing our code is so important, today we have many tools to help us write efficient tests.

In this article, we will discuss one such tool named Wiremock along with Junit 5, that we can use to simulate other services on which our app depends during integration testing of the Spring Boot application.

Let us start with WireMock.

What is WireMock?

WireMock is a library for stubbing and mocking web services. It constructs an HTTP server that acts as an actual web service. In simple terms, it is a mock server that is highly configurable and has the ability to return recorded HTTP responses for requests matching criteria.

WireMock has a feature called Record and Playback. It can create stub mappings automatically from requests it has received. Combined with its proxying feature this allows us to “record” stub mappings from interaction with existing APIs. When in action, we have the option to either call the external API and capture the response or use one of the earlier recorded stubs to reply to the caller without calling the external API.

Why do we need WireMock?

Suppose our SpringBoot Application calls external applications and this call to external services bears some cost. However, we can’t test the application without calling the external services. In this case, we can use WireMock to mock the external applications while testing the REST API that we are developing in SpringBoot.

WireMock can be used as a library or as a standalone process. In this tutorial, we will use it as a library. Later, we will see how we can use WireMock to mock the external applications while testing the REST API that we are developing using SpringBoot.

What is JUnit 5?

JUnit is a unit testing framework for the Java programming language. JUnit has been important in the development of test-driven development and is one of a family of unit testing frameworks, which is collectively known as xUnit that originated with SUnit.

Unlike previous versions of JUnit, JUnit 5 is composed of several different modules from three different sub-projects.

JUnit 5 = JUnit Platform + JUnit Jupiter + JUnit Vintage

The JUnit Platform serves as a foundation for launching testing frameworks on the JVM. It also defines the TestEngine API for developing a testing framework that runs on the platform. Furthermore, the platform provides a Console Launcher to launch the platform from the command line and the JUnit Platform Suite Engine for running a custom test suite using one or more test engines on the platform.

JUnit Jupiter is the combination of the new programming model and extension model for writing tests and extensions in JUnit 5. The Jupiter sub-project provides a TestEngine for running Jupiter based tests on the platform.

JUnit Vintage provides a TestEngine for running JUnit 3 and JUnit 4 based tests on the platform. It requires JUnit 4.12 or later to be present on the classpath or module path.

Let’s create a simple REST API using Spring Boot and JAVA Config. Then we will perform an integration test of our application using the JUnit 5 and WireMock.

Prerequisites

Before we begin, we must ensure that the following pre-requisites are installed on the local machine:

• Maven
• Java 17

Steps to Build

1. Set up the Spring Boot Project
2. Include Additional Maven Dependencies
3. Furnish the Configurations
4. Define a DTO Class
5. Define the Rest Controller Endpoints
6. Construct the Service Layer
7. Start the Application
8. Test the Application with Swagger-UI
9. Add Integration Test
10. Conclusion

1. Set up the Spring Boot Project

For this tutorial, we are using JDK 17 and Spring Boot 2.7.1 project. Use start.spring.io and create a sample project using the below artifacts:

• Spring Web: Build web, including RESTful, applications using Spring MVC. Uses Apache Tomcat as the default embedded container.

2. Include Additional Maven Dependencies

Add the additional dependencies, shown below, to our project to enable Swagger, Lombok, WireMock, and JUnit for our application.

<dependency>
    <groupId>org.springdoc</groupId>
    <artifactId>springdoc-openapi-ui</artifactId>
    <version>1.6.9</version>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.18.24</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.github.tomakehurst</groupId>
    <artifactId>wiremock-standalone</artifactId>
    <version>2.27.2</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter-engine</artifactId>
    <version>5.8.2</version>
    <scope>test</scope>
</dependency>

Swagger: Swagger is an open-source tool to develop APIs, interact with APIs, and mainly generate documentation for APIs.

Lombok: Java annotation library which helps to reduce boilerplate code.

WireMock and JUnit: for Integration testing.

3. Furnish the Configurations

3.1 Configure RestTemplate

To call an external API we need to create a RestTemplate instance. Let’s start with that. Spring Boot >= 1.4 no longer automatically defines a RestTemplate but instead defines a RestTemplateBuilder allowing us more control over the RestTemplate that gets created. We can inject the RestTemplateBuilder as an argument in our @Bean method to create a RestTemplate:

@Configuration
public class ApplicationConfig {
    @Bean
    public RestTemplate restTemplate(RestTemplateBuilder builder) {
        // Do any additional configuration here
        return builder.build();
    }
}

3.2 Configure application.yml

application.yml file contains all the configuration required by our application during runtime. It also contains externalized configurations. Here, we will just specify our application name and externalize the URL for the dependent service. We will use this URL later to connect to the external service.

spring:
  application:
    name: "Sample Spring Boot Application"

universitiesBaseURL: http://universities.hipolabs.com

3.3 Configuration to read external properties

We can bind the properties defined in the application.yml file to a Java class using @ConfigurationProperties. This annotation allows to map the entire properties and yml files into an object easily. The @ConfigurationProperties annotation takes a prefix parameter, and binds all the properties with the specified prefix to the Java class.

@Configuration
@ConfigurationProperties(prefix = "wiremock-config")
@Data
@Profile("integration")
public class WireMockConfig {
    private List<WireMockProxy> proxies;
}

In this article, we will use the WireMockConfig class to read all the configurations related to WireMock defined in application-integration.yml file. We will be defining this application-integration.yml file when we write the integration tests in Step 9. We will need these configurations during the Integration test.

4. Define a DTO Class

In this application, we will call some external API and then map the JSON response of the API to this DTO UniversityDTO.

@Data
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
public class UniversityDTO {
    private List<String> domains;
    private List<String> web_pages;
    private String name;
    private String country;
    private String alpha_two_code;
}

5. Define the Rest Controller Endpoints

Finally, let us write some APIs that can be exposed to the clients. This API will call some external APIs and return the results to the client. Here, we are using one of the free API listed in Apipheny for demonstration purposes. We will use the Universities List API to get a list of universities in a specified country. We will not be validating the response, as we are just simulating the scenario where in our application we may need to call some APIs, do certain processing on top of that and then return the results.

In the above code from line 22to 23 we are calling the method getUniversitiesForCountry of the service class, for fetching the list of all universities for the given input. Then we return the result to the caller.

Next, let’s see how we are getting the data from the external API.

6. Construct the Service Layer

Let us construct the Service Layer for our application. The service layer is required for encapsulating the business logic and also for centralizing the data access.

We will add a Service class named UniversityService inside the sub-package service of the root package. This class will contain the method getUniversitiesForCountry that will be used to fetch the universities for the given country.

As we can see above, UniversityService does not have much business logic other than calling some external service, and map the response to UniversityDTO and then return the response to the caller.

7. Start the Application

Till now we have written all the code required for our application to run. We will execute the below maven command from the root directory of the project to compile and build the jar.

mvn clean package

Once the build is complete then we can start the application. Here, we will use IntelliJ to run the application.

As we can see above, the application started successfully on port 8080.

8. Test the Application with Swagger-UI

Let’s test our universities endpoint using Swagger-UI. Once the application started successfully, we can see all the endpoints for our application using the Swagger-UI. Open the swagger UI and check the endpoint. The Swagger will open at the below URL.

http://localhost:8080/swagger-ui/index.html

Let’s now test the endpoint for some sample countries.

As we can see above, we got the response for the country India. So our application is up and running. Similarly, it can also be tested for other countries.

9. Add Integration Test

Let’s start writing some Integration tests for this application. During Integration tests, we don’t want to call the actual HTTP endpoint. This is less critical for this example, but imagine a scenario where we manipulate the data within the remote application using HTTP POST or PUT. We don’t want this to happen during our tests and instead talk to a stub.

Also, if we connect to the actual remote system for our integration test, the test outcome depends on the availability of the remote service and produces additional dependency. At the same time, we may also want to test different responses (e.g., for different scenarios and status codes).

We can resolve the above challenges of connecting to the external API during integration testing using WireMock. With WireMock, we can capture the actual response for a given URL and set the body and status code to whatever we need. Internally, WireMock spawns a local Jetty server that handles the remote calls during our integration tests.

9.1 Add the Configuration File

First, we need to add a configuration file name : application-integration.yml inside the “src/main/resources” folder. This file will contain all the configurations, which we need during the Integration test or we want to override during the Integration tests. Here, the term integration in the file name refers to the profile name which we will use during the tests. We have already defined the Java class WireMockConfig in step 3.3 above to read these configurations during runtime.

The file will contain the below configurations:

As we have seen above in step 5 we are calling the Universities List API to get a list of universities for a given country. So during our Integration tests, we want WireMock to listen to this URL and return responses from the stubs if present. Here, we are specifying the Mock Server to run at port : 9081 for the base URL : http://universities.hipolabs.com. Any request made to this URL will be intercepted by this Mock Server and will return a response from the source or stubs depending upon the value of the property : recording .

If recording is true then it will call the source and capture the response and return the response to the caller. If the recording is false then it will look into the mappings folder for the captured stubs. If there is a stub matching the required criteria, then it will return a response from the stub to the caller.

Note: We can see above that we have overridden the universitiesBaseURL to the proxy URL.

Here, we have taken proxies in a list format and not as a single entry, although we are calling only one API in this case. This is because if we have to call multiple external APIs, then we can just append the configuration below the current one and it will work.

9.2 Add Test Directories

We need to create some directories inside the “src/test/resources” folder for WireMock to record files.

As we can see above, we need to add the following directories for the integration test:

• __file (contains response errors and messages JSON)
• mappings (contains request and response JSON) directories that are needed by WireMock for recording purposes
• recorded directory is used to capture the actual response of the REST API which we will use to compare against our Integration test results

Let’s now start writing the Integration test for the REST API which we exposed earlier.

9.3 Add the Test Case

Let’s start with adding a new Integration Test class with the file name : SpringBootWiremockApplicationTests inside “src/test/java” folder.

In the above code:

• Set the active profile: In line 1 we are directing our tests to use the profile: integration by annotating the test class with the @ActiveProfiles annotation. By setting integration profile as the active profile, the tests will read the configurations from the file application-integration.yml which we defined in step 9.1.
• Look up main configuration class: In line 2 @SpringBootTest annotation tells Spring Boot to look for the main configuration class (one with @SpringBootApplication, for instance) and use that to start a Spring application context.
• Configure auto-configuration of MockMVC: We also have added the annotation @AutoConfigureMockMvc in line 4 because we’ll use MockMvc to call the “/api/university “ endpoint for the test. We could have called the service method directly depending on what kind of test we wanted to implement. In this article, we are trying to demonstrate an example of an integration test by calling the REST endpoints directly with MockMvc.
• Start WireMock server and record: From line 24 to 36 we defined a method called startRecording, which will be executed before every test case. It reads the configuration file defined in step 9.1. If recording flag is turned on then it will start the WireMock server in recording mode, else it will start in regular mode.
• Shut down WireMock server and stop recording: From line 39 to 48 we defined a method called stopRecording which will be executed after every test case. If recording flag is turned on then it will stop the recording before shutting down the WireMock server in recording mode or else it will just stop the server in regular mode.
• Provide WireMock server configuration: From line 124 to 133 we have provided some configuration for the WireMock server. It will start intercepting all the requests specified at the wiremock-config.proxies.url. In this tutorial we have specified only one URL : http://universities.hipolabs.com. If the recording is enabled then it will persist the stubs in the mappings folder. While matching against the stubs it will ignore the arrayOrder and extraElements in the response JSON.
• Define integration test case: From line 51 to 63 we defined the integration test case for the REST API “/api/university”. Here we have used the parameterized test case of JUnit 5. MockMvc is used to call the “/api/university “ endpoint. It then saves the response temporarily in JSON format to compare against the expected response. Here, we are using JsonAssert to compare the expected and actual JSON responses. Once the test case is successful then the current file is cleaned up automatically.

9.4 Record the Test Case

Let’s now see how easily we can record the response of external API and use it during the Integration test. To turn on the recording mode we just need to set recording : true in the configuration defined in step 9.1 above.

Now run the test case. When we run the test case, the response of the external API will be captured inside the mappings folder. The expected response of our API will be captured inside the recorded folder.

9.5 Replay the Test Case

Now everything is in place. We have the Integration Test case ready to be executed. We have all the recorded files in place. Let’s now run the Test case with recording: false .

With recording off, WireMock will try to search the mappings folder for one of the matched responses for the given request. In case it is able to find the response then will return that, else, it will throw an exception saying it’s not able to get any stubbed response.

Let’s modify the recorded response manually to see what happens when WireMock is not able to find the matched response.

Let’s say we update the country name in the response to india1 from india . to simulate the scenario when we don’t have the recorded file for a given request.

Now let’s run the test case again with recording: false

As we can see above, when we run our test case after modifying the response, we got the error Request was not matched. This time WireMock is not able to find any recordings. It also displayed the closest match it can find.

Conclusion

In this article, we saw how we can leverage WireMock in Spring Boot applications to record the requests and responses for external API for various scenarios, and then we can use that recorded files during Integration Test.

If you would like to refer to the full code, do check:

GitHub - sumanentc/spring-boot-with-wiremock: The purpose of this project is to demonstrate how we can use Spring Boot along with WireMock and JUnit 5 to perform Integration Testing with mocking external REST API calls.

References & Useful Readings

• JUnit 5 User Guide
• WireMock User Documentation
• Record and Playback
• JUnit 5 User Guide

Testing Spring Boot Application using WireMock and JUnit 5 was originally published in Crux Intelligence on Medium, where people are continuing the conversation by highlighting and responding to this story.

In one of my earlier tutorials, we have seen how we can optimize the performance of a FastAPI application using Async IO. To know more you can refer to Concurrency with FastAPI.

Async IO is suitable when we want to perform small BackgroundTasks or to perform tasks using concurrency and parallelism. However, when it comes to performing heavy background computations or complex tasks, we would ideally not like to run them as a part of the same process. As such, to perform these complex tasks as a separate process, we need specialized tools such as Celery.

Celery is a distributed task queue that simplifies the management of task distribution and processing. Task queues are used as a mechanism to distribute work across threads or machines. A task queue’s input is a unit of work called a task. Dedicated worker processes constantly monitor task queues for new work to perform.

As a task-queueing system, Celery works well with long running processes or small repeatable tasks working in batches. The types of problems Celery handles are common asynchronous tasks.

Celery communicates via messages, usually using a broker to mediate between clients and workers. To initiate a task the client adds a message to the queue, and the broker then delivers that message to a worker. RabbitMQ and Redis are the brokers transports completely supported by Celery. In our example, we will use RabbitMQ as broker transport.

RabbitMQ is the most widely deployed open-source message broker. RabbitMQ is lightweight and easy to deploy on-premises and in the cloud. It supports multiple messaging protocols. RabbitMQ can be deployed in distributed and federated configurations to meet high-scale, high-availability requirements. RabbitMQ is the default broker for Celery so it doesn’t require any additional dependencies or initial configuration, other than the URL location of the broker instance we want to use.

There are several tools available to monitor and inspect Celery clusters. Flower is a real-time web application monitoring and administration tool for Celery.

The following figure shows a simplified diagram of how various components interact. Here, we are using FastAPI as our Celery Client and RabbitMQ as a message broker and Result Backend.

1. The client sends a request to our FastAPI application.
2. FastAPI app sends the task message to the message broker.
3. Celery workers consume the messages from the message broker. After the task finishes, it saves the result to the Result Backend and updates the task status.
4. After sending the task to the message broker, the FastAPI app can also monitor the status of the task from the Result Backend.
5. Flower can also monitor the celery app tasks by processing messages on the message broker.

In this tutorial, we will see how we can integrate Celery into FastAPI application to perform asynchronous tasks without blocking user’s requests. Although the example which we will use here is very trivial, it is just for pure demonstration purposes only.

Some of the use cases where the application of Celery with FastAPI can be useful:

1. Sending out Emails as Background tasks in an app.
2. Processing the uploaded images in the background.
3. Offline training of ML models.
4. Periodic tasks like report generation or web scrapping.

The following steps illustrate, how we can use Celery along with FastAPI to perform asynchronous tasks:

1. Setup and Installation
2. Set up Message Broker
3. Add Celery
4. Add Celery Tasks
5. Add APIRouter
6. Start the Application and Celery Worker Server
7. Test the Application
8. Monitor the Tasks

Prerequisites

We require Python 3.9.2 with Pipenv and Git installed. Pipenv is a package and a virtual environment manager, which uses PIP under the hood. It provides more advanced features like version locking and dependency isolation between projects.

1. Setup and Installation

Once the prerequisites are in place we can begin creating our application.

To begin with our application, create a folder called fastapi-celery-rabbitmq-application in any directory on the disk for our project.

Navigate to the project folder.

Activate Virtual Environment

Once we are inside the project folder, execute the following commands to activate the VirtualEnv.

The virtual environment will now be activated, which will provide the required project isolation and version locking.

Install Dependencies

Next, we will install all the required dependencies using Pipenv present in requirements.txt.

After we execute the above commands, the required dependencies will be installed.

We can see now two files, which have been created inside our project folder, namely, Pipfile and Pipfile.lock.

• Pipfile contains all the names of the dependencies we just installed.
• Pipfile.lock is intended to specify, based on the dependencies present in Pipfile, which specific version of those should be used, avoiding the risks of automatically upgrading dependencies that depend upon each other and breaking your project dependency tree.

Note: Here, we have installed all the dependencies with specific versions, which worked on my machine while writing this tutorial. If we don’t specify any version then the latest version of that dependency will be installed, which might not be compatible with other dependencies.

The next step is to set up a Message Broker.

2. Set up Message Broker

Celery requires a solution to send and receive messages; usually, this comes in the form of a separate service called a message broker. There are several choices available, RabbitMQ is one of them.

One of the easiest ways to run RabbitMQ on our local machine is by using Homebrew. Homebrew is a popular package manager for MacOS. RabbitMQ formula is available from Homebrew’s core tap (out of the box).

1. To install the RabbitMQ server using Homebrew just execute the below command:

2. The server can then be started with rabbitmq-server in the foreground or with brew services start rabbitmq to have it run under launched in the background.

3. Once we start the server then we can log in to the server with default credentials(Username: guest, Password: guest) at http://localhost:15672.

For other installation options for RabbitMQ, check the below link:

RabbitMQ Documentation | RabbitMQ

Our message broker is up and running. Let’s configure Celery now.

3. Add Celery

We have already installed all the dependencies required for Celery to work. Let’s now configure it. In the root directory of the project, let’s create a file named main.py .

• In the above code from line 8 to 16 we have created a factory function called create_app(), which can be called any time, and returns a FastAPI app instance for us to use. In this function, we are calling a function called create_celery()to create a Celery app instance. Also, we have integrated the universities routing module, which will be explained further in this tutorial as we go ahead.
• The code from line 23 to 24 is required to run the FastAPI application using Uvicorn. FastAPI is the framework that we have used to build our API, and Uvicorn is the server that we will use to serve the requests.

Celery Configuration

Now start with the celery configuration by adding a file config/celery_config.py.

• The above file contains all the configurations required by Celery to run. From line 14 to 15 we have defined the Broker URL and Result backend. Here, we are using RabbitMQ, which we installed in step 2, as a broker as well as Result Backend.
• From line 17 to 23 we have defined all the Queues, which we will use in our application along with the default celery Queue. Instead of manually configuring routing rules at each task level, here we will use CELERY_TASK_ROUTES to configure the routing rules dynamically.
• We have added a helper method from line 6 to 10 fetch the routes per task at runtime.
• From line 32 to 39 we defined a function to load the celery configuration during startup.

Celery App Instance

Let’s now add one more file config/celery_utils.py that will contain the below code to create the celery app instance.

The above file defines a create_celery factory function that configures and then returns a Celery app instance. It will read all the celery related configurations defined earlier. We also updated some of the default celery configurations. This function is called from main.py while creating the FastAPI app instance.

From line 22 to 32 we defined one more helper function get_task_info to get the status and result of the asynchronous tasks submitted via Celery.

4. Add Celery Tasks

We have now all the Celery-related configurations in place. Let us now add some tasks which we want to execute asynchronously. We have to add a file celery_tasks/tasks.py which will contain all the tasks which we want to execute asynchronously.

In the above file we have defined two tasks get_all_universities_task and get_university_task . get_all_universities_task is used to get the all the universities for the list of countries provided as input. get_university_task is used to get the universities for a given country.

Here, we have used shared_task to make our code reusable, which requires current_app instance inside create_celery method instead of creating a new Celery instance. The @shared_task decorator returns a proxy that always uses the task instance in the current_app. It lets us create tasks that can be used by any app(s). Since Celery has built-in support for retrying the failed tasks, we have enabled it for our tasks. With retry option, we can now let the exception bubble up and let the decorator handle it.

5. Add APIRouter

Now, let’s add some entry points to test our application. We will start with a simple Sync entrypoint. This entry point will call some external APIs and return the results to the client. Here, we are using the same free API which I have used in one of my earlier tutorial. It’s listed in Apipheny. We will use the Universities List API to get a list of universities for a specified country. For this tutorial, we are fetching the list of universities provided as input to the API. We will not be validating the response, as we are just simulating the scenario where in our application we are performing some long running tasks to complete the request like we need to call multiple APIs, do certain processing on top of that and then return the results.

Let’s add a dedicated file for handling just universities requests as a module at /routers/universities.py.

We wanted to have the path operations related to universities separated from the rest of the code, to keep the code organized. But it’s still part of the same FastAPI application. We can create the path operations for any module using APIRouter. We have already included this router in our application in step 3 . Here we have added 4 different endpoints to test our application.

• get_universites : From line 14 to 22 we have defined a simple API which takes a list of countries as an input, and then it calls some external API to get the result. The result of each country is combined and then returned to the caller. This API is added just to check the API integration. It has nothing to do with Celery.
• get_universities_async: This API is used to demonstrate how we can use Celery to perform the long running tasks asynchronously. When we call this API, it submits a task message to the broker and returns the taskID for the same. Consumers then can use the other API to check the status of the submitted task or to get the task result. As we can see here, we just need to call the get_all_universities_task method defined earlier in step 4 . Celery task is called just by adding the .apply_async() after the function name. This will tell Celery to add new task to the queue. It internally submits the task to universities Queue. There will be a worker process which will listen to this Queue . The worker process will pickup the message, process the request and save the results in Result Backend.
• get_universities_parallel: This API is used to demonstrate how we can use Celery to split a large task into smaller subtasks and execute in parallel. When we call this API, we create a one task for each country provided as input. Then we group these tasks using Celery Group. Celery group is lazy so we must call it to take action and evaluate the group. Then we execute the group and wait for it to complete. Once all the tasks are completed then the group will be completed and we will get the response. After that we collect the response of the group and return it to the caller. This is a synchronous flow which uses Celery for parallel execution of subtasks.
• get_task_status: This API is used to get the status and result of the asynchronous tasks submitted using one of the API which we described earlier.

Let us have a look at the code which fetches the universities:

In line 5 we are using the HTTP client named httpx to send an HTTP request. HTTPX is a fully-featured HTTP client for Python 3, which provides sync and async APIs, and support for both HTTP/1.1 and HTTP/2.

Here, we are using the sync API to get the universities data. In the above code in line 6 we are calling the universities API to get the results. In line 7 we convert the response to JSON format. Then from line 9 to 11 , we map the JSON object to the Pydantic model University. And finally, we return the dictionary, which contains the country name as key and the list of universities as value.

Now our code is complete to run the FastAPI with Celery.

6. Start the Application and Celery Worker Server

We have already added the code in step 3 required to start the application.

FastAPI is the framework that we have used to build our API, and Uvicorn is the server that we will use to serve the requests. We have already installed Uvicorn. That will be our server. We can start the application by executing the below command:

python main.py

Once the application starts successfully, we can navigate to http://localhost:9000/docs. The system will bring up a page that looks something like this. We can see the endpoint in the Swagger-UI which was defined earlier.

Let’s now start the Celery worker server. We need to open a new terminal and go the project root directory. Then we need to install all the dependencies using the below command:

Pipenv install -r requirements.txt

Once all the dependencies are installed then we can execute the below command:

celery -A main.celery worker --loglevel=info -Q universities,university

The Celery worker will listen to the custom Queue : universities, university used in our application.

Everything is up and running now! Let’s test our application.

7. Test the Application

Let us test our universities endpoint with some sample countries to validate the external API integration. Here, we are getting all the universities for “Turkey” and “India”. We are not validating the response or data.

As we can see above, we got the universities list for the requested countries. Here, we are using the Universities List API to get a list of universities for the specified country.

Once we have tested that the external API interaction is working fine, let’s test the asynchronous task execution using Celery. Here, we will try to get the list of universities for the same set of countries provided as input.

Once we submit the request, we just get the task_id as an acknowledgment response. The request is getting processed asynchronously using Celery as a separate process. We can check the status of the task using the other API. With this architecture, we can handle a large number of requests. We can Enqueue the user requests and process them as per resource availability for our application.

Let’s check the status of the asynchronous task submitted above. We can use the below API with task_id as input.

Let’s now test the scenario where we want to split the user request into multiple subtasks, process them in parallel using Celery and then return the results.

This has the same request and response as the first API which we tested earlier. However, this will execute faster because we are getting the universities of each country in parallel using Celery. Here, we got the response in 0.58 sec as compared to 1.67 sec as per the first API.

8. Monitor the tasks

Till now we can see that using Celery we are submitting some tasks to be executed in the background. It would be good if we can see the progress of these using GUI. The easiest way to monitor the Celery app is via Flower. Using it, we can easily check the progress of all the tasks. We already have all the dependencies installed. We just need to start flower. To run the flower server on our local machine execute the below command from the project root directory:

celery -A main.celery flower --port=5555

We can now see our flower monitoring tool running on our local server at http://localhost:5555/ .

In the Dashboard, we can see the summary of all the tasks executed via Celery.

Once the tasks are submitted, we can also see the progress of each task by clicking on the tasks tab on the Flower navigation bar.

For example, we can see the state of each task as it goes from In Progress to Success or Failed. We can also see when the task started, its runtime and its response, and so on.

Conclusion

In this tutorial, we saw how we can integrate Celery with FastAPI application. Using Celery along with FastAPI application will not only improve the overall performance and efficiency but will help in better handling of background jobs for long-running tasks. We also saw if we want to monitor the tasks submitted via Celery then how we can do the same using Flower.

If you would like to refer to the full code, do check:

GitHub - sumanentc/fastapi-celery-rabbitmq-application: Sample FastAPI Application to demonstrate Async architecture with Celery, RabbitMQ and Flower

References & Useful Readings

• Celery - Distributed Task Queue - Celery 5.2.6 documentation
• Groups, Chords, Chains and Callbacks - Celery 2.6.0rc4 documentation
• Flower - Celery monitoring tool - Flower 1.0.1 documentation
• How to use the @shared_task decorator for class based tasks
• RabbitMQ Documentation | RabbitMQ
• RabbitMQ Documentation | RabbitMQ

Async Architecture with FastAPI, Celery, and RabbitMQ was originally published in Crux Intelligence on Medium, where people are continuing the conversation by highlighting and responding to this story.

In one of my earlier tutorials, we have seen how we can create a Python application using FastAPI. To know more you can refer to Building REST APIs using FastAPI. One of the main reasons to choose FastAPI over other frameworks is due to its very high performance, which is on par with NodeJS and Go.

In this tutorial, we will see how we can optimize the performance of our FastAPI application using concurrency and parallelism.

Understanding Concurrency and Parallelism

Concurrency

Concurrency is when two or more tasks can start, run, and complete in overlapping time periods. It doesn’t necessarily mean they’ll be running at the same instant, such as multitasking on a single-core machine.

Threading is a concurrent execution model whereby multiple threads take turns executing tasks. One process can contain multiple threads. It uses pre-emptive multitasking.

Async IO is a style of concurrent programming that is single-threaded, single-process design. It uses co-operative multitasking. We will be covering Async IO in Python in detail as we go ahead.

Parallelism

Parallelism is when tasks literally run at the same time, wherein we can split tasks up into smaller subtasks, which can be processed in parallel using multiple CPUs at the exact same time, such as, on a multicore processor.

Multiprocessing is a means to effect parallelism, and it entails spreading tasks over a computer’s central processing units (CPUs, or cores). The processes all run at the same time, essentially multiprocessing on different processors.

Concurrency and parallelism both relate to “different things happening more or less at the same time”.

Introduction to Async IO

Async IO is a style of concurrent programming introduced in Python 3.4. Asynchronous I/O is a form of input/output processing. While Synchronous I/O waits for an operation to complete before moving on to the next one, Asynchronous I/O permits other processing to continue before the transmission of one operation has finished.

Synchronous I/O is easier to code as it follows all the steps in a program to complete a task before moving on to the next. To simplify asynchronous code and make it as simple as synchronous code, Async IO uses coroutines and futures as there are no callbacks. It uses async and await keywords to manage and execute asynchronous code.

Coroutines are special functions that work similar to Python generators. It is a function that can suspend its execution before reaching return, and it can indirectly pass control to another coroutine for some time. On await, they release the flow of control back to the event loop. A coroutine needs to be scheduled to run on the event loop, once scheduled coroutines are wrapped in Tasks which is a type of Future.

Futures represent the result of a task that may or may not have been executed. This result may be an exception.

With FastAPI we can take the advantage of concurrency that is very commonly used for web development. We don’t have to worry about the Event loop or async/await management. What we need is to just do the following :

• Declare the first path functions as coroutines via async def wherever appropriate. If we do this wrong, FastAPI will still be able to handle it, we just won’t get the performance benefits.
• Declare particular points as awaitable via the await keyword within the coroutines.

We will see this in practice below when we come to the coding part.

In most of the applications, we perform some Input-output (I/O) operations which are slow in nature. To demonstrate the slow nature of Input-output (I/O) operations, in this tutorial we will create a FastAPI application that will internally call some external APIs to get some data. When we call external API we are performing some network (I/O). Then we will see how to utilize Async IO to improve the performance of our application.

The following steps illustrate, how we can use Async IO in FastAPI to improve performance:

1. Setup and Installation
2. Add a Sync EntryPoint
3. Add Middleware
4. Start the Application
5. Test the Application
6. Add an Async EntryPoint
7. Compare the Performance

Prerequisites

We require Python 3.9 with Pipenv and Git installed. Pipenv is a package and a virtual environment manager, which uses PIP under the hood. It provides more advanced features like version locking and dependency isolation between projects.

1. Setup and Installation

Once the prerequisites are in place we can begin creating our application.

To begin with, our application, create a folder called python-sample-fastapi-application in any directory on the disk for our project.

Navigate to the project folder.

b) Activate Virtual Environment

Once we are inside the project folder, execute the following commands to activate the VirtualEnv.

The virtual environment will now be activated, which will provide the required project isolation and version locking.

c) Install Dependencies

Next, install all the required dependencies using Pipenv as shown.

After we execute the above commands, the required dependencies will be installed.

We can see now two files, which have been created inside our project folder, namely, Pipfile and Pipfile.lock.

• Pipfile contains all the names of the dependencies we just installed.
• Pipfile.lock is intended to specify, based on the dependencies present in Pipfile, which specific version of those should be used, avoiding the risks of automatically upgrading dependencies that depend upon each other and breaking your project dependency tree.

Note: Here, we have installed all the dependencies with specific versions, which worked on my machine while writing this tutorial. If we don’t specify any version then the latest version of that dependency will be installed, which might not be compatible with other dependencies.

The next step is to add the Entrypoint for our application.

2. Add a Sync EntryPoint

In the root directory of the project, let’s create a file named main.py . Now, let’s add an entry point. The entry point will call some external APIs and return the results to the client. Here, we are using one of the free API listed in Apipheny for demonstration purposes. We will use the Universities List API to get a list of universities in a specified country. For this tutorial, we are fetching the list of universities in turkey, india and australia. We will not be validating the response, as we are just simulating the scenario where in our application we may need to call multiple APIs, do certain processing on top of that and then return the results.

Let us have a look at the code for adding the non-async endpoint:

In the above code from line 7 to 9 we are calling the method get_all_universities_for_country for fetching the list of all universities from three different countries and storing the result in the dictionary. After collecting the results for all the countries we are returning the results to the caller.

Next, let’s see how we are getting the data. We will add a file named universities.py in the root directory. This file will contain the function get_all_universities_for_country that will be used to fetch the universities for the given country.

Let us have a look at the code which fetches the universities:

In line 5 we are using the HTTP client named httpx to send an HTTP request. HTTPX is a fully-featured HTTP client for Python 3, which provides sync and async APIs, and support for both HTTP/1.1 and HTTP/2.

Here, we are using the sync API to get the universities data. Later on, we will use the async API.

In the above code in line 6 we are calling the universities API to get the results. In line 7 we convert the response to JSON format. Then from line 9 to 11 , we map the JSON object to the Pydantic model University.

And finally, we return the dictionary, which contains the country name as key and the list of universities as value.

Now our sync endpoint is ready. To check the response time of our API let’s add a middleware.

3. Add middleware

A middleware is a function that works with every request before it is processed by any specific path operation and also with every response before returning it.

Add the below middleware code in main.py

As we can see in the above code, the middleware function uses the decorator @app.middleware("http") at the top. It will intercept our entrypoint request and then forward the request to the corresponding path operation. We will use this middleware to calculate the time taken by our entrypoint and return the response time to the caller. Here, we are adding a response header X-Process-Time in line 6 which will contain the response time.

4. Start the Application

Till now we have written all the code required for our application to run. We just need to add the below code to main.py file.

FastAPI is the framework that we have used to build our API, and Uvicorn is the server that we will use to serve the requests. We have already installed Uvicorn. That will be our server. Our application is ready now. We can start the application by executing the below command:

python main.py

Once the application starts successfully, we can navigate to http://localhost:9000/docs. The system will bring up a page that looks something like this. We can see the endpoint in the Swagger-UI which was defined earlier.

5. Test the application

Let us test our universities endpoint and see, how much time does it take to complete the request.

As we can see in the response that the sync endpoint took around 2.0514 sec to get the list of universities for three countries.

6. Add an Async EntryPoint

Let’s now see how we can improve the performance of the REST API using Async IO.

We will add an async equivalent of the above endpoint. There is no functional difference between these two endpoints. It is just a different way of handling the request.

Let us have a look at the code for the async endpoint:

When we define a function with the async keyword, it is a special function called a coroutine. The reason why coroutines are special is that they can be paused internally, allowing the program to execute them in increments via multiple entry points for suspending and resuming execution. This is in contrast to normal functions which only have one entry point for execution.

When we use the await keyword, inside async function, it tells the program that this is a “suspendable point” in the coroutine. It will suspend execution at this point and can perform some other tasks. It will come back later once the results are available.

In the above code in line 7 we can see that, we are using asyncio.gather to run a sequence of awaitable objects (i.e. our coroutines) concurrently. Once we gather the response of all the coroutines, we are returning the response to the caller. The main difference here is that we are not waiting for the response of the first API to finish before calling the next API. It’s like we are calling all the APIs concurrently and waiting for the results.

Next, we will add one more function in universities.py called get_all_universities_for_country_async to get the universities in async mode.

The above function is declared with the async keyword, defining it as a coroutine. In the above code in line 3 we can see that here we are using the async API of httpx . async with httpx.AsyncClient() is the httpx context managed client for making async HTTP calls. Here, GET request is made with the await keyword, telling Python that this is a point where it can suspend execution and do something else. Once we receive the response then the code remains the same as sync call.

7. Compare the Performance

Now our async API is ready to be tested. Let us re-run the application. We can now see two endpoints for our application.

Let us run the async endpoint and compare the results.

As we can see that using Async IO we are able to get the response in 0.58 sec which earlier took around 2 sec

Conclusion

In this tutorial, we saw how we can take advantage of the Async IO style of programming in FastAPI to improve the performance of our REST APIs. If we have a scope in our application to use Async IO, then we should give it a try.

If you would like to refer to the full code, do check:

GitHub - sumanentc/python-sample-FastAPI-application: Sample Python application using FastAPI , Uvicorn and SQLAlchemy

References & Useful Readings

• HTTPX
• Concurrency and async / await - FastAPI
• What is the difference between concurrency and parallelism?
• asyncio - Asynchronous I/O
• PEP 492 -- Coroutines with async and await syntax
• GitHub - Hipo/university-domains-list: University Domains and Names Data List & API
• Free API - Huge List of Public APIs For Testing [No Key] - Apipheny
• Speed Up Your Python Program With Concurrency - Real Python

Concurrency with FastAPI was originally published in Crux Intelligence on Medium, where people are continuing the conversation by highlighting and responding to this story.

Apache Ignite is a memory-centric distributed database, caching, and processing platform for transactional, analytical, and streaming workloads delivering in-memory speeds at the petabyte scale.

Ignite provides an in-memory data store where each node in an Ignite cluster by default stores data in RAM. The data is kept in off-heap storage to ensure low latency and consistent access times. The system is multi-model, with the ability to support structured, semi-structured, and unstructured data.

Apache Ignite offers an elegant query API with the following components:

• Predicate-based Scan Query
  A scan query is a simple search query used to retrieve data from a cache in a distributed manner. When executed without parameters, a scan query returns all entries from the cache. Scan queries return entries that match a predicate, if specified. The predicate is applied to the remote nodes.
• ANSI 99-compliant SQL Query
  Apache Ignite SQL Grid is a distributed data grid where you can execute ANSI SQL-99-compliant SQLs (SELECT, UPDATE, INSERT, MERGE, and DELETE queries) to manipulate a cache.
• Lucene Index-based Text Query
  Lucene is a full-text search library in Java which makes it easy to add search functionality to an application or website. Ignite supports full-text queries based on the Apache Lucene engine.
• Continuous query
  Continuous queries allow registering a remote filter and a local listener for cache updates. Once the cache is updated and passes the filter criteria, then the cache update events will be sent to the node that executed the query, and the local listener will be notified. This helps us to take action in case the cache is updated on Ignite Cluster.

For SQL queries, Ignite supports in-memory indexing, so all the data lookups are extremely fast. If we are caching our data in off-heap memory, then query indexes will also be cached in off-heap memory as well.

In this tutorial, we will learn how to query the data stored in the Cache of Apache Ignite using query API. We will not cover Continuous Query in this tutorial.

We will use the following steps to explore Query-based search in Apache Ignite Cache:

1. Integrating Apache Ignite Server Node With Spring Boot
2. Enabling Caching
3. Query and QueryCursor
4. Scan Queries
5. SQL Queries
6. Text Queries
7. Testing the Application

Prerequisites

For this tutorial, we are using JDK 1.8 and Spring Boot 2.5.6 project with the following dependencies:

spring-boot-starter
spring-boot-starter-web
spring-boot-starter-test

Here, we will use Maven for dependency management. Along with the above dependencies, we will also use springdoc-openapi to automatically generate the OpenAPI 3 specification docs for our API. For that, we simply need to add the springdoc-openapi-ui dependency to our pom.xml :

<dependency>     
  <groupId>org.springdoc</groupId>
  <artifactId>springdoc-openapi-ui</artifactId>
   <version>1.5.2</version>
</dependency>

1. Integrating Apache Ignite Server Node With Spring Boot

The Apache Ignite server nodes act as containers for data and computations. Once interconnected, the server nodes will represent a distributed database (or data grid) that stores the data, participates in queries processing, compute execution, stream processing, and so on.

Ignite provides an implementation of JCache (JSR 107) specification. JCache provides a very simple and powerful API for data access. However, the specification omits any details about data distribution and consistency to allow vendors enough freedom for their own implementations. Here, we will use org.apache.ignite.cache.CacheManager to interact with the Cache.

We will use Spring XML-based configuration for Ignite.

The above codes create a CacheManager instance using the Ignite configuration XML file path specified in igniteConfigPath variable.

We also need to add the following Maven dependencies in our Spring Boot app to enable Ignite-based caching.

Here, we are using GridGain community edition version: 8.8.10. Once we are done with the above configuration, then we can start the Ignite node in embedded mode, that is, in the same JVM where the application is running.

2. Enabling Caching

After we have the Spring Boot app with embedded Ignite node configured, then we need to define our Cache. Here, we will define a sample Player cache, which we will use for querying purposes. We are using XML-based configuration to define our cache.

Once we declare our Player cache, then we have to define it.

The above code does the following:

We started off by defining the Player class with some attributes, which we have declared in our cache configuration. We have used the annotation @QuerySqlField on fields, which we will use for SQL queries. All fields which need to be involved in SQL clauses must have this annotation. We have also used the annotation @QueryTextField on fields that need to be indexed for full-text search using Lucene.

3. Query and QueryCursor

IgniteCache has several query methods, all of which require some subclass of the Query class. When we query IgniteCache it returns the QueryCursor.Query abstract class that represents an abstract paginated query to be executed on the distributed cache. We can also set the page size for the returned cursor via Query.setPageSize(...) method (default is 1024).

QueryCursor class represents the query result set and allows for transparent page-by-page iteration. Whenever we start iterating over the last page, it will automatically request the next page in the background. For cases when pagination is not needed, we can use the QueryCursor.getAll() method, which will fetch the whole query result and store it in a collection. If we use QueryCursor.getAll() method, it will also automatically close the cursor in the background.

To explore the search queries we will create a class called PlayerService. This will contain various methods required for the search operation.

4. Scan Queries

Apache Ignite’s key-value pair API is used to store objects in a cache and retrieve values using keys. It is a query API that lets us query objects using expressions. ScanQuery provides one of the implementations of Query API, which allows us to query the cache in distributed form using some user-defined predicate.

Let us explore the ScanQuery API now.

The above code does the following:

• We started off by creating a public method called scanQuerySearch in line 1 with search text input, which we will use to search in Player cache.
• In line 3, we retrieved an instance of the Player cache from Ignite node.
• Then we created an instance of ScanQuery in line 4, which takes IgniteBiPredicate as an argument. IgniteBiPredicate is a functional interface that accepts two parameters and returns a boolean. It is usually used to filter a collection of objects and can be used in lambda expression also. Here, we are going to use a Java 8 lambda expression to represent IgniteBiPredicate. The k represents the key of the cache and v is the value of Player. Our IgniteBiPredicate returns True only if any player stored in the cache qualifies the expression player.getTeam() EQ input text. The result is returned as a QueryCursor. It stores all qualified entries (key-value pairs).
• From line 7 to 14 we iterated through the results to locate if such an entry exists in the cache and then populated a local List of filtered Players.
• In line 15 we are returning the filtered list of Players to the caller.

The ScanQuery goes over each cache entry and applies the predicate, which may not always be the most efficient way to query objects, especially if the data size is large.

5. SQL Queries

The SQLQuery API supports ANSI-99 SQL queries against caches. This API allows SQL joins against collocated entries on the same node as well as non-collocated distributed nodes.

To tell Apache Ignite which fields are accessible for SQL queries, we need to define the metadata. Apache Ignite’s @QuerySqlField annotation does the trick. We can even index the fields for a faster query by setting the annotation value: @QuerySqlField(index=true). We have used this annotation in Step 2 for our Player cache to enable SQL queries.

Let us explore the SQLQuery API now.

The above code does the following:

• We started off by creating a public method called sqlQuerySearch in line 1 with search text input, which we will use to search in Player cache.
• In line 3, we retrieved an instance of the Player cache from Ignite node.
• Next, we extracted the table name from the cache in line 4 to 5. Once we have the table name, then we can form a SQL Query using the table name to search the cache.
• We created an instance of SqlFieldsQuery in line 6.
  The SqlFieldsQuery class provides another implementation of Query API. It is used for executing SQL statements and navigating through the results. It accepts a standard SQL query as its constructor​ parameter and executes it. SqlFieldsQuery is executed through the IgniteCache.query(SqlFieldsQuery) method, which returns a QueryCursor.
• From line 8 to 9 we iterated and extracted the results using getAll() method. If an entry exists in the cache, then we collect the results and return the response to the caller.

In replicated mode, SQL joins execute fast as all entries are replicated in all cluster nodes. However, in partitioned mode, a few nodes may not contain the primary backup, hence, distributed joins become complex and expensive. If we are planning to use distributed joins on partitioned tables then we should use Colocated joins or Hash joins.

6. Text Queries

The TextQuery API allows us to run a full-text search on stored objects in the cache. The TextQuery works on Lucene indexes. Elasticsearch and Apache Solr also use Lucene for indexing text.

Here too, we need to define the metadata to tell Apache Ignite which fields are to be enabled for Text search. The @QueryTextField annotation enables indexing along with Text Search. Also, our cache configuration needs to enable indexing by setting the setIndexedTypes. We have already set this in Step 2 during our cache configuration.

Let us explore the TextQuery API now.

The above code does the following:

• In line 1 we created a method called textSearch, which takes a text as an input and then searches that text in Player cache. It searches the fields in Player cache annotated with QueryTextField.
• Then in line 3, we retrieved an instance of the Player cache from Ignite node.
• We created an instance of TextQuery in line 4. Once we have an instance of TextQuery and IgniteCache we called the method searchPlayers to search the text in the cache.
• From line 25 to 29 we iterated through the results to locate if such an entry exists in the cache and then populated a local List of filtered Players.
• In line 30 we are returning the filtered list of Players to the caller.
• Similar to textSearch we also created a method called fuzzySearch from line 8 to 13 to perform Fuzzy search on Player cache. Fuzzy search determines whether there is any similarity between elements of the data. Let’s say we want to fetch players where the team name is ‘Barcelona’ or ‘Barcenola’. Then we just need to add a ‘~’ at the end of our search string and pass it to the TextQuery constructor. We are going to find the players where the text ‘Barcenola’ is present, and the fuzzy search will find a match between nola and lona and return all ‘Barcelona’ players.
• We can also perform Fuzzy search on specific fields. For this reason we added another method called fuzzySearchOnSpecificField from line 15 to 20 which takes fieldName and text as an input.
• In the previous methods textSearch and fuzzySearch, during search operation Ignite looked at all the indexes, name and team, to find data. However, we can also ask Ignite to look into specific index. We can see that in line 18 . In this method we ask Ignite to look at a specific index. It will search the input text only in the field provided by user. For example, if we are interested in name = ‘Neymar’, then the TextQuery can be configured as name:”Neymar”.

It is advisable to use indexes for querying entries. However, one drawback is that an index itself takes up space and may slow down the data modification as every time we modify an entry, the index needs to be rebuilt.

7. Testing the Application

Till now we have written most of the code required to perform various kinds of searches in Ignite Cache. Now let us write some code to populate the cache at startup . The PlayerService class contains various methods required to explore Query API in Ignite. Let us also add the below method to populate the cache.

The above code populates the Player Cache with Player name and Team. We can use these values for testing purposes.

After we have populated our cache with some sample values, then we added a Controller to enable users to perform various search operations.

The above code does the following:

It exposes API which can be used to test the various features of Ignite Query API that we discussed so far. It uses the PlayerService class to perform the search operations.

Once we build the project using mvn clean package then we can start the application using java -jar target/ignite-poc*.jar command.

After the application is started successfully, then we can use the Swagger-UI to test the application. The Swagger-UI comes up at http://localhost:9051/swagger-ui.html in local.

7.1 Search Player Cache with team name as input using SQLQuery API

Here, we are using SQLQuery to search all the Players whose team contains the given text : “Barcelona”

7.2 Search Player Cache with team name or player name that matches the text input using TextQuery API

Here, we are using TextQuery to search all the Players where either name or team contains the given text: “United”

7.3 Search Player Cache with team name that matches the text input using ScanQuery API

Here, we are using ScanQuery to search all the Players whose team contains the given text: “Barcelona”

7.4 Search Player Cache with team name or player name that matches the text input using TextQuery with Fuzzy search enabled

Here, we are using using Fuzzy TextQuery to search all the Players where either name or team contains the given text: ”Barcenona”

7.5 Search Player Cache with input field name and text using TextQuery API

Here, we are using Fuzzy TextQuery to search all Players with fieldName where fieldName contains the given text: “Messi”

If you would like to refer to the full code, do check:

GitHub - sumanentc/Ignite-Querying-Cache: Sample Spring Boot application to query the data stored in the Cache of Apache Ignite using query API.

Conclusion

To perform search operation on Key-Value pair, Cache is a great feature provided by Ignite. If Query API is used properly then the performance is also at par with key-value pair search. If there is a need for key-value search along with SQL Query support then Apache Ignite is one of the best choices currently available.

References & Useful Readings

• JCache Provider - Apache Ignite
• Apache Ignite With Spring Boot
• What Is GridGain?
• Apache Ignite Documentation
• Cache Queries

Querying Cache in Apache Ignite was originally published in Crux Intelligence on Medium, where people are continuing the conversation by highlighting and responding to this story.