tj&D"UddlmZddlmZmZmZmZddlmZm Z ddl m Z m Z m Z ddlmZmZddlmZddlmZmZddlmZdd lmZmZmZdd lmZdd lmZdd l m!Z!dd l"m#Z#m$Z$ddl%m&Z&ddl'm(Z( ddl)m*Z*n #e+$re,Z*YnwxYw ddl)m-Z-n #e+$re,Z-YnwxYwe dZ.e/e/e0dfe1e0e ffZ2d*dZ3Gddee*Z4Gdde4e-Z5e6hdZ7de8d<e6hd Z9de8d!<d+d%Z:d,d(Z;d)S)-) annotations) AsyncIteratorCallableIteratorSequence)fields is_dataclass)AnyTypeVarcast)UUIDuuid4)BaseCallbackHandler) BaseMessage ToolMessage)convert_to_messages)ChatGenerationChatGenerationChunk LLMResult) BaseModel)filter_to_user_tags)NS_SEP) TAG_HIDDEN TAG_NOSTREAM) StreamChunk)Command)_StreamingCallbackHandler)_V2StreamingCallbackHandlerT.objr return Sequence[Any]c\ttr!tSttr fdt jDStr0tt sfdtDSdS)zSExtract top-level field values from a state object (dict, BaseModel, or dataclass).c0g|]}t|S)getattr).0kr s X/var/www/html/CA-Chatbot/venv/lib/python3.11/site-packages/langgraph/pregel/_messages.py z!_state_values..+s!@@@AQ@@@c:g|]}t|jSr%)r&name)r'fr s r)r*z!_state_values..-s%:::QV$$:::r+r%) isinstancedictlistvaluesrtype model_fieldsr r)r s`r) _state_valuesr5&s#t;CJJLL!!! C # #;@@@@c)?@@@@ c  ;:c4#8#8;::::fSkk:::: 2r+ceZdZdZdZ ddd=d Zddd>dZd?dZd@dZdAdZ dddd dBd,Z dddd-dCd2Z dd3dDd5Z dd3dEd8Z dddd dFd:Zdd3dGd;Zdd3dEd<ZdS)HStreamMessagesHandlerzA callback handler that implements stream_mode=messages. Collects messages from: (1) chat model stream events; and (2) node outputs. TN parent_nsstreamCallable[[StreamChunk], None] subgraphsboolr9tuple[str, ...] | Noner!Nonecd||_||_i|_t|_||_dS)aConfigure the handler to stream messages from LLMs and nodes. Args: stream: A callable that takes a StreamChunk and emits it. subgraphs: Whether to emit messages from subgraphs. parent_ns: The namespace where the handler was created. We keep track of this namespace to allow calls to subgraphs that were explicitly requested as a stream with `messages` mode configured. Example: parent_ns is used to handle scenarios where the subgraph is explicitly streamed with `stream_mode="messages"`. ```python def parent_graph_node(): # This node is in the parent graph. async for event in some_subgraph(..., stream_mode="messages"): do something with event # <-- these events will be emitted return ... parent_graph.invoke(subgraphs=False) ``` N)r:r<metadatasetseenr9)selfr:r<r9s r)__init__zStreamMessagesHandler.__init__<s/> "*, $'EE "r+FdedupemetaMetamessagerrGc|r|j|jvrdS|j tt|_|j|j||dd||dffdS)Nrmessages)idrCstrraddr:)rDrHrJrGs r)_emitzStreamMessagesHandler._emitasy  CgjDI-- Fz! \\ IMM'* % % % KKa*wQ.@A B B B B Br+responser ct|tr|||ddSt|tr4|D]/}t|tr|||d0dSt |D]w}t|tr|||d0t|tr2|D]/}t|tr|||d0xdS)NTrF)r/rrQrr5rDrHrRvalueitems r)_find_and_emit_messagesz-StreamMessagesHandler._find_and_emit_messagesjs0 h , , @ JJtXdJ 3 3 3 3 3 ( + + @! 9 9e[119JJtU4J888 9 9'x00 @ @e[11@JJtU4J8888x00@ %@@%dK88@ JJtT$J???  @ @r+run_idr outputAsyncIterator[T]c|SNr%rDrXrYs r)tap_output_aiterz&StreamMessagesHandler.tap_output_aiterzs  r+ Iterator[T]c|Sr\r%r]s r)tap_output_iterz%StreamMessagesHandler.tap_output_iters r+) parent_run_idtagsrA serializeddict[str, Any]rLlist[list[BaseMessage]]rb UUID | Nonerclist[str] | NonerAdict[str, Any] | Nonekwargsc J|r|r t|vrttt|dt dd}|js t|dkr ||jkrdSt|x} | |d<||f|j |<dSdSdS)Nlanggraph_checkpoint_nsrrc) rtupler rOsplitrr<lenr9rrA) rDrdrLrXrbrcrArjns filtered_tagss r)on_chat_model_startz)StreamMessagesHandler.on_chat_model_starts  3T 3l$&>&>tC*C!DEEKKFSSTTB> c"ggkkbDN6J6J!4T!:!:: G#0 %'NDM& ! ! ! 3 3&>&>r+chunkrbrctokenrOruChatGenerationChunk | Nonec t|tsdS|j|x}r|||jdSdSr\)r/rrAgetrQrJ)rDrvrurXrbrcrjrHs r)on_llm_new_tokenz&StreamMessagesHandler.on_llm_new_tokens_%!455  F=$$V,, ,4 , JJtU] + + + + + , ,r+rbrc &|j|x}rY|jrR|jdrE|jdd}t|tr|||jd|j|ddSNrTrF)rAry generationsr/rrQrJpoprDrRrXrbrjrHgens r) on_llm_endz StreamMessagesHandler.on_llm_ends=$$V,, ,4 ?# ?(22?JJtS[J>>> &$'''''r+error BaseExceptionc <|j|ddSr\rArrDrrXrbrjs r) on_llm_errorz"StreamMessagesHandler.on_llm_error" &$'''''r+inputsc |ra|d|dkr6|r t|vr,ttt|dt dd}|jst|dkrdS||f|j |<t|D]} t| tr'| j |j| j >t| t rUt| ts@| D]=} t| tr&| j |j| j >dSdSdSdS)Nr-langgraph_noderlrmr)ryrrnr rOrorr<rprAr5r/rrNrCrPr) rDrdrrXrbrcrArjrqrUrVs r)on_chain_startz$StreamMessagesHandler.on_chain_startsv  7 6""hll3C&D&DDDE't33tC*C!DEEKKFSSTTB> c"ggkk%'NDM& !&v.. 7 7e[117x+ eh///x007E39O9O7 %77%dK887#w2 $ dg 6 6 6' 7 7DD33 7 7r+c |j|dx}rt|tr|||jdSt|t rgtd|DrN|D]I}t|tr|||j3|||JdS|||dSdS)Nc3@K|]}t|tVdSr\)r/r)r'rUs r) z5StreamMessagesHandler.on_chain_end..s=88/4 5'**888888r+)rArr/rrWupdaterany)rDrRrXrbrjrHrUs r) on_chain_endz"StreamMessagesHandler.on_chain_ends =$$VT22 24 =(G,, =,,T8?CCCCCHh// =C888@88855 =&BBE!%11B44T5<HHHH44T5AAAA BB,,T8<<<<< = =r+c <|j|ddSr\rrs r)on_chain_errorz$StreamMessagesHandler.on_chain_errorrr+r:r;r<r=r9r>r!r?)rHrIrJrrGr=r!r?rHrIrRr r!r?)rXr rYrZr!rZ)rXr rYr_r!r_)rdrerLrfrXr rbrgrcrhrArirjr r!r rvrOrurwrXr rbrgrcrhrjr r!r rRrrXr rbrgrjr r!r rrrXr rbrgrjr r!r )rdrerrerXr rbrgrcrhrArirjr r!r ) rRr rXr rbrgrjr r!r )__name__ __module__ __qualname____doc__ run_inlinerErQrWr^rarsrzrrrrrr%r+r)r7r71sJX-1 ############JINCCCCCC@@@@  &*!%*.3333332-1%)!% , , , , , ,(&* ( ( ( ( ( ((&* (((((( &*!%*.777777J&* ======<&* ((((((((r+r7cteZdZdZddddd)dZddd*fdZd+dZddd,d!Zddd-fd$Zddd%d.d(Z xZ S)/StreamMessagesHandlerV2azv2 variant of `StreamMessagesHandler`. Declaring `_V2StreamingCallbackHandler` as a base flips `BaseChatModel.invoke` to route through `_stream_chat_model_events` (firing `on_stream_event`) instead of `_stream` (firing `on_llm_new_token`). Inherits `on_stream_event` from the parent, which forwards protocol events onto the messages stream channel. Pregel attaches this class instead of the v1 handler only when `StreamingHandler` opts in via the internal `CONFIG_KEY_STREAM_MESSAGES_V2` config key; direct `graph.stream(stream_mode="messages")` callers keep the v1 AIMessageChunk shape. NrtrvrOrurwrXr rbrgrcrhrjr r!c dS)uIntentional no-op — v1 chunks are not used on v2-flagged runs. The v2 marker already steers `invoke` to the event generator, so `on_llm_new_token` should not fire under normal routing. This override stays a pass-through (no call to `super()`) to make the intent explicit and to guard against any caller (e.g. a node that calls `model.stream()` directly, which still fires the v1 callback) leaking AIMessageChunks onto a v2-flagged messages stream. Nr%)rDrvrurXrbrcrjs r)rzz(StreamMessagesHandlerV2.on_llm_new_tokensr+r8r:r;r<r=r9r>r?cvt|||t|_dS)Nr8)superrErB_streamed_run_ids)rDr:r<r9 __class__s r)rEz StreamMessagesHandlerV2.__init__)s6 i@@@,/EEr+rHrIrRct|tr/t|ts|||ddSt|trI|D]D}t|tr-t|ts|||dEdSt |D]}t|tr.t|ts|||dEt|trG|D]D}t|tr-t|ts|||dEdS)acLike the v1 handler, but skip ToolMessage from node outputs. Tool results belong on the tools channel / state in v3; v2-flagged streams must not replay finalized ToolMessages as chat tokens (see MessagesTransformer). Legacy v1-only `stream_mode="messages"` still emits ToolMessages (see subgraph streaming tests). TrFN)r/rrrQrr5rTs r)rWz/StreamMessagesHandlerV2._find_and_emit_messages3s h , , @Z+5V5V @ JJtXdJ 3 3 3 3 3 ( + + @! 9 9e[119*;;;9JJtU4J888  9 9 'x00 @ @e[11 @*;;; @JJtU4J8888x00@ %@@%dK88@ +BB@!JJtT$J??? @ @r+r{rc |j|x}r|jr|jdr|jdd}t|tr|||jvrV|jj%tt|j_|j |jjn| ||jd|j ||j|ddSr})rAryr~r/rrrJrNrOrrCrPrQdiscardrrs r)rz"StreamMessagesHandlerV2.on_llm_endPs=$$V,, ,4 C# C(22C!777;>1-0\\CKN ckn5555 4T BBB &&v... &$'''''r+rrc v|j|tj|f||d|dS)N)rXrb)rrrr)rDrrXrbrjrs r)rz$StreamMessagesHandlerV2.on_llm_errores^ &&v...  '        r+)rbrceventrec |j|x}r|ddkrK|j||d}|r|j|i|ddt |i}||dd||ffdSdS) uForward a protocol event from `stream_events(version="v3")` as a messages stream part. Fires once per `MessagesData` event (`message-start`, per-block `content-block-*`, `message-finish`). The transformer layer correlates events back to a single `ChatModelStream` via `metadata["run_id"]` — attached here so the v1 `stream_mode="messages"` output (which emits `(AIMessageChunk, metadata)` via `on_llm_new_token`) keeps its original metadata shape. Lives on the v2 handler rather than the v1 base: content-block events are a v2-only concept, and forwarding them only when the v2 handler is attached keeps the message channel's shape predictable for v1 callers. rz message-start message_idrMrXrrLN)rAryrrPrCrOr:) rDrrXrbrcrjrHmsg_idv2_metas r)on_stream_eventz'StreamMessagesHandlerV2.on_stream_eventus0=$$V,, ,4 A yy!!_44&**6222<00*IMM&)))8a8(CKK88G KKa*ug.>? @ @ @ @ @ A Ar+rrrrr) rrerXr rbrgrcrhrjr r!r ) rrrrrzrErWrrr __classcell__)rs@r)rrs  &-1%)!%      6-1 22222222@@@@D&* ((((((4&*         *&*!% #A#A#A#A#A#A#A#A#A#Ar+r>aitooluserhumansystemfunction assistantzfrozenset[str]_MESSAGE_ROLES>rrrremoverr_MESSAGE_TYPESrVr0r=cr|dtvp|dtvS)Nroler3)ryrr)rVs r)_is_message_dictrs/ 88F  ~ - S&1A1A^1SSr+rUr?ct|tr+|j"tt |_dSdSt|t rFt |r7|ds tt |d<dSdSt|trt|D]\}}t|tr(|j tt |_Bt|t rQt |rBt|gd}|j tt |_|||<dSdS)aACoerce message-like write values to typed BaseMessages with stable IDs. Called in put_writes() before DeltaChannel writes are submitted to the checkpointer. Without this the checkpoint may store raw dicts or id=None BaseMessages; every get_state() replay then produces a different UUID and the same message appears with a different ID in each LangSmith trace. Handles three input shapes: - BaseMessage objects: assign a UUID if id is None. - Dicts with a known "role" (OpenAI-style) or "type" (LangChain format) at the root level: stamp "id" into the dict in-place. The reducer's convert_to_messages call will forward the id to the resulting BaseMessage. - Lists of the above: apply the same logic to each element, replacing dict items with coerced BaseMessages so the shared list reference seen by checkpoint_pending_writes and the background thread both get typed messages. Mutating synchronously here (before the background thread is submitted) is safe: the serialised bytes always reflect the post-coercion state. NrNr) r/rrNrOrr0rryr1 enumerater)rUirVmsgs r)ensure_message_idsrsT(%%% 8 577||EHHH   E4 %5e%<%< yy 'egg,,E$KKK ' ' E4  ''  GAt$ ,, 7?!%''llDGD$'' , \\CFa    r+N)r r r!r")rVr0r!r=)rUr r!r?)< __future__rcollections.abcrrrr dataclassesrr typingr r r uuidr rlangchain_core.callbacksrlangchain_core.messagesrrlangchain_core.messages.utilsrlangchain_core.outputsrrrpydanticrlanggraph._internal._configrlanggraph._internal._constantsrlanggraph.constantsrrlanggraph.pregel.protocolrlanggraph.typesr!langchain_core.tracers._streamingr ImportErrorobjectrrrnrOr0rIr5r7r frozensetr__annotations__rrrr%r+r)rs>"""""""GGGGGGGGGGGG,,,,,,,, 888888<<<<<<<<======QQQQQQQQQQ;;;;;;11111188888888111111######'KKKKKKK''' &')MMMMMMM)))"() GCLL U38_d38n ,-O(O(O(O(O(/1JO(O(O(dUAUAUAUAUA35PUAUAUAv"+FFF"""+;;;"" TTTT######s$3A::BBBBB