The AI engineering community is currently obsessed with expanding LLM context windows to 1M+ tokens. Meanwhile, teams are burning massive amounts of compute and cloud egress costs because they are orchestrating multi-agent systems using JSON.
We are feeding state-of-the-art intelligence through a 20-year-old, heavily bloated web format.
ULMEN is a drop-in Python/Rust serialization engine that treats the LLM context window and network IPC as strict hardware constraints. By natively incorporating exact token-counting, string pooling, and semantic validation at the C/Rust boundary, ULMEN delivers Protobuf-level density without requiring pre-compiled schemas.
- Benchmarks
- At a Glance
- Surfaces
- Installation
- Quick Start
- API Reference
- Wire Format Constants
- Utilities
- Architecture
- Running Tests
- Format Specification
- Versioning
Benchmarks run on production-grade constraints (NVIDIA Tesla T4, 16GB VRAM):
-
44% LLM Token Reduction: Eliminates syntax bloat, saving approximately $59,000 per 10 million agent loops (vs. GPT-4o input costs).
-
3x Faster Reads: Deserializes heavily nested payloads natively faster than the C-optimized orjson and standard json.
-
4.1x Smaller IPC Footprint: The pooled binary format drastically reduces microservice network egress and Redis cache saturation.
-
The Semantic Firewall: Unlike generic formats that silently pass broken traces, the ULMEN-AGENT protocol automatically rejects orphaned tools, backwards steps, and invalid enums before they trigger LLM hallucinations.
ULMEN exposes four surfaces over a single data model:
Columnar binary format. Smallest on wire. Designed for storage and IPC. Supports delta encoding, bitpacking, RLE, string pooling, and zlib.
Line-oriented, diff-friendly, human-readable. Compatible with standard text tools. Uses the same pool and strategy system as binary.
LLM-native CSV surface. Every payload is self-describing via a typed header line. Language models can read and generate ULMEN without special training or prompt engineering.
Zero-materialisation streaming encode surface. Feed records one at a time
or in batches, then flush to an iterator of bytes chunks. The Rust backend
is selected automatically. Wire format is identical to batch binary encode —
every chunk is independently decodable. For truly unbounded streams use
stream_encode_windowed which encodes fixed-size windows into independent
sub-payloads, each decodable standalone.
Structured protocol for agentic AI communication. Typed record schemas for messages, tool calls, results, plans, observations, errors, memory, RAG chunks, hypotheses, and chain-of-thought steps.
Extended capabilities:
- Extended header fields: payload_id, parent_payload_id, agent_id, session_id, schema_version, context_window, context_used, meta_fields
- Meta fields appended to every row: parent_id, from_agent, to_agent, priority
- Context compression: completed_sequences, keep_types, sliding_window
- Priority-based retention: MUST_KEEP, KEEP_IF_ROOM, COMPRESSIBLE
- Unlimited context via chunk_payload, merge_chunks, build_summary_chain
- LLM output auto-repair via parse_llm_output
- Exact BPE token counting via count_tokens_exact (cl100k_base)
- Multi-agent routing via AgentRouter
- Cross-payload thread tracking via ThreadRegistry
- Append-only audit trail via ReplayLog
- Programmatic system prompt generation via generate_system_prompt
- ULMEN bridge: convert_agent_to_ulmen, convert_ulmen_to_agent
- Structured validation errors via ValidationError
- Context budget enforcement via ContextBudgetExceededError
- Streaming decode via decode_agent_stream
- Subgraph extraction by thread, step range, type
- Memory deduplication via dedup_mem, get_latest_mem
- MessagePack compatibility via encode_msgpack, decode_msgpack
git clone https://github.com/makroumi/ulmen
cd ulmen
pip install maturin
maturin develop --releasepip install -e .The library detects automatically whether the Rust extension is available and falls back to the pure Python implementation silently.
from ulmen import UlmenDict, UlmenDictRust, encode_ulmen_llm, decode_ulmen_llm
records = [
{"id": 1, "name": "Alice", "city": "London", "score": 98.5, "active": True},
{"id": 2, "name": "Bob", "city": "London", "score": 91.0, "active": False},
{"id": 3, "name": "Carol", "city": "Paris", "score": 87.3, "active": True},
]
# Binary (smallest)
ld = UlmenDict(records)
binary = ld.encode_binary_pooled()
zlib_ = ld.encode_binary_zlib()
# Text (human-readable)
text = ld.encode_text()
# ULMEN (LLM-native)
ulmen = encode_ulmen_llm(records)
back = decode_ulmen_llm(ulmen)
# Rust acceleration (drop-in, byte-identical)
ld_rust = UlmenDictRust(records)
binary = ld_rust.encode_binary_pooled()
text = ld_rust.encode_text()
ulmen = ld_rust.encode_ulmen_llm()from ulmen import (
encode_agent_payload,
decode_agent_payload,
decode_agent_payload_full,
validate_agent_payload,
compress_context,
chunk_payload,
merge_chunks,
build_summary_chain,
parse_llm_output,
count_tokens_exact,
AgentRouter,
ThreadRegistry,
ReplayLog,
generate_system_prompt,
convert_agent_to_ulmen,
convert_ulmen_to_agent,
dedup_mem,
get_latest_mem,
estimate_context_usage,
extract_subgraph,
extract_subgraph_payload,
make_validation_error,
AgentHeader,
ValidationError,
ContextBudgetExceededError,
)
records = [
{
"type": "msg", "id": "m1", "thread_id": "t1", "step": 1,
"role": "user", "turn": 1, "content": "Hello", "tokens": 5,
"flagged": False,
},
{
"type": "tool", "id": "tc1", "thread_id": "t1", "step": 2,
"name": "search", "args": '{"q":"ulmen"}', "status": "pending",
},
{
"type": "res", "id": "tc1", "thread_id": "t1", "step": 3,
"name": "search", "data": "ULMEN is fast", "status": "done",
"latency_ms": 42,
},
]
# Encode with extended header fields
payload = encode_agent_payload(
records,
thread_id="t1",
context_window=8000,
payload_id="uuid-abc",
parent_payload_id="uuid-prev",
agent_id="agent-alpha",
session_id="sess-001",
schema_version="1.0.0",
auto_context=True,
auto_payload_id=False,
enforce_budget=False,
)
# Decode (records only)
decoded = decode_agent_payload(payload)
# Decode (records + parsed header)
records_out, header = decode_agent_payload_full(payload)
print(header.payload_id)
print(header.context_used)
# Validate
ok, err = validate_agent_payload(payload)
# Validate with structured error object
ok, err = validate_agent_payload(payload, structured=True)
if not ok:
print(err.message, err.row, err.field, err.suggestion)
# Stream decode one record at a time
from ulmen import decode_agent_stream
for rec in decode_agent_stream(iter(payload.splitlines(keepends=True))):
print(rec["type"])
# Context compression
from ulmen.core._agent import COMPRESS_COMPLETED_SEQUENCES
compressed = compress_context(
records,
strategy=COMPRESS_COMPLETED_SEQUENCES,
preserve_cot=True,
)
# Memory deduplication
clean = dedup_mem(records)
latest = get_latest_mem(records, key="user_pref")
# Context usage estimation
usage = estimate_context_usage(records)
print(usage["tokens"], usage["by_type"])
# Chunking for unlimited context
chunks = chunk_payload(records, token_budget=2000, thread_id="t1", overlap=1)
merged = merge_chunks(chunks)
# Summary chain for unlimited context
chain = build_summary_chain(records, token_budget=2000, thread_id="t1")
# LLM output auto-repair
repaired = parse_llm_output(raw_llm_text)
repaired = parse_llm_output(raw_llm_text, strict=True)
# Exact token counting
n_tokens = count_tokens_exact(payload)
# Subgraph extraction
filtered = extract_subgraph(records, thread_id="t1", step_min=2, types=["tool","res"])
filtered_payload = extract_subgraph_payload(payload, types=["cot"])
# Multi-agent routing
router = AgentRouter()
router.register("planner", "executor", lambda rec: print(rec))
router.dispatch(records)
# Cross-payload thread tracking
registry = ThreadRegistry()
registry.add_payload("pid-1", records)
# Audit trail
log = ReplayLog()
log.append({"event": "encode", "payload_id": "pid-1"})
# System prompt generation
prompt = generate_system_prompt(include_examples=True, include_validation=True)
# ULMEN bridge
ulmen = convert_agent_to_ulmen(payload)
payload2 = convert_ulmen_to_agent(ulmen, thread_id="t1")
# Validation error payload
err_payload = make_validation_error("bad step", thread_id="t1")
# Context budget enforcement
try:
encode_agent_payload(records, context_window=10, enforce_budget=True)
except ContextBudgetExceededError as e:
print(e.overage)Pure Python record container. Zero runtime dependencies.
ld = UlmenDict(records)
ld.encode_text() # str ULMEN text format
ld.encode_binary() # bytes raw binary
ld.encode_binary_pooled() # bytes binary with full strategy selection
ld.encode_binary_zlib(level=6) # bytes binary + zlib, level 0-9
ld.encode_ulmen_llm() # str ULMEN format
ld.decode_text(text) # UlmenDict
ld.decode_binary(data) # UlmenDict
ld.decode_ulmen_llm(text) # UlmenDict
ld.to_json() # str standard JSON (NaN/inf replaced with null)
ld.append(record) # mutate, rebuilds pool, invalidates cache
len(ld) # number of records
ld.pool_size # number of interned strings
ld[0] # direct index accessExtended pool variant. Strategies always enabled.
ldf = UlmenDictFull(records, pool_size_limit=256)
ldf.encode_binary()
ldf.encode_text()
ldf.encode_ulmen_llm()Rust-accelerated drop-in replacements. Byte-identical output.
from ulmen import UlmenDictRust, UlmenDictFullRust, RUST_AVAILABLE
print(RUST_AVAILABLE)
ld = UlmenDictRust(records, optimizations=False, pool_size_limit=64)
ld.encode_text()
ld.encode_binary_pooled()
ld.encode_binary_zlib(level=6)
ld.encode_ulmen_llm()See ulmen.core._streaming for full API.
from ulmen import UlmenStreamEncoder, stream_encode, stream_encode_windowed
# One-shot
for chunk in stream_encode(records, chunk_size=65536):
socket.sendall(chunk)
# Stateful
enc = UlmenStreamEncoder(pool_size_limit=64, chunk_size=65536)
enc.feed(record)
enc.feed_many(records)
for chunk in enc.flush():
sink.write(chunk)
print(enc.rust_backed) # True when Rust extension active
# Unbounded windowed
for chunk in stream_encode_windowed(records, window_size=1000):
decode_binary_records(chunk)
from ulmen import (
encode_ulmen_llm,
decode_ulmen_llm,
encode_binary_records,
decode_binary_records,
encode_text_records,
decode_text_records,
build_pool,
detect_column_strategy,
)from ulmen import (
encode_agent_payload,
decode_agent_payload,
decode_agent_payload_full,
decode_agent_record,
encode_agent_record,
decode_agent_stream,
validate_agent_payload,
make_validation_error,
extract_subgraph,
extract_subgraph_payload,
AgentHeader,
ValidationError,
ContextBudgetExceededError,
)'encode_agent_payload' parameters:
| Parameter | Type | Description |
|---|---|---|
| records | list[dict] | Records to encode |
| thread_id | str or None | Written to header |
| context_window | int or None | Token budget declared in header |
| meta_fields | tuple | Extra fields appended to every row |
| auto_context | bool | Compute context_used automatically |
| enforce_budget | bool | Raise ContextBudgetExceededError if over budget |
| payload_id | str or None | Unique ID for this payload |
| parent_payload_id | str or None | Links to prior payload in chain |
| agent_id | str or None | ID of the producing agent |
| session_id | str or None | Session this payload belongs to |
| schema_version | str or None | Protocol version for negotiation |
| auto_payload_id | bool | Generate a UUID payload_id automatically |
from ulmen import compress_context, dedup_mem, get_latest_mem, estimate_context_usage
from ulmen.core._agent import (
COMPRESS_COMPLETED_SEQUENCES,
COMPRESS_KEEP_TYPES,
COMPRESS_SLIDING_WINDOW,
PRIORITY_MUST_KEEP,
PRIORITY_KEEP_IF_ROOM,
PRIORITY_COMPRESSIBLE,
)
compressed = compress_context(
records,
strategy=COMPRESS_COMPLETED_SEQUENCES,
keep_priority=PRIORITY_KEEP_IF_ROOM,
preserve_cot=True,
)
clean = dedup_mem(records)
latest = get_latest_mem(records, key="pref")
usage = estimate_context_usage(records)Strategies:
- completed_sequences: replace completed tool+res pairs with mem summaries
- keep_types: keep only specified record types
- sliding_window: keep recent records verbatim, summarize older ones
from ulmen import chunk_payload, merge_chunks, build_summary_chain
chunks = chunk_payload(
records,
token_budget=4000,
thread_id="t1",
overlap=2,
parent_payload_id="prev-id",
session_id="sess-1",
)
merged = merge_chunks(chunks)
chain = build_summary_chain(
records,
token_budget=4000,
thread_id="t1",
session_id="sess-1",
)from ulmen import parse_llm_output
repaired = parse_llm_output(raw_text)
repaired = parse_llm_output(raw_text, thread_id="t1", strict=True)Uses cl100k_base BPE (GPT-4 / Claude compatible). Falls back to character estimate when tiktoken is unavailable.
from ulmen import AgentRouter, validate_routing_consistency
router = AgentRouter()
router.register("agent_a", "agent_b", handler_fn)
router.dispatch(records)
router.dispatch_one(record)
ok, err = validate_routing_consistency(records)from ulmen import ThreadRegistry, merge_threads
registry = ThreadRegistry()
registry.add_payload("pid-1", records)
threads = registry.get_threads()
merged = merge_threads([payload1_records, payload2_records])from ulmen import ReplayLog
log = ReplayLog()
log.append({"event": "encode", "ts": 1234})
events = log.all()from ulmen import generate_system_prompt
prompt = generate_system_prompt(
include_examples=True,
include_validation=True,
)from ulmen import convert_agent_to_ulmen, convert_ulmen_to_agent
ulmen = convert_agent_to_ulmen(agent_payload)
payload = convert_ulmen_to_agent(ulmen, thread_id="t1")from ulmen.core._msgpack_compat import encode_msgpack, decode_msgpack
packed = encode_msgpack(records)
unpacked = decode_msgpack(packed)from ulmen import (
MAGIC, # b'LUMB'
VERSION, # bytes([3, 3])
T_STR_TINY, T_STR, T_INT, T_FLOAT, T_BOOL, T_NULL,
T_LIST, T_MAP, T_POOL_DEF, T_POOL_REF, T_MATRIX,
T_DELTA_RAW, T_BITS, T_RLE,
S_RAW, S_DELTA, S_RLE, S_BITS, S_POOL,
AGENT_MAGIC, # "ULMEN-AGENT v1"
AGENT_VERSION, # "1.0.0"
RECORD_TYPES, # frozenset of 10 type tags
FIELD_COUNTS, # dict[type -> total field count per row including common fields]
META_FIELDS, # ("parent_id", "from_agent", "to_agent", "priority")
COMPRESS_COMPLETED_SEQUENCES,
COMPRESS_KEEP_TYPES,
COMPRESS_SLIDING_WINDOW,
PRIORITY_MUST_KEEP, # 1
PRIORITY_KEEP_IF_ROOM, # 2
PRIORITY_COMPRESSIBLE, # 3
)from ulmen import (
estimate_tokens, # rough LLM token count (chars / 4)
deep_size, # recursive memory footprint in bytes
deep_eq, # structural equality handling NaN and inf
fnv1a, fnv1a_str, # FNV-1a 32-bit hash
)ulmen/
├── Cargo.lock
├── Cargo.toml
├── pyproject.toml
├── README.md
├── SPEC.md
├── src/
│ └── lib.rs
├── ulmen/
│ ├── __init__.py
│ ├── core.py
│ └── core/
│ ├── __init__.py
│ ├── _constants.py
│ ├── _primitives.py
│ ├── _strategies.py
│ ├── _text.py
│ ├── _binary.py
│ ├── _ulmen_llm.py
│ ├── _agent.py
│ ├── _api.py
│ ├── _repair.py
│ ├── _replay.py
│ ├── _routing.py
│ ├── _threading.py
│ ├── _tokens.py
│ ├── _msgpack_compat.py
│ └── _streaming.py
├── tests/
│ ├── conftest.py
│ ├── smoke_test_comprehensive.py
│ ├── integration/
│ │ ├── test_edge_cases.py
│ │ ├── test_init_coverage.py
│ │ └── test_rust_layer.py
│ ├── perf/
│ │ ├── test_benchmark.py
│ │ ├── test_size.py
│ │ └── test_speed.py
│ └── unit/
│ ├── test_agent.py
│ ├── test_core_coverage.py
│ ├── test_encoders.py
│ ├── test_ulmendict.py
│ ├── test_ulmen_llm.py
│ ├── test_msgpack_compat.py
│ ├── test_primitives.py
│ ├── test_repair.py
│ ├── test_replay.py
│ ├── test_routing.py
│ ├── test_strategies.py
│ ├── test_streaming.py
│ ├── test_threading.py
│ └── test_tokens.py
└── docs/
├── index.md
├── getting-started/
│ ├── installation.md
│ └── quickstart.md
├── guides/
│ ├── binary-format.md
│ ├── text-format.md
│ ├── ulmen.md
│ └── compression.md
├── reference/
│ ├── api.md
│ ├── constants.md
│ ├── primitives.md
│ └── benchmarks.md
├── agent/
│ ├── overview.md
│ ├── spec.md
│ └── system-prompt.md
└── internals/
├── architecture.md
└── wire-format.md
Design principle: the Python layer is the normative specification. The Rust layer is an optimization producing identical output at higher speed. All encode results are cached after the first call and invalidated on mutation.
pytest tests/ -v
pytest tests/ --cov=ulmen --cov-report=term-missing1,364 tests across unit, integration, performance, and smoke suites. 100% statement coverage across all modules. All tests pass with and without the Rust extension.
See SPEC.md for the complete wire format specification including all tag values, encoding rules, strategy selection logic, and full ULMEN and ULMEN-AGENT protocol details.
ULMEN is licensed under the Business Source License (BSL) 1.1.
I built this engine to solve a massive, expensive bottleneck in AI orchestration, and I want the builders, the indie hackers, and the startups to have it with zero friction.
The TL;DR:
-
Free for the 99%: ULMEN is completely free to use in production for any individual, open-source project, academic researcher, or commercial entity with less than $10,000,000 USD in gross annual revenue.
-
The Corporate Clause: If your legal entity (including affiliates) exceeds $10M USD in revenue, you must purchase a Commercial License.
-The Open Source Guarantee: To ensure this technology ultimately belongs to the community, every specific version of ULMEN automatically converts to the fully open-source Apache 2.0 License exactly four (4) years after its release.
Why BSL? The BSL protects the countless hours of systems architecture required to build and maintain ULMEN. It allows me to give a world-class engine away for free to the developers who need it most, while ensuring that trillion-dollar cloud providers and tech giants cannot strip-mine the project without contributing to its survival.
See the [LICENSE] file for the full legal text.
If your entity (including affiliates) exceeds the $10,000,000 USD gross annual revenue threshold, or if you wish to embed ULMEN into a commercial product that competes with it, you must acquire a Commercial License.
Commercial licenses grant you the right to bypass the BSL restrictions and use ULMEN in high-revenue production environments.
Contact: Reach out to 'elmehdi.makroumi@gmail.com' to request a commercial license agreement.