langchain-hs-graph 0.0.6.0 → 0.0.6.2

raw patch · 5 files changed

+300/−11 lines, 5 filesdep +conduitdep ~langchain-hs-corePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: conduit

Dependency ranges changed: langchain-hs-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +8/−0
• README.md +1/−1
• langchain-hs-graph.cabal +5/−3
• src/Langchain/Graph/StateGraph.hs +152/−7
• test/Test/Langchain/Graph/StateGraphSpec.hs +134/−0

@@ -2,6 +2,14 @@  All notable changes to this package will be documented in this file. +## 0.0.6.2 - 2026-11-14++> **Note**: `0.0.6.1` was skipped globally after `langchain-hs-ollama-v0.0.6.1`.++- Coordinated monorepo release alignment to `0.0.6.2`.+- Tightened `langchain-hs-core` dependency bound to `>= 0.0.6.2 && < 0.0.7`.+- Added StateGraph event streaming via `streamGraphEvents` with typed `GraphStreamEvent` constructors (`GraphStart`, `GraphNodeStart`, `GraphNodeChunk`, `GraphNodeEnd`, `GraphStateUpdate`, `GraphEnd`).+ ## 0.0.6.0 - 2026-10-02  - Updated bounds for GHC 9.12 and modern Stackage LTS releases.

@@ -18,7 +18,7 @@ ## Installation  ```cabal-build-depends: langchain-hs-graph >= 0.0.6 && < 0.0.7+build-depends: langchain-hs-graph >= 0.0.6.2 && < 0.0.7 ```  ## License

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           langchain-hs-graph-version:        0.0.6.0+version:        0.0.6.2 synopsis:       Graph-based stateful agent orchestration engine for langchain-hs description:    StateGraph, StateReducer, Checkpointer, and HITL interrupt/resume workflow patterns. category:       Web, AI, Control@@ -55,8 +55,9 @@       aeson >=2.0 && <3     , base >=4.17 && <5     , bytestring >=0.10 && <0.13+    , conduit ==1.3.*     , containers >=0.6 && <0.9-    , langchain-hs-core ==0.0.6.*+    , langchain-hs-core >=0.0.6.2 && <0.0.7     , mtl >=2.2 && <2.4     , sqlite-simple >=0.4.18 && <0.5     , stm ==2.5.*@@ -92,9 +93,10 @@     , aeson >=2.0 && <3     , base >=4.17 && <5     , bytestring >=0.10 && <0.13+    , conduit ==1.3.*     , containers >=0.6 && <0.9     , filepath-    , langchain-hs-core ==0.0.6.*+    , langchain-hs-core >=0.0.6.2 && <0.0.7     , langchain-hs-graph     , mtl >=2.2 && <2.4     , sqlite-simple >=0.4.18 && <0.5

@@ -49,20 +49,33 @@   , addChatNode   , addChatNodeWithConfig   , addSubGraphNode+  , addStreamingNode+  , addStreamingModelNode+  , addStreamingModelNodeWithConfig      -- * Compilation & Execution   , compileGraph   , runGraph   , runGraphWithConfig +    -- * Streaming Execution+  , streamGraph+  , streamGraphWithConfig+  , streamGraphWith+  , streamGraphStream+     -- * Pure State Reducers   , appendMessagesReducer   , replaceFieldReducer   ) where -import Control.Monad (when)-import Control.Monad.Except (ExceptT, MonadError, runExceptT, throwError)+import Control.Monad (void, when)+import Control.Monad.Except (ExceptT, MonadError, catchError, runExceptT, throwError) import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans (lift)+import Data.Aeson (ToJSON, Value, toJSON)+import Data.Conduit (ConduitT, fuseUpstream, runConduit, yield, (.|))+import qualified Data.Conduit.List as CL import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map import Data.Text (Text)@@ -70,8 +83,14 @@  import Langchain.Error (LangchainError, agentError, internalError) import Langchain.Model (ChatModel, Message, ModelConfig, extractMessageText, userMessage)-import qualified Langchain.Model as M (invoke)+import qualified Langchain.Model as M (invoke, stream) import Langchain.Runnable (Runnable (..), RunnableTree, interpret)+import Langchain.Stream+  ( EventStream+  , StreamEvent (NodeEnd, NodeStart)+  , liftChatStream+  , tapStreamMessage+  )  -- | Unique node identifier in a StateGraph type NodeId = Text@@ -120,6 +139,7 @@ -- | Declarative StateGraph specification data StateGraph s m = StateGraph   { graphNodes :: !(Map NodeId (Node s m))+  , graphStreamNodes :: !(Map NodeId (s -> ConduitT () StreamEvent m (Either LangchainError s)))   , graphEdges :: !(Map NodeId (Edge s m))   , graphReducer :: !(StateReducer s)   , graphConfig :: !GraphConfig@@ -128,6 +148,7 @@ -- | A validated, compiled StateGraph ready for execution data CompiledGraph s m = CompiledGraph   { compiledNodes :: !(Map NodeId (Node s m))+  , compiledStreamNodes :: !(Map NodeId (s -> ConduitT () StreamEvent m (Either LangchainError s)))   , compiledEdges :: !(Map NodeId (Edge s m))   , compiledReducer :: !(StateReducer s)   , compiledConfig :: !GraphConfig@@ -138,6 +159,7 @@ emptyStateGraph reducer =   StateGraph     { graphNodes = Map.empty+    , graphStreamNodes = Map.empty     , graphEdges = Map.empty     , graphReducer = reducer     , graphConfig = defaultGraphConfig@@ -282,6 +304,65 @@ addConditionalEdge fromNode condFn g =   g {graphEdges = Map.insert fromNode (ConditionalEdge condFn) (graphEdges g)} +-- | Reserved run identifier for streamed graph lifecycle events+graphRunId :: Text+graphRunId = "state-graph-run"++-- | Add a streaming node capable of emitting inner 'StreamEvent's during 'streamGraph'.+addStreamingNode ::+  Monad m =>+  NodeId ->+  (s -> ConduitT () StreamEvent m (Either LangchainError s)) ->+  StateGraph s m ->+  StateGraph s m+addStreamingNode name streamAct g =+  let syncAction state = runConduit (streamAct state `fuseUpstream` CL.sinkNull)+      n = Node {nodeId = name, nodeAction = syncAction}+   in g+        { graphNodes = Map.insert name n (graphNodes g)+        , graphStreamNodes = Map.insert name streamAct (graphStreamNodes g)+        }++addStreamingModelNodeMaybeConfig ::+  (ChatModel c, MonadIO m, MonadError LangchainError m) =>+  NodeId ->+  c ->+  Maybe (ModelConfig c) ->+  (s -> [Message]) ->+  (s -> Message -> s) ->+  StateGraph s m ->+  StateGraph s m+addStreamingModelNodeMaybeConfig name chatModel mbCfg toMessages mergeMessage =+  addStreamingNode name $ \state ->+    ( do+        finalMsg <- liftChatStream (M.stream chatModel (toMessages state) mbCfg) .| tapStreamMessage+        pure $ Right (mergeMessage state finalMsg)+    )+      `catchError` (pure . Left)++-- | Add a streaming ChatModel node that forwards inner 'LLMStart', 'LLMChunk', and 'LLMEnd' events.+addStreamingModelNode ::+  (ChatModel c, MonadIO m, MonadError LangchainError m) =>+  NodeId ->+  c ->+  (s -> [Message]) ->+  (s -> Message -> s) ->+  StateGraph s m ->+  StateGraph s m+addStreamingModelNode name chatModel = addStreamingModelNodeMaybeConfig name chatModel Nothing++-- | Add a streaming ChatModel node with explicit model configuration.+addStreamingModelNodeWithConfig ::+  (ChatModel c, MonadIO m, MonadError LangchainError m) =>+  NodeId ->+  c ->+  ModelConfig c ->+  (s -> [Message]) ->+  (s -> Message -> s) ->+  StateGraph s m ->+  StateGraph s m+addStreamingModelNodeWithConfig name chatModel cfg = addStreamingModelNodeMaybeConfig name chatModel (Just cfg)+ -- | Validate state graph invariants and compile to an executable 'CompiledGraph' compileGraph :: StateGraph s m -> Either LangchainError (CompiledGraph s m) compileGraph StateGraph {..}@@ -291,11 +372,51 @@       Right         CompiledGraph           { compiledNodes = graphNodes+          , compiledStreamNodes = graphStreamNodes           , compiledEdges = graphEdges           , compiledReducer = graphReducer           , compiledConfig = graphConfig           } +-- | Stream graph execution events ('NodeStart', inner node events, and 'NodeEnd').+streamGraph ::+  (ToJSON s, MonadIO m, MonadError LangchainError m) =>+  CompiledGraph s m ->+  NodeId ->+  s ->+  ConduitT () StreamEvent m s+streamGraph cg = streamGraphWithConfig (compiledConfig cg) cg++-- | Stream graph execution events with an explicit 'GraphConfig' override.+streamGraphWithConfig ::+  (ToJSON s, MonadIO m, MonadError LangchainError m) =>+  GraphConfig ->+  CompiledGraph s m ->+  NodeId ->+  s ->+  ConduitT () StreamEvent m s+streamGraphWithConfig = streamGraphWith toJSON++-- | Stream graph execution events using a custom state summary function (supports non-'ToJSON' states).+streamGraphWith ::+  (MonadIO m, MonadError LangchainError m) =>+  (s -> Value) ->+  GraphConfig ->+  CompiledGraph s m ->+  NodeId ->+  s ->+  ConduitT () StreamEvent m s+streamGraphWith = executeGraphWith True++-- | Stream graph execution as an 'EventStream m' discarding the final state.+streamGraphStream ::+  (ToJSON s, MonadIO m, MonadError LangchainError m) =>+  CompiledGraph s m ->+  NodeId ->+  s ->+  EventStream m+streamGraphStream cg entry s = void (streamGraph cg entry s)+ -- | Execute a CompiledGraph starting from the given entry NodeId runGraph ::   (MonadIO m, MonadError LangchainError m) =>@@ -313,7 +434,20 @@   NodeId ->   s ->   m s-runGraphWithConfig cfg cg =+runGraphWithConfig cfg cg entry initState =+  runConduit $+    executeGraphWith False (const (toJSON ())) cfg cg entry initState `fuseUpstream` CL.sinkNull++executeGraphWith ::+  (MonadIO m, MonadError LangchainError m) =>+  Bool ->+  (s -> Value) ->+  GraphConfig ->+  CompiledGraph s m ->+  NodeId ->+  s ->+  ConduitT () StreamEvent m s+executeGraphWith emitEvents encodeState cfg cg =   executeStep 0   where     CompiledGraph {..} = cg@@ -322,6 +456,15 @@     logMsg msg =       when logStepTransitions $ liftIO $ stepLogger msg +    emitNodeEvent mkEv nid st =+      when emitEvents $ yield (mkEv graphRunId nid (encodeState st))++    runNodeAction currentId node state+      | emitEvents = case Map.lookup currentId compiledStreamNodes of+          Just streamAct -> streamAct state+          Nothing -> lift (nodeAction node state)+      | otherwise = lift (nodeAction node state)+     executeStep stepCount currentId state       | stepCount >= recursionLimit = do           let errMsg =@@ -339,7 +482,8 @@           Just node -> do             logMsg $               "[StateGraph] Step " <> T.pack (show (stepCount + 1)) <> ": Entering node '" <> currentId <> "'"-            eNextState <- nodeAction node state+            emitNodeEvent NodeStart currentId state+            eNextState <- runNodeAction currentId node state             case eNextState of               Left err -> do                 logMsg $ "[StateGraph ERROR] Node '" <> currentId <> "' failed: " <> T.pack (show err)@@ -347,6 +491,7 @@               Right stepState -> do                 let mergedState = compiledReducer state stepState                 logMsg $ "[StateGraph] Node '" <> currentId <> "' executed successfully"+                emitNodeEvent NodeEnd currentId mergedState                 case Map.lookup currentId compiledEdges of                   Nothing -> do                     logMsg $ "[StateGraph] No outgoing edge from '" <> currentId <> "', terminating"@@ -355,7 +500,7 @@                     logMsg $ "[StateGraph] Static edge: '" <> currentId <> "' -> '" <> nextId <> "'"                     executeStep (stepCount + 1) nextId mergedState                   Just (ConditionalEdge cond) -> do-                    eNextId <- cond mergedState+                    eNextId <- lift $ cond mergedState                     case eNextId of                       Left err -> do                         logMsg $@@ -369,7 +514,7 @@               logMsg $ "[StateGraph] Passthrough edge: '" <> currentId <> "' -> '" <> nextId <> "'"               executeStep (stepCount + 1) nextId state             Just (ConditionalEdge cond) -> do-              eNextId <- cond state+              eNextId <- lift $ cond state               case eNextId of                 Left err -> throwError err                 Right nextId -> do

@@ -1,17 +1,38 @@ {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TypeFamilies #-}  module Test.Langchain.Graph.StateGraphSpec (tests) where  import Control.Concurrent.STM (atomically, modifyTVar', newTVarIO, readTVarIO) import Control.Monad.Except (ExceptT, runExceptT)+import Control.Monad.IO.Class (liftIO)+import Data.Conduit (fuseUpstream, runConduit, yield)+import qualified Data.Conduit.List as CL+import Data.IORef (modifyIORef', newIORef, readIORef) import Data.Text (Text) import qualified Data.Text as T import Langchain.Error (LangchainError, errorMessage) import Langchain.Graph.StateGraph+import Langchain.Model (ChatModel (..), assistantMessage, extractMessageText, userMessage) import Langchain.Runnable (RunnableTree (..), runPure, (|>>))+import Langchain.Stream (StreamEvent (..)) import Test.Tasty import Test.Tasty.HUnit+import Test.Tasty.QuickCheck +newtype MockStreamModel = MockStreamModel [Text]++instance ChatModel MockStreamModel where+  type ModelConfig MockStreamModel = ()++  invoke (MockStreamModel chunks) _ _ =+    pure $ assistantMessage (T.concat chunks)++  stream (MockStreamModel chunks) msgs _ = do+    yield $ LLMStart "mock-run" "mock-stream" msgs+    mapM_ (\c -> yield $ LLMChunk "mock-run" c Nothing) chunks+    yield $ LLMEnd "mock-run" (assistantMessage (T.concat chunks)) Nothing+ tests :: TestTree tests =   testGroup@@ -108,4 +129,117 @@             logs <- readTVarIO logsVar             assertBool "Recorded log messages" (length logs >= 4)             assertBool "Logged entering node 'a'" (any ("Entering node 'a'" `T.isInfixOf`) logs)+    , testCase "streamGraph emits NodeStart and NodeEnd in order" $ do+        let g =+              addEdge "n1" "n2" $+                addEdge "n2" endNodeId $+                  addTransformNode "n1" (<> "->n1") $+                    addTransformNode "n2" (<> "->n2") $+                      emptyStateGraph replaceFieldReducer+        case compileGraph g of+          Left err -> assertFailure $ "Compile failed: " ++ show err+          Right compiled -> do+            eventsRef <- newIORef ([] :: [StreamEvent])+            res <-+              runExceptT $+                runConduit $+                  streamGraph compiled "n1" ("start" :: Text)+                    `fuseUpstream` CL.mapM_ (\ev -> liftIO $ modifyIORef' eventsRef (++ [ev]))+            res @?= Right "start->n1->n2"+            events <- readIORef eventsRef+            let tags =+                  [ (if isStart then "start:" else "end:") <> nid+                  | ev <- events+                  , (isStart, nid) <- case ev of+                      NodeStart _ n _ -> [(True, n)]+                      NodeEnd _ n _ -> [(False, n)]+                      _ -> []+                  ]+            tags @?= ["start:n1", "end:n1", "start:n2", "end:n2"]+    , testCase "addStreamingModelNode emits LLM events between NodeStart and NodeEnd" $ do+        let model = MockStreamModel ["Hello ", "Graph"]+            mergeMsg _ = extractMessageText+            g =+              addEdge "llm" endNodeId $+                addStreamingModelNode "llm" model (\t -> [userMessage t]) mergeMsg $+                  emptyStateGraph replaceFieldReducer+        case compileGraph g of+          Left err -> assertFailure $ "Compile failed: " ++ show err+          Right compiled -> do+            eventsRef <- newIORef ([] :: [StreamEvent])+            res <-+              runExceptT $+                runConduit $+                  streamGraph compiled "llm" ("hi" :: Text)+                    `fuseUpstream` CL.mapM_ (\ev -> liftIO $ modifyIORef' eventsRef (++ [ev]))+            res @?= Right "Hello Graph"+            events <- readIORef eventsRef+            let summary = map eventTag events+            summary+              @?= [ "NodeStart:llm"+                  , "LLMStart"+                  , "LLMChunk:Hello "+                  , "LLMChunk:Graph"+                  , "LLMEnd"+                  , "NodeEnd:llm"+                  ]+    , testCase "streamGraph throws AgentError after NodeStart on recursionLimit" $ do+        let g =+              setRecursionLimit 1 $+                addEdge "loop" "loop" $+                  addTransformNode "loop" (+ (1 :: Int)) $+                    emptyStateGraph replaceFieldReducer+        case compileGraph g of+          Left err -> assertFailure $ "Compile failed: " ++ show err+          Right compiled -> do+            eventsRef <- newIORef ([] :: [StreamEvent])+            res <-+              runExceptT $+                runConduit $+                  streamGraph compiled "loop" (0 :: Int)+                    `fuseUpstream` CL.mapM_ (\ev -> liftIO $ modifyIORef' eventsRef (++ [ev]))+            case res of+              Left err ->+                assertBool "Mentions recursion limit" ("recursion limit exceeded" `T.isInfixOf` errorMessage err)+              Right val -> assertFailure $ "Expected recursion error, got: " ++ show val+            events <- readIORef eventsRef+            let starts = [n | NodeStart _ n _ <- events]+            starts @?= ["loop"]+    , testProperty "linear chain of n nodes emits strictly alternating NodeStart/NodeEnd pairs" $+        \(Positive rawN) -> ioProperty $ do+          let n = ((rawN - 1) `mod` 15) + 1+              nodeNames = ["node-" <> T.pack (show i) | i <- [1 .. n :: Int]]+              withNodes =+                foldr (\nid -> addTransformNode nid (+ (1 :: Int))) (emptyStateGraph replaceFieldReducer) nodeNames+              edges = zip nodeNames (drop 1 nodeNames ++ [endNodeId])+              withEdges = foldr (uncurry addEdge) withNodes edges+          case compileGraph withEdges of+            Left err -> pure $ counterexample (show err) False+            Right compiled -> do+              eventsRef <- newIORef ([] :: [StreamEvent])+              res <-+                runExceptT $+                  runConduit $+                    streamGraph compiled "node-1" (0 :: Int)+                      `fuseUpstream` CL.mapM_ (\ev -> liftIO $ modifyIORef' eventsRef (++ [ev]))+              events <- readIORef eventsRef+              let expected :: [(Text, Text)]+                  expected =+                    concat [[("start", nid), ("end", nid)] | nid <- nodeNames]+                  actual =+                    [ tag+                    | ev <- events+                    , tag <- case ev of+                        NodeStart _ nid _ -> [("start", nid)]+                        NodeEnd _ nid _ -> [("end", nid)]+                        _ -> []+                    ]+              pure $ res === Right n .&&. actual === expected     ]+  where+    eventTag (NodeStart _ nid _) = "NodeStart:" <> nid+    eventTag (NodeEnd _ nid _) = "NodeEnd:" <> nid+    eventTag (LLMStart {}) = "LLMStart"+    eventTag (LLMChunk _ c _) = "LLMChunk:" <> c+    eventTag (LLMEnd {}) = "LLMEnd"+    eventTag _ = "Other"