windlass.core.types

types

Canonical data model shared by every Windlass component.

These types are the lingua franca of the framework. A chunker produced by one plugin can feed a retriever from another because both speak :class:Chunk. Every model is a Pydantic v2 model, so it validates, serialises to JSON, and round-trips through checkpoints for free.

The rule of thumb: provider-specific shapes never cross a component boundary. Adapters translate into these types on the way in and out, and keep the untouched provider payload on .raw for callers who need it (Level 3 access).

Role

Bases: StrEnum

Who authored a :class:Message.

A :class:~enum.StrEnum, so members compare and format as their plain string value — Role.USER == "user" and f"{Role.USER}" both do what you would expect when the value reaches a provider's wire format.

FinishReason

Bases: StrEnum

Why the model stopped generating.

WindlassModel

Bases: BaseModel

Base model with the conventions every Windlass type shares.

• Extra keys are ignored rather than rejected, so a newer provider adapter can hand richer payloads to an older consumer without breaking it.
• Arbitrary types are allowed so raw can hold native SDK objects.
• Enum values serialise to their string form for clean JSON.

Usage

Bases: WindlassModel

Token accounting for one or more model calls.

Attributes:

Name	Type	Description
prompt_tokens	int	Tokens consumed by the prompt.
completion_tokens	int	Tokens produced by the model.
total_tokens	int	Sum of the two; computed when a provider omits it.
cached_tokens	int	Prompt tokens served from the provider's prompt cache.
cost_usd	float | None	Estimated spend, when the adapter can compute it.
calls	int	How many provider round-trips this usage aggregates.

ToolCall

Bases: WindlassModel

A model's request to invoke a tool.

Attributes:

Name	Type	Description
id	str	Provider-assigned correlation id; echoed back in the tool result.
name	str	Registered tool name.
arguments	dict[str, Any]	Parsed keyword arguments for the tool.
raw_arguments	str | None	The original JSON string, kept when parsing failed.

ToolResult

Bases: WindlassModel

The outcome of executing a :class:ToolCall.

Attributes:

Name	Type	Description
call_id	str	The :attr:ToolCall.id this answers.
name	str	Tool that produced the result.
content	str	String rendering handed back to the model.
data	Any	The unserialised Python return value, for programmatic access.
is_error	bool	True when the tool raised and the error text was returned.
duration_ms	float	Wall-clock execution time.

Message

Bases: WindlassModel

One turn in a conversation.

Attributes:

Name	Type	Description
role	Role	Author of the message.
content	str	Text body. May be empty when the turn is purely tool calls.
tool_calls	list[ToolCall]	Tools the assistant wants to invoke.
tool_call_id	str | None	Set on tool messages to correlate with a call.
name	str | None	Optional speaker name (multi-agent transcripts use this).
metadata	dict[str, Any]	Free-form annotations that never reach the provider.

Example

Message.user("hello") Message(role=, content='hello', ...)

system classmethod

system(content: str, **kw: Any) -> Message

Build a system message.

Source code in src\windlass\core\types.py

@classmethod
def system(cls, content: str, **kw: Any) -> Message:
    """Build a system message."""
    return cls(role=Role.SYSTEM, content=content, **kw)

user classmethod

user(content: str, **kw: Any) -> Message

Build a user message.

Source code in src\windlass\core\types.py

@classmethod
def user(cls, content: str, **kw: Any) -> Message:
    """Build a user message."""
    return cls(role=Role.USER, content=content, **kw)

assistant classmethod

assistant(content: str = '', **kw: Any) -> Message

Build an assistant message.

Source code in src\windlass\core\types.py

@classmethod
def assistant(cls, content: str = "", **kw: Any) -> Message:
    """Build an assistant message."""
    return cls(role=Role.ASSISTANT, content=content, **kw)

tool classmethod

tool(result: ToolResult) -> Message

Build the tool message that answers a :class:ToolCall.

Source code in src\windlass\core\types.py

@classmethod
def tool(cls, result: ToolResult) -> Message:
    """Build the ``tool`` message that answers a :class:`ToolCall`."""
    return cls(
        role=Role.TOOL,
        content=result.content,
        tool_call_id=result.call_id,
        name=result.name,
        metadata={"is_error": result.is_error},
    )

Completion

Bases: WindlassModel

A non-streaming model response.

Attributes:

Name	Type	Description
content	str	The generated text.
tool_calls	list[ToolCall]	Tools the model wants invoked, if any.
finish_reason	FinishReason	Why generation stopped.
model	str	Model identifier reported by the provider.
usage	Usage	Token accounting.
raw	Any	The untouched provider response object (Level 3 escape hatch).
latency_ms	float	Wall-clock time for the call.

to_message

to_message() -> Message

Render this completion as an assistant :class:Message.

Source code in src\windlass\core\types.py

def to_message(self) -> Message:
    """Render this completion as an assistant :class:`Message`."""
    return Message(role=Role.ASSISTANT, content=self.content, tool_calls=self.tool_calls)

StreamEvent

Bases: WindlassModel

One incremental update from a streaming call.

Attributes:

Name	Type	Description
type	Literal['text', 'tool_call', 'usage', 'done']	text for content deltas, tool_call when a call is complete, usage for accounting, done for the terminal event.
delta	str	Incremental text for text events.
tool_call	ToolCall | None	The completed call for tool_call events.
finish_reason	FinishReason | None	Present on the done event.
usage	Usage | None	Present on usage and done events when the provider reports it.
raw	Any	Untouched provider chunk.

Document

Bases: WindlassModel

A source document before chunking.

Attributes:

Name	Type	Description
id	str	Stable identifier. Defaults to a content hash so re-ingestion is idempotent.
content	str	Extracted plain text.
metadata	dict[str, Any]	Anything worth filtering or citing on later (page, author, section, url, ...). Kept alongside every chunk derived from it.
source	str | None	Where the document came from (path, URL, ...).
mimetype	str | None	Detected content type, when known.
created_at	float	Unix timestamp of ingestion.

Chunk

Bases: WindlassModel

A retrievable slice of a :class:Document.

Attributes:

Name	Type	Description
id	str	Stable identifier, derived from the document id and index.
content	str	The chunk text as it will be embedded and shown to the model.
metadata	dict[str, Any]	Document metadata plus chunk-level annotations.
document_id	str	The document this came from.
index	int	Zero-based position within the document.
embedding	list[float] | None	Optional dense vector; populated during ingestion.
parent_id	str | None	Set by hierarchical chunkers; points at the larger parent chunk that should be expanded at retrieval time.
start_char	int	Character offset in the source document.
end_char	int	Exclusive end offset in the source document.

with_embedding

with_embedding(vector: list[float]) -> Chunk

Return a copy carrying vector as its embedding.

Source code in src\windlass\core\types.py

def with_embedding(self, vector: list[float]) -> Chunk:
    """Return a copy carrying ``vector`` as its embedding."""
    return self.model_copy(update={"embedding": vector})

ScoredChunk

Bases: WindlassModel

A chunk together with its retrieval score.

Attributes:

Name	Type	Description
chunk	Chunk	The retrieved chunk.
score	float	Relevance score. Higher is better; the scale depends on the retriever (cosine similarity, BM25, reciprocal rank fusion, ...).
rank	int	One-based position in the result list.
retriever	str	Which retriever produced this hit. Hybrid search keeps this so you can see which leg contributed each result.

content property

content: str

Shortcut to chunk.content.

metadata property

metadata: dict[str, Any]

Shortcut to chunk.metadata.

Embedding

Bases: WindlassModel

A dense vector plus the text it encodes.

Attributes:

Name	Type	Description
vector	list[float]	The embedding values.
text	str	The text that was embedded.
model	str	Model identifier that produced the vector.

SearchResult

Bases: WindlassModel

The full result of a retrieval call.

Attributes:

Name	Type	Description
query	str	The query that was executed (post rewriting, if any).
hits	list[ScoredChunk]	Ranked chunks.
total	int	Number of candidates considered before truncation.
latency_ms	float	Wall-clock retrieval time.

texts

texts() -> list[str]

Return just the chunk texts, ranked.

Source code in src\windlass\core\types.py

def texts(self) -> list[str]:
    """Return just the chunk texts, ranked."""
    return [h.chunk.content for h in self.hits]

RAGAnswer

Bases: WindlassModel

The answer produced by a RAG pipeline.

Attributes:

Name	Type	Description
answer	str	The generated answer text.
question	str	The question that was asked.
contexts	list[ScoredChunk]	Chunks that were placed in the prompt, in prompt order.
usage	Usage	Token accounting for the generation call.
latency_ms	float	End-to-end wall-clock time including retrieval.
metadata	dict[str, Any]	Pipeline annotations (guardrail decisions, rewrites, ...).

sources property

sources: list[str]

Unique, order-preserving list of the sources cited in the context.

AgentStep

Bases: WindlassModel

One iteration of an agent's reason/act loop.

Attributes:

Name	Type	Description
index	int	Zero-based step number.
thought	str	Assistant text emitted during this step, if any.
tool_calls	list[ToolCall]	Tools the model requested.
tool_results	list[ToolResult]	Results of executing them.
usage	Usage	Tokens spent on this step.
node	str	Name of the graph node that ran (multi-agent / LangGraph).

AgentResponse

Bases: WindlassModel

The final result of an agent run.

Attributes:

Name	Type	Description
output	str	The agent's answer.
messages	list[Message]	Full transcript including tool traffic.
steps	list[AgentStep]	Structured per-iteration trace.
usage	Usage	Aggregate token accounting across the whole run.
latency_ms	float	End-to-end wall-clock time.
thread_id	str | None	Checkpoint thread, when checkpointing is enabled.
interrupted	bool	True when the run paused for human approval.
metadata	dict[str, Any]	Runtime annotations.

tool_calls property

tool_calls: list[ToolCall]

Every tool call made during the run, in order.

GuardrailResult

Bases: WindlassModel

Verdict from a guardrail check.

Attributes:

Name	Type	Description
allowed	bool	False when the content must be blocked.
content	str	Possibly rewritten content (redaction returns a sanitised copy).
detections	list[dict[str, Any]]	One entry per rule that fired.
rule	str | None	The first rule that blocked, if any.
stage	str	input or output.

modified property

modified: bool

True when the guardrail rewrote the content.

EvaluationResult

Bases: WindlassModel

Score for a single metric on a single sample.

Attributes:

Name	Type	Description
metric	str	Metric name, e.g. faithfulness.
score	float	Normalised score, conventionally in [0, 1].
passed	bool	Whether the score met the metric's threshold.
threshold	float | None	The threshold that was applied.
reason	str	Explanation, when the metric produces one.
sample_id	str	Identifier of the evaluated sample.

EvaluationReport

Bases: WindlassModel

Aggregate results of an evaluation run.

Attributes:

Name	Type	Description
results	list[EvaluationResult]	Every individual metric/sample score.
summary	dict[str, float]	Mean score per metric.
pass_rate	float	Fraction of results that passed their threshold.
samples	int	Number of samples evaluated.

new_id

new_id(prefix: str = '') -> str

Return a short unique identifier.

Parameters:

Name	Type	Description	Default
prefix	str	Optional namespace prefix, e.g. "doc" yields "doc_1f3a...".	''

Returns:

Type	Description
str	A URL-safe identifier that is unique per process invocation.

Source code in src\windlass\core\types.py

def new_id(prefix: str = "") -> str:
    """Return a short unique identifier.

    Args:
        prefix: Optional namespace prefix, e.g. ``"doc"`` yields ``"doc_1f3a..."``.

    Returns:
        A URL-safe identifier that is unique per process invocation.
    """
    raw = uuid.uuid4().hex[:16]
    return f"{prefix}_{raw}" if prefix else raw

content_id

content_id(text: str, prefix: str = '') -> str

Return a deterministic identifier derived from text.

Deterministic ids make ingestion idempotent: re-ingesting the same document overwrites the same rows instead of creating duplicates.

Parameters:

Name	Type	Description	Default
text	str	The content to hash.	required
prefix	str	Optional namespace prefix.	''

Returns:

Type	Description
str	A stable 16-character BLAKE2b digest, optionally prefixed.

Source code in src\windlass\core\types.py

def content_id(text: str, prefix: str = "") -> str:
    """Return a deterministic identifier derived from ``text``.

    Deterministic ids make ingestion idempotent: re-ingesting the same document
    overwrites the same rows instead of creating duplicates.

    Args:
        text: The content to hash.
        prefix: Optional namespace prefix.

    Returns:
        A stable 16-character BLAKE2b digest, optionally prefixed.
    """
    digest = hashlib.blake2b(text.encode("utf-8"), digest_size=8).hexdigest()
    return f"{prefix}_{digest}" if prefix else digest

normalize_messages

normalize_messages(
    prompt: str | Message | Sequence[Message] | Sequence[dict[str, Any]],
) -> list[Message]

Coerce any accepted prompt shape into a list of :class:Message.

Accepts a bare string, a single message, a sequence of messages, or a sequence of OpenAI-style dicts. This is what lets every Windlass entry point take "hello" as well as a full transcript.

Parameters:

Name	Type	Description	Default
prompt	str | Message | Sequence[Message] | Sequence[dict[str, Any]]	The prompt in any supported shape.	required

Returns:

Type	Description
list[Message]	A new list of validated messages.

Raises:

Type	Description
TypeError	If the input is not one of the supported shapes.

Source code in src\windlass\core\types.py

def normalize_messages(
    prompt: str | Message | Sequence[Message] | Sequence[dict[str, Any]],
) -> list[Message]:
    """Coerce any accepted prompt shape into a list of :class:`Message`.

    Accepts a bare string, a single message, a sequence of messages, or a
    sequence of OpenAI-style dicts. This is what lets every Windlass entry point
    take ``"hello"`` as well as a full transcript.

    Args:
        prompt: The prompt in any supported shape.

    Returns:
        A new list of validated messages.

    Raises:
        TypeError: If the input is not one of the supported shapes.
    """
    if isinstance(prompt, str):
        return [Message.user(prompt)]
    if isinstance(prompt, Message):
        return [prompt]
    if isinstance(prompt, Sequence):
        out: list[Message] = []
        for item in prompt:
            if isinstance(item, Message):
                out.append(item)
            elif isinstance(item, dict):
                out.append(Message.model_validate(item))
            else:
                raise TypeError(f"Cannot interpret {type(item).__name__} as a Message.")
        return out
    raise TypeError(f"Cannot interpret {type(prompt).__name__} as a prompt.")