DEV Community: firefrog

A Simple Look at TokPress: A Compressor That Uses an LLM's Tokenizer

firefrog — Sun, 06 Sep 2026 12:30:01 +0000

If you've ever compressed a bunch of small JSON log lines, you've probably noticed something annoying: gzip often makes them bigger, not smaller. A 200-byte log line has so little content that the compressor spends more bytes on its own bookkeeping than it saves.

TokPress is a tiny weekend project that tries a different trick: instead of compressing raw bytes, it first runs the text through the same tokenizer that OpenAI's models use (o200k_base), then compresses the tokens.

That's the whole idea. An LLM tokenizer has already learned which pieces of text are common — words, punctuation, code patterns. So the compressor gets a much better alphabet to work with for free.

What it actually does

Three simple steps:

Tokenize — turn the input into token ids using o200k_base.
Compress the tokens — run a simple LZ77 pass over the token ids.
Entropy-code — finish with rANS, an efficient entropy coder.

The result is a .tokz file that decompresses back to the exact original bytes — even binary data, since the tokenizer works on bytes, not just text.

import tokpress

compressed = tokpress.compress(payload)
original = tokpress.decompress(compressed)   # byte-exact

The trick that makes it useful: a shared dictionary

The real win comes when you have many records that look the same — log lines, API responses, telemetry. TokPress can train a small dictionary on a sample of your records, then every future record compresses against it:

tokpress train-dict mydict.tokdict samples.jsonl
tokpress compress new_record.json --dict mydict.tokdict -o new_record.tokz

With a trained dictionary, structured-log records went from a ratio of 0.800 to 0.2565 — about 3× smaller. And that's on records the dictionary had never seen.

There's an even simpler mode that needs no training at all: compress many records together as one stream, and the model learns as it goes:

packed = tokpress.compress_many(records)
records = tokpress.decompress_many(packed)

On 150 schema-similar JSON records, per-record compression summed to ratio 1.19 — the data literally grew. As one stream, it hit 0.0875. Same bytes, one header instead of 150.

Where it stands (honestly)

On prose it beats gzip and ties/beats zstd on some files (a 152KB prose file: 0.298 vs zstd's 0.324).
It still loses to zstd's trained dictionary by about 1.3×. zstd has had years of polish on dictionary training; this is a weekend project.
It's pure Python, so it's slow to compress. Decompression is fast (thousands of records/sec); compression is not.
Tokenization isn't magic. The tokenizer just reshapes the data; the real savings come from the entropy coding and the trained dictionary.

Try it

git clone https://github.com/LakoreAI/tokpress
cd tokpress && pip install -e .
python -c "import tokpress; print(len(tokpress.compress(b'{\"a\": 1}')))"

It's ~2900 lines, has 93 tests, and comes with an honest benchmark script. If you have a folder of repetitive logs or JSON, that's exactly the case it was built for.

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Agent harness or agent framework?

firefrog — Sat, 05 Sep 2026 12:30:02 +0000

In July I wrote Prompt, Context, Harness, Loop, which split an agent into four parts and argued that the harness — the thing that owns the loop and decides what the model sees — was the part nobody talked about.

I checked this week. Two of the largest projects in the space now use the word in their own one-line description. LangChain's deepagents calls itself "the batteries-included agent harness." Y Combinator's qm calls itself "multiplayer agent harness for work."

So the word arrived. What hasn't arrived is the distinction, and every comparison article I can find still files harnesses and frameworks under one heading called "AI agent tools." Those are different products that fail in different ways, and picking the wrong one costs you a rewrite.

An agent framework is a library you build an agent with — you own the loop. An agent harness is a runtime that owns the loop and calls your model. Frameworks give you control and hand you the hard problems; harnesses solve the hard problems and take the control. Most teams reach for a framework when they wanted a harness.

What actually separates a harness from a framework?

Take the four parts from the original post and ask, for each one, who owns it — you or the tool?

Part	Framework (LangChain, CrewAI, Mastra)	Harness (opencode, Codex, goose, qm)
Prompt	Yours entirely	Mostly the tool's; you get a config file
Context	Yours to assemble	The tool's, with a compaction strategy baked in
Harness	You write it	This is the product
Loop	You call it	The tool runs it and hands you a result

The row that decides everything is Loop. In a framework you call the library from inside your own loop. In a harness, the harness is the program and your code is a plugin.

That inversion is why "we'll start with a framework and swap it later" so often fails. You're not swapping a dependency; you're inverting control flow.

The landscape, with real numbers

Pulled from the GitHub API on 2026-08-12. Every one of these was pushed to the same day, so activity isn't a differentiator — they're all alive.

Project	Stars	Language	Self-described as
opencode	196,469	TypeScript	"The open source coding agent"
LangChain	144,078	Python	"The agent engineering platform"
Codex	105,518	Rust	"Lightweight coding agent that runs in your terminal"
CrewAI	56,986	Python	"Framework for orchestrating role-playing, autonomous AI agents"
goose	52,718	Rust	"Open source, extensible AI agent"
deepagents	27,688	Python	"The batteries-included agent harness"
Mastra	27,135	TypeScript	"The modern TypeScript framework for AI-powered applications"
qm	13,206	TypeScript	"Multiplayer agent harness for work"

Read the right-hand column as a taxonomy and it sorts itself. "Framework" and "platform" mean you own the loop. "Agent" and "harness" mean they do. The projects are telling you which one they are; the comparison articles just aren't reading it.

Note also that three of the eight are written in Rust or shipped as terminal binaries. That's not a language preference, it's a symptom: if your product is the loop, you are shipping a program, and programs care about startup time and single-binary distribution in a way libraries never do.

Which one do you actually want?

The honest test is a single question, and it isn't about features.

Does an agent run inside your application, or does your application run inside an agent?

If your product is a web service that occasionally needs to reason, you want a framework. The agent is a subroutine. You own the request lifecycle, your observability already exists, and you cannot hand control of the process to something else.

If your product is the agent — a coding assistant, a research tool, an autonomous worker — you want a harness. The loop, context management, tool dispatch and compaction are the hard parts, and they are also solved parts. Rebuilding them on top of a framework is the most common expensive mistake in this space, and it looks like progress for about six weeks.

The trap is that frameworks demo better. A framework demo is fifteen lines and a tidy diagram. A harness demo is a terminal, which looks like less work but is the finished product.

What does an agent harness actually solve for you?

Having built a small one, these are the parts that ate the time — and they're all invisible in a demo:

Context compaction. Not the summarisation call itself; deciding when, choosing what survives, and keeping the result stable enough not to destroy your prompt cache. I wrote about what a compaction does to your cache bill — it resets it entirely, which is correct behaviour and still expensive.

Tool result truncation. A tool returns 400KB of JSON. You cannot put that in the context and you cannot drop it. Every harness has a strategy here and they differ a lot.

Interruption and resumption. The user hits Ctrl-C mid-tool-call. What's the state? Frameworks mostly don't answer this because they assume you own the process.

Permission boundaries. Which tools can run without asking. qm's pitch is isolated workspaces per person, which is this problem at team scale.

Recovery from malformed output. The model emits a tool call with a missing required field. Retry, repair, or surface it? This one decides whether small models are usable at all — which is the subject of the benchmark I'm running next.

Each is a week of work and none of them is interesting. That's precisely the argument for not writing them yourself.

Where does the distinction break down?

Two honest complications, because a taxonomy that admits no exceptions is usually wrong.

deepagents is a harness shipped as a library. It calls itself a harness, and it is one, but it's a Python package you import rather than a binary you run. So it inverts control inside your process. That's a genuine third position and probably where the category is heading — harness semantics, library packaging.

LangChain calls itself a platform now, not a framework. With 144,078 stars and a decade of scope creep it contains both: primitives you build with, and higher-level runners that own the loop. The taxonomy applies to which part you use, not to the repo.

So the test isn't "which project is this?" It's "for the piece I'm about to depend on, who runs the loop?" Ask it per component and the answer stays useful.

What I'd tell someone starting today

Start with a harness and escape downward if you must. The reverse — start with a framework, grow into a harness — means rebuilding the five things above while shipping features, and everyone who does it says the same thing afterwards.

Read the one-line description literally. These projects are precise about what they are. "Platform," "framework," "harness," "agent" are load-bearing words chosen by people who know the difference.

Check who owns compaction before you commit. It's the single highest-leverage behaviour and the hardest to replace. If a tool won't tell you its compaction strategy, that's your answer.

Ignore star counts for this decision. opencode has 14× qm's stars and they're not solving the same problem. Popularity is a proxy for maturity, not fit — and every project here is actively maintained anyway.

TL;DR

The distinction is control flow, not features. In a framework you call the loop; in a harness the loop calls you. Swapping between them isn't a dependency change, it's an inversion.
The projects self-identify accurately. "Framework" and "platform" mean you own the loop (LangChain, CrewAI, Mastra); "agent" and "harness" mean they do (opencode, Codex, goose, qm, deepagents).
The test is one question: does an agent run inside your application, or your application inside an agent?
Harnesses solve five boring, expensive problems — compaction, tool-result truncation, interruption, permissions, malformed-output recovery. Each is a week you won't enjoy.
deepagents is the interesting edge case: harness semantics in library packaging, which may be where the whole category lands.

Data

GitHub API, 2026-08-12. Star counts and languages as reported.
All eight repositories had commits pushed on the day of collection.

Star counts move; the taxonomy doesn't. If you're reading this much later, the numbers are stale and the four-part test still works.

References

Prompt, Context, Harness, Loop: An Agent's Anatomy — the four-part split this post applies. Worth reading first if the terms above felt slippery.
yc-software/qm — the clearest example of harness-as-product, and the one that takes multi-user permission boundaries seriously.
langchain-ai/deepagents — a harness distributed as a library, from the project best known for the framework half of this distinction.

Next: I'm benchmarking whether small models can survive the malformed-output problem above — the one that decides if local agents are viable at all.

👉 Follow me: LinkedIn | GitHub

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Five ways to invalidate your prompt cache

firefrog — Wed, 02 Sep 2026 12:30:03 +0000

Part 2 of 2 on prompt caching. Part 1 covered the economics.

Part 1 established the prize: caching cut an 80-turn agent session from $54.08 to $6.91 in my cost model, an 87.2% saving on input tokens.

Once I had that number I got curious about the opposite question — not how much caching saves, but how easy it is to think you have it and not. That turned out to be the more useful half, because every way of losing it is silent.

This post is about how to lose it. Not by disabling caching — by writing a harness that looks completely correct and quietly never hits the cache. Every failure below is silent. No error, no warning, no degraded output. The agent works exactly as intended and costs several times more than it should.

Prompt caching is a prefix match, not a similarity match. Any change to the earliest part of a request invalidates every cached token after it. In my cost model, putting a timestamp in the system prompt cost 7.8× on an 80-turn session — not because the timestamp is expensive, but because it forfeits the entire discount on every turn.

What exactly does the cache match on?

Providers cache the longest common prefix between your request and something they recently processed.

Prefix. Byte-order, from the front. Not "a similar request," not "the same information in a different order." The comparison walks forward from token zero and stops at the first difference — and everything from that point on is fresh, at full price.

Which yields one design principle for the entire harness:

Stable content first. Volatile content last. Never change what came before.

Every mistake below is a violation of that one line.

1. A dynamic system prompt

The most expensive one-line mistake available, and it looks like good engineering.

Your system prompt sits at position zero. It's the longest-lived thing in the request. So you put useful context in it:

system = f"""You are a coding assistant.
Current time: {datetime.now().isoformat()}
Working directory: {os.getcwd()}
Available tools: {', '.join(discover_tools())}
"""

Every one of those three lines is a cache bomb.

The timestamp changes on every call. So the system prompt changes on every call. So the prefix diverges at token ~15 of a 10,000-token system prompt — and every token after it, including your entire conversation history, is billed fresh. Forever.

Turns	Stable prompt	Dynamic prompt	Penalty	Multiple
10	$0.33	$1.16	$0.83	3.5×
20	$0.79	$3.92	$3.13	4.9×
40	$2.19	$14.24	$12.05	6.5×
80	$6.91	$54.08	$47.17	7.8×

A timestamp doesn't cost you a timestamp. It costs you the whole discount, on every turn, for the rest of the session — and the penalty grows with session length, because you're back on the quadratic curve from part 1.

The fix is not to drop the information. It's to move it:

system = STATIC_PROMPT                     # never changes, caches beautifully
messages = [
    {"role": "system",  "content": system},
    *history,                              # append-only
    {"role": "user", "content": f"<context>time: {now}, cwd: {cwd}</context>\n{query}"},
]

Same information, delivered at the end of the request where volatility is free. The model reads it just as well.

Watch for the sneaky variants: a tool list assembled by iterating a set (non-deterministic order in some paths), a JSON dump with unsorted keys, a "user preferences" block refreshed from a database each call, anything with a request ID.

The tool-definition trap

Worth its own note, because tool schemas sit immediately after the system prompt — near the very front of the prefix, where instability is most expensive — and they're assembled programmatically, which is exactly where non-determinism creeps in.

Three ways I've seen it happen:

tools = [schema(t) for t in registry.values()]        # dict order: usually fine
tools = [schema(t) for t in discovered_plugins]       # filesystem order: not fine
tools = [schema(t) for t in enabled_tools]            # a set: order not guaranteed
json.dumps(tool_schema)                               # key order follows insertion

The failure mode is nasty because it's intermittent. A set iterates in a consistent order within one process, so your local test passes. Restart the process, or run on a machine with a different hash seed, and the order shifts — so the cache works fine in development and misses in production, or works for an hour and then stops after a redeploy.

Two defences, both cheap. Sort your tool list by name before serialising, and pass sort_keys=True when you dump any JSON that lands in the prompt. Then assert on it: hash the serialised system-prompt-plus-tools block at startup and log it. If that hash changes between two runs that should be identical, you have found your cache leak before it found your invoice.

2. History that isn't append-only

The second rule from that design principle, and the one that bites at exactly the wrong moment.

Your conversation history must only ever grow at the end. Editing, reordering, or removing an earlier message shifts the divergence point back to wherever you touched — invalidating everything after it.

Things that quietly do this:

Trimming old messages to stay under the context limit. Dropping the oldest turn changes the prefix at position one and invalidates the entire session cache. You saved context and torched the discount.
Re-rendering tool results with fresh formatting, a new timestamp, or a re-serialised payload.
Sorting or deduplicating history.
Injecting a reminder into the middle of the transcript rather than appending it.

The trimming case deserves attention because it's a genuine conflict: you may have to drop messages. Just know that a sliding window over history means you re-pay for the whole transcript every time the window slides. If you must shed context, do it rarely and in large chunks rather than one message per turn.

3. Compaction

Compaction — summarising the transcript into something short and continuing — invalidates the cache completely, by design. You've replaced the prefix with a different, shorter prefix. Nothing after position zero matches.

This one is not a bug. Compaction is doing exactly what it's supposed to, and the cache reset is the correct price for it.

But it should be a deliberate decision rather than a surprise, because the moment you compact you pay a full cache write on the new prefix and start the accumulation curve over. Two practical consequences:

Don't compact on a timer. Compact when the context genuinely needs it. Every compaction is a fresh write of everything.
Compact in bigger steps, less often. Two compactions cost two cold starts. Same context saved, twice the rewrite.

4. Letting the cache go cold (how long is a TTL?)

Caches expire. The TTL varies by provider and by tier — five minutes is a common default, an hour is often available at a higher write cost, and routed or brokered inference may inherit whatever the backend it landed on happens to offer.

That means wall-clock time is now a cost variable in your agent, which is an unusual thing to have to think about. Step away for coffee and come back, and your next message pays a full rewrite of the entire transcript.

Scenario, 80 turns	Cost	Extra
Uninterrupted	$6.91	—
One break at turn 40	$7.66	$0.75
Breaks at 20, 40, 60	$9.16	$2.25
A break every 10 turns	$12.16	$5.25

Late breaks cost more than early ones, because a cold start rewrites everything so far — and "everything so far" is bigger later. A break at turn 70 is far more expensive than the same break at turn 10.

This is also the argument for the longer TTL, but only conditionally — as part 1 showed, on an uninterrupted session the 1-hour option is a straight 2× markup on writes for no benefit. It pays at roughly three cold starts.

5. Assuming your provider does it for you

The silent one, and the reason to check rather than assume.

OpenAI caches eligible prefixes automatically.
Anthropic and Gemini require explicit cache markers in the request. Omit them and nothing is cached, forever, with no indication.
Conversation-state APIs typically handle it; raw chat-completions calls typically don't.
Routers and inference brokers may send consecutive requests to different backends. A warm cache lives on the machine that built it — if you're not pinned, you can miss a cache that exists. Consistent routing is worth real money here.

None of these produce an error. The only symptom is the bill.

How do you know if your cache is working?

Everything above is invisible without instrumentation, so this is the part I'd implement first — before any of the fixes.

Log cache-hit rate per request. Every major provider returns cached token counts in the usage block of the response. Read them and record the ratio. The field names differ — Anthropic splits cache_creation_input_tokens from cache_read_input_tokens, OpenAI reports cached_tokens nested inside its prompt token details — so normalise once at your client boundary rather than scattering provider checks through the codebase:

def cache_stats(usage):
    """Normalise the usage block into (fresh, written, read)."""
    read    = getattr(usage, "cache_read_input_tokens", 0) or \
              getattr(usage, "prompt_tokens_details", {}).get("cached_tokens", 0)
    written = getattr(usage, "cache_creation_input_tokens", 0)
    total   = getattr(usage, "input_tokens", 0) or getattr(usage, "prompt_tokens", 0)
    return total - read - written, written, read

fresh, written, read = cache_stats(response.usage)
log.info("cache_hit_rate=%.2f", read / max(fresh + written + read, 1))

That single ratio is the whole diagnostic. Everything else on this list shows up in it.

Put effective price per million on a dashboard. Divide total input spend by total input tokens. On a healthy long session it should sit far below list price — my model landed at $0.51/M against a $4.00 list. If yours hovers near list price, your cache is broken.

Watch the shape over a session. A healthy session starts cold, climbs as the transcript grows, and plateaus high. A sawtooth means expiries. A flat line near zero means something in your prefix changes every turn — go and look for a timestamp.

Test it deliberately. Send the same two-turn conversation twice and assert that the second turn reports cache hits. It's a cheap test and it catches every failure on this list.

Isn't this what semantic caching does?

A question that comes up immediately, and the two get conflated constantly, so it's worth separating them.

Semantic caching sits in your infrastructure. It embeds an incoming query, looks for a previously answered question that's close enough in vector space, and returns the stored answer without calling the model at all. It caches outputs, it's approximate, and it can be wrong — two questions can be neighbours in embedding space and still want different answers.

Prompt caching sits in the provider's infrastructure. It caches the processed state of an input prefix, it's an exact match, and it cannot change what the model returns — only what you're billed for the input.

They're complementary, not alternatives. Semantic caching can skip a call entirely, which beats any discount. Prompt caching makes the calls you do make dramatically cheaper. But only one of them can silently give a user a subtly wrong answer, so they deserve very different amounts of scrutiny before you ship them.

The order I'd do this in

Instrument first. Without a hit-rate number, everything else is guesswork.
Check whether your provider caches by default. One doc page; potentially a 7× bill difference.
Audit your system prompt for anything volatile. This is the big one, and it's usually a five-minute fix.
Verify history is append-only, especially wherever context-limit trimming happens.
Only then think about TTL tiers. It's the smallest lever on the list and the only one that costs money to pull.

TL;DR

The cache is a longest-common-prefix match. Stable content first, volatile content last, never modify what came before.
A timestamp in the system prompt cost 7.8× on an 80-turn session in my model — $6.91 becomes $54.08. Move volatile context into the final user message instead.
History must be append-only. Sliding-window trimming re-pays for the entire transcript every time the window moves.
Compaction resets the cache by design — legitimate, but don't do it on a timer.
Anthropic and Gemini don't cache unless you ask. There is no error when you forget; the only symptom is the invoice.

Reproduce this

python 3.13 · stdlib only
system prompt 10,000 tok · user +1,000/turn · assistant +3,000/turn
base input $4.00/M · write 1.25x (5-min) / 2.00x (1-hour) · read 0.10x

Figures are arithmetic on stated assumptions, not measurements of any provider's billing.

References

Anthropic — Prompt caching — cache breakpoints, TTL tiers, and the usage fields you need to compute a hit rate.
Google — Gemini context caching — a usefully different model: caches are explicit objects you create and reference, which makes the prefix contract impossible to ignore.
Kwon et al., Efficient Memory Management for Large Language Model Serving with PagedAttention, SOSP 2023 — explains why the match is a prefix rather than anything cleverer. The KV cache is stored in blocks, and two requests can share a block only while their token sequences are still identical. Everything in this post follows from that.

If you build agent harnesses, Prompt, Context, Harness, Loop covers where the harness sits in the first place — the cache lives in exactly the layer that post calls the harness.

👉 Follow me: LinkedIn | GitHub

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

Fun Project: I Built a Compressor That Thinks in Tokens

firefrog — Tue, 01 Sep 2026 06:51:33 +0000

Field notes from a weekend spent teaching an entropy coder to speak LLM.

LLMs spent billions of dollars learning the best subword dictionary that has ever existed. Somewhere around 200,000 pieces of language, ranked by how much they compress. And I kept looking at that dictionary and thinking: nobody is allowed to just use that for compression?

TokPress is a pure-Python lossless compressor that tokenizes input with OpenAI's o200k_base vocabulary, applies token-level LZ77, and entropy-codes the result with rANS. For many small, schema-homogeneous records (logs, JSON, telemetry) a trained dictionary takes the ratio to 0.2565, and a batch mode reaches 0.0875x — while compressing each record alone can inflate it past 1.0. It beats gzip on prose and loses honestly to zstd's trained dictionary.

So I built one. Pure Python, no native code, one weekend, way too many cups of coffee. It's called TokPress, it compresses by tokenizing with o200k_base (the tokenizer OpenAI's models use), applying token-level LZ77, and entropy-coding the result with rANS. And along the way it produced a number that still makes me grin: 150 JSON log lines, compressed as one stream, down to 0.0875× their size — while compressing each one alone made them bigger.

This is the honest version of how that happened. Including the parts that broke.

Why "many small records" is a genuine pain

Normal compressors like gzip and zstd are built for big files. They learn their model from the stream, which is great when you have megabytes to amortize it over. But the world is full of small, independent, schema-similar records: JSON log lines, telemetry events, API responses, package metadata. Each one is 200 to 500 bytes. Each one is on its own. As an engineer who spends most days building ML systems in production — and who has spent weekends benchmarking tokenizers — this is the shape of half the data I touch.

Compress a single 200-byte log line with gzip and you'll find the header and the dictionary setup cost more than the content you saved. In my measurements, a lone 70-byte JSON sample came out at 81 bytes compressed — a compressor that made things bigger. That's not a bug; it's the cold-start penalty of not having a model.

Zstandard's answer is dictionary mode: train a shared dictionary offline, then every record compresses against it. MongoDB and Amazon DocumentDB now ship exactly this (up to 5× better ratio on JSON documents). And in 2025 the IETF shipped RFC 9842, which put dictionary compression into HTTP itself — browsers, servers, the works. "Known-structure API responses" is literally named as the sweet spot.

So: shared dictionary + many small homogeneous records = a real, money-sized problem. I just wanted to add one twist — use a tokenizer as the dictionary.

Beat 1 — the idea: the tokenizer is the free dictionary

Here's the pitch. A BPE tokenizer's whole job is finding the pieces of language that compress well. "action" is one token. "click" is one token. The tokenizer has already done the hard vocabulary work — for free, at huge scale.

So: tokenize the input, run LZ77 over the token ids instead of bytes, then entropy-code with rANS. The token stream is a much better-shaped alphabet than raw bytes, because the multi-byte structure is already baked in.

import tokpress

compressed = tokpress.compress(payload)      # bytes or str
original = tokpress.decompress(compressed)    # byte-exact

There's one non-negotiable requirement: byte-exactness, even for arbitrary binary. Logs aren't always valid UTF-8. tiktoken's public encode takes a str, but its internal _encode_bytes/decode_bytes pair operates on raw bytes — including a lone 0xFF. That's what I use. It round-trips anything.

Beat 2 — the first wrong turn (and it was a good one)

My first instinct was a domain vocabulary. Train on JSON logs, mine the frequent pieces, get a smaller tokenizer tailored to my data. Sounds great. It silently failed.

The problem: a restricted vocabulary is only a valid tokenizer if it forms a complete hierarchical BPE merge chain — every token has to be the concatenation of two lower-ranked tokens. Mining pieces by longest-match gives you a bag of strings, not a chain. The encoder can't reproduce the merges, so the "tokens" don't round-trip the way you think. I threw the whole domain-profile system out and learned the hard way:

A vocabulary you can't merge is a vocabulary you can't trust.

Beat 3 — three real bugs, all found by being paranoid

Building the entropy coder was where the weekend's real work was. rANS at table precision 2^16 with a 64-bit state, and a selector that builds up to eight candidate encodings per record and keeps the smallest. That last bit is a nice design: min(candidates, key=len) — nothing is ever forced, the size decides.

And it turned out the size decided wrong in three delightful ways. All three were caught only because I looped fuzz inputs over and over instead of testing one fixed example:

The single-symbol truncation. A table with exactly one active symbol gets frequency exactly 65536 — 100% probability. That needs 17 bits. I was writing it in 16, silently truncating 65536 to 0 on the wire. Fix: transmit freq - 1.
The reverse-order escape. rANS encodes in reverse logical order. A two-event cascade (context table → fall through to order-0) needs its encode calls issued in the opposite micro-order from how the decoder consumes them. Get it backwards and it corrupts output only on inputs that exercise the escape path — invisible until it isn't.
The double-reverse. In adaptive-split mode, the escape list is built in a forward pass before the reverse encode loop — unlike every other mode. So I reversed it a second time, by analogy. Wrong. Only when a record had more than one escape did the values come out scrambled.

Three bugs, three regression tests, zero drama. This is the part of compression that doesn't make blog headers but is 90% of the actual work: the wire format must be a perfect mirror between encode and decode, and the only way to trust it is to fuzz it to death.

Beat 4 — the dictionary, and the number that made it click

The real win came from a TokDict: train once on a sample of schema-homogeneous records, then every future record gets three free gifts — an LZ priming buffer (match against other records' history), a baked order-0 rANS table (no per-record table on the wire), and order-1 context tables for the most common previous-token contexts, with an escape cascade so a record can always contain something the dictionary never saw.

tokpress train-dict mydict.tokdict sample1.json sample2.json ...
tokpress compress new_record.json --dict mydict.tokdict -o new_record.tokz
tokpress decompress new_record.tokz --dict mydict.tokdict -o restored.json

On 230 held-out structured-log records (trained on 184, tested on 46 it never saw):

Stage	ratio
per-record, no dictionary	0.800
+ TokDict order-0 table	0.284
+ order-1 context tables	0.2565
gzip -9	0.728
zstd -19, no dictionary	0.737
zstd -19 + matched trained dict	0.195

That's below gzip and dictionary-free zstd, and within 1.3× of zstd given the same training data — down from a 5–7× gap with no dictionary at all. Honest footnote: zstd's COVER/FastCover dictionary training is more mature than my concatenation-based priming, and it still wins on the same data. I'm not claiming otherwise.

And then the batch mode, which is the number I actually open the repo for. Instead of compressing each record on its own, concatenate them all and compress as one stream, with the record lengths stored in the header so decoding is still per-record exact:

packed = tokpress.compress_many(records)
records = tokpress.decompress_many(packed)   # byte-exact, per-record

On 150 schema-homogeneous JSON records (~11.6 KB): compressing each separately summed to ratio 1.19 — the records literally grew. As one adaptive stream: 0.0875. Same bytes, one header instead of 150, one adaptive model spanning the batch instead of 150 cold starts. That's the whole thesis of the project in a single number.

Beat 5 — the turn: tokenization is not the magic

Here's the uncomfortable part, and it's a theorem: tokenization cannot reduce the information content of a message. It's injective, so it preserves entropy. Every compression gain has to come from modeling, never from the transform itself.

I verified this the hard way. In bulk mode, without a trained dictionary, TokPress is just an LZ + entropy pipeline — and on long prose it now beats gzip and, on the short-prose corpus, even zstd outright (0.298 vs 0.324, and brotli 0.306) — but that's the entropy coding doing the work, not the tokenizer. Take the tokenizer away and the wins shrink. The tokenizer's job is to reshape the alphabet so a low-order model sees structure that would need a high-order byte model. It's a lever, not a free lunch.

This also tells you where TokPress sits next to the closest thing it has to a cousin. The parmar project measured "tokenize before you compress" across 452 configurations and found tiktoken-then-xz beats plain xz by 7–9.6% — but it pipes the token IDs into a byte-level compressor, and it explicitly left code, JSON, and logs untested. TokPress goes the other way: it entropy-codes the token IDs directly with rANS, and it's built for exactly those many-small-records, schema-homogeneous corpora parmar skipped. Same starting bet, opposite half of the design space — and the honest result is that on whole-file prose the direct route wins over piping tokens through xz, while neither is a substitute for a trained dictionary.

The other honest finding: a trained dictionary is schema-specific. Train on JSON logs, apply to Python code, and the dictionary buys you nothing over no dictionary at all (0.489 vs 0.485). It's a genuinely domain-specific artifact. Use it on data that resembles its training data, and only then.

Along the way I also learned the machine's own dirty secret: tiktoken's _encode_bytes routes valid UTF-8 through a regex (pat_str) and BPEs per piece — so a vocabulary trained with naive whole-input BPE fragments on piece boundaries. The fix was training the same way tiktoken encodes: pre-tokenize with the same regex, forbid merges across pieces. Once I did that, a JSON-trained vocabulary beat o200k_base on held-out JSON (0.162 vs 0.178). The project now ships tokpress train-vocab and tokpress fit (vocab + dictionary in one shot).

The toolbox, one week later

The weekend project grew a full CLI while I wasn't looking:

tokpress compress record.json -o record.tokz
tokpress decompress record.tokz -o restored.json
tokpress pack batch.tokz record1.json record2.json ...   # one adaptive stream
tokpress unpack batch.tokz out_dir/
tokpress read batch.tokz 7                                # O(1) random access
tokpress train-dict mydict.tokdict samples.jsonl
tokpress train-vocab myvocab.ranks corpus.txt
tokpress fit out corpus.txt                               # both at once
tokpress tokenize-stats file.txt                          # tokens/KB, entropy, MI

That last one is a little love letter to the LLM crowd: it reports the order-0/order-1 token entropy and adjacent-token mutual information of any corpus — compression as a tokenizer-quality signal, which Goldman et al. showed correlates with actual model performance. Tokenize your eval set, get a number that says something about how well the tokenizer fits it.

The whole thing is deliberately un-optimized, pure Python, ~2900 lines, 93 tests. Decompression is fast (thousands of records/sec); compression is slower (the encoder builds every candidate mode and keeps the smallest — thorough, not fast). I made the SymbolStats alphabet pass active-symbol-only one afternoon and got a 2.8× speedup with byte-identical output, which was the most satisfying 40 lines of the weekend.

Share your thoughts in the comments — I’d love to hear how this technology is impacting your industry.

👉 Be sure to press the like button and follow me. It would be a great motivation for me.

👉 Follow me: LinkedIn | GitHub

The takeaway

A tokenizer is a compressor's dictionary that someone else already trained. Reusing o200k_base as the alphabet is free leverage — just don't expect the tokenizer alone to be the win.
Many-small-homogeneous-records is its own regime. One stream over 150 records: 0.0875×. One record at a time: 1.19×. The container is the feature.
Modeling is everything; the transform is a lever. Tokenization preserves entropy — every gain is the entropy coder and the trained dictionary doing real prediction work.
Honest numbers beat impressive ones. It beats gzip on prose, ties/beats zstd on some corpora, and still loses to zstd's trained dictionary by 1.3×. I measured all of it, and the losing parts are the parts that made me learn the most.

If you want to poke at it, it's all open source: github.com/LakoreAI/tokpress. The bench harness (scripts/bench.py) is the source of truth for every number in this post — it round-trip-checks every ratio it prints. I'd love to see what it does on your logs.

References

The papers I kept going back to while building this — the entropy coder, the tokenizer, and the "compression is prediction" framing that got me started.

J. Duda, Asymmetric Numeral Systems, arXiv:0902.0271. The rANS coder — the entropy stage is a direct implementation of the range variant.
J. Ziv and A. Lempel, Compression of Individual Sequences via Variable-Rate Coding, IEEE Trans. Inf. Theory, 24(5), 1978. Token-level LZ77, plus the finite-length redundancy that makes small records the interesting regime.
P. Gage, A New Algorithm for Data Compression, The C Users Journal, 12(2), 1994. Byte-pair encoding — the ancestor of every subword tokenizer here.
A. Radford, J. Wu, et al., Language Models are Unsupervised Multitask Learners, OpenAI, 2019. The byte-level BPE tokenization that became o200k_base.
T. Kudo and J. Richardson, SentencePiece: A simple and language independent subword tokenizer and detokenizer for Neural Text Processing, EMNLP (demo), arXiv:1808.06226, 2018. The subword-modeling framing behind the whole tokenizer-as-dictionary idea.
G. Delétang, A. Ruoss, et al., Language Modeling Is Compression, arXiv:2309.10668, 2023. The prediction-compression equivalence — and the honest "no free lunch" reminder that the transform never carries the win.
C. S. K. Valmeekam, K. Narayanan, D. Kalathil, J.-F. Chamberland, S. Shakkottai, LLMZip: Lossless Text Compression using Large Language Models, arXiv:2306.04050, 2023. The LLM-as-predictor end of the spectrum; TokPress is the cheap, static-dictionary end of the same idea.
O. Goldman, A. Caciularu, M. Eyal, K. Cao, I. Szpektor, R. Tsarfaty, Unpacking Tokenization: Evaluating Text Compression and its Correlation with Model Performance, arXiv:2403.06265 (EMNLP Findings), 2024. The result behind tokpress tokenize-stats — tokenizer compression as a quality signal.
Y. Collet and M. Kucherawy, Zstandard Compression and the application/zstd Media Type, RFC 8478, 2018. The main comparison point, including its dictionary mode I keep losing to.
J. G. Cleary and I. H. Witten, Data Compression Using Adaptive Coding and Partial String Matching, IEEE Trans. Commun., 32(4), 1984. PPM — the escape-to-lower-order idea the per-record order-1 mode borrows.

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

How do you deduplicate a stream you don't control?

firefrog — Sat, 15 Aug 2026 11:33:33 +0000

Field notes, part 1 of 3. Data plumbing from an AI engineer's desk.*

My job title says AI. A meaningful share of my week is data.

Not the glamorous part — no model architecture, no eval harness. Just the plumbing that decides whether the model ever sees the right thing at the right time. And lately I've been stuck on a shape of problem I now see everywhere: something upstream is chatty, and something downstream is expensive.

The expensive thing varies: re-embedding a document, an LLM enrichment call, a search index that must be re-committed. The arithmetic doesn't. Upstream emits fifty events about one entity in twenty seconds, and downstream bills you fifty times for an answer that changed once.

Looking for how this is solved at scale, I landed on a PyCon DE talk by Mirano Tuk and Filip Bacic — How to Search Through 800 Billion Records in Real Time. Their scale is far past mine, but the pattern is small and ported straight onto my problems. These notes are me working it through on my own data.

Deduplicating a stream means collapsing repeated events about one key into a single unit of downstream work. A per-batch set mostly fails, because a batch is a time window whose width you don't control. A TTL buffer decouples that window from throughput — it cut my simulated stream from 904,298 messages to 200,398, a 4.51× reduction.

Why doesn't a per-batch set work?

First instinct: dedupe within each batch. You're already polling in batches, so throw the keys in a set and you're done.

for batch in consumer:
    for key in {msg.key for msg in batch}:
        process(key)

I simulated a stream shaped like the real thing — 200,000 distinct keys, each emitting a burst of updates over a few seconds, 904,298 messages across an hour:

Strategy	Effective window	Downstream work	Reduction
No dedup	—	904,298	1.00×
Per-batch set, 1,000	4.0s	548,524	1.65×
Per-batch set, 10,000	39.8s	260,396	3.47×
Per-batch set, 50,000	199.0s	212,122	4.26×
TTL buffer, 30s	30s	215,364	4.20×
TTL buffer, 60s	60s	200,398	4.51×
TTL buffer, 300s	300s	200,000	4.52×

Read the window column before the reduction column. That's the thing I got wrong when I first ran this.

A batch of 1,000 messages at 251 msg/s is a four-second window. A batch of 50,000 is a 199-second window. Per-batch dedup isn't a different technique from TTL dedup — it's the same technique with a window you didn't choose. Its width is batch_size / throughput.

Which means it moves in exactly the wrong direction. Traffic doubles, your window halves. The moment duplicates are most abundant is the moment your dedup window is narrowest. That's not a knob, it's a trapdoor.

The TTL buffer's entire contribution is decoupling the window from throughput. That's it. That's the idea.

Can a plain dict work as the buffer?

Here's the part that delighted me, because I expected to need a real data structure.

Since Python 3.7, regular dicts preserve insertion order — that's a language guarantee, not a CPython implementation detail. So a plain dict is a FIFO queue. The oldest entry is the first one you iterate. Which means expiry checking is O(1) amortised: look at the front, and if it hasn't expired, nothing behind it has either.

class Deduplicator:
    def __init__(self, ttl, on_evict):
        self.ttl = ttl
        self.on_evict = on_evict
        self.buf = {}                      # key -> expiry, in insertion order

    def add(self, key, now):
        self.expire(now)
        if key in self.buf:
            return                          # already scheduled, drop it
        self.buf[key] = now + self.ttl

    def expire(self, now):
        while self.buf:
            key = next(iter(self.buf))      # oldest entry
            if self.buf[key] > now:
                break                       # nothing behind it can be older
            del self.buf[key]
            self.on_evict(key)

That's the whole thing. No heap, no external store, no Redis.

I did wonder about Redis, and the answer is a nice piece of reasoning: if your key is also your partition key, every event for a given key lands on the same partition, and therefore the same consumer replica. The buffer is correctly local. Reaching for a shared store would add a network hop to enforce an invariant the partitioner already gives you for free.

The bug I'd have shipped

Look again at expire(). The processing happens on eviction — when the key leaves the buffer — not when it arrives.

My first version did the obvious thing and processed on arrival, treating the buffer purely as a "have I seen this?" filter. It looks equivalent. It isn't, and the gap is enormous:

Variant	Work	Keys stale at the end	% stale
ttl=5s, process on insert	425,508	58,239	29.1%
ttl=5s, process on evict	425,508	0	0.0%
ttl=30s, process on insert	215,364	141,879	70.9%
ttl=30s, process on evict	215,364	0	0.0%
ttl=60s, process on insert	200,398	155,650	77.8%
ttl=60s, process on evict	200,398	0	0.0%

Same amount of work. Wildly different correctness.

Processing on insert commits the first version of the record and then discards every update that arrives inside the window. If the last update for a key lands during its own TTL — which, given that updates arrive in bursts, is the common case, not the edge case — you have permanently stale data and no error anywhere to tell you.

At a 60-second TTL, 77.8% of keys ended the run holding a version that was not the latest one.

Flipping to process-on-eviction fixes it completely. You can't know which message is the last one for a key, but if you wait out the TTL and then read the current state, you get the state after the last message in that window. It's an approximation of "process the final update" that costs nothing extra.

The tell is that the work column is identical. This isn't a trade-off. The insert version is just wrong.

How long should the TTL be?

The TTL is latency you are choosing to add. A 60-second buffer means data becomes visible up to a minute after it arrives.

The returns die fast. Going 30s → 60s buys 4.20× → 4.51×. Going 60s → 300s buys 4.51× → 4.52×, for five times the latency. Almost all of the available win is in the first half-minute, because that's the width of the bursts. Set the TTL to roughly the width of your upstream's burst, and stop.

There's a floor you can't cross: 200,000 keys means at least 200,000 units of work. At a 300s TTL the buffer is doing literally nothing but adding delay.

What I'm taking to my own work

The reason I chased this down: the same shape sits underneath a lot of AI infrastructure, and I'd been solving it badly with cron jobs.

A document store where every edit triggers re-embedding. A user-activity stream where every event triggers a profile refresh through an LLM. Both are chatty-upstream, expensive-downstream, and both are places I'd previously reached for "just batch it hourly" — which is a TTL buffer with the worst possible TTL and no correctness story at all.

Process on eviction, never on insert. Same cost, and it's the difference between fresh and silently stale.
A batch is a window you didn't choose. If dedup matters, choose it explicitly.
Set the TTL from your burst width, not from a latency budget someone made up.

TL;DR

A per-batch set is a TTL buffer with a window of batch_size / throughput — it narrows exactly when traffic spikes and duplicates matter most.
A TTL buffer cut 904,298 messages to 200,398 units of work (4.51×) in my simulation, against a hard floor of 200,000.
Process on eviction, not insertion. At a 60s TTL, processing on insert left 77.8% of keys holding stale data for identical cost.
Since Python 3.7 a plain dict is a FIFO queue, so the buffer needs no heap and no Redis — the partitioner already guarantees a key's events reach one replica.

Reproduce this

python 3.13 · stdlib only · seed 20260810
200,000 keys · mean 4 updates each · 8s burst spread · 1 hour horizon

Next: Why committing Kafka offsets out of order loses data — the version of this buffer I'd have shipped drops 4,456 messages on the floor the first time it restarts.

Credit where it's due: the pattern in this note comes from Mirano Tuk and Filip Bacic's PyCon DE 2026 talk. The simulation, the numbers, and any mistakes are mine.

👉 Follow me: LinkedIn | GitHub

Originally published on ZyVOP

💡 For more articles like this, subscribe to the ZyVOP newsletter!

A Hands-On Guide to kalbee: Your First Kalman Filter (and Beyond)

firefrog — Sun, 19 Jul 2026 00:09:00 +0000

Everything you need to go from pip install to a working multi-object tracker, one runnable snippet at a time.

kalbee is a Python library for state estimation — the art of recovering a clean signal (position, velocity, temperature, whatever you're measuring) from noisy sensor data. This guide walks through it from the ground up. Every code block runs as-is; copy them into a file and follow along.

Install

pip install kalbee

The only runtime dependencies are NumPy and SciPy. Optional extras add object-detection (pip install "kalbee[yolo]") and plotting (pip install "kalbee[viz]") support.

The one idea you need: predict and update

Every filter in kalbee works the same way. You alternate between two steps:

predict() — advance the state forward in time using a motion model ("where do I think the object is now?").
update(z) — correct that prediction with a new measurement z ("what does the sensor actually say?").

The filter tracks two things: the state x (your best estimate) and the covariance P (how uncertain that estimate is). You read them back via kf.x and kf.P.

Your first filter

Let's track an object moving at roughly constant velocity, measuring only its (noisy) position. Instead of hand-building matrices, we use kalbee's ready-made models:

import numpy as np
from kalbee import KalmanFilter, rmse
from kalbee.models import constant_velocity, position_measurement_model

dt = 1.0
# Motion model: state is [position, velocity]
F, Q = constant_velocity(dt=dt, process_var=0.01, n_dims=1)
# Measurement model: we observe position only, with noise variance 4.0
H, R = position_measurement_model(order=1, n_dims=1, measurement_var=4.0)

# Simulate a noisy trajectory
rng = np.random.default_rng(0)
pos, vel = 0.0, 1.0
truths, measurements = [], []
for _ in range(50):
    pos += vel * dt
    truths.append(pos)
    measurements.append(pos + rng.standard_normal() * 2.0)  # std 2.0 -> var 4.0

# Create the filter: start at zero with high uncertainty (P = 100 * I)
kf = KalmanFilter(np.zeros((2, 1)), np.eye(2) * 100, F, Q, H, R)

estimates = []
for z in measurements:
    kf.predict(dt=dt)
    kf.update(np.array([[z]]))
    estimates.append(kf.x[0, 0])   # estimated position

print("Raw measurement RMSE:", round(rmse(np.array(measurements), np.array(truths)), 3))
print("Filtered RMSE:       ", round(rmse(np.array(estimates), np.array(truths)), 3))

Run it and you'll see the filtered error is meaningfully lower than the raw measurement error — the filter is smoothing out the noise. That's the whole game.

What the parameters mean

process_var — how much you trust the motion model. Higher = the filter reacts faster but is jumpier.
measurement_var (R) — how noisy your sensor is. Higher = the filter leans more on its predictions.
Initial P — your starting uncertainty. When in doubt, start large (like 100 * I); the filter converges quickly.

Tuning process_var vs measurement_var is the main knob you'll turn. (There's an automatic way to set them — see the EM section below.)

Going 2-D (and beyond)

Every model takes an n_dims argument. Want to track an object in a plane? Ask for two dimensions and the state becomes [x, vx, y, vy]:

F, Q = constant_velocity(dt=1.0, process_var=0.1, n_dims=2)   # 4x4
H, R = position_measurement_model(order=1, n_dims=2, measurement_var=0.5)

Need acceleration too? Use constant_acceleration (state becomes [pos, vel, acc] per axis). Tracking something that turns? constant_turn gives you a coordinated-turn model.

Nonlinear systems: EKF and UKF

When your motion or measurement isn't linear, swap in the Extended or Unscented Kalman Filter. Instead of matrices, you pass functions:

import numpy as np
from kalbee import UnscentedKalmanFilter

def f(x, dt):          # state transition
    F = np.array([[1, dt], [0, 1]])
    return F @ x

def h(x):              # measurement function
    return x[:1]       # observe position only

ukf = UnscentedKalmanFilter(
    state=np.zeros((2, 1)),
    covariance=np.eye(2) * 100,
    transition_covariance=np.eye(2) * 0.01,
    measurement_covariance=np.array([[4.0]]),
    transition_function=f,
    measurement_function=h,
)

ukf.predict(dt=1.0)
ukf.update(np.array([[1.2]]))
print(ukf.x)

The UKF needs no derivatives; the EKF (ExtendedKalmanFilter) is similar but also takes Jacobian functions. The same predict/update loop applies to all ten filters kalbee ships — including Particle, Ensemble, Information, Square-Root, and Vectorized filters.

Not sure which filter? Compare them

kalbee has a built-in experiment runner that races several filters on a synthetic signal:

from kalbee import run_experiment

report = run_experiment(
    signal="sine",                       # or "linear", "step", "maneuver"
    filters=["kf", "ekf", "ukf", "pf"],
    noise_std=0.5,
    seed=42,
)
print(report.summary())

You get a ranked table of position/velocity RMSE and a consistency metric (NEES) for each filter — a quick way to pick the right tool before committing.

Tracking many objects

To track multiple objects (people in a video, blips on a radar), wrap a filter in a MultiObjectTracker. You supply a small factory that builds a fresh filter for each new object:

import numpy as np
from kalbee import KalmanFilter, MultiObjectTracker
from kalbee.models import constant_velocity, position_measurement_model

F, Q = constant_velocity(dt=1.0, process_var=0.05, n_dims=2)
H, R = position_measurement_model(order=1, n_dims=2, measurement_var=0.5)

def new_track(z):
    # Seed state [x, vx, y, vy] at the detection, zero initial velocity.
    x0 = np.array([[z[0]], [0.0], [z[1]], [0.0]])
    return KalmanFilter(x0, np.eye(4) * 10.0, F, Q, H, R)

tracker = MultiObjectTracker(new_track, n_init=3, max_age=5, gate=6.0)

# Each frame, pass a (D, 2) array of detected positions:
for detections in detection_stream:
    confirmed = tracker.update(detections)
    for t in confirmed:
        print(f"id={t.id}  pos=({t.state[0,0]:.1f}, {t.state[2,0]:.1f})")

The tracker handles the hard parts for you: matching detections to existing tracks (Hungarian algorithm with distance gating) and managing each track's lifecycle. The knobs:

n_init — how many consecutive detections before a track is "confirmed" (filters out flicker).
max_age — how many missed frames before a track is deleted (handles occlusion).
gate — the maximum distance for a valid match.

Only confirmed tracks come back from update(). Feed it the box centers from a detector like YOLO and you have a complete tracking pipeline.

Let the data pick your noise values

Struggling to choose Q and R? Learn them from a recording with EM (Expectation-Maximization):

from kalbee import em_kalman
from kalbee.models import constant_velocity, position_measurement_model

F, _ = constant_velocity(dt=1.0, n_dims=1)
H, _ = position_measurement_model(order=1, n_dims=1)

# measurements: array of shape (T, m)
result = em_kalman(measurements, F, H, n_iter=50)

print("Learned Q:\n", result.Q)
print("Learned R:\n", result.R)
print("Converged:", result.converged)

Fit once on representative data, then plug result.Q and result.R into a live KalmanFilter. It's a principled alternative to hand-tuning.

A few tips

Shapes matter. States are column vectors (n, 1); measurements are (m, 1). When in doubt, reshape(-1, 1).
Start uncertain. A large initial P lets the filter trust early measurements and converge fast.
Watch consistency, not just error. If the average NEES is far from your state dimension, your Q/R are mistuned — the metrics module (nees, nis, log_likelihood) tells you.
Reproducibility is built in. Sampling filters (particle, ensemble) and the signal generators accept a seed/rng argument, so runs are deterministic when you want them to be.

Where to go next

The documentation has a dedicated page per filter with the underlying math and worked examples.
The examples/ folder has runnable scripts, including a full multi-object tracking demo and YOLO integration.
Everything shares the same predict/update interface — so once you've done this tutorial, the rest of the library is just variations on what you already know.

Happy estimating.

Meet kalbee: State Estimation and Target Tracking, Minus the Pain

firefrog — Tue, 14 Jul 2026 15:54:44 +0000

If you've ever tracked a car from jittery GPS, smoothed noisy YOLO boxes, or fused a handful of sensors, you know the drill. You either hand-roll filter equations and pray your covariance stays positive-definite, or you wrestle a heavyweight legacy toolkit built for a PhD thesis, not your project.

kalbee is the third option: a clean, modular, numerically stable Python library for modern state estimation and target tracking — one unified API across 10 filters, a smoother, and a full diagnostics suite.

What's inside

10 filters, one interface:

Kalman Filter — Joseph-form updates for a rock-solid baseline
Extended KF — analytical Jacobians for nonlinear systems
Unscented KF — sigma-point propagation, no Jacobians needed
Particle Filter — non-Gaussian / Monte Carlo estimation
Ensemble KF — built for high-dimensional states
Information Filter — the natural fit for multi-sensor fusion
Alpha-Beta-Gamma — lightweight fixed-gain tracking
Adaptive KF — estimates noise Q and R on the fly
Square-Root KF — Cholesky-based updates, roundoff-proof
Vectorized KF — batched multi-target tracking

Plus the IMM estimator (switching motion models), an RTS smoother for backward passes, and diagnostics — RMSE, NEES, NIS, log-likelihood — so you can actually prove your filter is consistent, not just eyeball it.

Stability you don't have to think about

Textbook Kalman filters die quietly. Roundoff error creeps in, P loses symmetry or positive-definiteness, and your estimate drifts or crashes. kalbee handles this for you:

Joseph-form updates keep P positive-semidefinite through aggressive measurement updates
Square-Root filtering propagates the Cholesky factor directly, sidestepping singularity
Singular-inversion fallbacks auto-regularize and fall back to pseudoinverses when things get numerically ugly

Showcase: tracking a car through YOLO boxes

Occlusions and sharp turns are where naive trackers fall apart. On a real video stream, tracking a vehicle against raw YOLO detections:

Filter                | Mean deviation (px) | Notes
----------------------|---------------------|---------------------------------
Standard Kalman       |        7.36         | Overshoots on turns
Square-Root Kalman    |        7.36         | Same accuracy, roundoff-robust
Particle Filter       |       57.84         | Drifts when YOLO loses confidence
IMM Blended           |        6.03         | Best — adapts to speed changes

Why IMM wins: it runs Constant-Velocity and Constant-Acceleration models in parallel. Cruising straight? It leans on Constant Velocity. The instant the car brakes or turns, weight shifts to Constant Acceleration, cutting lag. During occlusions, it extrapolates smoothly from velocity history.

Three lines to your first comparison

#pip install kalbee

from kalbee import run_experiment

report = run_experiment(
    signal="sine",
    filters=["kf", "ekf", "ukf", "pf"],
    noise_std=0.5,
)
print(report.summary())

One call, four filters, a ranked report. Swap signals, add filters, tune noise — iterate in seconds.

Try it

GitHub: https://github.com/MinLee0210/kalbee
Docs: https://minlee0210.github.io/kalbee/
Tutorial: the hands-on tracking_tutorial.ipynb

Building a robot, processing camera feeds, or researching sensor fusion — kalbee gives you the numerical stability of a serious toolkit with an interface you'll actually enjoy using.

wavio

firefrog — Sat, 11 Jul 2026 03:16:46 +0000

| Audio Fingerprinting Without the ML Tax
Most audio identification tools today reach for embeddings and neural nets. wavio doesn't.

It's a fast, deterministic acoustic fingerprinting library written in Rust — built on the same peak-based approach as Shazam, with none of the ML overhead. No embeddings, no models, no runtime. Just spectral peaks, combinatorial hashing, and raw speed.

How it works

wavio runs a straightforward DSP pipeline:

Each step is deterministic: the same input always produces the same fingerprint. That makes results reproducible and debuggable — no model drift, no version mismatches.

Why it's fast

In-memory & on-disk indexing — query thousands of tracks in microseconds
Zero unsafe code (#![forbid(unsafe_code)])
Optional parallelism via rayon

Benchmarks on synthetic 22,050 Hz audio (release build):

Task	Median
Fingerprint a 3-min track	88.6 ms
Index 1,000 tracks	1.04 ms
Query (1,000 lookups)	0.57 µs/query

Use it from Rust, Python, or the CLI

wavio ships as a Rust crate, a Python package (via PyO3/maturin), and a CLI:

wavio-cli index --db ./wavio.db ./music/
wavio-cli query --db ./wavio.db ./clip.wav

Who it's for

If you're building music identification, duplicate detection, or content matching and don't want to carry an ML stack for it, wavio gives you a lean, predictable alternative.

Check it out: github.com/MinLee0210/wavio

Enigmar

firefrog — Thu, 02 Jul 2026 14:40:52 +0000

| A High-Performance Enigma Machine Simulator in Rust, with Python Bindings

Link: https://lib.rs/crates/enigmar

Few machines capture the imagination like the Enigma. Used by the Wehrmacht in World War II and famously broken at Bletchley Park, its cipher logic — rotors, reflectors, plugboards, and a maddening quirk called double-stepping — is a rite of passage for anyone interested in classical cryptography.

Enigmar is an educational library that faithfully recreates the Enigma M3/M4 machines. It's written in Rust for speed and correctness, with Python bindings via PyO3 so you can experiment without leaving your notebook.

Why It's Interesting

Most Enigma simulators get the easy parts right and fumble the details. Enigmar focuses on historical accuracy:

Rotors I–VIII and Reflectors B, C, B-thin, C-thin, matching the real wiring diagrams
Correct double-stepping — the mechanical anomaly where the middle rotor sometimes steps twice in a row, a quirk of the real machine's ratchet mechanism that many simulators simplify away
Reciprocal encryption — because of how the reflector works, encrypting and decrypting are literally the same operation with the same settings
No letter self-encryption — a structural weakness of the real Enigma that cryptanalysts exploited; Enigmar reproduces it faithfully rather than "fixing" it

Under the hood, the core uses [u8; 26] lookup tables for zero-allocation, O(1) character mapping, so it's fast enough for large-scale experimentation or teaching demos.

How the Signal Path Works

Every keystroke travels through the plugboard, three rotors, a reflector, and back out through the rotors and plugboard again:

Input → Plugboard → Rotor III → Rotor II → Rotor I → Reflector
                                                          ↓
Output ← Plugboard ← Rotor III ← Rotor II ← Rotor I ←────┘

Before each character is processed, the right rotor always steps. If it's sitting at its notch, the middle rotor steps too. And if the middle rotor is at its notch, both it and the left rotor step together — the double-stepping anomaly that made the real machine's period shorter than naive rotor math would suggest.

Getting Started

Rust

[dependencies]
enigmar = { path = "." }

use enigmar::EnigmaBuilder;

fn main() {
    let mut machine = EnigmaBuilder::new()
        .rotor("I", 0, 0)
        .rotor("II", 0, 0)
        .rotor("III", 0, 0)
        .reflector("B")
        .plugboard("AV BS CG DL FU HZ IN KM OW RX")
        .build()
        .unwrap();

    let ciphertext = machine.process_string("HELLOWORLD");
    println!("Encrypted: {}", ciphertext);
}

Python

pip install maturin
maturin develop --features extension-module

from enigmar import EnigmaBuilder

builder = EnigmaBuilder()
builder.rotor("I", 0, 0)
builder.rotor("II", 0, 0)
builder.rotor("III", 0, 0)
builder.reflector("B")
builder.plugboard("AV BS CG DL FU HZ IN KM OW RX")
machine = builder.build()

ciphertext = machine.process_string("HELLOWORLD")
print(f"Encrypted: {ciphertext}")

# Save and restore state
key = machine.export_key()
machine.import_key(key)
machine.reset()

Decryption uses the exact same settings — build a fresh machine with identical rotors, reflector, and plugboard, then run the ciphertext back through it.

API at a Glance

EnigmaBuilder

Method	Description
`new()`	Create an empty builder
`.rotor(type, position, ring)`	Add a rotor, left to right. Types `"I"`–`"VIII"`
`.reflector(type)`	Set reflector: `"B"`, `"C"`, `"B-thin"`, `"C-thin"`
`.plugboard(pairs)`	Set up to 13 plug pairs, e.g. `"AB CD EF"`
`.build()`	Produce the `EnigmaMachine`

EnigmaMachine

Method	Description
`process_string(input)`	Encrypt/decrypt; non-alpha characters are dropped
`export_key()`	Serialize state to JSON
`import_key(key)`	Restore state from JSON
`reset()`	Reset rotors to their initial positions

Who It's For

Enigmar is built for learning: cryptography students exploring rotor ciphers, developers curious about PyO3-based Rust/Python interop, or anyone who wants to reproduce Bletchley-era encryption on modern hardware. The JSON key export also makes it easy to save, share, and reproduce exact machine configurations for teaching or testing.

Give it a try — encrypt something, export the key, and decrypt it back with a fresh machine instance. It's a small, satisfying way to see 1930s engineering behave exactly as designed.

How to build an autonomous news Generator with AI using Fluvio

firefrog — Mon, 02 Sep 2024 07:47:17 +0000

Building a News Bot with Fluvio

Introduction

In my previous article, I introduced the concept of event-driven architecture (EDA) and demonstrated its capabilities using Fluvio. I showcased how an application could leverage EDA to asynchronously send quotes from a publisher to subscribers at regular intervals.

In this article, I will expand upon my previous work by introducing additional features that enhance the application's functionality. I will delve into how to integrate a search engine to discover relevant quotes and utilize Large Language Models (LLMs) to summarize these quotes effectively. By combining these elements, I aim to create a more robust and informative application that leverages the power of EDA and AI, named Wipe.

Note: For better experience, I recommend reading my post on Substack. In addition, I am participating in a quest on Quira, please up vote for me. Here is the link

What is Wipe?

Tired of Falling Behind in the Fast-Paced World of AI? I understand the frustration of trying to keep up with the constant stream of new technologies and trends in the AI landscape. It can be overwhelming to stay informed about the latest developments while also focusing on building your projects.

Introducing Wipe, your AI-powered solution. By leveraging a powerful combination of search engines and Large Language Models, Wipe automatically curates the most relevant AI news and condenses it into concise summaries. No more sifting through countless articles. With Wipe, you can stay ahead of the curve and ensure your projects are always built with the latest insights and technologies.

Features

As an AI enthusiast, I've often found myself asking: How can I stay up-to-date with the rapid advancements in this field? Is there a way to capture the essence of countless articles in a matter of seconds?

Wipe, your AI-powered solution, offers the following benefits:

Real-time Updates: Stay informed about the latest AI trends and breakthroughs.

Instant Summarization: Use Large Language Models to quickly grasp the key points of articles.

Workflow

Data Ingestion: The Publisher continuously collects and ingests raw AI trend data from various sources.

Feature Extraction: The Feature component processes this data, extracting relevant features and insights through techniques like natural language processing and data analysis.
Content Refinement: The Feature component further refines the extracted content, summarizing key points or providing additional context.
Notification Distribution: The Notification component sends the processed and refined AI trend updates to interested Consumers. In these settings, fluvio handles this component very pretty.
Consumer Utilization: Consumers receive these updates and leverage them for their specific AI applications, such as model training, product development, or research.

Prerequisites

Event-driven Architecture (EDA)

I would love to remind myself a little bit about the EDA. Event-driven architecture is a design pattern where applications respond to events asynchronously. This allows for greater scalability and responsiveness compared to traditional request-response models. Events can be triggered by various sources, such as user actions, system changes, or external data feeds.

Inside event-driven architectures EDA is widely used in various domains, including:

Real-time data processing: Processing financial market data, IoT sensor data, and other time-sensitive information.
Microservices architecture: Decoupling services, facilitating asynchronous communication, and enabling independent scaling.
Serverless computing: Executing functions in response to events, such as file uploads, database changes, or API calls.

Tech-stack

At the core of Wipe lies a robust technological stack designed to deliver real-time updates and insightful summaries. Let's break down the key components:

Fluvio: As a high-performance streaming engine, Fluvio efficiently handles the continuous flow of data, ensuring that news articles are processed and delivered promptly. Its Rust-based ## architecture guarantees low latency and security.
Redis: This in-memory data store acts as a central hub, storing and retrieving data seamlessly between the publisher and consumer components.
Langchain: By providing a vast array of Large Language Models (LLMs), Langchain empowers Wipe to understand and summarize complex articles with exceptional accuracy.
Tavily Search: This AI-integrated search engine plays a crucial role in identifying relevant news articles, ensuring that Wipe delivers only the most pertinent information to its users.

Together, these components form a powerful synergy that enables Wipe to provide users with timely, accurate, and informative AI news updates.

Getting Started

Wipe relies heavily on Docker; therefore, you should utilize Docker to get the best result. First and foremost, let’s clone the repository:

git clone <https://github.com/MinLee0210/Wipe.git>
cd ./Wipe
pip install -r requirements.txt

Environment Setup

To setup the environment for the project, you must to:

Install the Fluvio (view my previous article or on Fluvio’s website).
Install Redis (Wipe used Redis on Docker, I’ll leave the link here).
Get API keys from Tavily (a must) and a LLM’s provider that you want to use (Gemini, Groq, OpenAI), remember to change the configuration from config.yaml file.

Make sure to create a .env file that follows this structure:

TAVILY_API_KEY=""
GEMINI_API_KEY=""
GROQ_API_KEY=""

Experiment Setup

I can not show everything of the code in this blog; however, I will show 3 most important components of the app and the logic of the news_features .

The WipeProducer.

The WipeConsumer.

The WipeDB.

Before deep dive into those 3 components, I will show the get_latest_trends()function that allows getting the latest trends in the AI field.

TREND = "What is the latest trend in AI 2024?"

def get_latest_trend() -> tuple[list[Article], list[Event]]:
    """
    Retrieves the latest trend in AI and returns a list of summarized articles.

    Returns:
        list[Article]: A list of Article objects containing summaries of relevant news.
    """

    # Find relevant URLs
    urls = [result["url"] for result in searcher.run(TREND)["results"]]

    # Filter out unsupported URLs and scrape content
    docs = []
    for url in urls:
        try:
            docs.append(scraper.run(url))
        except ValueError:
            continue  # Skip unsupported URLs


    # Process and summarize articles
    articles, events = [], []
    for idx, doc in enumerate(docs):
        metadata = doc[0].metadata
        content = clean(doc[0].page_content)

        summary_prompt = SUMMARY_ARTICLE.format(article=content)
        summary = llm.invoke(summary_prompt).content  

        # Create Article object with metadata and summary
        article = Article(summary=summary, **metadata)
        articles.append(article)
        # Create Event object with Article's information
        event_title = f"Latest Trend in AI (2024) - {article.title}"  # Modify title creation if needed
        event = Event(title=event_title, 
                    article_id=article.id)
        events.append(event)

    return (articles, events)

The TREND is defined to be strict and related to our topic ai-trends. However, it can be extended further into any topics that you want it to be. The flow of the algorithm behaves as follows:

It gets trends from a bunch of websites that are relevant based on Tavily Search Engine.

Those websites are then scrapped via LangChain’s WebBaseLoader and summarized via an LLM (I used Gemini; additionally, you can use another library to simplify this stage, such as ScrapeGraphAI).

The processed documents are then fed into 2 objects: Event and Article. The former is sent to the Consumer to notify them there are new trends that are gathered successfully; the latter is saved in the database, and based on the Consumer’s choice, the Article is then read by getting it from the database.

WipeProducer

Here is the code of the Producer:


"""
A simple Fluvio producer that produces records to a topic.
"""
import subprocess

from fluvio import Fluvio

class WipeProducer:
    """
    A class to produce records to a Fluvio topic.

    Attributes:
    ----------
    topic_name : str
        The name of the topic to produce to.
    partition : int
        The partition to produce to.
    producer : Fluvio.topic_producer
        The Fluvio producer object.

    Methods:
    -------
    produce_records(num_records)
        Produces a specified number of records to the topic.
    flush()
        Flushes the producer to ensure all records are sent.
    """
    ROLE = "producer"

    def __init__(self, topic_name: str, partition: int):
        """
        Initializes the FluvioProducer object.

        Parameters:
        ----------
        topic_name : str
            The name of the topic to produce to.
        partition : int
            The partition to produce to.
        """
        self.topic_name = topic_name
        self.partition = partition
        self.producer = Fluvio.connect().topic_producer(topic_name)

    def produce_records(self, event: str) -> None:
        """
        Produces a specified event.

        Parameters:
        ----------
        event : str
            The information of the event
        """
        try:
                self.producer.send_string(event)

        except Exception as e:
            print(f"Error producing records: {e}")

    def flush(self) -> None:
        """
        Flushes the producer to ensure all records are sent.
        """
        try:
            self.producer.flush()
            print("Producer flushed successfully")
        except Exception as e:
            print(f"Error flushing producer: {e}")

    def __create_topic(self, topic_name:str):
        """
        Create a topic. 

        Parameters: 
        ----------
        topic_name: str
            The name of the topic
        """
        try:
            shell_cmd = ['fluvio', 'topic', 'create', topic_name]
            subprocess.run(shell_cmd, check=True)
        except subprocess.CalledProcessError as e:
            print(f'Command {e.cmd} failed with error {e.returncode}')

The Producer object has 3 main methods:

Connect to the chosen topic via the constructor.
Send records to the Consumer.
Flushes the producer to ensure all records are sent.

I also provide a method for Python code to execute shell commands, allowing creating topics via the Producer interface.

The logic of the Producer is defined as:

producer = WipeProducer(topic_name=config["pubsub"]["topic"],
                            partition=config["pubsub"]["partition"])

# ===== PRODUCER'S METHODS =====
def pub_produce_articles():
    """
    Publishes summarized articles to the defined topic.
    """
    trends = get_latest_trend() #   (articles, events)
    for article, event in zip(trends[0], trends[1]):

        event_str = json_to_str(event.json())
        producer.produce_records(event_str)  # Serialize event to JSON

        article_str = json_to_str(article.json())
        wipe_db.set_article(id=article.id, 
                            role=producer.ROLE, 
                            content=article_str)
    producer.flush()

WipeConsumer

The implementation of the WipeConsumer is as follows:


"""
A simple Fluvio consumer that consumes records from a topic.
"""

from datetime import datetime
from fluvio import Fluvio, Offset

class WIPEConsumer:
    """
    A class to consume records from a Fluvio topic.

    Attributes:
    ----------
    role : str
        The role of the consumer (in this case, 'customer').
    topic_name : str
        The name of the topic to consume from.
    partition : int
        The partition to consume from.
    consumer : Fluvio.partition_consumer
        The Fluvio consumer object.

    Methods:
    -------
    consume_records(num_records)
        Consumes a specified number of records from the topic.
    """

    ROLE = 'customer'

    def __init__(self, topic_name: str, partition: int):
        """
        Initializes the WIPEConsumer object.

        Parameters:
        ----------
        topic_name : str
            The name of the topic to consume from.
        partition : int
            The partition to consume from.
        """
        self.topic_name = topic_name
        self.partition = partition
        self.consumer = Fluvio.connect().partition_consumer(topic_name, partition)
        self.notification = []

    def consume_records(self, num_records: int) -> None:
        """
        Consumes a specified number of records from the topic.

        Parameters:
        ----------
        num_records : int
            The number of records to consume.
        """
        try:
            for idx, record in enumerate(self.consumer.stream(Offset.from_end(num_records))):
                print(f"Record {idx+1}: {record.value_string()}: timestamp: {datetime.now()}")
                self.notification.append(record.value_string())
                if idx >= num_records - 1:
                    break
        except Exception as e:
            print(f"Error consuming records: {e}")

    def flush(self): 
        """
        Delete the notification
        """
        self.notification = []

The Consumer object has 3 main methods:

Create a connection to the Producer based on the chosen topic via the constructor.
Consume the records: Wipe sets the Consumer to consume the 1-latest article from the Producer.
Delete notification: Everytime the Producer creates an article, it will notify its consumers. The Consumer will then store it in the self.notification; based on the Consumer’s choice, the article is then read by getting it from the database.

The logic of the Consumer is:

# ===== CONSUMER'S METHODS =====
def sub_catch_articles():
    """
    Consumes events from the topic and processes them (implementation pending).
    """
    logger.info("[CONSUMER]: Catch events from Producer")
    consumer.consume_records(config["pubsub"]["num_records_consume"])

def sub_read_articles(): 
    """
    Retrieve articles from the database based on the events.
    Note: 
        Assume the Consumer chose the 1-latest event. 
    """

    logger.info("[CONSUMER]: Get event")
    event = consumer.notification[-1]

    while not isinstance(event, dict):
        event = str_to_json(event)

    logger.info("[CONSUMER]: Get article")
    article_id = event['article_id']
    article = wipe_db.get_article(id=article_id, 
                                  role=consumer.ROLE)

    logger.info(f"[CONSUMER]: Reading\n{article}")

WipeDB

For the sake of simplicity, the implementation of the database is quite simple: get and set articles.


"""
Define database logic for WIPE.
"""

import redis

AUTHORIZED_METHODS = {
    'get': ['customer', 'producer'], 
    'set': 'producer'
}

class WIPEDB(object):
    """
    A class to handle database operations for WIPE.

    Attributes:
    ----------
    server : redis.Redis
        The Redis database connection.

    Methods:
    -------
    get_article(id, role)
        Retrieves an article from the database.
    set_article(id, role)
        Sets an article in the database.
    """

    def __init__(self, db_config: dict):
        """
        Initializes the WIPEDB object.

        Parameters:
        ----------
        db_config : dict
            A dictionary containing the Redis database configuration.
        """
        self.server = self.__set_db_connection(db_config)

    def get_article(self, id: str, role: str) -> str:
        """
        Retrieves an article from the database.

        Parameters:
        ----------
        id : str
            The ID of the article to retrieve.
        role : str
            The role of the user requesting the article.

        Returns:
        -------
        str
            The article content if the user is authorized, otherwise None.
        """
        if role not in AUTHORIZED_METHODS['get']:
            return None
        try:
            return self.server.get(id)
        except redis.exceptions.RedisError as e:
            raise e

    def set_article(self, id: str, role: str, content: str) -> bool:
        """
        Sets an article in the database.

        Parameters:
        ----------
        id : str
            The ID of the article to set.
        role : str
            The role of the user setting the article.
        content : str
            The content of the article.

        Returns:
        -------
        bool
            True if the article was set successfully, otherwise False.
        """
        if role != AUTHORIZED_METHODS['set']:
            return False
        try:
            self.server.set(id, content)
            return True
        except redis.exceptions.RedisError as e:
            raise e

    def __set_db_connection(self, db_config: dict) -> redis.Redis:
        """
        Establishes a connection to the Redis database.

        Parameters:
        ----------
        db_config : dict
            A dictionary containing the Redis database configuration.

        Returns:
        -------
        redis.Redis
            The Redis database connection.
        """
        try:
            server = redis.Redis(**db_config)
            return server
        except redis.ConnectionError as r_ce:
            raise r_ce

To make the app more fun, Wipe constructs the app to have the authorization for interacting with the database. For instance, the Consumer is restricted to make a set method to the database.

Code in Action

From the producer, the code be like:

./producer.py

import time

from controller.pubsub import pub_produce_articles


while True: 
    pub_produce_articles()  # Avg of 35 secs per call.
    time.sleep(10)
For every 10 seconds, the Producer will look up the Internet for the latest AI trends.

The Consumer will also get the notification after every 10 seconds:

./consumer.py

import time
import random
from controller.pubsub import sub_catch_articles, sub_read_articles

while True: 
    sub_catch_articles()
    time.sleep(10)

    rand_idx = random.randint(a=0, b=10)
    if rand_idx % 2 == 0: 
        sub_read_articles()

For a more engaging experience, I incorporated a random element into the notification system. If the random outcome met specific conditions, the Consumer would be shown the article.

To run the app, simply type

python producer.py &
python cosumer.py

Result

The Producer automatically gets the latest trends from the Internet and uses AI to summarize the website every 10 seconds. After the summarization is done, it makes an event to notify its customers.

The Customer retrieves a notification from its producer. In this experiment, I set it to randomly choose whether to "read" the news from the notification or not.

Conclusion

In this article, I have successfully extended the capabilities of the event-driven architecture (EDA) application introduced in the previous installment.

By integrating a search engine and utilizing Large Language Models (LLMs), the application, now named Wipe, has become a more comprehensive and informative tool. The ability to discover relevant quotes and generate concise summaries enhances the user experience and provides valuable insights into the vast world of AI.

The successful implementation of these features demonstrates the versatility and power of EDA in creating robust and scalable applications.

How to build an event-driven architecture with Fluvio

firefrog — Wed, 28 Aug 2024 03:29:20 +0000

Introduction

Get started on a journey into the world of event-driven architecture with Fluvio. This powerful platform offers a streamlined approach to building real-time, scalable, and resilient applications. By leveraging Fluvio's capabilities, you can unlock the full potential of event-driven design and create innovative solutions that meet the demands of today's dynamic environments.

In this guide, we'll delve into the intricacies of Fluvio, exploring its key features, benefits, and practical implementation strategies. You'll learn how to utilize the power of event-driven architecture to build applications that are responsive, scalable, and efficient.

Some information

Event-driven architecture

Imagine you're hosting a party. You want to notify everyone when the pizza arrives. Instead of shouting to each guest individually, you could simply announce it once, and everyone who's interested in pizza will hear and react accordingly.

This is essentially the concept of event-driven architecture. It's a design pattern where components of a system communicate by producing and consuming events. Think of it as a way to create a more dynamic and responsive system, similar to how your party guests react to your announcement.

Now, let's introduce Pub/Sub.

Imagine you're the party host (the publisher). When the pizza arrives, you publish an event called "Pizza Is Here.". Your guests (the subscribers) can subscribe to this event. When they hear your announcement (the event), they'll take action (e.g., grab a slice).

In a pub/sub system, the publisher sends out events, and subscribers can choose to listen to specific events. This decouples the components, making the system more scalable, flexible, and resilient.

Here's a more technical breakdown:

Publisher: Produces events and sends them to a message broker.
Message Broker: Stores and distributes events to interested subscribers.
Subscriber: Consumes events and takes appropriate actions.

Imagine a social media platform. When a user posts a new message, that's an event. Other users who follow that user can subscribe to their posts and receive notifications whenever a new message is published.

Key benefits of Pub/Sub:

Scalability: handles large volumes of events efficiently.
Flexibility: Allows for dynamic subscriptions and decoupled components.
Resilience: Ensures messages are delivered even if components fail.
Real-time updates: Enables real-time communication and updates.

Note: I found an interesting video that can help you easily understand the concept; here is the link.

Fluvio

Fluvio's exceptional performance and efficiency make it a standout choice for real-time data processing. Its low-latency capabilities ensure that data is processed swiftly, enabling applications to respond to events in a timely manner. Furthermore, Fluvio's lightweight design and optimized architecture minimize resource consumption, making it suitable for even the most resource-constrained environments.

Fluvio's rich API support and customizable stream processing capabilities make it a developer's dream. With client libraries available for popular programming languages, you can easily integrate Fluvio into your existing applications. The platform's programmability allows you to tailor data processing pipelines to meet your specific requirements, ensuring maximum flexibility and control.

Moreover, Fluvio's WebAssembly integration enables you to securely execute custom stream processing logic, providing a powerful and efficient way to extend the platform's capabilities.

Code in Action

Please read the article via this website for detailed implementation and better visualizations.

Conclusion

In this article, we talked about one of the greatest architecture in programming: Pub/Sub, a fundamental component of event-driven architecture. It provides a robust and scalable foundation for event-driven architectures, enabling loosely coupled, asynchronous communication between components. In addition, we used Fluvio to demonstrate the architecture by allowing the publisher to generate quote every 7 seconds to the Consumer. Clearly, this framework provides us an easy approach to event-driven architecture.

If you guys want me to continue this approach in LLMs applications or develop it further,. You guys can comment to let me know!

Thank you for reading this article; I hope it added something to your knowledge bank! Just before you leave:

👉 Be sure to press the like button and follow me. It would be a great motivation for me.

👉 More details of the code refer to: Github

👉 Follow me: LinkedIn | Github

Boost Your RAG Performance with Tavily Search API

firefrog — Wed, 31 Jul 2024 09:54:12 +0000

LLMs and RAG systems have shown to be advantageous over time. They not only provide engaging discussions that deliver helpful information, but they also open up new avenues for tailored and intelligent applications, transforming areas ranging from customer service to scientific research. Despite their unique and powerful skills, there is evidence that they can produce plausible-sounding but inaccurate information, particularly when confronted with unclear questions or a lack of relevant data. Furthermore, they have demonstrated a lack of knowledge updates, causing them to occasionally present "old" information.

To mitigate those issues, the ability to connect to reliable and up-to-date resources is essential. Using an additional tool to retrieve external knowledge can help RAG and LLMs access up-to-date information, mitigating hallucinations and enhancing factual accuracy.

The Tavily Search API is suitable for that job. It is a search engine designed specifically for LLMs and RAG, with the goal of providing efficient, rapid, and permanent search results. Tavily specializes in improving search results for AI developers and autonomous AI agents. Furthermore, Tavily uses private financial, coding, news, and other internal data sources to supplement web content. As a result, Tavily empowers developers to build more accurate, insightful, and contextually aware AI applications.

We will talk about the Tavily Search API, diving into its functionalities and how it leverages AI for enhanced search. The structure of this writing is as follows:

Understanding the Power of Tavily Search API: A quick overview of the Tavily Search API, including why it is important and how it works.
Code in Action: Start with a basic code example showcasing a simple search query using Tavily.
Conclusion.

The link of the writing left below 👇👇👇