windlass.providers.llm.fake

fake

A deterministic, dependency-free LLM for tests, demos and CI.

FakeLLM is not a toy. It is the reason the entire Windlass test suite runs offline in a few seconds with no API keys, and the reason pip install windlass alone gives you a working end-to-end pipeline to learn from.

It can:

• replay a scripted list of responses;
• answer via a callable, so a test can assert on the prompt it received;
• emit tool calls on cue, exercising the whole agent loop;
• record every call it received for assertions.

Example

llm = FakeLLM(responses=["first", "second"]) llm.complete("a").content, llm.complete("b").content ('first', 'second') llm.calls[0][-1].content 'a'

FakeLLM

FakeLLM(
    responses: Sequence[str] | str | None = None,
    *,
    handler: Callable[[list[Message], list[dict[str, Any]] | None], Any] | None = None,
    tool_calls: Sequence[Sequence[ToolCall]] | None = None,
    cycle: bool = False,
    latency: float = 0.0,
    model: str = "fake-1",
    **config: Any
)

Bases: LLM

A scripted language model.

Parameters:

Name	Type	Description	Default
responses	Sequence[str] | str | None	Texts to return, one per call. Exhausting the list cycles back to the start when cycle is True, otherwise the last entry repeats.	None
handler	Callable[[list[Message], list[dict[str, Any]] | None], Any] | None	Callable receiving (messages, tools) and returning either a string or a full :class:~windlass.core.types.Completion. Takes precedence over responses.	None
tool_calls	Sequence[Sequence[ToolCall]] | None	Tool calls to emit, one list per call. Use this to drive an agent through a specific tool-use path in a test.	None
cycle	bool	Whether responses wraps around.	False
latency	float	Artificial delay per call in seconds, for testing timeouts and progress indicators.	0.0
model	str	Reported model name.	'fake-1'
**config	Any	Forwarded to :class:~windlass.interfaces.llm.LLM.	{}

Attributes:

Name	Type	Description
calls	list[list[Message]]	Every message list this model was asked to complete.
tool_schemas	list[list[dict[str, Any]] | None]	The tool definitions passed on each call.

Example

Driving an agent through one tool call, then a final answer::

llm = FakeLLM(
    responses=["", "The weather is sunny."],
    tool_calls=[[ToolCall(name="weather", arguments={"city": "Paris"})], []],
)

Source code in src\windlass\providers\llm\fake.py

def __init__(
    self,
    responses: Sequence[str] | str | None = None,
    *,
    handler: Callable[[list[Message], list[dict[str, Any]] | None], Any] | None = None,
    tool_calls: Sequence[Sequence[ToolCall]] | None = None,
    cycle: bool = False,
    latency: float = 0.0,
    model: str = "fake-1",
    **config: Any,
) -> None:
    super().__init__(model=model, **config)
    if isinstance(responses, str):
        responses = [responses]
    self.responses: list[str] = list(responses or ["This is a fake response."])
    self.handler = handler
    self.tool_calls: list[list[ToolCall]] = [list(t) for t in (tool_calls or [])]
    self.cycle = cycle
    self.latency = latency
    self.calls: list[list[Message]] = []
    self.tool_schemas: list[list[dict[str, Any]] | None] = []
    self._counter = itertools.count()

default_model classmethod

default_model() -> str

Return "fake-1".

Source code in src\windlass\providers\llm\fake.py

@classmethod
def default_model(cls) -> str:
    """Return ``"fake-1"``."""
    return "fake-1"

agenerate async

agenerate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> Completion

Return the next scripted response.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation, recorded on :attr:calls.	required
tools	list[dict[str, Any]] | None	Tool definitions, recorded on :attr:tool_schemas.	None
**kwargs	Any	Ignored.	{}

Returns:

Type	Description
Completion	The scripted completion.

Source code in src\windlass\providers\llm\fake.py

async def agenerate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> Completion:
    """Return the next scripted response.

    Args:
        messages: The conversation, recorded on :attr:`calls`.
        tools: Tool definitions, recorded on :attr:`tool_schemas`.
        **kwargs: Ignored.

    Returns:
        The scripted completion.
    """
    self.calls.append(list(messages))
    self.tool_schemas.append(tools)
    if self.latency:
        await asyncio.sleep(self.latency)

    index = next(self._counter)

    if self.handler is not None:
        produced = self.handler(messages, tools)
        if isinstance(produced, Completion):
            return produced
        return Completion(
            content=str(produced), model=self.model, usage=self._fake_usage(messages)
        )

    calls = self.tool_calls[index] if index < len(self.tool_calls) else []
    text = self._next_text(index)
    return Completion(
        content=text,
        tool_calls=list(calls),
        finish_reason=FinishReason.TOOL_CALLS if calls else FinishReason.STOP,
        model=self.model,
        usage=self._fake_usage(messages, text),
        raw={"fake": True, "index": index},
    )

astream_generate async

astream_generate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> AsyncIterator[StreamEvent]

Stream the scripted response one word at a time.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required
tools	list[dict[str, Any]] | None	Tool definitions.	None
**kwargs	Any	Ignored.	{}

Yields:

Type	Description
AsyncIterator[StreamEvent]	Word-level text events, then any tool calls, then done.

Source code in src\windlass\providers\llm\fake.py

async def astream_generate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]:
    """Stream the scripted response one word at a time.

    Args:
        messages: The conversation.
        tools: Tool definitions.
        **kwargs: Ignored.

    Yields:
        Word-level text events, then any tool calls, then ``done``.
    """
    completion = await self.agenerate(messages, tools=tools, **kwargs)
    words = completion.content.split(" ")
    for position, word in enumerate(words):
        if not word and len(words) == 1:
            break
        delta = word if position == len(words) - 1 else f"{word} "
        if self.latency:
            await asyncio.sleep(self.latency / max(1, len(words)))
        yield StreamEvent(type="text", delta=delta)
    for call in completion.tool_calls:
        yield StreamEvent(type="tool_call", tool_call=call)
    yield StreamEvent(
        type="done", finish_reason=completion.finish_reason, usage=completion.usage
    )

last_prompt

last_prompt() -> str

Return the text of the most recent user message.

Returns:

Type	Description
str	The last user message's content, or "" when there was none.

Raises:

Type	Description
IndexError	If the model has not been called yet.

Source code in src\windlass\providers\llm\fake.py

def last_prompt(self) -> str:
    """Return the text of the most recent user message.

    Returns:
        The last user message's content, or ``""`` when there was none.

    Raises:
        IndexError: If the model has not been called yet.
    """
    messages = self.calls[-1]
    for message in reversed(messages):
        if message.role.value == "user":
            return message.content
    return ""

reset

reset() -> None

Clear recorded calls and rewind the response script.

Source code in src\windlass\providers\llm\fake.py

def reset(self) -> None:
    """Clear recorded calls and rewind the response script."""
    self.calls.clear()
    self.tool_schemas.clear()
    self._counter = itertools.count()

EchoLLM

EchoLLM(prefix: str = '', **config: Any)

Bases: LLM

A model that echoes its input.

Useful for wiring tests where you need to prove that a prompt reached the model intact — the answer is the prompt.

Parameters:

Name	Type	Description	Default
prefix	str	Prepended to the echoed text.	''
**config	Any	Forwarded to :class:~windlass.interfaces.llm.LLM.	{}

Example

EchoLLM().complete("hello there").content 'hello there' EchoLLM(prefix="You said: ").complete("hi").content 'You said: hi'

Source code in src\windlass\providers\llm\fake.py

def __init__(self, prefix: str = "", **config: Any) -> None:
    super().__init__(model="echo-1", **config)
    self.prefix = prefix

default_model classmethod

default_model() -> str

Return "echo-1".

Source code in src\windlass\providers\llm\fake.py

@classmethod
def default_model(cls) -> str:
    """Return ``"echo-1"``."""
    return "echo-1"

agenerate async

agenerate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> Completion

Echo the last user message.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required
tools	list[dict[str, Any]] | None	Ignored — this model does not call tools.	None
**kwargs	Any	Ignored.	{}

Returns:

Type	Description
Completion	A completion containing the echoed text.

Source code in src\windlass\providers\llm\fake.py

async def agenerate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> Completion:
    """Echo the last user message.

    Args:
        messages: The conversation.
        tools: Ignored — this model does not call tools.
        **kwargs: Ignored.

    Returns:
        A completion containing the echoed text.
    """
    text = next(
        (m.content for m in reversed(messages) if m.role.value == "user"),
        messages[-1].content if messages else "",
    )
    return Completion(content=f"{self.prefix}{text}", model=self.model)