Executive Summary

The purpose of this document is to teach a learner who has finished the four Draughts AI courses how to build an agent framework of their own and use it with a language model. An agent here means a language model that is given a goal, a set of tools and a loop, and decides for itself which tools to call until the goal is met. The scope covers the parts of such a framework, written from scratch in Python: a model adapter, tools made from ordinary functions, the agent loop, checks on every answer, a notebook for memory, teams of agents, a person's approval, traces, evaluation and a nightly scheduled run. The course first applies the framework to draughts, then to a task unrelated to draughts, and finally maps the same ideas onto an established library. Hosted model services, web pages for the agents and agents that change files without approval are out of scope.

All code in this course was built and tested before the course was written, using the Course 3 application as its starting point and parts of Courses 2 and 4. As in Course 2, no language model could be downloaded in the environment used to write the course, so every agent was tested against a stand-in that follows fixed rules and can be told to make mistakes on purpose. Every part of the framework that does not depend on a model's judgement was therefore tested directly, and the course gives each learner the same evaluation to measure their own model. The matters assessed are whether each agent uses its tools before answering, whether its answers can be checked automatically, whether a person stays in control of costly and lasting actions, and whether the framework carries over to work that has nothing to do with draughts. Sources are listed in the appendix.

Module 0: Before You Start

What you will build

In Course 2 the coach answered one question at a time: the player asked, the model called a tool, and the model answered. In this course you take the next step and build agents: language models that are given a goal rather than a question, and that decide for themselves which tools to call, in what order, until the goal is met.

You will build the framework itself, a small Python package called agentkit, so that nothing about agents stays a mystery. Then you will use it three times, and finally set it to run every night on your server.

• A team of draughts agents. A director runs a tournament between the computer players from Courses 1 and 3, an analyst finds the biggest mistakes in the games, and a reporter writes a short report that a person approves before it is saved

• An agent for any project. A helper that answers questions about a folder of files, citing the file and line for each claim. It has nothing to do with draughts, which shows the framework is general

• The same agent in a library. The director again, built with LangChain, so you can see exactly what a library does for you and what it does not

What you need

• The finished Courses 1 to 4, and Ollama with a tool-capable model from Course 2

• Python 3 and Node.js on your computer, as before. The framework itself needs nothing beyond the Python standard library

• The GitHub course, if you want to deploy with git pull as it describes

Start by making a copy of your Course 3 folder called draughts-agents, and open the copy in VS Code. Course 3 is the right starting point because it has all five computer players: easy, medium and hard from Course 1, and learned and neural from Course 3. Then copy two files from your other courses into the new folder.

• static/js/notation.js from your Course 2 folder, so moves can be named as in Course 2

• ratings.py from your Course 4 folder, so the tournament can give Elo ratings exactly as Course 4 does

Your copy includes the venv folder, but a virtual environment cannot be copied: the copy still points at the folder where it was first made, so anything you install would go there instead. In the copy, delete the venv folder, then create and activate a new one exactly as in Course 1, and install the packages with pip install -r requirements.txt.

Module 1: What Makes an Agent

From a coach to an agent

The coach in Course 2 already had the ingredients of an agent: a model, tools, and a loop that ran the tools the model asked for. Three things turn that into an agent.

• A goal instead of a question. "Run a tournament and report on it" needs several tools, in an order that depends on what each one returns

• More freedom. The model chooses which tool to call next, and how many times, within limits that you set

• More at stake. A coach that is wrong gives a bad tip. An agent that is wrong may spend an hour of computing time, or save a misleading report that people rely on

That last point shapes everything in this course. The more an agent can do, the more carefully its work must be checked, and the clearer it must be where a person stays in charge.

The parts of an agent framework

Every agent framework, whether you write it or download it, has the same parts. You will write each one in the next modules.

Figure 1 shows the agent loop that ties these parts together.

When not to use an agent

An agent is the right tool when the steps depend on what is found along the way. When you already know the steps, plain code is cheaper, faster and easier to test. For that reason the draughts team in Module 6 is run first as a pipeline, in which our code decides the order and the model only does the parts that need language, and only then as a team led by a manager agent that decides the order itself. Keep asking which parts of a job really need a model's judgement.

Check your progress

You should be able to name the parts of an agent framework, explain what each does, and say why a pipeline is often a better choice than letting a model decide the order.

Module 2: Talking to Any Model

One method for every model

The coach in Course 2 called Ollama directly. An agent framework should not care which model it talks to, so agentkit puts every model behind the same single method: chat(messages, tools), which returns one message. Any class with that method can be plugged in. This idea is called an adapter.

Create a folder called agentkit with the file below. OllamaModel talks to Ollama exactly as the coach did, through its /api/chat interface, with the model's private reasoning removed. ScriptedModel is a pretend model that gives a fixed list of replies, which the tests use.

File: agentkit/models.py

"""
models.py - talking to a language model.

Every model adapter has one method:

    chat(messages, tools) -> message

`messages` is the conversation so far and `tools` the list of tool
descriptions. The reply is one message: either text, or a request to call
tools. Because the agent only ever calls chat(), any model can be plugged in
by writing a new class with that one method. That is what makes the
framework "pluggable".
"""

import json
import os
import re
import urllib.request


def clean(text):
    """Some models wrap their private reasoning in <think> tags. Keep only the answer."""
    return re.sub(r"<think>.*?</think>", "", text or "", flags=re.S).strip()


def tidy_calls(message):
    """Some models send tool arguments as text instead of JSON. Always give the agent a dictionary."""
    for call in message.get("tool_calls") or []:
        args = call["function"].get("arguments") or {}
        if isinstance(args, str):
            args = json.loads(args or "{}")
        call["function"]["arguments"] = args
    return message


class OllamaModel:
    """A model running in Ollama, on this computer or this server (as in Course 2)."""

    def __init__(self, model=None, url=None, timeout=None):
        self.model = model or os.environ.get("AGENT_MODEL", "qwen3")
        self.url = url or os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
        self.timeout = float(timeout or os.environ.get("AGENT_TIMEOUT", "120"))

    def chat(self, messages, tools):
        body = json.dumps({"model": self.model, "messages": messages, "tools": tools,
                           "stream": False}).encode("utf-8")
        request = urllib.request.Request(self.url + "/api/chat", data=body,
                                         headers={"Content-Type": "application/json"})
        with urllib.request.urlopen(request, timeout=self.timeout) as response:
            message = json.loads(response.read())["message"]
        message["content"] = clean(message.get("content"))
        return tidy_calls(message)


class ScriptedModel:
    """A pretend model that gives a fixed list of replies, one per call. For tests."""

    def __init__(self, replies):
        self.replies = list(replies)
        self.seen = []                     # what the agent sent, so tests can inspect it

    def chat(self, messages, tools):
        self.seen.append({"messages": [dict(m) for m in messages], "tools": tools})
        return tidy_calls(dict(self.replies.pop(0)))

The model name and the address of Ollama are set by the environment variables AGENT_MODEL and OLLAMA_URL, with the same defaults as Course 2. The model you choose must support tool calling; check its page in Ollama's library, as in Course 2.

The stand-in

The stand-in from Course 2 knew about the coach's three tools only. An agent framework needs a stand-in that works with any tools, so this course has a new one. It is not a language model: it calls each tool in the agent's list once, fills in arguments from the task and from earlier results, and then answers by listing what the tools returned. A setting makes it invent a number now and then, so that we can prove the checks catch it.

Create a folder called devtools and copy devtools/standin.py into it from the project download. You do not need to study its code; the comment at the top, below, explains its rules.

From devtools/standin.py

"""
devtools/standin.py - a STAND-IN for a language model, for testing only.

This is not a language model. It follows fixed rules, so that every other part
of the framework - the agent loop, the tools, the checks, the team, the trace
and the evaluation - can be built and tested without one. The rules are:

  1. Call each tool in the agent's tool list once, in order, skipping any
     whose required arguments it cannot fill.
  2. Fill arguments from the task's Inputs, then from simple values earlier
     tools returned (such as a tournament_id). A text argument called task,
     query, question or text gets the task itself, plus those values.
  3. When no tools are left, answer by listing what the tools returned or,
     if the last tool was another agent, by passing on that agent's answer.
  4. MISTAKE_RATE (0 to 1) makes the first answer sometimes contain a number
     no tool reported, imitating a model that invents things. The checks
     should catch every one, and the retry should then fix it.

A real model writes far better prose, and does not always call every tool.
That is why the evaluation (Module 8) must be run again with your own model.
"""

devtools/fake_ollama.py puts the same stand-in behind a small Flask web server that answers like Ollama, on port 11435. It is included in the project download. You need it only to test OllamaModel itself, and the library in Module 9, without a real model.

Check your progress

You can explain what an adapter is, and why the agent code never needs to know which model it is talking to.

Module 3: Tools from Functions

Letting the function describe itself

In Course 2 each tool's description was written by hand as JSON. That works for three tools; it becomes a burden for twenty, and the description can drift away from the code. agentkit instead reads the description from the function itself, using a decorator called @tool.

• The function's name becomes the tool's name

• The first paragraph of its docstring becomes the description the model reads

• Each argument's type hint, such as int or list[str], becomes its JSON Schema type

• Arguments with a default value are optional; the rest are required

• An Args: section in the docstring describes each argument

Never trust the arguments

A model can ask for anything: a tool that does not exist, a word where a number should be, or a tournament of a thousand games. The ToolBox checks every call against the tool's description before the function runs. Limits that the type hint cannot express, such as a smallest and largest number or a list of allowed values, go in limits.

Just as important, ToolBox.run never raises an error. Anything that goes wrong comes back to the model as {"error": ...}, so the model can read what happened and try again, and the program keeps going. Finally, a tool marked needs_approval only runs if an approver says yes. Module 6 uses this to keep a person in charge of how much computing time the agents may spend.

File: agentkit/tools.py

"""
tools.py - turning ordinary Python functions into tools a model can call.

In Course 2 we wrote each tool's description by hand, as JSON. Here the
@tool decorator writes it for us from the function itself:

  * the function's name becomes the tool's name;
  * the first paragraph of its docstring becomes the description;
  * each argument's type hint becomes its JSON Schema type;
  * arguments with a default value are optional, the rest are required;
  * an "Args:" section in the docstring describes each argument.

Limits on the arguments (a smallest and largest number, a list of allowed
values) go in `limits`, and the ToolBox refuses any call that breaks them
before the function runs. A model is never trusted to stay within limits.
"""

import inspect
import json
import typing

JSON_TYPES = {str: "string", int: "integer", float: "number", bool: "boolean",
              list: "array", dict: "object"}


def schema_for(annotation):
    """The JSON Schema for one Python type hint, such as int or list[str]."""
    origin = typing.get_origin(annotation)
    if origin is list:
        (item,) = typing.get_args(annotation) or (str,)
        return {"type": "array", "items": schema_for(item)}
    return {"type": JSON_TYPES.get(annotation, "string")}


def read_docstring(fn):
    """Split a docstring into the description and the per-argument notes under 'Args:'."""
    doc = inspect.getdoc(fn) or ""
    description, _, args_part = doc.partition("Args:")
    notes = {}
    for line in args_part.splitlines():
        name, sep, text = line.strip().partition(":")
        if sep and name.isidentifier():
            notes[name] = text.strip()
    return " ".join(description.split()), notes


class Tool:
    def __init__(self, fn, needs_approval=False, limits=None):
        self.fn = fn
        self.name = fn.__name__
        self.needs_approval = needs_approval
        self.description, notes = read_docstring(fn)
        hints = typing.get_type_hints(fn)
        properties, required = {}, []
        for name, param in inspect.signature(fn).parameters.items():
            prop = schema_for(hints.get(name, str))
            if name in notes:
                prop["description"] = notes[name]
            prop.update((limits or {}).get(name, {}))
            if param.default is inspect.Parameter.empty:
                required.append(name)
            else:
                prop["default"] = param.default
            properties[name] = prop
        self.parameters = {"type": "object", "properties": properties, "required": required}

    def schema(self):
        """The description sent to the model, in the format Ollama expects."""
        return {"type": "function", "function": {
            "name": self.name, "description": self.description, "parameters": self.parameters}}

    def __call__(self, *args, **kwargs):
        return self.fn(*args, **kwargs)


def tool(fn=None, *, needs_approval=False, limits=None):
    """Use as @tool, or as @tool(needs_approval=True, limits={...})."""
    if fn is None:
        return lambda f: Tool(f, needs_approval, limits)
    return Tool(fn, needs_approval, limits)


def check_value(name, value, rule):
    """Problems with one argument, compared with its schema. An empty list means it is fine."""
    expected = rule.get("type")
    python_types = {"string": str, "integer": int, "number": (int, float),
                    "boolean": bool, "array": list, "object": dict}
    ok_type = isinstance(value, python_types.get(expected, object))
    if expected in ("integer", "number") and isinstance(value, bool):
        ok_type = False                    # in Python, True is also the number 1
    if not ok_type:
        return [f"{name} should be of type {expected}, not {type(value).__name__}"]
    problems = []
    if "enum" in rule and value not in rule["enum"]:
        problems.append(f"{name} must be one of {rule['enum']}")
    if "minimum" in rule and value < rule["minimum"]:
        problems.append(f"{name} must be at least {rule['minimum']}")
    if "maximum" in rule and value > rule["maximum"]:
        problems.append(f"{name} must be at most {rule['maximum']}")
    if expected == "array":
        if "minItems" in rule and len(value) < rule["minItems"]:
            problems.append(f"{name} needs at least {rule['minItems']} items")
        if "maxItems" in rule and len(value) > rule["maxItems"]:
            problems.append(f"{name} may have at most {rule['maxItems']} items")
        for item in value:
            problems += check_value(f"each item of {name}", item, rule.get("items", {}))
    return problems


def validate(args, parameters):
    problems = [f"missing argument: {n}" for n in parameters["required"] if n not in args]
    for name, value in args.items():
        if name not in parameters["properties"]:
            problems.append(f"unknown argument: {name}")
        else:
            problems += check_value(name, value, parameters["properties"][name])
    return problems


class ToolBox:
    """The tools one agent may use, and the only way that agent can run them."""

    def __init__(self, tools=()):
        self.tools = {t.name: t for t in tools}

    def schemas(self):
        return [t.schema() for t in self.tools.values()]

    def run(self, name, args, approver=None):
        """
        Run one tool call from the model. Never raises: anything that goes
        wrong comes back as {"error": ...}, so the model can read it and try
        again, and the program keeps going.
        """
        if name not in self.tools:
            return {"error": f"There is no tool called {name}. Tools: {', '.join(self.tools)}"}
        chosen = self.tools[name]
        problems = validate(args, chosen.parameters)
        if problems:
            return {"error": "The arguments were not accepted: " + "; ".join(problems)}
        if chosen.needs_approval and not (approver and approver(name, args)):
            return {"error": f"A person did not approve {name}, so it was not run."}
        try:
            result = chosen(**args)
            json.dumps(result)             # the result must be something we can send as JSON
        except Exception as error:         # a broken tool must not stop the agent
            return {"error": f"{name} failed: {type(error).__name__}: {error}"}
        return result

Try it

In a terminal in your project folder, start Python and describe a small function of your own.

python
>>> from agentkit import tool
>>> @tool
... def add(a: int, b: int = 1) -> dict:
...     """Add two numbers."""
...     return {"total": a + b}
...
>>> add.schema()
>>> from agentkit import ToolBox
>>> ToolBox([add]).run("add", {"a": "two"})

The last line returns an error message rather than crashing: a should be of type integer, not str.

Check your progress

You can write a tool with a docstring and type hints, read the schema the decorator makes from it, and explain why the ToolBox checks arguments before running the function.

Module 4: The Agent Loop and Its Checks

The loop

agent.py is the heart of the framework, and it follows Figure 1 exactly. It is the coach's loop from Course 2, written once so that every agent can reuse it, with three additions.

• A step limit. max_steps stops a confused model calling tools for ever. Every reply from the model counts as one step

• A check and a retry. When the model answers, the agent runs its check. If there are problems, the model is told what they are and may try once more. A second failure is reported, never hidden

• A record. Each Result keeps the answer, every tool call with its arguments and result, the problems found and how long it took. facts() collects the simple values the tools returned, such as a tournament_id, so that they can be passed to the next agent

File: agentkit/agent.py

"""
agent.py - the agent loop.

An agent is a language model, some instructions, a set of tools and a loop:

  1. send the task, the conversation so far and the tool list to the model;
  2. if the model asks for tools, run them and add the results to the
     conversation;
  3. repeat until the model writes an answer, or the step limit is reached;
  4. check the answer. If the check finds problems, tell the model what they
     are and let it try again, a limited number of times.

This is the same loop as the coach in Course 2. What is new is that it is
written once, here, and every agent in this course reuses it.
"""

import json
import time
from dataclasses import dataclass, field

from .tools import ToolBox


@dataclass
class Result:
    agent: str
    answer: str
    tool_log: list = field(default_factory=list)   # every tool call: name, args, result
    problems: list = field(default_factory=list)   # what the check found; empty means passed
    stopped: str = "answered"                       # or "step limit reached"
    steps: int = 0
    seconds: float = 0.0
    retries_used: int = 0

    @property
    def passed(self):
        return self.stopped == "answered" and not self.problems

    def facts(self):
        """Simple values the tools returned (such as a tournament_id), for the next agent."""
        found = {}
        for entry in self.tool_log:
            if isinstance(entry["result"], dict):
                for key, value in entry["result"].items():
                    if isinstance(value, (str, int, float, bool)) and key not in ("error", "answer", "passed"):
                        found[key] = value
        return found


class Agent:
    def __init__(self, name, instructions, tools=(), model=None, max_steps=8,
                 check=None, retries=1, approver=None, tracer=None):
        self.name = name
        self.instructions = instructions
        self.toolbox = ToolBox(tools)
        self.model = model
        self.max_steps = max_steps    # stops a confused model calling tools for ever
        self.check = check            # a function (answer, tool_log) -> list of problems
        self.retries = retries        # how many times the model may fix a failed check
        self.approver = approver      # asked before any tool marked needs_approval
        self.tracer = tracer          # records every step, if given

    def record(self, **event):
        if self.tracer:
            self.tracer.write(agent=self.name, **event)

    def run(self, task, inputs=None):
        started = time.time()
        text = task if not inputs else f"{task}\n\nInputs: {json.dumps(inputs)}"
        messages = [{"role": "system", "content": self.instructions},
                    {"role": "user", "content": text}]
        self.record(kind="task", text=text)
        tool_log, retries_left, answer, problems = [], self.retries, "", []
        for step in range(1, self.max_steps + 1):
            message = self.model.chat(messages, self.toolbox.schemas())
            calls = message.get("tool_calls") or []
            if not calls:
                answer = message.get("content", "")
                problems = self.check(answer, tool_log) if self.check else []
                self.record(kind="answer", step=step, text=answer, problems=problems)
                if problems and retries_left > 0:
                    retries_left -= 1
                    messages.append({"role": "assistant", "content": answer})
                    messages.append({"role": "user", "content":
                                     "Your answer was checked against the tool results and has "
                                     "these problems: " + " ".join(problems) +
                                     " Write the answer again, using only facts from the tools."})
                    continue
                return Result(self.name, answer, tool_log, problems, "answered", step,
                              round(time.time() - started, 2), self.retries - retries_left)
            messages.append(message)
            for call in calls:
                name = call["function"]["name"]
                args = call["function"].get("arguments") or {}
                result = self.toolbox.run(name, args, self.approver)
                tool_log.append({"tool": name, "args": args, "result": result})
                self.record(kind="tool", step=step, tool=name, args=args, result=result)
                messages.append({"role": "tool", "tool_name": name,
                                 "content": json.dumps(result)})
        self.record(kind="stopped", step=self.max_steps)
        return Result(self.name, answer, tool_log, problems or ["The step limit was reached."],
                      "step limit reached", self.max_steps, round(time.time() - started, 2),
                      self.retries - retries_left)

The checks

The coach's checker made sure that every move it mentioned had been reported by the engine. checks.py keeps that check and adds three more. Each check is a function that takes the answer and the tool log and returns a list of problems, and all_of joins several into one.

File: agentkit/checks.py

"""
checks.py - checking an agent's answer against what its tools really returned.

Grounding (Course 2) makes mistakes less likely; it does not make them
impossible. So, as with the coach, every answer is checked by ordinary code
before anyone relies on it. Each check is a function

    check(answer, tool_log) -> list of problems

and an empty list means the answer passed. all_of() joins several checks.
"""

import json
import re

NUMBER = re.compile(r"(?<![\w.])\d+(?:\.\d+)?")
MOVE = re.compile(r"\b[a-h][1-8](?:[-x][a-h][1-8])+\b")     # c3-d4, c3xe5xg7 (Course 2)


def tool_text(tool_log):
    """Everything the tools were asked and everything they said, as one piece of text."""
    return json.dumps([[t["args"], t["result"]] for t in tool_log])


def used_a_tool(answer, tool_log):
    if not tool_log:
        return ["The answer was written without calling any tool."]
    return []


def numbers_grounded(answer, tool_log):
    """Every number in the answer must appear somewhere in the tools' arguments or results."""
    confirmed = {float(n) for n in NUMBER.findall(tool_text(tool_log))}
    unconfirmed = [n for n in NUMBER.findall(answer) if float(n) not in confirmed]
    if unconfirmed:
        return ["These numbers did not come from any tool: " + ", ".join(dict.fromkeys(unconfirmed)) + "."]
    return []


def moves_grounded(answer, tool_log):
    """Every move mentioned must have been reported by a tool, as the coach's check did."""
    confirmed = set(MOVE.findall(tool_text(tool_log)))
    unconfirmed = [m for m in MOVE.findall(answer) if m not in confirmed]
    if unconfirmed:
        return ["These moves did not come from any tool: " + ", ".join(dict.fromkeys(unconfirmed)) + "."]
    return []


def succeeded(*names):
    """A check that each named tool was called at least once without an error."""
    def check(answer, tool_log):
        ok = {t["tool"] for t in tool_log if not (isinstance(t["result"], dict) and "error" in t["result"])}
        missing = [n for n in names if n not in ok]
        if missing:
            return ["These tools never ran successfully: " + ", ".join(missing) + "."]
        return []
    return check


def not_empty(answer, tool_log):
    return [] if answer.strip() else ["There was no answer."]


def all_of(*checks):
    def check(answer, tool_log):
        problems = []
        for c in checks:
            problems += c(answer, tool_log)
        return problems
    return check

Notice that numbers_grounded is strict on purpose: a number that appears only in the question does not count. If an answer says "the top 3 mistakes", the 3 must be an argument the agent passed to a tool. Being strict produces some false alarms, but a false alarm costs a retry, while a missed invention costs a wrong report.

Check your progress

You can follow one run of the loop through Figure 1, and explain what happens when an answer fails its check once, and twice.

Module 5: The Draughts Tools

The engine for agents

The agents never play or judge draughts themselves. As with the coach, every fact comes from the engine, run with Node.js. First, add the function below to static/js/ai.js, just before the return line at the end, and add scoreMoves to the names that line returns. It scores every legal move, as Course 2 did, but uses Course 3's evaluation functions.

From static/js/ai.js

  /*
   * Score every legal move for the side to move, best first (added in Course 5,
   * as Course 2 did for its coach). Each move is searched with a full window,
   * so every score is exact for the depth searched. Scores are from the point
   * of view of the side to move.
   */
  function scoreMoves(state, depth, evaluator = 'hand') {
    const ctx = { player: state.turn, nodes: 0, evaluate: evaluatorFor(evaluator) };
    const scored = RulesLib.legalMoves(state).map(move => ({
      move,
      score: minimax(RulesLib.applyMove(state, move), depth - 1, -Infinity, Infinity, false, ctx)
    }));
    scored.sort((a, b) => b.score - a.score);
    return scored;
  }

  return { WEIGHTS, LEVELS, WIN_SCORE, evaluateWith, chooseMove, scoreMoves };

Then create engine/arena.js. It answers three commands. tournament plays every pair of players, each opening twice with the colours swapped, using the fair match rules and the seeded random numbers from Course 3, and keeps every move. review replays the games and scores every move against the alternatives, so the biggest mistakes can be found. players lists the computer players.

File: engine/arena.js

/*
 * engine/arena.js - the game engine, packaged for the agents (Course 5).
 *
 * The agents never play or judge draughts themselves. When they need a fact,
 * a Python tool runs this file with Node.js, sends one JSON request on
 * standard input and reads one JSON answer from standard output, exactly as
 * the coach's tools did in Course 2.
 *
 *   { "command": "tournament", "players": ["easy", "medium"], "pairs": 1, "seed": 1 }
 *   { "command": "review", "games": [ ... ], "depth": 4, "how_many": 3 }
 *
 * The tournament reuses the fair match rules from Course 3 (train/match.js):
 * every opening is played twice with the colours swapped, and a seeded random
 * number generator means the same request always gives the same games.
 */

const Rules = require('../static/js/rules.js');
const AI = require('../static/js/ai.js');
const Notation = require('../static/js/notation.js');
const { seeded, randomOpening } = require('../train/match.js');

const MAX_PLIES = 200;          // very long games count as draws, as in Course 3

// Play one game and keep every move, so it can be reviewed later.
function playRecorded(start, blackName, whiteName, rand) {
  let state = start;
  const moves = [];
  for (let ply = 0; ply < MAX_PLIES; ply++) {
    const outcome = Rules.result(state);
    if (outcome) return { winner: outcome.winner, reason: outcome.reason, moves };
    const name = state.turn === 'b' ? blackName : whiteName;
    const move = AI.chooseMove(state, AI.LEVELS[name], rand);
    moves.push(Notation.moveName(move));
    state = Rules.applyMove(state, move);
  }
  return { winner: 'draw', reason: `stopped after ${MAX_PLIES} moves`, moves };
}

// Every player meets every other player. Each pairing plays `pairs` openings,
// each opening twice with the colours swapped.
function tournament(request) {
  const players = request.players;
  for (const p of players) {
    if (!AI.LEVELS[p]) return { error: `Unknown player: ${p}. Known players: ${Object.keys(AI.LEVELS).join(', ')}` };
  }
  const rand = seeded(request.seed || 1);
  const games = [];
  for (let i = 0; i < players.length; i++) {
    for (let j = i + 1; j < players.length; j++) {
      for (let k = 0; k < (request.pairs || 1); k++) {
        const start = randomOpening(rand);
        const opening = start;                    // remembered so a review can replay it
        for (const [black, white] of [[players[i], players[j]], [players[j], players[i]]]) {
          const game = playRecorded(start, black, white, rand);
          games.push({ number: games.length + 1, black, white, start: opening, ...game });
        }
      }
    }
  }
  return { games };
}

// How good was each move? Replay every game and score each move against the
// alternatives. The biggest drops are the mistakes worth reporting.
function review(request) {
  const depth = request.depth || 4;
  const found = [];
  for (const game of request.games) {
    let state = { board: game.start.board.map(r => Array.from(r)), turn: game.start.turn,
                  quietPlies: game.start.quietPlies || 0 };
    game.moves.forEach((name, index) => {
      const legal = Rules.legalMoves(state);
      const move = Notation.findMove(legal, name);
      if (legal.length > 1) {
        const scored = AI.scoreMoves(state, depth);
        const mine = scored.find(s => Notation.moveName(s.move) === name);
        const entry = {
          game: game.number,
          move_number: index + 1,
          side: state.turn === 'b' ? 'Black' : 'White',
          player: state.turn === 'b' ? game.black : game.white,
          played: name,
          engine_best_move: Notation.moveName(scored[0].move)
        };
        const half = AI.WIN_SCORE / 2;
        if (scored[0].score >= half && mine.score < half) {
          entry.verdict = 'missed a forced win';
          entry.size = Infinity;
        } else if (mine.score <= -half && scored[0].score > -half) {
          entry.verdict = 'walked into a forced loss';
          entry.size = Infinity;
        } else {
          // 100 points is about one man, so report the loss in men.
          entry.lost_men = Math.round((scored[0].score - mine.score) / 10) / 10;
          entry.size = entry.lost_men;
        }
        if (entry.size >= 0.5) found.push(entry);
      }
      state = Rules.applyMove(state, move);
    });
  }
  // Biggest mistakes first; ties in the order they were played.
  found.sort((a, b) => b.size - a.size || a.game - b.game || a.move_number - b.move_number);
  const mistakes = found.slice(0, request.how_many || 3).map(({ size, ...m }) => m);
  return { depth, mistakes_found: found.length, mistakes };
}

function run(request) {
  switch (request.command) {
    case 'tournament': return tournament(request);
    case 'review': return review(request);
    case 'players': return { players: AI.LEVELS };
    default: return { error: 'Unknown command: ' + request.command };
  }
}

module.exports = { run };

if (require.main === module) {
  let input = '';
  process.stdin.on('data', chunk => { input += chunk; });
  process.stdin.on('end', () => {
    try {
      process.stdout.write(JSON.stringify(run(JSON.parse(input))));
    } catch (e) {
      process.stdout.write(JSON.stringify({ error: String(e.message || e) }));
    }
  });
}

The tools

agents/draughts_tools.py turns the engine into four tools. Look for these details as you read it.

• Limits on cost. run_tournament accepts two to five known players and at most three openings per pair, and it is marked needs_approval, because it is the one tool that can take minutes

• Ratings reused. standings gives each player an Elo rating from the tournament's games, using ratings.compute from Course 4 unchanged

• Safe ids. A tournament_id must look like 20261028-140501, so a model can never use one to reach another file, such as ../../ratings

• Saved work. Tournaments are saved in data/tournaments, and a review, which is slow, is saved with its tournament, so the analyst and the reporter see the same mistakes

Create a folder called agents with an empty file called __init__.py, then add the tools.

File: agents/draughts_tools.py

"""
agents/draughts_tools.py - the tools the draughts agents may use.

Each tool is an ordinary Python function. The facts come from the engine
(engine/arena.js, run with Node.js) and from the Elo ratings of Course 4
(ratings.py); the language model only decides which tool to call and writes
about the results.

Tournaments are saved in data/tournaments/<tournament_id>.json, so later
tools, later agents and the person reading the report all see the same games.
"""

import json
import os
import re
import subprocess
import sys
from datetime import datetime, timedelta

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, BASE_DIR)

import ratings  # noqa: E402  (Course 4)
from agentkit import tool  # noqa: E402

ARENA_JS = os.path.join(BASE_DIR, "engine", "arena.js")
TOURNAMENT_DIR = os.path.join(BASE_DIR, "data", "tournaments")
NODE = os.environ.get("NODE_BINARY", "node")
PLAYERS = ["easy", "medium", "hard", "learned", "neural"]
WHERE_FROM = {"easy": "Course 1", "medium": "Course 1", "hard": "Course 1",
              "learned": "Course 3", "neural": "Course 3"}
SAFE_ID = re.compile(r"^\d{8}-\d{6}$")      # an id is only ever digits, so never a path


def arena(request, timeout=900):
    done = subprocess.run([NODE, ARENA_JS], input=json.dumps(request),
                          capture_output=True, text=True, timeout=timeout)
    if done.returncode != 0:
        raise RuntimeError("The engine failed: " + done.stderr.strip()[:200])
    answer = json.loads(done.stdout)
    if "error" in answer:
        raise ValueError(answer["error"])
    return answer


def path_for(tournament_id):
    if not SAFE_ID.match(str(tournament_id)):
        raise ValueError("A tournament_id looks like 20261028-140501.")
    path = os.path.join(TOURNAMENT_DIR, tournament_id + ".json")
    if not os.path.exists(path):
        raise ValueError(f"There is no tournament {tournament_id}.")
    return path


def load(tournament_id):
    with open(path_for(tournament_id), encoding="utf-8") as f:
        return json.load(f)


def save(tournament):
    os.makedirs(TOURNAMENT_DIR, exist_ok=True)
    path = os.path.join(TOURNAMENT_DIR, tournament["tournament_id"] + ".json")
    with open(path, "w", encoding="utf-8") as f:
        json.dump(tournament, f)


@tool
def list_players() -> dict:
    """List the computer players that can enter a tournament, with how far each searches ahead and how it judges positions."""
    levels = arena({"command": "players"})["players"]
    return {"players": [{"name": n, "depth": levels[n]["depth"],
                         "random_chance": levels[n]["randomChance"],
                         "evaluation": levels[n]["evaluator"], "built_in": WHERE_FROM[n]}
                        for n in PLAYERS]}


@tool(needs_approval=True,
      limits={"players": {"items": {"type": "string", "enum": PLAYERS}, "minItems": 2, "maxItems": 5},
              "pairs": {"minimum": 1, "maximum": 3},
              "seed": {"minimum": 1, "maximum": 1000000}})
def run_tournament(players: list[str], pairs: int = 1, seed: int = 1) -> dict:
    """Play a round-robin tournament between computer players. Every pair of players plays each opening twice, with the colours swapped. Returns a tournament_id for the other tools.

    Args:
        players: the names of the players taking part
        pairs: how many openings each pair of players plays (each is played twice)
        seed: any whole number; the same seed always gives the same games
    """
    if len(set(players)) != len(players):
        raise ValueError("Each player may only be entered once.")
    games = arena({"command": "tournament", "players": players, "pairs": pairs, "seed": seed})["games"]
    when = datetime.now()
    while os.path.exists(os.path.join(TOURNAMENT_DIR, when.strftime("%Y%m%d-%H%M%S") + ".json")):
        when += timedelta(seconds=1)       # two tournaments in the same second get different ids
    tournament_id = when.strftime("%Y%m%d-%H%M%S")
    save({"tournament_id": tournament_id, "players": players, "pairs": pairs,
          "seed": seed, "games": games})
    return {"tournament_id": tournament_id, "games_played": len(games),
            "results": [{"game": g["number"], "black": g["black"], "white": g["white"],
                         "winner": {"b": g["black"], "w": g["white"]}.get(g["winner"], "draw"),
                         "moves": len(g["moves"])} for g in games]}


@tool
def standings(tournament_id: str) -> dict:
    """The tournament table: games, wins, draws, losses and points for each player (a win is 1 point, a draw half), and an Elo rating calculated from these games alone.

    Args:
        tournament_id: the id returned by run_tournament
    """
    t = load(tournament_id)
    table = {p: {"player": p, "games": 0, "wins": 0, "draws": 0, "losses": 0} for p in t["players"]}
    for g in t["games"]:
        for colour, name in (("b", g["black"]), ("w", g["white"])):
            row = table[name]
            row["games"] += 1
            if g["winner"] == "draw":
                row["draws"] += 1
            elif g["winner"] == colour:
                row["wins"] += 1
            else:
                row["losses"] += 1
    elo = ratings.compute(g for g in t["games"])
    rows = []
    for row in table.values():
        row["points"] = row["wins"] + row["draws"] / 2
        row["elo"] = round(elo[row["player"]]["rating"])
        rows.append(row)
    rows.sort(key=lambda r: (-r["points"], -r["elo"], r["player"]))
    return {"tournament_id": tournament_id, "games_played": len(t["games"]), "table": rows,
            "elo_note": f"Everyone starts at {ratings.START}; K is {ratings.K}, as in Course 4."}


@tool(limits={"how_many": {"minimum": 1, "maximum": 10}})
def find_mistakes(tournament_id: str, how_many: int = 3) -> dict:
    """Replay every game in a tournament and find the biggest mistakes: moves that the engine scores well below the best move available at the time.

    Args:
        tournament_id: the id returned by run_tournament
        how_many: how many of the biggest mistakes to return
    """
    t = load(tournament_id)
    key = f"review-{how_many}"
    if key not in t:                       # reviewing is slow, so keep the answer
        t[key] = arena({"command": "review", "games": t["games"], "depth": 4, "how_many": how_many})
        save(t)
    review = t[key]
    return {"tournament_id": tournament_id, "review_depth": review["depth"],
            "mistakes_found": review["mistakes_found"], "biggest_mistakes": review["mistakes"],
            "note": "move_number counts every move by either side after the random opening; "
                    "lost_men is how much worse than the best move, where 1 is about one man."}

Try it from the terminal

You can call the tools directly, with no model involved.

python
>>> from agents import draughts_tools as dt
>>> t = dt.run_tournament(players=["easy", "medium"], pairs=1, seed=1)
>>> dt.standings(t["tournament_id"])
>>> dt.find_mistakes(t["tournament_id"])

In the course workspace, a full tournament of all five players with one opening per pair played 20 games in 52.7 seconds, and reviewing them took 4.1 seconds. Your times will depend on your computer.

Check your progress

You can run a tournament from the terminal, read its standings, and explain why run_tournament needs approval while standings does not.

Module 6: A Team of Agents

Three agents, one job each

agents/draughts_team.py defines three agents. Each one has short, numbered instructions, only the tools it needs, and a check that asks two questions: are the words grounded, and was the job done?

• The director runs the tournament and reads the standings

• The analyst finds the biggest mistakes

• The reporter writes the report, and reads the notebook to compare with earlier tournaments

Giving each agent one job and a few tools keeps its instructions short, makes its mistakes easier to spot in a trace, and means a cheaper or smaller model may be enough for the simpler jobs.

Pipeline or manager

Figure 2 shows the two ways the team can work. In the pipeline, our code runs the director, then the analyst, then the reporter, passing the tournament_id along. In the manager version, a fourth agent is given the other three as tools, through agent_as_tool, and the model decides whom to ask and in what order.

File: agentkit/team.py

"""
team.py - several agents working together, with a person in charge.

There are two common ways to organise a team of agents:

  Pipeline        Our code decides the order: agent A, then B, then C. Each
                  agent's results are handed to the next. Predictable and easy
                  to test; use it when you already know the steps.

  Manager         One agent is given the other agents as tools, and the model
                  decides whom to ask, and in what order. Flexible, but harder
                  to predict; use it when the steps depend on what is found.

agent_as_tool() makes the manager pattern possible: it wraps a whole agent so
that it looks like any other tool.
"""

from .tools import Tool


class Pipeline:
    def __init__(self, steps, stop_on_failure=True):
        self.steps = steps                    # a list of (agent, task) pairs
        self.stop_on_failure = stop_on_failure

    def run(self, inputs):
        """Run each agent in turn. Facts and answers from earlier agents are passed on."""
        shared = dict(inputs)
        results = []
        for agent, task in self.steps:
            result = agent.run(task, shared)
            results.append(result)
            shared.update(result.facts())
            shared[f"{agent.name}_answer"] = result.answer
            if self.stop_on_failure and not result.passed:
                break
        return results


def agent_as_tool(agent, description):
    """Wrap an agent so that another agent (a manager) can call it like a tool."""
    def run_agent(task: str) -> dict:
        result = agent.run(task)
        return {"answer": result.answer, "passed": result.passed,
                "problems": result.problems, **result.facts()}

    run_agent.__name__ = f"ask_{agent.name}"
    run_agent.__doc__ = f"""{description}

    Args:
        task: what you want this agent to do, in a sentence or two, including any ids it needs
    """
    return Tool(run_agent)


def ask_person(tool_name, args):
    """Approve a tool call by asking the person at the keyboard."""
    reply = input(f"The agent wants to run {tool_name} with {args}. Allow it? [y/N] ")
    return reply.strip().lower() in ("y", "yes")


def refuse_all(tool_name, args):
    """For unattended runs: nothing that needs approval is ever run."""
    return False

Start with the pipeline. It is predictable and easy to test, and a failed check stops it before the next agent builds on a mistake. Use a manager only when the order of the steps really depends on what is found, and measure it in Module 8 before you trust it.

Memory that a person has approved

The reporter can search a notebook of notes from earlier tournaments, using the recall tool. The notebook is long-term memory: unlike the conversation, it lasts between runs. A wrong note, however, would mislead every later report, so the agents can only read it. A note is written by ordinary code, from the standings, and only after a person has approved the report.

File: agentkit/memory.py

"""
memory.py - a notebook that lasts between runs.

An agent's conversation is its short-term memory: it is lost when the run
ends. The Notebook is long-term memory: short notes saved in a JSON file and
found again by keyword search.

A wrong note would mislead every later run, so in this course the agents can
only READ the notebook. Notes are written by ordinary code, after a person has
approved the work they summarise.
"""

import json
import os
import re
from datetime import datetime, timezone

from .tools import tool

WORD = re.compile(r"[a-z0-9]+")


class Notebook:
    def __init__(self, path):
        self.path = path

    def notes(self):
        if not os.path.exists(self.path):
            return []
        with open(self.path, encoding="utf-8") as f:
            return json.load(f)

    def add(self, text, tags=()):
        notes = self.notes()
        notes.append({"when": datetime.now(timezone.utc).isoformat(timespec="seconds"),
                      "text": text, "tags": list(tags)})
        os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(notes, f, indent=1)

    def search(self, query, limit=3):
        """Notes sharing the most words with the query; the newest first when equal."""
        wanted = set(WORD.findall(query.lower()))
        scored = []
        for i, note in enumerate(self.notes()):
            words = set(WORD.findall((note["text"] + " " + " ".join(note["tags"])).lower()))
            overlap = len(wanted & words)
            if overlap:
                scored.append((overlap, i, note))
        scored.sort(key=lambda s: (-s[0], -s[1]))
        return [note for _, _, note in scored[:limit]]

    def recall_tool(self):
        """A read-only tool that lets an agent search this notebook."""
        notebook = self

        @tool
        def recall(query: str) -> dict:
            """Search the notebook of earlier, approved results for notes about a subject.

            Args:
                query: words to look for, such as the names of players
            """
            found = notebook.search(query)
            return {"notes_found": len(found), "notes": found}

        return recall

The team

File: agents/draughts_team.py

"""
agents/draughts_team.py - three agents that run, study and report on a tournament.

  director   runs the tournament and reads the standings
  analyst    finds the biggest mistakes in the games
  reporter   writes the report, using the notebook to compare with earlier runs

build_pipeline() puts them in a fixed order. build_manager() gives them to a
fourth agent, the manager, as tools, and lets the model decide the order.
"""

import os

from agentkit import (Agent, Notebook, Pipeline, agent_as_tool, all_of, moves_grounded,
                      numbers_grounded, succeeded, used_a_tool)

from .draughts_tools import BASE_DIR, find_mistakes, list_players, run_tournament, standings

NOTEBOOK = Notebook(os.path.join(BASE_DIR, "data", "notebook.json"))
GROUNDED = all_of(used_a_tool, numbers_grounded, moves_grounded)

DIRECTOR = """You run draughts tournaments between computer players.
1. If you are not sure of the players' names, call list_players.
2. Call run_tournament with the players, pairs and seed you are given.
3. Call standings with the tournament_id that run_tournament returns.
Then reply in no more than three sentences: the tournament_id, who finished first and with how many points.
Use only facts the tools return."""

ANALYST = """You study the games of one draughts tournament.
Call find_mistakes with the tournament_id you are given.
Then list the biggest mistakes, one sentence each: who made it, in which game and at which move number, the move played, the engine's better move, and how much it cost.
Moves are written like c3-d4 or c3xe5. Use only facts the tools return."""

REPORTER = """You write a short tournament report for people learning about AI.
1. Call standings and find_mistakes with the tournament_id you are given.
2. Call recall with the players' names, to find notes about earlier tournaments.
Write the report in Markdown, under 250 words, in plain British English, with: a table of the standings, the biggest mistakes, and, only if recall found earlier notes, one sentence comparing with them.
Every number and every move you write must come from a tool. Do not add advice or opinions."""

MANAGER = """You manage a team of three agents that produce a draughts tournament report.
1. Ask the director to run the tournament, passing on the players, pairs and seed.
2. Ask the analyst to find the mistakes, passing on the tournament_id.
3. Ask the reporter to write the report, passing on the tournament_id.
Reply with the reporter's report exactly as it was written."""


def build_agents(model, approver=None, tracer=None):
    common = {"model": model, "tracer": tracer}
    # Each check asks two things: are the words grounded, and was the job done?
    director = Agent("director", DIRECTOR, [list_players, run_tournament, standings],
                     check=all_of(GROUNDED, succeeded("run_tournament", "standings")),
                     approver=approver, **common)
    analyst = Agent("analyst", ANALYST, [find_mistakes],
                    check=all_of(GROUNDED, succeeded("find_mistakes")), **common)
    reporter = Agent("reporter", REPORTER, [standings, find_mistakes, NOTEBOOK.recall_tool()],
                     check=all_of(GROUNDED, succeeded("standings", "find_mistakes")), **common)
    return director, analyst, reporter


def build_pipeline(model, approver=None, tracer=None):
    director, analyst, reporter = build_agents(model, approver, tracer)
    return Pipeline([
        (director, "Run a tournament between these players, then report the standings."),
        (analyst, "Find the biggest mistakes in this tournament."),
        (reporter, "Write the tournament report."),
    ])


def build_manager(model, approver=None, tracer=None):
    director, analyst, reporter = build_agents(model, approver, tracer)
    team = [
        agent_as_tool(director, "The director runs a tournament and reports the standings and the tournament_id."),
        agent_as_tool(analyst, "The analyst finds the biggest mistakes in a tournament, given its tournament_id."),
        agent_as_tool(reporter, "The reporter writes the finished report for a tournament, given its tournament_id."),
    ]
    return Agent("manager", MANAGER, team, model=model, tracer=tracer, max_steps=6,
                 check=all_of(GROUNDED, succeeded("ask_reporter")))


def remember(report_inputs, table):
    """Called by run_team.py only after a person approves the report."""
    placing = "; ".join(f"{r['player']} {r['points']} points, Elo {r['elo']}" for r in table)
    NOTEBOOK.add(f"Tournament {report_inputs['tournament_id']} with players "
                 f"{', '.join(report_inputs['players'])}: {placing}.", tags=report_inputs["players"])

A person in charge

run_team.py runs the team from the command line. A person stays in charge at two points: before the tournament is played, because it costs computing time, and before the report is saved and remembered. With --unattended, used for the nightly run in Module 10, no one is asked. Instead a budget allows tournaments of up to --max-games games, and the report is saved only if every check passed.

File: run_team.py

"""
run_team.py - run the draughts agents from the command line.

    python run_team.py --players easy medium hard
    python run_team.py --players easy medium --manager
    python run_team.py --players easy medium --standin       (no model needed)
    python run_team.py --players easy medium hard --unattended --daily-seed

A person stays in charge at two points:
  * before the tournament is played, because it costs computing time
    (unattended runs instead allow up to --max-games games and no more);
  * before the report is saved and remembered in the notebook
    (unattended runs save it only if every check passed).
"""

import argparse
import os
import sys
import urllib.error
from datetime import datetime

from agentkit import OllamaModel, Tracer, ask_person
from agents.draughts_team import build_manager, build_pipeline, remember
from agents.draughts_tools import BASE_DIR, standings

REPORT_DIR = os.path.join(BASE_DIR, "reports")


def games_in(args):
    n = len(args["players"])
    return n * (n - 1) * args.get("pairs", 1)       # every pair, each opening twice


def budget(max_games):
    """An approver for unattended runs: allow a tournament only if it is small enough."""
    def approve(tool_name, args):
        return tool_name == "run_tournament" and games_in(args) <= max_games
    return approve


def main(argv=None):
    parser = argparse.ArgumentParser(description="Run the draughts agent team.")
    parser.add_argument("--players", nargs="+", default=["easy", "medium", "hard"])
    parser.add_argument("--pairs", type=int, default=1)
    parser.add_argument("--seed", type=int, default=1)
    parser.add_argument("--daily-seed", action="store_true",
                        help="use today's date as the seed, so each night's games differ")
    parser.add_argument("--manager", action="store_true", help="let a manager agent decide the order")
    parser.add_argument("--standin", action="store_true", help="use the stand-in instead of a model")
    parser.add_argument("--unattended", action="store_true", help="no questions; for scheduled runs")
    parser.add_argument("--max-games", type=int, default=20)
    a = parser.parse_args(argv)
    if a.daily_seed:
        a.seed = int(datetime.now().strftime("%y%m%d"))     # 28 October 2026 gives 261028

    if a.standin:
        sys.path.insert(0, os.path.join(BASE_DIR, "devtools"))
        from standin import StandInModel
        model = StandInModel()
    else:
        model = OllamaModel()
    approver = budget(a.max_games) if a.unattended else ask_person
    run_name = datetime.now().strftime("%Y%m%d-%H%M%S")
    tracer = Tracer(os.path.join(BASE_DIR, "data", "traces", run_name + ".jsonl"))
    inputs = {"players": a.players, "pairs": a.pairs, "seed": a.seed}

    try:
        if a.manager:
            result = build_manager(model, approver, tracer).run("Produce the tournament report.", inputs)
            results = [result]
            facts = {}
            for entry in result.tool_log:
                if isinstance(entry["result"], dict):
                    facts.update(entry["result"])
        else:
            results = build_pipeline(model, approver, tracer).run(inputs)
            facts = {}
            for r in results:
                facts.update(r.facts())
    except (urllib.error.URLError, ConnectionError, TimeoutError) as error:
        print(f"Could not reach the model: {error}. Is Ollama running?")
        return 2

    for r in results:
        status = "passed" if r.passed else "PROBLEMS: " + " ".join(r.problems)
        print(f"{r.agent}: {r.steps} steps, {len(r.tool_log)} tool calls, {r.seconds}s, {status}")
    print(f"Trace: {tracer.path}")

    final = results[-1]
    if "tournament_id" not in facts or not final.answer:
        print("No report was produced.")
        return 1
    print("\n" + final.answer + "\n")

    if a.unattended:
        keep = all(r.passed for r in results)
    else:
        keep = input("Save this report and remember its results? [y/N] ").strip().lower() in ("y", "yes")
    if not keep:
        print("Not saved.")
        return 1
    os.makedirs(REPORT_DIR, exist_ok=True)
    path = os.path.join(REPORT_DIR, f"report-{facts['tournament_id']}.md")
    with open(path, "w", encoding="utf-8") as f:
        f.write(final.answer + "\n")
    table = standings(facts["tournament_id"])["table"]
    remember({"tournament_id": facts["tournament_id"], "players": a.players}, table)
    print(f"Saved {path}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

Traces

Every run writes a trace to data/traces: one line of JSON for every task, tool call and answer, in order. When an agent goes wrong, the trace tells you why far more often than the answer does. The short script show_trace.py prints a trace in a form people can read.

File: agentkit/trace.py

"""
trace.py - a record of every step an agent takes.

When an agent goes wrong, the answer alone rarely tells you why. A trace
does: it holds every task, tool call, tool result and answer, in order, one
JSON object per line (a format called JSON Lines).

To read a trace:   python show_trace.py data/traces/<file>.jsonl
"""

import json
import os
import time


class Tracer:
    def __init__(self, path):
        self.path = path
        os.makedirs(os.path.dirname(path) or ".", exist_ok=True)

    def write(self, **event):
        event["time"] = round(time.time(), 2)
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps(event) + "\n")


def short(value, limit=160):
    text = json.dumps(value) if not isinstance(value, str) else value
    return text if len(text) <= limit else text[:limit] + " ..."


def show(path):
    with open(path, encoding="utf-8") as f:
        for line in f:
            e = json.loads(line)
            if e["kind"] == "task":
                print(f"\n[{e['agent']}] TASK {short(e['text'])}")
            elif e["kind"] == "tool":
                print(f"[{e['agent']}] step {e['step']}: {e['tool']}({short(e['args'], 80)})"
                      f" -> {short(e['result'])}")
            elif e["kind"] == "answer":
                flag = "passed" if not e["problems"] else "PROBLEMS: " + " ".join(e["problems"])
                print(f"[{e['agent']}] step {e['step']}: ANSWER ({flag})\n    {short(e['text'], 400)}")
            else:
                print(f"[{e['agent']}] {e['kind']}")

Run it

Try the team with the stand-in first, then with your model. The stand-in's report is a plain list of what the tools returned; your model should write a proper report in the same facts.

python run_team.py --players easy medium --standin
python run_team.py --players easy medium hard
python run_team.py --players easy medium hard --manager
python show_trace.py data/traces/<the file named at the end of the run>

The program asks before playing the tournament and again before saving. Run it a second time and the reporter's recall finds the first tournament's note.

Check your progress

You have run the pipeline and the manager, approved one report, and read its trace. You can explain why the agents may read the notebook but not write to it.

Module 7: Beyond Draughts

An agent for any project

Nothing in agentkit knows about draughts. To prove it, this module builds an agent for a common job at work: answering questions about a folder of files, such as a software project, with the file and line for every claim. Point it at your Course 4 folder, or at any project of your own.

The helper has three read-only tools: search_files, read_file and list_files. Its check, files_grounded, makes sure that every file named in the answer was returned by a tool, so a reader can open each source and check the claim.

Fencing the agent in

An agent that can read files must not be able to read every file. The tools are fenced into the project folder.

• Every path is turned into its real location, and refused if it is outside the folder. That stops tricks such as ../../etc/passwd

• Folders such as .git, venv and data are never listed or read

• Files that usually hold secrets, such as .env or anything ending in .key or .pem, are never read

• Large files are skipped, and read_file returns at most 120 lines at a time

• Paths are always shown with forward slashes, on Windows too, so that the check compares like with like

File: agents/codebase_helper.py

"""
agents/codebase_helper.py - the same framework, used for a job that has
nothing to do with draughts: answering questions about a folder of files.

Point it at any project folder. It can search, list and read files there,
and nowhere else. Every file it names in its answer must be one its tools
actually returned, so a reader can check each claim at its source.

    python ask_codebase.py "Where is the step limit set?"
    python ask_codebase.py --folder ../draughts-adapt "How are Elo ratings updated?"
"""

import os
import re

from agentkit import Agent, all_of, tool, used_a_tool

SKIP_FOLDERS = {".git", "venv", "__pycache__", "node_modules", "data"}
SKIP_FILES = re.compile(r"(^\.env|\.key$|\.pem$|secret)", re.I)    # never read secrets
TEXT_FILES = re.compile(r"\.(py|js|md|txt|json|html|css|conf|wsgi|cfg|toml|ya?ml)$")
FILE_NAME = re.compile(r"[\w./-]+\.(?:py|js|md|txt|json|html|css|conf|wsgi|cfg|toml|ya?ml)\b")
MAX_FILE_BYTES = 200_000


def make_tools(root):
    root = os.path.realpath(root)

    def inside(path):
        """The real location of `path`, refused if it is outside the folder or not allowed."""
        full = os.path.realpath(os.path.join(root, path))
        if os.path.commonpath([root, full]) != root:
            raise PermissionError("Only files inside the project folder may be read.")
        parts = os.path.relpath(full, root).split(os.sep)
        if SKIP_FOLDERS & set(parts) or SKIP_FILES.search(parts[-1]):
            raise PermissionError("That file or folder is not available to this agent.")
        return full

    def shown(path):
        """Paths are always shown with forward slashes, on Windows too, so answers match them."""
        return os.path.relpath(path, root).replace(os.sep, "/")

    def all_files():
        for folder, subfolders, files in os.walk(root):
            subfolders[:] = sorted(d for d in subfolders if d not in SKIP_FOLDERS)
            for name in sorted(files):
                if TEXT_FILES.search(name) and not SKIP_FILES.search(name):
                    yield shown(os.path.join(folder, name))

    @tool
    def search_files(text: str) -> dict:
        """Search every text file in the project for lines containing the words in the text. Returns the best-matching lines, with file and line number.

        Args:
            text: the words to look for
        """
        words = {w for w in re.findall(r"[a-z_]{4,}", text.lower())}
        hits = []
        for path in all_files():
            full = inside(path)
            if os.path.getsize(full) > MAX_FILE_BYTES:
                continue
            with open(full, encoding="utf-8", errors="replace") as f:
                for number, line in enumerate(f, 1):
                    score = sum(w in line.lower() for w in words)
                    if score:
                        hits.append((score, path, number, line.strip()[:160]))
        hits.sort(key=lambda h: (-h[0], h[1], h[2]))
        return {"lines_found": len(hits),
                "best_matches": [{"file": p, "line": n, "text": t} for _, p, n, t in hits[:10]]}

    @tool
    def list_files(folder: str = ".") -> dict:
        """List the text files in the project, or in one folder of it.

        Args:
            folder: a folder inside the project, such as agentkit
        """
        prefix = shown(inside(folder))
        files = [p for p in all_files() if prefix == "." or p.startswith(prefix + "/")]
        return {"files_listed": len(files), "files": files[:200]}

    @tool(limits={"start": {"minimum": 1}, "count": {"minimum": 1, "maximum": 120}})
    def read_file(path: str, start: int = 1, count: int = 60) -> dict:
        """Read lines from one file in the project, with their line numbers.

        Args:
            path: the file, as listed by list_files or search_files
            start: the first line to read
            count: how many lines to read
        """
        full = inside(path)
        if not os.path.isfile(full):
            raise FileNotFoundError(f"There is no file {path}.")
        with open(full, encoding="utf-8", errors="replace") as f:
            lines = f.readlines()
        chosen = lines[start - 1:start - 1 + count]
        return {"file": shown(full), "first_line": start, "total_lines": len(lines),
                "lines": [f"{start + i}: {line.rstrip()}" for i, line in enumerate(chosen)]}

    return [search_files, read_file, list_files]


def files_grounded(answer, tool_log):
    """Every file named in the answer must have come back from a tool."""
    seen = " ".join(str(t["result"]) for t in tool_log)
    answer = answer.replace("\\", "/")              # agentkit\agent.py and agentkit/agent.py are the same file
    unknown = [name for name in FILE_NAME.findall(answer) if name not in seen]
    if unknown:
        return ["These files were not returned by any tool: " + ", ".join(dict.fromkeys(unknown)) + "."]
    return []


INSTRUCTIONS = """You answer questions about the files in one software project.
1. Call search_files with the important words from the question.
2. Read the most promising files with read_file. Use list_files if you need to see what exists.
Answer in no more than five sentences of plain British English. After the answer, list your sources as file:line, one per line.
Only name files that the tools returned. If the files do not answer the question, say so."""


def build_helper(model, root, tracer=None):
    return Agent("codebase_helper", INSTRUCTIONS, make_tools(root), model=model, tracer=tracer,
                 max_steps=8, check=all_of(used_a_tool, files_grounded))

ask_codebase.py runs the helper from the command line.

python ask_codebase.py --standin "Where is the step limit set?"
python ask_codebase.py "Where is the step limit set?"
python ask_codebase.py --folder ../draughts-adapt "How are Elo ratings updated?"

Building your own agent

The same recipe works for any agent you build next.

1. Write down the goal, and which steps need judgement and which do not. Give the steps that do not to plain code

2. Write each tool as a small function with a clear docstring, type hints and limits. Make tools read-only wherever you can

3. Mark any tool that costs money, takes a long time or changes something lasting as needs_approval, and decide who approves it

4. Write a check for the answer: which facts must come from a tool, and which tools must have succeeded

5. Write short, numbered instructions, and test the agent with a scripted model before using a real one

6. Measure it over many runs, as in Module 8, before anyone relies on it

Check your progress

You have asked the helper a question about one of your own projects, and checked one of its sources by opening the file at the line it gave.

Module 8: Evaluation and Testing

Why one good run proves nothing

A model can be lucky once. eval/run_eval.py runs the pipeline once for each of a series of seeds and counts, for each agent, how often its answer passed every check first time, how often it needed its retry, and how often it failed altogether, with the average number of steps and the average time. It keeps tournaments small, easy against medium with one opening, because the model, not the engine, is being measured, and it keeps its games out of your data folder.

python eval/run_eval.py --standin
python eval/run_eval.py --standin --mistake-rate 0.3
python eval/run_eval.py --runs 10

Testing the test

Before trusting a measurement of your model, check that the checks themselves work. The stand-in can be told to invent a number in its first answer some of the time. Every invention should be caught, and the retry should then fix it. These are the results from the course workspace, with 20 runs each.

With the mistake rate at 0.3 the stand-in invented a number 21 times, and the checks caught all 21. The exact figures change from run to run, because each tournament's id comes from the clock, and the stand-in's choices depend on the conversation.

Measuring your real model

Run the evaluation with your model, then fill in a table like the one below. Run it again whenever you change a model, an agent's instructions or a tool's description, and keep the change only if the numbers improve.

The tests

The project has three new test files, and keeps the four from Course 3. None of them needs a language model.

Run them from the project folder.

node tests/test_arena.js
python tests/test_agentkit.py
python tests/test_agents.py

All 33 new tests passed in the course workspace, as did the 19 tests kept from Course 3.

Check your progress

You have run the evaluation with the stand-in at a mistake rate of 0.3 and seen every invention caught, then run it with your own model and recorded the results.

Module 9: The Same Ideas in a Library

Why build it yourself first

Libraries for agents save a lot of typing, but they hide the parts that matter. Now that you have built each part, you can read a library's documentation and know what it must be doing underneath. This module rebuilds the director with LangChain, one widely used library, and compares the two.

Mapping the parts

The director, again

The file below uses the very same tool functions as Module 5. Install the library into your project's virtual environment first, from library/requirements.txt, which pins the versions tested.

File: library/langchain_director.py

"""
library/langchain_director.py - the director agent again, built with LangChain.

Module 9 compares this with agents/draughts_team.py. The tools are the very
same Python functions; only the loop around them now comes from a library.

    pip install -r library/requirements.txt
    python library/langchain_director.py
"""

import functools
import os
import sys

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, BASE_DIR)

from langchain.agents import create_agent  # noqa: E402
from langchain_core.tools import StructuredTool  # noqa: E402
from langchain_ollama import ChatOllama  # noqa: E402

from agents import draughts_tools as dt  # noqa: E402
from agents.draughts_team import DIRECTOR, GROUNDED  # noqa: E402

# Our @tool objects keep the original function in .fn, with its type hints and
# docstring, which is all LangChain needs to describe it to the model. Two
# things our ToolBox did must be added back by hand:
#   * limits: LangChain does not see our `limits`, so run_tournament checks
#     its own budget here;
#   * errors: in the version tested, a tool that raised an error stopped the
#     whole run, so safe() turns errors into a message for the model.
MAX_GAMES = 20


def safe(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        try:
            return fn(*args, **kwargs)
        except Exception as error:
            return {"error": f"{type(error).__name__}: {error}"}
    return wrapper


def run_tournament(players: list[str], pairs: int = 1, seed: int = 1) -> dict:
    n = len(players)
    if n * (n - 1) * pairs > MAX_GAMES:
        return {"error": f"Refused: more than {MAX_GAMES} games."}
    return dt.run_tournament.fn(players, pairs, seed)


run_tournament.__doc__ = dt.run_tournament.fn.__doc__

tools = [StructuredTool.from_function(safe(f), parse_docstring=True)
         for f in (dt.list_players.fn, run_tournament, dt.standings.fn)]

model = ChatOllama(model=os.environ.get("AGENT_MODEL", "qwen3"),
                   base_url=os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434"))

director = create_agent(model, tools=tools, system_prompt=DIRECTOR)

def check(state):
    """Our own checks from agentkit, run on LangChain's messages: they still apply."""
    tool_log = [{"tool": m.name, "args": {}, "result": m.content}
                for m in state["messages"] if m.type == "tool"]
    return GROUNDED(state["messages"][-1].content, tool_log)


if __name__ == "__main__":
    task = ('Run a tournament between these players, then report the standings.\n\n'
            'Inputs: {"players": ["easy", "medium"], "pairs": 1, "seed": 1}')
    # recursion_limit caps the steps, as max_steps does in our Agent.
    state = director.invoke({"messages": [{"role": "user", "content": task}]},
                            config={"recursion_limit": 12})
    print(state["messages"][-1].content)
    print("\nCheck:", check(state) or "passed")
pip install -r library/requirements.txt
python library/langchain_director.py

Check your progress

You have run the LangChain director, and can name two things the library did for you and two things you still had to add yourself.

Module 10: Running the Team Every Night

Part A: Put the project on the server

The agents run as a scheduled job, not as part of the website, so they live in their own folder, /opt/draughts-agents. Copy the project there, or clone it and update it with git pull as the GitHub course describes. Then create a virtual environment. The agents need only the Python standard library, Node.js and Ollama, which Course 2 installed.

sudo mkdir -p /opt/draughts-agents
sudo chown $USER /opt/draughts-agents
cd /opt/draughts-agents
python3 -m venv venv
sudo chown -R www-data:www-data /opt/draughts-agents

Part B: A service and a timer

systemd can start a program on a schedule. It needs two files, both in the project's deploy folder. The service says what to run and as which user; the timer says when.

File: deploy/draughts-agents.service

# deploy/draughts-agents.service - runs the agent team once, unattended.
# Copy to /etc/systemd/system/ (Module 10). The timer below starts it each night.
[Unit]
Description=Draughts agent team: nightly tournament report
After=network-online.target ollama.service
Wants=network-online.target

[Service]
Type=oneshot
User=www-data
WorkingDirectory=/opt/draughts-agents
Environment=OLLAMA_URL=http://127.0.0.1:11434
Environment=AGENT_MODEL=qwen3
Environment=NODE_BINARY=/usr/bin/node
ExecStart=/opt/draughts-agents/venv/bin/python run_team.py --unattended --daily-seed --players easy medium hard learned neural --max-games 20

File: deploy/draughts-agents.timer

# deploy/draughts-agents.timer - starts draughts-agents.service at 02:30 every night.
# Persistent=true runs a missed night as soon as the server is back on.
[Unit]
Description=Run the draughts agent team every night

[Timer]
OnCalendar=*-*-* 02:30:00
Persistent=true

[Install]
WantedBy=timers.target

The run uses --daily-seed, so each night's games differ, and --unattended, so the budget decides whether the tournament may be played and the checks decide whether the report is saved. Copy both files into place and start the timer.

sudo cp deploy/draughts-agents.service deploy/draughts-agents.timer /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable --now draughts-agents.timer
systemctl list-timers draughts-agents.timer

Part C: Check that it works

Run the service once by hand, then read what it printed. Reports appear in /opt/draughts-agents/reports, and traces in data/traces.

sudo systemctl start draughts-agents.service
journalctl -u draughts-agents.service --since today

Troubleshooting

Check your progress

The timer is listed by systemctl list-timers, one run started by hand has finished, and its report is in the reports folder.

Module 11: Where to Go Next

• Write a new tool. Give the analyst a tool that replays one game move by move, and measure whether its reports improve

• Compare models. Run the evaluation with two or three tool-capable models, and choose using the pass rate, the steps and the time together

• Give each agent its own model. The analyst's job is simpler than the reporter's. Measure whether a smaller model is enough for it

• Show the reports on the website. Add a Flask page that lists the approved reports in reports/

• Build an agent for your own work. Follow the recipe in Module 7 for a task you do every week, keeping every tool read-only to begin with

Known limitations

• The checks are narrow. They catch invented numbers, moves and files, not real ones put in the wrong place, and not poor reasoning

• The stand-in is not a model. Every result in this course comes from the stand-in. Your model's results are the ones that count

• The notebook search is simple. It matches words, so it finds "medium" but not "the middle level"

• One tournament is a small sample. Twenty games show which player did best that night, not which player is stronger

Glossary of Terms

Appendix: Sources