Python API reference¶

The stable user-facing Python namespace is the set of names exported from cognoxium. Import public types from the top level:

import cognoxium as cx

frame = cx.CognitionFrame.from_records(records)
profile = cx.profiles.openai("o200k_base")

Modules such as cognoxium.frame and helper functions not exported at the top level are implementation details in 0.x releases. Underscore-prefixed attributes and metadata keys are private.

Where to look¶

Task	Principal API
Build or load context	`CognitionFrame.from_records`, `from_source`, `from_arrow`, `read_jsonl`, `read_ipc`, `read_parquet`
Execute or export	`collect`, `collect_records`, `to_records`, `to_arrow`, `to_pandas`, `to_polars`
Tabular planning	`filter`, `select`, `with_columns`, `join`, `group_by`, `sort`, `limit`
AI context planning	`dedupe`, `rank`, `redact`, `quarantine`, `demote`, `pack`
Expressions	`col`, `lit`, `now`, `Expr`
Schema	`ContextItem`, `Payload`, `SourceRef`, enums
Budget and output	`Boundary`, `Overflow`, `ContextPack`, `PackManifest`, `ItemCost`
Token accounting	`TokenProfile`, `profiles.approximate`, `profiles.openai`, `profiles.huggingface`
Recovery	`BudgetExceeded`, `PolicyViolation`, `SchemaError`, `UnsupportedContent`, `PluginError`
Extension points	`Source`, `Ranker`, `Policy`, `Renderer`, `PythonTokenizer`
Localization	`locale`, `get_locale`, `set_default_locale`, `Diagnostic.render`

Do not instantiate CognitionFrame with its _rows and _plan dataclass fields. Use a from_* classmethod. GroupedCognitionFrame is returned by group_by() and normally is not instantiated directly.

Cognoxium public Python API.

class cognoxium.Boundary(kind, target)¶

Bases: object

Describe where packed context will be consumed.

Boundary values are immutable and are recorded in the manifest as "kind:target". The built-in external-boundary checks require provenance and reject restricted or secret-shaped content. Expiry, quarantine, and supplied policies apply to both boundary kinds.

Parameters:

kind (str) – Boundary classification. Use external() or internal() for supported built-in behavior.
target (str) – Human-readable destination identifier, such as "openai" or "application".

Notes

A boundary records policy intent; it does not perform network communication or call a model.

kind¶

target¶

classmethod external(target)¶

Create a boundary for context leaving the application trust domain.

Parameters:: target – Destination identifier used in manifests and diagnostics.
Returns:: An immutable boundary whose kind is "external".
Return type:: Boundary

Examples

>>> import cognoxium as cx
>>> str(cx.Boundary.external("openai"))
'external:openai'

classmethod internal(target='application')¶

Create a boundary for consumption within the application trust domain.

Internal packing still enforces expiry, quarantine, required-item handling, and custom policies. The external-only provenance, sensitivity, and secret-pattern checks do not run.

Parameters:: target – Destination identifier. The default is "application".
Returns:: An immutable boundary whose kind is "internal".
Return type:: Boundary

exception cognoxium.BudgetExceeded(*, budget, required_tokens, token_profile, item_costs, excluded_items, partial_manifest)¶

Bases: CognoxiumError

Report required context that cannot fit without violating its contract.

Cognoxium raises this diagnostic only after excluding ineligible non-required items and, when requested, attempting eligible text truncation. Applications can inspect structured fields to reduce the required set, increase the budget, or explicitly retry with a listed strategy. Never parse the localized exception message for recovery logic.

Parameters:

budget – Requested token ceiling.
required_tokens – Required-item cost, including profile base and per-item envelope tokens.
token_profile – Profile used for the failed calculation.
item_costs – Required-item costs in input order, including cumulative totals.
excluded_items – Non-required items excluded before or not evaluated because required allocation failed.
partial_manifest – Auditable failure-state manifest. Its total_tokens may exceed its budget.

overflow_tokens¶: required_tokens - budget.

token_profile_id, token_profile_fingerprint: Identity of the accounting rules used for the failure.

exceeded_at_item_id¶: First required item whose cumulative cost crossed the budget, if available.

available_strategies¶: ("error",) plus "truncate_truncatable" when a required item has opted into the flag. Truncation can still fail when eligible text cannot be shortened enough.

partial_manifest¶: Manifest captured at the failure point.

Notes

The inherited code is "CX_BUDGET_001". The diagnostic parameters and attributes are stable machine interfaces; render() provides localized text.

class cognoxium.CognitionFrame(_rows, _plan=())¶

Bases: object

An immutable, lazily evaluated collection of context records.

A frame owns an immutable snapshot of its input rows. Transformation methods return a new frame with an additional plan step; they do not execute the step or modify the original. Execution occurs when a collection, conversion, write, or pack() method is called.

Rows retain the canonical ContextItem schema unless a tabular operation, such as select() or GroupedCognitionFrame.agg(), removes required fields. A frame without both id and payload remains useful through collect_records(), but cannot be collected as context items or passed to AI-specific operations.

Examples

Build a lazy plan and execute it explicitly:

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records(
...     [{"id": "guide", "payload": "Rust setup", "priority": 2}]
... )
>>> planned = frame.filter(cx.col("priority") > 0).limit(1)
>>> len(frame.collect())
1
>>> [item.id for item in planned.collect()]
['guide']

classmethod from_records(records)¶

Create a frame from Python mappings or typed context items.

The iterable is consumed immediately and normalized into an immutable tuple. Mapping values are validated through ContextItem.from_record(); omitted fields receive the model defaults, including an automatically generated id and untrusted trust.

Parameters:: records – Context items or mappings accepted by ContextItem.from_record().
Returns:: A frame with no pending transformation steps.
Return type:: CognitionFrame
Raises:: SchemaError – If a record contains an invalid payload, enum value, timestamp, hash, or field value.

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records(
...     [{"id": "answer", "payload": {"value": 42}, "sources": ["app://result"]}]
... )
>>> frame.collect()[0].id
'answer'

classmethod from_source(source)¶

Load records from a source plugin and create a frame.

source.load() is called immediately. The returned iterable is then handled exactly as in from_records(); source loading itself is not part of the lazy query plan.

Parameters:

source – An object implementing the Cognoxium Source protocol, including load().

Returns:

A frame containing the source’s normalized records.

Return type:

CognitionFrame

Raises:

SchemaError – If any loaded record is invalid.
AttributeError – If source does not provide load().

classmethod read_jsonl(path)¶

Read UTF-8 JSON Lines records into a frame.

Empty lines are ignored. Each non-empty line must contain one JSON object compatible with from_records().

Parameters:

path – Path to the JSONL file.

Returns:

A frame containing the normalized records.

Return type:

CognitionFrame

Raises:

OSError – If the file cannot be opened or read.
json.JSONDecodeError – If a non-empty line is not valid JSON.
SchemaError – If a decoded record is not a valid context item.

Examples

Read a file created by write_jsonl():

frame = CognitionFrame.read_jsonl("context.jsonl")

classmethod read_ipc(path)¶

Read a Cognoxium Arrow IPC file.

Parameters:

path – Path to an Arrow IPC file written with write_ipc().

Returns:

A frame reconstructed from the canonical Arrow schema.

Return type:

CognitionFrame

Raises:

ImportError – If PyArrow is unavailable. Install Cognoxium with the arrow extra.
OSError – If the file cannot be opened or mapped.
SchemaError – If reconstructed rows violate the context schema.

classmethod read_parquet(path)¶

Read Cognoxium records from Parquet.

Parameters:

path – Path to a Parquet file containing the canonical Cognoxium schema.

Returns:

A frame reconstructed from the file.

Return type:

CognitionFrame

Raises:

ImportError – If PyArrow is unavailable. Install Cognoxium with the arrow extra.
OSError – If the file cannot be read.
SchemaError – If reconstructed rows violate the context schema.

classmethod from_arrow(value)¶

Create a frame from an Arrow-compatible table value.

Objects providing to_pylist() are accepted directly. Other inputs are passed to pyarrow.table. Canonical Cognoxium payload structs use the numeric kind discriminator and are restored to Payload values before validation.

Parameters:

value – A PyArrow table, record batch, or value accepted by pyarrow.table.

Returns:

A frame containing normalized context records.

Return type:

CognitionFrame

Raises:

ImportError – If conversion requires PyArrow and it is unavailable.
SchemaError – If converted rows violate the context schema.
ValueError – If a canonical payload contains an unknown kind discriminator.

explain()¶

Return a compact description of the pending logical plan.

Returns:: A stable, human-readable sequence beginning with Scan[context records].
Return type:: str

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records([]).limit(2)
>>> frame.explain()
'Scan[context records] -> Limit[2]'

collect_records()¶

Execute the plan and return independent row dictionaries.

Unlike collect(), this method also supports tabular results that no longer contain the complete context schema.

Returns:: Materialized rows in plan order. Mutating the returned list or dictionaries does not mutate the frame.
Return type:: list[dict[str, Any]]

collect()¶

Execute the plan and materialize typed context items.

Returns:: Validated items in plan order.
Return type:: list[ContextItem]
Raises:: SchemaError – If a tabular operation removed id or payload, or produced an invalid item.

to_records()¶

Execute the plan and return JSON-compatible records where possible.

Canonical context rows use ContextItem.to_record(). Non-context tabular rows are converted recursively so that payloads, timestamps, enums, and plugin objects with to_dict() have portable representations.

Returns:: Materialized records detached from the frame.
Return type:: list[dict[str, Any]]

write_jsonl(path)¶

Execute the plan and write deterministic UTF-8 JSON Lines.

Parameters:

path – Destination file. An existing file is replaced.

Raises:

OSError – If the destination cannot be opened or written.
TypeError – If a value cannot be serialized as JSON.

write_ipc(path)¶

Execute the plan and write the canonical Arrow schema to an IPC file.

Parameters:

path – Destination file. An existing file is replaced.

Raises:

ImportError – If PyArrow is unavailable. Install Cognoxium with the arrow extra.
SchemaError – If the materialized rows are not complete context items.
OSError – If the destination cannot be written.

write_parquet(path)¶

Execute the plan and write the canonical Arrow schema to Parquet.

Parameters:

path – Destination file. Existing-file behavior follows pyarrow.parquet.write_table.

Raises:

ImportError – If PyArrow is unavailable. Install Cognoxium with the arrow extra.
SchemaError – If the materialized rows are not complete context items.
OSError – If the destination cannot be written.

to_arrow()¶

Execute the plan and return a canonical PyArrow table.

Text, JSON, binary, and reference payloads are represented by the stable payload struct. The JSON child field carries Arrow’s arrow.json extension metadata.

Returns:

A table with Cognoxium schema version 1 metadata.

Return type:

pyarrow.Table

Raises:

ImportError – If PyArrow is unavailable. Install Cognoxium with the arrow extra.
SchemaError – If the materialized rows are not complete context items.

to_pandas()¶

Execute the plan and return a pandas DataFrame.

Returns:: A DataFrame built from to_records().
Return type:: pandas.DataFrame
Raises:: ImportError – If pandas is unavailable. Install Cognoxium with the pandas extra.

to_polars()¶

Execute the plan and return a Polars DataFrame.

Returns:: A DataFrame built from to_records().
Return type:: polars.DataFrame
Raises:: ImportError – If Polars is unavailable. Install Cognoxium with the polars extra.

filter(predicate)¶

Add a lazy row filter without changing the current frame.

Parameters:: predicate – A Cognoxium expression or callable receiving one row mapping. Rows for which the result is truthy are retained. Callables should treat the mapping as read-only.
Returns:: A new frame containing the pending filter step.
Return type:: CognitionFrame

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records([{"id": "a", "payload": "keep"}])
>>> recent = frame.filter(cx.col("priority") >= 0)
>>> [item.id for item in recent.collect()]
['a']

select(*columns)¶

Add a lazy projection of named columns.

Unknown names produce None values. Selecting without both id and payload creates a tabular-only frame: use collect_records(), to_records(), to_pandas(), or to_polars(), rather than collect() or pack().

Parameters:: *columns – One or more column names, in output order.
Returns:: A new frame containing the pending projection.
Return type:: CognitionFrame
Raises:: ValueError – If no columns are provided.

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records(
...     [{"id": "guide", "payload": "Rust", "priority": 2}]
... )
>>> frame.select("id", "priority").collect_records()
[{'id': 'guide', 'priority': 2.0}]

with_columns(**values)¶

Add or replace columns with lazy expressions, callables, or constants.

Assignments are evaluated in keyword order for each row. Consequently, a later expression can read a column created earlier in the same call. The input frame and its rows remain unchanged.

Parameters:: **values – Mapping from output names to Expr objects, row callables, or constant values.
Returns:: A new frame containing the pending column assignments.
Return type:: CognitionFrame

Notes

Replacing canonical fields can make rows invalid as ContextItem objects. Invalid results can still be inspected with collect_records().

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records(
...     [{"id": "guide", "payload": "Rust", "priority": 2}]
... )
>>> frame.with_columns(selected=cx.col("priority") > 1).select(
...     "id", "selected"
... ).collect_records()
[{'id': 'guide', 'selected': True}]

sort(by, *, descending=False)¶

Add a lazy stable sort by one column.

Parameters:

by – Column name used as the sort key. A missing column is treated as None.
descending – Reverse the ascending result when True.

Returns:

A new frame containing the pending sort.

Return type:

CognitionFrame

Raises:

TypeError – If non-null values in the selected column cannot be ordered together.

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records(
...     [{"id": "guide", "payload": "Rust", "priority": 2}]
... )
>>> [item.id for item in frame.sort("priority", descending=True).collect()]
['guide']

limit(count)¶

Add a lazy row-count limit.

Parameters:: count – Maximum number of rows to retain. Zero produces an empty result.
Returns:: A new frame containing the pending limit.
Return type:: CognitionFrame
Raises:: ValueError – If count is negative.

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records([{"id": "guide", "payload": "Rust"}])
>>> len(frame.limit(1).collect())
1

join(other, *, on, how='inner', suffix='_right')¶

Add a lazy inner or left equijoin with another frame.

The right frame is materialized when this frame is executed. The join key appears once. A right-side name that collides with a left-side name is renamed by appending suffix. Duplicate keys may produce multiple output rows.

Parameters:

other – Right-hand frame.
on – Column compared with equality on both sides. Missing values are represented by None and therefore match other missing values.
how – Either "inner" or "left".
suffix – Suffix for colliding right-side columns.

Returns:

A new frame containing the pending join.

Return type:

CognitionFrame

Raises:

ValueError – If how is not "inner" or "left".

Notes

Join output is packable only when its final id and payload values still form a valid context item.

group_by(*columns)¶

Create a deterministic grouping over materialized row values.

Parameters:: *columns – One or more group-key columns.
Returns:: A grouping descriptor awaiting GroupedCognitionFrame.agg().
Return type:: GroupedCognitionFrame
Raises:: ValueError – If no group-key columns are provided.

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records([{"id": "guide", "payload": "Rust"}])
>>> rows = frame.group_by("role").agg(item_count=("count", "*")).collect_records()
>>> rows[0]["item_count"]
1

dedupe()¶

Add deterministic exact-content deduplication.

Rows are grouped by their canonical content_hash. For each duplicate group, the oldest created_at value wins; an ID lexical comparison breaks ties. Sources and lineage are merged, while trust, sensitivity, retention, expiry, priority, truncatability, and metadata conflicts follow Cognoxium’s conservative merge rules. The decision is recorded in _cognoxium_dedupe metadata for the pack manifest.

Returns:: A new frame containing the pending exact-deduplication step.
Return type:: CognitionFrame
Raises:: SchemaError – If a row is not a complete, valid context item.

Notes

This method does not perform semantic or approximate similarity matching.

rank(query, ranker=None)¶

Add relevance scores to item metadata without reordering rows.

The built-in ranker computes lexical query-term overlap for text and JSON payloads. Supplying a ranker calls ranker.score_batch(query, items) once at execution. Scores are stored as floats in metadata["_rank_score"] and influence later budget selection.

Parameters:

query – Search text passed to the ranker.
ranker – Optional object implementing the Cognoxium Ranker protocol. When omitted, the deterministic lexical-v1 ranker is used.

Returns:

A new frame containing the pending ranking step.

Return type:

CognitionFrame

Raises:

SchemaError – If a row is not a complete, valid context item.
PluginError – If the plugin returns the wrong number of scores or any non-numeric score.

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records([{"id": "guide", "payload": "Rust"}])
>>> ranked = frame.rank("Rust").collect()
>>> ranked[0].metadata["_rank_score"]
1.0

redact(pattern, replacement='[REDACTED]')¶

Add regex replacement for text payloads.

Non-text payloads pass through unchanged. When a replacement occurs, Cognoxium recomputes the content hash, records the item ID in parent_ids, and stores the replacement count in metadata["redaction_count"].

Parameters:

pattern – A string pattern compiled with re.compile(), or a compiled regular expression.
replacement – Replacement template passed to re.Pattern.subn().

Returns:

A new frame containing the pending redaction step.

Return type:

CognitionFrame

Raises:

re.error – If a string pattern is invalid.
SchemaError – If a row is not a complete, valid context item.

Notes

Redaction is an explicit text transformation, not a complete secret-detection guarantee.

quarantine(predicate)¶

Mark matching items as quarantined without dropping them.

Quarantined items remain inspectable in the frame. Packing excludes them; if a quarantined item is required, packing raises PolicyViolation rather than silently dropping it.

Parameters:: predicate – A Cognoxium expression or row callable selecting items to quarantine.
Returns:: A new frame containing the pending trust update.
Return type:: CognitionFrame
Raises:: SchemaError – If a selected row is not a complete, valid context item.

demote(predicate, *, to=Retention.PREFERRED)¶

Change the retention class of matching items.

Parameters:

predicate – A Cognoxium expression or row callable selecting items to update.
to – Target retention enum or one of "required", "preferred", and "optional". The default is preferred.

Returns:

A new frame containing the pending retention update.

Return type:

CognitionFrame

Raises:

ValueError – If a string target is not a supported retention value.
SchemaError – If a selected row is not a complete, valid context item.

Notes

Despite the method name, any retention class may be supplied. In particular, setting to="required" promotes matching items and gives them the no-drop packing contract.

pack(*, budget, token_profile=None, boundary=None, overflow=None, policies=())¶

Execute the plan and select a policy-compliant context pack.

Required items are never silently removed. Preferred candidates are considered before optional ones, then rank score, priority, token density, and ID determine selection. External boundaries exclude expired, quarantined, provenance-free, restricted, and secret-pattern-matching optional content. A violation affecting a required item raises an error instead.

Parameters:

budget – Positive maximum token count, including the token profile’s message envelope cost.
token_profile – Tokenizer and envelope accounting profile. Defaults to the estimated approximate profile; choose an exact profile when enforcing a provider-specific limit.
boundary – Trust boundary for policy evaluation. Defaults to Boundary.internal().
overflow – Required-content overflow behavior. Defaults to Overflow.error(). Explicit Overflow.truncate_truncatable() may shorten only truncatable text items.
policies – Additional policy plugins evaluated after the built-in boundary checks.

Returns:

Immutable selected items and an auditable packing manifest.

Return type:

ContextPack

Raises:

ValueError – If budget is not positive or an option is invalid.
SchemaError – If a row is not a complete, valid context item.
BudgetExceeded – If required context cannot fit under the selected overflow strategy.
PolicyViolation – If policy rejects a required item.
UnsupportedContent – If a required binary payload cannot be rendered for the boundary target.
PluginError – If a tokenizer or policy plugin violates its protocol.

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records(
...     [{"id": "guide", "payload": "Rust", "sources": ["app://guide"]}]
... )
>>> pack = frame.pack(budget=128, token_profile=cx.profiles.approximate())
>>> pack.manifest.total_tokens <= pack.manifest.budget
True

exception cognoxium.CognoxiumError(diagnostic)¶

Bases: Exception

Base exception carrying a machine-readable diagnostic.

render(language=None)¶

class cognoxium.ContextItem(id, payload, mime_type='text/plain', role=Role.USER, sources=(), content_hash='', trust=Trust.UNTRUSTED, sensitivity=Sensitivity.PUBLIC, retention=Retention.OPTIONAL, priority=0.0, created_at=<factory>, expires_at=None, parent_ids=(), merged_from_ids=(), metadata=<factory>, truncatable=False, min_tokens=0)¶

Bases: object

One typed unit of model context.

id¶

payload¶

mime_type¶

role¶

sources¶

content_hash¶

trust¶

sensitivity¶

retention¶

priority¶

created_at¶

expires_at¶

parent_ids¶

merged_from_ids¶

metadata¶

truncatable¶

min_tokens¶

classmethod from_record(record)¶

evolve(**changes)¶

to_record()¶

class cognoxium.ContextPack(items, manifest)¶

Bases: object

Provider-neutral, immutable context selected for one model call.

Parameters:

items (tuple[cognoxium.models.ContextItem, ...]) – Selected context items in their original input order.
manifest (cognoxium.packing.PackManifest) – Auditable token, boundary, exclusion, deduplication, and truncation decision.

Notes

Conversion methods only build Python values. They do not make network requests, hold API keys, or validate that a provider SDK accepts a particular model’s roles. Binary payloads are stored by Cognoxium but are not silently converted by the v1 text renderers.

items¶

manifest¶

to_text()¶

Render items as role-prefixed plain text.

Each item becomes "[role] payload" and items are separated by one blank line. JSON is emitted in canonical serialized form and references emit their URI.

Returns:: The rendered context, or an empty string for an empty pack.
Return type:: str
Raises:: UnsupportedContent – If any selected item has a binary payload.

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records([{"id": "q", "payload": "Why?"}])
>>> frame.pack(budget=20).to_text()
'[user] Why?'

to_openai()¶

Convert the pack to explicit OpenAI-style text input messages.

System, developer, user, and assistant roles are preserved. Tool items become user messages prefixed with "[tool result]\n". Every message contains one {"type": "input_text", "text": ...} content part. JSON and references use their text representations.

Returns:: Newly allocated message dictionaries in item order.
Return type:: list[dict[str, Any]]
Raises:: UnsupportedContent – If any selected item has a binary payload.

Notes

This method performs a provider mapping only. Callers remain responsible for selecting a compatible API operation and model.

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records(
...     [{"id": "q", "payload": "Why?", "role": "user"}]
... )
>>> frame.pack(budget=20).to_openai()
[{'role': 'user', 'content': [{'type': 'input_text', 'text': 'Why?'}]}]

to_anthropic()¶

Convert the pack to separate Anthropic-style system text and messages.

System and developer payloads are concatenated into the top-level system string with blank lines. Assistant items keep the assistant role. User and tool items use the user role, with tool payloads prefixed by "[tool result]\n". Message content contains one {"type": "text", "text": ...} part.

Returns:: {"system": str, "messages": list} in original item order within each section.
Return type:: dict[str, Any]
Raises:: UnsupportedContent – If any selected item has a binary payload.

Notes

Adjacent messages are not merged or alternated. Callers must satisfy any additional SDK, API, or model-specific conversation requirements.

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records(
...     [
...         {"id": "s", "payload": "Be concise.", "role": "system"},
...         {"id": "q", "payload": "Why?", "role": "user"},
...     ]
... )
>>> result = frame.pack(budget=40).to_anthropic()
>>> result["system"]
'Be concise.'
>>> result["messages"][0]
{'role': 'user', 'content': [{'type': 'text', 'text': 'Why?'}]}

render(renderer)¶

Delegate conversion to a renderer plugin.

Parameters:: renderer – An object implementing render(pack). Cognoxium passes this exact immutable pack.
Returns:: The plugin’s result without additional conversion or validation.
Return type:: Any
Raises:: AttributeError – If the object does not provide render.

Notes

Exceptions raised by the renderer propagate unchanged.

class cognoxium.Diagnostic(code, help_code, parameters=<factory>, captured_locale=<factory>)¶

Bases: object

A stable diagnostic code with localizable presentation.

code¶

help_code¶

parameters¶

captured_locale¶

render(language=None)¶

to_dict()¶

class cognoxium.Expr(evaluator, description)¶

Bases: object

A composable expression evaluated against a context record.

evaluator¶

description¶

evaluate(row)¶

is_null()¶

is_in(values)¶

class cognoxium.GroupedCognitionFrame(frame, columns)¶

Bases: object

A deterministic grouping awaiting one or more aggregations.

Instances are created by CognitionFrame.group_by(). Calling agg() executes the source frame and returns a materialized CognitionFrame of ordinary tabular rows.

The frame attribute stores the source plan and columns stores the group keys in output order.

frame¶

columns¶

agg(**aggregations)¶

Aggregate each group into one tabular row.

Each keyword names an output column. Its value is either (function, source_column) or a function name, in which case the output name is also used as the source column. Supported functions are count, sum, min, max, and list. The special source "*" makes count include every row; other counts ignore null values.

Parameters:

**aggregations – Output names mapped to aggregation specifications.

Returns:

A materialized frame containing one row per group, ordered by first group occurrence.

Return type:

CognitionFrame

Raises:

ValueError – If an aggregation function is unsupported, or min or max receives no non-null values.
TypeError – If a group key is unhashable or values do not support the requested aggregation.

Notes

Aggregated rows normally omit id and payload and are therefore intended for CognitionFrame.collect_records(), pandas, or Polars inspection. Add valid canonical fields before calling CognitionFrame.collect() or CognitionFrame.pack().

Examples

>>> import cognoxium as cx
>>> frame = cx.CognitionFrame.from_records(
...     [
...         {"id": "a", "payload": "one", "role": "user"},
...         {"id": "b", "payload": "two", "role": "user"},
...     ]
... )
>>> frame.group_by("role").agg(items=("count", "*")).collect_records()
[{'role': <Role.USER: 'user'>, 'items': 2}]

class cognoxium.ItemCost(id, tokens, cumulative_tokens, retention, priority, truncatable, min_tokens)¶

Bases: object

Per-required-item token accounting attached to BudgetExceeded.

Parameters:

id (str) – Context item identifier.
tokens (int) – Item token cost, including the profile’s per-item envelope cost.
cumulative_tokens (int) – Profile base tokens plus required-item costs through this item in input order.
retention (str) – Serialized retention value, normally "required".
priority (float) – Priority used by packing and truncation decisions.
truncatable (bool) – Whether the item opted into truncation.
min_tokens (int) – Minimum content-token floor requested by the item.

id¶

tokens¶

cumulative_tokens¶

retention¶

priority¶

truncatable¶

min_tokens¶

to_dict()¶

Return a JSON-compatible snapshot of the accounting fields.

Returns:: A newly allocated dictionary suitable for diagnostics or logging.
Return type:: dict[str, Any]

class cognoxium.Overflow(strategy='error', side='tail', marker='\n…[truncated]')¶

Bases: object

Configure recovery when required context exceeds the token budget.

Parameters:

strategy (str) – "error" or "truncate_truncatable". Prefer the class methods to construct a supported strategy.
side (str) – For truncation, "tail" keeps the beginning and removes the end; "head" keeps the end and removes the beginning.
marker (str) – Text appended or prepended at the removed side. Its token cost counts against the budget.

Notes

Truncation is opt-in and applies only to required text items marked truncatable=True. JSON, reference, binary, and non-truncatable items are never partially shortened. Required items are never automatically dropped or demoted.

strategy¶

side¶

marker¶

classmethod error()¶

Return the default strategy that raises BudgetExceeded.

Returns:: An immutable error strategy.
Return type:: Overflow

classmethod truncate_truncatable(*, side='tail', marker='\n…[truncated]')¶

Allow eligible required text items to be shortened before failing.

Parameters:

side – "tail" removes trailing text; "head" removes leading text.
marker – Text inserted at the removed side and included in token accounting.

Returns:

An immutable truncation strategy.

Return type:

Overflow

Raises:

ValueError – If side is not "head" or "tail".

Examples

>>> import cognoxium as cx
>>> strategy = cx.Overflow.truncate_truncatable(side="tail")
>>> strategy.strategy
'truncate_truncatable'

class cognoxium.PackManifest(schema_version, library_version, budget, total_tokens, token_profile_id, token_profile_fingerprint, estimated, boundary, selected=(), excluded=(), dedupe_groups=(), truncations=())¶

Bases: object

Immutable, auditable record of a packing decision.

Parameters:

schema_version (str) – Version of the manifest serialization contract.
library_version (str) – Cognoxium version that produced the decision.
budget (int) – Requested token ceiling.
total_tokens (int) – Selected token cost on success. In a partial overflow manifest, this is the required cost that could not fit and may exceed budget.
token_profile_id (str) – Stable profile identifier.
token_profile_fingerprint (str) – Fingerprint covering tokenizer and envelope settings.
estimated (bool) – Whether token accounting is approximate. Consumers requiring an exact provider limit should reject estimated manifests or choose a suitable exact profile.
boundary (str) – Serialized kind:target trust boundary.
selected (tuple[collections.abc.Mapping[str, Any], ...]) – Selected item records containing IDs, token costs, retention, and content hashes.
excluded (tuple[collections.abc.Mapping[str, Any], ...]) – Excluded item records with machine-readable reasons and available decision details.
dedupe_groups (tuple[collections.abc.Mapping[str, Any], ...]) – Exact-deduplication lineage and merge decisions propagated from frame metadata.
truncations (tuple[collections.abc.Mapping[str, Any], ...]) – Original and replacement hashes, token changes, and truncation sides.

Notes

The manifest records a decision; it is not a cryptographic signature or proof that model provider envelope rules have remained unchanged.

schema_version¶

library_version¶

budget¶

total_tokens¶

token_profile_id¶

token_profile_fingerprint¶

estimated¶

boundary¶

selected¶

excluded¶

dedupe_groups¶

truncations¶

to_dict()¶

Return a serialization-friendly copy of the complete manifest.

Returns:: A new top-level dictionary with tuple collections converted to lists of dictionaries.
Return type:: dict[str, Any]

class cognoxium.Payload(kind, value, uri=None, digest=None, size=None)¶

Bases: object

A future-compatible context payload.

kind¶

value¶

uri¶

digest¶

size¶

classmethod text(value)¶

classmethod json(value)¶

classmethod binary(value)¶

classmethod reference(uri, *, digest=None, size=None)¶

classmethod from_value(value)¶

canonical_bytes(mime_type)¶

text_value()¶

to_dict()¶

class cognoxium.PayloadKind(*values)¶

Bases: str, Enum

Discriminator for the stable multimodal payload structure.

TEXT = 'text'¶

JSON = 'json'¶

BINARY = 'binary'¶

REFERENCE = 'reference'¶

exception cognoxium.PerformanceWarning¶

Bases: UserWarning

Warns about an intentionally slower but compatible execution path.

exception cognoxium.PluginError(*, plugin, reason)¶

Bases: CognoxiumError

Raised for invalid plugin behavior.

class cognoxium.Policy(*args, **kwargs)¶

Bases: Protocol

Returns a reason code when an item cannot cross a boundary.

id¶

evaluate(item, boundary)¶

exception cognoxium.PolicyViolation(*, item_id, rule_id, boundary, reason_code)¶

Bases: CognoxiumError

Raised when a required item cannot cross a trust boundary.

class cognoxium.PythonTokenizer(*args, **kwargs)¶

Bases: Protocol

Fallback tokenizer contract. Implementations must be deterministic.

fingerprint¶

thread_safe¶

count_batch(values)¶

class cognoxium.Ranker(*args, **kwargs)¶

Bases: Protocol

Scores a batch of context items for a query.

fingerprint¶

score_batch(query, items)¶

class cognoxium.Renderer(*args, **kwargs)¶

Bases: Protocol

Converts a provider-neutral pack into an SDK-compatible structure.

id¶

render(pack)¶

class cognoxium.Retention(*values)¶

Bases: str, Enum

Packing contract ordered from optional to required.

OPTIONAL = 'optional'¶

PREFERRED = 'preferred'¶

REQUIRED = 'required'¶

class cognoxium.Role(*values)¶

Bases: str, Enum

Provider-neutral conversational role.

SYSTEM = 'system'¶

DEVELOPER = 'developer'¶

USER = 'user'¶

ASSISTANT = 'assistant'¶

TOOL = 'tool'¶

exception cognoxium.SchemaError(diagnostic)¶

Bases: CognoxiumError

Raised when a context record violates the public schema.

class cognoxium.Sensitivity(*values)¶

Bases: str, Enum

Information sensitivity ordered from public to restricted.

PUBLIC = 'public'¶

INTERNAL = 'internal'¶

CONFIDENTIAL = 'confidential'¶

RESTRICTED = 'restricted'¶

class cognoxium.Source(*args, **kwargs)¶

Bases: Protocol

Loads context records without requiring a framework integration.

id¶

load()¶

class cognoxium.SourceRef(uri, kind='external', name=None, digest=None)¶

Bases: object

A stable provenance reference.

uri¶

kind¶

name¶

digest¶

classmethod from_value(value)¶

to_dict()¶

class cognoxium.TokenProfile(id, encoding='approximate', envelope_tokens_per_item=0, base_tokens=0, estimated=True, python_tokenizer=None, tokenizer_json=None)¶

Bases: object

Tokenizer plus provider-envelope accounting rules.

id¶

encoding¶

envelope_tokens_per_item¶

base_tokens¶

estimated¶

python_tokenizer¶

tokenizer_json¶

property fingerprint¶

count_batch(values, *, content_hashes=None)¶

item_costs(values, *, content_hashes=None)¶

truncate(value, max_content_tokens, side='tail')¶: Truncate at a Unicode code-point boundary using monotonic token counts.

class cognoxium.Trust(*values)¶

Bases: str, Enum

Application assertion about content trustworthiness.

TRUSTED = 'trusted'¶

UNTRUSTED = 'untrusted'¶

QUARANTINED = 'quarantined'¶

exception cognoxium.UnsupportedContent(*, kind, renderer)¶

Bases: CognoxiumError

Raised when a renderer cannot represent a payload kind.

cognoxium.col(name)¶: Create a column expression.

cognoxium.get_locale()¶: Return the active task-local locale.

cognoxium.lit(value)¶: Create a literal expression.

cognoxium.locale(value)¶: Temporarily select a locale without leaking across async tasks.

cognoxium.now()¶: Capture the current UTC time once when the expression is created.

cognoxium.set_default_locale(value)¶: Set the process default locale used outside a locale() context.

Token accounting profiles.

cognoxium.profiles.approximate()¶: Return a deterministic UTF-8 byte estimate profile.

cognoxium.profiles.openai(encoding='o200k_base')¶: Return an OpenAI-compatible content tokenizer with conservative envelope cost.

cognoxium.profiles.huggingface(tokenizer, *, profile_id='huggingface:custom')¶: Create a native JSON profile or a fallback batched Python profile.

Rust API¶

The public Rust API is generated with cargo doc --workspace --no-deps and published under rust/latest. See Rust API reference for local generation, release-specific auditing, and scope.