277 AI Engineer Interview Questions & Answers

from __future__ import annotations

import asyncio
import random
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, Optional


@dataclass
class RetryConfig:
    # Bound the retry loop so one logical request cannot retry forever.
    max_attempts: int = 6
    base_delay: float = 0.5
    max_delay: float = 30.0
    timeout: float = 30.0


class APIError(Exception):
    # Provider-neutral exception used by the runnable example.
    def __init__(
        self,
        message: str,
        *,
        status_code: Optional[int] = None,
        retry_after: Optional[float] = None,
    ) -> None:
        super().__init__(message)
        self.status_code = status_code
        self.retry_after = retry_after


def is_transient(exc: BaseException) -> bool:
    # Timeouts and connection failures may succeed when tried again later.
    if isinstance(exc, (asyncio.TimeoutError, ConnectionError, OSError)):
        return True

    # Retry rate limits and server errors, but fail fast for other 4xx errors.
    status = getattr(exc, "status_code", None)
    if status is not None:
        if status == 429 or 500 <= status < 600:
            return True
        return False

    # Unknown failures are not retried by default.
    return False


def _parse_retry_after(exc: Exception) -> Optional[float]:
    # Read a provider retry hint without depending on a specific SDK type.
    raw = getattr(exc, "retry_after", None)
    if raw is None:
        return None

    try:
        value = float(raw)
        # Ignore negative hints because a wait cannot be negative.
        return value if value >= 0 else None
    except (TypeError, ValueError):
        # Invalid hints fall back to exponential backoff with jitter.
        return None


async def call_llm_with_retry(
    call: Callable[[str], Awaitable[Any]],
    cfg: RetryConfig,
    idempotency_key: str,
) -> Any:
    """Call an LLM API with bounded retries and one stable idempotency key."""

    last_exc: Optional[BaseException] = None

    # Try at most max_attempts times and return immediately on success.
    for attempt in range(1, cfg.max_attempts + 1):
        try:
            # Reuse the same key for every attempt of this logical request.
            return await asyncio.wait_for(
                call(idempotency_key),
                timeout=cfg.timeout,
            )
        except asyncio.CancelledError:
            # Cancellation must stop immediately and must never become a retry.
            raise
        except Exception as exc:
            # Remember the current failure in case defensive fallback is needed.
            last_exc = exc

            # Permanent failures should fail fast instead of repeating the request.
            if not is_transient(exc):
                raise

            # Do not calculate or sleep after the final allowed failed attempt.
            if attempt >= cfg.max_attempts:
                raise

            retry_after = _parse_retry_after(exc)
            if retry_after is not None:
                # Honor the server hint, but keep it inside the configured cap.
                wait = min(retry_after, cfg.max_delay)
            else:
                # Increase the possible delay exponentially, then cap it.
                backoff_cap = min(
                    cfg.base_delay * (2 ** (attempt - 1)),
                    cfg.max_delay,
                )

                # Full jitter spreads clients across the whole allowed range.
                wait = random.uniform(0.0, backoff_cap)

            # Sleep only when the failure is transient and another attempt remains.
            await asyncio.sleep(max(0.0, wait))

    # Defensive fallback. Normal execution returns or raises inside the loop.
    if last_exc is not None:
        raise last_exc
    raise RuntimeError("Retry loop ended without a result")


class FakeLLMAPI:
    """Deterministic fake that follows the diagram's 429 -> 503 -> 200 flow."""

    def __init__(self) -> None:
        self.attempt: int = 0

    async def __call__(self, idempotency_key: str) -> str:
        # Count each API attempt while accepting the same idempotency key.
        self.attempt += 1
        print(f"attempt={self.attempt}, key={idempotency_key}")

        if self.attempt == 1:
            # Attempt 1: rate limited with Retry-After: 2 seconds.
            raise APIError(
                "Too Many Requests",
                status_code=429,
                retry_after=2.0,
            )

        if self.attempt == 2:
            # Attempt 2: temporary server failure, so normal backoff is used.
            raise APIError("Service Unavailable", status_code=503)

        # Attempt 3: success stops the retry loop immediately.
        return "200 OK"


async def main() -> None:
    # One stable key represents this one logical LLM request.
    idempotency_key = "48e2f9b7-c71a-4c44-8e48-a8c7c2b5a3e1"
    cfg = RetryConfig()
    fake_api = FakeLLMAPI()

    # Seed randomness only so the local demonstration is repeatable.
    random.seed(7)

    response = await call_llm_with_retry(
        fake_api,
        cfg,
        idempotency_key,
    )
    print("response:", response)


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

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from functools import total_ordering
import heapq
import math
from typing import Protocol, Sequence


Vector = list[float]


class EmbeddingProvider(Protocol):
    async def embed(self, texts: Sequence[str]) -> list[Vector]:
        """Return one embedding vector for each supplied text."""
        ...


@dataclass(frozen=True)
class SearchResult:
    id: str
    score: float


@total_ordering
class _ReverseId:
    """Reverse ID order so a larger ID is worse when scores tie."""

    def __init__(self, value: str) -> None:
        self.value = value

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, _ReverseId):
            return NotImplemented
        return self.value == other.value

    def __lt__(self, other: _ReverseId) -> bool:
        # Reverse normal string order inside the min-heap.
        # This makes the larger ID reach the root first on an equal score.
        return self.value > other.value


class SemanticSearch:
    def __init__(self, provider: EmbeddingProvider) -> None:
        self._provider = provider

        # Each stored entry is: (document ID, normalized embedding vector).
        self._docs: list[tuple[str, Vector]] = []

        # The first valid vector sets the index dimension.
        self._dim: int | None = None

    @staticmethod
    def _l2_normalize(vector: Sequence[float]) -> Vector:
        # Calculate the Euclidean length used by L2 normalization.
        norm = math.sqrt(sum(value * value for value in vector))

        # Avoid division by zero for an all-zero embedding.
        if norm == 0.0:
            return [0.0 for _ in vector]

        # Unit-length vectors let cosine similarity use a dot product.
        return [value / norm for value in vector]

    async def add_texts(
        self,
        items: Sequence[tuple[str, str]],
        batch: int = 32,
    ) -> None:
        """Embed documents in batches and store normalized vectors."""
        # Empty input does not change the index.
        if not items:
            return

        # A non-positive batch size would make batching invalid.
        if batch <= 0:
            raise ValueError("batch must be greater than zero")

        texts = [text for _, text in items]

        # Send several texts per async provider call.
        for start in range(0, len(texts), batch):
            chunk = texts[start : start + batch]
            vectors = await self._provider.embed(chunk)

            # The provider must return one vector for every text in this batch.
            if len(vectors) != len(chunk):
                raise ValueError("Embedding provider returned the wrong vector count")

            # Pair every returned vector with its original document ID.
            for (doc_id, _), vector in zip(
                items[start : start + batch],
                vectors,
            ):
                normalized = self._l2_normalize(vector)

                # The first vector defines the shared embedding dimension.
                if self._dim is None:
                    self._dim = len(normalized)

                # Later vectors must have the same dimension.
                elif len(normalized) != self._dim:
                    raise ValueError(
                        f"Dimension mismatch: expected {self._dim}, got {len(normalized)}"
                    )

                # Store the ID beside its normalized embedding.
                self._docs.append((doc_id, normalized))

    async def search(
        self,
        query: str,
        top_k: int = 5,
    ) -> list[SearchResult]:
        """Return stable top-k document IDs and cosine scores."""
        # An empty index has no possible matches.
        if not self._docs:
            return []

        # Asking for zero or fewer results also returns nothing.
        if top_k <= 0:
            return []

        # Embed the query through the same provider boundary.
        query_vectors = await self._provider.embed([query])

        # Defensive handling if the provider returns no query vector.
        if not query_vectors:
            return []

        # The query request should produce exactly one vector.
        if len(query_vectors) != 1:
            raise ValueError("Embedding provider returned the wrong query vector count")

        # Normalize the query with the same rule used for documents.
        query_vector = self._l2_normalize(query_vectors[0])

        # Dot products require equal vector dimensions.
        if self._dim is None or len(query_vector) != self._dim:
            raise ValueError(f"Query dimension {len(query_vector)} != index dimension {self._dim}")

        limit = min(top_k, len(self._docs))

        # The heap stores at most k candidates.
        # The root is the current worst selected candidate.
        heap: list[tuple[float, _ReverseId, str]] = []

        for doc_id, doc_vector in self._docs:
            # Both vectors are normalized.
            # Their dot product is therefore the cosine score for nonzero vectors.
            score = sum(
                query_value * doc_value
                for query_value, doc_value in zip(
                    query_vector,
                    doc_vector,
                )
            )

            # A smaller document ID is better when two scores are equal.
            entry = (score, _ReverseId(doc_id), doc_id)

            # Fill the heap until it contains k candidates.
            if len(heap) < limit:
                heapq.heappush(heap, entry)
                continue

            # Replace the current worst candidate only when this one is better.
            if entry > heap[0]:
                heapq.heapreplace(heap, entry)

        # Convert the heap into the stable output order shown in the diagram:
        # score descending, then document ID ascending.
        selected = [(score, doc_id) for score, _, doc_id in heap]
        selected.sort(key=lambda item: (-item[0], item[1]))

        # Return structured IDs and scores.
        return [SearchResult(id=doc_id, score=round(score, 6)) for score, doc_id in selected]


class FakeEmbeddingProvider:
    """Deterministic provider used only for the runnable example."""

    def __init__(self) -> None:
        # The query vector is [1, 0].
        # Every document vector below has length 1.
        # Its first value is therefore its cosine score with the query.
        self._vectors: dict[str, Vector] = {
            "async in Python": [1.0, 0.0],
            "Python async tips": self._unit_vector(0.72),
            "Vector databases explained": self._unit_vector(0.86),
            "What is cosine similarity?": self._unit_vector(0.20),
            "D5 example document": self._unit_vector(0.45),
            "Guide to LLMs": self._unit_vector(-0.12),
        }

    @staticmethod
    def _unit_vector(score: float) -> Vector:
        # Build a unit vector [score, y] where score^2 + y^2 = 1.
        second = math.sqrt(max(0.0, 1.0 - score * score))
        return [score, second]

    async def embed(self, texts: Sequence[str]) -> list[Vector]:
        # Return known vectors so the example is deterministic.
        # No external model call is made.
        return [list(self._vectors[text]) for text in texts]


async def main() -> None:
    provider = FakeEmbeddingProvider()
    search = SemanticSearch(provider)

    # Build the small in-memory document index used by the diagram example.
    await search.add_texts(
        [
            ("D1", "Python async tips"),
            ("D2", "Vector databases explained"),
            ("D3", "What is cosine similarity?"),
            ("D5", "D5 example document"),
            ("D7", "Guide to LLMs"),
        ]
    )

    # Run the exact query and top-3 request shown in the diagram.
    results = await search.search("async in Python", top_k=3)

    # Expected stable output:
    # D2: 0.86
    # D1: 0.72
    # D5: 0.45
    for result in results:
        print(f"{result.id}: {result.score:.2f}")


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

from __future__ import annotations

import asyncio
import math
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol, TypeVar


T = TypeVar("T")


@dataclass(frozen=True)
class Document:
    source: str
    page: int | None
    text: str


@dataclass(frozen=True)
class Chunk:
    chunk_id: str
    text: str
    source: str
    page: int | None
    start_char: int
    end_char: int


@dataclass(frozen=True)
class RetrievedChunk(Chunk):
    score: float


@dataclass(frozen=True)
class Source:
    source: str
    page: int | None
    chunk_id: str
    score: float


@dataclass(frozen=True)
class Answer:
    answer: str
    sources: list[Source]
    no_evidence: bool


class DocumentLoader(Protocol):
    async def load(self, paths: list[Path]) -> list[Document]: ...


class EmbeddingModel(Protocol):
    async def embed(self, texts: list[str]) -> list[list[float]]: ...


class VectorDatabase(Protocol):
    async def upsert(
        self,
        chunks: list[Chunk],
        vectors: list[list[float]],
    ) -> None: ...

    async def search(
        self,
        query_vector: list[float],
        k: int,
    ) -> list[RetrievedChunk]: ...


class Generator(Protocol):
    async def generate(self, query: str, context: str) -> str: ...


class EvidencePolicy(Protocol):
    def accepts(self, results: list[RetrievedChunk]) -> bool: ...


async def call_with_timeout_and_retries(
    operation: Callable[[], Awaitable[T]],
    *,
    timeout_seconds: float = 2.0,
    max_attempts: int = 3,
) -> T:
    # Retry only transient timeout/connection failures. The retry count is bounded.
    for attempt in range(1, max_attempts + 1):
        try:
            # asyncio.timeout cancels an operation that exceeds the per-call limit.
            async with asyncio.timeout(timeout_seconds):
                return await operation()
        except (TimeoutError, ConnectionError):
            if attempt == max_attempts:
                raise

            # Back off briefly before the next bounded retry.
            await asyncio.sleep(0.05 * (2 ** (attempt - 1)))

    raise RuntimeError("unreachable")


class FakeDocumentLoader:
    """Deterministic loader used only by the runnable example."""

    async def load(self, paths: list[Path]) -> list[Document]:
        # Map the example file names to fixed source text and page provenance.
        fixtures = {
            "guide.pdf": Document(
                source="guide.pdf",
                page=2,
                text=(
                    "RAG uses retrieval to bring relevant documents into the "
                    "model context before generation."
                ),
            ),
            "notes.md": Document(
                source="notes.md",
                page=5,
                text=("Retrieved evidence grounds the answer and lets the system return sources."),
            ),
            "handbook.md": Document(
                source="handbook.md",
                page=7,
                text=(
                    "A bounded context keeps only a limited amount of retrieved "
                    "evidence in the prompt."
                ),
            ),
        }

        # Fail clearly if the caller asks for a source outside this deterministic demo.
        return [fixtures[path.name] for path in paths]


class DeterministicEmbeddingModel:
    """Deterministic fake that creates the diagram's illustrative similarity scores."""

    _vectors = {
        "What is RAG?": [1.0, 0.0],
        (
            "RAG uses retrieval to bring relevant documents into the "
            "model context before generation."
        ): [0.92, math.sqrt(1.0 - 0.92**2)],
        ("Retrieved evidence grounds the answer and lets the system return sources."): [
            0.87,
            math.sqrt(1.0 - 0.87**2),
        ],
        ("A bounded context keeps only a limited amount of retrieved evidence in the prompt."): [
            0.73,
            math.sqrt(1.0 - 0.73**2),
        ],
    }

    async def embed(self, texts: list[str]) -> list[list[float]]:
        # Return fixed normalized vectors so the runnable example is repeatable.
        try:
            return [self._vectors[text] for text in texts]
        except KeyError as exc:
            raise ValueError("No deterministic embedding fixture for this text.") from exc


class InMemoryVectorDatabase:
    """Exact-scan vector database used only for the runnable example."""

    def __init__(self) -> None:
        self._rows: list[tuple[Chunk, list[float]]] = []

    async def upsert(
        self,
        chunks: list[Chunk],
        vectors: list[list[float]],
    ) -> None:
        # Keep each vector beside the chunk text and provenance metadata.
        self._rows = list(zip(chunks, vectors, strict=True))

    async def search(
        self,
        query_vector: list[float],
        k: int,
    ) -> list[RetrievedChunk]:
        scored: list[RetrievedChunk] = []

        for chunk, vector in self._rows:
            # The demo vectors are normalized, so dot product equals cosine similarity.
            score = sum(left * right for left, right in zip(query_vector, vector, strict=True))
            scored.append(
                RetrievedChunk(
                    chunk_id=chunk.chunk_id,
                    text=chunk.text,
                    source=chunk.source,
                    page=chunk.page,
                    start_char=chunk.start_char,
                    end_char=chunk.end_char,
                    score=score,
                )
            )

        # Return the top-k candidates from highest to lowest similarity.
        scored.sort(key=lambda item: item.score, reverse=True)
        return scored[:k]


class DeterministicGenerator:
    """Deterministic fake used only to keep the example provider-neutral and testable."""

    async def generate(self, query: str, context: str) -> str:
        # A real generator would receive the same query and bounded context.
        if not context.strip():
            raise ValueError("Context must not be empty on the evidence-present path.")

        # Keep the fake grounded in the deterministic retrieved evidence.
        if "RAG uses retrieval" not in context:
            raise ValueError("Expected supporting RAG evidence in the bounded context.")

        return (
            "RAG uses retrieved, relevant documents to ground the model and improve answer quality."
        )


class AcceptNonEmptyEvidence:
    """Deterministic test policy; production can apply any configured relevance rule."""

    def accepts(self, results: list[RetrievedChunk]) -> bool:
        # The real policy may use scores or other retrieval signals.
        return bool(results)


class BasicRAG:
    def __init__(
        self,
        *,
        loader: DocumentLoader,
        embedding_model: EmbeddingModel,
        vector_db: VectorDatabase,
        generator: Generator,
        evidence_policy: EvidencePolicy,
    ) -> None:
        self.loader = loader
        self.embedding_model = embedding_model
        self.vector_db = vector_db
        self.generator = generator
        self.evidence_policy = evidence_policy

    async def ingest(self, paths: list[Path]) -> list[Document]:
        # Keep file loading behind an async boundary with timeout and bounded retries.
        return await call_with_timeout_and_retries(lambda: self.loader.load(paths))

    def chunk(
        self,
        documents: list[Document],
        *,
        size: int = 800,
        overlap: int = 120,
    ) -> list[Chunk]:
        # Validate the overlap so every loop iteration moves forward.
        if size <= 0 or overlap < 0 or overlap >= size:
            raise ValueError("Require size > 0 and 0 <= overlap < size.")

        chunks: list[Chunk] = []
        next_id = 1

        for document in documents:
            start = 0

            while start < len(document.text):
                end = min(start + size, len(document.text))

                # Keep source, page, and character offsets with every chunk.
                chunk_id = "ck" if next_id == 3 else f"c{next_id}"
                chunks.append(
                    Chunk(
                        chunk_id=chunk_id,
                        text=document.text[start:end],
                        source=document.source,
                        page=document.page,
                        start_char=start,
                        end_char=end,
                    )
                )
                next_id += 1

                if end == len(document.text):
                    break

                # Move forward while keeping the requested overlap.
                start = end - overlap

        return chunks

    async def embed(self, texts: list[str]) -> list[list[float]]:
        # Keep the embedding provider behind the same timeout/retry wrapper.
        return await call_with_timeout_and_retries(lambda: self.embedding_model.embed(texts))

    async def index(
        self,
        chunks: list[Chunk],
        vectors: list[list[float]],
    ) -> None:
        # Store vectors together with chunk text and provenance.
        await call_with_timeout_and_retries(lambda: self.vector_db.upsert(chunks, vectors))

    async def retrieve(
        self,
        query: str,
        *,
        k: int = 5,
    ) -> list[RetrievedChunk]:
        # Embed the query with the same model used for document chunks.
        query_vector = (await self.embed([query]))[0]

        # Retrieve the top-k candidates from the vector database.
        return await call_with_timeout_and_retries(lambda: self.vector_db.search(query_vector, k))

    def has_sufficient_evidence(
        self,
        results: list[RetrievedChunk],
    ) -> bool:
        # Keep the evidence rule configurable instead of inventing one fixed threshold.
        return self.evidence_policy.accepts(results)

    def build_context(
        self,
        results: list[RetrievedChunk],
        *,
        max_tokens: int = 800,
    ) -> tuple[str, list[RetrievedChunk]]:
        # This deterministic demo counts whitespace-separated words as test tokens.
        # Production code should use the generator model's real tokenizer.
        context_parts: list[str] = []
        used_results: list[RetrievedChunk] = []
        seen_chunk_ids: set[str] = set()
        used_tokens = 0

        for result in results:
            # Deduplicate by chunk ID before adding retrieved evidence.
            if result.chunk_id in seen_chunk_ids:
                continue

            words = result.text.split()
            remaining = max_tokens - used_tokens
            if remaining <= 0:
                break

            selected_words = words[:remaining]
            if not selected_words:
                continue

            # Carry provenance into the bounded context for later source reporting.
            context_parts.append(
                f"[source={result.source}, page={result.page}, "
                f"chunk_id={result.chunk_id}, score={result.score:.2f}]\n"
                + " ".join(selected_words)
            )
            used_results.append(result)
            seen_chunk_ids.add(result.chunk_id)
            used_tokens += len(selected_words)

        return "\n\n".join(context_parts), used_results

    async def generate(
        self,
        query: str,
        context: str,
        used_results: list[RetrievedChunk],
    ) -> Answer:
        # Generate only on the evidence-present path and only from bounded context.
        answer_text = await call_with_timeout_and_retries(
            lambda: self.generator.generate(query, context)
        )

        # The example answer cites the first two supporting chunks shown in the diagram.
        cited_results = used_results[:2]
        sources = [
            Source(
                source=result.source,
                page=result.page,
                chunk_id=result.chunk_id,
                score=result.score,
            )
            for result in cited_results
        ]

        return Answer(
            answer=answer_text,
            sources=sources,
            no_evidence=False,
        )

    def no_evidence_answer(self) -> Answer:
        # Do not call the generator when retrieved evidence fails the configured policy.
        return Answer(
            answer=(
                "I don't have enough information in the provided sources to answer this question."
            ),
            sources=[],
            no_evidence=True,
        )

    async def answer(
        self,
        query: str,
        *,
        k: int = 5,
        max_tokens: int = 800,
    ) -> Answer:
        # Retrieval happens first. Evidence gating happens only after top-k retrieval.
        results = await self.retrieve(query, k=k)

        if not self.has_sufficient_evidence(results):
            return self.no_evidence_answer()

        # Only accepted evidence is assembled into bounded context.
        context, used_results = self.build_context(
            results,
            max_tokens=max_tokens,
        )

        if not context:
            return self.no_evidence_answer()

        return await self.generate(query, context, used_results)


async def main() -> None:
    rag = BasicRAG(
        loader=FakeDocumentLoader(),
        embedding_model=DeterministicEmbeddingModel(),
        vector_db=InMemoryVectorDatabase(),
        generator=DeterministicGenerator(),
        evidence_policy=AcceptNonEmptyEvidence(),
    )

    # Ingest the three deterministic example sources.
    paths = [Path("guide.pdf"), Path("notes.md"), Path("handbook.md")]
    documents = await rag.ingest(paths)

    # Chunk with provenance, embed every chunk, and index vectors plus metadata.
    chunks = rag.chunk(documents, size=800, overlap=120)
    vectors = await rag.embed([chunk.text for chunk in chunks])
    await rag.index(chunks, vectors)

    # Run the same query and retrieval settings shown in the diagram.
    answer = await rag.answer(
        "What is RAG?",
        k=5,
        max_tokens=800,
    )

    print(answer.answer)
    for source in answer.sources:
        print(
            f"- {source.source}, p. {source.page}, "
            f"chunk {source.chunk_id}, score {source.score:.2f}"
        )


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

from __future__ import annotations

import ast
import asyncio
import json
import operator
import time
from collections.abc import Awaitable, Callable
from typing import Any, Literal, TypedDict


class ToolCall(TypedDict):
    name: str
    args: dict[str, Any]


class Step(TypedDict, total=False):
    action: ToolCall
    observation: str


class Message(TypedDict):
    role: Literal["system", "user", "assistant", "tool"]
    content: str


class Tool(TypedDict):
    name: str
    description: str
    input_schema: dict[str, Any]
    func: Callable[[dict[str, Any]], Awaitable[Any]]
    timeout: float


class State(TypedDict):
    messages: list[Message]
    steps: list[Step]
    tools: dict[str, Tool]
    step: int
    max_steps: int
    start_time: float


class FinalAction(TypedDict):
    type: Literal["final"]
    answer: str


class ToolAction(TypedDict):
    type: Literal["tool_call"]
    name: str
    args: dict[str, Any]


ModelAction = FinalAction | ToolAction
Model = Callable[[list[Message]], Awaitable[ModelAction]]


_ALLOWED_BIN_OPS: dict[type[ast.operator], Callable[[float, float], float]] = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
}

_ALLOWED_UNARY_OPS: dict[type[ast.unaryop], Callable[[float], float]] = {
    ast.UAdd: operator.pos,
    ast.USub: operator.neg,
}


def safe_eval(expression: str) -> float:
    """Evaluate only simple numeric arithmetic. Never use eval() or exec()."""
    # Parse text into a syntax tree without executing the expression.
    tree = ast.parse(expression, mode="eval")

    def visit(node: ast.AST) -> float:
        # Numeric constants are safe leaf values. Booleans are rejected.
        if (
            isinstance(node, ast.Constant)
            and isinstance(node.value, (int, float))
            and not isinstance(node.value, bool)
        ):
            return float(node.value)

        # Only explicitly allowlisted binary operators may run.
        if isinstance(node, ast.BinOp) and type(node.op) in _ALLOWED_BIN_OPS:
            left = visit(node.left)
            right = visit(node.right)
            return _ALLOWED_BIN_OPS[type(node.op)](left, right)

        # Allow only unary plus and unary minus.
        if isinstance(node, ast.UnaryOp) and type(node.op) in _ALLOWED_UNARY_OPS:
            return _ALLOWED_UNARY_OPS[type(node.op)](visit(node.operand))

        # Reject calls, names, attributes, and every other AST node.
        raise ValueError("Unsupported calculator expression")

    return visit(tree.body)


def validate_args(schema: dict[str, Any], args: dict[str, Any]) -> None:
    """Validate the small object schemas used by this example."""
    # Tool arguments must be a JSON-like object.
    if not isinstance(args, dict):
        raise ValueError("Tool arguments must be an object")

    required = schema.get("required", [])
    properties = schema.get("properties", {})

    # Every required field must be present before the tool runs.
    for name in required:
        if name not in args:
            raise ValueError(f"Missing required argument: {name}")

    # Reject unknown fields when additionalProperties is false.
    if schema.get("additionalProperties") is False:
        unknown = set(args) - set(properties)
        if unknown:
            raise ValueError(f"Unknown arguments: {sorted(unknown)}")

    # Check the string and integer types used by the tool schemas.
    for name, value in args.items():
        expected = properties.get(name, {}).get("type")
        if expected == "string" and not isinstance(value, str):
            raise ValueError(f"{name} must be a string")
        if expected == "integer" and (not isinstance(value, int) or isinstance(value, bool)):
            raise ValueError(f"{name} must be an integer")


async def calculator_tool(args: dict[str, Any]) -> dict[str, str]:
    # Arguments were validated before this function is called.
    result = safe_eval(args["expr"])

    # Match the diagram example by displaying 36 instead of 36.0.
    text = str(int(result)) if result.is_integer() else str(result)
    return {"result": text}


def tool_schema_message(tools: dict[str, Tool]) -> Message:
    # Give the model only the allowlisted tool names and their input contracts.
    schemas = {
        name: {
            "description": tool["description"],
            "input_schema": tool["input_schema"],
        }
        for name, tool in tools.items()
    }
    return {
        "role": "system",
        "content": (
            "Available tools and schemas: "
            + json.dumps(schemas, ensure_ascii=False)
            + ". Return one structured action. Use type=final with an answer, "
            "or type=tool_call with a tool name and args."
        ),
    }


def build_prompt(state: State) -> list[Message]:
    # State already contains the system tool schemas and full conversation.
    return state["messages"]


async def execute_tool(
    action: ToolCall,
    tools: dict[str, Tool],
) -> dict[str, Any]:
    # The registry is the allowlist. Unknown tool names never execute.
    tool = tools.get(action["name"])
    if tool is None:
        return {"ok": False, "error": "Unknown tool"}

    try:
        # Validate typed arguments before calling the tool implementation.
        validate_args(tool["input_schema"], action["args"])

        # Bound each async tool call with its configured timeout.
        result = await asyncio.wait_for(
            tool["func"](action["args"]),
            timeout=tool["timeout"],
        )
        return {"ok": True, "result": result}
    except TimeoutError:
        return {"ok": False, "error": "Tool timeout"}
    except Exception as exc:
        # Validation and tool failures become observations for the next turn.
        return {"ok": False, "error": str(exc)}


async def run_agent(
    user_input: str,
    model: Model,
    tools: dict[str, Tool],
    max_steps: int = 8,
) -> str:
    # Initialize explicit state before the first model turn.
    state: State = {
        "messages": [
            tool_schema_message(tools),
            {"role": "user", "content": user_input},
        ],
        "steps": [],
        "tools": tools,
        "step": 0,
        "max_steps": max_steps,
        "start_time": time.monotonic(),
    }

    # The bounded loop permits at most max_steps model turns.
    for _ in range(max_steps):
        prompt = build_prompt(state)
        action = await model(prompt)

        # A final action is the normal successful stop condition.
        if action["type"] == "final":
            return action["answer"]

        # Build the structured tool call requested by the model.
        tool_call: ToolCall = {
            "name": action["name"],
            "args": action["args"],
        }

        # Record the requested action in the observable message history.
        state["messages"].append(
            {
                "role": "assistant",
                "content": json.dumps(
                    {"type": "tool_call", **tool_call},
                    ensure_ascii=False,
                ),
            }
        )

        # Execute the allowlisted tool and turn success or failure into an observation.
        result = await execute_tool(tool_call, tools)
        observation = json.dumps(result, ensure_ascii=False)

        # Update both message history and the explicit observable step log.
        state["messages"].append({"role": "tool", "content": observation})
        state["steps"].append({"action": tool_call, "observation": observation})
        state["step"] += 1

    # Do not make another model or tool call after max_steps is reached.
    return "Stop: reached max steps without a final answer."


class FakeModel:
    """Deterministic model used only to run the diagram example."""

    def __init__(self) -> None:
        self.calls = 0

    async def __call__(self, messages: list[Message]) -> ModelAction:
        self.calls += 1

        # First request the exact calculator call shown in the diagram.
        if self.calls == 1:
            return {
                "type": "tool_call",
                "name": "calculator",
                "args": {"expr": "0.15 * 240"},
            }

        # Then return the final response after the calculator observation exists.
        return {
            "type": "final",
            "answer": "15% of 240 is 36.",
        }


async def fake_web_search(args: dict[str, Any]) -> list[dict[str, str]]:
    # This deterministic fake is for tests. It is not presented as live web data.
    query = args["query"]
    max_results = args.get("max_results", 3)
    results = [
        {
            "title": "TEST ONLY",
            "url": "https://example.test/",
            "snippet": f"Deterministic fake result for query: {query}",
        }
    ]
    return results[:max_results]


def make_tools() -> dict[str, Tool]:
    # Only these two registered names can be executed by the controller.
    return {
        "calculator": {
            "name": "calculator",
            "description": "Evaluate a safe arithmetic expression.",
            "input_schema": {
                "type": "object",
                "properties": {"expr": {"type": "string"}},
                "required": ["expr"],
                "additionalProperties": False,
            },
            "func": calculator_tool,
            "timeout": 10.0,
        },
        "web_search": {
            "name": "web_search",
            "description": "Search through an injected async function.",
            "input_schema": {
                "type": "object",
                "properties": {"query": {"type": "string"}, "max_results": {"type": "integer"}},
                "required": ["query"],
                "additionalProperties": False,
            },
            "func": fake_web_search,
            "timeout": 10.0,
        },
    }


async def main() -> None:
    # Run the same deterministic example shown in the diagram.
    answer = await run_agent(
        user_input="What is 15% of 240?", model=FakeModel(), tools=make_tools(), max_steps=8
    )
    print(answer)


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