fluxmend

Streaming FSM validation + layered repair for LLM outputs with embedded structured components.

LLM structured output · streaming JSON validation · XML/CFG/regex FSM · Pydantic · schema-aware repair · LLM-as-repairer · token-stream parser · char-level validation · OpenAI / Anthropic compatible

Why

When an LLM streams text with embedded structured blocks (JSON inside <shop>...</shop> tags, XML config, regex-validated fields), the structured parts must be valid — but LLMs emit malformed JSON more often than vendors admit: missing braces, trailing commas, True instead of true, field-name typos, wrong types. Waiting until the stream ends to validate means discovering errors too late, and naive json.loads on partial buffers fails on every token boundary.

fluxmend validates each character as it streams — no buffering until close — and repairs common errors in three layers (rule-based → json-repair library → LLM-as-repairer), emitting audit events at every step. Markdown around the components passes through untouched.

Features

• Char-level streaming validation — FSM-driven, no full-buffer parse needed
• Layered repair — Local rules → json-repair → LLM Repair → Checkpoint Rollback
• Multi-format — JSON, XML, regex, CFG (lark), custom FSM plugins
• Schema-aware — Pydantic classes, JSON Schema dicts, or programmatic DSL
• Per-tag handlers — transform each parsed instance with a user function
• Async repair mode — defer LLM Repair to background, non-blocking streams
• Token-boundary safe — repairs deferred until </tag> close to avoid double brackets
• OpenAI / Anthropic compatible — wrap any SDK via the LLMClient Protocol
• Typed and tested — mypy strict, 310+ tests, ruff clean

Quick start

from pydantic import BaseModel
from fluxmend import Fluxmend


class Shop(BaseModel):
    id: int
    name: str


with Fluxmend(schemas=[("shop", Shop)]) as guard:
    for chunk in llm_stream("Recommend a shop"):
        for event in guard.feed(chunk):
            if event.type == "text":
                print(event.content, end="")
            elif event.type == "repair_applied":
                r = event.content
                print(f"\n[repair layer={r.layer}] {r.original!r} -> {r.repaired!r}")

result = guard.result  # {"shop": [Shop(id=101, name="test")]}

Works with any text source — agno, pydantic-ai, langgraph, raw LLM API, or plain text.

Architecture

Two layers:

• Enhancement Layer (optional) — repairs broken open tags (shop> → <shop>) and tag-name typos (<shp> → <shop>) before the core layer sees them
• Core Layer — DetectionFSM recognizes tag boundaries, GrammarValidator runs char-level FSM per format, layered repair kicks in on errors

Layered repair runs in two phases (see Layered repair below for details): fast schema-aware rules during streaming, full chain (json-repair → LLM Repair → rollback) on </tag> close.

Layered repair

During streaming (no bracket insertion — token boundaries haven't closed yet):

1. Schema-aware Local Repair — bool/null case (True → true), type inference ("42" → 42)
2. Field-name correction — edit-distance-1 match against schema properties (nearby_poi → nearby_pois)

On </tag> close (safe to insert brackets):

3. json-repair — library-based syntax repair (missing brackets, trailing commas, quotes)
4. LLM Repair — asks an LLM to rewrite the broken component with schema context
5. Fallback — emit original text, component_end verified=False

Install

pip install fluxmend

Optional extras:

pip install "fluxmend[cfg]"    # lark-based CFG support
pip install "fluxmend[dev]"    # pytest, mypy, ruff, pre-commit

Usage

Fluxmend (recommended)

from fluxmend import Fluxmend

guard = Fluxmend(schemas=[("shop", Shop), ("map", MapMark)])

for chunk in text_stream:
    for event in guard.feed(chunk):
        ...

result = guard.close()  # {"shop": [Shop(...)], "map": [MapMark(...)]}

Supports with statement:

with Fluxmend(schemas=[...]) as guard:
    guard.feed(chunk)
result = guard.result

With LLM Repair

from fluxmend.llm import OpenAICompatibleClient
from openai import OpenAI

guard = Fluxmend(
    schemas=[("shop", Shop)],
    try_times=2,
    llm_client=OpenAICompatibleClient(OpenAI(api_key=...)),
)

Instance pool (multi-request / production)

For web frameworks (FastAPI, Flask, Django), use FluxmendPool to pre-create instances and reuse them across requests. Grammars (compiled JSON Schemas) are created once at pool init and reused — no per-request schema compilation overhead.

from fluxmend import FluxmendPool

# Init once at startup
pool = FluxmendPool(
    schemas=[("shop", Shop), ("map", MapMark)],
    try_times=2,
    llm_client=client,
    handlers={"shop": process_shop},
    pool_size=10,
)

# Per request (sync)
with pool.acquire() as guard:
    for chunk in text_stream:
        guard.feed(chunk)
    result = guard.close()

# Per request (async — FastAPI / Starlette)
async with pool.aacquire() as guard:
    for chunk in text_stream:
        await guard.afeed(chunk)
    result = await guard.aclose()

Thread-safe: queue.Queue handles checkout/return. Each checkout gets a reset() instance — no cross-request state leaks.

Async repair mode (non-blocking)

Defer LLM Repair to a background executor so multi-component streams don't block on each </tag>:

guard = Fluxmend(
    schemas=[("shop", Shop), ("map", MapMark)],
    try_times=2,
    llm_client=client,
    async_repair=True,  # ← component_end emits "pending" immediately
)

@structured decorator (optional)

from fluxmend import structured

@structured(Metric, tag="metric")
def generate(prompt: str):
    for chunk in agent.run(prompt, stream=True):
        yield chunk.content

metric = generate.collect("report latency")

Multi-component

guard = Fluxmend(schemas=[("shop", Shop), ("map", MapMark)])
# Automatically detects <shop> and <map> tags in the stream

Per-tag handlers

Pass a handlers dict to transform each parsed instance. The handler receives a dict (BaseModel instances are dumped via model_dump(); dict schemas pass through as-is). Its return value is stored in result[tag] instead of the raw instance. Handler exceptions are swallowed: the failed instance is stored as "" (empty string) so result[tag] length matches the component count (zip-safe for frontends), and a handler_error event is emitted with {"tag": str, "error": str} content for diagnosis.

def process_shop(shop: dict) -> dict:
    return {"id": shop["id"], "name_upper": shop["name"].upper()}

guard = Fluxmend(
    schemas=[("shop", Shop), ("map", MapMark)],
    handlers={"shop": process_shop},  # map keeps default behavior
)
result = guard.close()
# result["shop"] = [{"id": 1, "name_upper": "..."}, ...]  # handler results
# result["map"]  = [MapMark(...), ...]                    # raw instances

# Check handler failures via events:
errors = [e for e in guard.events if e.type == "handler_error"]
# errors[i].content == {"tag": "shop", "error": "some error message"}

Format plugins

Format	Schema input	FSM	Repair
json	JSON Schema / Pydantic class / DSL Term	hand-written pushdown automaton	json-repair + schema-aware
xml	XSD string	hand-written stack-based FSM	rule-based whitelists
regex	pattern string	re-backed permissive streamer	LLM only
cfg	lark grammar string	lark LALR(1)	LLM only
custom	pre-compiled FSM instance	user-supplied	LLM only

Use cases

• Agent frameworks — validate tool-call JSON streaming from Claude/GPT before passing to tools
• RAG pipelines — repair malformed metadata blocks embedded in markdown responses
• Chat UIs — display verified text in real-time, defer repair events for logging
• Batch eval — run 1000s of LLM calls, count repair rates, audit failures via events
• Multi-modal streams — mix JSON + XML + free text in one stream, validate each by tag

Testing

pytest tests/ -v              # 310+ tests
mypy src/fluxmend             # strict type checks
ruff check src/fluxmend tests

License

Apache-2.0