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:

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.

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¶

Type alias for Literal["user", "assistant", "system", "tool"].

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