Models API Reference

API reference for all core data models in anchor. Every model listed here is a Pydantic BaseModel (unless otherwise noted) and can be imported directly from anchor.

ContextItem

The atomic unit of context flowing through the pipeline. Items are frozen (immutable after creation) to prevent context poisoning bugs.

from anchor import ContextItem

class ContextItem(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    content: str
    source: SourceType
    score: float = Field(default=0.0, ge=0.0, le=1.0)
    priority: int = Field(default=5, ge=1, le=10)
    token_count: int = Field(default=0, ge=0)
    metadata: dict[str, Any] = Field(default_factory=dict)
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))

Fields:

[!TIP] Since ContextItem is frozen, use item.model_copy(update={"score": 0.9}) to create a modified copy.

Example

from anchor import ContextItem, SourceType

item = ContextItem(
    content="Context engineering is the discipline of building dynamic systems.",
    source=SourceType.RETRIEVAL,
    score=0.95,
    priority=5,
    token_count=9,
)
print(item.id)        # auto-generated UUID
print(item.content)   # "Context engineering is..."

ContextWindow

A complete context window ready for formatting. Token-aware and priority-ranked.

Properties

Methods

add_item(item) -> bool

Add an item if it fits within the token budget. Returns True if added, False if it would exceed the budget.

add_items_by_priority(items) -> list[ContextItem]

Add items sorted by priority (highest first, ties broken by score). Returns the overflow list -- items that did not fit.

ContextResult

The final output of the context pipeline.

QueryBundle

Encapsulates a query as it flows through the pipeline.

[!TIP] ContextPipeline.build() accepts a plain string and wraps it in a QueryBundle automatically.

TokenBudget

Manages token budget across all context sources with per-source allocations. Unallocated tokens form a shared pool.

Validation: The sum of all allocations[].max_tokens plus reserve_tokens must not exceed total_tokens.

Properties

Methods

get_allocation(source) -> int

Get the max tokens for a source type. Falls back to the shared pool if the source has no explicit allocation.

get_overflow_strategy(source) -> OverflowStrategy

Get the overflow strategy for a source type. Returns "truncate" for sources without an explicit allocation.

Example

from anchor import TokenBudget, BudgetAllocation, SourceType

budget = TokenBudget(
    total_tokens=8192,
    reserve_tokens=1200,
    allocations=[
        BudgetAllocation(source=SourceType.SYSTEM, max_tokens=800, priority=10),
        BudgetAllocation(source=SourceType.RETRIEVAL, max_tokens=3000, priority=5),
    ],
)
print(budget.shared_pool)                              # 3192
print(budget.get_allocation(SourceType.RETRIEVAL))     # 3000
print(budget.get_allocation(SourceType.MEMORY))        # 3192 (shared pool)

BudgetAllocation

Token allocation for a specific source type within a TokenBudget.

Budget Factory Functions

Three factory functions return pre-configured TokenBudget instances. All accept a single max_tokens: int parameter and raise ValueError if it is not positive.

Enums

SourceType

The origin type of a context item. A StrEnum.

from anchor import SourceType

OverflowStrategy

Type alias for Literal["truncate", "drop"]. Controls behavior when a source exceeds its budget allocation.

• "truncate" -- include items up to the cap (sorted by priority/score).
• "drop" -- if total for that source exceeds the cap, all items are dropped.

Role

Message role for conversation turns. A StrEnum (imported from anchor.llm.models).

from anchor.llm import Role

[!TIP] Since Role is a StrEnum, str(Role.USER) returns "user" and Role.USER == "user" is True.

MemoryType

Classification of memory entries by cognitive type. A StrEnum.

Memory Models

ConversationTurn

A single turn in a conversation.

MemoryEntry

A persistent memory entry with relevance tracking.

from anchor import MemoryEntry

class MemoryEntry(BaseModel):
    id: str                          # Auto-generated UUID
    content: str                     # (required) Memory content text
    relevance_score: float = 0.5     # 0.0 to 1.0
    access_count: int = 0
    last_accessed: datetime           # Auto-set to now(UTC)
    created_at: datetime              # Auto-set to now(UTC)
    tags: list[str] = []
    metadata: dict[str, Any] = {}
    memory_type: MemoryType = MemoryType.SEMANTIC
    user_id: str | None = None       # Optional user scope
    session_id: str | None = None    # Optional session scope
    expires_at: datetime | None = None
    updated_at: datetime              # Auto-set to now(UTC)
    content_hash: str = ""           # Auto-computed MD5
    source_turns: list[str] = []     # IDs of source conversation turns
    links: list[str] = []            # IDs of linked memory entries

Property: is_expired -> bool -- True if expires_at is set and in the past.

Method: touch() -> MemoryEntry -- returns a copy with incremented access_count and refreshed last_accessed.

Diagnostics Models

PipelineDiagnostics

A TypedDict (with total=False) describing the diagnostics dict produced by the pipeline. All keys are optional.

StepDiagnostic

A TypedDict for a single pipeline step.

Streaming Models

Models for tracking LLM streaming responses. These are data containers that pair with context pipeline results.

StreamDelta

A single text delta from a streaming LLM response.

StreamUsage

Token usage from a completed streaming response.

StreamResult

Accumulated result from a completed streaming response.

See Also

• Pipeline Guide -- how models fit into the pipeline flow
• Pipeline API Reference -- ContextPipeline and step factories
• Exceptions Reference -- error hierarchy