Metadata-Version: 2.4
Name: fluxmend
Version: 0.1.0
Summary: Streaming FSM validation + layered repair for LLM outputs with embedded structured components
Author: fluxmend Developers
License: Apache-2.0
Project-URL: Homepage, https://github.com/luvrix/fluxmend
Project-URL: Repository, https://github.com/luvrix/fluxmend
Project-URL: Documentation, https://github.com/luvrix/fluxmend#readme
Project-URL: Issues, https://github.com/luvrix/fluxmend/issues
Project-URL: Changelog, https://github.com/luvrix/fluxmend/releases
Keywords: llm,streaming,fsm,structured-output,validation,repair,json,pydantic,schema-aware,openai,anthropic
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: <3.14,>=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pydantic>=2.0
Requires-Dist: typing_extensions>=4.7
Requires-Dist: json-repair>=0.30
Provides-Extra: cfg
Requires-Dist: lark>=1.1; extra == "cfg"
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: mypy>=1.0; extra == "dev"
Requires-Dist: ruff>=0.1; extra == "dev"
Requires-Dist: pre-commit; extra == "dev"
Requires-Dist: beartype; extra == "dev"
Dynamic: license-file

# fluxmend

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

[![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue)](https://www.python.org/)
[![License: Apache-2.0](https://img.shields.io/badge/license-Apache--2.0-blue)](LICENSE)
[![Tests](https://img.shields.io/badge/tests-310%2B-green)](#testing)
[![mypy strict](https://img.shields.io/badge/mypy-strict-blue)](#testing)
[![ruff](https://img.shields.io/badge/ruff-checked-green)](#testing)

> 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

```python
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

![fluxmend architecture](docs/architecture.png)

**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](#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

```bash
pip install fluxmend
```

Optional extras:

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

## Usage

### Fluxmend (recommended)

```python
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:

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

### With LLM Repair

```python
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.

```python
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>`:

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

### @structured decorator (optional)

```python
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

```python
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.

```python
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

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

## License

Apache-2.0
