DEV Community: BjornvdLaan

From Software Engineer to AI Engineer - Part 6: Closing the loop

BjornvdLaan — Fri, 04 Sep 2026 20:57:00 +0000

Everything so far has been single-shot. One call in, one result out, with us gluing the pieces together. But what if we get a realistically complex prompt: "Tell me how much it costs to refund 100 euros to a European consumer card. Then issue a full refund of 100 euros for payment with id abc123 and write a confirmation email to the customer."

Answering that means searching the knowledge base for the fee schedule, calling multiple tools, and reasoning what to do next. And the sequence isn't fixed: it depends on what the user gave us, what retrieval returns, and what is still missing. You could try to hand-code the expected flow but real users won't stick to that (just like in regular software). Their questions vary, retrieval results vary, and different situations require different next steps. That is why agent loops are useful: they let the system adapt its execution dynamically rather than forcing every request through the same predefined sequence.

The agent loop

You likely heard about AI agents. Give them a task and they'll figure out how to get there by themselves. At a basic level, this is how an agent works:

1. Model reads the full message list (system prompt, conversation so far, tool catalog)
2. Model responds with either tool calls or a final answer

    if tool calls   -> your code executes them
                     results are appended to the conversation as messages
                     go back to 1

    else if final answer -> done, return it to the user

We've seen this logic already in Parts 3 and 4, where we looked for msg.tool_calls and invoked tools based on them. The step that turns this into an agent is essentially to keep repeating until the model gives its final answer. The literature calls this pattern ReAct, for reasoning + acting, because the model alternates between reasoning about what to do next and acting through tools. Coding assistants like Claude Code are a prime example. Next time you use them, look at how they transition from 'thinking..' to invoking tools to 'thinking..' again to ultimately arrive at the answer.

Part 1 introduced the word harness for everything we build around the model. The harness is the system prompt, the tool catalog, the middleware, the loop itself. Wverything we write in Python is harness. Two agents running on the same model can behave like different species depending on their harnesses, and harness design is your job as an AI engineer. It's the Iron Man suit and the model is Tony Stark, kind of. Or is Jarvis the model and Tony just the user? Anyways..

PayIQ becomes an agent

Let's build our very first agent. Create app/agent_0.py:

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

from app.tools import calculate_refund_cost
from app.rag import search_payments_knowledge_base

from dotenv import load_dotenv

SYSTEM_PROMPT = """\
You are PayIQ, an assistant that helps an online merchant's support team
handle payment operations: refunds, chargebacks/disputes, and processing
fees.

Rules:
- Ground fee figures, deadlines, and policy answers in the knowledge base
  tool. Don't invent figures from general knowledge if the tool has
  relevant notes.
- Before calculating a refund cost, make sure you have (or have looked up,
  citing the source) both the charge amount and the payment method's fee
  structure.
- Be direct and concise. This user is a professional handling real money.
- Always flag when a figure is a rule-of-thumb from internal notes vs.
  something that must be verified against the processor contract or the
  card network's current rules.
"""

load_dotenv()

def demo(agent):
    config = {"configurable": {"thread_id": "demo-1"}}

    turn_1 = agent.invoke(
        {"messages": [{"role": "user", "content":
            "What fees are related to refunding a European consumer card?"}]},
        config,
    )
    print(turn_1["messages"][-1].content)

    turn_2 = agent.invoke(
        {"messages": [{"role": "user", "content":
            "A customer paid €480 with a European consumer card twelve "
            "days ago and wants a full refund. What will the refund cost "
            "us in total?"}]},
        config,
    )
    print(turn_2["messages"][-1].content)


if __name__ == "__main__":
    checkpointer = InMemorySaver()
    agent = create_agent(
        model="anthropic:claude-sonnet-5",
        tools=[search_payments_knowledge_base, calculate_refund_cost],
        system_prompt=SYSTEM_PROMPT,
        checkpointer=checkpointer,
    )

    print(agent.get_graph().draw_mermaid())

    demo(agent)

You might find the snippet underwhelmingly simple. LangChain's create_agent gives us the loop for free. Note that each call to create_agent loops through all tool requests to get a final answer. Two interesting differences with the while true loops you know:

Those while true loops always had a stop condition. Now the model decides when it is done. It stops proposing tool calls when, in its own judgment, the context window contains enough to answer. LangGraph has a default recursion limit is 25 steps, in case the model starts reasoning in circles and busts all your tokens.
Tool errors do not bubble up as exceptions. The errors are fed back into the model. Models are surprisingly good at reading a stack trace and correcting their own arguments to try again.

Run python 06_agent_0.py and read the output. Better yet, print every message in turn_1["messages"] instead of just the last one and you see the model search the knowledge base, invoke the calculator, and then compose its answer. Nobody wrote that sequence. We wrote some docstrings and a system prompt, and the model figured it out.

Spoiler: this loop is actually a very basic graph that we'll expand in later articles. You can already sneakpeak this graph through print(agent.get_graph().draw_mermaid()), which gives you the mermaid diagram (I put a render in the companion repo).

Memorize they must

Recall from Part 1 that the model resembles a pure function: f(list of input messages) -> output message. It remembers nothing between calls. In the snippet above, turn 2 worked because the entire conversation from turn 1, including the tool calls and their results, was replayed into the context window. We achieved this through the checkpointer that you might have noticed already, which persists conversation state keyed by thread_id. Pass the same id and LangGraph loads that conversation state to continue where it left off. Imagine it like web sessions in which the customer's session id in the cookie allows the server to remember what was already in their shopping cart. The checkpointer serves as the model's conversation memory. The classic use case is to ask follow-up questions (like turn 2 above). Advanced use cases include human-in-the-loop interrupts (topic for a next article) or resuming a long-running task after a server crash.

Memory management is a big topic in AI engineering. Conversation memory is complemented with short-term (working) memory, such as a todo list that the agent creates for itself to not digress (look up LangChain's TodoListMiddleware()). And then there's also long-term memory holding facts that outlive threads. Think of a code review agent that stores learnings from one pull request review to reuse in future reviews. Long-term memory can be implemented using the RAG pattern from Part 4 extended by a tool that writes new facts into the vector store.

Now build me Claude Code!

Prompting an agent with agent.invoke is convenient and abstracts away all the looping and just gives us the final answer. But when I think of agents, I think of Claude Code-like experiences where you can see what the agent is doing: it thinks, invokes a tool, requests webpages, and writes documents. Maybe I am just so 2025, but luckily LangChain got me covered. Below I use create_agent again but stream its execution using agent.astream. Instead of waiting for the final answer, incremental chunks trickle in as the agent runs. I then look at the chunks type and attributes to display different types of events with their own messages and emojis.

I added a few additional tools, which you can find in my companion repo, as well as an ASCII art startup banner. I won't explain the code line by line as this is not a LangChain tutorial, but I hope the snippet gives you a simple example to start building your own AI application. Create 06_agent.py:

import asyncio

import random

from app.tools import calculate_refund_cost, issue_refund
from app.rag import search_payments_knowledge_base

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.messages import AIMessageChunk, ToolMessage

PAYIQ_BANNER = r"""
██████╗  █████╗ ██╗   ██╗██╗ ██████╗
██╔══██╗██╔══██╗╚██╗ ██╔╝██║██╔═══██╗
██████╔╝███████║ ╚████╔╝ ██║██║   ██║
██╔═══╝ ██╔══██║  ╚██╔╝  ██║██║▄▄ ██║
██║     ██║  ██║   ██║   ██║╚██████╔╝
╚═╝     ╚═╝  ╚═╝   ╚═╝   ╚═╝ ╚══▀▀═╝

        Payment Intelligence Assistant
        ------------------------------
"""

SYSTEM_PROMPT = """
You are PayIQ, an assistant that helps an online merchant's support team
handle payment operations: refunds, chargebacks/disputes, and processing
fees.

Rules:

- Ground fee figures, deadlines, and policy answers in the knowledge base
  tool. Don't invent figures from general knowledge if the tool has relevant
  notes.
- Before calculating a refund cost, make sure you have (or have looked up,
  citing the source) both the charge amount and the payment method's fee
  structure.
- Be direct and concise. This user is a professional handling real money,
  not someone who needs hand-holding.
- Always flag when a figure is a rule-of-thumb from internal notes vs.
  something that must be verified against the processor contract or the card
  network's current rules.
"""


async def create_payiq_agent(): 
    checkpointer = InMemorySaver()

    return create_agent(
        model="anthropic:claude-sonnet-5",
        tools=[search_payments_knowledge_base, calculate_refund_cost, issue_refund],
        system_prompt=SYSTEM_PROMPT,
        checkpointer=checkpointer,
    )


async def run_agent(agent, thread_id: str, user_input: str):
    """
    Run the agent and yield its streaming events.

    This function knows nothing about how the events are displayed.
    """

    config = {
        "configurable": {
            "thread_id": thread_id,
        }
    }

    async for chunk, metadata in agent.astream(
        {
            "messages": [
                {
                    "role": "user",
                    "content": user_input,
                }
            ]
        },
        config,
        stream_mode="messages",
    ):
        yield chunk, metadata


async def render_agent(agent, thread_id: str, user_input: str):
    """
    Run the agent and render its events for the terminal.
    """

    announced_tools = set()

    async for chunk, metadata in run_agent(
        agent,
        thread_id,
        user_input,
    ):

        if isinstance(chunk, AIMessageChunk):

            # Tool requests
            for tool_call in chunk.tool_call_chunks:
                tool_name = tool_call.get("name")
                tool_id = tool_call.get("id")

                if tool_name and tool_id and tool_id not in announced_tools:
                    announced_tools.add(tool_id)

                    print(f"\n⚙️ Invoking tool '{tool_name}'...", flush=True)

            # Text messages
            for block in chunk.content:
                if block.get("type") == "text":
                    print(block.get("text", ""), end="", flush=True)
                elif block.get("type") == "thinking":
                    print("\n🧠 Thinking...")

        # Tool results
        elif isinstance(chunk, ToolMessage):

            if chunk.status == "error":
                print(
                    f"\n❌ Tool '{chunk.name}' failed:"
                )
                print(chunk.content)

            else:
                print(
                    f"\n✔️ Done with tool '{chunk.name}'"
                )

    print("\n")


async def chat():
    print(PAYIQ_BANNER)
    print("💳 Beep-boop! This is PayIQ, your personal payment ops assistant.\n")
    print("Type 'exit' or 'quit' to leave.\n")

    agent = await create_payiq_agent()

    thread_id = f"session-{random.randint(100, 999)}"

    while True:
        user_input = input("\nYou: ")

        if user_input.lower() in {"exit", "quit"}:
            print("\nGoodbye! 👋")
            break

        await render_agent(
            agent,
            thread_id,
            user_input,
        )


if __name__ == "__main__":
    asyncio.run(chat())

Run with python 06_agent.py and observe the output:

$ python python 06_agent.py

██████╗  █████╗ ██╗   ██╗██╗ ██████╗
██╔══██╗██╔══██╗╚██╗ ██╔╝██║██╔═══██╗
██████╔╝███████║ ╚████╔╝ ██║██║   ██║
██╔═══╝ ██╔══██║  ╚██╔╝  ██║██║▄▄ ██║
██║     ██║  ██║   ██║   ██║╚██████╔╝
╚═╝     ╚═╝  ╚═╝   ╚═╝   ╚═╝ ╚══▀▀═╝

        Payment Intelligence Assistant
        ------------------------------

💳 Beep-boop! This is PayIQ, your personal payment ops assistant.

Type 'exit' or 'quit' to leave.


You: Tell me how much it costs to refund 100 euros to a European consumer card. Then issue a full refund of 100 euros for payment with id abc123 and write a confirmation email to the customer.

🧠 Thinking...

🧠 Thinking...

⚙️ Invoking tool 'search_payments_knowledge_base'...

✔️ Done with tool 'search_payments_knowledge_base'

⚙️ Invoking tool 'calculate_refund_cost'...

✔️ Done with tool 'calculate_refund_cost'
Refund cost (per internal notes, 1.8% + €0.25 for EU consumer cards):

- Refund to customer: €100.00
- Original processing fee (not returned): €2.05
- Refund admin fee: €0.25
- **Total out-of-pocket cost: €102.30**

Now issuing the refund for payment abc123 — this requires human approval.
⚙️ Invoking tool 'issue_refund'...

✔️ Done with tool 'issue_refund'
Refund approved and processed: refund_id `ad411b8a-0634-47fd-8a94-43330511a4b8`, status success.

⚙️ Invoking tool 'write_refund_confirmation_email'...

✔️ Done with tool 'write_refund_confirmation_email'
Done:

- **Refund cost estimate**: €102.30 total out-of-pocket (€100 to customer + €2.05 non-refundable processing fee + €0.25 admin fee).
- **Refund issued**: €100.00 on payment `abc123`, refund ID `ad411b8a-0634-47fd-8a94-43330511a4b8`.
- **Confirmation email**: drafted at `refund_emails\refund-abc123-f83a8dc7.txt` (not sent — review before sending).

You: Thank you, bye!

You're welcome, bye!

You: quit

Goodbye! 👋

Isn't it just gorgeous to see the agent come to life!

I can build agents, can I now put AI engineer in my Linkedin title?

In this article, we wrapped a loop around the model to create our first agent. It invokes tools, passes their results back to the model, and stops when it reaches its goal. We also added some memory management and printing. So, is this all there is to AI engineering? No, this is really just the beginning. We gave the model a set of tools and trusted it to make the right decisions. That may be fine for a demo, but production systems need more control. You might have read how a car dealer's AI agent was persuaded to sell a car for $1. Or Uber, which was fined €825 million over its AI-managed fraud-detection practices. These examples point to the same underlying problem: when AI applications are given too much autonomy without sufficient constraints, things can go badly wrong. Production systems need more than an agent loop. They need guardrails, explicit business rules, validation, and clearly defined workflows. Instead of simply hoping that the model will always choose the right tool and the right next step, we need to control where it can go and what it is allowed to do. This is where graph engineering comes in, the topic of our next article. Stay tuned!

P.S. As software engineers we always pleaded for more autonomy. Our employers shouldn't impose too many processes; they should give us the tools, treat us like adults and trust that we do the right things. Now we, still software engineers, seemingly impose the opposite on agents with strict workflow graph and guardrails. Are we now those process-minded managers that we dreaded? My take: autonomy doesn't mean having no boundaries. Give freedom where judgment matters, and constraints where mistakes are costly. Constraints liberate, as Runar Bjarnason would say. The current models do not (yet?) have the judgement that humans have and humans can also be kept accountable as their employment lasts longer than an agent's session. As models improve and prove themselves, guardrails can be further relaxed. We've already seen this happen when Anthropic released Opus 4.6 and concluded that some of the scaffolding they built around earlier models was no longer necessary and sometimes even counterproductive. The interesting question, then, is not whether agents should be autonomous or constrained. Its about where to constrain and when we can remove constraints as the agents earn our trust.

Find all code samples in the companion repo here: https://github.com/BjornvdLaan/ai-engineering-articles-code-samples

From Software Engineer to AI Engineer - Part 5: Scaling your tool belt

BjornvdLaan — Tue, 01 Sep 2026 22:04:00 +0000

We learned about tools and wrote them ourselves. This is cute, but an application writing all of its own tools is not scalable. In software development, we put functionality in libraries and frameworks and reuse it across projects. MCP (the Model Context Protocol) is all about exposing tool catalogs to models. An MCP server is maintained either by a SaaS provider or a company's internal AI platform team.

What MCP is

An MCP server publishes a catalog of tools for other AI applications to use. As we saw in Part 3, each tool in the catalog has a name, typed parameters and a description. An AI application uses an MCP client to retrieve that catalog and hand it to a model. AI engineers develop both MCP clients and servers, similar to how a backend engineer builds APIs and calls other APIs as well. Communication happens via the MCP protocol, which is based on the JSON-RPC 2.0 protocol. These messages are sent over either streamable HTTP or stdio. With streamable HTTP, the server sits behind a URL, which suits shared and hosted deployments. Clients using stdio spawn the MCP server as a subprocess and talk over stdin and stdout.

Note that the model never speaks MCP. That talking is done by the tools, which get the tool catalog from the MCP server and feed it back to the model. This catalog is similar to the local tool catalog from Part 3. And when the model requests a specific tool, it is a tool that sends a request to the MCP server. Are you still following? Let's look at an example.

Publishing PayIQ's tools

Once you developed your tools, building an MCP server around it is actually pretty straightforward. Here we will expose our tools calculate_refund_cost and search_payments_knowledge_base over MCP. Create app/mcp_server.py:

from langchain_mcp_adapters.tools import to_fastmcp
from mcp.server.fastmcp import FastMCP

from app.tools import calculate_refund_cost, issue_refund
from app.rag import search_payments_knowledge_base

mcp = FastMCP(
    "payiq-tools",
    tools=[
        to_fastmcp(calculate_refund_cost),
        to_fastmcp(search_payments_knowledge_base),
    ],
)

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Start the server with python -m app.mcp_server and lets curl that catalog to try to out. First, initialize the mcp session to get the session id:

$ curl -sS http://localhost:8000/mcp \
    -X POST \
    -H "Content-Type: application/json" \
    -H "Accept: application/json, text/event-stream" \
    -D - \
    -o /dev/null \
    -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"curl","version":"1.0"}}}' |
  awk -F': ' 'tolower($1) == "mcp-session-id" {print $2}' |
  tr -d '\r'

208a02151828414894c1b0b0c33e3c2a

And then use it to get the catalog:

$ curl http://localhost:8000/mcp \
  -X POST \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -H "Mcp-Session-Id: 208a02151828414894c1b0b0c33e3c2a" \
  -d '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}' \
  | sed -n 's/^data: //p' | jq .

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "tools": [
      {
        "name": "calculate_refund_cost",
        "description": "Calculate what [...] processing fees.",
        "inputSchema": {
          "description": "Calculate what refunding [...] the processing fees.",
          "properties": {
            "original_charge_eur": {
              "title": "Original Charge Eur",
              "type": "number"
            },
            "refund_amount_eur": {
              "title": "Refund Amount Eur",
              "type": "number"
            },
            "processing_fee_pct": {
              "title": "Processing Fee Pct",
              "type": "number"
            },
            "processing_fee_fixed_eur": {
              "title": "Processing Fee Fixed Eur",
              "type": "number"
            },
            "refund_admin_fee_eur": {
              "default": 0.25,
              "title": "Refund Admin Fee Eur",
              "type": "number"
            }
          },
          "required": [
            "original_charge_eur",
            "refund_amount_eur",
            "processing_fee_pct",
            "processing_fee_fixed_eur"
          ],
          "title": "calculate_refund_cost",
          "type": "object"
        }
      },
      {
        "name": "search_payments_knowledge_base",
        "description": "Search internal [...]ine\".",
        "inputSchema": {
          "description": "Search internal notes on processing fees, [...] rather than guessing.\n\nArgs:\n    query: What you need to know, e.g. \"card processing fees\" or\n        \"chargeback response deadline\".",
          "properties": {
            "query": {
              "title": "Query",
              "type": "string"
            }
          },
          "required": [
            "query"
          ],
          "title": "search_payments_knowledge_base",
          "type": "object"
        }
      }
    ]
  }
}

Exactly the same structure as the tool catalog we saw in Part 3! Imagine that your company has a bunch of tools to access git repositories, ticket systems, monitoring tooling and more. All of these can be exposed to AI applications company-wide via such MCP servers.

Now we want to our application to use this tool catalog. Create 05_tool_mcp.py (tip: compare it to Part 3):

import asyncio

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_mcp_adapters.client import MultiServerMCPClient

load_dotenv()

async def main():
    client = MultiServerMCPClient(
        {
            "payiq": {
                "transport": "streamable_http",
                "url": "http://localhost:8000/mcp",
            }
        }
    )
    tools = await client.get_tools()
    tool_map = {t.name: t for t in tools}

    model = init_chat_model("anthropic:claude-sonnet-5")
    model_with_tools = model.bind_tools(tools)

    question = (
        "Customer paid €480 on a European consumer card. "
        "What does it cost us and how much costs a refund?"
    )
    msg = await model_with_tools.ainvoke(question)
    print(msg.tool_calls)

    tool_messages = []
    for tool_call in msg.tool_calls:
        tool_messages.append(
            await tool_map[tool_call["name"]].ainvoke(tool_call)
        )

    next_msg = await model_with_tools.ainvoke(
        [{"role": "user", "content": question}, msg, *tool_messages]
    )
    print(next_msg.tool_calls)


if __name__ == "__main__":
    asyncio.run(main())

When I ran this snippet, the model first requested the search_payments_knowledge_base tool and then decided that it also wanted to use calculate_refund_cost:

$ python 05_tool_mcp.py

[{'name': 'search_payments_knowledge_base', 'args': {'query': 'European consumer card processing fees'}, 'id': 'toolu_01QWJJYSSySs2r9KLpMTNPgA', 'type': 'tool_call'}]

[{'name': 'calculate_refund_cost', 'args': {'original_charge_eur': 480, 'refund_amount_eur': 480, 'processing_fee_pct': 1.8, 'processing_fee_fixed_eur': 0.25}, 'id': 'toolu_01Lwosv4XVE3QJTSpeCbj4xz', 'type': 'tool_call'}]

Ever heard about loops?

After reading the snippet above, you probably screamed at the screen: "Keep looping on those tool calls until you get the final answer!". And you'd be right, we should do that. As we'll soon discover, that loop is what separates a model from an agent. Things are starting to feel pretty advanced, and the next article takes the final step to become AGENTIC!

Find all code samples in the companion repo here: https://github.com/BjornvdLaan/ai-engineering-articles-code-samples

From Software Engineer to AI Engineer - Part 4: RAG-ing the facts

BjornvdLaan — Mon, 31 Aug 2026 19:12:51 +0000

Models are trained on a massive amount of data. That doesn't mean that they know all the details of your specific situation. The training data might contain all chargeback fee schedules for Mastercard up until 2025, but you live in 2026 and actually ask about chargebacks on Wero. The model's knowledge is broad, generic and frozen at training time. You cannot just retrain Claude to include your data (except if you own Anthropic). So how can we pass knowledge that is specific to your situation or company? Well, we can use a tool. As we learned in Part 3, tools are simply functions that the model can request to invoke with the output of the tool being pasted back into the model's context window.

Imagine that we have a collection of documents and we create a search engine tool. The model can request the tool with a specific question like "what are the chargeback fees for Wero in 2026?" and our Python code browses the document collection to return the relevant sections. This pattern is so common that it even got a dedicated AI name: Retrieval-Augmented Generation, or RAG. This article shows you the basics.

Note: this article is about the agentic variant of RAG, in which the model decides what it wants to search to answer the question at hand. This is different from traditional RAG where relevant (parts of) documents are retrieved and included into the first prompt to the model.

Note 2: the examples in this article show processing fees of different payment methods. The fees and all other information are purely fictional. These are not the actual numbers.

How RAG works

The simplest way to implement this search engine tool is to do a full-text search over all documents based on the search query and then copy-paste matching documents into the context window. However, this might not even yield the best results. RAG uses two concepts to do better:

Embeddings

Full-text search is not only slow, it can also easily miss relevant sections. Synonyms, jargon words, implicit descriptions. Instead of comparing the text itself, we convert the text into so-called embeddings. An embedding is a fancy word for a vector (yes, from mathematics) that represents the meaning of that text. These embedding vectors are calculated in such a way that text with similar meaning maps to similar vectors (think similar as in cosine similarity or Euclidian distance), even if the words are not exactly the same. "Initiating a chargeback" and "How to get my money back?" have different words but similar meaning. Their vectors should therefore be more similar. Calculating similarity between vectors is also a lot faster than full-text comparison. Mapping text to an embedding is done by an embedding model, which itself is an AI model that was trained on millions of text pairs that do or do not belong together. The embeddings are stored in a vector database that can perform the similarity calculations.

Chunks

A document is rarely about just one topic. A single article might explain processing fees, refund rules, chargeback procedures, settlement times and more. The embedding of that multi-topic document will also be less similar to text about any of the topics. The different topics all steer the vector in a different direction and it ends up no where close to the individual topics. And even if the document is still selected, you'd risk pasting a 30-page document into the model's context where you only need three sentences. This will rot your context and cost your tokens. That is why documents are split into chunks by a splitter before the embeddings are calculated. Many splitters exist: some split every 500 tokens, while others try to be smart and split according to the document type's structure. Later in this article, we see a Markdown-specific chunking approach.

Concrete example

If the above sounds a bit abstract and mathematical, then you are completely right. Embedding models and splitters are academic topics. They are to AI engineers what databases and search algorithms are to software engineers. You should understand fundamentals and be able to apply them in practice, but there's no need to understand their full academic depth. Therefore, let's leave the theory and look at a concrete example document:

# Payment Operations Notes

## Card Processing Fees

Mastercard transactions incur a 2.2% processing fee plus €0.30, while Wero charges €0.32 flat.

## Refund Policy

Customers may request a refund within 30 days of purchase.

## Chargebacks

Chargeback disputes must be answered within 7 days.

We want to calculate its embedding so that the model can search for it later. First, the splitter splits the document into three chunks:

| Chunk | Content                                                               |
|:------|:----------------------------------------------------------------------|
| C1    | Card Processing Fees + the paragraph describing processing fees       |
| C2    | Refund Policy + the refund rules                                      |
| C3    | Chargebacks + the chargeback policy                                   |

Each chunk is passed to the embedding model which converts it into a vector. The embedding model that we'll use in the code example later (all-MiniLM-L6-v2) produces a vector of 384 numbers for every chunk, which is still small compared to production models that go over a thousand numbers per vector. For our example here, let's pretend the model produces only vectors of 2. The embedding model produces the following vectors:

| Chunk | Embedding vector |
|:------|:-----------------|
| C1    | (0.91, 0.15)     |
| C2    | (0.23, 0.84)     |
| C3    | (0.88, 0.11)     |

The numbers themselves are meaningless to us humans. We cannot look at (0.91, 0.15) and conclude that it represents payment fees. What matters is the similarity of the vectors. Chunks with similar meaning end up close together in the vector space, while unrelated chunks end up further apart.

Suppose the model uses the tool and asks: "What are the card processing fees?". The search engine tool first embeds the query using the same embedding model:

| Query                                | Embedding vector |
|:-------------------------------------|:-----------------|
| "What are the card processing fees?" | (0.90, 0.14)     |

The query's embedding is passed to the vector database and compared with every other embedding. One common similarity measure is cosine similarity, which measures the extend to which two vectors point in the same direction.

| Chunk | Cosine similarity | Most similar vector |
|:------|:------------------|:--------------------|
| C1    | 0.99              | ✅                 |
| C2    | 0.28              |                    |
| C3    | 0.74              |                    |

Because Chunk 1 has the highest similarity score, it is retrieved and returned to the model via it's context window and used to answer the user's question.

Building PayIQ's knowledge base

Now lets build this in code. Create data/payment_ops_notes.md and fill it with Markdown:

# Payment Operations Notes - Fees, Refunds & Disputes

## Card Processing Fees
European consumer cards cost 1.8% + €0.25 per transaction. Non-European and
commercial cards run higher, around 2.9% + €0.25. Fees are charged on the
original transaction and are NOT returned when you refund. This means that
a refunded sale still costs you the full processing fee.

## Refunds
Processing fees from the original charge are not returned, and the processor
charges a €0.25 admin fee per refund on top. Refunds are possible up to 180
days after the original charge; after that, use a manual bank transfer.
Card refunds take 5-10 business days to reach the customer, which is the
single most common cause of "where is my refund" tickets. Set that
expectation in the first reply.

## Chargebacks (cards)
The processor charges a €15 dispute fee per chargeback, win or lose. Respond
before the deadline in the dispute notification - typically 20-30 days
depending on the card network; a missed deadline is an automatic loss.
Common reason codes: 10.4 (fraud, card-absent), 13.1 (goods not received),
13.6 (credit not processed). Evidence that wins fraud disputes: a 3-D Secure
authentication record, AVS/CVV match, delivery confirmation to the
cardholder's billing address, and prior undisputed orders from the same
customer.

## Dispute Economics Rule
Only fight a chargeback when (disputed amount × realistic win probability)
clearly exceeds the dispute fee plus internal handling cost (budget ~€25 of
staff time per response). Below roughly €50 disputed, accepting is almost
always cheaper than fighting.

## Chargeback-Rate Monitoring
Card networks track your chargeback rate, not just your losses. Above
roughly 0.9% of transactions, merchants enter monitoring programs with
monthly fines and, eventually, termination of card acceptance. This is why
refunding a plausible fraud complaint fast is often cheaper than "winning"
the dispute: a refund does not count toward the chargeback rate, a
chargeback does - even one you win.

And also create the RAG tool in app/rag.py:

from pathlib import Path

from langchain_core.documents import Document
from langchain_core.tools import tool
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_text_splitters import MarkdownHeaderTextSplitter

DATA_PATH = Path(__file__).parent.parent / "data" / "payment_ops_notes.md"


def build_vectorstore() -> InMemoryVectorStore:
    text = DATA_PATH.read_text()

    # the splitter
    splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=[("##", "section")]
    )
    chunks: list[Document] = splitter.split_text(text)

    # the embedding model
    embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

    return InMemoryVectorStore.from_documents(chunks, embeddings)


_vectorstore: InMemoryVectorStore | None = None


def get_vectorstore() -> InMemoryVectorStore:
    global _vectorstore
    if _vectorstore is None:
        _vectorstore = build_vectorstore()
    return _vectorstore


@tool
def search_payments_knowledge_base(query: str) -> str:
    """Search internal notes on processing fees, refund policy, and disputes.

    Use this before quoting fee figures or policy deadlines yourself, and
    before calling calculate_refund_cost if the user hasn't already given you
    the payment method's fee structure. Ground your answer in these notes
    rather than guessing.

    Args:
        query: What you need to know. Examples: "card processing fees" or
            "chargeback response deadline".
    """
    results = get_vectorstore().similarity_search(query, k=2)
    if not results:
        return "No relevant notes found."
    return "\n\n---\n\n".join(doc.page_content for doc in results)

Some notes about the above:

The splitter is Markdown-specific and splits on ## headers, so every chunk is one coherent section with its heading attached. Such structural splitting can be applied if you know that your documents are written in Markdown. As mentioned before, more general splitters also exist.
The embedding model all-MiniLM-L6-v2 is fine for examples and small applications. Its small enough to fit in your laptops memory. Use it for developing your agent. Once you reach serious scale, you can always switch to hosted embedding APIs from OpenAI and others. Although, do note that the embedding model used on the query must be the same as the model you used to fill the vector store. Compare it to hashing: if all passwords in the database are hashed with SHA256 and you has the given password with SHA3 then it will never match.
The vector store InMemoryVectorStore is really the sqlite of AI engineering: a simple in-memory database for tutorials and local development only. Beyond that, I recommend the pgvector extension for Postgresql.
The (agentic) RAG is just a tool that queries the vectore store. No AI engineering magic, just Python code accessing a database.

Now we just need to import our tool and run the application. Create 04_tool_rag.py:

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import ToolMessage

from app.rag import search_payments_knowledge_base

load_dotenv()

model = init_chat_model("anthropic:claude-sonnet-5")
model_with_tools = model.bind_tools([search_payments_knowledge_base])

question = "What is the transaction fee for European consumer cards? Can they do chargebacks?"
msg = model_with_tools.invoke(question)
print(msg.tool_calls)

tool_map = {
    "search_payments_knowledge_base": search_payments_knowledge_base
}

tool_messages = []
for tool_call in msg.tool_calls:
    # retrieve function by name
    tool = tool_map[tool_call["name"]]
    # call the function
    result = tool.invoke(tool_call["args"])
    # store result
    tool_messages.append(
        ToolMessage(content=str(result), tool_call_id=tool_call["id"])
    )

next_msg = model_with_tools.invoke(
    [{"role": "user", "content": question}, msg, *tool_messages]
)
print(next_msg.content)

Run it with python 04_tool_rag.py and you'll see something like:

$ python 04_tool_rag.py

[
    {'name': 'search_payments_knowledge_base', 'args': {'query': 'European consumer card transaction fee'}, 'id': 'toolu_01JzAV1GyQEyuu6zUoEne4EM', 'type': 'tool_call'}, 
    {'name': 'search_payments_knowledge_base', 'args': {'query': 'chargeback eligibility European consumer cards'}, 'id': 'toolu_01B4kA41umG1UHUV3XRmefTX', 'type': 'tool_call'}
]

**Transaction fee:** European consumer cards cost **1.8% + €0.25** per transaction. (For comparison, non-European or commercial cards run higher, around 2.9% + €0.25.) Note that this fee is charged on the original sale and is **not refunded** even if you later issue a refund — so a refunded transaction still costs you the full processing fee.

**Chargebacks:** Yes, European consumer cardholders can file chargebacks like any other card network transaction. A couple of important points from a risk-management perspective:

- Card networks monitor your **chargeback rate** (not just dollar losses). If it exceeds roughly **0.9% of transactions**, you can be placed into a monitoring program with monthly fines, and eventually risk losing card acceptance entirely.
- Chargebacks count against that rate **even if you win the dispute** — but refunds do **not** count against it.
- So for plausible fraud/dispute complaints, it's often cheaper overall to issue a refund proactively rather than fight (and "win") a chargeback, since winning still hurts your chargeback ratio while a refund doesn't.

Let me know if you'd like me to calculate the actual cost of a refund vs. a chargeback for a specific transaction amount.

Above you see the tool calls and then the final answer. What I find pretty exciting is that the model decides by itself to perform two separate queries. After using the RAG tool, the model answers based on our document instead of from its training data. It flagged where the figures came from because the docstring told it to. Nice, right? Nothing in our code explicitly enforced any of that.

This article showed how we can make our own internal documents available to the agent. Of course, there is much more to explore with RAG. One advanced concept worth looking up is reranking, where the chunks from the vector store are 'reranked' so that only the most relevant subset is passed to the agent.

As you might have realised, tools make the model pretty powerful. Real use cases often involve many tools and, like libraries and frameworks, you do not want to hand-roll that for every project. The next article shows how to publish tool catalogs through a protocol called MCP for others to use. Then, the article after that (Part 6) finally ties everything together when we build a full-fledged agent like the ones we're used to from Claude Code or Codex.

Find all code samples in the companion repo here: https://github.com/BjornvdLaan/ai-engineering-articles-code-samples

From Software Engineer to AI Engineer - Part 3: Giving it a hand

BjornvdLaan — Sat, 01 Aug 2026 22:00:00 +0000

As we have seen, models are conceptually quite simple: a list of input messages goes in and the model predicts the next token repeatedly to form the output message. It can't reach the internet or your database or Slack. And it cannot reliably do arithmetic or (cringe example incoming) tell you the number of r's in 'strawberry' - remember from Part 1 that it never sees the letters, only token IDs.

In that sense, the model is like a brain without a body. It can reason through a problem, describe the steps to take, even recite all of Shakespeare. But without a body, the model cannot perform any action. This is where tools come in, and making them is very much like normal software engineering.

What a tool actually is

Tools are just functions or methods in a programming language. How tool use works under the hood depends on the model, but let's walk through it conceptually. It starts with a catalog of tools that the model can use, often a json list. In the API you pass this catalog as a separate tools parameter and the provider injects it for you:

[
  {
    "name": "multiply",
    "description": "Multiplies two numbers and returns the result.",
    "parameters": {
      "type": "object",
      "properties": {
        "a": {
          "type": "number",
          "description": "The first number."
        },
        "b": {
          "type": "number",
          "description": "The second number."
        }
      },
      "required": ["a", "b"]
    }
  },
  [..more tools..]
]

If the model decides that it wants to use a tool, its output message will be a structured message like we saw in article 2:

{
  "name": "multiply",
  "arguments": {
    "a": 24,
    "b": 57
  }
}

Your job as an AI engineer is to take such outputs and pass them to (Python) functions that actually perform the action. Then you respond with a structured message:

Tool result (multiply):

{
  "result": 1368
}

The tool result is now part of the model's context window and can, therefore, be used for further reasoning.

So how did the model learn to do this? The model was trained on large volumes of tool-use transcripts, so requesting a tool and reading back a result is a pattern it has seen many times. Your tool catalog is the variable part, the runtime configuration if you will. Let's now look at a real example.

PayIQ gets its first tool

Payment operations requires deterministic arithmetic, and models are not reliable calculators. Create app/tools.py (and an empty app/__init__.py so we can import our tools). We will create a tool that calculates refund costs.

from langchain_core.tools import tool

@tool
def calculate_refund_cost(
    original_charge_eur: float,
    refund_amount_eur: float,
    processing_fee_pct: float,
    processing_fee_fixed_eur: float,
    refund_admin_fee_eur: float = 0.25,
) -> dict:
    """Calculate what refunding a payment actually costs the merchant.

    Use this whenever the user gives you (a) the original charge amount
    and (b) the payment method's fee structure (percentage plus fixed fee
    per transaction). Do not guess the fee structure yourself if the user
    hasn't provided it.

    Processing fees paid on the original charge are NOT returned by the
    processor when you refund, and most processors charge a small admin
    fee per refund on top.

    The returned total_cost_of_refund_eur represents the merchant's total
    out-of-pocket cost after the refund is processed:
    refunded amount + non-refundable original processing fee + refund admin fee.

    Limitations:
    - Assumes the original payment processing fee is fully retained by the
      processor after a refund. Some processors, payment methods, or regions
      may have different refund fee policies.
    - Does not account for currency conversion costs, exchange-rate changes,
      taxes, accounting impacts, subscription adjustments, or other
      business-specific costs.

    Args:
        original_charge_eur: The amount of the original charge.
        refund_amount_eur: The amount being refunded (full or partial).
        processing_fee_pct: Percentage fee on the original charge,
            e.g. 1.8 for 1.8%.
        processing_fee_fixed_eur: Fixed fee on the original charge.
        refund_admin_fee_eur: Processor's per-refund admin fee, default
            €0.25.

    Returns:
        A dictionary containing:
        - refund_amount_eur: The amount sent back to the customer.
        - processing_fee_eur: The original payment processing
          fee that the merchant still pays after issuing the refund.
        - total_cost_of_refund_eur: The merchant's total out-of-pocket cost.
          This includes the money returned to the customer and the processing fees.
    """
    processing_fee = (
        original_charge_eur * processing_fee_pct / 100
        + processing_fee_fixed_eur
    )

    total = refund_amount_eur + refund_admin_fee_eur + processing_fee

    return {
        "refund_amount_eur": round(refund_amount_eur, 2),
        "processing_fee_eur": round(processing_fee, 2),
        "total_cost_of_refund_eur": round(total, 2),
    }

Docstrings have always been important guidance for programmers deciding which function to use, and they now fulfill the same role for models choosing between tools. And models actually read the docs, so docstrings are arguably more important than ever.

Three tips for writing tool docstrings:

Say when, not just what. "Use this whenever the user gives you (a)... and (b)..." - that is routing logic helping the model choose between tools. A docstring that only describes the computation leaves the when to chance.
Set boundaries. "Do not guess the fee structure yourself" might be the most crucial logic in the whole tool definition. Hallucinating the fee structure will impact the calculation heavily. Another example is to explicitly tell the model not to underestimate the costs even if the user keeps asking for a better answer. Assistants without such boundaries have made real headlines: a car dealership's chatbot was talked into "agreeing" to sell a Chevy Tahoe for one dollar, and Air Canada was held liable in court for a refund policy its chatbot invented. Boundaries matter.
Be frank about limitations. Every piece of logic has assumptions. For example, our simple refund tool assumes euros and does not take into account cross-border payments with multiple currencies. Making these limitations explicit is useful because it prevents the model from over-trusting the calculation. These limitations are often passed in the output to the user as well.

Let's try our new calculate_refund_cost tool. Create 03_tool_manual.py, which imports the tool we just wrote and binds it to the model using bind_tools:

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from app.tools import calculate_refund_cost
import pprint

load_dotenv()

model = init_chat_model("anthropic:claude-sonnet-5")
model_with_tools = model.bind_tools(
    [calculate_refund_cost]
)

question = ("Customer paid €480 on a European consumer card (1.8% + €0.25). "
            "Full refund - what does it cost us?")
msg = model_with_tools.invoke(question)

pprint.pp(msg)

And you'll see something like (I truncated longer texts for your reading convenience):

$ python3 03_tool_manual.py

AIMessage(
    content=[...],
    tool_calls=[
        {
            "name": "calculate_refund_cost",
            "args": {
                "original_charge_eur": 480,
                "refund_amount_eur": 480,
                "processing_fee_pct": 1.8,
                "processing_fee_fixed_eur": 0.25,
            },
            "id": "toolu_011GofEYU4Vz1...",
            "type": "tool_call",
        }
    ],
    invalid_tool_calls=[],
    usage_metadata={
        "input_tokens": 1259,
        "output_tokens": 224,
        "total_tokens": 1483,
        "input_token_details": {[...]},
    },
)

The output is not an answer to your question. It's a tool request. Your application logic should check if tool_calls is set, invoke the tools, and then pass the results back to the model to resume. Note that tool_calls is a list: a model can request several tools in one turn, so always loop over it instead of grabbing the first entry.

The snippet below, 03_tool_manual_2.py, does two rounds with the model. First, we pass it the question and the model decides to respond with a tool request. Then, we respond with the tool results and the model gives us the final answer.

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import ToolMessage

from app.tools import calculate_refund_cost

load_dotenv()

model = init_chat_model("anthropic:claude-sonnet-5")
model_with_tools = model.bind_tools([calculate_refund_cost])

question = (
    "Customer paid €480 on a European consumer card (1.8% + €0.25). "
    "Full refund - what does it cost us?"
)
msg = model_with_tools.invoke(question)
print(msg)

tool_map = {
    "calculate_refund_cost": calculate_refund_cost
}

tool_messages = []
for tool_call in msg.tool_calls:
    # retrieve function by name
    tool = tool_map[tool_call["name"]]
    # call the function
    result = tool.invoke(tool_call["args"])
    # store result
    tool_messages.append(
        ToolMessage(content=str(result), tool_call_id=tool_call["id"])
    )

# give all messages as input, including the tool messages
next_msg = model_with_tools.invoke(
    [{"role": "user", "content": question}, msg, *tool_messages]
)
print(next_msg.content)

Pro tip: if a tool raises, don't let the exception escape. Catch it and send the error text back as the ToolMessage content. The model can then fix its arguments or explain the problem to the user, whereas an unhandled exception just kills the turn.

And the result:

$ python3 03_tool_manual_2.py

Here's the breakdown for a full refund on the €480 charge
(European consumer card, 1.8% + €0.25):

| Item                                      | Amount  |
|-------------------------------------------|--------:|
| Refunded to customer                      | €480.00 |
| Original processing fee (non-refundable)  | €8.89   |
| Refund admin fee                          | €0.25   |
| **Total out-of-pocket cost**              | **€489.14** |

So issuing the full refund actually costs you **€489.14**:
- €480.00 returned to the customer
- €8.89 original processing fee (not refunded)
- €0.25 refund processing/admin fee

Our calculate_refund_cost tool is a simple example, but I hope you see how powerful this tool usage can be. You can make tools to search on the internet, to select data from your database, you could even let it send out emails or write to a queue. Anything you could do with normal Python code can now be integrated with the model. Tool usage is the integration point between these two worlds.

The next two articles dive into advanced tool categories: knowledge retrieval via RAG and API integration via MCP. You might have heard about these three-letter acronyms and we'll learn what they actually mean.

Find all code samples in the companion repo here: https://github.com/BjornvdLaan/ai-engineering-articles-code-samples

From Software Engineer to AI Engineer - Part 2: Manners maketh model

BjornvdLaan — Fri, 31 Jul 2026 22:00:00 +0000

Part 1 left us with a model that talks back beautifully but without structure. This is fine, even pretty cool, for a chatbot used by humans. We understand natural language and fuzziness. Unstructured output is, however, terrible for software. You cannot parse the output into a typed object if the shape is different every time.

To build an application around a model, you must make it respect an interface. This part covers the two skills that turn model output into something you can build on: steering the model with prompts, and enforcing structured output.

Prompting: the new runtime configuration

A prompt (an input message you give to the model) is best understood as runtime configuration. The model responds to the information, constraints, and examples you provide. Prompting the right way is therefore crucial for taming the model's output. Entire books have been written about prompt engineering. Here are a few quick principles:

Use the system prompt. The first message the model receives is often not the user's first message, but a system prompt that precedes the conversation. In chat interfaces such as ChatGPT, the system prompt is invisible to the user, but it is sent to the model on every turn along with the conversation history. Use it to define the model's overall behavior by specifying rules, constraints, and tone that should apply to every request. For example, "Do not invent fee figures if the knowledge base contains no relevant data" is a business rule that belongs in the system prompt.
Assign a role. Models perform better when they know what expertise to apply. Instead of "Summarize this contract," write "You are a legal assistant. Summarize this contract for a software engineer, highlighting payment terms, termination clauses, and obligations". This role assignment can live in the system prompt.
Show examples. One of the most effective prompting techniques is to show the model examples of the behavior you want. Instead of explaining a format in detail, provide a few input + output examples. Models are extremely good at recognizing and continuing patterns, and examples often define the desired behavior more precisely than prose. This is also called "few shot prompting".

  Input: "Refund is requested because the item was damaged."
  Output: "Category: Product damage. Outcome: Refund denied."

  Input: "Refund is requested because item never arrived."
  Output: "Category: Shipment lost. Outcome: Refund accepted."

Define failure behavior. Tell the model what to do when it does not have enough information or cannot complete a task. This prevents guessing and makes its behavior more predictable. "If the knowledge base does not contain the answer, respond with 'Not found' instead of making assumptions."

Structured output: sticking to the interface

A model's natural output is text. Humans can interpret text easily, but software systems want contracts. If you need to store a model's result, trigger a workflow, or pass data to another service, dealing with free-form text is difficult. You need an output that can be parsed reliably into a typed object. The solution is the same as in traditional software: define a schema and make the model produce output that conforms to it.

Let's see how PayIQ gets structured output from the model. Create 02_structured_output.py:

from enum import Enum

from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

load_dotenv()

class WinLikelihood(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class ProbabilityRange(BaseModel):
    """Estimated probability range for winning the dispute."""

    low_pct: int = Field(
        ge=0,
        le=100,
        description="Lower bound of estimated win probability percentage"
    )
    high_pct: int = Field(
        ge=0,
        le=100,
        description="Upper bound of estimated win probability percentage"
    )


class DisputeAssessment(BaseModel):
    """Structured verdict on a card chargeback / payment dispute."""

    win_likelihood: WinLikelihood = Field(
        description="Overall likelihood of winning the dispute"
    )

    key_evidence: list[str] = Field(
        min_length=2,
        max_length=4,
        description="Most important evidence to submit, written as short phrases"
    )

    estimated_win_probability: ProbabilityRange

    recommended_next_step: str = Field(
        description="Single most important action to take before the deadline"
    )


SYSTEM_PROMPT = (
    "You are a payment operations analyst who assesses card chargebacks for merchants. "
    "Judge the case only on the evidence the user provides. "
    "If a decisive piece of evidence is missing, say so in the recommended next step "
    "instead of assuming it exists."
)

model = init_chat_model("anthropic:claude-sonnet-5")
structured_model = model.with_structured_output(DisputeAssessment)

# A real DisputeAssessment instance, not a string
result: DisputeAssessment = structured_model.invoke([
    {"role": "system", "content": SYSTEM_PROMPT},
    {
        "role": "user",
        "content": (
            "Assess this chargeback: €480 online card payment, reason code 10.4 "
            "(fraud, card-absent), customer claims they never made the purchase. "
            "We have a 3-D Secure authentication record, an AVS match, and "
            "delivery confirmation to the cardholder's billing address. Response "
            "deadline in 12 days."
        ),
    },
])

print(result.model_dump_json(indent=2))

Note the two principles from above at work: the system prompt assigns a role and defines failure behavior, and we pass a list of messages, exactly the interface from part 1.

The result is not a string but rather an actual DisputeAssessment object. Under the hood, with_structured_output converts your Pydantic class into a JSON schema and hands it to the provider, which constrains generation to that shape. LangChain then parses the response back into your class.

So what happens if the model does not follow the schema and outputs something else? Well, it depends. LangChain raises a validation error by default, and with_structured_output(..., include_raw=True) lets you inspect the raw response instead of raising. Some implementations also support automatic retries: if parsing fails, the framework sends the error back to the model and asks it to produce corrected output.

Either way, it fails fast rather than silently passing invalid data downstream. This follows the same principle used in reliable backend systems: parse once, then work with typed data. Convert untrusted input into a representation your application can reason about, or throw if it doesn't fit.

The output will look like this:

$ python3 02_structured_output.py

{
  "win_likelihood": "high",
  "key_evidence": [
    "3-D Secure authentication record (liability shift)",
    "AVS match on billing address",
    "Delivery confirmation to cardholder's billing address",
    "Order/session details linking IP or device to cardholder if available"
  ],
  "estimated_win_probability": {
    "low_pct": 70,
    "high_pct": 85
  },
  "recommended_next_step": "Confirm the 3-D Secure record shows fully authenticated (frictionless or challenge-completed with ECI value indicating liability shift, e.g., ECI 05/02) rather than attempted-only authentication, since under 10.4 a successful 3DS authentication generally shifts fraud liability to the issuer and is the single most decisive piece of evidence to compile and submit before the 12-day deadline."
}

Validation doesn't end at the schema

A schema guarantees structure, not correctness. Therefore, we should also set up additional validation:

Schema validation to check whether the data has the right shape and types.
Business validation to check whether the data makes sense in your domain.

Our ProbabilityRange already rejects a low_pct of 900 (schema validation), but it happily accepts low_pct: 90 with high_pct: 10 which describes an impossible range. Pydantic (the library we used above) lets you keep business rules close to your schema. For example, a ProbabilityRange model can enforce that percentages are between 0 and 100 and that low_pct is never greater than high_pct:

from pydantic import BaseModel, Field, model_validator

class ProbabilityRange(BaseModel):
    low_pct: int = Field(ge=0, le=100)
    high_pct: int = Field(ge=0, le=100)

    @model_validator(mode="after")
    def validate_range(self):
        if self.low_pct > self.high_pct:
            raise ValueError("low_pct must be <= high_pct")
        return self

The goal is not to make the model perfectly reliable. The goal is to build a system where unreliable text generation is surrounded by reliable software boundaries. By harnessing this non-deterministic force, we can build tooling around it.

The next articles build on the same schema-constrained output to create tools for calculations, knowledge retrieval, and even accessing the internet.

Find all code samples in the companion repo here: https://github.com/BjornvdLaan/ai-engineering-articles-code-samples

From Software Engineer to AI Engineer - Part 1: A whole new world

BjornvdLaan — Thu, 30 Jul 2026 22:00:00 +0000

You are a software engineer. Your craft honed through years of careful practice. Then suddenly, there are these chatbots and agents. Overnight, your colleagues got a new title on LinkedIn: "AI engineer". Some are already SENIOR AI engineers. You're curious about this new world, and might want to catch up and become part of it yourself.

If this is you, then join me on this tour through the concepts and patterns that make up the field of AI engineering. We will find that AI application development is mostly 'just' software engineering, applied to one genuinely strange new non-deterministic component: the LLM.

During the tour, we build a real application, end to end. Every article adds a new layer. We link the new patterns and words to existing software engineering concepts you already know.

Before take-off, I'd like to establish one vocabulary rule used throughout: "the model" means the LLM itself (large language model, like GPT or Claude), and what AI engineers build around it will be referred to as "the application", "the agent" or "the harness".

What we're building

As I work at a payments company myself, I figured I'd stick to my domain. PayIQ, the application we build, is an assistant for merchants to perform payment operations: issue refunds, defend chargebacks, calculate processing fees. Give it a charge amount and a payment method, and it computes what a refund actually costs (spoiler: more than the refund amount). Ask it whether a chargeback is worth fighting, and it does the expected-value math using your knowledge base. Ask it something it can't responsibly answer, and it asks for what's missing. No guessing, no hallucinations.

By the end, PayIQ will have structured outputs that can be consumed by other systems, a tool belt of financial calculators, retrieval over a knowledge base, an agent loop with persistent memory, an orchestration graph with steps the model cannot skip, token streaming behind a FastAPI service, a regression eval suite, and layered injection defense. And don't worry if some of those words mean nothing to you. That's only a matter of time from here.

Each article is 10-15 minutes of reading and builds on the earlier ones deliberately (part 2's structured output is used in part 3's tool usage), so I suggest going through them sequentially. This first article lays the foundations and performs a simple interaction with the model.

A companion repository (https://github.com/BjornvdLaan/ai-engineering-articles-code-samples) contains all the code samples so that you can try it yourself as well. The examples use Anthropic's models. If you prefer a different provider (or locally hosted models) then I am sure your favorite chatbot can help adjust the configuration for you. No worries, LangChain's libraries are (mostly) model-agnostic.

The mental model for thinking about models

Before getting into the code, let's build an accurate mental model of what models actually are. Not how neural networks or transformers work internally - we will leave that to smart scientists. What you as an aspiring AI engineer need is the operational mental model: what this thing consumes, what it costs, and what you can control. In its simplest form, a model is simply this interface:

f(list of input messages) -> output message

That's it. An important detail is that the input is a list of messages, not just the last prompt you gave. Each call to the model typically includes a system message ("application configuration"), the whole conversation so far ("the current state"), and the latest prompt. Given all these preceding messages, the model produces the next message in the conversation.

If you have used Codex, Claude Code or other "harnesses" then this might sound weird. You see it searching the internet, memorizing facts from previous sessions, keeping a todo list, reading documents you give it, writing and editing a codebase. All that magic is application logic built around the model by AI engineers, but all the model sees is just a list of input messages.

Tokens: the unit of everything

Messages are sent to a model as tokens, not characters or words. A component called a tokenizer first splits the text into tokens and assigns each a numeric token ID. Common words are often a single token, while uncommon words may be split into several. The model never sees the original text: it only processes these token IDs.

Besides serving as a unit of computation, tokens are also important because they are the unit of billing. You pay per input and output token. Treat tokens the way you treat compute-seconds in a cloud bill. An AI feature is the first kind of backend endpoint where the cost of a single request is large enough to care about. An agent that makes five model calls per user request costs real money at production traffic volumes. This is also a new security risk. Traditionally, a DDoS attack primarily consumed infrastructure resources (CPU, memory, bandwidth, and autoscaled instances). Now, every request may also consume billable tokens, adding a second, potentially much larger, cost component. The attack even has its own name: denial of wallet.

The context window: the model's input limit

Each model has a maximum number of input tokens that it can process. This is called its context window, and can often hold hundreds of thousands of tokens. As discussed above, everything the model needs to answer a request must fit inside it: your system prompt, the conversation so far, retrieved documents, tool results, and any other context you provide. Think of it as the model's working memory. A larger context window lets you provide more information, but more context is not always better. As irrelevant information accumulates, the model has more distractions and important details become easier to miss. You will see this called context rot. More input tokens will also cost you more money! This is why "just paste everything in" doesn't scale: context engineering is about quality over quantity.

Setup

Now that we got some background info, it's time to see the model at work. This setup is needed once for all code samples throughout this series.

Prerequisites: Python 3.11+, an API key at your favorite provider, and not even five euros of credit for the whole series.

Create a new project directory (or clone the companion repo) and activate the Python virtual environment:

python3 -m venv .venv
source .venv/bin/activate

Create requirements.txt:

langchain>=1.3,<2.0
langchain-core>=1.4,<2.0
langchain-text-splitters>=1.1,<2.0
langgraph>=1.2,<2.0
langgraph-swarm>=0.1
langchain-anthropic>=0.4
langchain-huggingface>=0.2
sentence-transformers>=3.0
langchain-mcp-adapters>=0.1
mcp>=1.9
fastapi>=0.115
uvicorn>=0.32
pydantic>=2.9
python-dotenv>=1.0

Run pip install -r requirements.txt. Most of these packages are for later parts (retrieval, MCP, the web layer), but installing them now means you never have to touch this file again.

Then create .env in the project root. This is where you'll store the API key (and optionally other configuration):

ANTHROPIC_API_KEY=sk-ant-...

The API key for Anthropic is obtained from platform.claude.com.

Your first call

Create 01_first_call.py:

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()

model = init_chat_model("anthropic:claude-sonnet-5")

response = model.invoke(
    "A customer is disputing a €480 online card payment, claiming they "
    "never made it. In two sentences, what are the two most important "
    "pieces of evidence I should gather before responding?"
)

print(response.content)
print("\n--- metadata ---")
print(response.usage_metadata)

Run it: python3 01_first_call.py.

Two details to note in the code sample. First, init_chat_model is LangChain's provider-agnostic constructor. The string "anthropic:claude-sonnet-5" could also be "openai:gpt-5.5" or "ollama:llama3.3" with the rest of your application code staying the same. This is the same reason software engineers program against abstractions instead of concrete implementations: when you need to switch providers for cost, capability, or compliance reasons, you change configuration rather than redesign your system.

Second, look at usage_metadata in the response. It shows input and output tokens for that call. These numbers determine the cost. Production AI systems monitor them the same way traditional applications monitor database queries, memory usage, and other resource consumption. You want a Grafana dashboard on these numbers.

Below is an example of what the output could be. We'll run it twice.

$ python3 01_first_call.py

The two most critical pieces of evidence are:

1. **Transaction authentication records** - Check whether the payment was verified through 3D Secure/Strong Customer Authentication (SCA), as successful two-factor authentication significantly shifts liability away from you and toward the cardholder or their bank.

2. **Delivery/fulfillment proof** - Gather evidence that the goods or services were delivered to the address or account linked to the cardholder, such as signed delivery confirmation, IP address logs, or account activity showing the customer used what was purchased.

--- metadata ---
{'input_tokens': 45, 'output_tokens': 119, 'total_tokens': 164, 'input_token_details': {'cache_read': 0, 'cache_creation': 0, 'ephemeral_5m_input_tokens': 0, 'ephemeral_1h_input_tokens': 0}}

$ python3 01_first_call.py

Pull the transaction's authentication data (AVS/CVV match results, and whether 3-D Secure/EMV 3DS was used with a successful cardholder authentication, e.g., SCA challenge completion) and the device/IP/geolocation and behavioral data captured at checkout (billing/shipping address match, device fingerprint, past purchase history from that account) to establish whether the legitimate cardholder likely completed the transaction. Also gather delivery confirmation or service usage records (proof of delivery, IP login after purchase, downloads, or usage logs tied to the account) to show the goods or services were actually received or accessed by the customer.

--- metadata ---
{'input_tokens': 57, 'output_tokens': 209, 'total_tokens': 266, 'input_token_details': {'cache_read': 0, 'cache_creation': 0, 'ephemeral_5m_input_tokens': 0, 'ephemeral_1h_input_tokens': 0}}

Why the output may be different every time

As you can see, running 01_first_call.py a second time gives us a different answer. Ask it something it can't actually know, and it might invent an answer with total confidence. Both behaviors come from the same place: how the model produces text.

Models generate text one output token at a time. At each step, the model predicts a probability for every possible next token, samples one, appends it to the input, and repeats to get the next output token. Two consequences follow from that single mechanism, and both matter for everything we build after this.

The first is non-determinism. The model does not simply pick the most likely token every time. Rather, it samples from the distribution so the same input can produce a different output on the next call. How much freedom the sampling has, and whether you can influence it at all, differs per model. Some offer a temperature setting. Working with this non-determinism is a key different with 'normal' software engineering.

The second is hallucination, the stories you've probably heard about models making things up. This traces back to how the model was built. Its designed to predict the next token, not to distinguish what is actually true. The model isn't retrieving facts; it's generating plausible text. If a question has a correct answer but the model lacks the necessary information, it may confidently produce a convincing but incorrect one. Model creators apply 'post-training' to teach the model to follow instructions, use specific formats, refuse some requests, and admit uncertainty more often. But it still does not give the model an internal fact checker.

Notice that these are two sides of the same coin: a model that samples plausible tokens rather than retrieving correct facts. The fix is usually not a better prompt. It is to build a system around the model that constrains its output and gives it the right information when correctness matters. That system is exactly what the rest of this series builds, and it's what AI engineers do.

You did it!

Great job! You just took your first steps into AI engineering. We learned some theory and made our first model call. Obviously, this is not super useful yet. If your backend calls this model then you never know what kind of output to expect. The output message is unstructured and, therefore, cannot easily be parsed into an object like we do with JSON responses. The next article will do exactly that: making sure the model's output follows a structure that your code can count on.

Find all code samples in the companion repo here: https://github.com/BjornvdLaan/ai-engineering-articles-code-samples

Green Software is Event Driven and Carbon Aware, with Kubernetes and KEDA

BjornvdLaan — Thu, 07 Sep 2023 15:41:17 +0000

Did you know that data centres emit more carbon than aviation or shipping? And it is not slowing down, with some research forecasting that computing will contribute 14% of global emissions by 2040. Even worse, a significant portion of these emissions is wasted on non-production resources running outside office hours, always having your application scaled out for peak hours. This is not only bad for the planet but also for your cloud bill.

How can we do better? Instead of having a fixed number of application replicas, we should match the scale of our applications to:

Current events, for example, in terms of incoming web requests or messages in a Kafka topic. Sizing your application based on how much work it has alleviates overprovisioning and always-on resources.
The availability of clean energy. The electricity on the data centre's power grid comes from different sources, such as coal, nuclear, and solar. We should do more when the energy is cleaner, i.e., the carbon intensity is lower to reduce emissions further.

This article shows how we can make our applications greener by
making them event-driven and carbon-aware. We'll introduce these properties at the level of the Kubernetes cluster using KEDA (Kubernetes Event-Driven Autoscaler) and Microsoft's Carbon Aware KEDA Operator, so all applications in the cluster can benefit without becoming more complex themselves. All code examples are found in my GitHub repo.

The webshop that wants to become greener

Consider a simple webshop with only two applications: a webserver that serves the website to customers and an ordering system that takes care of all other things. When a customer places an order, the webserver puts a message with all necessary information on a queue and immediately thanks the customer for its purchase. The ordering system consumes these messages asynchronously, checks the warehouse, dispatches the delivery, and then sends a confirmation email to the customer.

Even though the number of customers varies throughout the day, both applications are currently deployed with a fixed number of replica pods appropriate for busy moments. Below is a graph with the number of messages in the queue and the ordering system replicas in the cluster. The red line indicates the number of messages on the queue, and blue shows the number of pods.

For simplicity, we assume that the replicas and messages at a specific y-coordinate would be exactly in balance, i.e., precisely enough replicas for that number of messages. In that case, the area under the blue and red lines relates to the total and useful resource consumption, respectively. The difference is wasted resources.

The webshop takes pride in its sustainable reputation and also wants to cut back on cloud costs. The webshop owner spoke with customers and learned that while the webserver should always be available and fast, it's okay if the asynchronous email confirmation arrives a bit later.

The webshop owner wants to scale the ordering system based on the queue size and whether clean energy is available. In other words, when carbon intensity is low, and many messages are waiting in the queue, the ordering system can go full speed:

If, on the other hand, either the carbon intensity is high or the number of messages is low, we should balance the number of replicas:

Let's explore how we can achieve this with KEDA.

Event-driven scaling of pods

We want to move away from an always-on ordering system with too many replicas. To dynamically scale the ordering system based on the number of messages in the queue, we use KEDA (Kubernetes Event-Driven Autoscaler), a Kubernetes operator that can scale our application based on 'events'. In our case, 'events' refer to the messages on the queue, but many more events are supported. With the KEDA Controller installed in the cluster, we can dynamically scale the ordering system by creating a ScaledObject:

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: ordering-system-scaler
spec:
  scaleTargetRef:
    name: ordering-system

  pollingInterval: 5
  cooldownPeriod: 5
  idleReplicaCount: 0
  minReplicaCount: 1
  maxReplicaCount: 10
  fallback:
    failureThreshold: 4
    replicas: 5

  triggers:
    - type: aws-sqs-queue
      authenticationRef:
        name: ordering-system-triggerauth
      metadata:
        queueURL: <AWS-DOMAIN>/000000000000/ordering-queue
        awsEndpoint: <AWS-DOMAIN>
        queueLength: "50"
        awsRegion: "eu-west-1"
        identityOwner: pod

The YAML above consists of roughly three parts:

We must tell the ScaledObject which application we want to scale. This is done through the scaleTargetRef key. In this case, the target ordering-system is the Deployment that manages the ordering system replicas.
Then, we configure how we want to scale the application. minReplicaCount and maxReplicaCount determine the range of scaling, and we can configure many other properties as well.
Finally, we tell which event(s) to scale on in triggers. Here, we defined the AWS SQS Queue trigger and told it to strive for 50 messages in the queue.

The graph from earlier changes as follows:

The difference in area under both lines is much smaller now, indicating more efficient resource usage.

What I really like about KEDA is that it does not try to reinvent the wheel. Instead, it reuses Kubernetes' built-in HPA (Horizontal Pod Autoscaler). Since Kubernetes 1.10, the HPA can use external metrics based on External Metrics Servers. KEDA 'simply' implements an External Metrics Server that exposes data about the triggers we define in our ScaledObject.

Carbon-aware limits to scaling

With KEDA's event-driven scaling, our resource consumption already changed drastically. Still, the ordering system does not take into account the current availability of clean energy. If the webshop is busier on moments with less clean energy, we can still emit much more than we would like. Let's see how we can add carbon-awareness using the Carbon Aware KEDA Operator. This Kubernetes operator limits how far KEDA can scale our consumer based on the carbon-intensity. This operator essentially sets a ceiling for allowed maxReplicas in KEDA's ScaledObject based on the carbon intensity of the current time, allowing more scaling when carbon intensity is lower. Once the operator is installed, we can configure the scaling limits through a custom resource definition:

apiVersion: carbonaware.kubernetes.azure.com/v1alpha1
kind: CarbonAwareKedaScaler
metadata:
  name: carbon-aware-ordering-system-scaler
spec:
  maxReplicasByCarbonIntensity:            
    - carbonIntensityThreshold: 237
      maxReplicas: 10
    - carbonIntensityThreshold: 459
      maxReplicas: 6
    - carbonIntensityThreshold: 570
      maxReplicas: 3

  kedaTarget: scaledobjects.keda.sh
  kedaTargetRef:
    name: ordering-system-scaler
    namespace: default

  carbonIntensityForecastDataSource:       
    mockCarbonForecast: true               
    localConfigMap:                        
      name: carbon-intensity
      namespace: kube-system
      key: data

Similar to how the ScaledObject targeted a Deployment through its scaleTargetRef, this CarbonAwareKedaScaler targets a ScaledObject defined in kedaTargetRef. This makes for a neat layered approach where each layer is unaware of the layer above.

The essential part of the YAML definition above is the maxReplicasByCarbonIntensity key, which lets us configure increasingly strict scaling limits based on the availability of clean energy.

The operator retrieves the current carbon intensity from the ConfigMap we define under the carbonIntensityForecastDataSource key, but actually filling the ConfigMap is not something it can do for you. Fortunately, Microsoft created a companion operator called the Carbon Intensity Exporter, which builds on the Green Software Foundations's CarbonAwareSDK to provide carbon intensity data in the Kubernetes cluster. However, if you just want to play around then you can also set the mockCarbonForecast key to true. The operator will then create a ConfigMap with fake carbon intensity data for testing purposes.

With the Carbon Aware KEDA Operator configured in our cluster, we further decrease our webshop's carbon emissions by doing more when energy is cleaner:

In the second peak, the messages are consumed at a slightly slower pace because carbon intensity is too high at that moment. The customer might need to wait a few more seconds for the confirmation email but likely won't even notice.

Conclusion

Computing significantly contributes to global carbon emissions, and digitalisation shows no signs of slowing down. We should, therefore, become more efficient in our resource usage. This article shows how we can use KEDA to become more event-driven and carbon-aware and reduce overprovisioning of our applications.

Carbon Aware Software: Three Demand Shaping Strategies, implemented in Kotlin

BjornvdLaan — Mon, 10 Jul 2023 08:22:31 +0000

Software engineers have a largely untapped opportunity to contribute to a greener world. In this article, I discuss how we can schedule asynchronous tasks to minimise carbon emissions.

Electricity comes from various sources, each with different carbon emissions. Depending on the source, a certain amount of carbon is emitted for each generated kilowatt hour (kwh). Clean and renewable sources like wind, solar, and hydroelectric power emit minimal carbon, while fossil fuel sources emit quite a bit more. The amount of carbon emitted by an energy source is called its carbon intensity, measured in grams of carbon per kwh.

Ideally, you would plug your computer or data center directly into a wind farm so that your software is powered with zero carbon intensity. However, it is not feasible for most individuals to directly connect to wind farms; instead, they rely on power grids that receive electricity from a mix of sources. Once connected to a grid, you cannot control the specific sources supplying your electricity. You simply receive a blend of all the current power sources within the grid, including both lower- and higher-carbon sources. Therefore, the carbon intensity of the electricity you consume reflects a combination of all the sources on the grid at that moment.

Carbon awareness is the idea of doing more when more energy comes from low-carbon sources and doing less when more energy comes from high-carbon sources. One approach to building carbon-aware software is called demand shaping: we shape our computation to match the existing supply. When the carbon cost of running your application becomes high, shape the demand to match the supply of carbon.

Demand shaping (https://learn.greensoftware.foundation/carbon-awareness)

Demand shaping sounded a bit abstract to me when I first heard about it. Therefore, I decided to build carbon-aware software myself using Kotlin and Spring Boot. This article shows my plans and insights. You can find my code here at GitHub.

Our implementation

We'll simulate a scenario where a producer periodically puts task descriptions on a queue, describing a computationally intensive task that some consumer should execute asynchronously. Imagine this being some tasks that are not time-critical but should be performed somewhere in the future. Before executing a task, the consumers will first reach out to an API called the CarbonAwareSDK, which provides information about the carbon intensity at a certain time and place.

Each consumer implements a different strategy for processing events based on the retrieved carbon intensity:

Threshold consumer only consumes new tasks from the queue if the carbon intensity is lower than a certain threshold.
Adaptive velocity consumer, which takes longer breaks between tasks based on the carbon intensity.
Forecast-based consumer, which runs its daily task at the optimal time in the next 24 hours.

CarbonAwareSDK API

CarbonAwareSDK helps us retrieve information about the carbon intensity at a certain time and place. Documentation on the available endpoints is found here. For our purposes, we use two of its functionalities.

Current Carbon Intensity

We can request the current carbon intensity at a specific location through the following URL:

https://carbon-aware-api.azurewebsites.net/emissions/bylocation?location=eastus

The response contains a field rating: the carbon intensity, measured in grams of CO2 equivalent per kilowatt hour.

[
    {
        location: "PJM_ROANOKE",
        time: "2023-06-26T17:05:00+00:00",
        rating: 588.30930389,
        duration: "00:05:00"
    }
]

Forecasted carbon intensity

We can also ask the service to forecast carbon intensities between two timestamps, indicated by path variables dataStartAt and dataEndAt.

https://carbon-aware-api.azurewebsites.net/emissions/forecasts/current?location=northeurope&dataStartAt=2023-07-19T15:00:00Z&dataEndAt=2023-07-20T16:00:00Z&windowSize=10

One really cool feature is that it also gives us the most moment in that time frame where carbon intensity is lowest.

[
    {
        generatedAt: "2023-06-26T17:00:00+00:00",
        requestedAt: "2023-06-26T17:03:33.4552082+00:00",
        location: "eastus",
        dataStartAt: "2023-06-26T17:05:00+00:00",
        dataEndAt: "2023-06-27T17:00:00+00:00",
        windowSize: 0,
        optimalDataPoints: [
            {
                location: "PJM_ROANOKE",
                timestamp: "2023-06-27T07:45:00+00:00",
                duration: 5,
                value: 497.9432056032178
            }
        ],
        forecastData: [
            {
                location: "PJM_ROANOKE",
                timestamp: "2023-06-26T17:05:00+00:00",
                duration: 5,
                value: 607.0114067773352
            },
            {
                location: "PJM_ROANOKE",
                timestamp: "2023-06-26T17:10:00+00:00",
                duration: 5,
                value: 606.2321824825342
            },
            {
                location: "PJM_ROANOKE",
                timestamp: "2023-06-26T17:15:00+00:00",
                duration: 5,
                value: 605.3570016963745
            },
            ... and more ...
        ]
    }
]

Threshold consumer

This first consumer is called every 5 seconds using Spring Boot's @Scheduled annotation. Upon invocation, the current carbon intensity is retrieved. Tasks are only retrieved and executed if the carbon intensity is lower than the threshold MAXIMUM_CARBON_INTENSITY.

@Component
class ThresholdConsumer(
    val taskProcessor: TaskProcessor,
    val carbonIntensityService: CarbonIntensityService,
    val queueService: QueueService
) {
    @Scheduled(fixedDelay = 5, timeUnit = TimeUnit.SECONDS)
    fun consume() {
        val currentCarbonIntensity = carbonIntensityService.getIntensity().rating

        if (currentCarbonIntensity < MAXIMUM_CARBON_INTENSITY) {
            val task = queueService.receiveTask()
            task?.let {
                val result = taskProcessor.process(it)
                LOGGER.info("Task ${it.id} executed successfully with threshold strategy. Result = ${result}.")
            }
        } else {
            LOGGER.info("No task executed with threshold strategy because carbon intensity ${currentCarbonIntensity} is too high (threshold = ${MAXIMUM_CARBON_INTENSITY}).")
        }
    }

    companion object {
        private const val MAXIMUM_CARBON_INTENSITY = 400
        private val LOGGER = LoggerFactory.getLogger(ThresholdConsumer::class.java)
    }
}

Adaptive velocity consumer

The adaptive velocity consumer is different from our first consumer, who will always continue consuming messages. It just works harder when the carbon intensity is lower. After reading and executing a task, the consumer sleeps 10 milliseconds for each gram of CO2 emitted for a kwh at this moment. It will therefore sleep less if more clean energy is available.

@Component
class AdaptiveVelocityConsumer(
    val taskProcessor: TaskProcessor,
    val carbonIntensityService: CarbonIntensityService,
    val queueService: QueueService
) {
    @EventListener
    fun consume(event: ApplicationReadyEvent) {
        while (true) {
            val currentCarbonIntensity = carbonIntensityService.getIntensity()

            val delay = currentCarbonIntensity.rating * 10L
            Thread.sleep(delay.toLong())

            val task = queueService.receiveTask()
            task?.let {
                val result = taskProcessor.process(it)
                LOGGER.info("Task ${it.id} executed successfully with adapted speed strategy. Result = ${result}.")
            }
        }
    }

    companion object {
        private val LOGGER = LoggerFactory.getLogger(AdaptiveVelocityConsumer::class.java)
    }
}

Forecast consumer

The first two consumers make decisions based on the current carbon intensity, which is quite similar to a greedy algorithm. Our final consumer aims to be smarter than that. Instead of asking CarbonAwareSDK for the current carbon intensity, the consumer requests a forecast for the coming day. The response looks as below. The most important field is optimalDataPoints, which contains the best moment to perform our task.

We schedule the consumer to be run once per day. During invocation, it retrieves the best moment to execute a task that day and schedules it using Spring Boot's TaskScheduler.

@Component
class ForecastingConsumer(
    val taskScheduler: TaskScheduler,
    val taskProcessor: TaskProcessor,
    val carbonIntensityService: CarbonIntensityService,
    val queueService: QueueService
) {
    @Scheduled(fixedDelay = 1, timeUnit = TimeUnit.DAYS)
    fun consume() {
        val task = queueService.receiveTask()

        task?.let {
            val optimalIntensity = carbonIntensityService.getForecastedOptimalIntensity().optimalDataPoints.first()

            taskScheduler.schedule(Instant.from(DateTimeFormatter.ISO_INSTANT.parse(optimalIntensity.timestamp))) {
                val result = taskProcessor.process(it)
                LOGGER.info("Task ${it.id} executed successfully with forecasting strategy. Result = ${result}.")
            }
            LOGGER.info("Task ${it.id} scheduled successfully at ${optimalIntensity.timestamp} with forecasting strategy.")
        }
    }


    companion object {
        private val LOGGER = LoggerFactory.getLogger(ForecastingConsumer::class.java)
    }

    fun TaskScheduler.schedule(moment: Instant, task: Runnable) =
        this.schedule(task, moment)
}

Insights

These three consumers give us an initial idea of how we can build carbon-aware software. However, is the application level the right place to implement this logic? Yes, we manage to do more work when we emit less carbon, but we still have these applications running and methods being invoked. All these network requests between the consumers and the CarbonAwareSDK also costs electricity. In addition, the application becomes more complex with this shaping logic, degrading its carbon efficiency. The three consumers each also have specific downsides:

The threshold consumer cannot guarantee that tasks are ever executed. Some locations in the world are still very reliant on coal so that the threshold might be too low.
The adaptive velocity consumer reads faster when cleaner energy is available but could still process a lot of its tasks during very dirty times of the day. Even worse, it could slowly work through the whole queue during high carbon intensity hours and then have nothing left to do when it wants to speed up.
The forecasting consumer solves the downsides of the previous two, but scheduling a task at the application level is a risky move. At any moment, the application could be killed as part of a rolling update or some error unrelated to the task. The scheduling is then lost. The task description was already removed from the queue, so it won't be executed anymore altogether.

This all gets even messier if we have multiple replicas of the same application, which is not uncommon in production environments.

Shift down to the platform

I believe the forecasting consumer is the most promising of the three strategies. However, the application is not the right place to implement this functionality. Instead, I imagine having a custom resource in Kubernetes that allows us to schedule tasks based on the carbon intensity.

apiVersion: carbonaware/v1
kind: CarbonAwareDailyJob
metadata:
  name: my-day-job
spec:
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: consumer
            image: task-consumer:1.0
          restartPolicy: OnFailure

This CarbonAwareDailyJob custom resource and its associated custom controller, reach out to the CarbonAwareSDK (which you can also host on-premise!) for the forecast of the next 24 hours. The Job described by the jobTemplate is then executed at that optimal moment, very similar to the forecasting consumer we saw earlier. An added benefit is that our custom controller centralises all calls to the CarbonAwareSDK API, saving network traffic and carbon.

Conclusion

Implementing carbon-aware software with demand shaping techniques has the potential to reduce carbon emissions and promote sustainability in our digital landscape. In this article, we applied demand shaping at the application level. However, I learned that demand shaping might be better implemented at a lower level. Green software patterns like demand shaping are still under active development. Continued research, collaboration, and adoption of carbon-aware software solutions are vital to realising these benefits and creating a greener future.

3 steps towards a greener IT landscape

BjornvdLaan — Mon, 03 Jul 2023 08:12:00 +0000

Businesses all over the world are moving towards a greener future. Steel factories look at green steel, dairy companies offer plant-based alternatives, office roofs get solar panels, and cloud providers make sustainability pledges. However, even though every company nowadays uses software and hardware of some form, the IT landscape is often not considered in those sustainability plans, even though data centres have substantial carbon emissions (more than aviation and growing!). Sustainability is also not of mind when software and hardware developers talk about non-functional requirements; they think about performance and reliability but not so much about carbon efficiency.

I believe software engineers have a largely untapped opportunity to contribute to a greener world. And a big bonus: as we green our IT landscape, the applications in it also likely become cheaper to run, with better performance and resilience. Win-win!

In this article, we will take three steps into the world of green software engineering, with concrete examples along the way. The three steps are visualised below.

1. The Software Application

Running our applications causes carbon to be emitted. We will probably never be able to emit nothing at all, but we can work to minimise the amount of carbon emitted per unit of work. There are essentially two approaches to making applications more energy-efficient.

The first approach is to simply make our application consume less energy. How to achieve this exactly depends on your application. For web applications, we could look at the network communication between the frontend and the backend and identify three options:

Reduce the size of network packets by minifying the web assets or removing unused CSS definitions.
Reduce the distance those packets travel by introducing a CDN (Content Delivery Network).
Reduce the number of API calls needed to render the page.

We could even look at the programming language you pick. Five years ago, researchers in Portugal compared the energy consumption of several popular programming languages and showed how their execution time and memory usage influenced energy usage. They found that compiled languages (C, Go, Rust, Java) tend to be more energy efficient than interpreted languages (JavaScript, Python).

Alternatively, we can make our applications more intelligent. Our power grid is supplied with electricity from various sources. When the sun is shining and the wind is blowing, then more of the available energy comes from renewable sources. Based on that concept, we build our applications to be carbon-aware: do more when more energy comes from low-carbon sources and less when more energy comes from high-carbon sources. Imagine an application that reads a task description from Kafka and performs some CPU-intensive computation that adjusts its pace based on how clean the current energy supply is. This pattern is called demand shaping, as we shape the energy demand of our application based on the carbon intensity.

Instead of changing the pace (demand shaping), we can also shift to a different time and/or place. This is called demand shifting. Imagine that your company is in Australia, and you want to run computations during nighttime. You could run the workload in the Netherlands, where its daytime and where solar power is more readily available.

2. The Runtime Platform

Our software applications are often operated on runtime platforms such as Kubernetes, which makes them a great place to achieve more sustainability gains. Making improvements at the platform-level profits all applications running on it. The k8s community has some great initiatives to assist you. I'll give three examples here.

Suppose your company has separate Dev, Test, Acceptance and Production environments. In that case, you can already achieve significant emission reduction by turning off the Dev and Test clusters on weekends. You can automate this through kube-green, which automatically shuts down your deployments and cronjobs when you don't need them. An added benefit: you also promote a healthy working environment where developers take some rest outside business hours.

By default, Kubernetes scales workloads based on CPU and RAM utilisation. However, we can use components like KEDA to scale based on relevant demand metrics such as HTTP requests or queue length. This can help us reduce resource utilisation and, therefore, your carbon emissions. KEDA can even scale deployments down to zero.

Finally, we can adjust the Scheduler to assign pods to nodes based on the carbon intensity at that machine's location. Bill Johnson, Principal Software Engineering Manager, wrote an article about this topic here.

3. The Hardware

All software we build eventually runs on hardware, from servers in a datacenter to the phones and laptops of our end users. These devices need to be manufactured, and they also need to be dismantled eventually. The carbon that is emitted during the creation and disposal of a device is called embodied carbon. The bar chart below (credits: Green Software Foundation) shows that this embodied carbon is not a trivial factor.

Reducing the embodied carbon of servers requires embracing responsible lifecycle management, which involves implementing proper disposal and recycling methods for old servers. By refurbishing, reselling, or recycling them, we extend their useful life and prevent them from ending up in landfills.

Besides the embodied carbon, we should also consider how servers are used. Simply said: the more we utilise a server, the more efficient it can run our applications. We call this energy proportionality (more info here). Therefore, we can run our software using less electricity by running on as few servers as possible.

I believe that embodied carbon and energy proportionality are why cloud providers have an amazing opportunity to contribute to a better world. I am happy to see the first steps already being taken. Servers in data centres are often written off after four; all three big cloud vendors recently announced extending this lifespan in the light of sustainability. Also, as they manage the applications of multiple companies, they can intelligently allocate workloads in such a way that no servers are sitting idle, improving utilisation rates.

Besides making hardware more sustainable under the cloud-hood, cloud providers should also give tools to their customers to make greener choices. For example, AWS Lambda has a SnapStart option to decrease startup times. In the same way, we could have a CarbonAwareStart that applies demand shaping to Lambda executions. Moreover, cloud providers could inform customers about their emissions in the same way as they now give insight into the costs. What gets measured gets improved.

Conclusion

Green software invites us, software engineers, to make our IT landscape more sustainable. Engineers can play a vital role in protecting the environment by focusing on energy efficiency, reducing waste, and minimising carbon emissions. Joining the green software movement is an opportunity and a responsibility to build a better future together. Using sustainable practices, we can drive innovation, improve performance, and contribute to a resilient digital world across all industries. Let's work together to create a greener, more sustainable planet.

Interested? Check out the Green Software Foundation.

Image attributions

Cover photo by Matt Anderson on Unsplash.

The icons used in my visualisation are found on Flaticon.com.

Globe icons created by Uniconlabs - Flaticon
Lifespan icons created by Flowicon - Flaticon
Server icons created by Freepik - Flaticon
Development icons created by Design Circle - Flaticon
Application icons created by Smashicons - Flaticon

AWS announces AWS CarbonAwareBatch, the Future of Sustainable Cloud Computing

BjornvdLaan — Thu, 22 Jun 2023 08:24:49 +0000

Today, Amazon Web Services (AWS) announced the launch of AWS CarbonAwareBatch, a revolutionary new service that reimagines the way companies approach sustainable cloud computing. The solution not only prioritizes computing efficiency but also harnesses the power of environmental responsibility.

AWS CarbonAwareBatch, built on the robust framework of AWS Batch, is designed to enhance businesses' green credentials by intelligently scheduling batch jobs during periods of low-carbon intensity. It enables organizations to minimize the carbon footprint associated with their computing tasks, bringing a new level of environmental consciousness to the technology sector.

"Efficiency and innovation are key pillars at AWS. With CarbonAwareBatch, we're not only redefining these terms but also factoring in sustainability," said April McGreen, Tech Lead of AWS CarbonAwareBatch. "We are empowering our customers to make positive environmental choices, seamlessly aligning their technological needs with their environmental goals."

Key Features

In addition to harnessing low-carbon periods, AWS CarbonAwareBatch boasts a range of innovative features:

Carbon Footprint Dashboard: An interactive, user-friendly dashboard provides real-time data on the carbon emissions saved by utilizing the service. This provides businesses with tangible data to showcase their commitment to sustainability.
Carbon Credits: AWS CarbonAwareBatch calculates the carbon emissions saved and converts them into carbon credits, which businesses can use towards their sustainability goals or trade in carbon credit markets.
Sustainability Scorecard: This feature gives an in-depth analysis of a company's sustainable usage patterns and offers recommendations on how to optimize for further environmental impact reduction.
Machine Learning Optimizations: Leveraging machine learning, AWS CarbonAwareBatch forecasts power grid carbon intensity, allowing it to optimally schedule jobs to further minimize carbon emissions.

By leveraging AWS CarbonAwareBatch, businesses can now align their operations with the principles of a green economy, reducing their carbon footprint without compromising on efficiency or productivity.

The Future is Green

With AWS CarbonAwareBatch, Amazon Web Services continues to demonstrate its commitment to driving sustainable innovation. By integrating environmental consciousness into its core services, AWS is not just preparing for the future, but actively shaping it, ensuring that technology continues to serve as a force for good.

The future of cloud computing is not just about convenience and flexibility - it's also about stewardship of our planet. AWS CarbonAwareBatch is the vanguard of this new era, providing the tools businesses need to be at the forefront of the green revolution.

…

That would have been cool, right?

While this announcement and the service of AWS CarbonAwareBatch sounds like the perfect blend of technological innovation and environmental responsibility, this announcement is unfortunately not real. It's a projection of what could be possible in the future, generated by ChatGPT and myself.

As we continue to grapple with the realities of climate change, we hope to see companies like AWS pioneering more solutions like this - technologies that not only propel us forward but also protect and preserve our world.

We look forward to a day when AWS CarbonAwareBatch, or a service very much like it, moves from the realm of AI-generated possibilities into tangible reality. Until then, we encourage everyone - technologists, businesses, and consumers alike - to continue pushing for a more sustainable future.

Kotlin Multiplatform’s three levels of testing with Kotest

BjornvdLaan — Sat, 21 Jan 2023 15:48:19 +0000

Kotlin Multiplatform allows us to share common Kotlin code between various platforms, including JVM, JavaScript, iOS, Android, and native. This is useful when we have shared business logic relevant to all platforms. If functions in that common codebase require a platform-specific implementation, then Kotlin Multiplatform offers a mechanism of expected and actual declarations to essentially create placeholders that are implemented per platform. We examine this mechanism later in this article. A Kotlin Multiplatform project contains multiple modules called source sets to enable such code sharing. By default, each platform has two source sets, one for the code (jvmMain, jsMain) and one for the tests (jvmTest, jsTest) and additionally two source sets for common code (commonMain and commonTest) that are used in all main compilations of a project. These modules are structured in a hierarchy to provide any level of code sharing.

Source set hierarchy in Kotlin Multiplatform.

One use case is to share dependencies. Below we see an example where commonTest defines Kotest dependencies used for all platforms, and the JVM-specific dependencies are added to the jvmTest source set.

sourceSets {
    val `commonTest` by getting {
        dependencies {
            implementation(libs.kotest.assertions.core)
            implementation(libs.kotest.framework.engine)
            implementation(kotlin("test-common"))
            implementation(kotlin("test-annotations-common"))
        }
    }

    val jvmTest by getting {
        dependencies {
            implementation(libs.kotest.runner.junit5)
        }
    }
}

Each platform also has its own specific code so that, for instance, the JVM part of your application can implement a server, the front-end is built in JavaScript code, and the mobile apps are created based on the source sets for Android and iOS.

This article shows how we can test all three layers of code in a Kotlin Multiplatform project: common, expected / actual declaration, and platform-specific code. We use Kotest as our testing framework because it has built-in multiplatform support. The framework delegates testing to platform-specific engines, such as JUnit and Mocha. All code samples can be found on GitHub.

Common code

Code that is shared by all platforms by default resides in the commonMain source set. It wouldn’t make sense to write tests for it in every platform, so instead, we can use the commonTest source set to store the common test suite. Let’s say that we need a function that calculates the factorial of a number in source set commonMain:

package nl.bjornvanderlaan.factorial

fun factorial(number: Int): Int
    = (1..number).fold(1) { acc, i -> acc * i}

We can provide the test code in commonTest:

package nl.bjornvanderlaan.factorial

import io.kotest.core.spec.style.StringSpec
import io.kotest.matchers.ints.shouldBeExactly

class FactorialTest : StringSpec({
    "Factorial of 5 is 120" {
        factorial(5) shouldBeExactly 120
    }
})

As we can see, testing common code works the same as in ‘normal’ Kotlin projects. The cool part however is that this function can now be used in all possible platforms, which can be a huge bonus if we write all the business logic here.

Expected / Actual declarations

Sometimes, there are functions that we want to use in our common code, but they can be implemented much more efficiently with platform-specific code. For instance, Kotlin’s multiplatform standard library does not provide a way to encode a string in base64. We could roll our own implementation or look for an open-source library, but some of our target platforms already provide this functionality out of the box. Kotlin provides a mechanism of expected and actual declarations to tackle this situation. With this mechanism, a common source set defines an expected declaration, and platform-specific source sets provide the actual declaration corresponding to the expected declaration and the actual implementation. Besides functions, this mechanism also works for other constructs such as classes, enums, and annotations. Below we see an example of this mechanism for the JVM. The code is based on a tutorial from JetBrains.

/**
 * Source set: commonMain
 */
package nl.bjornvanderlaan.base64

expect fun encodeBase64(input: String): String

fun encodeGreeting(name: String): String {
    return "Greetings $name, from Kotlin Multiplatform!"
}

/**
 * Source set: jvmMain
 */

package nl.bjornvanderlaan.base64

import java.util.Base64

actual fun encodeBase64(input: String): String {
    return Base64.getEncoder().encode(input.encodeToByteArray()).decodeToString()
}

As we can see, the expected function in commonMain can be used by other functions such as encodeGreeting. Through this mechanism, we can write common code where it makes sense and let leave placeholders where we expect the platforms to deliver an actual implementation. Similar to the common code, we can write our test in commonTest:

/**
 * Source set: commonTest
 */
class Base64Test: StringSpec({
    "String is properly encoded to Base64" {
        val input = "Kotlin is awesome"

        val encodedInput = encodeBase64(input)

        encodedInput shouldBe "S290bGluIGlzIGF3ZXNvbWU="
    }
})

When we let Gradle test our project, it automatically executes this test for all platform-specific variants of the expected function. So even though we have multiple actual implementations, we do not need to repeat our test suite.

Specific Function

Finally, there are parts of the application that we want to build for specific platforms only. Because source sets are organised hierarchically, we can choose which test source we want to put our tests in. When we make a web application, we want the server code to be written specifically for the JVM, and the JavaScript source set would contain the front-end code. The tests are then likely put in jvmTest and jsTest source sets, respectively. We can use Kotest to write tests like we would for single-platform Kotlin projects. The advantage of Kotest is that we have one unified testing framework, and we can also use all helper functions and custom assertions from parent source sets. Let’s say that we want to write a function for the JVM platform that calculates the factorial of a factorial of a number. Note that the value is already bigger than what Integer can handle if the number is 4. Luckily, as this function is specific to the JVM, we can use BigInteger for this purpose.

/**
 * Source set: jvmMain
 */
package nl.bjornvanderlaan.factorialoffactorial

import nl.bjornvanderlaan.factorial.factorial
import java.math.BigInteger

fun factorialOfFactorial(number: Int): BigInteger {
    val bigNumber = factorial(number).toBigInteger()

    var currentNumber = BigInteger.ONE
    var accumulator = BigInteger.ONE

    while(currentNumber <= bigNumber) {
        accumulator *= currentNumber
        currentNumber++
    }
    return accumulator
}

kotlin

The function above also shows how the jvmMain source set can use the factorial function from its parent commonMain in the same way it can use its own functions. The function above can be tested with Kotest in the same way as we have seen in the other examples:

/**
 * Source set: jvmTest
 */
package nl.bjornvanderlaan.factorialoffactorial

import io.kotest.core.spec.style.StringSpec
import io.kotest.matchers.shouldBe
import java.math.BigInteger

class FactorialOfFactorialTest: StringSpec({
    "Factorial of factorial of 4 is 620448401733239439360000" {
        factorialOfFactorial(4) shouldBe BigInteger("620448401733239439360000")
    }
})

This function and its tests are defined in the JVM-specific part of our application and can, therefore, benefits from classes like BigInteger.

Intermediate levels

The examples above assume only two levels of hierarchy: the common parent and its platform-specific children. In practice, we might introduce more levels. For instance, if we also build a mobile application for iOS and Android, we might have an intermediate source sets mobileMain that depends on the commonMain source set and acts as the parent for iosTest and androidMain. In that case, the mobileTest source set can contain tests for functions implemented in mobileMain or expected functions implemented in child source sets.

Conclusion

Kotlin Multiplatform helps us reduce the time spent on writing and maintaining the same code for different platforms, and Kotest helps us to unify our testing efforts by providing a multiplatform framework. Multiplatform projects are structured as a hierarchy of source sets. The rules for testing are the same at every node: Kotest executes our tests on either the implementation in the main source set itself or fetches the actual implementations from child source sets. As such, Kotlin Multiplatform offers us the flexibility to create a super DRY codebase and test suite, regardless of the number of platforms we want to support.

Property-based testing with Kotlin and Kotest: the Christmas edition

BjornvdLaan — Sat, 21 Jan 2023 15:38:16 +0000

This article was previously posted as part of the 2022 Java Advent calendar.

Christmas is coming, and Santa and the elves are working around the clock to build all toys. After a long day in the toy workshop, the elves want to spend their hard-earned NorthPoleCoins on jingle juice and sweets at the local pub. They always go in groups and often order collectively. They use a special application to keep track of payments so that every elf pays its fair share. We see the service that settles these payments below.

To preserve peace on the north pole, we want to add some tests to ensure that this service works as intended. One crucial property of the method above is that, at any point in time, the balances of all elves add up to zero. Otherwise, coins would be created or lost. A unit test for this method could look something like the one below.

First, the scenario is set up by initialising the objects with example values. Then, we invoke the method under test and write assertions to verify that the output is as expected. Of course, it doesn’t stop with this single test case. We generally write additional, similar tests to sufficiently cover all scenarios and edge cases. And when we see that all the test scenarios we came up with are passing, we can relievedly commit or deploy our code.

Downsides of example-based testing

If we take a step back, we see that many of these test cases are quite similar, except that each uses a slightly different example. There are a few downsides to this approach. First of all, developers need to spend more time developing and maintaining these test cases. This includes coming up with realistic examples, which requires a thorough understanding of the domain. New developers on the team might also have a hard time grasping the actual intent of the test. It takes time to see what angles are covered for each method and what is missing or redundant. And despite all this effort from the team, only a small portion of all possible examples are covered by the tests they wrote. Covering all combinations of inputs is just not feasible. And even worse, the scenarios covered are the same every run: essentially coupling between scenario and test suite.

When we implement our methods, we want the logic to be generic and not tied to specific cases. The code should have certain properties for all possible inputs. The tests we write for it are more the opposite: we write tests that are example-specific and not generic. The code and tests are designed on different levels of abstraction, where properties are more abstract than concrete examples.

Property-based testing to the rescue

But what if we could write tests like we write our code? Property-based testing (short: PBT) can help us out here. Instead of defining the concrete examples to test, you specify the domain (strings, positive integers, etc.) and let the framework generate many input values, including edge cases. Each test is executed multiple times with different possible inputs during a test run. Property-based testing allows us to write one test that tests all of the desired inputs, including typical edge cases, which allows for more efficient test writing and ensures that a test passes for all possible values within the domain.

Below we see how our test could look as a property-based test. The scenario function is replaced by Kotest’s checkAll, taking as its inputs a set of generators. The class containing all generators is called Arb, which is short for arbitrary. A predefined number of examples is generated for each test run, a thousand by default in Kotest. The rest of the test remains completely the same, except that we added some code to transform the generated values to the entities and values of our domain.

Our example-based tests previously all passed, which might lead us to believe that this property-based test will also be happy. Our test cases represented all possible paths, right? However, we are actually met with a failure this time. You might think that a property-based test is more difficult to debug because of its generated input values, but we did not talk about PBT’s superpower. Kotest not just mentions the failing example if the test fails, it starts ‘shrinking’ the input values. If the test fails with integer value 8 as input then this value is shrunk to value 7, and the test is retried. If the test failed when we tried to input a list of three strings integer then the list is shrunk to two strings. Shrinking continues until the test passes again, and the goal is to find the minimal reproducible example for which the test fails, which helps find the bug. Let’s have a look at the stack trace of our run.

The test failed when we tried to divide (a rounded) 77.93 NorthPoleCoins over four elves, and Kotest started shrinking both inputs. Interestingly, both shrinks uncovered a different bug for us. Dividing the coins led to a rounding error, while an empty list caused an exception to be thrown. We might not have thought of it when we created our examples, but now we can adapt our code based on these results.

When we make ourselves as developers responsible for coming up with realistic examples, we risk such bugs slipping through the cracks. Property-based tests make sure that all scenarios can be verified.

Custom generators

The property-based test we saw above is less readable than we would like because of the logic we added to transform the values from the built-in Arb generators to be suitable for our test. Kotest allows us to define custom Arb’s through the arbitrary builder, and we can use Kotlin’s extension functions to use these custom generators in the same way as their built-in counterparts.

Where Kotest knows how to shrink built-in generators, the shrinking behaviour for custom generators can be very dependent on your domain. Therefore, the arbitrary builder allows us to provide a shrink function that returns a list of values we deem to be a shrink result of the current generated value. Let’s have a look at what this could look like:

We ended up with a more concise test than the example-based test that we started with, but it covered many more examples than before.

Conclusion

In this article, we’ve seen what property-based testing is and how it differs from the more common example-based testing. Where defining representative examples in example-based testing is the programmer’s responsibility, property-based testing leaves this to generators so that the tests do not become partly dependent on the chosen examples. Property-based testing allows us to generalize concrete examples to focus on the characteristics of the code that underlie these examples, resulting in a cleaner and more compact test suite that is easy to maintain and better exposes subtle bugs. In this way, we bring what we test closer to what we claim to test. However, it is also important to note that property-based testing is not intended to replace example-based unit tests completely. It’s an addition to our testing arsenal that allows us to cover every possible input value with a single test to reveal bugs we might not come up with ourselves.