LangDiff provides two main modules for streaming structured data and tracking changes:
The parser module contains streaming-aware data types and the core parser for processing token streams.
Object- Represents a streaming JSON objectList- Represents a streaming JSON arrayString- Represents a streaming string valueAtom- Represents atomic values (numbers, booleans, null)Parser- Processes token streams and triggers callbacks
- Event Callbacks: All streaming types support
on_startandon_completecallbacks. Additionally,Objectsupportson_update, whileListandStringsupporton_append - Type Safety: Full type hints and generic support for compile-time checking
- Pydantic Integration: Convert streaming models to Pydantic models via
to_pydantic()
The tracker module provides change tracking capabilities for generating JSON Patch diffs.
ChangeTracker- Abstract base for change trackingJSONPatchChangeTracker- Standard JSON Patch trackingEfficientJSONPatchChangeTracker- Enhanced tracking withappendoperations
track_change()- Wrap objects for automatic change trackingapply_change()- Apply JSON Patch operations to objects
import langdiff as ld
# Define schema
class Response(ld.Object):
title: ld.String
items: ld.List[ld.String]
# Set up callbacks
response = Response()
@response.title.on_append
def on_title_chunk(chunk: str):
print(f"Title: {chunk}")
# Parse stream
with ld.Parser(response) as parser:
for token in stream:
parser.push(token)# Track changes to any object
obj, diff_buf = ld.track_change(UI(items=[]))
# Make modifications
obj.items.append("new item")
# Get JSON Patch operations
changes = diff_buf.flush()