ޠj\ SSKJr SSKrSSKrSSKJrJr SSKJ r J r SSK J r SSK Jr SSKJr SSKJr SS KJrJr SS KJrJrJr SS KJrJrJrJr S r"S S\ \\\ \ \ 45r g)) annotationsN)CallableSequence)AnyGeneric) PendingWrite)_DeltaSnapshot)SelfMISSING) BaseChannelValue)_get_overwrite_operators_equal _strip_extras)EmptyChannelError ErrorCodeInvalidUpdateErrorcreate_error_message) DeltaChannelc^\rSrSr%SrSrS\S'SSS.SU4SjjjjrSS jr\ SS j5r \ SS j5r SS jr SS jr SSjrSSjrSSjrSSjrSSjrSrU=r$)ragReducer channel that stores only a sentinel in checkpoint blobs and reconstructs state by replaying ancestor writes through the reducer. !!! warning "Beta" `DeltaChannel` is in beta. The API and on-disk representation may change in future releases. Threads written with `DeltaChannel` today are expected to remain readable, but the surrounding contract (`BaseCheckpointSaver.get_delta_channel_history`, the `_DeltaSnapshot` blob shape, the `counters_since_delta_snapshot` metadata field) is not yet stable. The reducer receives the current accumulated value and a batch of writes in one call: `reducer(state, [write1, write2, ...]) -> new_state`. Reducers must be deterministic and batching-invariant (associative across folds): applying two consecutive write batches separately must produce the same state as applying their concatenation once: reducer(reducer(state, xs), ys) == reducer(state, xs + ys) This lets LangGraph replay checkpointed writes in larger batches than they were originally produced without changing reconstructed state. Snapshot cadence is driven by two counters: per-channel update count and total supersteps since last snapshot. `create_checkpoint` writes a full `_DeltaSnapshot` blob when EITHER the update count reaches `snapshot_frequency` OR the supersteps count reaches the system-wide `DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT` bound (default 5000), bounding replay depth even for channels that stop receiving writes. Parameters: reducer: `(state, list[writes]) -> new_state`. Must be deterministic and batching-invariant as described above. typ: The value type (e.g. `list`, `dict`). Inferred automatically from the outer type when used inside `Annotated[T, DeltaChannel(...)]`. snapshot_frequency: Every Nth update to this channel writes a snapshot blob (default `1000`). Must be a positive int. )valuereducersnapshot_frequencyz Value | Anyrirc">US::a[SU35eUc[n[TU] U5 XlX0l[ U5nU[RR[RR4;a[nU[RR[RR4;a[nU[RR[RR4;a[ nX l[$Ulg)Nrz/snapshot_frequency must be a positive int, got ) ValueErrorlistsuper__init__rrr collectionsabcrMutableSequenceSet MutableSetsetMappingMutableMappingdicttypr r)selfrr+r __class__s R/var/www/html/gaurav/venv/lib/python3.13/site-packages/langgraph/channels/delta.pyr!DeltaChannel.__init__Es  "ABTAUV  ;C  "4C  ;??++[__-L-LM MC ;??&& (B(BC CC ;??**KOO,J,JK KC! c[U[5(dgURUR:wag[URUR5$)NF) isinstancerrrr)r,others r.__eq__DeltaChannel.__eq___s>%..  " "e&>&> > emm<CZZ =S  r0c"URURURURS9nURUlU[ LaUR5UlU$[U[5(aUR UlU$XlU$)zInitialize from a stored blob. Blob types: * `MISSING`: start empty; caller replays writes. * `_DeltaSnapshot(value)`: restore value directly from snapshot. * plain value (migration from old `BinaryOperatorAggregate` blobs): use directly. r) r-rr+rrAr rr2r )r, checkpointrDs r.from_checkpointDeltaChannel.from_checkpointvsnn LL$((t7N7N ((   CI   N 3 3"((CI #I r0cjUVVs/sHu p#UPM nnnU(dgURnSn[U5HIups[U5upU(dMU b[R"U 5OUR 5nUS-nMK XFSn U (aUR XZ5UlgUUlgs snnf)zApply ancestor writes oldest-to-newest via a single reducer call. If any write is an Overwrite, the last one in the sequence acts as the reset point: its value becomes the new base and only writes after it are passed to the reducer. Nr)r enumeraterrBrCr+r) r,writes_vvaluesbasestartiis_owow_value remainings r. replay_writesDeltaChannel.replay_writess$**61!6* zzf%DA,Q/OEu/7/Cuzz(+A & 6N 6?T\\$2 T +sB/cU(dgSn[U5HCup4[U5upVU(dMUb#[S[RS9n[ U5eUnME Ub>[X5uphUb[ R"U5OUR5Ul gUR[LaUR5O URn URU [U55Ul g)NFz4Can receive only one Overwrite value per super-step.)message error_codeT) rLrrrINVALID_CONCURRENT_GRAPH_UPDATErrBrCr+rr rr) r,rP overwrite_idxrSrOrTrNmsgoverwrite_valuerQs r.updateDeltaChannel.updates$( f%DA%a(HEu ,. V#,#L#LC-S11 ! &  $!/0E!F A#. ?+XXZ J !ZZ72txxz \\$V 5 r0cTUR[La [5eUR$r7)rr rr9s r.getDeltaChannel.gets! :: #% %zzr0c&UR[L$r7)rr r9s r. is_availableDeltaChannel.is_availableszz((r0c[$)a_Return stored representation: always `MISSING`. Snapshot decisions live in `create_checkpoint` (which has the channel version) and write `_DeltaSnapshot(ch.get())` directly into `channel_values`. For non-snapshot steps the channel does not appear in `channel_values`; reconstruction walks ancestor writes via the saver's `get_delta_channel_history`. r r9s r.rGDeltaChannel.checkpoints r0)rrr+rr7)rz#Callable[[Any, Sequence[Any]], Any]r+ztype[Value] | NonerintreturnNone)r3objectrkbool)rkr)rkr )rGrrkr )rMzSequence[PendingWrite]rkrl)rPz Sequence[Any]rkrn)rkrn)__name__ __module__ __qualname____firstlineno____doc__ __slots____annotations__r!r4propertyr:r>rCrHrWr`rcrfrG__static_attributes__ __classcell__)r-s@r.rrs&P;I  #'" #' "4" "  "  ""4=*J(4 )  r0r)! __future__rcollections.abcr"rCrBrrtypingrrlanggraph.checkpoint.baser langgraph.checkpoint.serde.typesr typing_extensionsr langgraph._internal._typingr langgraph.channels.baser rlanggraph.channels.binoprrrlanggraph.errorsrrrr__all__rr0r.rsY".2;"/6TT q75>;sC}#=qr0